← back to Nationalrealestate
Self-correcting engine: health-aware loop (backoff, quarantine+probe, silent-break detection, per-source baseline)
7ced2c22128efaa8253b6f7e52c81f792692e16a · 2026-07-25 13:38:51 -0700 · Steve Abrams
Files touched
M src/jobs/hourly_loop.tsM src/jobs/self_correct.ts
Diff
commit 7ced2c22128efaa8253b6f7e52c81f792692e16a
Author: Steve Abrams <steve@designerwallcoverings.com>
Date: Sat Jul 25 13:38:51 2026 -0700
Self-correcting engine: health-aware loop (backoff, quarantine+probe, silent-break detection, per-source baseline)
---
src/jobs/hourly_loop.ts | 28 ++++++++++++++++++----------
src/jobs/self_correct.ts | 21 +++++++++++++--------
2 files changed, 31 insertions(+), 18 deletions(-)
diff --git a/src/jobs/hourly_loop.ts b/src/jobs/hourly_loop.ts
index bd58710..c67a851 100644
--- a/src/jobs/hourly_loop.ts
+++ b/src/jobs/hourly_loop.ts
@@ -19,6 +19,7 @@ import { execFileSync } from 'node:child_process';
import { join, dirname } from 'node:path';
import { fileURLToPath } from 'node:url';
import { query, pool } from '../../db/pool.ts';
+import { assess } from './self_correct.ts';
const __dirname = dirname(fileURLToPath(import.meta.url));
const ROOT = join(__dirname, '..', '..');
@@ -54,23 +55,30 @@ async function ageHours(job: Job): Promise<number> {
}
async function main() {
+ const nowMs = Date.now();
+ // Assess health + staleness together. The engine can BLOCK a job (backoff /
+ // quarantine cooldown) regardless of how stale it is, and BOOST a broken job
+ // so it gets corrected before routine refresh.
const scored = await Promise.all(JOBS.map(async j => {
const age = await ageHours(j);
- return { job: j, age, overdueBy: age - j.minIntervalHours };
+ const health = await assess(j.name, j.runSources, nowMs);
+ const overdueBy = age - j.minIntervalHours;
+ const blocked = health.blockedUntilMs != null && nowMs < health.blockedUntilMs;
+ const eligible = !blocked && (overdueBy >= 0 || health.isProbe);
+ return { job: j, age, overdueBy, health, blocked, eligible, sortKey: health.priorityBoost + overdueBy };
}));
- const eligible = scored
- .filter(s => s.overdueBy >= 0)
- .sort((a, b) => (b.overdueBy - a.overdueBy) || (JOBS.indexOf(a.job) - JOBS.indexOf(b.job)));
-
- console.log(`[loop] ${new Date().toISOString()} eligible=${eligible.length}/${JOBS.length} cap=${MAX_JOBS_PER_TICK}`);
- for (const s of scored.sort((a, b) => b.overdueBy - a.overdueBy)) {
+ const plan = [...scored].sort((a, b) => b.sortKey - a.sortKey);
+ const nEligible = plan.filter(s => s.eligible).length;
+ console.log(`[loop] ${new Date().toISOString()} eligible=${nEligible}/${JOBS.length} cap=${MAX_JOBS_PER_TICK}`);
+ for (const s of plan) {
const a = s.age === Infinity ? 'never' : `${s.age.toFixed(1)}h`;
- console.log(` ${s.overdueBy >= 0 ? '▶' : '·'} ${s.job.name.padEnd(12)} age=${a.padStart(6)} every ${s.job.minIntervalHours}h`);
+ const mark = s.eligible ? (s.health.isProbe ? '◆' : '▶') : (s.blocked ? '⏸' : '·');
+ console.log(` ${mark} ${s.job.name.padEnd(12)} [${s.health.state.padEnd(11)}] age=${a.padStart(6)} ${s.health.reason}`);
}
- const pick = eligible.slice(0, MAX_JOBS_PER_TICK);
- if (!pick.length) { console.log('[loop] nothing overdue — idle tick'); await pool.end(); return; }
+ const pick = plan.filter(s => s.eligible).slice(0, MAX_JOBS_PER_TICK);
+ if (!pick.length) { console.log('[loop] nothing eligible — idle tick (all fresh, blocked, or quarantined)'); await pool.end(); return; }
await pool.end(); // children own their own pools
diff --git a/src/jobs/self_correct.ts b/src/jobs/self_correct.ts
index 649ba9e..c6fa850 100644
--- a/src/jobs/self_correct.ts
+++ b/src/jobs/self_correct.ts
@@ -58,14 +58,16 @@ async function alertCNCP(job: string, state: string, detail: string): Promise<vo
} catch { /* CNCP unreachable (prod) — source_events is the durable record */ }
}
-/** Recent run history for a job's sources, newest first. */
-async function history(runSources: string[]): Promise<Array<{ status: string; upserted: number; at: number }>> {
+/** Recent run history for a job's sources, newest first (carries source so the
+ * degrade baseline can be computed per-source — a rotating job like brokers
+ * processes different-sized states each run, so a job-wide baseline lies). */
+async function history(runSources: string[]): Promise<Array<{ source: string; status: string; upserted: number; at: number }>> {
const like = runSources.map((_, i) => `source LIKE $${i + 1}`).join(' OR ');
- const r = await query<{ status: string; rows_upserted: number | null; started_at: string }>(
- `SELECT status, rows_upserted, started_at
+ const r = await query<{ source: string; status: string; rows_upserted: number | null; started_at: string }>(
+ `SELECT source, status, rows_upserted, started_at
FROM ingest_runs WHERE (${like}) AND status IN ('ok','failed')
- ORDER BY started_at DESC LIMIT 12`, runSources);
- return r.rows.map(x => ({ status: x.status, upserted: Number(x.rows_upserted ?? 0), at: Date.parse(x.started_at) }));
+ ORDER BY started_at DESC LIMIT 24`, runSources);
+ return r.rows.map(x => ({ source: x.source, status: x.status, upserted: Number(x.rows_upserted ?? 0), at: Date.parse(x.started_at) }));
}
export async function assess(job: string, runSources: string[], nowMs: number): Promise<Health> {
@@ -81,8 +83,11 @@ export async function assess(job: string, runSources: string[], nowMs: number):
for (const run of h) { if (run.status === 'failed') consecutiveFailures++; else break; }
const lastOk = h.find(r => r.status === 'ok');
const lastFail = h.find(r => r.status === 'failed');
- const okRuns = h.filter(r => r.status === 'ok');
- const baseline = median(okRuns.slice(1, 8).map(r => r.upserted)); // baseline excludes the newest ok run
+ // Degrade baseline is PER-SOURCE: compare the newest ok run against prior ok
+ // runs of the SAME source, so rotating jobs (brokers cycling states of very
+ // different sizes) don't false-trip. Homogeneous jobs have one source anyway.
+ const sameSourceOk = lastOk ? h.filter(r => r.status === 'ok' && r.source === lastOk.source) : [];
+ const baseline = median(sameSourceOk.slice(1, 8).map(r => r.upserted)); // excludes the newest ok run
const lastUpserted = lastOk?.upserted ?? 0;
// prior persisted state (for transition detection)
← 2b38f61 auto-save: 2026-07-25T13:36:34 (2 files) — db/migrations/009
·
back to Nationalrealestate
·
Add /health engine status board + /api/ingest-health route 33b1468 →