Fold the sentinel into the scraper and drop the sentinel service (BL-0046)
The stall watchdog now runs inside the scraper (ScraperService.checkStalls, every 30 s from the work loop): same rule as before, one checkpoint-safe restart, quarantine on the second stall, missing checkpoint is no evidence, and paused time no longer counts. Stalls are recorded as scraper incidents (failure.kind 'stall'). The private /api/recovery endpoint is gone. Dropped with the service: reader/scraper-unreachable incidents (supervisor and Docker handle crashes; the health page shows "Scraper: unreachable"), duplicate job-failure incidents and the unused /metrics endpoint. The stack is now reader + scraper (+ optional peer-gateway). The supervisor test moved to runtime/test (npm run test:runtime stays). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
88555d4780
commit
b2bb34174c
37 files changed
+163
-311
No files matched your search
@@ -22,7 +22,7 @@ Runtime recovery is documented in `../docs/resilience.md`. Request-level 5xx ret
|
||||
remain limited to 30/60 seconds; after those are exhausted the daemon can resume the
|
||||
same checkpoint once after a five-minute source cooldown. No automatic access-error
|
||||
retry is permitted. Unavailable titles and parser failures are job-scoped; repeated
|
||||
independent parser failures stop the source lane. Sentinel never changes adapters.
|
||||
independent parser failures stop the source lane. The stall watchdog never changes adapters.
|
||||
|
||||
```text
|
||||
public metadata -> cached series metadata
|
||||
|
||||
@@ -42,7 +42,6 @@ async function body(req) {
|
||||
if (url.pathname === '/api/queue/clear-completed') { const result = await service.clearCompleted(String(input.source || '')); return json(res, 200, { ...result, state: await service.publicState() }); }
|
||||
if (url.pathname === '/api/queue/reorder') { await service.reorder(input.ids); return json(res, 200, await service.publicState()); }
|
||||
if (url.pathname === '/api/control') { await service.control(String(input.action || ''), input.id); return json(res, 200, await service.publicState()); }
|
||||
if (url.pathname === '/api/recovery') return json(res, 200, await service.recover(input));
|
||||
if (url.pathname === '/api/settings') { await service.updateSettings(input); return json(res, 200, await service.publicState()); }
|
||||
if (url.pathname === '/api/discovery/run') { await service.enqueueDiscoveryChecks('manual'); return json(res, 200, await service.publicState()); }
|
||||
if (url.pathname === '/api/library/auto-update') { await service.setMangaAutoUpdate(String(input.id || ''), !!input.enabled); return json(res, 200, await service.publicState()); }
|
||||
|
||||
+31
-14
@@ -124,6 +124,9 @@ class ScraperService {
|
||||
this.resolveAdapter = options.adapterFor || adapterFor;
|
||||
this.adapterList = options.adapters || adapters;
|
||||
this.retryDelayMs = options.retryDelayMs ?? 300000;
|
||||
this.stallMs = options.stallMs ?? 900000;
|
||||
this.stallObservations = new Map();
|
||||
this.stallCheckedAt = 0;
|
||||
this.runtimeHeartbeat = Date.now();
|
||||
}
|
||||
|
||||
@@ -738,22 +741,32 @@ class ScraperService {
|
||||
throw new Error('Unknown control action');
|
||||
}
|
||||
|
||||
async recover(input) {
|
||||
const item = this.state.queue.find(entry => entry.id === input.id);
|
||||
if (!item) throw new Error('Job not found');
|
||||
// Compare observations so a stale watcher cannot disturb a progressing worker.
|
||||
if (Object.values(this.state.currentIds).includes(item.id)) {
|
||||
// Stall watchdog: an active job whose checkpoint has not moved for stallMs (or one
|
||||
// chapter delay + 3 min, if longer) is stopped at its next checkpoint and requeued
|
||||
// once; a second stall quarantines it. A missing checkpoint proves nothing, and time
|
||||
// spent paused does not count.
|
||||
async checkStalls(now = Date.now()) {
|
||||
const activeIds = new Set(Object.values(this.state.currentIds || {}));
|
||||
for (const id of this.stallObservations.keys()) if (!activeIds.has(id)) this.stallObservations.delete(id);
|
||||
for (const item of this.state.queue) {
|
||||
if (!activeIds.has(item.id)) continue;
|
||||
const job = await this.jobState(item);
|
||||
if (!job) return { changed: false };
|
||||
if (progressKey(job) !== input.progressKey || this.state.paused || item.stopReason) return { changed: false };
|
||||
if (Date.now() - Date.parse(job?.updated || item.started) < 900000) return { changed: false };
|
||||
if ((item.stallRecoveries || 0) >= 1) {
|
||||
item.stopReason = 'quarantined';
|
||||
} else { item.stallRecoveries = (item.stallRecoveries || 0) + 1; item.stopReason = 'recovering'; }
|
||||
await this.event('warning', `Sentinel requested recovery at the next checkpoint for ${item.title}`, { id: item.id, source: item.source });
|
||||
return { changed: true };
|
||||
if (!job) { this.stallObservations.delete(item.id); continue; }
|
||||
const key = progressKey(job), prior = this.stallObservations.get(item.id);
|
||||
if (!prior || prior.key !== key) { this.stallObservations.set(item.id, { key, at: now }); continue; }
|
||||
if (this.state.paused || this.state.sourcePaused[item.source] || item.stopReason) { prior.at = now; continue; }
|
||||
if (now - prior.at < Math.max(this.stallMs, Number(job.rate?.chapterMaxMs || 0) + 180000)) continue;
|
||||
this.stallObservations.delete(item.id);
|
||||
const minutes = Math.round((now - prior.at) / 60000), quarantine = (item.stallRecoveries || 0) >= 1;
|
||||
const failure = { kind: 'stall', scope: 'job', retryable: !quarantine, message: quarantine
|
||||
? `No progress for ${minutes} minutes after a restart; quarantined`
|
||||
: `No progress for ${minutes} minutes; restarting from the last checkpoint` };
|
||||
if (quarantine) { item.stopReason = 'quarantined'; item.failure = failure; item.error = failure.message; }
|
||||
else { item.stallRecoveries = (item.stallRecoveries || 0) + 1; item.stopReason = 'recovering'; }
|
||||
this.state.incidents.push({ id: crypto.randomUUID(), at: new Date(now).toISOString(), jobId: item.id, source: item.source, title: item.title, failure, state: 'open' });
|
||||
this.state.incidents = this.state.incidents.slice(-200);
|
||||
await this.event('warning', `${item.title}: ${failure.message}`, { id: item.id, title: item.title, source: item.source });
|
||||
}
|
||||
return { changed: false };
|
||||
}
|
||||
|
||||
recomputeStatus() {
|
||||
@@ -876,6 +889,10 @@ class ScraperService {
|
||||
const prior = this.state.status;
|
||||
this.recomputeStatus();
|
||||
if (prior !== this.state.status) await this.persist();
|
||||
if (Date.now() - this.stallCheckedAt >= 30000) {
|
||||
this.stallCheckedAt = Date.now();
|
||||
await this.checkStalls().catch(error => console.error('[scraper] stall check failed:', safeMessage(error)));
|
||||
}
|
||||
await sleep(500);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -29,12 +29,45 @@ test('status and watchdog read the checkpoint for each source layout', async t =
|
||||
}
|
||||
});
|
||||
|
||||
test('daemon refuses recovery when the checkpoint cannot be read', async () => {
|
||||
const service = Object.create(ScraperService.prototype);
|
||||
service.state = { paused: false, currentIds: { mangapill: 'job' },
|
||||
queue: [{ id: 'job', source: 'mangapill', status: 'downloading' }] };
|
||||
service.jobState = async () => null;
|
||||
const result = await service.recover({ id: 'job', progressKey: '[0,0,"",0,""]' });
|
||||
assert.deepEqual(result, { changed: false });
|
||||
function stallService(job) {
|
||||
const service = new ScraperService({ stallMs: 100 });
|
||||
service.state = { paused: false, sourcePaused: {}, currentIds: { a: 'job' }, incidents: [],
|
||||
queue: [{ id: 'job', source: 'a', title: 'Manga', status: 'downloading' }] };
|
||||
service.jobState = async () => job.value;
|
||||
service.event = async () => {};
|
||||
return service;
|
||||
}
|
||||
|
||||
test('stall check leaves progressing and deliberately paused jobs alone', async () => {
|
||||
const job = { value: { completed: 0, current: { downloadedPages: 1 } } };
|
||||
const service = stallService(job), item = service.state.queue[0];
|
||||
await service.checkStalls(0);
|
||||
job.value = { completed: 0, current: { downloadedPages: 2 } };
|
||||
await service.checkStalls(500000);
|
||||
service.state.paused = true;
|
||||
await service.checkStalls(1000000);
|
||||
assert.equal(item.stopReason, undefined);
|
||||
// Paused time does not count: the clock restarts on resume.
|
||||
service.state.paused = false;
|
||||
await service.checkStalls(1000050);
|
||||
assert.equal(item.stopReason, undefined);
|
||||
await service.checkStalls(1190000);
|
||||
assert.equal(item.stopReason, 'recovering');
|
||||
assert.equal(item.stallRecoveries, 1);
|
||||
assert.equal(service.state.incidents[0].failure.kind, 'stall');
|
||||
});
|
||||
|
||||
test('a second stall quarantines the job', async () => {
|
||||
const service = stallService({ value: { completed: 2, total: 5 } }), item = service.state.queue[0];
|
||||
item.stallRecoveries = 1;
|
||||
await service.checkStalls(0); await service.checkStalls(1000000);
|
||||
assert.equal(item.stopReason, 'quarantined');
|
||||
assert.equal(item.failure.kind, 'stall');
|
||||
});
|
||||
|
||||
test('missing checkpoint does not trigger a false stall recovery', async () => {
|
||||
const service = stallService({ value: null });
|
||||
for (const now of [0, 1000000, 2000000]) await service.checkStalls(now);
|
||||
assert.equal(service.state.queue[0].stopReason, undefined);
|
||||
assert.equal(service.state.incidents.length, 0);
|
||||
});
|
||||
Reference in new issue
Block a user