IMBENJI.NET/Duriin-API
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: let replay recover from dead letter jobs

Browse Source
This commit is contained in:
ImBenji
2026-08-08 23:32:41 +01:00
parent 0344d5ca97
commit 2c023c8962
3 changed files with 104 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
workers/replayWorker.js
+40 -18
View File
@@ -23,7 +23,7 @@ function articleTimeColumns(archiveDb) {
const columns = new Set(archiveDb.prepare('PRAGMA table_info(articles)').all().map((row) => row.name));
const candidates = ['pub_date_effective', 'pub_date', 'ingested_at'].filter((name) => columns.has(name));
if (!candidates.length) throw new Error('archive articles need publication or ingestion timestamps for replay');
return { columns, effective: `COALESCE(${candidates.join(', ')})` };
return { columns, effective: candidates.length === 1 ? candidates[0] : `COALESCE(${candidates.join(', ')})` };
}
function replayPrompt(article) {
@@ -52,23 +52,45 @@ function scheduleNext(db, archiveDb, run) {
const { columns, effective } = articleTimeColumns(archiveDb);
const content = columns.has('content') ? "content IS NOT NULL AND content != ''" : '1=1';
const indexFilter = columns.has('is_index_page') ? 'AND (is_index_page = 0 OR is_index_page IS NULL)' : '';
const cursorFilter = run.cursor_effective_at
? `AND (datetime(${effective}) > datetime(?) OR (datetime(${effective}) = datetime(?) AND id > ?))`
: '';
const params = [run.watermark_at];
if (run.cursor_effective_at) params.push(run.cursor_effective_at, run.cursor_effective_at, run.cursor_article_id);
const article = archiveDb.prepare(`
SELECT id, title, description, content, ${effective} AS effective_at
FROM articles
WHERE ${content} ${indexFilter} AND datetime(${effective}) <= datetime(?) ${cursorFilter}
ORDER BY datetime(${effective}) ASC, id ASC LIMIT 1
`).get(...params);
if (!article) return null;
enqueueJob(db, {
jobType: 'replay_article', lane: 'historical', priority: 1, entityType: 'article', entityId: article.id,
idempotencyKey: `replay:${run.id}:article:${article.id}`,
});
return article;
let cursorEffectiveAt = run.cursor_effective_at;
let cursorArticleId = run.cursor_article_id;
for (let skipped = 0; skipped < 100; skipped += 1) {
const cursorFilter = cursorEffectiveAt
? `AND (datetime(${effective}) > datetime(?) OR (datetime(${effective}) = datetime(?) AND id > ?))`
: '';
const params = [run.watermark_at];
if (cursorEffectiveAt) params.push(cursorEffectiveAt, cursorEffectiveAt, cursorArticleId);
const article = archiveDb.prepare(`
SELECT id, title, description, content, ${effective} AS effective_at
FROM articles
WHERE ${content} ${indexFilter} AND datetime(${effective}) <= datetime(?) ${cursorFilter}
ORDER BY datetime(${effective}) ASC, id ASC LIMIT 1
`).get(...params);
if (!article) return null;
const idempotencyKey = `replay:${run.id}:article:${article.id}`;
const existing = db.prepare('SELECT status, last_error FROM autonomy_jobs WHERE idempotency_key = ?').get(idempotencyKey);
if (existing && ['complete', 'dead_letter'].includes(existing.status)) {
db.prepare(`
UPDATE autonomy_replay_runs
SET cursor_article_id=?, cursor_effective_at=?, last_error=?, updated_at=datetime('now')
WHERE id=?
`).run(article.id, article.effective_at, existing.status === 'dead_letter'
? `Skipped dead-letter replay job for article ${article.id}: ${existing.last_error || 'unknown error'}`
: run.last_error, run.id);
cursorEffectiveAt = article.effective_at;
cursorArticleId = article.id;
continue;
}
enqueueJob(db, {
jobType: 'replay_article', lane: 'historical', priority: 1, entityType: 'article', entityId: article.id,
idempotencyKey,
});
return article;
}
throw new Error('replay scheduler skipped too many terminal jobs in one pass');
}
async function runReplayWorker({ archivePath, intelligencePath, workerId = `replay-${os.hostname()}-${process.pid}`, pollMs = 15000 } = {}) {