Files
MaashkinoozheandClaude Opus 5.5 f8f21ab42a Backup: probe the target with a real write instead of access(W_OK)
On the server the backup folder is a Synology NFS share with "map all
users to admin": access(W_OK) returns EACCES for uid 1000 although
writes succeed, so the first scheduled run failed with "not usable".
Create and remove a probe file instead; test that it leaves nothing.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-22 18:12:41 +02:00

403 lines
20 KiB
JavaScript
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
'use strict';
/*
* Built-in backup of the reader's DATA tree into BACKUP (see docs/backup.md).
*
* One run: pause the scraper queue through its control API, wait (bounded) until no
* job is active, write a gzip'd tar of DATA into BACKUP, resume the queue only if this
* run paused it, then apply retention. No container stops. db.json comes from the
* reader's in-memory state; scraper/queue.json is stored with `paused: true` so a
* restored copy never starts downloading by itself. Runtime/lock/pid/temp files are
* left out. Retention only ever deletes `circus-YYYYMMDD-HHMMSS.tar.gz` files directly
* inside BACKUP. No dependencies: the tar writer/reader below is plain ustar + PAX.
*/
const fs = require('fs'), fsp = fs.promises, path = require('path'), zlib = require('zlib');
const { Readable } = require('stream'), { pipeline } = require('stream/promises');
const ARCHIVE = /^circus-(\d{4})(\d{2})(\d{2})-(\d{2})(\d{2})(\d{2})\.tar\.gz$/;
const PART = /^circus-\d{8}-\d{6}\.tar\.gz\.part$/;
const EXCLUDE = [/^runtime.*\.json$/, /\.lock$/, /\.pid$/, /\.tmp$/];
const DEFAULT_SETTINGS = { enabled: false, time: '03:30', retention: { mode: 'count', value: 14 } };
const RETENTION_LIMITS = { count: [1, 365], days: [1, 3650] };
const CATCH_UP_MS = 24 * 3600e3;
class BackupError extends Error { constructor(code, msg) { super(msg); this.code = code; } }
const delay = ms => new Promise(resolve => setTimeout(resolve, ms));
const pad = n => String(n).padStart(2, '0');
const stampOf = d => `${d.getFullYear()}${pad(d.getMonth() + 1)}${pad(d.getDate())}-${pad(d.getHours())}${pad(d.getMinutes())}${pad(d.getSeconds())}`;
const archiveTime = name => { const m = ARCHIVE.exec(name); return m ? new Date(+m[1], m[2] - 1, +m[3], +m[4], +m[5], +m[6]).getTime() : NaN; };
const message = e => String((e && e.message) || e).slice(0, 500);
function validateSettings(input, current = DEFAULT_SETTINGS) {
const next = { ...current, retention: { ...current.retention } };
if (input.enabled !== undefined) {
if (typeof input.enabled !== 'boolean') throw new BackupError(400, 'enabled must be true or false');
next.enabled = input.enabled;
}
if (input.time !== undefined) {
if (!/^([01]\d|2[0-3]):[0-5]\d$/.test(String(input.time))) throw new BackupError(400, 'The backup time must be HH:MM');
next.time = String(input.time);
}
if (input.retention !== undefined) {
const mode = input.retention && input.retention.mode, value = Number(input.retention && input.retention.value);
const limits = RETENTION_LIMITS[mode];
if (!limits) throw new BackupError(400, 'Retention is "count" or "days"');
if (!Number.isInteger(value) || value < limits[0] || value > limits[1]) throw new BackupError(400, `Keep ${limits[0]}–${limits[1]} ${mode === 'count' ? 'backups' : 'days'}`);
next.retention = { mode, value };
}
return next;
}
// Next local occurrence of HH:MM strictly after `now` (process TZ).
function nextOccurrence(time, now) {
const [hh, mm] = time.split(':').map(Number), d = new Date(now);
const at = new Date(d.getFullYear(), d.getMonth(), d.getDate(), hh, mm, 0, 0);
if (at.getTime() <= now) at.setDate(at.getDate() + 1);
return at.getTime();
}
// ---------- tar (ustar + PAX path records), enough for `tar -xzf` everywhere ----------
function octal(buf, offset, length, value) { buf.write(value.toString(8).padStart(length - 1, '0') + '\0', offset, length, 'ascii'); }
function rawHeader(name, { size = 0, mode = 0o644, uid = 0, gid = 0, mtime = 0, type = '0' }) {
const b = Buffer.alloc(512);
b.write(name, 0, 100, 'utf8');
octal(b, 100, 8, mode & 0o7777); octal(b, 108, 8, uid & 0o7777777); octal(b, 116, 8, gid & 0o7777777);
octal(b, 124, 12, size); octal(b, 136, 12, Math.max(0, Math.floor(mtime / 1000)));
b.write(' ', 148, 8, 'ascii');
b.write(type, 156, 1, 'ascii');
b.write('ustar\0' + '00', 257, 8, 'ascii');
let sum = 0; for (const byte of b) sum += byte;
b.write(sum.toString(8).padStart(6, '0') + '\0 ', 148, 8, 'ascii');
return b;
}
const padding = size => Buffer.alloc((512 - (size % 512)) % 512);
function paxRecord(key, value) {
const body = ` ${key}=${value}\n`, bytes = Buffer.byteLength(body);
let length = bytes + String(bytes).length;
if (String(length).length !== String(bytes).length) length = bytes + String(length).length;
return `${length}${body}`;
}
function tarEntry(name, data, meta) {
const parts = [];
// Names longer than the ustar field or non-ASCII go into a PAX header first.
if (Buffer.byteLength(name) > 100 || /[^\x20-\x7e]/.test(name)) {
const pax = Buffer.from(paxRecord('path', name), 'utf8');
parts.push(rawHeader('PaxHeader/' + name.replace(/[^\x20-\x7e]/g, '_').slice(-80), { size: pax.length, type: 'x', mtime: meta.mtime }), pax, padding(pax.length));
name = name.replace(/[^\x20-\x7e]/g, '_').slice(-100);
}
parts.push(rawHeader(name, { ...meta, size: data ? data.length : 0 }));
if (data && data.length) parts.push(data, padding(data.length));
return parts;
}
// Streaming reader for verification and tests: [{ name, type, size, mode, mtime, data? }].
async function readArchive(file, { contents = false } = {}) {
const entries = []; let pending = Buffer.alloc(0), current = null, paxPath = null, ended = false;
const consume = chunk => {
pending = pending.length ? Buffer.concat([pending, chunk]) : chunk;
for (;;) {
if (ended) return;
if (current) {
if (pending.length < current.need) return;
const body = pending.subarray(0, current.size);
if (current.entry.type === 'x') paxPath = /(?:^|\n)\d+ path=([^\n]*)\n/.exec(body.toString('utf8'))?.[1] ?? null;
else { if (contents) current.entry.data = Buffer.from(body); entries.push(current.entry); }
pending = pending.subarray(current.need); current = null; continue;
}
if (pending.length < 512) return;
const h = pending.subarray(0, 512); pending = pending.subarray(512);
if (h.every(byte => byte === 0)) { ended = true; return; }
let sum = 0; for (let i = 0; i < 512; i++) sum += i >= 148 && i < 156 ? 32 : h[i];
if (sum !== parseInt(h.toString('ascii', 148, 156), 8)) throw new Error('Corrupt tar header');
const field = (o, l) => h.toString('utf8', o, o + l).replace(/\0.*$/s, '');
const size = parseInt(field(124, 12), 8) || 0, type = field(156, 1) || '0';
const prefix = field(345, 155), name = prefix ? prefix + '/' + field(0, 100) : field(0, 100);
const entry = { name: type === 'x' ? name : (paxPath ?? name), type, size, mode: parseInt(field(100, 8), 8), mtime: parseInt(field(136, 12), 8) * 1000 };
if (type !== 'x') paxPath = null;
current = { entry, size, need: size + padding(size).length };
}
};
await pipeline(fs.createReadStream(file), zlib.createGunzip(), async function* (source) { for await (const chunk of source) consume(chunk); });
if (!ended) throw new Error('Archive ends without the tar end marker');
return entries;
}
class BackupService {
/**
* dataRoot DATA (read) backupRoot BACKUP (rw) or '' = feature off
* stateFile settings + history snapshotDb () => db.json text from memory
* scraper { state(), control(action) } or null
*/
constructor({ dataRoot, backupRoot, stateFile, snapshotDb, scraper = null, now = Date.now, log = console,
idleTimeoutMs = 15 * 60e3, pollMs = 2000, resumeRetryMs = 5000 }) {
this.dataRoot = path.resolve(dataRoot);
this.backupRoot = backupRoot ? path.resolve(backupRoot) : '';
this.stateFile = stateFile || path.join(this.dataRoot, 'backup.json');
this.snapshotDb = snapshotDb; this.scraper = scraper; this.now = now; this.log = log;
this.idleTimeoutMs = idleTimeoutMs; this.pollMs = pollMs; this.resumeRetryMs = resumeRetryMs;
this.state = { version: 1, settings: { ...DEFAULT_SETTINGS }, nextAt: null, last: null, history: [], pending: null };
this.running = null; this.archives = []; this.targetError = null; this.timer = null;
}
load() {
try { Object.assign(this.state, JSON.parse(fs.readFileSync(this.stateFile, 'utf8'))); }
catch (e) { if (e.code !== 'ENOENT') this.log.warn(`[backup] ignoring unreadable ${this.stateFile}: ${e.message}`); }
this.state.settings = validateSettings(this.state.settings || {}, DEFAULT_SETTINGS);
this.state.history = Array.isArray(this.state.history) ? this.state.history.slice(-10) : [];
return this;
}
persist() {
fs.writeFileSync(this.stateFile + '.tmp', JSON.stringify(this.state, null, 2) + '\n');
fs.renameSync(this.stateFile + '.tmp', this.stateFile);
}
// A run interrupted by a restart may have left the queue paused on its behalf.
async recoverInterrupted() {
const pending = this.state.pending;
if (!pending) return;
this.state.pending = null;
const result = { at: pending.startedAt, trigger: pending.trigger, error: 'Interrupted by a reader restart', warnings: [] };
if (pending.pausedByUs) {
try { await this.resume(); result.warnings.push('Queue resumed after the interrupted backup'); }
catch (e) { result.warnings.push(`Queue left paused: ${message(e)}`); }
}
this.record(result);
}
async refreshTarget() {
if (!this.backupRoot) { this.targetError = null; this.archives = []; return; }
try {
const st = await fsp.stat(this.backupRoot);
if (!st.isDirectory()) throw new Error('not a directory');
// Probe with a real write: access(W_OK) is unreliable on NFS with squashed ids (Synology
// "map all users to admin" denies it although writes succeed).
const probe = path.join(this.backupRoot, `.circus-backup-probe-${process.pid}`);
await fsp.writeFile(probe, '');
await fsp.rm(probe, { force: true });
const names = (await fsp.readdir(this.backupRoot)).filter(name => ARCHIVE.test(name)).sort();
this.archives = await Promise.all(names.map(async name => ({ name, size: (await fsp.stat(path.join(this.backupRoot, name))).size, at: archiveTime(name) })));
this.targetError = null;
} catch (e) { this.targetError = `Backup folder ${this.backupRoot} is not usable: ${e.code || message(e)}`; this.archives = []; }
}
publicState() {
const s = this.state;
return {
configured: !!this.backupRoot, path: this.backupRoot || null, targetError: this.targetError,
settings: s.settings, nextAt: s.settings.enabled && this.backupRoot ? s.nextAt : null,
running: this.running && { ...this.running }, last: s.last, history: s.history.slice().reverse(),
archives: { count: this.archives.length, bytes: this.archives.reduce((n, a) => n + a.size, 0),
newest: this.archives.length ? this.archives[this.archives.length - 1].name : null },
};
}
updateSettings(input) {
this.state.settings = validateSettings(input || {}, this.state.settings);
this.state.nextAt = this.state.settings.enabled ? nextOccurrence(this.state.settings.time, this.now()) : null;
this.persist();
return this.publicState();
}
start(intervalMs = 30000) {
this.load();
this.refreshTarget().then(() => this.recoverInterrupted()).catch(e => this.log.error('[backup]', message(e)));
this.timer = setInterval(() => this.tick(), intervalMs);
this.timer.unref?.();
return this;
}
stop() { clearInterval(this.timer); }
// Scheduled runs: catch up a missed slot from the last day, never older ones
// (e.g. right after restoring an old backup.json).
async tick() {
const s = this.state.settings, now = this.now();
if (!s.enabled || !this.backupRoot || this.running) return null;
if (!this.state.nextAt) { this.state.nextAt = nextOccurrence(s.time, now); this.persist(); return null; }
if (now < this.state.nextAt) return null;
const due = now - this.state.nextAt < CATCH_UP_MS;
this.state.nextAt = nextOccurrence(s.time, now); this.persist();
return due ? this.run('schedule').catch(() => null) : null;
}
/** Starts a run and resolves with its result; rejects at once when no run is possible. */
run(trigger = 'manual') {
if (!this.backupRoot) return Promise.reject(new BackupError(409, 'Backups are off: set BACKUP to a mounted folder'));
if (this.running) return Promise.reject(new BackupError(409, 'A backup is already running'));
this.running = { startedAt: this.now(), trigger, phase: 'starting' };
return this.perform(trigger).finally(() => { this.running = null; });
}
async perform(trigger) {
const started = this.now(), result = { at: started, trigger, warnings: [] };
let part = null, pausedByUs = false;
try {
await this.refreshTarget();
if (this.targetError) throw new Error(this.targetError);
await this.removeStaleParts();
// 1. Pause the queue (unless already paused) and wait for the active chapter.
let scraperState = null;
if (this.scraper) {
try { scraperState = await this.scraper.state(); }
catch (e) { result.warnings.push(`Scraper unreachable, queue not paused: ${message(e)}`); }
}
if (scraperState) {
result.queueWasPaused = !!scraperState.paused;
if (!scraperState.paused) {
this.state.pending = { startedAt: started, trigger, pausedByUs: true }; this.persist();
await this.scraper.control('pause'); pausedByUs = true;
}
this.running.phase = 'waiting for the scraper to finish its current step';
await this.waitIdle(scraperState);
}
// 2. Snapshot DATA into BACKUP/<name>.part, then rename.
this.running.phase = 'writing archive';
let stamp = started, name, file;
do { name = `circus-${stampOf(new Date(stamp))}.tar.gz`; file = path.join(this.backupRoot, name); stamp += 1000; }
while (fs.existsSync(file)); // two runs within one second must not overwrite each other
part = file + '.part';
const stats = await this.writeArchive(part);
this.running.phase = 'verifying archive';
const entries = await readArchive(part);
const files = entries.filter(e => e.type === '0').length;
if (files !== stats.files) throw new Error(`Verification read ${files} files, expected ${stats.files}`);
await fsp.rename(part, file); part = null;
Object.assign(result, { file: name, size: (await fsp.stat(file)).size, files, dataBytes: stats.bytes });
} catch (e) {
result.error = message(e);
if (part) await fsp.rm(part, { force: true }).catch(() => {});
} finally {
// 3. Resume only what this run paused.
if (pausedByUs) {
this.running.phase = 'resuming queue';
try { await this.resume(); result.queueResumed = true; }
catch (e) { result.warnings.push(`Queue could not be resumed, resume it in Settings → Scraper: ${message(e)}`); }
}
if (this.state.pending) this.state.pending = null;
}
// 4. Retention only after a successful run, so failures never thin out old copies.
if (!result.error) {
this.running.phase = 'applying retention';
try { result.deleted = await this.applyRetention(result.file); }
catch (e) { result.warnings.push(`Retention failed: ${message(e)}`); }
}
result.durationMs = this.now() - started;
await this.refreshTarget();
this.record(result);
if (result.error) this.log.error(`[backup] failed: ${result.error}`);
else this.log.log(`[backup] ${result.file} ${result.size} bytes in ${result.durationMs} ms`);
return result;
}
record(result) {
this.state.last = result;
this.state.history = [...this.state.history, result].slice(-10);
this.persist();
}
async waitIdle(initial) {
const deadline = this.now() + this.idleTimeoutMs;
let state = initial, lastError = null;
for (;;) {
if (state && !(state.active || []).length && state.status !== 'pausing') return;
if (this.now() >= deadline) {
throw new Error(`The scraper did not become idle within ${Math.round(this.idleTimeoutMs / 60000)} min${lastError ? ` (${lastError})` : ''}; no backup written`);
}
await delay(this.pollMs);
try { state = await this.scraper.state(); lastError = null; }
catch (e) { state = null; lastError = message(e); }
}
}
async resume() {
let error;
for (let attempt = 0; attempt < 3; attempt++) {
try { await this.scraper.control('resume'); return; }
catch (e) { error = e; if (attempt < 2) await delay(this.resumeRetryMs); }
}
throw error;
}
// Files of DATA in a stable order, skipping excluded names, symlinks and a nested BACKUP.
async collect(dir = this.dataRoot, rel = '') {
const out = [];
const entries = (await fsp.readdir(dir, { withFileTypes: true })).sort((a, b) => (a.name < b.name ? -1 : a.name > b.name ? 1 : 0));
for (const entry of entries) {
const full = path.join(dir, entry.name), relative = rel ? rel + '/' + entry.name : entry.name;
if (entry.isDirectory()) {
if (full === this.backupRoot) continue;
out.push({ full, relative, dir: true });
out.push(...await this.collect(full, relative));
} else if (entry.isFile() && !EXCLUDE.some(re => re.test(entry.name))) out.push({ full, relative, dir: false });
}
return out;
}
// Contents of one file as it goes into the archive.
async contentOf(item) {
if (item.relative === 'db.json' && this.snapshotDb) return Buffer.from(this.snapshotDb());
if (item.relative === path.relative(this.dataRoot, this.stateFile).split(path.sep).join('/')) {
const { pending, ...rest } = this.state; void pending;
return Buffer.from(JSON.stringify(rest, null, 2) + '\n');
}
const data = await fsp.readFile(item.full);
if (item.relative === 'scraper/queue.json') {
const queue = JSON.parse(data.toString('utf8'));
queue.paused = true; queue.status = 'paused';
return Buffer.from(JSON.stringify(queue, null, 2) + '\n');
}
return data;
}
async writeArchive(part) {
const items = await this.collect(), stats = { files: 0, bytes: 0 };
// The in-memory db is archived even before its first debounced write.
if (this.snapshotDb && !items.some(item => item.relative === 'db.json')) items.unshift({ full: path.join(this.dataRoot, 'db.json'), relative: 'db.json', dir: false, memory: true });
const manifest = { kind: 'mangacircus-data-backup', version: 1, created: new Date(this.now()).toISOString(),
excluded: 'runtime*.json, *.lock, *.pid, *.tmp', queuePaused: true };
const self = this;
async function* tar() {
yield* tarEntry('.circus-backup.json', Buffer.from(JSON.stringify(manifest, null, 2) + '\n'), { mtime: self.now() });
for (const item of items) {
let st;
try { st = item.memory ? { mode: 0o644, uid: process.getuid?.() ?? 0, gid: process.getgid?.() ?? 0, mtimeMs: self.now() } : await fsp.stat(item.full); }
catch (e) { if (e.code === 'ENOENT') continue; throw e; }
const meta = { mode: st.mode, uid: st.uid, gid: st.gid, mtime: st.mtimeMs };
if (item.dir) { yield* tarEntry(item.relative + '/', null, { ...meta, type: '5' }); continue; }
let data;
try { data = await self.contentOf(item); } catch (e) { if (e.code === 'ENOENT') continue; throw new Error(`${item.relative}: ${message(e)}`); }
stats.files++; stats.bytes += data.length;
yield* tarEntry(item.relative, data, meta);
}
yield Buffer.alloc(1024);
}
await pipeline(Readable.from(tar()), zlib.createGzip(), fs.createWriteStream(part, { flags: 'wx' }));
const handle = await fsp.open(part, 'r+'); // fsync before the rename makes it visible
try { await handle.sync(); } finally { await handle.close(); }
stats.files++; // the manifest
return stats;
}
async removeStaleParts() {
for (const name of await fsp.readdir(this.backupRoot)) if (PART.test(name)) await fsp.rm(path.join(this.backupRoot, name), { force: true });
}
// Deletes only our own archive names directly inside BACKUP; never the newest one.
async applyRetention(keepName) {
const { mode, value } = this.state.settings.retention;
const names = (await fsp.readdir(this.backupRoot)).filter(name => ARCHIVE.test(name)).sort();
let doomed;
if (mode === 'count') doomed = names.slice(0, Math.max(0, names.length - value));
else { const cutoff = this.now() - value * 864e5; doomed = names.filter(name => archiveTime(name) < cutoff); }
doomed = doomed.filter(name => name !== keepName && name !== names[names.length - 1]);
for (const name of doomed) {
const target = path.join(this.backupRoot, name);
if (path.dirname(target) !== this.backupRoot) continue;
await fsp.rm(target);
}
return doomed.length;
}
}
module.exports = { BackupService, BackupError, readArchive, validateSettings, nextOccurrence, DEFAULT_SETTINGS, ARCHIVE };