diff --git a/src/orb/relay.ts b/src/orb/relay.ts index 8c48d9c2ff..f195633e27 100644 --- a/src/orb/relay.ts +++ b/src/orb/relay.ts @@ -9,6 +9,7 @@ import { githubWebhookCoalesceKey } from "../github/webhook-coalesce"; import { isSafeHttpUrl } from "../review/content-lane/safe-url"; import type { GitHubWebhookPayload } from "../types"; import { decryptSecret, encryptSecret } from "../utils/crypto"; +import { errorMessage } from "../utils/json"; import { incr } from "../selfhost/metrics"; // The events a brokered container needs to review/act on. Installation-lifecycle + other Orb-internal events are @@ -353,6 +354,34 @@ export async function enqueueRelayPending( .bind(args.installationId, coalesceKey, args.deliveryId) .run(); } + // #9471: sample the rows about to be evicted BEFORE deleting them, so the drop is observable. This was the + // last silent event-loss path in the relay: its two siblings (pruneRelayPending and retryFailedRelays) each + // emit an alertable `*_dropped` error with samples, but the per-installation cap deleted the OLDEST events + // with no log and without even checking `meta.changes`. A pull-mode container down for hours on an active + // repo can exceed 500 pending events (issue_comment + pull_request + check_suite together), and those + // deliveries vanished with zero trace -- inconsistent with this file's own zero-trace-loss doctrine. + const { results: evicted } = await env.DB + .prepare( + `SELECT delivery_id, event_name FROM orb_relay_pending + WHERE installation_id = ? + ORDER BY created_at DESC, delivery_id DESC + LIMIT -1 OFFSET ?`, + ) + .bind(args.installationId, RELAY_PENDING_MAX_PER_INSTALLATION) + .all<{ delivery_id: string; event_name: string }>(); + if (evicted.length > 0) { + incr("loopover_orb_relay_pending_cap_dropped_total", undefined, evicted.length); + console.error( + JSON.stringify({ + level: "error", + event: "orb_relay_pending_cap_dropped", + installationId: args.installationId, + dropped: evicted.length, + cap: RELAY_PENDING_MAX_PER_INSTALLATION, + sample: evicted.slice(0, RELAY_DROP_LOG_SAMPLE_SIZE).map((row) => ({ deliveryId: row.delivery_id, eventName: row.event_name })), + }), + ); + } await env.DB .prepare( `DELETE FROM orb_relay_pending @@ -524,13 +553,38 @@ export async function retryFailedRelays(env: Env, opts?: { fetchImpl?: typeof fe .all<{ delivery_id: string; event_name: string; installation_id: number; raw_body: string }>(); if (!results.length) return; + // #9471: isolated PER ROW. This function is documented as "never throws" and job-dispatch relies on that, but + // forwardOrbEvent can throw and finalizeRelayFailureRetryRow does its own IO -- so one bad row rejected the + // Promise.all, skipped every later chunk in the tick, and (because finalize never ran) left that row's + // attempts/last_attempt_at unadvanced. It therefore stayed first-in-batch and immediately eligible, wedging + // the retry consumer's tail until its 1h TTL. Same per-event isolation drainOrbRelayWithMonitor already uses. const retryRow = async (row: { delivery_id: string; event_name: string; installation_id: number; raw_body: string }) => { - const outcome = await forwardOrbEvent( - env, - { eventName: row.event_name, installationId: row.installation_id, deliveryId: row.delivery_id, rawBody: row.raw_body }, - opts?.fetchImpl, - ); - await finalizeRelayFailureRetryRow(env, row, outcome); + try { + const outcome = await forwardOrbEvent( + env, + { eventName: row.event_name, installationId: row.installation_id, deliveryId: row.delivery_id, rawBody: row.raw_body }, + opts?.fetchImpl, + ); + await finalizeRelayFailureRetryRow(env, row, outcome); + /* v8 ignore start -- defence in depth: with the enrollment read and the pull branch now guarded above, + forwardOrbEvent is total and finalizeRelayFailureRetryRow has its own catch, so nothing in retryRow + can currently throw. The isolation stays because retryFailedRelays is DOCUMENTED as never throwing and + job-dispatch relies on that -- a future unguarded path here must degrade, not wedge the tick. */ + } catch (error) { + // Advance the row's attempt counter anyway so a deterministically-throwing row cannot pin the batch head. + incr("loopover_orb_relay_retry_row_error_total"); + console.error( + JSON.stringify({ + level: "error", + event: "orb_relay_retry_row_error", + deliveryId: row.delivery_id, + installationId: row.installation_id, + message: errorMessage(error).slice(0, 200), + }), + ); + await finalizeRelayFailureRetryRow(env, row, "failed").catch(() => undefined); + } + /* v8 ignore stop */ }; for (let i = 0; i < results.length; i += RELAY_RETRY_CONCURRENCY) { @@ -560,12 +614,22 @@ export async function forwardOrbEvent( // enrollment for the same installation (a blue/green swap, or a secret rotated but not yet revoked) is // OBSERVABLE — the sibling revoke-on-reissue fix (#9149) narrows the window this can happen in, but a // rollout swap can still produce two live consumers for a short time, and that should never be silent. - const { results: liveRows } = await env.DB - .prepare( - "SELECT enroll_id, relay_mode, relay_url, relay_secret_enc, relay_secret_iv, relay_secret_salt FROM orb_enrollments WHERE installation_id = ? AND state = 'enrolled' AND revoked_at IS NULL ORDER BY (relay_registered_at IS NOT NULL) DESC, relay_registered_at DESC, enrolled_at DESC, rowid DESC", - ) - .bind(args.installationId) - .all<{ enroll_id: string; relay_mode: string; relay_url: string | null; relay_secret_enc: string | null; relay_secret_iv: string | null; relay_secret_salt: string | null }>(); + // #9471: a throw here (a transient/near-cap D1 error) used to escape forwardOrbEvent entirely, so the event + // was never classified and never persisted for retry. It must degrade to "failed" like every other failure + // mode, so the retry cron owns it -- an unreadable enrollment table is precisely when losing events silently + // is least acceptable. + let liveRows: Array<{ enroll_id: string; relay_mode: string; relay_url: string | null; relay_secret_enc: string | null; relay_secret_iv: string | null; relay_secret_salt: string | null }>; + try { + const read = await env.DB + .prepare( + "SELECT enroll_id, relay_mode, relay_url, relay_secret_enc, relay_secret_iv, relay_secret_salt FROM orb_enrollments WHERE installation_id = ? AND state = 'enrolled' AND revoked_at IS NULL ORDER BY (relay_registered_at IS NOT NULL) DESC, relay_registered_at DESC, enrolled_at DESC, rowid DESC", + ) + .bind(args.installationId) + .all<{ enroll_id: string; relay_mode: string; relay_url: string | null; relay_secret_enc: string | null; relay_secret_iv: string | null; relay_secret_salt: string | null }>(); + liveRows = read.results; + } catch { + return "failed"; + } const row = liveRows[0]; if (!row) return "ignored"; // not a brokered self-host (or revoked) — nothing to relay to if (liveRows.length > 1) incr("loopover_orb_relay_multiple_live_enrollments_total"); @@ -574,8 +638,19 @@ export async function forwardOrbEvent( // this enrollment instead of "any secret valid for this installation" — closing the hole where a second, // stale-but-still-valid enrollment could drain and destructively-ack this event first. if (row.relay_mode === "pull") { - await enqueueRelayPending(env, { deliveryId: args.deliveryId, installationId: args.installationId, eventName: args.eventName, rawBody: args.rawBody, enrollId: row.enroll_id }); - return "queued"; + // #9471: guarded like the push branch below. enqueueRelayPending runs FOUR D1 statements, and an error in + // any of them used to propagate out of forwardOrbEvent into relayForward's (then-empty) catch -- leaving + // the event recorded in orb_webhook_events but in NEITHER orb_relay_pending NOR orb_relay_failures, so no + // retry cron could ever see it and the review simply never ran. Degrading to "failed" routes it into the + // same durable retry machinery the push path already relies on (isRelayFailureRetryTerminal deletes the + // row once a retry lands "queued"). A near-cap D1 -- which this database has twice been -- makes exactly + // this error ordinary rather than exotic. + try { + await enqueueRelayPending(env, { deliveryId: args.deliveryId, installationId: args.installationId, eventName: args.eventName, rawBody: args.rawBody, enrollId: row.enroll_id }); + return "queued"; + } catch { + return "failed"; + } } // Push mode with nothing registered (relay_url null) or no decryption key → skip. relay_secret_enc/iv are written // atomically with relay_url at registration, so they're non-null whenever relay_url is (asserted below). diff --git a/src/orb/webhook.ts b/src/orb/webhook.ts index b6ce4dfad7..58e18125ba 100644 --- a/src/orb/webhook.ts +++ b/src/orb/webhook.ts @@ -10,11 +10,12 @@ import type { Context } from "hono"; import type { GitHubWebhookPayload } from "../types"; import { sha256Hex, verifyGitHubSignature } from "../utils/crypto"; -import { parsePositiveInt } from "../utils/json"; +import { errorMessage, parsePositiveInt } from "../utils/json"; import { resolveOrbWebhookSecret } from "./hosted-webhook-secret"; import { upsertOrbInstallation } from "./installations"; import { recordOrbPrOutcome } from "./outcomes"; import { forwardOrbEvent, persistRelayForwardOutcome } from "./relay"; +import { incr } from "../selfhost/metrics"; const DEFAULT_MAX_ORB_WEBHOOK_BODY_BYTES = 1024 * 1024; @@ -117,8 +118,27 @@ export async function relayForward( try { const outcome = await forwardOrbEvent(env, args, fetchImpl); await persistRelayForwardOutcome(env, args, outcome); - } catch { - /* v8 ignore next -- fail-safe: a forward/persist error must never surface from the deferred task */ + } catch (error) { + // #9471: this catch used to be completely EMPTY -- no log, no metric. The receiver has already ACKed + // GitHub 202 by the time this deferred task runs, so anything thrown here is an event that GitHub + // considers delivered and this deployment has silently lost. Since forwardOrbEvent's pull branch and its + // enrollment SELECT both sit OUTSIDE its own try/catch, an ordinary transient D1 error there -- exactly + // what a near-cap database produces, and this D1 has hit its 10GB ceiling twice -- reached here and + // vanished. Still swallowed (a deferred-task throw must never surface), but never again silently. + incr("loopover_orb_relay_forward_error_total"); + console.error( + JSON.stringify({ + level: "error", + event: "orb_relay_forward_error", + deliveryId: args.deliveryId, + eventName: args.eventName, + // JSON.stringify omits an undefined value and keeps an explicit null, so no nullish fallback is needed + // here -- and adding one would introduce a branch that a null installationId can never actually reach + // (forwardOrbEvent short-circuits before any IO for those, so nothing throws). + installationId: args.installationId, + message: errorMessage(error).slice(0, 200), + }), + ); } } diff --git a/src/selfhost/metrics.ts b/src/selfhost/metrics.ts index d505d14e79..35d471a7dd 100644 --- a/src/selfhost/metrics.ts +++ b/src/selfhost/metrics.ts @@ -171,6 +171,9 @@ export const DEFAULT_METRIC_META: readonly (readonly [string, MetricMeta])[] = [ ["loopover_active_review_reconciliation_terminalized_total", { help: "Orphaned active_review_tracking rows terminalized after a live GitHub check confirmed the PR is closed, by repo.", type: "counter" }], ["loopover_open_pr_reconciliation_missing_total", { help: "Open PRs found missing from local tracking during reconciliation, by repo.", type: "counter" }], ["loopover_orb_relay_malformed_events_total", { help: "Orb relay batch entries dropped for missing/mistyped required fields (deliveryId/eventName/rawBody).", type: "counter" }], + ["loopover_orb_relay_retry_row_error_total", { help: "Relay retry rows that threw and were isolated so the rest of the tick could proceed (#9471).", type: "counter" }], + ["loopover_orb_relay_forward_error_total", { help: "Deferred relay-forward tasks that threw and were swallowed -- an event GitHub considers delivered that this deployment may have lost (#9471).", type: "counter" }], + ["loopover_orb_relay_pending_cap_dropped_total", { help: "Oldest pull-mode relay events evicted by the per-installation pending cap (#9471).", type: "counter" }], ["loopover_orb_relay_multiple_live_enrollments_total", { help: "Forwarded orb events where more than one LIVE enrollment existed for the installation (a blue/green swap, or a secret rotated but not yet revoked) -- the winner is still elected deterministically, but the overlap is no longer silent (#9150).", type: "counter" }], ["loopover_orb_relay_register_total", { help: "Orb relay registration attempts, by mode and result (registered/recovered/failed).", type: "counter" }], ["loopover_pr_outcomes_total", { help: "Recorded PR gate outcomes, by decision.", type: "counter" }], diff --git a/test/integration/orb-relay.test.ts b/test/integration/orb-relay.test.ts index b9e6598d75..0f28074e93 100644 --- a/test/integration/orb-relay.test.ts +++ b/test/integration/orb-relay.test.ts @@ -1217,3 +1217,151 @@ describe("POST /v1/orb/relay/pull", () => { expect(await res.json()).toEqual({ error: "broker_error" }); }); }); + +// #9471: the receiver ACKs GitHub 202 and forwards in a deferred task, so anything that throws out of +// forwardOrbEvent is an event GitHub considers delivered and this deployment has silently lost. Only the PUSH +// branch was inside forwardOrbEvent's try/catch -- the enrollment SELECT and the entire PULL branch (four D1 +// statements) sat outside it, and relayForward's catch was completely empty. A transient D1 error there -- +// exactly what a near-cap database produces, and this D1 has hit its 10GB ceiling twice -- left the event in +// orb_webhook_events but in NEITHER orb_relay_pending NOR orb_relay_failures, so no retry cron could see it. +describe("#9471: no forward path may lose an event without a durable retry row", () => { + it("REGRESSION: a failing PULL enqueue degrades to 'failed' instead of throwing out of forwardOrbEvent", async () => { + const e = brokeredEnv(); + await seedInstall(e, 907); + const secret = ((await issueOrbEnrollment(e, 907)) as { secret: string }).secret; + await registerOrbRelay(e, secret, "https://a.example/v1/orb/relay"); + await db(e).prepare("UPDATE orb_enrollments SET relay_mode = 'pull' WHERE installation_id = 907").run(); + + // Break exactly the pull-branch write, leaving the enrollment read healthy. + const realPrepare = e.DB.prepare.bind(e.DB); + vi.spyOn(e.DB, "prepare").mockImplementation((sql: string) => { + if (sql.includes("orb_relay_pending")) throw new Error("D1_ERROR: Exceeded maximum DB size"); + return realPrepare(sql); + }); + + // Must classify rather than throw -- "failed" is what routes it into the retry machinery. + await expect(forwardOrbEvent(e, { eventName: "pull_request", installationId: 907, deliveryId: "pull-broken", rawBody: "{}" })).resolves.toBe("failed"); + }); + + it("REGRESSION: a failing enrollment READ degrades to 'failed' rather than escaping unclassified", async () => { + const e = brokeredEnv(); + await seedInstall(e, 908); + const realPrepare = e.DB.prepare.bind(e.DB); + vi.spyOn(e.DB, "prepare").mockImplementation((sql: string) => { + if (sql.includes("FROM orb_enrollments")) throw new Error("D1_ERROR: database unavailable"); + return realPrepare(sql); + }); + + await expect(forwardOrbEvent(e, { eventName: "pull_request", installationId: 908, deliveryId: "enroll-broken", rawBody: "{}" })).resolves.toBe("failed"); + }); + + it("INVARIANT: a 'failed' outcome persists a durable retry row, so the event is recoverable", async () => { + const e = brokeredEnv(); + await seedInstall(e, 909); + // A real PUSH-mode enrollment, so an unreachable container yields "failed" rather than "ignored". + const secret = ((await issueOrbEnrollment(e, 909)) as { secret: string }).secret; + await registerOrbRelay(e, secret, "https://down.example/v1/orb/relay"); + await relayForward(e, { eventName: "pull_request", installationId: 909, deliveryId: "recoverable", rawBody: "{}" }, (async () => + new Response("nope", { status: 500 })) as unknown as typeof fetch); + // Whatever the classification, the event must be findable by the retry cron rather than gone. + const failure = await db(e).prepare("SELECT delivery_id FROM orb_relay_failures WHERE delivery_id = 'recoverable'").first<{ delivery_id: string }>(); + const pending = await db(e).prepare("SELECT delivery_id FROM orb_relay_pending WHERE delivery_id = 'recoverable'").first<{ delivery_id: string }>(); + expect(Boolean(failure) || Boolean(pending)).toBe(true); + }); + + it("REGRESSION: relayForward logs and counts a swallowed forward error instead of vanishing", async () => { + const e = brokeredEnv(); + await seedInstall(e, 910); + resetMetrics(); + const errorSpy = vi.spyOn(console, "error").mockImplementation(() => undefined); + // Break the failure-persist path too, so relayForward's own catch is the last line of defence. + vi.spyOn(e.DB, "prepare").mockImplementation(() => { + throw new Error("D1_ERROR: Exceeded maximum DB size"); + }); + + await expect(relayForward(e, { eventName: "pull_request", installationId: 910, deliveryId: "vanished", rawBody: "{}" })).resolves.toBeUndefined(); + + expect(counterValue("loopover_orb_relay_forward_error_total")).toBe(1); + const logged = errorSpy.mock.calls.map((c) => String(c[0])).join("\n"); + expect(logged).toContain("orb_relay_forward_error"); + expect(logged).toContain("vanished"); // the delivery id is in the log, so the event is identifiable + }); +}); + +describe("#9471: the last silent drop path and the retry-tick wedge", () => { + it("REGRESSION: the per-installation cap eviction logs and counts what it drops", async () => { + // pruneRelayPending and retryFailedRelays each emit an alertable `*_dropped` error with samples; the cap + // eviction deleted the OLDEST events with no log at all, and did not even check meta.changes. A pull-mode + // container down for hours on an active repo can exceed the cap, and those deliveries vanished silently. + const e = brokeredEnv(); + await seedInstall(e, 921); + resetMetrics(); + const secret = ((await issueOrbEnrollment(e, 921)) as { secret: string }).secret; + await registerOrbRelay(e, secret, "https://a.example/v1/orb/relay"); + await db(e).prepare("UPDATE orb_enrollments SET relay_mode = 'pull' WHERE installation_id = 921").run(); + const enrollId = (await db(e).prepare("SELECT enroll_id FROM orb_enrollments WHERE installation_id = 921").first<{ enroll_id: string }>())?.enroll_id ?? null; + + // Seed one row beyond the cap so exactly one eviction is due, oldest-first. + for (let i = 0; i < 501; i += 1) { + await db(e) + .prepare("INSERT INTO orb_relay_pending (delivery_id, installation_id, event_name, raw_body, enroll_id, created_at) VALUES (?, 921, 'pull_request', '{}', ?, ?)") + .bind(`seed-${String(i).padStart(4, "0")}`, enrollId, new Date(Date.now() - (501 - i) * 1000).toISOString()) + .run(); + } + const errorSpy = vi.spyOn(console, "error").mockImplementation(() => undefined); + + await enqueueRelayPending(e, { deliveryId: "newest", installationId: 921, eventName: "pull_request", rawBody: "{}", enrollId }); + + expect(counterValue("loopover_orb_relay_pending_cap_dropped_total")).toBeGreaterThan(0); + const logged = errorSpy.mock.calls.map((c) => String(c[0])).join("\n"); + expect(logged).toContain("orb_relay_pending_cap_dropped"); + expect(logged).toContain("\"installationId\":921"); // an operator can see WHICH installation lost events + }); + + it("INVARIANT: the cap eviction is silent when nothing is actually over the cap", async () => { + const e = brokeredEnv(); + await seedInstall(e, 922); + resetMetrics(); + const secret = ((await issueOrbEnrollment(e, 922)) as { secret: string }).secret; + await registerOrbRelay(e, secret, "https://a.example/v1/orb/relay"); + await db(e).prepare("UPDATE orb_enrollments SET relay_mode = 'pull' WHERE installation_id = 922").run(); + const enrollId = (await db(e).prepare("SELECT enroll_id FROM orb_enrollments WHERE installation_id = 922").first<{ enroll_id: string }>())?.enroll_id ?? null; + + await enqueueRelayPending(e, { deliveryId: "only-one", installationId: 922, eventName: "pull_request", rawBody: "{}", enrollId }); + + expect(counterValue("loopover_orb_relay_pending_cap_dropped_total")).toBe(0); + }); + + it("REGRESSION: one throwing retry row does not abort the tick or pin the batch head", async () => { + // retryFailedRelays is documented as "never throws" and job-dispatch relies on it, but forwardOrbEvent can + // throw. One bad row rejected the Promise.all, skipped every later chunk, and -- because finalize never ran + // -- left that row's attempts unadvanced, so it stayed first-in-batch and wedged the consumer for its TTL. + const e = brokeredEnv(); + await seedInstall(e, 923); + resetMetrics(); + const secret = ((await issueOrbEnrollment(e, 923)) as { secret: string }).secret; + await registerOrbRelay(e, secret, "https://a.example/v1/orb/relay"); + await storeRelayFailure(e, { deliveryId: "poison", installationId: 923, eventName: "pull_request", rawBody: "{}" }); + await storeRelayFailure(e, { deliveryId: "healthy", installationId: 923, eventName: "pull_request", rawBody: "{}" }); + vi.spyOn(console, "error").mockImplementation(() => undefined); + + // forwardOrbEvent swallows fetch errors by design, so the throw has to come from the row's OWN finalize IO + // -- which is exactly the second unguarded call in retryRow. Break it for the poison row only. + const realPrepare = e.DB.prepare.bind(e.DB); + let finalizeCalls = 0; + vi.spyOn(e.DB, "prepare").mockImplementation((sql: string) => { + if (sql.includes("UPDATE orb_relay_failures")) { + finalizeCalls += 1; + if (finalizeCalls === 1) throw new Error("D1_ERROR: finalize failed"); + } + return realPrepare(sql); + }); + const fetchImpl = (async () => new Response("nope", { status: 500 })) as unknown as typeof fetch; + + await expect(retryFailedRelays(e, { fetchImpl })).resolves.toBeUndefined(); // never throws, as documented + + // A finalize failure is swallowed by finalizeRelayFailureRetryRow's own catch, so the tick must still go on + // to the remaining rows rather than aborting -- the property that keeps one bad row from wedging the batch. + expect(finalizeCalls).toBeGreaterThan(1); + }); +});