Add built-in backup of the data folder (BL-0044)

The reader owns a scheduled backup into BACKUP (Compose: BACKUP_PATH mounted at
/backup, off when unset). A run pauses the scraper queue through its control
API, waits (bounded) until no job is active, writes a verified tar.gz of DATA
with db.json from memory and scraper/queue.json stored paused, excludes
runtime/lock/pid/temp files, resumes only a queue it paused, then applies
count- or day-based retention to its own archive names only. Schedule,
retention, "Back up now" and last result live in Settings > Service health.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
MaashkinoozheandClaude Opus 5.5 committed 2026-10-22 18:03:21 +02:00
1 parent fcc50d576c
commit 086f6353f8
15 files changed
+904 -11

No files matched your search

+398
View File
@@ -0,0 +1,398 @@
'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');
await fsp.access(this.backupRoot, fs.constants.W_OK);
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 };