← back to Nationalrealestate
auto-save: 2026-07-25T13:36:34 (2 files) — db/migrations/009_source_health.sql src/jobs/self_correct.ts
2b38f61cb39381717f1db88f0c7683151954d4aa · 2026-07-25 13:36:35 -0700 · Steve Abrams
Files touched
A db/migrations/009_source_health.sqlA src/jobs/self_correct.ts
Diff
commit 2b38f61cb39381717f1db88f0c7683151954d4aa
Author: Steve Abrams <steve@designerwallcoverings.com>
Date: Sat Jul 25 13:36:35 2026 -0700
auto-save: 2026-07-25T13:36:34 (2 files) — db/migrations/009_source_health.sql src/jobs/self_correct.ts
---
db/migrations/009_source_health.sql | 31 ++++++++
src/jobs/self_correct.ts | 150 ++++++++++++++++++++++++++++++++++++
2 files changed, 181 insertions(+)
diff --git a/db/migrations/009_source_health.sql b/db/migrations/009_source_health.sql
new file mode 100644
index 0000000..aa41445
--- /dev/null
+++ b/db/migrations/009_source_health.sql
@@ -0,0 +1,31 @@
+-- M-SC1: self-correcting engine — per-job health state.
+-- No BEGIN/COMMIT here — migrate.ts wraps each file in a transaction.
+--
+-- The hourly loop recomputes each job's health from ingest_runs every tick and
+-- persists it here so it can (a) back off / quarantine broken sources instead of
+-- hammering them, (b) detect a SILENT break (run 'ok' but row count collapsed vs
+-- baseline), and (c) de-dupe alerts (only fire on a state TRANSITION, not every
+-- tick). This table is the engine's memory between ticks.
+
+CREATE TABLE source_health (
+ job TEXT PRIMARY KEY, -- loop job name (listings, brokers, ...)
+ state TEXT NOT NULL DEFAULT 'unknown', -- healthy|degraded|retrying|quarantined|unknown
+ consecutive_failures INT NOT NULL DEFAULT 0,
+ last_ok_at TIMESTAMPTZ,
+ last_failure_at TIMESTAMPTZ,
+ quarantined_at TIMESTAMPTZ,
+ baseline_upserted NUMERIC, -- median rows_upserted of recent OK runs
+ last_upserted NUMERIC, -- most recent OK run's row count
+ alerted_state TEXT, -- last state we alerted on (transition de-dupe)
+ updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+
+-- Append-only self-correction event log (quarantine / recovery / degraded / probe).
+CREATE TABLE source_events (
+ id SERIAL PRIMARY KEY,
+ job TEXT NOT NULL,
+ event TEXT NOT NULL, -- degraded|quarantined|recovered|probe|retry
+ detail TEXT,
+ at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+CREATE INDEX idx_source_events_job ON source_events (job, at DESC);
diff --git a/src/jobs/self_correct.ts b/src/jobs/self_correct.ts
new file mode 100644
index 0000000..649ba9e
--- /dev/null
+++ b/src/jobs/self_correct.ts
@@ -0,0 +1,150 @@
+/**
+ * Self-correcting engine — turns the hourly loop from a dumb timer into a
+ * health-aware controller. Each tick, for every job, we read its recent
+ * ingest_runs history and derive a HEALTH STATE, then tell the loop how to act:
+ *
+ * healthy → run on normal staleness.
+ * degraded → last run was 'ok' but rows_upserted collapsed vs baseline
+ * (a SILENT break — URL drift, selector rot). Keep running but
+ * alert; the source is "up" yet lying.
+ * retrying → last run FAILED. Not eligible until an exponential backoff
+ * window elapses (1h·2^(n-1), capped 12h) — don't hammer a
+ * flapping source, but keep trying.
+ * quarantined → N consecutive failures. Blocked for a cooldown (24h), then
+ * released as a single PROBE to test recovery. Auto-recovers
+ * (back to healthy) the moment a probe succeeds.
+ *
+ * State is persisted in source_health (memory between ticks) and every
+ * transition is logged to source_events + best-effort surfaced to CNCP. Alerts
+ * fire ONLY on a worsening/recovery TRANSITION, never every tick.
+ */
+import { query } from '../../db/pool.ts';
+
+const QUARANTINE_AFTER = Number(process.env.USRE_QUARANTINE_AFTER || 3); // consecutive failures
+const COOLDOWN_H = Number(process.env.USRE_QUARANTINE_COOLDOWN_H || 24);
+const BACKOFF_BASE_H = Number(process.env.USRE_BACKOFF_BASE_H || 1);
+const BACKOFF_CAP_H = Number(process.env.USRE_BACKOFF_CAP_H || 12);
+const DEGRADE_FRACTION = Number(process.env.USRE_DEGRADE_FRACTION || 0.3); // ok run < 30% baseline = silent break
+const DEGRADE_MIN_BASE = Number(process.env.USRE_DEGRADE_MIN_BASE || 10); // ignore tiny-count sources
+
+export type Health = {
+ job: string;
+ state: 'healthy' | 'degraded' | 'retrying' | 'quarantined' | 'unknown';
+ blockedUntilMs: number | null; // engine says: don't run before this (backoff / cooldown)
+ isProbe: boolean; // quarantined but cooldown elapsed → run once to test recovery
+ priorityBoost: number; // added to overdueBy so breakage gets fixed before routine refresh
+ reason: string;
+};
+
+function median(xs: number[]): number {
+ if (!xs.length) return 0;
+ const s = [...xs].sort((a, b) => a - b);
+ const m = Math.floor(s.length / 2);
+ return s.length % 2 ? s[m] : (s[m - 1] + s[m]) / 2;
+}
+
+async function logEvent(job: string, event: string, detail: string): Promise<void> {
+ await query(`INSERT INTO source_events (job, event, detail) VALUES ($1,$2,$3)`, [job, event, detail]);
+}
+
+/** Best-effort CNCP parking-lot card (Mac2 only; silently no-ops on Kamatera). */
+async function alertCNCP(job: string, state: string, detail: string): Promise<void> {
+ try {
+ await fetch('http://127.0.0.1:3333/api/parking-lot', {
+ method: 'POST', headers: { 'Content-Type': 'application/json' },
+ body: JSON.stringify({ project: 'usrealestate', title: `ingest source ${state}: ${job}`, note: detail }),
+ signal: AbortSignal.timeout(2000),
+ });
+ } 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 }>> {
+ 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
+ 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) }));
+}
+
+export async function assess(job: string, runSources: string[], nowMs: number): Promise<Health> {
+ const h = await history(runSources);
+ if (!h.length) {
+ await query(
+ `INSERT INTO source_health (job, state) VALUES ($1,'unknown')
+ ON CONFLICT (job) DO UPDATE SET state='unknown', updated_at=NOW()`, [job]);
+ return { job, state: 'unknown', blockedUntilMs: null, isProbe: false, priorityBoost: 0, reason: 'no runs yet' };
+ }
+
+ let consecutiveFailures = 0;
+ 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
+ const lastUpserted = lastOk?.upserted ?? 0;
+
+ // prior persisted state (for transition detection)
+ const prev = await query<{ state: string; alerted_state: string | null; quarantined_at: string | null }>(
+ `SELECT state, alerted_state, quarantined_at::text FROM source_health WHERE job=$1`, [job]);
+ const prevAlerted = prev.rows[0]?.alerted_state ?? null;
+ const prevQuarantinedAt = prev.rows[0]?.quarantined_at ? Date.parse(prev.rows[0].quarantined_at) : null;
+
+ // classify
+ let state: Health['state'];
+ let blockedUntilMs: number | null = null;
+ let isProbe = false;
+ let priorityBoost = 0;
+ let reason = '';
+ let quarantinedAtMs: number | null = prevQuarantinedAt;
+
+ if (consecutiveFailures >= QUARANTINE_AFTER) {
+ state = 'quarantined';
+ quarantinedAtMs = prevQuarantinedAt ?? (lastFail?.at ?? nowMs); // pin the start of quarantine
+ const releaseAt = quarantinedAtMs + COOLDOWN_H * 3_600_000;
+ if (nowMs >= releaseAt) { isProbe = true; priorityBoost = 5000; reason = `probe after ${COOLDOWN_H}h cooldown (${consecutiveFailures} fails)`; }
+ else { blockedUntilMs = releaseAt; reason = `quarantined ${consecutiveFailures} fails, cooldown until ${new Date(releaseAt).toISOString()}`; }
+ } else if (consecutiveFailures > 0) {
+ state = 'retrying';
+ const backoffH = Math.min(BACKOFF_CAP_H, BACKOFF_BASE_H * 2 ** (consecutiveFailures - 1));
+ const readyAt = (lastFail?.at ?? nowMs) + backoffH * 3_600_000;
+ priorityBoost = 3000;
+ if (nowMs < readyAt) { blockedUntilMs = readyAt; reason = `retry backoff ${backoffH}h (fail #${consecutiveFailures})`; }
+ else reason = `retry ready (fail #${consecutiveFailures}, waited ${backoffH}h)`;
+ } else if (baseline >= DEGRADE_MIN_BASE && lastUpserted < baseline * DEGRADE_FRACTION) {
+ state = 'degraded';
+ priorityBoost = 1000;
+ reason = `SILENT break: last ok upserted ${lastUpserted} vs baseline ${baseline} (<${DEGRADE_FRACTION * 100}%)`;
+ } else {
+ state = 'healthy';
+ quarantinedAtMs = null;
+ reason = `ok (last ${lastUpserted}, baseline ${baseline || 'n/a'})`;
+ }
+
+ await query(
+ `INSERT INTO source_health (job, state, consecutive_failures, last_ok_at, last_failure_at,
+ quarantined_at, baseline_upserted, last_upserted, updated_at)
+ VALUES ($1,$2,$3,$4,$5,$6,$7,$8,NOW())
+ ON CONFLICT (job) DO UPDATE SET state=$2, consecutive_failures=$3, last_ok_at=$4,
+ last_failure_at=$5, quarantined_at=$6, baseline_upserted=$7, last_upserted=$8, updated_at=NOW()`,
+ [job, state, consecutiveFailures,
+ lastOk ? new Date(lastOk.at) : null, lastFail ? new Date(lastFail.at) : null,
+ quarantinedAtMs ? new Date(quarantinedAtMs) : null, baseline || null, lastUpserted]);
+
+ // alert only on a real transition (worsening or recovery)
+ const bad = (s: string) => s === 'degraded' || s === 'quarantined' || s === 'retrying';
+ if (state !== prevAlerted) {
+ if (bad(state) && state !== 'retrying') { // retrying is noisy/expected; alert on degraded+quarantined
+ await logEvent(job, state, reason); await alertCNCP(job, state, reason);
+ await query(`UPDATE source_health SET alerted_state=$2 WHERE job=$1`, [job, state]);
+ } else if (state === 'healthy' && prevAlerted && bad(prevAlerted)) {
+ await logEvent(job, 'recovered', `back to healthy from ${prevAlerted}`); await alertCNCP(job, 'recovered', reason);
+ await query(`UPDATE source_health SET alerted_state='healthy' WHERE job=$1`, [job]);
+ } else {
+ await query(`UPDATE source_health SET alerted_state=$2 WHERE job=$1`, [job, state]);
+ }
+ }
+
+ return { job, state, blockedUntilMs, isProbe, priorityBoost, reason };
+}
← 8b31083 Add hourly self-balancing ingest loop + Google Places discov
·
back to Nationalrealestate
·
Self-correcting engine: health-aware loop (backoff, quaranti 7ced2c2 →