From 644adb52ac35afdc003ceb758e1e86d4acbca57c Mon Sep 17 00:00:00 2001 From: ag-linden Date: Tue, 30 Jun 2026 21:22:22 -0500 Subject: [PATCH] fix(jobs): recover stale webhook deliveries --- apps/jobs/package.json | 4 +- apps/jobs/src/consumers/deliveries.test.ts | 126 +++++++++++++++++++++ apps/jobs/src/consumers/deliveries.ts | 72 +++++++++--- apps/jobs/src/lib/constants.ts | 2 + apps/jobs/vitest.config.mjs | 16 +++ pnpm-lock.yaml | 3 + 6 files changed, 209 insertions(+), 14 deletions(-) create mode 100644 apps/jobs/src/consumers/deliveries.test.ts create mode 100644 apps/jobs/vitest.config.mjs diff --git a/apps/jobs/package.json b/apps/jobs/package.json index e6deb0b4..77520368 100644 --- a/apps/jobs/package.json +++ b/apps/jobs/package.json @@ -4,7 +4,8 @@ "scripts": { "dev": "wrangler dev", "deploy": "wrangler deploy --minify", - "cf-typegen": "wrangler types --env-interface CloudflareBindings" + "cf-typegen": "wrangler types --env-interface CloudflareBindings", + "test": "vitest run" }, "dependencies": { "@marble/db": "workspace:*", @@ -20,6 +21,7 @@ }, "devDependencies": { "@cloudflare/workers-types": "^4.20260508.1", + "vitest": "^3.2.4", "wrangler": "^4.90.0" } } diff --git a/apps/jobs/src/consumers/deliveries.test.ts b/apps/jobs/src/consumers/deliveries.test.ts new file mode 100644 index 00000000..d025c4d0 --- /dev/null +++ b/apps/jobs/src/consumers/deliveries.test.ts @@ -0,0 +1,126 @@ +import { describe, expect, it, vi } from "vitest"; + +vi.mock("@marble/events", () => ({ + buildWebhookPayload: vi.fn(), + serializeEventType: vi.fn(), +})); +vi.mock("@/lib/db", () => ({ createDbClient: vi.fn() })); +vi.mock("@/lib/formats", () => ({ buildWebhookRequestBody: vi.fn() })); +vi.mock("@/lib/signing", () => ({ signPayload: vi.fn() })); +vi.mock("@/lib/usage", () => ({ + checkWebhookUsage: vi.fn(), + recordWebhookUsage: vi.fn(), + sendWebhookUsageAlert: vi.fn(), +})); + +import { WEBHOOK_DELIVERY_STALE_SENDING_MS } from "@/lib/constants"; +import { claimWebhookDeliveryAttempt } from "./deliveries"; + +function createDb({ + activeClaimCount, + staleClaimCount, + status, +}: { + activeClaimCount: number; + staleClaimCount: number; + status?: string; +}) { + const updateMany = vi + .fn() + .mockResolvedValueOnce({ count: activeClaimCount }) + .mockResolvedValueOnce({ count: staleClaimCount }); + const findUnique = vi + .fn() + .mockResolvedValue(status === undefined ? null : { status }); + + return { + webhookDelivery: { + updateMany, + findUnique, + }, + }; +} + +describe("claimWebhookDeliveryAttempt", () => { + it("claims pending or retrying deliveries for a new attempt", async () => { + const now = new Date("2026-06-30T12:00:00.000Z"); + const db = createDb({ activeClaimCount: 1, staleClaimCount: 0 }); + + await expect( + claimWebhookDeliveryAttempt(db as never, "delivery_1", now) + ).resolves.toBe(true); + + expect(db.webhookDelivery.updateMany).toHaveBeenCalledTimes(1); + expect(db.webhookDelivery.updateMany).toHaveBeenCalledWith({ + where: { + id: "delivery_1", + status: { in: ["pending", "retrying"] }, + }, + data: { + status: "sending", + attemptCount: { increment: 1 }, + lastAttemptAt: now, + }, + }); + expect(db.webhookDelivery.findUnique).not.toHaveBeenCalled(); + }); + + it("reclaims stale sending deliveries instead of acknowledging them", async () => { + const now = new Date("2026-06-30T12:00:00.000Z"); + const db = createDb({ activeClaimCount: 0, staleClaimCount: 1 }); + + await expect( + claimWebhookDeliveryAttempt(db as never, "delivery_1", now) + ).resolves.toBe(true); + + expect(db.webhookDelivery.updateMany).toHaveBeenCalledTimes(2); + expect(db.webhookDelivery.updateMany).toHaveBeenNthCalledWith(2, { + where: { + id: "delivery_1", + status: "sending", + OR: [ + { + lastAttemptAt: { + lt: new Date( + now.getTime() - WEBHOOK_DELIVERY_STALE_SENDING_MS + ), + }, + }, + { lastAttemptAt: null }, + ], + }, + data: { + status: "sending", + attemptCount: { increment: 1 }, + lastAttemptAt: now, + }, + }); + expect(db.webhookDelivery.findUnique).not.toHaveBeenCalled(); + }); + + it("keeps active sending deliveries on the queue retry path", async () => { + const now = new Date("2026-06-30T12:00:00.000Z"); + const db = createDb({ + activeClaimCount: 0, + staleClaimCount: 0, + status: "sending", + }); + + await expect( + claimWebhookDeliveryAttempt(db as never, "delivery_1", now) + ).rejects.toThrow("Webhook delivery delivery_1 is already sending"); + }); + + it("skips terminal or missing deliveries", async () => { + const now = new Date("2026-06-30T12:00:00.000Z"); + const db = createDb({ + activeClaimCount: 0, + staleClaimCount: 0, + status: "success", + }); + + await expect( + claimWebhookDeliveryAttempt(db as never, "delivery_1", now) + ).resolves.toBe(false); + }); +}); diff --git a/apps/jobs/src/consumers/deliveries.ts b/apps/jobs/src/consumers/deliveries.ts index e7ee9343..2c70f57f 100644 --- a/apps/jobs/src/consumers/deliveries.ts +++ b/apps/jobs/src/consumers/deliveries.ts @@ -1,5 +1,8 @@ import { buildWebhookPayload, serializeEventType } from "@marble/events"; -import { WEBHOOK_DELIVERY_TIMEOUT_MS } from "@/lib/constants"; +import { + WEBHOOK_DELIVERY_STALE_SENDING_MS, + WEBHOOK_DELIVERY_TIMEOUT_MS, +} from "@/lib/constants"; import { createDbClient } from "@/lib/db"; import { buildWebhookRequestBody } from "@/lib/formats"; import { signPayload } from "@/lib/signing"; @@ -37,19 +40,9 @@ async function processDelivery( env: Env, deliveryId: string ) { - const claim = await db.webhookDelivery.updateMany({ - where: { - id: deliveryId, - status: { in: ["pending", "retrying"] }, - }, - data: { - status: "sending", - attemptCount: { increment: 1 }, - lastAttemptAt: new Date(), - }, - }); + const claimed = await claimWebhookDeliveryAttempt(db, deliveryId); - if (claim.count === 0) { + if (!claimed) { return; } @@ -220,3 +213,56 @@ async function processDelivery( throw new Error(`Webhook returned ${response.status}`); } + +export async function claimWebhookDeliveryAttempt( + db: ReturnType, + deliveryId: string, + now = new Date() +) { + const claim = await db.webhookDelivery.updateMany({ + where: { + id: deliveryId, + status: { in: ["pending", "retrying"] }, + }, + data: { + status: "sending", + attemptCount: { increment: 1 }, + lastAttemptAt: now, + }, + }); + + if (claim.count > 0) { + return true; + } + + const staleCutoff = new Date( + now.getTime() - WEBHOOK_DELIVERY_STALE_SENDING_MS + ); + const staleClaim = await db.webhookDelivery.updateMany({ + where: { + id: deliveryId, + status: "sending", + OR: [{ lastAttemptAt: { lt: staleCutoff } }, { lastAttemptAt: null }], + }, + data: { + status: "sending", + attemptCount: { increment: 1 }, + lastAttemptAt: now, + }, + }); + + if (staleClaim.count > 0) { + return true; + } + + const delivery = await db.webhookDelivery.findUnique({ + where: { id: deliveryId }, + select: { status: true }, + }); + + if (delivery?.status === "sending") { + throw new Error(`Webhook delivery ${deliveryId} is already sending`); + } + + return false; +} diff --git a/apps/jobs/src/lib/constants.ts b/apps/jobs/src/lib/constants.ts index 3bdcac09..7f34b86c 100644 --- a/apps/jobs/src/lib/constants.ts +++ b/apps/jobs/src/lib/constants.ts @@ -8,6 +8,8 @@ export const IMPORT_JOB_RETENTION_DAYS = 30; export const IMPORT_STALE_JOB_DAYS = 1; export const WEBHOOK_DELIVERY_RETENTION_DAYS = 30; export const WEBHOOK_DELIVERY_TIMEOUT_MS = 15_000; +export const WEBHOOK_DELIVERY_STALE_SENDING_MS = + WEBHOOK_DELIVERY_TIMEOUT_MS * 2; // The dashboard currently renders a 30-day chart plus the previous 30-day // comparison from raw API request rows. export const API_REQUEST_RETENTION_DAYS = 60; diff --git a/apps/jobs/vitest.config.mjs b/apps/jobs/vitest.config.mjs new file mode 100644 index 00000000..fa8471e9 --- /dev/null +++ b/apps/jobs/vitest.config.mjs @@ -0,0 +1,16 @@ +import { dirname, resolve } from "node:path"; +import { fileURLToPath } from "node:url"; +import { defineConfig } from "vitest/config"; + +const root = dirname(fileURLToPath(import.meta.url)); + +export default defineConfig({ + resolve: { + alias: { + "@": resolve(root, "src"), + }, + }, + test: { + include: ["src/**/*.test.ts"], + }, +}); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index f0d120c9..88ade36b 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -339,6 +339,9 @@ importers: '@cloudflare/workers-types': specifier: ^4.20260508.1 version: 4.20260626.1 + vitest: + specifier: ^3.2.4 + version: 3.2.6(@types/debug@4.1.13)(@types/node@24.13.2)(jiti@2.7.0)(lightningcss@1.32.0)(yaml@2.9.0) wrangler: specifier: ^4.90.0 version: 4.105.0(@cloudflare/workers-types@4.20260626.1)(bufferutil@4.1.0)