Recover stale webhook deliveries

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"
   }
 }

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);
+  });
+});

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<typeof createDbClient>,
+  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;
+}

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;

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"],
+  },
+});