66 lines
4.1 KiB
JavaScript
66 lines
4.1 KiB
JavaScript
'use strict';
|
|
const { progressKey } = require('../scraper/lib/failures');
|
|
|
|
// No source requests, filesystem checkpoint edits, shell execution or Docker socket.
|
|
class Monitor {
|
|
constructor(options = {}) {
|
|
this.now = options.now || Date.now;
|
|
this.stallMs = options.stallMs || 900000;
|
|
this.observations = new Map();
|
|
this.incidents = options.incidents || [];
|
|
this.state = { status: 'starting', services: {}, incidents: this.incidents };
|
|
}
|
|
incident(key, message, extra = {}) {
|
|
let incident = this.incidents.find(i => i.key === key && i.state === 'open');
|
|
if (!incident) {
|
|
incident = { key, message, state: 'open', at: new Date(this.now()).toISOString(), ...extra };
|
|
this.incidents.push(incident);
|
|
if (this.incidents.length > 200) this.incidents.splice(0, this.incidents.length - 200);
|
|
}
|
|
return incident;
|
|
}
|
|
resolve(key) {
|
|
for (const i of this.incidents) if (i.key === key && i.state === 'open') { i.state = 'resolved'; i.resolvedAt = new Date(this.now()).toISOString(); }
|
|
}
|
|
async observe(reader, scraper, recover) {
|
|
const now = this.now();
|
|
this.state.checkedAt = new Date(now).toISOString();
|
|
this.state.services = { reader: !!reader?.ok, scraper: !!scraper };
|
|
if (reader?.ready === false) this.incident('reader:scan', 'Reader is alive but its library scan failed; inspect storage and scan diagnostics', { service: 'reader' });
|
|
else if (reader?.ok) this.resolve('reader:scan');
|
|
for (const [service, healthy] of Object.entries(this.state.services)) {
|
|
if (healthy) this.resolve(`service:${service}`);
|
|
else this.incident(`service:${service}`, `${service} is unreachable or unhealthy; inspect its supervisor`, { service });
|
|
}
|
|
if (!scraper) { this.state.status = 'degraded'; return this.state; }
|
|
const ids = new Set();
|
|
for (const entry of scraper.active || []) {
|
|
const id = entry.current.id; ids.add(id);
|
|
// Without a checkpoint, a null progress key stays constant forever and
|
|
// cannot prove a stall. Do not quarantine a live worker on that basis.
|
|
if (!entry.job) { this.observations.delete(id); this.resolve(`stall:${id}`); continue; }
|
|
const key = progressKey(entry.job), prior = this.observations.get(id);
|
|
if (!prior || prior.key !== key) { this.observations.set(id, { key, at: now }); this.resolve(`stall:${id}`); continue; }
|
|
if (scraper.paused || scraper.sourcePaused?.[entry.source]) continue;
|
|
const allowance = Math.max(this.stallMs, Number(entry.job?.rate?.chapterMaxMs || 0) + 180000);
|
|
if (now - prior.at < allowance) continue;
|
|
const incident = this.incident(`stall:${id}`, 'Job progress has stopped; recovery is limited to one checkpoint-safe attempt', { jobId: id, source: entry.source, title: entry.current.title });
|
|
if (!incident.attempted && recover) {
|
|
// Persist attempted in the daemon via stopReason/stallRecoveries; monitor records audit too.
|
|
incident.attempted = true;
|
|
try { incident.recovery = await recover({ id, progressKey: key }); }
|
|
catch { incident.recovery = { changed: false, error: 'Recovery API unavailable' }; }
|
|
}
|
|
}
|
|
for (const id of this.observations.keys()) if (!ids.has(id)) { this.observations.delete(id); this.resolve(`stall:${id}`); }
|
|
const failures = (scraper.queue || []).filter(item => ['error', 'quarantined'].includes(item.status));
|
|
for (const item of failures) this.incident(`job:${item.id}`, item.failure?.message || item.error || 'Job needs attention', { jobId: item.id, source: item.source, kind: item.failure?.kind || 'unknown' });
|
|
for (const i of this.incidents) if (i.key.startsWith('job:') && !failures.some(item => i.key === `job:${item.id}`)) this.resolve(i.key);
|
|
this.state.sourcePaused = scraper.sourcePaused || {};
|
|
this.state.active = (scraper.active || []).map(e => ({ source: e.source, id: e.current.id, title: e.current.title, completed: e.job?.completed, total: e.job?.total, updated: e.job?.updated }));
|
|
this.state.status = this.incidents.some(i => i.state === 'open') ? 'attention' : scraper.paused ? 'paused' : 'healthy';
|
|
return this.state;
|
|
}
|
|
}
|
|
module.exports = { Monitor };
|