From 73f9987f9209cb0924d45b933389653249b5eefd Mon Sep 17 00:00:00 2001 From: JSONbored <49853598+JSONbored@users.noreply.github.com> Date: Mon, 6 Jul 2026 07:38:08 -0700 Subject: [PATCH] fix(selfhost): isolate per-event enqueue failures in the Orb relay drain loop (#3813) drainOrbRelayWithMonitor called args.enqueue() with no try/catch around it. enqueueWebhookByEnv swallows its own anticipated failures and returns a string result, but a D1/Postgres write failure inside one of its two unwrapped recordWebhookEvent calls throws uncaught -- aborting the entire remaining batch for that drain tick. Every event after the failing one was silently never attempted. Wrap the enqueue call per event: a throw is now treated exactly like the existing non-throwing "enqueue_failed" result -- logged, counted, and NOT acked (so the relay redelivers it), with the loop continuing to the next event instead of aborting the batch. --- src/selfhost/monitored-work.ts | 34 ++++++++++++--- test/unit/selfhost-monitored-work.test.ts | 50 +++++++++++++++++++++++ 2 files changed, 78 insertions(+), 6 deletions(-) diff --git a/src/selfhost/monitored-work.ts b/src/selfhost/monitored-work.ts index 682167d000..a26a87b540 100644 --- a/src/selfhost/monitored-work.ts +++ b/src/selfhost/monitored-work.ts @@ -86,12 +86,34 @@ export async function drainOrbRelayWithMonitor(args: { result: events.length > 0 ? "events" : "empty", }); for (const ev of events) { - const result = await args.enqueue( - args.env, - ev.deliveryId, - ev.eventName, - ev.rawBody, - ); + // #audit-orb-relay-enqueue-isolation: an enqueue can throw uncaught (e.g. a D1/Postgres write failure + // inside recordWebhookEvent, not just the anticipated failures enqueueWebhookByEnv already returns as a + // string result) -- that must not abort the REST of this batch, or every event after the failing one + // is silently never attempted this tick. Isolate per event and treat a throw exactly like the existing + // non-throwing "enqueue_failed" result: don't ack (the relay redelivers it next drain) and keep going. + let result: EnqueueWebhookResult; + try { + result = await args.enqueue( + args.env, + ev.deliveryId, + ev.eventName, + ev.rawBody, + ); + } catch (error) { + incr("gittensory_orb_webhook_total", { + event: orbRelayMetricEvent(ev.eventName), + result: "enqueue_failed", + }); + console.error( + JSON.stringify({ + level: "error", + event: "orb_relay_enqueue_threw", + eventName: ev.eventName, + error: error instanceof Error ? error.message : String(error), + }), + ); + continue; + } incr("gittensory_orb_webhook_total", { event: orbRelayMetricEvent(ev.eventName), result, diff --git a/test/unit/selfhost-monitored-work.test.ts b/test/unit/selfhost-monitored-work.test.ts index 1643d4ba8d..adccf01535 100644 --- a/test/unit/selfhost-monitored-work.test.ts +++ b/test/unit/selfhost-monitored-work.test.ts @@ -148,6 +148,56 @@ describe("self-host monitored recurring work", () => { expect(metrics).toContain('gittensory_orb_webhook_total{event="check_suite",result="duplicate"} 1'); }); + it("REGRESSION (#audit-orb-relay-enqueue-isolation): an enqueue that throws for one event does not abort the rest of the batch", async () => { + const state: OrbRelayDrainState = { pendingAck: [], lastDrainAtMs: null }; + const drain = vi.fn().mockResolvedValue([ + { deliveryId: "ok-1", eventName: "pull_request", rawBody: "{}" }, + { deliveryId: "throws-2", eventName: "issues", rawBody: "{}" }, + { deliveryId: "ok-3", eventName: "check_suite", rawBody: "{}" }, + ]); + const enqueue = vi + .fn() + .mockResolvedValueOnce("queued") + .mockRejectedValueOnce(new Error("D1 write error")) + .mockResolvedValueOnce("queued"); + const errors = vi.spyOn(console, "error").mockImplementation(() => undefined); + + await drainOrbRelayWithMonitor({ + state, + relayEnv: {}, + env: {} as Env, + drain, + enqueue, + }); + + // All 3 events were attempted -- ok-3 was still reached even though throws-2 (the 2nd) rejected. + expect(enqueue).toHaveBeenCalledTimes(3); + expect(enqueue).toHaveBeenNthCalledWith(3, {}, "ok-3", "check_suite", "{}"); + // throws-2 is NOT acked (the relay redelivers it next drain), but both successful events are. + expect(state.pendingAck).toEqual(["ok-1", "ok-3"]); + const logged = errors.mock.calls.map((c) => String(c[0])).find((line) => line.includes("orb_relay_enqueue_threw")); + expect(logged).toBeDefined(); + expect(JSON.parse(logged!)).toMatchObject({ level: "error", event: "orb_relay_enqueue_threw", eventName: "issues", error: "D1 write error" }); + const metrics = await renderMetrics(); + expect(metrics).toContain('gittensory_orb_webhook_total{event="pull_request",result="queued"} 1'); + expect(metrics).toContain('gittensory_orb_webhook_total{event="issues",result="enqueue_failed"} 1'); + expect(metrics).toContain('gittensory_orb_webhook_total{event="check_suite",result="queued"} 1'); + errors.mockRestore(); + }); + + it("logs a non-Error enqueue rejection by stringifying it (the false ternary arm)", async () => { + const state: OrbRelayDrainState = { pendingAck: [], lastDrainAtMs: null }; + const drain = vi.fn().mockResolvedValue([{ deliveryId: "throws-1", eventName: "pull_request", rawBody: "{}" }]); + const enqueue = vi.fn().mockRejectedValueOnce("not an Error instance"); + const errors = vi.spyOn(console, "error").mockImplementation(() => undefined); + + await drainOrbRelayWithMonitor({ state, relayEnv: {}, env: {} as Env, drain, enqueue }); + + const logged = errors.mock.calls.map((c) => String(c[0])).find((line) => line.includes("orb_relay_enqueue_threw")); + expect(JSON.parse(logged!)).toMatchObject({ error: "not an Error instance" }); + errors.mockRestore(); + }); + it("clears previous Orb relay acks and stays quiet when the broker has no events", async () => { const state: OrbRelayDrainState = { pendingAck: ["previous-delivery"], lastDrainAtMs: null }; const drain = vi.fn().mockResolvedValue([]);