`src/jobs/self_correct.ts`

159 lines
/**
 * 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 (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<{ 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 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, stableVolume = false): 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');
  // 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)
  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 (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}%)`;
  } 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 };
}