From 243d43abf1f6163aa00972a6aa803a41fda65f0d Mon Sep 17 00:00:00 2001 From: JSONbored <49853598+JSONbored@users.noreply.github.com> Date: Sun, 26 Jul 2026 22:49:29 -0700 Subject: [PATCH] fix(orb): stop four silent alarm gaps in relay-drain, crash handling, DLQ, and reversal signals Four related observability blind spots where the alarm was off exactly when it mattered: - Relay-drain staleness (#9128): a drain that only ever throws stamped no failure signal at all, and "never drained" read a flat -1 that could never cross the firing threshold. Counts drain failures (loopover_orb_relay_drain_consecutive_failures), ages "never drained" from process boot instead of -1 in both the seconds-since-last gauge and isOrbRelayRegistrationAlerting, and applies the same boot-aging fix to the clock-skew sibling gauge. Adds LoopoverOrbRelayDrainFailing to catch a flapping drain the existing 30m no-progress alert can't see. - PostHog vs. unhandled rejections (#9133): enableExceptionAutocapture captured a rejection but never rethrew or exited, silently downgrading a fatal crash-and-restart into a dead worker loop inside a still-"healthy" process. installSelfHostCrashHandlers (new) is now the sole, unconditional source of truth for uncaughtException/unhandledRejection, installed first thing in main(); enableExceptionAutocapture is now off so posthog-node never competes with it. - Queue/metrics blind spots during a DB incident (#9139): a throwing gauge sampler vanished entirely (no series, no counter); every queue-backlog alert silently went inactive during exactly the DB outage they exist to catch. Counts sampler failures (loopover_metrics_sampler_errors_total) and emits a -1 sentinel so absence is visible. Adds a generic up==0 rule for the always-co-started observability-bundle exporters, and an absent() rule for a dead backup exporter. Backs loopover_dlq_dead_lettered_recent with the self-host queue's own recentDeadCount (the cloud-worker audit_events source is unreachable on self-host). Writes a review_audit dead_lettered row from the self-host dead-letter path. Fixes alertmanager.yml's LoopOverTargetDown/LoopoverTargetDown capitalization mismatch and extends the metric-name-reference drift test to cover alertmanager.yml's alertname references too. - Orphaned review_targets (#9136): repoints the anomaly-alerter's reversal and DLQ signals off the dead review_targets table onto the live review_audit ledger directly (also fixing a target_id namespace mismatch that meant the old join could never match even while review_targets was live). Adds checkReviewSourceFreshness, a generalizable staleness check + alert so the next table a downstream module silently stops writing is loud, not silent. byStatus/manualRate/ stuckRetryable/failed, computeCalibration, submitter-reputation.ts, and ams-miner-cohort.ts remain review_targets-sourced -- deferred, see the PR description for the full scope. Closes #9128 Closes #9133 Closes #9139 Advances #9136 --- alertmanager/alertmanager.yml | 2 +- .../src/lib/selfhost-env-reference.ts | 4 +- prometheus/rules/alerts.yml | 102 ++++++++++ src/review/ops.ts | 147 ++++++++++++--- src/selfhost/backend-contracts.ts | 6 + src/selfhost/clock-skew.ts | 23 ++- src/selfhost/dlq-recent.ts | 17 +- src/selfhost/metrics.ts | 37 +++- src/selfhost/monitored-work.ts | 53 +++++- src/selfhost/pg-queue.ts | 70 ++++++- src/selfhost/posthog.ts | 18 +- src/selfhost/process-lifecycle.ts | 103 ++++++++++ src/selfhost/sqlite-queue.ts | 55 +++++- src/server.ts | 61 +++++- .../alerts-metric-name-references.test.ts | 44 +++++ test/unit/clock-skew.test.ts | 18 +- test/unit/dlq-recent.test.ts | 19 ++ test/unit/ops.test.ts | 173 ++++++++++++++++- test/unit/selfhost-metrics.test.ts | 60 ++++++ test/unit/selfhost-monitored-work.test.ts | 104 ++++++++-- test/unit/selfhost-pg-queue.test.ts | 100 ++++++++++ test/unit/selfhost-posthog.test.ts | 7 +- test/unit/selfhost-process-lifecycle.test.ts | 178 ++++++++++++++++++ test/unit/selfhost-sqlite-queue.test.ts | 114 +++++++++++ 24 files changed, 1414 insertions(+), 101 deletions(-) create mode 100644 src/selfhost/process-lifecycle.ts create mode 100644 test/unit/selfhost-process-lifecycle.test.ts diff --git a/alertmanager/alertmanager.yml b/alertmanager/alertmanager.yml index 2d2568cf41..cd8fb77654 100644 --- a/alertmanager/alertmanager.yml +++ b/alertmanager/alertmanager.yml @@ -172,7 +172,7 @@ receivers: # # # When the whole target is down, silence its derivative warnings (5xx, latency, queue). # - source_matchers: -# - alertname="LoopOverTargetDown" +# - alertname="LoopoverTargetDown" # target_matchers: # - severity="warning" # equal: ["job"] diff --git a/apps/loopover-ui/src/lib/selfhost-env-reference.ts b/apps/loopover-ui/src/lib/selfhost-env-reference.ts index f963f7bb77..688eb50763 100644 --- a/apps/loopover-ui/src/lib/selfhost-env-reference.ts +++ b/apps/loopover-ui/src/lib/selfhost-env-reference.ts @@ -227,7 +227,7 @@ export const SELFHOST_ENV_REFERENCE_ROWS: SelfHostEnvReferenceRow[] = [ }, { name: "GITHUB_APP_SLUG", - firstReference: "src/queue/processors.ts", + firstReference: "src/selfhost/pg-queue.ts", }, { name: "GITHUB_CACHE_TTL_SECONDS", @@ -689,7 +689,7 @@ export const SELFHOST_ENV_REFERENCE_MARKDOWN = [ "| `FOREGROUND_LIVENESS_MAX_RELEASE_PER_SWEEP` | `src/selfhost/foreground-liveness.ts` |", "| `GITHUB_APP_ID` | `src/selfhost/orb-collector.ts` |", "| `GITHUB_APP_PRIVATE_KEY` | `src/selfhost/orb-collector.ts` |", - "| `GITHUB_APP_SLUG` | `src/queue/processors.ts` |", + "| `GITHUB_APP_SLUG` | `src/selfhost/pg-queue.ts` |", "| `GITHUB_CACHE_TTL_SECONDS` | `src/server.ts` |", "| `GITHUB_INSTALLATION_CONCURRENCY_DEFER_MS` | `src/selfhost/installation-concurrency-admission.ts` |", "| `GITHUB_INSTALLATION_CONCURRENCY_ENABLED` | `src/selfhost/installation-concurrency-admission.ts` |", diff --git a/prometheus/rules/alerts.yml b/prometheus/rules/alerts.yml index e413db252b..87db6c002e 100644 --- a/prometheus/rules/alerts.yml +++ b/prometheus/rules/alerts.yml @@ -39,6 +39,49 @@ groups: description: "Prometheus has failed to scrape {{ $labels.instance }} (job {{ $labels.job }}) for over 2m. The app is unreachable or not serving /metrics." runbook: "Check `docker compose ps` and `docker compose logs loopover`. Look for a missing selfhost_listening log line or a crash-loop (restart count climbing)." + - alert: LoopoverObservabilityExporterDown + # #9139: the generic up==0 counterpart to LoopoverTargetDown above, for the OTHER scrape jobs that + # previously had no liveness rule at all -- an exporter dying showed as Grafana "No data", + # indistinguishable from a profile that was simply never enabled. + # + # Deliberately scoped to node-exporter/cadvisor/redis/observability-stack ONLY, not every job in + # prometheus.yml -- those four ship in the SAME `--profile observability` bundle as Prometheus/ + # Alertmanager themselves (docker-compose.yml's own `profiles:` list for each), so they are always + # co-started whenever this alerting stack is running at all: `up==0` here can only mean the exporter + # itself died, never "the operator hasn't enabled that add-on". postgres/qdrant/gpu/rees/browserless/ + # backup are each gated behind their OWN SEPARATE profile and legitimately read `up==0` FOREVER on an + # install that never opted into that add-on (prometheus.yml's own per-job comments document this) -- + # a blanket `up{job!="loopover"}==0` would misfire permanently on the common case of "I didn't enable + # Qdrant". Those jobs are covered instead by their own metric-VALUE rules elsewhere in this file + # (the Postgres group, LoopoverQdrantErrorRateHigh, and LoopoverBackupMissing/…Stale/the new + # absent()-based rule below), which naturally stay inactive (empty result set) when the exporter was + # never scraped at all, rather than reading a real, continuous `up==0` series. + expr: up{job=~"node-exporter|cadvisor|redis|observability-stack"} == 0 + for: 10m + labels: + severity: warning + annotations: + summary: "loopover observability exporter {{ $labels.job }}/{{ $labels.instance }} is down" + description: "Prometheus has failed to scrape {{ $labels.instance }} (job {{ $labels.job }}) for over 10m. This exporter ships in the same --profile observability bundle as Prometheus/Alertmanager, so it should always be reachable while this stack is running." + runbook: "Check `docker compose ps` for the corresponding service (node-exporter / cadvisor / redis-exporter / prometheus / alertmanager / loki / tempo / grafana / otel-collector) and `docker compose logs `." + + - alert: LoopoverMetricsSamplerFailing + # #9139: renderMetrics' own gauge/gaugeVector catch blocks previously swallowed a failing sampler + # completely silently -- no series, no counter, nothing. Every queue-backlog gauge is a live DB read + # (loopover_queue_pending, pressureSignals()'s pool.query(...) calls, …), so a Postgres outage or + # pool exhaustion made `/metrics` still return 200 (non-DB counters/gauges render fine) while every + # queue-backlog alert below silently evaluated over an empty set and went INACTIVE -- exactly the + # moment they exist to catch. Any occurrence over 15m means at least one metric is currently + # invisible to Prometheus. + expr: increase(loopover_metrics_sampler_errors_total[15m]) > 0 + for: 5m + labels: + severity: warning + annotations: + summary: "a loopover metrics sampler is failing on scrape" + description: "{{ $labels.metric }} failed to sample {{ $value | printf \"%.0f\" }} time(s) over the last 15m (sustained 5m). That gauge is either reading a -1 sentinel (loopover_metrics_sampler_errors_total's own fix) or, for a gaugeVector, emitting zero series this scrape -- either way, treat every OTHER alert reading {{ $labels.metric }} as currently blind." + runbook: "Most queue/DB gauges are live Postgres reads -- check Postgres reachability/pool exhaustion first (see the loopover-postgres group and LoopoverPostgresConnectionPressure). If {{ $labels.metric }} keeps failing after the DB recovers, check for a bug in that specific sampler." + # ── Job queue / worker health ───────────────────────────────────────────── - name: loopover-jobs rules: @@ -322,6 +365,31 @@ groups: description: "The backup profile is exposing metrics, but {{ $labels.target }} has no retained backup file after 2h." runbook: "Run `docker compose --profile backup run --rm backup sh /backup.sh` and inspect the loopover-backups volume. For Postgres, confirm DATABASE_URL is available to the backup service." + - alert: LoopoverBackupExporterMissing + # #9139: LoopoverBackupMissing above REQUIRES the exporter to be up and actively reporting a zero + # count -- exactly backwards the moment the exporter itself dies (crashes, the sidecar container + # exits, the backup volume becomes unreadable): loopover_backup_files then has ZERO SAMPLES at all + # (not a reported 0), so the `== 0` comparison evaluates over an empty set and the existing rule goes + # silently INACTIVE, even though "we can no longer see whether backups exist" is arguably WORSE than + # "we can see there are zero". absent() catches exactly that gap. + # + # DELIBERATE TRADE-OFF, read before enabling the backup profile: this also reads true, permanently, + # on ANY install that has never run `--profile backup` at all (the metric then has zero samples for + # the same structural reason -- the exporter was never scraped, not just currently down) -- unlike + # every OTHER optional-profile rule in this file, which stays silent until you opt in. That is + # intentional here, mirroring the boot-time sqliteBackupAdvisory nag ("no acknowledged backup is a + # data-loss SPOF"): an install running with genuinely zero backup visibility should be flagged, not + # quietly accepted. If you haven't enabled the backup profile yet, either enable it or remove/comment + # this rule for your deployment. + expr: absent(loopover_backup_files{target=~"postgres|sqlite"}) + for: 2h + labels: + severity: warning + annotations: + summary: "no loopover backup visibility at all -- the backup exporter is not reporting" + description: "loopover_backup_files has had zero samples for over 2h -- either the backup profile was never enabled, or the backup-exporter sidecar died/lost the backups volume." + runbook: "If you intend to run backups, start the profile: `docker compose --profile backup up -d backup-exporter backup`. If it's already enabled, check `docker compose ps backup-exporter` and `docker compose logs backup-exporter` for a crash-loop or a lost volume mount." + - alert: LoopoverBackupStale # Default backup loop is daily. 26h allows one missed scrape/restart window before warning. expr: | @@ -415,6 +483,22 @@ groups: description: "Either the consecutive registration-failure streak has reached {{ $value | printf \"%.0f\" }}, or the pull-mode drain loop hasn't completed in over 30m. A lone registration timeout alone would not trigger this." runbook: "Check loopover_orb_relay_register_total{result=\"failed\"} by mode for the failure pattern, and confirm ORB_BROKER_URL / ORB_ENROLLMENT_SECRET are still valid. If pull mode, verify the drain loop itself isn't crash-looping (selfhost_orb_relay_register_failed logs at level=error)." + - alert: LoopoverOrbRelayDrainFailing + # #9128: the rule above catches a drain loop gone TOTALLY quiet (nothing completed in 30m). This + # catches the narrower gap it can't see: a FLAPPING pull-mode drain (a 4xx, a broker-side schema + # change, an ack-payload rejection on a meaningful fraction of ticks) that still succeeds often + # enough to keep resetting loopover_orb_relay_drain_seconds_since_last, so the 30m no-progress + # window above never trips even though a real fraction of every drain tick is failing. Any + # occurrence over 15m is worth a look, same absolute-increase style as LoopoverDeadLetterJobsGrowing. + expr: increase(loopover_orb_relay_drains_total{result="failed"}[15m]) > 0 + for: 5m + labels: + severity: warning + annotations: + summary: "loopover orb relay drain is failing on a meaningful fraction of ticks" + description: "{{ $value | printf \"%.0f\" }} drain tick(s) threw over the last 15m (sustained 5m). loopover_orb_relay_drain_consecutive_failures shows the current unbroken streak." + runbook: "Tail logs around orb_relay_drain for the thrown error (4xx from the broker, a schema change, an ack-payload rejection). If loopover_orb_relay_drain_consecutive_failures keeps climbing without resetting, this will also eventually trip LoopoverOrbRelayRegistrationStuck once no drain has succeeded for 30m." + # ── HTTP serving health (status label + duration histogram, both live in src/server.ts) ─ # loopover_http_requests_total carries a status="2xx|3xx|4xx|5xx" label (seeded at zero per # class so every series exists from boot), and loopover_http_request_duration_seconds is a @@ -539,6 +623,24 @@ groups: description: "{{ $value | printf \"%.0f\" }} {{ $labels.kind }} detection(s) over the last 2h for {{ $labels.repo }} (sustained 5m). Check the ops_anomaly structured log for the full detail line." runbook: "Tail logs for level=error event=ops_anomaly repo={{ $labels.repo }} for the human-readable anomaly text. A review_burst or review_failure_burst usually means a stuck-CI finalize loop or a sweep retry storm -- see #orb-ci-stuck-repeat / #review-burst-blind-spot." + - alert: LoopoverReviewSourceStale + # #9136: the generalizable fix -- review_targets had NO live writer anywhere for months after the + # 2026-06-22 convergence cutover before anyone noticed, silently zeroing the Discord anomaly alerter + # and (slowly) the reputation signal. loopover_review_source_fresh (checkReviewSourceFreshness) + # checks whether each tracked table still has a row inside ITS OWN consumer's window -- this fires + # the moment one goes stale instead of requiring a human to notice a suspiciously-quiet dashboard. + # 6h tolerates review_audit's occasional quiet stretch on a low-traffic repo without paging on noise; + # review_targets is EXPECTED to already read stale (it has no live writer at all -- see #9136's own + # scope decision) and reads as a routine, informational warning until it's repointed or restored. + expr: loopover_review_source_fresh == 0 + for: 6h + labels: + severity: warning + annotations: + summary: "loopover review source table {{ $labels.table }} has gone stale" + description: "{{ $labels.table }} has had no row inside its own {{ $labels.window_days }}-day consumer window for over 6h (sustained). Any reader still treating it as live is silently reading empty/zero results." + runbook: "For review_audit: check that parity-wire.ts / outcomes-wire.ts are still writing (gate_decision / pr_outcome / reversal_* rows) -- if this fires for review_audit, the anomaly alerter has likely gone silently inert again, the exact #9136 shape. For review_targets: this is a KNOWN, expected-stale table (no live writer) -- see #9136 for the tracked remainder (submitter-reputation.ts, ams-miner-cohort.ts) before treating this as a new incident." + # ── Host clock sync (#3811) ─────────────────────────────────────────────── - name: loopover-system-health rules: diff --git a/src/review/ops.ts b/src/review/ops.ts index 706e391645..0b279b804a 100644 --- a/src/review/ops.ts +++ b/src/review/ops.ts @@ -185,7 +185,27 @@ const NON_TERMINAL = new Set(["queued", "reviewing", "error_retryable"]); const ANOMALY_WINDOW = "-7 days"; // DLQ spike = a RECENT burst of dead-letters whose targets HAVEN'T recovered. const DLQ_WINDOW = "-6 hours"; -const DLQ_RECOVERED_STATUSES = "('merged', 'closed', 'commented', 'manual', 'ignored')"; + +// #9136: `review_targets` has NO live writer anywhere in this codebase (the 2026-06-22 convergence cutover +// orphaned it -- see src/db/repo-identity-rename.ts / src/review/public-stats.ts's own comments) — every +// query below that used to join or read it saw a permanently-empty table, silently zeroing the reversals +// and DLQ anomaly signals. `review_audit` (migration 0049) IS live: parity-wire.ts writes 'gate_decision' +// rows on every finalized verdict, and outcomes-wire.ts writes 'pr_outcome'/'reversal_reverted'/ +// 'reversal_reopened' rows on the realized outcome. Repointing reversals + DLQ (the two signals detectAnomalies +// can ACT on directly) onto review_audit alone -- parsing repo/number out of its own target_id instead of +// joining review_targets for them -- also fixes a SEPARATE bug the join had: review_audit.target_id is +// `owner/repo#123` (reviewAuditTargetId, outcomes-wire.ts) while review_targets.id is `project:kind:owner/repo#123` +// (rowId, above) -- a different namespace the join could never actually match, even while review_targets was +// still live. byStatus/verdictRows/failedRows below are UNCHANGED (still review_targets-sourced, and so still +// silently zero) -- deferred; see the PR description for the full scope decision and #9136 for the tracked +// remainder (manualRate/stuckRetryable/failed/calibration bins, submitter-reputation.ts, ams-miner-cohort.ts). +function parseReviewAuditTargetId(targetId: string): { repo: string; number: number } | null { + const hashIndex = targetId.lastIndexOf("#"); + if (hashIndex <= 0) return null; + const repo = targetId.slice(0, hashIndex); + const number = Number(targetId.slice(hashIndex + 1)); + return Number.isInteger(number) && number > 0 ? { repo, number } : null; +} // ── Injected runtime-gate deps (config invariants + kill-switch/circuit-breaker flags + AI errors) ─── @@ -223,32 +243,43 @@ export async function computeAgentHealth(env: Env, config: OpsAgentConfig, deps: WHERE project = ? AND status = 'error' AND updated_at > datetime('now', ?) ORDER BY updated_at DESC LIMIT ?`, ).bind(slug, ANOMALY_WINDOW, LIST_CAP).all<{ number: number; repo: string; verdict: string | null; last_error: string | null }>(), - // Recent human reversals of a bot auto-action. A reopened bot-close the gate SUBSEQUENTLY - // RE-TERMINALIZED (terminal_at AFTER the reopen) is excluded — the gate re-reviewed and ACTED on it. + // #9136: repointed off review_targets (see this file's own header comment above). Recent human reversals + // of a bot auto-action, read directly from review_audit's own reversal_* rows -- repo/number parsed from + // target_id, no join. A reopened bot-close the gate SUBSEQUENTLY re-acted on (a LATER gate_decision row + // for the same target_id) is excluded, mirroring the review_targets.terminal_at check this replaces. storage(env).prepare( - `SELECT t.number AS number, t.repo AS repo, t.status AS status, a.event_type AS event_type - FROM review_audit a JOIN review_targets t ON t.id = a.target_id + `SELECT a.target_id AS target_id, a.event_type AS event_type + FROM review_audit a WHERE a.project = ? AND a.event_type IN ('reversal_reverted', 'reversal_reopened') AND a.created_at > datetime('now', ?) - AND NOT (a.event_type = 'reversal_reopened' AND t.terminal_at IS NOT NULL AND t.terminal_at > a.created_at) + AND NOT ( + a.event_type = 'reversal_reopened' AND EXISTS ( + SELECT 1 FROM review_audit g + WHERE g.project = a.project AND g.target_id = a.target_id + AND g.event_type = 'gate_decision' AND g.created_at > a.created_at + ) + ) ORDER BY a.created_at DESC LIMIT ?`, - ).bind(slug, ANOMALY_WINDOW, LIST_CAP).all<{ number: number; repo: string; status: string; event_type: string }>(), - // Auto-actions in the SAME 7d window — the rate denominator. - storage(env).prepare(`SELECT COUNT(*) AS n FROM review_targets WHERE project = ? AND status IN ('merged', 'closed') AND terminal_at > datetime('now', ?)`).bind(slug, ANOMALY_WINDOW).first<{ n: number }>(), - // RECENT, UNRECOVERED dead-letter events, WITH the PR. + ).bind(slug, ANOMALY_WINDOW, LIST_CAP).all<{ target_id: string; event_type: string }>(), + // #9136: auto-actions in the SAME 7d window — the reversalRate denominator. Repointed onto review_audit's + // own gate_decision rows (decision IN merge/close) instead of review_targets' terminal-status count. storage(env).prepare( - `SELECT t.number AS number, t.repo AS repo, t.verdict AS verdict, t.last_error AS last_error - FROM review_audit a JOIN review_targets t ON t.id = a.target_id - WHERE a.project = ? AND a.event_type = 'dead_lettered' AND a.created_at > datetime('now', ?) - AND t.status NOT IN ${DLQ_RECOVERED_STATUSES} - ORDER BY a.created_at DESC LIMIT ?`, - ).bind(slug, DLQ_WINDOW, LIST_CAP).all<{ number: number; repo: string; verdict: string | null; last_error: string | null }>(), - // TRUE count of recent UNRECOVERED dead-letters — a separate COUNT(*) so a storm of >LIST_CAP isn't - // undercounted, and so recovered targets never inflate it. + `SELECT COUNT(*) AS n FROM review_audit WHERE project = ? AND event_type = 'gate_decision' AND decision IN ('merge', 'close') AND created_at > datetime('now', ?)`, + ).bind(slug, ANOMALY_WINDOW).first<{ n: number }>(), + // #9136: repointed off review_targets. RECENT dead-letter events, read directly from review_audit (the + // event type pg-queue.ts's self-host dead-letter path now writes, #9139) -- repo/number/lastError parsed + // from target_id + summary, no join. Unlike the review_targets-joined query this replaces, this does NOT + // exclude a target that later recovered (review_targets' terminal-status recheck has no live equivalent + // here) -- a conservative direction change: it can only make the alert fire on a real dead-letter more + // readily, never mask one. + storage(env).prepare( + `SELECT target_id, summary FROM review_audit + WHERE project = ? AND event_type = 'dead_lettered' AND created_at > datetime('now', ?) + ORDER BY created_at DESC LIMIT ?`, + ).bind(slug, DLQ_WINDOW, LIST_CAP).all<{ target_id: string; summary: string | null }>(), + // TRUE count of recent dead-letters — a separate COUNT(*) so a storm of >LIST_CAP isn't undercounted. storage(env).prepare( - `SELECT COUNT(*) AS n FROM review_audit a JOIN review_targets t ON t.id = a.target_id - WHERE a.project = ? AND a.event_type = 'dead_lettered' AND a.created_at > datetime('now', ?) - AND t.status NOT IN ${DLQ_RECOVERED_STATUSES}`, + `SELECT COUNT(*) AS n FROM review_audit WHERE project = ? AND event_type = 'dead_lettered' AND created_at > datetime('now', ?)`, ).bind(slug, DLQ_WINDOW).first<{ n: number }>(), ]); const byStatus: Record = {}; @@ -259,8 +290,27 @@ export async function computeAgentHealth(env: Env, config: OpsAgentConfig, deps: const nonTerminal = Object.entries(byStatus).reduce((sum, [s, n]) => (NON_TERMINAL.has(s) ? sum + n : sum), 0); const recentAutoActions = recentActionsRow?.n ?? 0; const failedTargets: FailedTarget[] = (failedRows.results ?? []).map((r) => ({ number: r.number, repo: r.repo, verdict: r.verdict, lastError: r.last_error })); - const reversedTargets: ReversedTarget[] = (reversedRows.results ?? []).map((r) => ({ number: r.number, repo: r.repo, status: r.status, eventType: r.event_type })); - const dlqTargets: FailedTarget[] = (dlqRows.results ?? []).map((r) => ({ number: r.number, repo: r.repo, verdict: r.verdict, lastError: r.last_error })); + // #9136: repo/number parsed from review_audit's own target_id (no review_targets join -- see this file's + // header comment). A target_id this malformed to parse is skipped (filtered out) rather than crashing the + // whole snapshot over one bad row -- defensive only; every writer of this column (outcomes-wire.ts, + // pg-queue.ts/sqlite-queue.ts's dead-letter path) always stamps the well-formed `owner/repo#n` shape. + const reversedTargets: ReversedTarget[] = (reversedRows.results ?? []) + .map((r) => { + const parsed = parseReviewAuditTargetId(r.target_id); + if (!parsed) return null; + // No live review_targets status to read anymore -- derived purely from the reversal's OWN event_type, + // which already tells us what the PR's terminal status must have been: a reverted action was a MERGE + // (only a merge can be "reverted" by a separate revert PR); a reopened action was a CLOSE. + const status = r.event_type === "reversal_reverted" ? "merged" : "closed"; + return { number: parsed.number, repo: parsed.repo, status, eventType: r.event_type }; + }) + .filter((t): t is ReversedTarget => t !== null); + const dlqTargets: FailedTarget[] = (dlqRows.results ?? []) + .map((r): FailedTarget | null => { + const parsed = parseReviewAuditTargetId(r.target_id); + return parsed ? { number: parsed.number, repo: parsed.repo, verdict: null, lastError: r.summary } : null; + }) + .filter((t): t is FailedTarget => t !== null); const reversals = reversedTargets.length; return { byStatus, @@ -478,3 +528,54 @@ export async function handleInternalCalibration(request: Request, env: Env, conf if (denied) return denied; return Response.json({ project: config.slug, calibration: await computeCalibration(env, config) }); } + +// ── Source-table freshness (#9136, the generalizable fix) ─────────────────────────────────────────── + +/** One source table's freshness verdict: does it have a row inside ITS OWN consumer's window, not just + * "any row ever". `fresh: false` on a read error too (a missing/dropped table is itself a staleness + * signal — fail CLOSED, never masquerade a broken read as "everything's fine"). */ +export interface ReviewSourceFreshnessCheck { + table: string; + /** The consuming window (days) a stale table would fall outside of -- the same window value the real + * consumer(s) use, so this check fails at EXACTLY the moment those consumers' own queries would start + * reading an empty result. */ + windowDays: number; + fresh: boolean; +} + +/** Every table an ops/reputation module treats as a LIVE, windowed source, with that consumer's own + * window. `review_targets` was silently orphaned by the 2026-06-22 convergence cutover (no live writer + * anywhere — see src/db/repo-identity-rename.ts / src/review/public-stats.ts's own comments) and nobody + * noticed for months; this is the generalizable fix so the NEXT orphaning is loud, not silent. + * - review_targets: submitter-reputation.ts's own REPUTATION_WINDOW_DAYS (90) — this table's newest row + * is frozen at the 2026-06-22 cutover, so it will fall outside this window (and read permanently stale + * here) around 2026-09-20 if nothing else changes; see #9136 for the full scope decision on that module. + * - review_audit: this module's own ANOMALY_WINDOW (7 days) — IS live today (parity-wire.ts writes + * 'gate_decision' rows, outcomes-wire.ts writes the outcome/reversal types), so this should read fresh + * in steady state. If both writers ever stop, this is what catches the alerter going silently inert + * again, exactly the shape review_targets' own orphaning took. + */ +const REVIEW_SOURCE_FRESHNESS_SOURCES: ReadonlyArray<{ table: string; timestampColumn: string; windowDays: number }> = [ + { table: "review_targets", timestampColumn: "terminal_at", windowDays: 90 }, + { table: "review_audit", timestampColumn: "created_at", windowDays: 7 }, +]; + +/** Check every tracked source table for a row inside its own consuming window. Read-only; a per-table + * query failure degrades that ONE table to `fresh: false` (fail closed) rather than throwing the whole + * check — a broken table is exactly the condition this exists to surface, not to hide behind an + * unhandled rejection. */ +export async function checkReviewSourceFreshness(env: Env): Promise { + return Promise.all( + REVIEW_SOURCE_FRESHNESS_SOURCES.map(async ({ table, timestampColumn, windowDays }): Promise => { + try { + const row = await storage(env) + .prepare(`SELECT 1 AS x FROM ${table} WHERE ${timestampColumn} IS NOT NULL AND ${timestampColumn} > datetime('now', ?) LIMIT 1`) + .bind(`-${windowDays} days`) + .first<{ x: number }>(); + return { table, windowDays, fresh: row != null }; + } catch { + return { table, windowDays, fresh: false }; + } + }), + ); +} diff --git a/src/selfhost/backend-contracts.ts b/src/selfhost/backend-contracts.ts index a108b6d1c7..89db5a8198 100644 --- a/src/selfhost/backend-contracts.ts +++ b/src/selfhost/backend-contracts.ts @@ -37,6 +37,12 @@ export interface DurableQueue { drain(): Promise; size(): Promise; deadCount(): Promise; + /** #9139: jobs dead-lettered within the trailing `windowMs` -- a RATE-style window over `dead_at`, unlike + * deadCount()'s standing depth. Backs loopover_dlq_dead_lettered_recent on self-host, whose cloud-worker + * source (audit_events' `github_app.dlq_dead_lettered`, written only by the Cloudflare `queue()` handler's + * processDlqBatch) is structurally unreachable here -- server.ts calls `worker.fetch`/`worker.scheduled` + * but never `worker.queue`. */ + recentDeadCount(windowMs: number): Promise; /** Jobs currently claimed and mid-flight (status='processing') -- distinct from size(), which also * includes still-pending work. See #selfhost-queue-liveness's own observability additions. */ processingCount(): Promise; diff --git a/src/selfhost/clock-skew.ts b/src/selfhost/clock-skew.ts index 03f9385788..c92afba66c 100644 --- a/src/selfhost/clock-skew.ts +++ b/src/selfhost/clock-skew.ts @@ -12,6 +12,13 @@ let lastSkewSeconds = 0; // signal below so an old sample can't silently look current if token-mint activity — the only thing that // refreshes lastSkewSeconds — stalls (#7000). let lastSkewSampleAtMs: number | null = null; +// #9128 (sibling audit): module-load time, the fallback "since" reference for clockSkewSampleAgeSeconds +// when no sample has ever landed -- mirrors the SAME fix applied to the relay-drain "never happened" gauge, +// which read a flat -1 forever (never ageing into a threshold no matter how long the underlying condition +// persisted). No alert currently reads this gauge (confirmed: zero references in prometheus/rules/alerts.yml +// and zero dashboard panels), so today this closes a latent hole rather than an active one -- but the SAME +// shape would silently defeat any FUTURE alert added on this metric, exactly as it did for relay-drain. +let moduleLoadedAtMs = Date.now(); /** * Update the last-observed clock-skew sample from a GitHub response's `Date` header. Positive means @@ -35,17 +42,21 @@ export function clockSkewSecondsSample(): number { } /** - * Seconds since the last successful clock-skew sample, or a -1 sentinel when none has landed yet — the same - * "never sampled" convention as {@link d1DatabaseSizeBytesSample} (src/selfhost/d1-size-probe.ts). Lets an - * operator tell a fresh reading apart from an old sample the token-mint path simply hasn't refreshed (#7000). + * Seconds since the last successful clock-skew sample -- or, before any sample has ever landed, seconds + * since this module was loaded (#9128: previously a flat -1 sentinel that never aged, the same "never + * happened" shape that let the relay-drain staleness alarm go permanently quiet). No alert reads this gauge + * today, so this only closes a latent hole, but it means a future threshold on it behaves correctly from + * the start rather than needing its own follow-up fix. */ export function clockSkewSampleAgeSeconds(): number { - if (lastSkewSampleAtMs === null) return -1; - return (Date.now() - lastSkewSampleAtMs) / 1000; + const sinceMs = lastSkewSampleAtMs ?? moduleLoadedAtMs; + return (Date.now() - sinceMs) / 1000; } -/** Test-only: reset the module-level sample between tests. */ +/** Test-only: reset the module-level sample between tests, including the #9128 boot-time reference (so a + * test can control "time since load" precisely, matching resetPostHogForTest-style module resets elsewhere). */ export function resetClockSkewForTest(): void { lastSkewSeconds = 0; lastSkewSampleAtMs = null; + moduleLoadedAtMs = Date.now(); } diff --git a/src/selfhost/dlq-recent.ts b/src/selfhost/dlq-recent.ts index f4e34a2f78..48ccacdc92 100644 --- a/src/selfhost/dlq-recent.ts +++ b/src/selfhost/dlq-recent.ts @@ -11,10 +11,23 @@ export function isoNowMinus(windowMs: number, now: number = Date.now()): string return new Date(now - windowMs).toISOString(); } +/** The subset of DurableQueue (backend-contracts.ts) this gauge needs -- avoids importing the full interface + * (and its Queue/D1-shaped siblings) into a module whose only job is one scrape-time count. */ +export type RecentDeadCountSource = { recentDeadCount(windowMs: number): Promise }; + /** Scrape-time sample of DLQ dead-letters within the trailing window. Swallows a query error so a transient DB - * hiccup degrades the sample to 0 rather than rejecting and breaking the whole `/metrics` scrape. */ -export async function sampleRecentDeadLetters(env: Env, now: number = Date.now()): Promise { + * hiccup degrades the sample to 0 rather than rejecting and breaking the whole `/metrics` scrape. + * + * `selfHostQueue` (#9139), when provided, is used EXCLUSIVELY instead of the `audit_events`-based cloud-worker + * path: `github_app.dlq_dead_lettered` is written only by `processDlqBatch` (src/queue/dlq.ts), reached only + * from the Cloudflare `queue()` handler (src/index.ts) -- server.ts (the self-host runtime) calls + * `worker.fetch`/`worker.scheduled` but never `worker.queue`, so that source is a structural, permanent 0 on + * every self-hosted instance regardless of how many jobs actually dead-letter. The self-host queue backends + * (sqlite-queue.ts / pg-queue.ts) dead-letter via `UPDATE ... SET status='dead'` on their own jobs table + * instead (see pg-queue.ts's dead-letter path) -- `recentDeadCount` reads that directly. */ +export async function sampleRecentDeadLetters(env: Env, now: number = Date.now(), selfHostQueue?: RecentDeadCountSource): Promise { try { + if (selfHostQueue) return await selfHostQueue.recentDeadCount(DLQ_RECENT_WINDOW_MS); return await countRecentDeadLetters(env, isoNowMinus(DLQ_RECENT_WINDOW_MS, now)); } catch { return 0; diff --git a/src/selfhost/metrics.ts b/src/selfhost/metrics.ts index fca3d29bf5..76bd49551d 100644 --- a/src/selfhost/metrics.ts +++ b/src/selfhost/metrics.ts @@ -107,7 +107,8 @@ export const DEFAULT_METRIC_META: readonly (readonly [string, MetricMeta])[] = [ ["loopover_orb_relay_drains_total", { help: "Orb relay drain outcomes.", type: "counter" }], ["loopover_orb_relay_drain_skipped_total", { help: "Pull-mode orb relay drain ticks skipped because the previous tick was still in flight.", type: "counter" }], ["loopover_orb_relay_register_consecutive_failures", { help: "Current consecutive orb relay registration failure streak, reset to 0 on any success.", type: "gauge" }], - ["loopover_orb_relay_drain_seconds_since_last", { help: "Seconds since the pull-mode orb relay drain loop last completed successfully, or -1 if never (or in push mode).", type: "gauge" }], + ["loopover_orb_relay_drain_consecutive_failures", { help: "Current consecutive pull-mode orb relay drain failure streak (drain threw), reset to 0 on any completed drain tick; always 0 in push mode.", type: "gauge" }], + ["loopover_orb_relay_drain_seconds_since_last", { help: "Seconds since the pull-mode orb relay drain loop last completed successfully, or since process boot if it never has; -1 in push mode, where there is no drain loop.", type: "gauge" }], ["loopover_orb_webhook_total", { help: "Orb webhook outcomes.", type: "counter" }], ["loopover_orb_config_push_received_total", { help: "Config-push relay rows received and logged by the pull-drain loop (#7523).", type: "counter" }], ["loopover_ai_requests_total", { help: "AI provider request outcomes.", type: "counter" }], @@ -178,6 +179,8 @@ export const DEFAULT_METRIC_META: readonly (readonly [string, MetricMeta])[] = [ ["loopover_review_memory_suppressed_total", { help: "Review-memory entries suppressed from surfacing, by repo.", type: "counter" }], ["loopover_rees_enrich_requests_total", { help: "REES /v1/enrich call outcomes, by status (ok/empty/http_error/timeout/exception/skipped_auth_rejected).", type: "counter" }], ["loopover_rees_enrich_request_duration_seconds", { help: "REES /v1/enrich call duration in seconds, for calls that were actually attempted (excludes the auth-rejected circuit-breaker skip).", type: "histogram" }], + ["loopover_metrics_sampler_errors_total", { help: "Scrape-time gauge sampler failures, by metric name -- a failing sampler previously emitted no series at all, silently. Any occurrence means that metric's value is currently invisible to Prometheus this scrape (see the sentinel gauges' own -1-on-failure convention).", type: "counter" }], + ["loopover_review_source_fresh", { help: "1 when a review/ops/reputation source table has a row inside its own consumer's window, 0 when stale -- labeled by table and window_days. review_targets was silently orphaned by the 2026-06-22 convergence cutover for months before anyone noticed; this makes the next such orphaning loud instead.", type: "gauge" }], ]; const metricMeta = new Map(DEFAULT_METRIC_META); @@ -315,21 +318,29 @@ export function observe(name: string, value: number, labels?: Labels, buckets: n h.count += 1; } -/** Render the registry in Prometheus text exposition format. */ +/** Render the registry in Prometheus text exposition format. Counters render LAST (#9139): a gauge/ + * gaugeVector sampler failure below increments loopover_metrics_sampler_errors_total AS PART OF the same + * render call, so counters must be read after those loops run, not before -- otherwise a THIS-scrape + * failure would only ever show up starting with the NEXT scrape's output, one full interval late. */ export async function renderMetrics(): Promise { const lines: string[] = []; const emittedMeta = new Set(); - for (const [k, v] of counters) { - pushMetricMeta(lines, emittedMeta, metricNameFromSeriesKey(k)); - lines.push(`${k} ${v}`); - } for (const [name, sample] of gauges) { try { const value = await sample(); pushMetricMeta(lines, emittedMeta, name); lines.push(`${name} ${value}`); } catch { - /* a failing sampler must not break the scrape */ + // #9139: a throwing sampler previously emitted NO series at all -- indistinguishable in Grafana from + // "idle"/"no data", and every alert rule reading it simply evaluated over an empty set (INACTIVE, not + // FIRING) at exactly the DB-incident moment it exists to catch (loopover_queue_pending and friends are + // all live DB reads -- see server.ts). Counting the failure AND still emitting a -1 sentinel series -- + // the SAME "impossible for a healthy gauge, so its absence is itself visible" convention + // loopover_clock_skew_sample_age_seconds / loopover_d1_database_size_bytes already use for "never + // sampled" -- turns a silently-empty scrape into an actionable, alertable signal. + incr("loopover_metrics_sampler_errors_total", { metric: name }); + pushMetricMeta(lines, emittedMeta, name); + lines.push(`${name} -1`); } } for (const [name, sample] of gaugeVectors) { @@ -343,7 +354,12 @@ export async function renderMetrics(): Promise { lines.push(`${seriesKey(name, publicLabelsForMetric(name, labels))} ${value}`); } } catch { - /* a failing sampler must not break the scrape */ + // #9139: same failure-visibility fix as the plain-gauge loop above, but a gaugeVector's label SET is + // unknown at failure time (that's the whole point of it), so there's no single value to sentinel -- + // counting the failure is still the actionable half; the metric name simply emits zero series this + // scrape, same as its own pre-existing "legitimately empty" case just above. + incr("loopover_metrics_sampler_errors_total", { metric: name }); + pushMetricMeta(lines, emittedMeta, name); } } for (const h of histograms.values()) { @@ -356,6 +372,11 @@ export async function renderMetrics(): Promise { lines.push(`${seriesKey(`${h.name}_sum`, h.labels)} ${h.sum}`); lines.push(`${seriesKey(`${h.name}_count`, h.labels)} ${h.count}`); } + // Rendered last (#9139) -- see this function's own header comment. + for (const [k, v] of counters) { + pushMetricMeta(lines, emittedMeta, metricNameFromSeriesKey(k)); + lines.push(`${k} ${v}`); + } return `${lines.join("\n")}\n`; } diff --git a/src/selfhost/monitored-work.ts b/src/selfhost/monitored-work.ts index 4a0dbcd6db..4af75bca6e 100644 --- a/src/selfhost/monitored-work.ts +++ b/src/selfhost/monitored-work.ts @@ -21,6 +21,13 @@ export type OrbRelayDrainState = { // isOrbRelayRegistrationAlerting so a registration failure streak below the alert threshold can still // be judged against real evidence the relay connection is (or isn't) making progress. lastDrainAtMs: number | null; + // #9128: consecutive drain THROWS (a 4xx, a broker schema change, an ack-payload rejection) -- distinct + // from registration's OWN consecutiveFailures (OrbRelayRegistrationState), which only ever tracks the + // registration call, never the drain loop. Reset to 0 on any drain call that completes without throwing + // (mirrors lastDrainAtMs's own "even a zero-event poll is progress" reset). Optional (not set by every + // existing state literal, e.g. server.ts's `{ pendingAck: [], lastDrainAtMs: null }`) -- read as 0 when + // absent, rather than forcing every construction site to know about this field. + consecutiveFailures?: number; }; type OrbRelayEnv = { @@ -95,12 +102,29 @@ export async function drainOrbRelayWithMonitor(args: { "orb-relay-drain", { jobType: "orb-relay-drain", pendingAckCount: args.state.pendingAck.length }, async () => { - const events = await args.drain(args.relayEnv, args.state.pendingAck); + let events: OrbRelayEvent[]; + try { + events = await args.drain(args.relayEnv, args.state.pendingAck); + } catch (error) { + // #9128: a drain that throws on EVERY tick (a 4xx, a broker-side schema change, an ack-payload + // rejection) previously left NO trace behind -- lastDrainAtMs correctly stays untouched (no + // progress to claim), but nothing ELSE moved either: loopover_orb_relay_drains_total only ever + // incremented on the success path, so there was no failure counter to notice, and + // isOrbRelayRegistrationAlerting's own consecutiveFailures tracks REGISTRATION, not drain -- a + // healthy registration with a permanently-broken drain had no failure signal anywhere. Count the + // throw (both a result="failed" series on the existing counter and a dedicated streak, mirroring + // OrbRelayRegistrationState's own consecutiveFailures) before rethrowing, unchanged, to the + // reentrancy-guarded caller. + args.state.consecutiveFailures = (args.state.consecutiveFailures ?? 0) + 1; + incr("loopover_orb_relay_drains_total", { result: "failed" }); + throw error; + } args.state.pendingAck = []; // A successful round-trip (even zero events) proves the broker link itself is alive -- stamped // BEFORE the per-event enqueue loop so a downstream enqueue failure still counts as drain progress // (the relay connection, not the local queue, is what registration-alerting cares about). args.state.lastDrainAtMs = args.nowMs ?? Date.now(); + args.state.consecutiveFailures = 0; incr("loopover_orb_relay_drains_total", { result: events.length > 0 ? "events" : "empty", }); @@ -215,21 +239,26 @@ export const ORB_RELAY_DRAIN_NO_PROGRESS_WINDOW_MS = 30 * 60_000; /** Pull-mode registration alert gate (#selfhost-runtime-drift follow-up): a lone registration timeout is * routine degraded telemetry, NOT an error, as long as the drain loop is still making progress -- so this * only reports "actually stuck" (as opposed to "one hiccup") when EITHER the failure streak has crossed - * {@link ORB_RELAY_REGISTER_UNHEALTHY_FAILURE_STREAK}, OR a KNOWN prior drain has gone stale for over - * {@link ORB_RELAY_DRAIN_NO_PROGRESS_WINDOW_MS}. `drainLastAtMs` is `null` when there is no drain-progress - * evidence to judge yet (push mode has no drain loop at all; a pull-mode container may simply not have - * reached its first drain tick) -- treated as "insufficient signal to escalate on this basis", not as - * "stuck", so a lone registration hiccup at boot can't alert before the drain loop has had a chance to - * prove itself either way. */ + * {@link ORB_RELAY_REGISTER_UNHEALTHY_FAILURE_STREAK}, OR the last known drain progress (a completed + * drain tick, or -- #9128 -- process boot if there has never been one) has gone stale for over + * {@link ORB_RELAY_DRAIN_NO_PROGRESS_WINDOW_MS}. `drainLastAtMs` is `null` when there is no COMPLETED + * drain tick yet -- push mode has no drain loop at all (bootAtMs is unused in that branch, since the + * streak check above always applies for push), and a pull-mode container may simply not have reached its + * first drain tick, OR every tick since boot has thrown (#9128's own trigger: a broker-side schema change + * or ack-payload rejection breaks every single attempt, so lastDrainAtMs never once gets stamped). Either + * way this now ages from {@link bootAtMs} instead of returning `false` forever -- a lone hiccup at boot + * still can't alert before the no-progress window has elapsed, but a container that NEVER completes a + * drain tick for the whole window is exactly "actually stuck", not "insufficient signal". */ export function isOrbRelayRegistrationAlerting(args: { consecutiveFailures: number; drainLastAtMs: number | null; + bootAtMs: number; nowMs?: number; }): boolean { if (args.consecutiveFailures >= ORB_RELAY_REGISTER_UNHEALTHY_FAILURE_STREAK) return true; - if (args.drainLastAtMs === null) return false; const nowMs = args.nowMs ?? Date.now(); - return nowMs - args.drainLastAtMs > ORB_RELAY_DRAIN_NO_PROGRESS_WINDOW_MS; + const lastProgressAtMs = args.drainLastAtMs ?? args.bootAtMs; + return nowMs - lastProgressAtMs > ORB_RELAY_DRAIN_NO_PROGRESS_WINDOW_MS; } /** Recurring wrapper around the retryable relay-registration attempt (#selfhost-runtime-drift): a bare @@ -238,12 +267,15 @@ export function isOrbRelayRegistrationAlerting(args: { * network request (`registered` / `failed`) — `already_registered` / `backoff` / `skipped` are silent no-ops * so a healthy or intentionally-idle container does not spam logs/PostHog every tick. `drainState` is the * pull-mode drain loop's shared state (omitted/undefined in push mode, where there is no drain loop) -- its - * `lastDrainAtMs` feeds the no-progress-window half of {@link isOrbRelayRegistrationAlerting}. */ + * `lastDrainAtMs` feeds the no-progress-window half of {@link isOrbRelayRegistrationAlerting}. `bootAtMs` + * (#9128) is this process's own start time -- the fallback "since" reference isOrbRelayRegistrationAlerting + * ages a never-completed drain from, rather than treating "no drain evidence yet" as permanently benign. */ export async function registerOrbRelayWithMonitor(args: { env: OrbRelayRegisterEnv; state: OrbRelayRegistrationState; register: (env: OrbRelayRegisterEnv, state: OrbRelayRegistrationState) => Promise; drainState?: OrbRelayDrainState; + bootAtMs: number; log?: (line: string) => void; nowMs?: number; }): Promise { @@ -275,6 +307,7 @@ export async function registerOrbRelayWithMonitor(args: { isOrbRelayRegistrationAlerting({ consecutiveFailures: args.state.consecutiveFailures, drainLastAtMs: args.drainState?.lastDrainAtMs ?? null, + bootAtMs: args.bootAtMs, ...(args.nowMs !== undefined ? { nowMs: args.nowMs } : {}), }); (alerting ? console.error : console.warn)( diff --git a/src/selfhost/pg-queue.ts b/src/selfhost/pg-queue.ts index 1f460cd8c1..f5e4932662 100644 --- a/src/selfhost/pg-queue.ts +++ b/src/selfhost/pg-queue.ts @@ -243,6 +243,46 @@ export interface PgQueueOptions { backgroundConcurrency?: number; } +/** #9139/#9136: append a `dead_lettered` review_audit row for a queue job that just permanently failed -- + * the ops.ts anomaly-alerter's DLQ signal (`event_type = 'dead_lettered'`) reads this table, but NOTHING + * wrote that event type anywhere in the codebase before this fix (only 'gate_decision' and the outcome/ + * reversal types were ever written), so `h.dlqCount` was permanently 0 and the "N review(s) DEAD-LETTERED" + * alert could never fire. + * + * Scoped to jobs with an identifiable repo+PR (extractPayloadContext) -- a maintenance/export job with no + * PR context (orb-export, a cron sweep, ...) has nothing sensible to attribute a REVIEW dead-letter to, and + * writing one anyway would fabricate a target the ops dashboard/alert can't meaningfully act on. Those + * still increment loopover_jobs_dead_total exactly as before; this is additive, review-scoped detail on + * top, not a replacement for the general counter. + * + * target_id uses the SAME `owner/repo#pr` shape outcomes-wire.ts's reviewAuditTargetId stamps (the + * target_id NAMESPACE MISMATCH #9136 also fixes: review_targets.id used a different, incompatible + * `project:kind:repo#pr` shape, so a join against it could never match this table's rows even when + * review_targets was still live). `project` mirrors alerts-wire.ts's own GITHUB_APP_SLUG fallback so the + * row is scoped the same way every other ops.ts read already is. Best-effort: a write failure here must + * never mask the job's own already-successful dead-letter transition above. */ +async function recordReviewAuditDeadLetter(pool: Pool, payload: string, errMsg: string): Promise { + const context = extractPayloadContext(payload); + if (!context?.repo || context.pr_number === undefined) return; + const project = process.env.GITHUB_APP_SLUG?.trim() || "loopover"; + const targetId = `${context.repo.slice(0, 200)}#${context.pr_number}`; + try { + await pool.query( + `INSERT INTO review_audit (id, project, target_id, event_type, decision, source, head_sha, summary, created_at) + VALUES ($1, $2, $3, 'dead_lettered', NULL, 'selfhost-queue', NULL, $4, $5)`, + [`dead_lettered:${targetId}:${Date.now()}:${Math.random().toString(36).slice(2, 8)}`, project, targetId, errMsg.slice(0, 500), new Date().toISOString()], + ); + } catch (auditError) { + console.warn( + JSON.stringify({ + event: "review_audit_dead_letter_record_error", + targetId, + message: errorMessageWithCause(auditError).slice(0, 160), + }), + ); + } +} + export function createPgQueue( pool: Pool, consume: (message: JobMessage) => Promise, @@ -515,6 +555,23 @@ export function createPgQueue( return Number((await pool.query(`SELECT COUNT(*) AS c FROM ${TABLE} WHERE status='dead'`)).rows[0].c); } + /** #9139: jobs dead-lettered within the trailing `windowMs` -- the self-host counterpart to + * loopover_dlq_dead_lettered_recent's cloud-worker source (countRecentDeadLetters, which reads + * `github_app.dlq_dead_lettered` audit_events written only by processDlqBatch, itself reached only from + * the Cloudflare `queue()` handler that server.ts never calls). Unlike deadCount() above (the STANDING + * depth), this is a RATE-style window over `dead_at`, so it reads 0 once a self-host operator's own + * reviveDeadLetterJobs()/manual replay clears the backlog, exactly like the cloud-side gauge already + * behaves once the DLQ consumer catches up. */ + async function recentDeadCount(windowMs: number): Promise { + return Number( + ( + await pool.query(`SELECT COUNT(*) AS c FROM ${TABLE} WHERE status='dead' AND dead_at IS NOT NULL AND dead_at > $1`, [ + Date.now() - windowMs, + ]) + ).rows[0].c, + ); + } + async function listDeadLetterJobs(limit: number, offset: number): Promise { const res = await pool.query( `SELECT id, payload, attempts, last_error, created_at, dead_at @@ -668,10 +725,12 @@ export function createPgQueue( /** Wraps reviveDeadLetterJobs() for the setInterval callback below, which has no rejection handler of its * own -- a transient pool/driver/metric failure here would otherwise surface as an unhandled promise - * rejection and can terminate the process (fatal when POSTHOG_API_KEY is unset, since initPostHog's - * enableExceptionAutocapture only installs a handler when PostHog is configured), exactly the failure mode - * pump()'s own try/catch above guards against for the main poll loop. A failed revive tick just waits for - * the next interval, same as a failed poll tick waits for the next poll. + * rejection and terminate the process, uniformly regardless of PostHog config (#9133: + * installSelfHostCrashHandlers is now the sole, unconditional crash-and-restart contract -- previously + * this specific async/rejection case was fatal only when POSTHOG_API_KEY was UNSET, since initPostHog's + * enableExceptionAutocapture, when configured, silently captured the rejection without ever exiting), + * exactly the failure mode pump()'s own try/catch above guards against for the main poll loop. A failed + * revive tick just waits for the next interval, same as a failed poll tick waits for the next poll. * * Also wrapped in a PostHog monitor heartbeat (#1824): dead-letter revival stopping SILENTLY (the timer * never fires again, e.g. after an unexpected process-level disruption) is worse than one throwing tick -- @@ -1470,6 +1529,7 @@ export function createPgQueue( [attempts, errMsg, Date.now(), job.id], ); await recordQueueMetric("loopover_jobs_dead_total"); + await recordReviewAuditDeadLetter(pool, job.payload, errMsg); console.error( JSON.stringify({ level: "error", @@ -1649,6 +1709,7 @@ export function createPgQueue( ); }, deadCount, + recentDeadCount, async processingCount() { return Number( ( @@ -1699,6 +1760,7 @@ export function createPgQueue( ); if ((update.rowCount ?? 0) > 0) { await recordQueueMetric("loopover_jobs_dead_total"); + await recordReviewAuditDeadLetter(pool, row.payload, "processing lease expired repeatedly; dead-lettered"); capturePostHogError(new Error("self-host queue job wedged past its lease repeatedly"), { kind: "job_dead", reason: "processing_lease_exhausted", diff --git a/src/selfhost/posthog.ts b/src/selfhost/posthog.ts index 6af1ca1828..b368af6c3f 100644 --- a/src/selfhost/posthog.ts +++ b/src/selfhost/posthog.ts @@ -154,11 +154,19 @@ export async function initPostHog(env: NodeJS.ProcessEnv): Promise { // still exists for explicit drain-before-exit, and PostHog's own recommended client.shutdown() covers // graceful process shutdown. before_send: scrubPostHogEvent, - // Matches Sentry's own Node SDK default posture (Sentry.init() installs global uncaughtException/ - // unhandledRejection handlers unless explicitly disabled) -- a safety net for genuinely-unhandled cases - // beyond the explicit capturePostHogError/capturePostHogReviewFailure call sites below, per PostHog's - // own documented recommendation for Node.js error tracking. - enableExceptionAutocapture: true, + // #9133: OFF, not Sentry's own default-on posture. posthog-node's own autocapture is NOT a safe drop-in + // for Sentry.init()'s equivalent default here: its unhandledRejection listener captures but never + // rethrows or exits (Node's --unhandled-rejections=throw only escalates a rejection to a fatal + // uncaughtException when NO unhandledRejection listener exists at all -- installing this one silently + // downgrades every unhandled rejection to a captured telemetry event with no crash, no Docker restart, + // and no recovery of whatever in-flight job the crash used to reclaim). Its uncaughtException listener + // DOES exit correctly on its own -- but only when it is the SOLE such listener (a foreign listener makes + // it skip its own process.exit(1) entirely), which would make server.ts's own crash-handler installation + // (src/selfhost/process-lifecycle.ts) silently change posthog-node's behavior for that event depending + // on install order. installSelfHostCrashHandlers is now the SOLE, unconditional source of truth for + // BOTH events (installed in server.ts's main(), regardless of whether telemetry is configured), so + // posthog-node must never install its own competing listeners for either. + enableExceptionAutocapture: false, }); active = true; return true; diff --git a/src/selfhost/process-lifecycle.ts b/src/selfhost/process-lifecycle.ts new file mode 100644 index 0000000000..768718feb7 --- /dev/null +++ b/src/selfhost/process-lifecycle.ts @@ -0,0 +1,103 @@ +// Self-host process crash-safety (#9133). server.ts previously registered NO uncaughtException/ +// unhandledRejection handlers of its own -- the reasoning (removed by this fix, see the corrected comment +// in server.ts) was that PostHog's own `enableExceptionAutocapture` (posthog-node) already installs +// equivalent handlers when telemetry is configured, so no manual `process.on` wiring was needed for the +// crash case. That holds for `uncaughtException` ONLY: posthog-node's addUncaughtExceptionListener counts +// OTHER `uncaughtException` listeners and calls its own `process.exit(1)` only when it is the sole one. It +// does NOT hold for `unhandledRejection` -- posthog-node's addUnhandledRejectionListener captures the +// rejection but never rethrows or exits. Node 22's default `--unhandled-rejections=throw` escalates a +// rejection to an uncaught exception ONLY when no `unhandledRejection` listener is registered at all; +// installing PostHog's own (via enableExceptionAutocapture) silently downgrades every unhandled rejection +// into a captured telemetry event with no crash, no Docker restart, and no recovery of whatever in-flight +// job the crash used to reclaim (`loopover_jobs_recovered_total`) -- an absorbing state: the worker loop is +// dead, the process stays alive, and `/health` keeps reporting 200. +// +// FIX: this module is the SOLE, unconditional source of truth for the crash-and-restart contract -- the +// SAME two handlers are installed regardless of whether telemetry is configured, so the contract never +// depends on PostHog's own state. `initPostHog` (posthog.ts) sets `enableExceptionAutocapture: false` +// specifically so posthog-node never installs its OWN competing listeners for these two events: this +// avoids a double-captured exception (its internal listener firing IN ADDITION to this one) AND sidesteps +// the foreign-listener-count heuristic entirely -- a foreign `uncaughtException` listener (this module's +// own) would otherwise make posthog-node silently skip ITS `process.exit(1)`, quietly making this module +// responsible for that anyway. Better to own the whole contract outright than depend on that heuristic. +// +// Mirrors packages/loopover-miner/lib/process-lifecycle.ts's injectable-dependency shape (process/log/exit/ +// captureError), adapted for a single long-running server rather than a one-shot CLI: no SIGINT/SIGTERM or +// cleanup-resource registry here -- server.ts's own graceful `shutdown()` already owns those, unchanged. + +/** The subset of `process` the handlers use; injectable for tests. */ +export type ProcessLike = { + on: (event: string, listener: (...args: unknown[]) => void) => unknown; + exit: (code?: number) => void; +}; + +export type InstallSelfHostCrashHandlersOptions = { + process?: ProcessLike; + /** Structured log line, forwarded to PostHog by the existing installPostHogStructuredLogForwarding wiring + * when telemetry is active (a `level` field is required for that forwarding to pick it up). Defaults to + * `console.error`. */ + log?: (line: string) => void; + exit?: (code: number) => void; + /** Best-effort telemetry capture (`capturePostHogError` at the real call site) -- a no-op when telemetry + * isn't configured. Synchronous; never expected to throw. Belt-and-suspenders alongside `log` (which + * already reaches PostHog via structured-log-forwarding when active) so capture doesn't depend on that + * wiring having run first. */ + captureError?: (error: unknown, context: Record) => void; + /** Awaited before exit so a captured/queued telemetry event has a chance to actually leave the process -- + * `process.exit()` tears the event loop down immediately otherwise, which would make capture a near-total + * no-op in practice for a batching client. No-op default. Never expected to throw/reject. */ + flush?: () => Promise; + /** Reinstall even if handlers were already installed (mainly for tests). */ + force?: boolean; +}; + +let handlersInstalled = false; + +/** Render any thrown/rejected value as a single log-safe string, preferring an Error's stack -- mirrors the + * miner's own describeError exactly. */ +function describeError(value: unknown): string { + if (value instanceof Error) return value.stack ?? value.message; + return String(value); +} + +/** + * Install the uncaughtException/unhandledRejection crash handlers once. Both events are treated IDENTICALLY + * (log, best-effort capture, flush, exit non-zero) -- the point of this fix is that BOTH must terminate the + * process regardless of whether PostHog is configured, not just whichever one posthog-node's own + * (now-disabled) autocapture happened to handle correctly. No-op (returns false) if already installed + * unless `options.force` is set. All of `process`, `log`, `exit`, `captureError`, and `flush` are + * injectable so this is unit-testable without ever registering a REAL process-level handler. + */ +export function installSelfHostCrashHandlers(options: InstallSelfHostCrashHandlersOptions = {}): boolean { + const proc = options.process ?? (process as unknown as ProcessLike); + const log = typeof options.log === "function" ? options.log : (line: string) => console.error(line); + const exit = typeof options.exit === "function" ? options.exit : (code: number) => proc.exit(code); + const captureError = typeof options.captureError === "function" ? options.captureError : () => {}; + const flush = typeof options.flush === "function" ? options.flush : async () => {}; + + if (handlersInstalled && options.force !== true) return false; + handlersInstalled = true; + + const handleFatal = (kind: "uncaughtException" | "unhandledRejection") => async (error: unknown): Promise => { + log( + JSON.stringify({ + level: "fatal", + event: `selfhost_${kind}`, + error: describeError(error).slice(0, 4000), + }), + ); + captureError(error, { kind }); + await flush(); + exit(1); + }; + + proc.on("uncaughtException", handleFatal("uncaughtException")); + proc.on("unhandledRejection", handleFatal("unhandledRejection")); + + return true; +} + +/** Test-only: clear the installed flag so each test starts from a clean lifecycle. */ +export function resetSelfHostCrashHandlersForTest(): void { + handlersInstalled = false; +} diff --git a/src/selfhost/sqlite-queue.ts b/src/selfhost/sqlite-queue.ts index 6b3dc33771..3c6b72f089 100644 --- a/src/selfhost/sqlite-queue.ts +++ b/src/selfhost/sqlite-queue.ts @@ -149,6 +149,35 @@ export interface SqliteQueueOptions { backgroundConcurrency?: number; } +/** #9139/#9136: append a `dead_lettered` review_audit row for a queue job that just permanently failed -- + * mirrors pg-queue.ts's identical helper (see its own doc comment for the full "why": ops.ts's DLQ + * anomaly signal reads `event_type = 'dead_lettered'`, but nothing wrote that event type anywhere before + * this fix). Scoped to jobs with an identifiable repo+PR; a job with no PR context has nothing sensible to + * attribute a REVIEW dead-letter to and still increments loopover_jobs_dead_total exactly as before. + * target_id uses the SAME `owner/repo#pr` shape outcomes-wire.ts's reviewAuditTargetId stamps. Best-effort: + * a write failure here must never mask the job's own already-successful dead-letter transition above. */ +function recordReviewAuditDeadLetter(driver: SqliteDriver, payload: string, errMsg: string): void { + const context = extractPayloadContext(payload); + if (!context?.repo || context.pr_number === undefined) return; + const project = process.env.GITHUB_APP_SLUG?.trim() || "loopover"; + const targetId = `${context.repo.slice(0, 200)}#${context.pr_number}`; + try { + driver.query( + `INSERT INTO review_audit (id, project, target_id, event_type, decision, source, head_sha, summary, created_at) + VALUES (?, ?, ?, 'dead_lettered', NULL, 'selfhost-queue', NULL, ?, ?)`, + [`dead_lettered:${targetId}:${Date.now()}:${Math.random().toString(36).slice(2, 8)}`, project, targetId, errMsg.slice(0, 500), new Date().toISOString()], + ); + } catch (auditError) { + console.warn( + JSON.stringify({ + event: "review_audit_dead_letter_record_error", + targetId, + message: errorMessageWithCause(auditError).slice(0, 160), + }), + ); + } +} + export function createSqliteQueue( driver: SqliteDriver, consume: (message: JobMessage) => Promise, @@ -305,11 +334,13 @@ export function createSqliteQueue( } /** Wraps reviveDeadLetterJobs() for the setInterval callback below, which has no error handler of its own -- - * a transient driver/metric failure here would otherwise surface as an uncaught exception and can terminate - * the process (fatal when POSTHOG_API_KEY is unset, since initPostHog's enableExceptionAutocapture only - * installs a handler when PostHog is configured), exactly the failure mode pump()'s own try/catch above - * guards against for the main poll loop. A failed revive tick just waits for the next interval, same as a - * failed poll tick waits for the next poll. + * a transient driver/metric failure here would otherwise surface as an uncaught exception and terminate the + * process, uniformly regardless of PostHog config (#9133: installSelfHostCrashHandlers is now the sole, + * unconditional crash-and-restart contract -- previously this was fatal only when POSTHOG_API_KEY was + * UNSET, since initPostHog's enableExceptionAutocapture, when configured, silently captured-without-exiting + * the async/rejection case specifically), exactly the failure mode pump()'s own try/catch above guards + * against for the main poll loop. A failed revive tick just waits for the next interval, same as a failed + * poll tick waits for the next poll. * * Also wrapped in a PostHog monitor heartbeat (#1824): dead-letter revival stopping SILENTLY (the timer * never fires again) is worse than one throwing tick -- a crashed tick self-reports via capturePostHogError @@ -730,6 +761,18 @@ export function createSqliteQueue( ); } + /** #9139: jobs dead-lettered within the trailing `windowMs` -- see backend-contracts.ts's DurableQueue + * doc comment for the full "why" (the cloud-worker source this backs on self-host is unreachable here). */ + async function recentDeadCount(windowMs: number): Promise { + return Number( + ( + driver.query(`SELECT COUNT(*) AS c FROM ${TABLE} WHERE status='dead' AND dead_at IS NOT NULL AND dead_at > ?`, [ + Date.now() - windowMs, + ]).rows[0] as { c: number } + ).c, + ); + } + async function listDeadLetterJobs(limit: number, offset: number): Promise { const { rows } = driver.query( `SELECT id, payload, attempts, last_error, created_at, dead_at @@ -1100,6 +1143,7 @@ export function createSqliteQueue( [attempts, errMsg, Date.now(), job.id], ); recordQueueMetric(driver, "loopover_jobs_dead_total"); + recordReviewAuditDeadLetter(driver, job.payload, errMsg); console.error( JSON.stringify({ level: "error", @@ -1287,6 +1331,7 @@ export function createSqliteQueue( ); }, deadCount, + recentDeadCount, async processingCount() { return Number( ( diff --git a/src/server.ts b/src/server.ts index 4fc4227141..8cb7fec2f7 100644 --- a/src/server.ts +++ b/src/server.ts @@ -99,7 +99,9 @@ import { runOrbExportWithMonitor, runScheduledLoopWithMonitor, withOrbRelayDrainReentrancyGuard, + type OrbRelayDrainState, } from "./selfhost/monitored-work"; +import { installSelfHostCrashHandlers } from "./selfhost/process-lifecycle"; import { currentOtelTraceParent, initOpenTelemetry, @@ -118,6 +120,7 @@ import { setLocalReviewContextReader, } from "./signals/focus-manifest-loader"; import { probeReesSecretAtStartup } from "./review/enrichment-wire"; +import { checkReviewSourceFreshness } from "./review/ops"; import { sampleRecentDeadLetters } from "./selfhost/dlq-recent"; import type { JobMessage } from "./types"; @@ -316,15 +319,28 @@ function resolveReviewAuditBinding(): R2Bucket | undefined { } async function main(): Promise { + // #9133: the crash-and-restart contract (an uncaught exception or unhandled rejection kills the process + // so Docker's restart policy recovers it, reclaiming any stale in-flight job) must be independent of + // whether telemetry is configured -- installed FIRST, before anything else in boot, so it also covers a + // failure during preflight/config loading itself, not just the steady-state server loop. See + // src/selfhost/process-lifecycle.ts's own header comment for the full "why" (the PREVIOUS reasoning here + // -- that PostHog's own enableExceptionAutocapture already installs equivalent handlers -- was correct for + // uncaughtException but silently wrong for unhandledRejection, which posthog-node's own handler captures + // but never rethrows or exits). + /* v8 ignore next -- importing this entrypoint starts the Node server; handler-installation logic itself is unit-tested in selfhost-process-lifecycle.test.ts. */ + installSelfHostCrashHandlers({ captureError: (error, context) => capturePostHogError(error, context), flush: flushPostHog }); loadFileSecrets(); /* v8 ignore next -- importing this entrypoint starts the Node server; pure validation is covered in selfhost-preflight tests. */ assertSelfHostPreflight(process.env); // Error tracking (#1468, epic #8286): opt-in via POSTHOG_API_KEY -- the same var #6235's MCP telemetry // already reads -- a complete no-op when unset. REPLACES the old Sentry sink entirely (2026-07-25 epic - // correction: full replacement, not a parallel-run). enableExceptionAutocapture (set inside initPostHog) - // already installs its own uncaughtException/unhandledRejection handlers per PostHog's own documented - // Node.js setup, so no manual process.on wiring is needed here for the crash case; only structured-log - // forwarding needs an explicit install. + // correction: full replacement, not a parallel-run). enableExceptionAutocapture is now OFF (#9133; set + // inside initPostHog) -- installSelfHostCrashHandlers above is the sole, unconditional source of truth + // for the uncaughtException/unhandledRejection crash contract, so posthog-node never installs its OWN + // competing listeners for either event (avoiding both a double-captured exception and posthog-node's own + // foreign-listener-count heuristic for uncaughtException, which would otherwise silently skip ITS + // process.exit(1) the moment it saw a foreign listener -- this module's own). Structured-log forwarding is + // unaffected and still needs its own explicit install below. // // #6325 follow-up: initialized HERE, before every boot-time advisory below (emptyConfigDirAdvisory / // sqliteBackupAdvisory / publicOriginReachabilityAdvisory all "warn LOUDLY" via console.error, which @@ -845,7 +861,10 @@ async function main(): Promise { gauge("loopover_queue_pending", () => backend.queue.size()); gauge("loopover_queue_dead", () => backend.queue.deadCount()); - gauge("loopover_dlq_dead_lettered_recent", () => sampleRecentDeadLetters(env)); + // #9139: pass backend.queue's own recentDeadCount so this reads the self-host dead-letter path (the + // audit_events-based cloud-worker source this gauge otherwise reads is unreachable here -- see + // sampleRecentDeadLetters's own doc comment). + gauge("loopover_dlq_dead_lettered_recent", () => sampleRecentDeadLetters(env, undefined, backend.queue)); gauge("loopover_queue_processing", () => backend.queue.processingCount()); const durableJobMetric = async (name: string): Promise => Number((await backend.queue.stats())[name] ?? 0); @@ -903,6 +922,16 @@ async function main(): Promise { gauge("loopover_d1_database_size_bytes", () => d1DatabaseSizeBytesSample()); gaugeVector("loopover_d1_table_row_count", () => d1TableRowCountSamples()); gauge("loopover_signal_snapshots_rows_per_key", () => d1SignalSnapshotsRowsPerKeySample()); + // #9136: the generalizable fix — the NEXT review-source orphaning (a table a downstream module treats as + // live, silently stops being written) must be loud, not silent, the way review_targets' own 2026-06-22 + // orphaning went unnoticed for months. 1 = fresh (a row inside that table's own consumer's window), 0 = + // stale. See checkReviewSourceFreshness's own doc comment for exactly which tables/windows are tracked. + gaugeVector("loopover_review_source_fresh", async () => + (await checkReviewSourceFreshness(env)).map((check) => ({ + labels: { table: check.table, window_days: String(check.windowDays) }, + value: check.fresh ? 1 : 0, + })), + ); // Backlog-vs-fresh-intake fairness lanes (#selfhost-lane-observability, see queue-fairness.ts): the SAME // `foreground_lane` classification the claim-time fairness mechanism itself consults, so an operator can see // whether a stuck-looking queue is actually a real, unresolved PR-review backlog (high backlog-convergence @@ -1224,7 +1253,8 @@ async function main(): Promise { // attempt can consult `relayDrainState.lastDrainAtMs` -- a single registration timeout must not alert // while the drain loop is still proving the relay connection itself is alive (#selfhost-runtime-drift // follow-up). Stays undefined in push mode, where there is no drain loop. - const relayDrainState = process.env.ORB_RELAY_MODE === "pull" ? { pendingAck: [] as string[], lastDrainAtMs: null as number | null } : undefined; + const relayDrainState: OrbRelayDrainState | undefined = + process.env.ORB_RELAY_MODE === "pull" ? { pendingAck: [], lastDrainAtMs: null, consecutiveFailures: 0 } : undefined; // Brokered self-host: register our relay target with the central Orb (best-effort). PUSH mode (default) // registers a public relay URL the Orb POSTs to; PULL mode (ORB_RELAY_MODE=pull) registers no URL and the @@ -1246,6 +1276,7 @@ async function main(): Promise { env: orbRelayEnv, state: orbRelayRegistrationState, register: registerOrbRelayTargetWithRetry, + bootAtMs: startedAt, ...(relayDrainState ? { drainState: relayDrainState } : {}), }).catch((error) => { capturePostHogError(error, { kind: "orb_relay_register" }, "orb_relay_register"); @@ -1256,9 +1287,21 @@ async function main(): Promise { // an operator staring at the registration-failures counter alone can't tell "one hiccup" from "actually // stuck" -- these two gauges are the SAME two signals that gate, sampled live at scrape time. gauge("loopover_orb_relay_register_consecutive_failures", () => orbRelayRegistrationState.consecutiveFailures); - gauge("loopover_orb_relay_drain_seconds_since_last", () => - relayDrainState?.lastDrainAtMs == null ? -1 : Math.floor((Date.now() - relayDrainState.lastDrainAtMs) / 1000), - ); + // #9128: a dedicated DRAIN failure streak, distinct from the registration one above -- a flapping drain + // (succeeds often enough that seconds_since_last below never crosses the no-progress window) previously + // had no signal of its own at all. 0 in push mode (relayDrainState undefined) -- there is no drain loop. + gauge("loopover_orb_relay_drain_consecutive_failures", () => relayDrainState?.consecutiveFailures ?? 0); + gauge("loopover_orb_relay_drain_seconds_since_last", () => { + // #9128: -1 in push mode ONLY -- there is no drain loop at all, so the metric genuinely doesn't apply + // (unchanged from before this fix). In PULL mode, a null lastDrainAtMs (never once completed a drain + // tick, whether because boot just happened or because every tick has thrown) now ages from process + // BOOT instead of reading a flat -1 forever -- so a never-drained pull-mode instance still climbs past + // LoopoverOrbRelayRegistrationStuck's existing >1800s threshold on its own, the same as a + // previously-draining instance that's gone stale. + if (!relayDrainState) return -1; + const sinceMs = relayDrainState.lastDrainAtMs ?? startedAt; + return Math.floor((Date.now() - sinceMs) / 1000); + }); /* v8 ignore stop */ // D1 size/row-count observability probe (#3810): a no-op everywhere until an operator sets all three diff --git a/test/unit/alerts-metric-name-references.test.ts b/test/unit/alerts-metric-name-references.test.ts index 3057a99e12..fb8d6f76e1 100644 --- a/test/unit/alerts-metric-name-references.test.ts +++ b/test/unit/alerts-metric-name-references.test.ts @@ -64,6 +64,50 @@ function findUnknownMetricReferences(doc: AlertsDoc, registeredNames: ReadonlySe const registeredNames = new Set(DEFAULT_METRIC_META.map(([name]) => name)); +// #9139: alertmanager.yml's inhibit-rule EXAMPLES reference a Prometheus alertname by string +// (`alertname="…"`), entirely inside YAML comments (every inhibit_rules block ships commented-out until an +// operator opts in) -- so parseYaml can never see them; a plain regex scan over the RAW file text is the +// only way to catch a stale/misspelled name before an operator uncomments it and gets a rule that silently +// never inhibits (the exact live bug: `alertname="LoopOverTargetDown"`, capital O, against the real rule +// `LoopoverTargetDown`). +const ALERTNAME_PATTERN = /alertname="([A-Za-z0-9_]+)"/g; + +/** Every `alertname="…"` value referenced anywhere in `text` (comments included), in order of appearance. */ +function alertnameReferences(text: string): string[] { + return [...text.matchAll(ALERTNAME_PATTERN)].map((match) => match[1]!); +} + +/** Every referenced alertname that does NOT match a real `alert:` rule name in `knownAlertNames`. */ +function findUnknownAlertnameReferences(text: string, knownAlertNames: ReadonlySet): string[] { + return alertnameReferences(text).filter((name) => !knownAlertNames.has(name)); +} + +describe("alertmanager.yml alertname references (#9139)", () => { + const alertsDoc = parseYaml(readFileSync("prometheus/rules/alerts.yml", "utf8")) as AlertsDoc; + const knownAlertNames = new Set(alertsDoc.groups.flatMap((group) => group.rules.map((rule) => rule.alert))); + + it("resolves every alertname reference in the real alertmanager.yml (including commented inhibit-rule examples) to a real alerts.yml rule", () => { + const raw = readFileSync("alertmanager/alertmanager.yml", "utf8"); + // Sanity: the file actually contains at least one alertname reference to check -- otherwise this + // assertion would trivially pass even if the commented example were deleted entirely. + expect(alertnameReferences(raw).length).toBeGreaterThan(0); + expect(findUnknownAlertnameReferences(raw, knownAlertNames)).toEqual([]); + }); + + it("REGRESSION (#9139): flags the exact prior live bug -- a capitalization mismatch against the real rule name", () => { + const fixture = '# - source_matchers:\n# - alertname="LoopOverTargetDown"\n'; + expect(findUnknownAlertnameReferences(fixture, knownAlertNames)).toEqual(["LoopOverTargetDown"]); + // The corrected spelling is a real rule and is not flagged (the fix this regression test pins). + expect(findUnknownAlertnameReferences('alertname="LoopoverTargetDown"', knownAlertNames)).toEqual([]); + }); + + it("does not flag a fabricated, never-registered alertname", () => { + expect(findUnknownAlertnameReferences('alertname="TotallyMadeUpAlertThatDoesNotExist"', knownAlertNames)).toEqual([ + "TotallyMadeUpAlertThatDoesNotExist", + ]); + }); +}); + describe("alert annotation metric-name references (#5816)", () => { it("references only registered metrics, a recognized wildcard family, or a documented external prefix in the real alerts.yml", () => { const doc = parseYaml(readFileSync("prometheus/rules/alerts.yml", "utf8")) as AlertsDoc; diff --git a/test/unit/clock-skew.test.ts b/test/unit/clock-skew.test.ts index 0cf72b347f..1a852aeb49 100644 --- a/test/unit/clock-skew.test.ts +++ b/test/unit/clock-skew.test.ts @@ -54,8 +54,13 @@ describe("clock-skew", () => { }); describe("clock-skew sample age (#7000)", () => { - it("reports the -1 never-sampled sentinel before any successful sample", () => { - expect(clockSkewSampleAgeSeconds()).toBe(-1); + it("ages from the reset/module-load time (not a flat -1) before any successful sample (#9128)", () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-07-06T12:00:00.000Z")); + resetClockSkewForTest(); // re-captures the boot-time reference under the fake clock + expect(clockSkewSampleAgeSeconds()).toBe(0); // just reset -- no time has passed yet + vi.setSystemTime(new Date("2026-07-06T12:10:00.000Z")); + expect(clockSkewSampleAgeSeconds()).toBe(600); // 10 minutes later, still never sampled }); it("reports the sample age in seconds after a successful sample, growing as time passes", () => { @@ -76,12 +81,15 @@ describe("clock-skew sample age (#7000)", () => { expect(clockSkewSampleAgeSeconds()).toBe(60); // still measured from the 12:00:00 sample, not "just now" }); - it("resetClockSkewForTest restores the age to the -1 never-sampled sentinel", () => { + it("resetClockSkewForTest re-arms the never-sampled reference point to the moment of reset", () => { vi.useFakeTimers(); vi.setSystemTime(new Date("2026-07-06T12:00:00.000Z")); recordClockSkewFromResponse(new Response(null, { headers: { date: "Mon, 06 Jul 2026 12:00:00 GMT" } })); - expect(clockSkewSampleAgeSeconds()).not.toBe(-1); + vi.setSystemTime(new Date("2026-07-06T12:05:00.000Z")); + expect(clockSkewSampleAgeSeconds()).toBe(300); resetClockSkewForTest(); - expect(clockSkewSampleAgeSeconds()).toBe(-1); + expect(clockSkewSampleAgeSeconds()).toBe(0); // never-sampled again, re-based at THIS moment, not -1 + vi.setSystemTime(new Date("2026-07-06T12:06:00.000Z")); + expect(clockSkewSampleAgeSeconds()).toBe(60); // ages from the reset point, still never sampled }); }); diff --git a/test/unit/dlq-recent.test.ts b/test/unit/dlq-recent.test.ts index 0bea3e6e0a..d7ddb70c3d 100644 --- a/test/unit/dlq-recent.test.ts +++ b/test/unit/dlq-recent.test.ts @@ -33,5 +33,24 @@ describe("dlq-recent gauge helpers (#2083)", () => { vi.spyOn(repositories, "countRecentDeadLetters").mockRejectedValue(new Error("db down")); expect(await sampleRecentDeadLetters({} as Env)).toBe(0); }); + + // #9139: on self-host, the audit_events-based cloud-worker source above is structurally always 0 (the + // Cloudflare `queue()` handler that writes it is never invoked) -- a passed selfHostQueue reads the + // self-host queue's OWN dead-letter path instead, and EXCLUSIVELY (never falls through to the D1 path). + describe("selfHostQueue arm (#9139)", () => { + it("reads recentDeadCount from the self-host queue instead of countRecentDeadLetters when provided", async () => { + const countRecentDeadLettersSpy = vi.spyOn(repositories, "countRecentDeadLetters"); + const selfHostQueue = { recentDeadCount: vi.fn().mockResolvedValue(4) }; + + expect(await sampleRecentDeadLetters({} as Env, undefined, selfHostQueue)).toBe(4); + expect(selfHostQueue.recentDeadCount).toHaveBeenCalledWith(DLQ_RECENT_WINDOW_MS); + expect(countRecentDeadLettersSpy).not.toHaveBeenCalled(); + }); + + it("degrades to 0 when the self-host queue's own recentDeadCount throws", async () => { + const selfHostQueue = { recentDeadCount: vi.fn().mockRejectedValue(new Error("pool exhausted")) }; + expect(await sampleRecentDeadLetters({} as Env, undefined, selfHostQueue)).toBe(0); + }); + }); }); }); diff --git a/test/unit/ops.test.ts b/test/unit/ops.test.ts index a537158a5c..966520c188 100644 --- a/test/unit/ops.test.ts +++ b/test/unit/ops.test.ts @@ -1,6 +1,7 @@ import { describe, expect, it, vi } from "vitest"; import { buildCalibrationBins, + checkReviewSourceFreshness, computeAgentHealth, computeCalibration, defaultOpsHealthDeps, @@ -261,14 +262,16 @@ function healthEnv(): Env { bind() { return { first: async () => { - if (sql.includes("status IN ('merged', 'closed')")) return { n: 2 }; // recent auto-actions denominator + // #9136: recentAutoActions repointed onto review_audit's own gate_decision rows. + if (sql.includes("event_type = 'gate_decision'") && sql.includes("decision IN ('merge', 'close')")) return { n: 2 }; if (sql.includes("event_type = 'dead_lettered'") && sql.includes("COUNT(*)")) return { n: 0 }; return { n: 0 }; }, all: async () => { if (sql.includes("GROUP BY status")) return { results: [{ status: "merged", n: 8 }, { status: "manual", n: 2 }, { status: "queued", n: 1 }] }; if (sql.includes("GROUP BY verdict")) return { results: [{ verdict: "merge", n: 8 }, { verdict: "manual", n: 2 }] }; - if (sql.includes("reversal_reverted")) return { results: [{ number: 99, repo: "o/r", status: "merged", event_type: "reversal_reverted" }] }; + // #9136: repo/number parsed from target_id (owner/repo#n), no review_targets join. + if (sql.includes("reversal_reverted")) return { results: [{ target_id: "o/r#99", event_type: "reversal_reverted" }] }; if (sql.includes("event_type IN ('reviewed', 'shadow_reviewed')")) return { results: [{ target_id: "t1", decision: "merge", summary: "ok", created_at: "2026-06-13T00:00:00Z" }] }; return { results: [] }; }, @@ -645,13 +648,14 @@ describe("computeAgentHealth empty-ledger fallbacks", () => { bind() { return { first: async () => { - if (sql.includes("status IN ('merged', 'closed')")) return { n: 1 }; + if (sql.includes("event_type = 'gate_decision'") && sql.includes("decision IN ('merge', 'close')")) return { n: 1 }; if (sql.includes("event_type = 'dead_lettered'") && sql.includes("COUNT(*)")) return {}; // no n → `?? dlqTargets.length` return {}; }, all: async () => { - // dead-letter display sample (has rows) — and a row with verdict/last_error null - if (sql.includes("event_type = 'dead_lettered'")) return { results: [{ number: 7, repo: "o/r", verdict: null, last_error: null }] }; + // #9136: dead-letter display sample (has rows) — target_id/summary, no review_targets join + // — and a row with a null summary (lastError null). + if (sql.includes("event_type = 'dead_lettered'")) return { results: [{ target_id: "o/r#7", summary: null }] }; return {}; }, }; @@ -662,6 +666,165 @@ describe("computeAgentHealth empty-ledger fallbacks", () => { } as unknown as Env; const h = await computeAgentHealth(env, healthConfig); expect(h.dlqTargets).toHaveLength(1); + expect(h.dlqTargets?.[0]).toEqual({ number: 7, repo: "o/r", verdict: null, lastError: null }); expect(h.dlqCount).toBe(1); // fell back to dlqTargets.length }); }); + +// ── #9136: the generalizable fix — the NEXT orphaning (a table a downstream module treats as live, +// silently stopped being written) must be loud, not silent, the way review_targets' own 2026-06-22 +// orphaning was for months. Real D1 (createTestEnv applies every migration), not a hand-rolled mock, so +// the actual `datetime('now', ?)` window arithmetic is exercised for real. ───────────────────────────── +describe("checkReviewSourceFreshness (#9136)", () => { + it("reads review_targets and review_audit as STALE when both are empty (the ground state right now)", async () => { + const env = createTestEnv(); + const checks = await checkReviewSourceFreshness(env); + expect(checks).toEqual([ + { table: "review_targets", windowDays: 90, fresh: false }, + { table: "review_audit", windowDays: 7, fresh: false }, + ]); + }); + + it("reads review_audit as FRESH when it has a row inside its 7-day window (the live, steady-state case)", async () => { + const env = createTestEnv(); + await env.DB.prepare( + `INSERT INTO review_audit (id, project, target_id, event_type, created_at) VALUES (?, ?, ?, 'gate_decision', datetime('now'))`, + ) + .bind("a1", "loopover", "o/r#1") + .run(); + const checks = await checkReviewSourceFreshness(env); + expect(checks.find((c) => c.table === "review_audit")).toEqual({ table: "review_audit", windowDays: 7, fresh: true }); + }); + + it("reads review_audit as STALE when its only row is OUTSIDE the 7-day window (the boundary arm)", async () => { + const env = createTestEnv(); + await env.DB.prepare( + `INSERT INTO review_audit (id, project, target_id, event_type, created_at) VALUES (?, ?, ?, 'gate_decision', datetime('now', '-8 days'))`, + ) + .bind("a2", "loopover", "o/r#2") + .run(); + const checks = await checkReviewSourceFreshness(env); + expect(checks.find((c) => c.table === "review_audit")).toEqual({ table: "review_audit", windowDays: 7, fresh: false }); + }); + + it("reads review_targets as FRESH when it has a row inside its 90-day window", async () => { + const env = createTestEnv(); + await env.DB.prepare( + `INSERT INTO review_targets (id, project, kind, repo, number, terminal_at) VALUES (?, ?, 'pull_request', ?, 1, datetime('now'))`, + ) + .bind("loopover:pull_request:o/r#1", "loopover", "o/r") + .run(); + const checks = await checkReviewSourceFreshness(env); + expect(checks.find((c) => c.table === "review_targets")).toEqual({ table: "review_targets", windowDays: 90, fresh: true }); + }); + + it("reads review_targets as STALE once its newest row falls outside the 90-day window (the 2026-09-20 cliff this exists to catch)", async () => { + const env = createTestEnv(); + await env.DB.prepare( + `INSERT INTO review_targets (id, project, kind, repo, number, terminal_at) VALUES (?, ?, 'pull_request', ?, 1, datetime('now', '-91 days'))`, + ) + .bind("loopover:pull_request:o/r#1", "loopover", "o/r") + .run(); + const checks = await checkReviewSourceFreshness(env); + expect(checks.find((c) => c.table === "review_targets")).toEqual({ table: "review_targets", windowDays: 90, fresh: false }); + }); + + it("fails CLOSED (stale) on a read error, e.g. a dropped/missing table, rather than throwing", async () => { + const env = { + DB: { + prepare() { + throw new Error("no such table: review_audit"); + }, + }, + } as unknown as Env; + const checks = await checkReviewSourceFreshness(env); + expect(checks.every((c) => c.fresh === false)).toBe(true); + }); +}); + +// ── #9136: repo/number parsed from review_audit's own target_id (parseReviewAuditTargetId) ──────── +describe("computeAgentHealth target_id parsing (#9136)", () => { + it("derives 'closed' status for a reversal_reopened row (the other ternary arm vs 'merged')", async () => { + const env = { + DB: { + prepare(sql: string) { + return { + bind() { + return { + first: async () => ({ n: 0 }), + all: async () => { + if (sql.includes("reversal_reverted")) return { results: [{ target_id: "o/r#5", event_type: "reversal_reopened" }] }; + return {}; + }, + }; + }, + }; + }, + }, + } as unknown as Env; + const h = await computeAgentHealth(env, healthConfig); + expect(h.reversedTargets?.[0]).toEqual({ number: 5, repo: "o/r", status: "closed", eventType: "reversal_reopened" }); + }); + + it("filters out a reversal row whose target_id can't be parsed, without dropping a well-formed sibling", async () => { + const env = { + DB: { + prepare(sql: string) { + return { + bind() { + return { + first: async () => ({ n: 0 }), + all: async () => { + if (sql.includes("reversal_reverted")) { + return { + results: [ + { target_id: "no-hash-at-all", event_type: "reversal_reverted" }, // no '#' -> null + { target_id: "#5", event_type: "reversal_reverted" }, // hashIndex === 0 -> null + { target_id: "o/r#not-a-number", event_type: "reversal_reverted" }, // NaN -> null + { target_id: "o/r#12", event_type: "reversal_reverted" }, // well-formed + ], + }; + } + return {}; + }, + }; + }, + }; + }, + }, + } as unknown as Env; + const h = await computeAgentHealth(env, healthConfig); + expect(h.reversedTargets).toHaveLength(1); + expect(h.reversedTargets?.[0]).toEqual({ number: 12, repo: "o/r", status: "merged", eventType: "reversal_reverted" }); + }); + + it("filters out a dead-letter row whose target_id can't be parsed", async () => { + const env = { + DB: { + prepare(sql: string) { + return { + bind() { + return { + first: async () => ({ n: 0 }), + all: async () => { + if (sql.includes("event_type = 'dead_lettered'")) { + return { + results: [ + { target_id: "unparseable", summary: "boom" }, + { target_id: "o/r#9", summary: "ok" }, + ], + }; + } + return {}; + }, + }; + }, + }; + }, + }, + } as unknown as Env; + const h = await computeAgentHealth(env, healthConfig); + expect(h.dlqTargets).toHaveLength(1); + expect(h.dlqTargets?.[0]).toEqual({ number: 9, repo: "o/r", verdict: null, lastError: "ok" }); + }); +}); diff --git a/test/unit/selfhost-metrics.test.ts b/test/unit/selfhost-metrics.test.ts index 0c0fddddc3..94e231694e 100644 --- a/test/unit/selfhost-metrics.test.ts +++ b/test/unit/selfhost-metrics.test.ts @@ -194,6 +194,53 @@ describe("metrics registry (#982)", () => { incr("ok_total"); expect((await renderMetrics())).toContain("ok_total 1"); }); + + // #9139: a throwing sampler previously vanished entirely -- no series, no counter, no trace it ever + // existed this scrape. Every queue-backlog alert reads exactly this shape of gauge (a live DB read), so a + // DB incident silently deactivated the alerts meant to catch it. Now the failure is itself counted and the + // gauge still emits a -1 sentinel (the same "impossible for a healthy gauge" convention as + // loopover_clock_skew_sample_age_seconds), so its absence is visible instead of silent. + describe("failing gauge sampler visibility (#9139)", () => { + it("counts the failure by metric name and emits a -1 sentinel series instead of vanishing", async () => { + registerMetricMeta("bad_gauge", { help: "A gauge that throws.", type: "gauge" }); + gauge("bad_gauge", () => { + throw new Error("db down"); + }); + + const out = await renderMetrics(); + expect(out).toContain("bad_gauge -1"); + expect(out).toContain('loopover_metrics_sampler_errors_total{metric="bad_gauge"} 1'); + }); + + it("accumulates across repeated scrapes (a sustained failure, not a one-shot)", async () => { + gauge("still_bad", () => { + throw new Error("still down"); + }); + + await renderMetrics(); + await renderMetrics(); + const out = await renderMetrics(); + expect(out).toContain('loopover_metrics_sampler_errors_total{metric="still_bad"} 3'); + }); + + it("does not touch the sampler-errors counter for a gauge that succeeds (the other arm)", async () => { + gauge("healthy_gauge", () => 42); + + const out = await renderMetrics(); + expect(out).toContain("healthy_gauge 42"); + expect(out).not.toContain("loopover_metrics_sampler_errors_total"); + }); + + it("counts an async gauge sampler's rejection the same as a sync throw", async () => { + gauge("bad_async_gauge", async () => { + throw new Error("async db down"); + }); + + const out = await renderMetrics(); + expect(out).toContain("bad_async_gauge -1"); + expect(out).toContain('loopover_metrics_sampler_errors_total{metric="bad_async_gauge"} 1'); + }); + }); }); describe("gaugeVector (#selfhost-lane-observability)", () => { @@ -273,6 +320,19 @@ describe("gaugeVector (#selfhost-lane-observability)", () => { expect(await renderMetrics()).toContain("ok_total 1"); }); + // #9139: same failure-visibility fix as the plain gauge() case, adapted for a vector's unknown-at-failure + // label set -- there's no single value to sentinel, so only the failure counter is the actionable signal. + it("counts a failing gaugeVector sampler by metric name, even though it has no single value to sentinel (#9139)", async () => { + registerMetricMeta("bad_vector_gauge", { help: "A vector gauge that throws.", type: "gauge" }); + gaugeVector("bad_vector_gauge", () => { + throw new Error("db down"); + }); + + const out = await renderMetrics(); + expect(out).toContain('loopover_metrics_sampler_errors_total{metric="bad_vector_gauge"} 1'); + expect(out).not.toMatch(/^bad_vector_gauge\{/m); + }); + it("re-registering the same name replaces the sampler", async () => { gaugeVector("replaced_v", () => [{ labels: { x: "1" }, value: 1 }]); gaugeVector("replaced_v", () => [{ labels: { x: "2" }, value: 2 }]); diff --git a/test/unit/selfhost-monitored-work.test.ts b/test/unit/selfhost-monitored-work.test.ts index 992eb02f62..fddbd40dc8 100644 --- a/test/unit/selfhost-monitored-work.test.ts +++ b/test/unit/selfhost-monitored-work.test.ts @@ -387,32 +387,104 @@ describe("self-host monitored recurring work", () => { expect(state.lastDrainAtMs).toBeNull(); }); + describe("drain failure counting (#9128)", () => { + it("counts a drain throw: increments the failed result series and the consecutive-failure streak, then rethrows", async () => { + const state: OrbRelayDrainState = { pendingAck: [], lastDrainAtMs: null }; + const drain = vi.fn().mockRejectedValue(new Error("broker down")); + + await expect( + drainOrbRelayWithMonitor({ state, relayEnv: {}, env: {} as Env, drain, enqueue: vi.fn() }), + ).rejects.toThrow("broker down"); + + expect(state.consecutiveFailures).toBe(1); + expect(await renderMetrics()).toContain('loopover_orb_relay_drains_total{result="failed"} 1'); + }); + + it("accumulates the consecutive-failure streak across repeated throws (a drain that only ever throws)", async () => { + const state: OrbRelayDrainState = { pendingAck: [], lastDrainAtMs: null }; + const drain = vi.fn().mockRejectedValue(new Error("still down")); + + for (let i = 0; i < 3; i++) { + await expect( + drainOrbRelayWithMonitor({ state, relayEnv: {}, env: {} as Env, drain, enqueue: vi.fn() }), + ).rejects.toThrow("still down"); + } + + // REGRESSION (#9128): a drain that only ever throws must produce a firing condition -- both the + // consecutive-failure streak and lastDrainAtMs staying null (ageable from boot, see server.ts's own + // gauge and isOrbRelayRegistrationAlerting) are exactly that condition. + expect(state.consecutiveFailures).toBe(3); + expect(state.lastDrainAtMs).toBeNull(); + expect(await renderMetrics()).toContain('loopover_orb_relay_drains_total{result="failed"} 3'); + }); + + it("resets the consecutive-failure streak to 0 on the next drain that completes without throwing", async () => { + const state: OrbRelayDrainState = { pendingAck: [], lastDrainAtMs: null, consecutiveFailures: 2 }; + const drain = vi.fn().mockResolvedValue([]); + + await drainOrbRelayWithMonitor({ state, relayEnv: {}, env: {} as Env, drain, enqueue: vi.fn(), nowMs: 5_000 }); + + expect(state.consecutiveFailures).toBe(0); + expect(state.lastDrainAtMs).toBe(5_000); + }); + + it("treats an absent consecutiveFailures field as 0 on the first throw (the ?? nullish arm)", async () => { + // No consecutiveFailures key at all -- mirrors every pre-existing state literal in this test file / server.ts. + const state: OrbRelayDrainState = { pendingAck: [], lastDrainAtMs: null }; + expect(state.consecutiveFailures).toBeUndefined(); + const drain = vi.fn().mockRejectedValue(new Error("broker down")); + + await expect( + drainOrbRelayWithMonitor({ state, relayEnv: {}, env: {} as Env, drain, enqueue: vi.fn() }), + ).rejects.toThrow("broker down"); + + expect(state.consecutiveFailures).toBe(1); + }); + }); + describe("isOrbRelayRegistrationAlerting", () => { it("does not alert below the failure streak with no drain-progress evidence yet (a lone boot-time hiccup)", () => { - expect(isOrbRelayRegistrationAlerting({ consecutiveFailures: 1, drainLastAtMs: null, nowMs: 1_000 })).toBe(false); - expect(isOrbRelayRegistrationAlerting({ consecutiveFailures: 2, drainLastAtMs: null, nowMs: 1_000 })).toBe(false); + expect(isOrbRelayRegistrationAlerting({ consecutiveFailures: 1, drainLastAtMs: null, bootAtMs: 1_000, nowMs: 1_000 })).toBe(false); + expect(isOrbRelayRegistrationAlerting({ consecutiveFailures: 2, drainLastAtMs: null, bootAtMs: 1_000, nowMs: 1_000 })).toBe(false); }); it("does not alert below the failure streak while a known drain is still fresh", () => { expect( - isOrbRelayRegistrationAlerting({ consecutiveFailures: 1, drainLastAtMs: 1_000, nowMs: 1_000 + ORB_RELAY_DRAIN_NO_PROGRESS_WINDOW_MS }), + isOrbRelayRegistrationAlerting({ consecutiveFailures: 1, drainLastAtMs: 1_000, bootAtMs: 0, nowMs: 1_000 + ORB_RELAY_DRAIN_NO_PROGRESS_WINDOW_MS }), ).toBe(false); // exactly at the window boundary — not yet OVER it }); it("alerts once the consecutive-failure streak reaches the threshold, regardless of drain freshness", () => { - expect(isOrbRelayRegistrationAlerting({ consecutiveFailures: 3, drainLastAtMs: Date.now(), nowMs: Date.now() })).toBe(true); - expect(isOrbRelayRegistrationAlerting({ consecutiveFailures: 4, drainLastAtMs: null, nowMs: 1_000 })).toBe(true); + expect(isOrbRelayRegistrationAlerting({ consecutiveFailures: 3, drainLastAtMs: Date.now(), bootAtMs: 0, nowMs: Date.now() })).toBe(true); + expect(isOrbRelayRegistrationAlerting({ consecutiveFailures: 4, drainLastAtMs: null, bootAtMs: 1_000, nowMs: 1_000 })).toBe(true); }); it("alerts once a known last-drain timestamp goes stale past the no-progress window, even below the streak threshold", () => { expect( - isOrbRelayRegistrationAlerting({ consecutiveFailures: 1, drainLastAtMs: 0, nowMs: ORB_RELAY_DRAIN_NO_PROGRESS_WINDOW_MS + 1 }), + isOrbRelayRegistrationAlerting({ consecutiveFailures: 1, drainLastAtMs: 0, bootAtMs: 0, nowMs: ORB_RELAY_DRAIN_NO_PROGRESS_WINDOW_MS + 1 }), ).toBe(true); }); it("defaults nowMs to the real clock when omitted", () => { - expect(isOrbRelayRegistrationAlerting({ consecutiveFailures: 0, drainLastAtMs: Date.now() })).toBe(false); - expect(isOrbRelayRegistrationAlerting({ consecutiveFailures: 0, drainLastAtMs: Date.now() - ORB_RELAY_DRAIN_NO_PROGRESS_WINDOW_MS - 1 })).toBe(true); + expect(isOrbRelayRegistrationAlerting({ consecutiveFailures: 0, drainLastAtMs: Date.now(), bootAtMs: 0 })).toBe(false); + expect(isOrbRelayRegistrationAlerting({ consecutiveFailures: 0, drainLastAtMs: Date.now() - ORB_RELAY_DRAIN_NO_PROGRESS_WINDOW_MS - 1, bootAtMs: 0 })).toBe(true); + }); + + // REGRESSION (#9128): a pull-mode instance that has NEVER completed a single drain tick (every attempt + // throws since boot) must still escalate once it's been stuck past the no-progress window -- the exact + // "-1 never fires" hole this issue fixes. Ages from bootAtMs, not a flat "insufficient signal" false. + describe("never-drained-since-boot ages from bootAtMs, not a permanent false (#9128)", () => { + it("does not alert while still within the no-progress grace period since boot", () => { + expect( + isOrbRelayRegistrationAlerting({ consecutiveFailures: 1, drainLastAtMs: null, bootAtMs: 0, nowMs: ORB_RELAY_DRAIN_NO_PROGRESS_WINDOW_MS }), + ).toBe(false); // exactly at the boundary — not yet OVER it + }); + + it("alerts once a NEVER-drained instance has been stuck past the no-progress window since boot", () => { + expect( + isOrbRelayRegistrationAlerting({ consecutiveFailures: 1, drainLastAtMs: null, bootAtMs: 0, nowMs: ORB_RELAY_DRAIN_NO_PROGRESS_WINDOW_MS + 1 }), + ).toBe(true); + }); }); }); @@ -425,7 +497,7 @@ describe("self-host monitored recurring work", () => { state.attempts = 1; // the injected register() already bumped attempts before returning const register = vi.fn().mockResolvedValue({ status: "registered" }); - await registerOrbRelayWithMonitor({ env: { ORB_RELAY_MODE: "push" }, state, register, log }); + await registerOrbRelayWithMonitor({ env: { ORB_RELAY_MODE: "push" }, state, register, bootAtMs: 0, log }); expect(mocks.withPostHogMonitor).toHaveBeenCalledWith( "orb-relay-register", @@ -446,7 +518,7 @@ describe("self-host monitored recurring work", () => { state.attempts = 3; // two prior failed attempts before this one succeeded const register = vi.fn().mockResolvedValue({ status: "registered" }); - await registerOrbRelayWithMonitor({ env: { ORB_RELAY_MODE: "pull" }, state, register, log }); + await registerOrbRelayWithMonitor({ env: { ORB_RELAY_MODE: "pull" }, state, register, bootAtMs: 0, log }); expect(log).toHaveBeenCalledWith( JSON.stringify({ event: "selfhost_orb_relay_register_recovered", mode: "pull", attempts: 3 }), @@ -463,7 +535,8 @@ describe("self-host monitored recurring work", () => { state.consecutiveFailures = 1; const register = vi.fn().mockResolvedValue({ status: "failed", reason: "http_500" }); - await registerOrbRelayWithMonitor({ env: { ORB_RELAY_MODE: "pull" }, state, register }); + // bootAtMs = right now: the process just started, well inside the no-progress grace period (#9128). + await registerOrbRelayWithMonitor({ env: { ORB_RELAY_MODE: "pull" }, state, register, bootAtMs: Date.now() }); expect(warnSpy).toHaveBeenCalledWith( JSON.stringify({ level: "warn", event: "selfhost_orb_relay_register_failed", mode: "pull", error: "http_500", attempts: 1, consecutiveFailures: 1 }), @@ -491,6 +564,7 @@ describe("self-host monitored recurring work", () => { state, register, drainState, + bootAtMs: 0, nowMs: 1_000 + 60_000, // well inside ORB_RELAY_DRAIN_NO_PROGRESS_WINDOW_MS }); @@ -512,7 +586,7 @@ describe("self-host monitored recurring work", () => { const drainState: OrbRelayDrainState = { pendingAck: [], lastDrainAtMs: 1_000 }; // still draining fine const register = vi.fn().mockResolvedValue({ status: "failed", reason: "http_500" }); - await registerOrbRelayWithMonitor({ env: { ORB_RELAY_MODE: "pull" }, state, register, drainState, nowMs: 2_000 }); + await registerOrbRelayWithMonitor({ env: { ORB_RELAY_MODE: "pull" }, state, register, drainState, bootAtMs: 0, nowMs: 2_000 }); expect(errorSpy).toHaveBeenCalledWith( JSON.stringify({ level: "error", event: "selfhost_orb_relay_register_failed", mode: "pull", error: "http_500", attempts: 3, consecutiveFailures: 3 }), @@ -539,6 +613,7 @@ describe("self-host monitored recurring work", () => { state, register, drainState, + bootAtMs: 0, nowMs: ORB_RELAY_DRAIN_NO_PROGRESS_WINDOW_MS + 1, }); @@ -561,7 +636,7 @@ describe("self-host monitored recurring work", () => { state.consecutiveFailures = 1; const register = vi.fn().mockResolvedValue({ status: "failed" }); - await registerOrbRelayWithMonitor({ env: {}, state, register }); + await registerOrbRelayWithMonitor({ env: {}, state, register, bootAtMs: 0 }); expect(errorSpy).toHaveBeenCalledWith( JSON.stringify({ level: "error", event: "selfhost_orb_relay_register_failed", mode: "push", error: "unknown", attempts: 1, consecutiveFailures: 1 }), @@ -577,7 +652,7 @@ describe("self-host monitored recurring work", () => { const log = vi.fn(); for (const status of ["skipped", "already_registered", "backoff"] as const) { const register = vi.fn().mockResolvedValue({ status }); - await registerOrbRelayWithMonitor({ env: {}, state: freshState(), register, log }); + await registerOrbRelayWithMonitor({ env: {}, state: freshState(), register, bootAtMs: 0, log }); } expect(log).not.toHaveBeenCalled(); expect(await renderMetrics()).not.toContain("loopover_orb_relay_register_total"); @@ -592,6 +667,7 @@ describe("self-host monitored recurring work", () => { env: { ORB_RELAY_MODE: "push" }, state, register: vi.fn().mockResolvedValue({ status: "registered" }), + bootAtMs: 0, }); expect(consoleLog).toHaveBeenCalledWith( JSON.stringify({ event: "selfhost_orb_relay_register", mode: "push", attempts: 1 }), diff --git a/test/unit/selfhost-pg-queue.test.ts b/test/unit/selfhost-pg-queue.test.ts index 137d8360d5..5fcd7690f0 100644 --- a/test/unit/selfhost-pg-queue.test.ts +++ b/test/unit/selfhost-pg-queue.test.ts @@ -3432,6 +3432,84 @@ describe("createPgQueue (durable #977)", () => { ); }); + // #9139/#9136: nothing wrote review_audit's `dead_lettered` event type before this fix, so ops.ts's DLQ + // anomaly signal was permanently 0. Scoped to jobs with an identifiable repo+PR (regateJob); a plain + // no-context payload has nothing sensible to attribute a review dead-letter to. + describe("review_audit dead-letter recording (#9139/#9136)", () => { + afterEach(() => { + delete process.env.GITHUB_APP_SLUG; + }); + + it("writes a dead_lettered review_audit row (default 'loopover' project) when the payload has repo+PR context", async () => { + const m = makePool(); + m.enqueueJob("1", regateJob(null, 42), 0); + const q = createPgQueue(m.pool, async () => { + throw new Error("ai review failed"); + }, { maxRetries: 1, backoffMs: () => 0 }); + await q.init(); + await q.drain(); + + expect(m.pool.query).toHaveBeenCalledWith( + expect.stringContaining("INSERT INTO review_audit"), + expect.arrayContaining([ + expect.any(String), + "loopover", + "jsonbored/gittensory#42", + expect.stringContaining("ai review failed"), + expect.any(String), + ]), + ); + }); + + it("uses GITHUB_APP_SLUG as the project when set (the other ?? arm)", async () => { + process.env.GITHUB_APP_SLUG = "custom-slug"; + const m = makePool(); + m.enqueueJob("1", regateJob(null, 1), 0); + const q = createPgQueue(m.pool, async () => { + throw new Error("boom"); + }, { maxRetries: 1, backoffMs: () => 0 }); + await q.init(); + await q.drain(); + + expect(m.pool.query).toHaveBeenCalledWith( + expect.stringContaining("INSERT INTO review_audit"), + expect.arrayContaining(["custom-slug"]), + ); + }); + + it("does not write a review_audit row when the payload has no repo/PR context", async () => { + const m = makePool(); + m.enqueueJob("1", { type: "orb-export" }, 0); + const q = createPgQueue(m.pool, async () => { + throw new Error("boom"); + }, { maxRetries: 1, backoffMs: () => 0 }); + await q.init(); + await q.drain(); + + expect(m.pool.query).not.toHaveBeenCalledWith(expect.stringContaining("INSERT INTO review_audit"), expect.anything()); + expect(await q.deadCount()).toBe(3); // makePool's generic COUNT(*) fallback -- the job itself still dead-lettered normally + }); + + it("swallows a review_audit write failure without breaking the job's own dead-letter transition", async () => { + const m = makePool(); + m.enqueueJob("1", regateJob(null, 7), 0); + const realImpl = (m.pool.query as unknown as { getMockImplementation(): (...args: unknown[]) => unknown }).getMockImplementation(); + vi.spyOn(m.pool, "query").mockImplementation(async (sql: unknown, params?: unknown[]) => { + if (String(sql).includes("INSERT INTO review_audit")) throw new Error("connection reset"); + return realImpl!(sql, params); + }); + const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => undefined); + const q = createPgQueue(m.pool, async () => { + throw new Error("boom"); + }, { maxRetries: 1, backoffMs: () => 0 }); + await q.init(); + await q.drain(); + + expect(warnSpy.mock.calls.some((c) => String(c[0]).includes("review_audit_dead_letter_record_error"))).toBe(true); + warnSpy.mockRestore(); + }); + }); + it("size() and deadCount() return numeric counts", async () => { const { pool } = makePool(); // makePool returns { c: "3" } for COUNT queries @@ -3441,6 +3519,28 @@ describe("createPgQueue (durable #977)", () => { expect(await q.deadCount()).toBe(3); }); + // #9139: backs loopover_dlq_dead_lettered_recent on self-host (the cloud-worker audit_events source is + // structurally unreachable there -- see dlq-recent.ts's own doc comment). + it("recentDeadCount() returns a numeric count scoped to status='dead' AND dead_at within the window", async () => { + const m = makePool(); + // makePool returns { c: "3" } for any COUNT(*) query. + const q = createPgQueue(m.pool, async () => undefined); + await q.init(); + + const windowMs = 15 * 60 * 1000; + const before = Date.now(); + expect(await q.recentDeadCount(windowMs)).toBe(3); + + const calls = (m.fn as unknown as ReturnType).mock.calls; + const call = calls.find((c: unknown[]) => String(c[0]).includes("status='dead' AND dead_at IS NOT NULL")); + expect(call).toBeDefined(); + const bound = call![1] as number[]; + expect(bound).toEqual([expect.any(Number)]); + // The bound cutoff is "now - windowMs", not a fixed constant -- sanity-check it's in the right ballpark. + expect(bound[0]).toBeLessThanOrEqual(before - windowMs); + expect(bound[0]).toBeGreaterThan(before - windowMs - 5_000); + }); + it("stats() returns persisted queue metric counts", async () => { const m = makePool(); const q = createPgQueue(m.pool, async () => undefined); diff --git a/test/unit/selfhost-posthog.test.ts b/test/unit/selfhost-posthog.test.ts index 501919926b..9be69db015 100644 --- a/test/unit/selfhost-posthog.test.ts +++ b/test/unit/selfhost-posthog.test.ts @@ -71,10 +71,13 @@ describe("initPostHog", () => { expect(mocks.PostHog).toHaveBeenCalledWith("phc_test_key", expect.objectContaining({ host: "https://eu.i.posthog.com" })); }); - it("enables exception autocapture and installs the before_send scrubber", async () => { + it("disables posthog-node's own exception autocapture (#9133) and installs the before_send scrubber", async () => { await initPostHog({ POSTHOG_API_KEY: "phc_test_key" } as unknown as NodeJS.ProcessEnv); const options = mocks.getLastOptions(); - expect(options.enableExceptionAutocapture).toBe(true); + // #9133: OFF, not Sentry's own default-on posture -- server.ts's installSelfHostCrashHandlers is now the + // sole, unconditional uncaughtException/unhandledRejection crash contract, so posthog-node must never + // install its OWN competing listeners for either event (see posthog.ts's own comment for the full "why"). + expect(options.enableExceptionAutocapture).toBe(false); expect(options.before_send).toBe(scrubPostHogEvent); }); diff --git a/test/unit/selfhost-process-lifecycle.test.ts b/test/unit/selfhost-process-lifecycle.test.ts new file mode 100644 index 0000000000..39c263c6d1 --- /dev/null +++ b/test/unit/selfhost-process-lifecycle.test.ts @@ -0,0 +1,178 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { + installSelfHostCrashHandlers, + resetSelfHostCrashHandlersForTest, + type ProcessLike, +} from "../../src/selfhost/process-lifecycle"; + +type Listener = (...args: unknown[]) => void; + +/** A fake `process` that captures the last-registered listener per event so tests can invoke it directly -- + * mirrors packages/loopover-miner/lib/process-lifecycle.ts's own test helper (miner-process-lifecycle.test.ts). */ +function makeFakeProcess() { + const handlers = new Map(); + const exit = vi.fn(); + const proc: ProcessLike = { + on(event: string, listener: Listener) { + handlers.set(event, listener); + return proc; + }, + exit, + }; + return { proc, handlers, exit }; +} + +const SIGNAL_EVENTS = ["uncaughtException", "unhandledRejection"]; + +/** Run `fn`, then strip any listeners it added to the REAL process (only relevant to the default-process test). */ +function withRealProcessCleanup(fn: () => void) { + const before = new Map(SIGNAL_EVENTS.map((event) => [event, new Set(process.rawListeners(event))])); + try { + fn(); + } finally { + for (const event of SIGNAL_EVENTS) { + for (const listener of process.rawListeners(event)) { + if (!before.get(event)?.has(listener)) process.removeListener(event, listener as Listener); + } + } + } +} + +beforeEach(() => resetSelfHostCrashHandlersForTest()); +afterEach(() => { + resetSelfHostCrashHandlersForTest(); + vi.restoreAllMocks(); +}); + +describe("self-host process crash handlers (#9133)", () => { + it("installs uncaughtException/unhandledRejection once and reports whether it did", () => { + const { proc } = makeFakeProcess(); + expect(installSelfHostCrashHandlers({ process: proc, log: vi.fn(), exit: vi.fn() })).toBe(true); + // Already installed, no force -> no-op. + expect(installSelfHostCrashHandlers({ process: proc, log: vi.fn(), exit: vi.fn() })).toBe(false); + // force reinstalls. + expect(installSelfHostCrashHandlers({ process: proc, log: vi.fn(), exit: vi.fn(), force: true })).toBe(true); + }); + + it("logs a structured fatal line and exits 1 on an uncaught exception", async () => { + const { proc, handlers, exit } = makeFakeProcess(); + const log = vi.fn(); + installSelfHostCrashHandlers({ process: proc, log, exit }); + + const error = new Error("kaboom"); + await handlers.get("uncaughtException")?.(error); + + const logged = JSON.parse(log.mock.calls[0]?.[0] as string); + expect(logged).toMatchObject({ level: "fatal", event: "selfhost_uncaughtException" }); + expect(logged.error).toContain(error.stack); + expect(exit).toHaveBeenCalledWith(1); + }); + + // REGRESSION (#9133): this is the exact bug the issue fixes -- an unhandled rejection must terminate the + // process exactly like an uncaught exception does, independent of whether telemetry is configured. Before + // this fix, server.ts registered NO handler of its own for this event at all. + it("REGRESSION (#9133): logs a structured fatal line and exits 1 on an unhandled rejection, same as uncaughtException", async () => { + const { proc, handlers, exit } = makeFakeProcess(); + const log = vi.fn(); + installSelfHostCrashHandlers({ process: proc, log, exit }); + + await handlers.get("unhandledRejection")?.("a rejected reason, not necessarily an Error"); + + const logged = JSON.parse(log.mock.calls[0]?.[0] as string); + expect(logged).toMatchObject({ + level: "fatal", + event: "selfhost_unhandledRejection", + error: "a rejected reason, not necessarily an Error", + }); + expect(exit).toHaveBeenCalledWith(1); + }); + + it("falls back to an Error's message when it has no stack", async () => { + const { proc, handlers, exit } = makeFakeProcess(); + const log = vi.fn(); + installSelfHostCrashHandlers({ process: proc, log, exit }); + + const error = new Error("stackless"); + Object.defineProperty(error, "stack", { value: undefined }); + await handlers.get("uncaughtException")?.(error); + + const logged = JSON.parse(log.mock.calls[0]?.[0] as string); + expect(logged.error).toBe("stackless"); + }); + + it("calls the injected captureError with the error and a kind tag for both event types (telemetry configured)", async () => { + const { proc, handlers } = makeFakeProcess(); + const captureError = vi.fn(); + installSelfHostCrashHandlers({ process: proc, log: vi.fn(), exit: vi.fn(), captureError }); + + const error = new Error("kaboom"); + await handlers.get("uncaughtException")?.(error); + expect(captureError).toHaveBeenCalledWith(error, { kind: "uncaughtException" }); + + await handlers.get("unhandledRejection")?.("plain reason"); + expect(captureError).toHaveBeenCalledWith("plain reason", { kind: "unhandledRejection" }); + }); + + it("defaults captureError and flush to no-ops when telemetry is NOT configured, and still exits 1 (the other arm)", async () => { + const { proc, handlers, exit } = makeFakeProcess(); + installSelfHostCrashHandlers({ process: proc, log: vi.fn(), exit }); + + await expect(handlers.get("uncaughtException")?.(new Error("no telemetry"))).resolves.toBeUndefined(); + expect(exit).toHaveBeenCalledWith(1); + + exit.mockClear(); + await expect(handlers.get("unhandledRejection")?.("no telemetry either")).resolves.toBeUndefined(); + expect(exit).toHaveBeenCalledWith(1); + }); + + it("AWAITS flush before exiting -- exit() must not fire while telemetry is still draining", async () => { + const { proc, handlers, exit } = makeFakeProcess(); + let resolveFlush: () => void = () => {}; + const flushPending = new Promise((resolve) => { + resolveFlush = resolve; + }); + const flush = vi.fn(() => flushPending); + installSelfHostCrashHandlers({ process: proc, log: vi.fn(), flush }); + + const handled = handlers.get("uncaughtException")?.(new Error("kaboom")); + await Promise.resolve(); // let the handler's synchronous-until-await portion run + expect(exit).not.toHaveBeenCalled(); // flush() has not resolved yet + + resolveFlush(); + await handled; + expect(exit).toHaveBeenCalledWith(1); // only fires once the awaited flush actually resolved + }); + + it("actually invokes console.error as the default log sink when none is injected", async () => { + const { proc, handlers, exit } = makeFakeProcess(); + const errorSpy = vi.spyOn(console, "error").mockImplementation(() => undefined); + installSelfHostCrashHandlers({ process: proc, exit }); + + await handlers.get("uncaughtException")?.(new Error("default log sink")); + + expect(errorSpy).toHaveBeenCalledWith(expect.stringContaining("default log sink")); + errorSpy.mockRestore(); + }); + + it("uses console.error as the default log sink and process.exit as the default exit", () => { + withRealProcessCleanup(() => { + const errorSpy = vi.spyOn(console, "error").mockImplementation(() => undefined); + expect(installSelfHostCrashHandlers({ force: true })).toBe(true); + for (const event of SIGNAL_EVENTS) { + expect(process.rawListeners(event).length).toBeGreaterThan(0); + } + errorSpy.mockRestore(); + }); + }); + + it("truncates a very long error description to 4000 chars so a runaway stack can't blow out the log line", async () => { + const { proc, handlers } = makeFakeProcess(); + const log = vi.fn(); + installSelfHostCrashHandlers({ process: proc, log, exit: vi.fn() }); + + await handlers.get("unhandledRejection")?.("x".repeat(10_000)); + + const logged = JSON.parse(log.mock.calls[0]?.[0] as string); + expect(logged.error.length).toBe(4000); + }); +}); diff --git a/test/unit/selfhost-sqlite-queue.test.ts b/test/unit/selfhost-sqlite-queue.test.ts index 291191f606..3a0bdfdfef 100644 --- a/test/unit/selfhost-sqlite-queue.test.ts +++ b/test/unit/selfhost-sqlite-queue.test.ts @@ -1986,6 +1986,120 @@ describe("createSqliteQueue (durable #980)", () => { }); }); + // #9139/#9136: nothing wrote review_audit's `dead_lettered` event type before this fix, so ops.ts's DLQ + // anomaly signal was permanently 0. Scoped to jobs with an identifiable repo+PR (regateJob); a plain + // no-context payload (msg()) has nothing sensible to attribute a review dead-letter to. + describe("review_audit dead-letter recording (#9139/#9136)", () => { + afterEach(() => { + delete process.env.GITHUB_APP_SLUG; + }); + + it("writes a dead_lettered review_audit row (default 'loopover' project) when the payload has repo+PR context", async () => { + const driver = makeDriver(); + const spy = vi.spyOn(driver, "query"); + const q = createSqliteQueue(driver, async () => { + throw new Error("ai review failed"); + }, { maxRetries: 1, backoffMs: () => 0 }); + await q.binding.send(regateJob(null, 42)); + await q.drain(); + + const call = spy.mock.calls.find((c) => String(c[0]).includes("INSERT INTO review_audit")); + expect(call).toBeDefined(); + expect(call![1]).toEqual([ + expect.any(String), + "loopover", + "jsonbored/gittensory#42", + expect.stringContaining("ai review failed"), + expect.any(String), + ]); + }); + + it("uses GITHUB_APP_SLUG as the project when set (the other ?? arm)", async () => { + process.env.GITHUB_APP_SLUG = "custom-slug"; + const driver = makeDriver(); + const spy = vi.spyOn(driver, "query"); + const q = createSqliteQueue(driver, async () => { + throw new Error("boom"); + }, { maxRetries: 1, backoffMs: () => 0 }); + await q.binding.send(regateJob(null, 1)); + await q.drain(); + + const call = spy.mock.calls.find((c) => String(c[0]).includes("INSERT INTO review_audit")); + expect(call![1]).toEqual(expect.arrayContaining(["custom-slug"])); + }); + + it("does not write a review_audit row when the payload has no repo/PR context", async () => { + const driver = makeDriver(); + const spy = vi.spyOn(driver, "query"); + const q = createSqliteQueue(driver, async () => { + throw new Error("boom"); + }, { maxRetries: 1, backoffMs: () => 0 }); + await q.binding.send(msg("orb-export")); + await q.drain(); + + expect(spy.mock.calls.some((c) => String(c[0]).includes("INSERT INTO review_audit"))).toBe(false); + expect(await q.deadCount()).toBe(1); // the job itself still dead-lettered normally + }); + + it("swallows a review_audit write failure without breaking the job's own dead-letter transition", async () => { + const driver = makeDriver(); + const realQuery = driver.query.bind(driver); + vi.spyOn(driver, "query").mockImplementation((sql: string, params: unknown[]) => { + if (String(sql).includes("INSERT INTO review_audit")) throw new Error("no such table: review_audit"); + return realQuery(sql, params); + }); + const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => undefined); + const q = createSqliteQueue(driver, async () => { + throw new Error("boom"); + }, { maxRetries: 1, backoffMs: () => 0 }); + await q.binding.send(regateJob(null, 7)); + await q.drain(); + + expect(await q.deadCount()).toBe(1); // the job's own dead-letter transition still succeeded + expect(warnSpy.mock.calls.some((c) => String(c[0]).includes("review_audit_dead_letter_record_error"))).toBe(true); + warnSpy.mockRestore(); + }); + }); + + // #9139: backs loopover_dlq_dead_lettered_recent on self-host -- the cloud-worker audit_events source is + // structurally unreachable there (see dlq-recent.ts's own doc comment). + describe("recentDeadCount (#9139)", () => { + it("counts a job dead-lettered within the trailing window", async () => { + const driver = makeDriver(); + const q = createSqliteQueue(driver, async () => { + throw new Error("boom"); + }, { maxRetries: 1, backoffMs: () => 0 }); + await q.binding.send(msg("x")); + await q.drain(); + expect(await q.deadCount()).toBe(1); + + expect(await q.recentDeadCount(15 * 60 * 1000)).toBe(1); + }); + + it("excludes a job whose dead_at falls OUTSIDE the trailing window (the boundary arm)", async () => { + const driver = makeDriver(); + const q = createSqliteQueue(driver, async () => { + throw new Error("boom"); + }, { maxRetries: 1, backoffMs: () => 0 }); + await q.binding.send(msg("x")); + await q.drain(); + expect(await q.deadCount()).toBe(1); // still dead in the STANDING count... + + // ...but backdate dead_at well outside a 15m window -- recentDeadCount is a RATE-style window, not + // deadCount's unbounded depth, so an old dead-letter must stop counting here even though it's still + // sitting in the table. + driver.query("UPDATE _selfhost_jobs SET dead_at = ? WHERE status = 'dead'", [Date.now() - 20 * 60 * 1000]); + + expect(await q.recentDeadCount(15 * 60 * 1000)).toBe(0); + }); + + it("reads 0 when nothing has ever dead-lettered (no rows to match at all)", async () => { + const driver = makeDriver(); + const q = createSqliteQueue(driver, async () => undefined); + expect(await q.recentDeadCount(15 * 60 * 1000)).toBe(0); + }); + }); + it("retries then dead-letters after maxRetries", async () => { const driver = makeDriver(); let calls = 0;