[object Object]

← back to Nationalrealestate

Self-correct: gate silent-break detection behind stableVolume flag (paging/rotating jobs have variable counts, don't false-degrade)

9e65b7731f1f1d10ea1481655f9f45969006e036 · 2026-07-26 07:52:43 -0700 · Steve Abrams

Files touched

Diff

commit 9e65b7731f1f1d10ea1481655f9f45969006e036
Author: Steve Abrams <steve@designerwallcoverings.com>
Date:   Sun Jul 26 07:52:43 2026 -0700

    Self-correct: gate silent-break detection behind stableVolume flag (paging/rotating jobs have variable counts, don't false-degrade)
---
 src/jobs/hourly_loop.ts  | 9 ++++++---
 src/jobs/self_correct.ts | 7 +++++--
 2 files changed, 11 insertions(+), 5 deletions(-)

diff --git a/src/jobs/hourly_loop.ts b/src/jobs/hourly_loop.ts
index 591e5db..40c4398 100644
--- a/src/jobs/hourly_loop.ts
+++ b/src/jobs/hourly_loop.ts
@@ -33,16 +33,19 @@ type Job = {
   minIntervalHours: number;
   // ingest_runs.source value(s) that mark this job as having run; freshest wins.
   runSources: string[];
+  // true only for FIXED-workload jobs — enables the silent-break (degrade) check.
+  // Paging/rotating jobs have variable per-run counts, so leave false for them.
+  stableVolume?: boolean;
 };
 
 // Order = tie-break priority when equally overdue (listings freshest first).
 const JOBS: Job[] = [
-  { name: 'listings',   script: 'src/ingest/listings/engine.ts',    args: ['all'], minIntervalHours: 2,  runSources: ['coldwellbanker', 'realtytexas'] },
+  { name: 'listings',   script: 'src/ingest/listings/engine.ts',    args: ['all'], minIntervalHours: 2,  runSources: ['coldwellbanker', 'realtytexas'], stableVolume: true },
   { name: 'places-seed',script: 'src/ingest/places/seed.ts',        args: [],      minIntervalHours: 2,  runSources: ['google_places'] },
   { name: 'firm-crawl', script: 'src/crawl/firm_front_page.ts',     args: [],      minIntervalHours: 4,  runSources: ['firm_crawl'] },
   { name: 'discover',   script: 'src/enrich/firm_website_discovery.ts', args: [],  minIntervalHours: 4,  runSources: ['firm_discovery'] },
   { name: 'brokers',    script: 'src/jobs/broker_rotate.ts',        args: [],      minIntervalHours: 6,  runSources: ['%_dre', '%_trec', '%_dos', '%_dbpr', '%_idfpr', '%_dcp', '%_dpr'] },
-  { name: 'commercial', script: 'src/ingest/commercial/engine.ts',  args: ['la'], minIntervalHours: 24, runSources: ['la_assessor'] },
+  { name: 'commercial', script: 'src/ingest/commercial/engine.ts',  args: ['la'], minIntervalHours: 24, runSources: ['la_assessor'], stableVolume: true },
   { name: 'parcels-sd', script: 'src/ingest/parcels/engine.ts',     args: ['sandiego-county'], minIntervalHours: 1, runSources: ['sandiego_ca'] },
 ];
 
@@ -62,7 +65,7 @@ async function main() {
   // so it gets corrected before routine refresh.
   const scored = await Promise.all(JOBS.map(async j => {
     const age = await ageHours(j);
-    const health = await assess(j.name, j.runSources, nowMs);
+    const health = await assess(j.name, j.runSources, nowMs, j.stableVolume);
     const overdueBy = age - j.minIntervalHours;
     const blocked = health.blockedUntilMs != null && nowMs < health.blockedUntilMs;
     const eligible = !blocked && (overdueBy >= 0 || health.isProbe);
diff --git a/src/jobs/self_correct.ts b/src/jobs/self_correct.ts
index c6fa850..8cc5e89 100644
--- a/src/jobs/self_correct.ts
+++ b/src/jobs/self_correct.ts
@@ -70,7 +70,7 @@ async function history(runSources: string[]): Promise<Array<{ source: string; st
   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> {
+export async function assess(job: string, runSources: string[], nowMs: number, stableVolume = false): Promise<Health> {
   const h = await history(runSources);
   if (!h.length) {
     await query(
@@ -117,7 +117,10 @@ export async function assess(job: string, runSources: string[], nowMs: number):
     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) {
+  } else if (stableVolume && baseline >= DEGRADE_MIN_BASE && lastUpserted < baseline * DEGRADE_FRACTION) {
+    // Only meaningful for FIXED-workload jobs. Paging/rotating jobs (parcels-sd,
+    // brokers, places-seed) have legitimately variable per-run counts, so the
+    // degrade check is off for them — a real break there shows as a failed run.
     state = 'degraded';
     priorityBoost = 1000;
     reason = `SILENT break: last ok upserted ${lastUpserted} vs baseline ${baseline} (<${DEGRADE_FRACTION * 100}%)`;

← 854bc7a Generalize SD adapter Poway->county-wide sweep (1.09M parcel  ·  back to Nationalrealestate  ·  CT broker upgrade: full DCP salesperson roster (eqtn-rppv), 61b8b5c →