diff --git a/src/db/repositories.ts b/src/db/repositories.ts index 86c423ea62..5c44e02949 100644 --- a/src/db/repositories.ts +++ b/src/db/repositories.ts @@ -4954,7 +4954,11 @@ export async function recordWebhookEvent( installationId?: number | undefined; repositoryFullName?: string | undefined; payloadHash: string; - status: "queued" | "processed" | "error"; + // "superseded": a coalescable delivery (e.g. a pr-refresh) whose queue row was overwritten by a later + // redelivery sharing the same job_key before either was claimed — written directly by the self-host queue + // backends (pg-queue.ts / sqlite-queue.ts) at coalesce time, not through this function, but included here so + // the full set of terminal statuses this column can hold is documented in one place (#audit-webhook-supersede-trace). + status: "queued" | "processed" | "error" | "superseded"; errorSummary?: string; }, ): Promise { diff --git a/src/selfhost/pg-queue.ts b/src/selfhost/pg-queue.ts index f2461bb725..9b9158d5e3 100644 --- a/src/selfhost/pg-queue.ts +++ b/src/selfhost/pg-queue.ts @@ -819,6 +819,38 @@ export function createPgQueue( return due.length; } + // #audit-webhook-supersede-trace: best-effort, never blocks a coalesce on a write hiccup -- the row it marks + // is purely an audit trace (webhook_events), not the actual job data, so a failure here must not resurrect the + // "abort the whole enqueue" class of bug this whole issue exists to close. `oldPayload` is the row's payload + // BEFORE it gets overwritten by the coalesce; `incomingMessage` is what it's about to become. Only a + // github-webhook delivery has a webhook_events row at all (rag-index-repo etc. never do), and only when the + // superseded id genuinely differs from the surviving one (defense-in-depth against a same-id no-op). + async function markSupersededWebhookEvent(oldPayload: string, incomingMessage: JobMessage): Promise { + let old: { type?: unknown; deliveryId?: unknown } | null; + try { + old = JSON.parse(oldPayload) as { type?: unknown; deliveryId?: unknown }; + } catch { + return; + } + if (old?.type !== "github-webhook" || typeof old.deliveryId !== "string") return; + const incomingDeliveryId = incomingMessage.type === "github-webhook" ? incomingMessage.deliveryId : undefined; + if (old.deliveryId === incomingDeliveryId) return; + try { + await pool.query( + `UPDATE webhook_events SET status='superseded', processed_at=$2 WHERE delivery_id=$1 AND status='queued'`, + [old.deliveryId, new Date().toISOString()], + ); + } catch (error) { + console.error( + JSON.stringify({ + level: "error", + event: "webhook_supersede_mark_failed", + error: errorMessageWithCause(error), + }), + ); + } + } + async function enqueue( message: JobMessage, delaySeconds: number, @@ -920,11 +952,17 @@ export function createPgQueue( if (key) { const existing = ( await pool.query( - `SELECT id FROM ${TABLE} WHERE status='pending' AND job_key=$1 ORDER BY priority DESC, run_after DESC, id LIMIT 1`, + `SELECT id, payload FROM ${TABLE} WHERE status='pending' AND job_key=$1 ORDER BY priority DESC, run_after DESC, id LIMIT 1`, [key], ) - ).rows[0] as { id: string } | undefined; + ).rows[0] as { id: string; payload: string } | undefined; if (existing) { + // #audit-webhook-supersede-trace: the row about to be overwritten below may itself be a github-webhook + // delivery (e.g. a "PR opened" pr-refresh coalesce) whose own webhook_events row was written as 'queued' + // BEFORE it ever reached this coalesce -- overwriting the payload here discards that delivery's id + // forever, so nothing would ever advance its webhook_events row past 'queued'. Mark it superseded FIRST, + // while the OLD payload (and its deliveryId) is still readable. + await markSupersededWebhookEvent(existing.payload, message); // See the supersededKeyPrefix branch above: created_at is preserved across a coalesced re-enqueue so the // maintenance trickle clock reflects genuine wait time, not the most recent re-request. await pool.query( diff --git a/src/selfhost/sqlite-queue.ts b/src/selfhost/sqlite-queue.ts index 9409bd6f1f..58279f3258 100644 --- a/src/selfhost/sqlite-queue.ts +++ b/src/selfhost/sqlite-queue.ts @@ -483,6 +483,38 @@ export function createSqliteQueue( } } + // #audit-webhook-supersede-trace: best-effort, never blocks a coalesce on a write hiccup -- the row it marks + // is purely an audit trace (webhook_events), not the actual job data, so a failure here must not resurrect the + // "abort the whole enqueue" class of bug this whole issue exists to close. `oldPayload` is the row's payload + // BEFORE it gets overwritten by the coalesce; `incomingMessage` is what it's about to become. Only a + // github-webhook delivery has a webhook_events row at all (rag-index-repo etc. never do), and only when the + // superseded id genuinely differs from the surviving one (defense-in-depth against a same-id no-op). + function markSupersededWebhookEvent(oldPayload: string, incomingMessage: JobMessage): void { + let old: { type?: unknown; deliveryId?: unknown } | null; + try { + old = JSON.parse(oldPayload) as { type?: unknown; deliveryId?: unknown }; + } catch { + return; + } + if (old?.type !== "github-webhook" || typeof old.deliveryId !== "string") return; + /* v8 ignore next -- defensive: jobCoalesceKey partitions its key format strictly by message.type, so a + * github-webhook job_key can only ever be matched by another github-webhook message; the non-webhook arm + * is unreachable through this call site, not load-bearing. */ + const incomingDeliveryId = incomingMessage.type === "github-webhook" ? incomingMessage.deliveryId : undefined; + if (old.deliveryId === incomingDeliveryId) return; + try { + driver.query(`UPDATE webhook_events SET status='superseded', processed_at=? WHERE delivery_id=? AND status='queued'`, [new Date().toISOString(), old.deliveryId]); + } catch (error) { + console.error( + JSON.stringify({ + level: "error", + event: "webhook_supersede_mark_failed", + error: errorMessageWithCause(error), + }), + ); + } + } + function enqueue(message: JobMessage, delaySeconds: number): void { const now = Date.now(); const payload = JSON.stringify(message); @@ -572,10 +604,16 @@ export function createSqliteQueue( } if (key) { const existing = driver.query( - `SELECT id FROM ${TABLE} WHERE status='pending' AND job_key=? ORDER BY priority DESC, run_after DESC, id LIMIT 1`, + `SELECT id, payload FROM ${TABLE} WHERE status='pending' AND job_key=? ORDER BY priority DESC, run_after DESC, id LIMIT 1`, [key], - ).rows[0] as { id: number } | undefined; + ).rows[0] as { id: number; payload: string } | undefined; if (existing) { + // #audit-webhook-supersede-trace: the row about to be overwritten below may itself be a github-webhook + // delivery (e.g. a "PR opened" pr-refresh coalesce) whose own webhook_events row was written as 'queued' + // BEFORE it ever reached this coalesce -- overwriting the payload here discards that delivery's id + // forever, so nothing would ever advance its webhook_events row past 'queued'. Mark it superseded FIRST, + // while the OLD payload (and its deliveryId) is still readable. + markSupersededWebhookEvent(existing.payload, message); // See the supersededKeyPrefix branch above: created_at is preserved across a coalesced re-enqueue so the // maintenance trickle clock reflects genuine wait time, not the most recent re-request. driver.query( diff --git a/test/unit/selfhost-pg-queue.test.ts b/test/unit/selfhost-pg-queue.test.ts index 12f9753bca..5aafbf0532 100644 --- a/test/unit/selfhost-pg-queue.test.ts +++ b/test/unit/selfhost-pg-queue.test.ts @@ -558,6 +558,31 @@ describe("createPgQueue (durable #977)", () => { ); }); + it("REGRESSION (#audit-webhook-supersede-trace): marks the superseded delivery's webhook_events row instead of leaving it stuck at 'queued' forever", async () => { + const m = makePool(); + const q = createPgQueue(m.pool, async () => undefined); + await q.init(); + m.fn.mockResolvedValueOnce({ rows: [{ id: "existing", payload: JSON.stringify(ciWebhook("ci-1", "check_run")) }], rowCount: 1 }); + + await q.binding.send(ciWebhook("ci-2", "check_run"), { delaySeconds: 1 }); + + expect(m.pool.query).toHaveBeenCalledWith( + expect.stringContaining("UPDATE webhook_events SET status='superseded'"), + ["ci-1", expect.any(String)], + ); + }); + + it("does NOT mark superseded when the coalesced-away job was not a github-webhook delivery (e.g. a scheduled sweep re-arm)", async () => { + const m = makePool(); + const q = createPgQueue(m.pool, async () => undefined); + await q.init(); + m.fn.mockResolvedValueOnce({ rows: [{ id: "existing", payload: JSON.stringify({ type: "refresh-registry", requestedBy: "schedule" }) }], rowCount: 1 }); + + await q.binding.send(msg("refresh-registry")); + + expect(m.pool.query).not.toHaveBeenCalledWith(expect.stringContaining("UPDATE webhook_events SET status='superseded'"), expect.anything()); + }); + it("does not reset created_at when coalescing a re-enqueue into an existing pending row (regression for #selfhost-runtime-drift)", async () => { // created_at anchors the maintenance trickle's age clock (maintenance-admission.ts). If a coalesced // re-enqueue reset it, a periodic scheduler re-requesting the same still-pending maintenance job faster diff --git a/test/unit/selfhost-sqlite-queue.test.ts b/test/unit/selfhost-sqlite-queue.test.ts index 46f9884c03..360174c848 100644 --- a/test/unit/selfhost-sqlite-queue.test.ts +++ b/test/unit/selfhost-sqlite-queue.test.ts @@ -871,6 +871,94 @@ describe("createSqliteQueue (durable #980)", () => { }); }); + it("REGRESSION (#audit-webhook-supersede-trace): marks a superseded pr-refresh delivery's webhook_events row instead of leaving it stuck at 'queued' forever", async () => { + const driver = makeDriver(); + driver.exec(` + CREATE TABLE webhook_events ( + delivery_id TEXT PRIMARY KEY, + event_name TEXT NOT NULL, + action TEXT, + installation_id INTEGER, + repository_full_name TEXT, + payload_hash TEXT NOT NULL, + status TEXT NOT NULL, + error_summary TEXT, + received_at TEXT NOT NULL, + processed_at TEXT + ); + `); + driver.query( + "INSERT INTO webhook_events (delivery_id, event_name, payload_hash, status, received_at) VALUES (?, 'pull_request', 'hash-1', 'queued', '2026-07-06T00:00:00.000Z')", + ["pr-1"], + ); + const q = createSqliteQueue(driver, async () => undefined); + + await q.binding.send(prWebhook("pr-1"), { delaySeconds: 60 }); + await q.binding.send(prWebhook("pr-2"), { delaySeconds: 1 }); // coalesces into pr-1's row, overwriting its payload + + const rows = driver.query("SELECT payload FROM _selfhost_jobs ORDER BY id", []).rows as Array<{ payload: string }>; + expect(rows).toHaveLength(1); // coalesced into one row + expect(JSON.parse(rows[0]!.payload).deliveryId).toBe("pr-2"); // pr-2's payload survives -- pr-1's is gone from the queue + const event = driver.query("SELECT status FROM webhook_events WHERE delivery_id = ?", ["pr-1"]).rows[0] as { status: string } | undefined; + expect(event?.status).toBe("superseded"); // pr-1's trace row is marked, not left stuck at 'queued' forever + }); + + it("fails safe when the webhook_events table itself errors: the coalesce still completes and only logs a warning", async () => { + const driver = makeDriver(); + // No webhook_events table -- the UPDATE inside markSupersededWebhookEvent will throw "no such table", which + // must be caught and logged, never allowed to abort the coalesce/enqueue itself. + const q = createSqliteQueue(driver, async () => undefined); + const errors = vi.spyOn(console, "error").mockImplementation(() => undefined); + + await q.binding.send(prWebhook("pr-1"), { delaySeconds: 60 }); + await q.binding.send(prWebhook("pr-2"), { delaySeconds: 1 }); + + const rows = driver.query("SELECT payload FROM _selfhost_jobs ORDER BY id", []).rows as Array<{ payload: string }>; + expect(rows).toHaveLength(1); // the coalesce itself still completed despite the missing table + expect(JSON.parse(rows[0]!.payload).deliveryId).toBe("pr-2"); + expect(errors.mock.calls.some((call) => String(call[0]).includes("webhook_supersede_mark_failed"))).toBe(true); + errors.mockRestore(); + }); + + it("does not mark superseded (no-op) when the coalesce is against the SAME deliveryId (defense-in-depth)", async () => { + const driver = makeDriver(); + driver.exec(` + CREATE TABLE webhook_events ( + delivery_id TEXT PRIMARY KEY, + event_name TEXT NOT NULL, + payload_hash TEXT NOT NULL, + status TEXT NOT NULL, + received_at TEXT NOT NULL + ); + `); + driver.query( + "INSERT INTO webhook_events (delivery_id, event_name, payload_hash, status, received_at) VALUES (?, 'pull_request', 'hash-1', 'queued', '2026-07-06T00:00:00.000Z')", + ["same-id"], + ); + const q = createSqliteQueue(driver, async () => undefined); + + await q.binding.send(prWebhook("same-id"), { delaySeconds: 60 }); + await q.binding.send(prWebhook("same-id"), { delaySeconds: 1 }); // re-coalesces against its own delivery id + + const event = driver.query("SELECT status FROM webhook_events WHERE delivery_id = ?", ["same-id"]).rows[0] as { status: string } | undefined; + expect(event?.status).toBe("queued"); // untouched -- there is nothing genuinely superseded here + }); + + it("tolerates an unparseable existing payload (a corrupted row) without throwing", async () => { + const driver = makeDriver(); + const q = createSqliteQueue(driver, async () => undefined); + driver.query( + `INSERT INTO _selfhost_jobs (payload, status, attempts, run_after, created_at, priority, job_key) VALUES (?, 'pending', 0, ?, ?, 5, ?)`, + ["not valid json{", Date.now() + 60_000, Date.now(), `github-webhook:pr-refresh:jsonbored/gittensory#1629@${"a".repeat(40)}`], + ); + + await expect(q.binding.send(prWebhook("pr-2"), { delaySeconds: 1 })).resolves.toBeUndefined(); + + const rows = driver.query("SELECT payload FROM _selfhost_jobs ORDER BY id", []).rows as Array<{ payload: string }>; + expect(rows).toHaveLength(1); // still coalesced into one row despite the corrupted existing payload + expect(JSON.parse(rows[0]!.payload).deliveryId).toBe("pr-2"); + }); + it("lets a pending full RAG index absorb later repo incrementals", async () => { const driver = makeDriver(); const q = createSqliteQueue(driver, async () => undefined);