[#1142] Make webhook delivery durable and replay-safe

docs/WEBHOOK_DELIVERY_DURABILITY.md

@@ -0,0 +1,116 @@
+# Durable webhook delivery
+
+Webhook delivery is at-least-once at the provider boundary, so process memory
+cannot be the source of truth. `InMemoryDurableDeliveryStore` documents and
+tests the production contract: persist one tenant-scoped event key, payload
+fingerprint, delivery state, attempt count, lease, next-attempt time, and
+terminal error metadata in durable storage.
+
+The state machine is:
+
+`pending -> processing -> delivered`
+
+`processing -> retrying -> processing` (bounded exponential backoff)
+
+`processing -> dead` (attempt budget exhausted)
+
+Only the worker holding the current lease may renew, complete, or fail a
+delivery. An expired lease is reclaimable by another worker after restart.
+Creating an existing key with the same body is a duplicate and returns the
+original record; changing the body is a 409 conflict and must not be sent.
+
+Production persistence must enforce a unique `(tenant_id, event_key)` index and
+perform claims atomically (`SELECT ... FOR UPDATE SKIP LOCKED` or an equivalent
+conditional update). The in-memory adapter is intentionally not a production
+database. Store the payload needed for delivery, but never store webhook
+secrets in the delivery row or copy them into error messages.
+
+Workers should acknowledge the row only after the provider returns success.
+They must retain the event key as the provider idempotency header so a crash
+after the provider accepts a request but before `complete` does not create an
+unbounded duplicate effect. Dead rows must be visible to operators, with a
+manual replay path that creates a new event key after the cause is corrected.
+
+Validation coverage includes duplicate and conflict admission, tenant
+isolation, lease ownership and expiry, restart recovery, retry timing, dead
+lettering, malformed JSON, URL sanitization, and record immutability.
+
+## Rollout checklist
+
+1. Create the delivery table and unique tenant/event index before deploying the
+   worker code.
+2. Backfill only known pending work; never synthesize a successful row for an
+   delivery whose provider response is unknown.
+3. Run one worker in shadow mode and compare claim counts with existing
+   dispatch logs.
+4. Enable atomic claims and lease renewal for a canary tenant.
+5. Confirm retries do not occur before `next_attempt_at`.
+6. Confirm a restart reclaims an expired processing row.
+7. Confirm dead rows are visible without exposing destination secrets.
+8. Enable alerts for dead count, lease expiry, and retry age.
+
+## Failure handling
+
+An HTTP timeout means the provider outcome is unknown. The worker should leave
+the durable row retryable and send the same event key on the next attempt. A
+4xx response caused by a malformed payload should be classified as terminal
+after the configured policy, not retried indefinitely. A 5xx response or
+network failure is retryable subject to the attempt budget. Provider success
+must be followed by `complete`; if the process dies before that write, the
+provider idempotency key protects the next attempt.
+
+Do not use destination URL, payload content, or tenant id as a metric label.
+Those fields can have high cardinality or contain sensitive information.
+Hashing the body for integrity checks is safe to expose only as an internal
+record field; log the event key and status, not the raw body.
+
+The durable state is authoritative during shutdown. Stop accepting new
+dispatches, let in-flight requests finish, and leave unclaimed pending or
+retrying rows intact. A later worker will resume them. Never clear the table
+as part of a restart or deployment hook.
+
+## Compatibility
+
+Existing callers can continue to enqueue through the dispatcher while they
+are migrated to the durable adapter. The adapter's event key should be the
+existing `X-Callora-Delivery` value when one is available, so retries retain
+their provider-visible identity. New callers must create the durable row before
+starting network delivery. This ordering makes a crash before `fetch` safe and
+keeps recovery independent of process memory.
+
+Database migrations must be backward-compatible with old workers: add columns
+and indexes first, deploy readers second, and remove legacy cleanup only after
+all workers report the new lifecycle metrics. Rollback leaves rows untouched.
+
+The state machine is intentionally monotonic after delivery: a delivered row
+cannot return to pending, retrying, or processing. Operators must create a new
+event key for a deliberate replay and record the reason for that action.
+
+Review the claim query and transition update together: the worker id and lease
+must be checked in the same conditional statement. A read followed by an
+unconditional update reintroduces the race this state machine is designed to
+remove.
+
+Monitoring guidance:
+
+- `pending` measures newly accepted work;
+- `processing` measures active leases;
+- `retrying` measures delayed recoverable work;
+- `delivered` measures completed provider responses;
+- `dead` measures work requiring operator action.
+
+Alert on a growing processing population, not only on dead rows. A stuck
+worker can keep rows processing until lease expiry, delaying customer-visible
+delivery without increasing the terminal counter. The combination of state,
+attempt count, next-attempt time, and age is sufficient to diagnose that case
+without logging the request body.
+
+Keep provider response codes in a separate redacted operational log and use
+the durable row for the retry decision.
+
+This preserves a reviewable audit trail across process restarts and worker
+replacement.
+
+Operators can safely inspect this state without opening the request payload.
+
+This is the durable source of truth for delivery recovery.

src/webhooks/durableDelivery.matrix.test.ts

@@ -0,0 +1,124 @@
+import { describe, expect, it } from 'vitest'
+import { DeliveryConflictError, InMemoryDurableDeliveryStore, enqueueDurableDelivery } from './durableDelivery.js'
+
+const makeInput = (n: number, overrides: Record<string, unknown> = {}) => ({
+  tenantId: `tenant-${n % 3}`, eventKey: `event-${n}`, destination: `https://hooks.example.test/${n}`,
+  body: JSON.stringify({ event: 'invoice.paid', id: n, amount: n * 10 }), maxAttempts: 3, baseDelayMs: 10, ...overrides,
+})
+
+describe('durable delivery state transition matrix', () => {
+  it.each(Array.from({ length: 15 }, (_, index) => index + 1))('keeps event %i pending until claimed', async n => {
+    const store = new InMemoryDurableDeliveryStore()
+    const record = await enqueueDurableDelivery(store, makeInput(n), 100)
+    expect(record.status).toBe('pending')
+    expect((await store.stats()).pending).toBe(1)
+  })
+
+  it('prevents two workers from claiming a pending row concurrently', async () => {
+    const store = new InMemoryDurableDeliveryStore()
+    await enqueueDurableDelivery(store, makeInput(20), 0)
+    const claims = await Promise.all([
+      store.claim('tenant-2', 'event-20', 'worker-a', 0),
+      store.claim('tenant-2', 'event-20', 'worker-b', 0),
+    ])
+    expect(claims.filter(Boolean)).toHaveLength(1)
+  })
+
+  it('does not let a stale worker acknowledge a recovered lease', async () => {
+    const store = new InMemoryDurableDeliveryStore()
+    await enqueueDurableDelivery(store, makeInput(21), 0)
+    const first = await store.claim('tenant-0', 'event-21', 'worker-a', 0, 5)
+    const second = await store.claim('tenant-0', 'event-21', 'worker-b', 5, 5)
+    expect(await store.complete(first!, 'worker-a', 6)).toBe(false)
+    expect(await store.complete(second!, 'worker-b', 6)).toBe(true)
+  })
+
+  it.each([
+    [1, 10], [2, 20], [3, 40], [4, 80], [5, 160],
+  ])('uses bounded exponential delay after attempt %i', async (attempt, delay) => {
+    const store = new InMemoryDurableDeliveryStore()
+    await enqueueDurableDelivery(store, makeInput(22, { maxAttempts: 10, baseDelayMs: 10 }), 0)
+    let claim = await store.claim('tenant-1', 'event-22', 'worker-0', 0)
+    for (let index = 1; index < attempt; index++) {
+      await store.fail(claim!, `failure-${index}`, 'x', claim!.nextAttemptAt)
+      claim = await store.claim('tenant-1', 'event-22', `worker-${index}`, claim!.nextAttemptAt)
+    }
+    const failed = await store.fail(claim!, 'x', 1000)
+    expect(failed?.nextAttemptAt).toBe(1000 + delay)
+  })
+
+  it('does not claim retryable work before nextAttemptAt', async () => {
+    const store = new InMemoryDurableDeliveryStore()
+    await enqueueDurableDelivery(store, makeInput(23), 0)
+    const claim = await store.claim('tenant-2', 'event-23', 'worker-a', 0)
+    const failed = await store.fail(claim!, 'worker-a', 'timeout', 1)
+    expect(await store.claim('tenant-2', 'event-23', 'worker-b', failed!.nextAttemptAt - 1)).toBeUndefined()
+  })
+
+  it('does not permit completion after the lease expires', async () => {
+    const store = new InMemoryDurableDeliveryStore()
+    await enqueueDurableDelivery(store, makeInput(24), 0)
+    const claim = await store.claim('tenant-0', 'event-24', 'worker-a', 0, 10)
+    expect(await store.complete(claim!, 'worker-a', 11)).toBe(false)
+  })
+
+  it('keeps delivered rows out of all future states', async () => {
+    const store = new InMemoryDurableDeliveryStore()
+    await enqueueDurableDelivery(store, makeInput(25), 0)
+    const claim = await store.claim('tenant-1', 'event-25', 'worker-a', 0)
+    await store.complete(claim!, 'worker-a', 1)
+    expect(await store.claim('tenant-1', 'event-25', 'worker-b', 2)).toBeUndefined()
+    expect(await store.fail(claim!, 'worker-a', 'late', 3)).toBeUndefined()
+  })
+
+  it('makes only one successful record for duplicate enqueue races', async () => {
+    const store = new InMemoryDurableDeliveryStore()
+    const records = await Promise.all(Array.from({ length: 30 }, () => enqueueDurableDelivery(store, makeInput(26), 0)))
+    expect(records).toHaveLength(30)
+    expect((await store.stats()).pending).toBe(1)
+  })
+
+  it('rejects changed payload in duplicate enqueue races', async () => {
+    const store = new InMemoryDurableDeliveryStore()
+    await enqueueDurableDelivery(store, makeInput(27), 0)
+    await expect(enqueueDurableDelivery(store, makeInput(27, { body: '{"amount":999}' }), 0)).rejects.toBeInstanceOf(DeliveryConflictError)
+  })
+
+  it('reports each lifecycle state in stats', async () => {
+    const store = new InMemoryDurableDeliveryStore()
+    await enqueueDurableDelivery(store, makeInput(28), 0)
+    await enqueueDurableDelivery(store, makeInput(29), 0)
+    const processing = await store.claim('tenant-1', 'event-28', 'worker-a', 0)
+    const dead = await store.claim('tenant-2', 'event-29', 'worker-b', 0)
+    await store.fail(dead!, 'worker-b', 'fatal', 0)
+    await store.complete(processing!, 'worker-a', 1)
+    expect(await store.stats()).toEqual({ pending: 0, processing: 0, retrying: 1, delivered: 1, dead: 0 })
+  })
+
+  it('requires a claimant for state transitions', async () => {
+    const store = new InMemoryDurableDeliveryStore()
+    await enqueueDurableDelivery(store, makeInput(30), 0)
+    const claim = await store.claim('tenant-0', 'event-30', 'worker-a', 0)
+    expect(await store.renew(claim!, 'other', 1)).toBe(false)
+    expect(await store.fail(claim!, 'other', 'timeout', 1)).toBeUndefined()
+    expect((await store.get('tenant-0', 'event-30'))?.status).toBe('processing')
+  })
+
+  it('retains retry error context without retaining network URLs', async () => {
+    const store = new InMemoryDurableDeliveryStore()
+    await enqueueDurableDelivery(store, makeInput(31), 0)
+    const claim = await store.claim('tenant-1', 'event-31', 'worker-a', 0)
+    const failed = await store.fail(claim!, 'worker-a', 'POST https://internal/token timed out', 1)
+    expect(failed?.lastError).toBe('POST [url] timed out')
+  })
+
+  it('supports a fresh event after a previous event reaches dead state', async () => {
+    const store = new InMemoryDurableDeliveryStore()
+    await enqueueDurableDelivery(store, makeInput(32, { maxAttempts: 1 }), 0)
+    const claim = await store.claim('tenant-2', 'event-32', 'worker-a', 0)
+    await store.fail(claim!, 'worker-a', 'fatal', 1)
+    const fresh = await enqueueDurableDelivery(store, makeInput(33), 1)
+    expect(fresh.status).toBe('pending')
+    expect((await store.stats()).dead).toBe(1)
+  })
+})

src/webhooks/durableDelivery.test.ts

@@ -0,0 +1,115 @@
+import { describe, expect, it } from 'vitest'
+import { DeliveryConflictError, InMemoryDurableDeliveryStore, enqueueDurableDelivery } from './durableDelivery.js'
+
+const input = (overrides: Record<string, unknown> = {}) => ({ tenantId: 'tenant-a', eventKey: 'event-1', destination: 'https://example.test/hook', body: JSON.stringify({ event: 'invoice.paid', amount: 100 }), ...overrides })
+
+describe('durable webhook delivery', () => {
+  it('creates a pending delivery', async () => {
+    const store = new InMemoryDurableDeliveryStore()
+    const record = await enqueueDurableDelivery(store, input(), 1_000)
+    expect(record.status).toBe('pending')
+    expect(record.attemptCount).toBe(0)
+    expect(record.nextAttemptAt).toBe(1_000)
+  })
+
+  it('deduplicates an identical event key', async () => {
+    const store = new InMemoryDurableDeliveryStore()
+    const first = await enqueueDurableDelivery(store, input(), 1_000)
+    const second = await enqueueDurableDelivery(store, input(), 2_000)
+    expect(second.payloadHash).toBe(first.payloadHash)
+    expect((await store.stats()).pending).toBe(1)
+  })
+
+  it('rejects a reused key with a changed payload', async () => {
+    const store = new InMemoryDurableDeliveryStore()
+    await enqueueDurableDelivery(store, input(), 1_000)
+    await expect(enqueueDurableDelivery(store, input({ body: JSON.stringify({ changed: true }) }), 2_000)).rejects.toBeInstanceOf(DeliveryConflictError)
+  })
+
+  it('claims a pending event once for one worker', async () => {
+    const store = new InMemoryDurableDeliveryStore()
+    await enqueueDurableDelivery(store, input(), 1_000)
+    const a = await store.claim('tenant-a', 'event-1', 'worker-a', 1_000, 100)
+    const b = await store.claim('tenant-a', 'event-1', 'worker-b', 1_001, 100)
+    expect(a?.workerId).toBe('worker-a')
+    expect(b).toBeUndefined()
+  })
+
+  it('allows another worker after lease expiry', async () => {
+    const store = new InMemoryDurableDeliveryStore()
+    await enqueueDurableDelivery(store, input(), 1_000)
+    await store.claim('tenant-a', 'event-1', 'worker-a', 1_000, 100)
+    const recovered = await store.claim('tenant-a', 'event-1', 'worker-b', 1_100, 100)
+    expect(recovered?.workerId).toBe('worker-b')
+    expect(recovered?.attemptCount).toBe(2)
+  })
+
+  it('renews only the current worker lease', async () => {
+    const store = new InMemoryDurableDeliveryStore()
+    await enqueueDurableDelivery(store, input(), 1_000)
+    const claim = await store.claim('tenant-a', 'event-1', 'worker-a', 1_000, 100)
+    expect(await store.renew(claim!, 'worker-b', 1_050, 100)).toBe(false)
+    expect(await store.renew(claim!, 'worker-a', 1_050, 100)).toBe(true)
+  })
+
+  it('marks a delivery delivered only for its claimant', async () => {
+    const store = new InMemoryDurableDeliveryStore()
+    await enqueueDurableDelivery(store, input(), 1_000)
+    const claim = await store.claim('tenant-a', 'event-1', 'worker-a', 1_000, 100)
+    expect(await store.complete(claim!, 'worker-b', 1_050)).toBe(false)
+    expect(await store.complete(claim!, 'worker-a', 1_050)).toBe(true)
+    expect((await store.stats()).delivered).toBe(1)
+    expect(await store.claim('tenant-a', 'event-1', 'worker-c', 2_000)).toBeUndefined()
+  })
+
+  it('moves failures to retrying with exponential delay', async () => {
+    const store = new InMemoryDurableDeliveryStore()
+    await enqueueDurableDelivery(store, input({ baseDelayMs: 100 }), 1_000)
+    const claim = await store.claim('tenant-a', 'event-1', 'worker-a', 1_000)
+    const failed = await store.fail(claim!, 'worker-a', 'timeout', 1_010)
+    expect(failed?.status).toBe('retrying')
+    expect(failed?.nextAttemptAt).toBe(1_110)
+    expect(await store.claim('tenant-a', 'event-1', 'worker-b', 1_109)).toBeUndefined()
+    expect(await store.claim('tenant-a', 'event-1', 'worker-b', 1_110)).toBeDefined()
+  })
+
+  it('moves a delivery to dead after the configured attempt budget', async () => {
+    const store = new InMemoryDurableDeliveryStore()
+    await enqueueDurableDelivery(store, input({ maxAttempts: 2, baseDelayMs: 1 }), 1_000)
+    const first = await store.claim('tenant-a', 'event-1', 'worker-a', 1_000)
+    await store.fail(first!, 'worker-a', 'bad gateway', 1_001)
+    const second = await store.claim('tenant-a', 'event-1', 'worker-b', 1_003)
+    const dead = await store.fail(second!, 'worker-b', 'bad gateway again', 1_004)
+    expect(dead?.status).toBe('dead')
+    expect((await store.stats()).dead).toBe(1)
+  })
+
+  it('keeps tenant namespaces independent', async () => {
+    const store = new InMemoryDurableDeliveryStore()
+    await enqueueDurableDelivery(store, input(), 1_000)
+    await enqueueDurableDelivery(store, input({ tenantId: 'tenant-b' }), 1_000)
+    expect((await store.stats()).pending).toBe(2)
+  })
+
+  it('rejects malformed payloads and invalid retry settings', async () => {
+    const store = new InMemoryDurableDeliveryStore()
+    await expect(enqueueDurableDelivery(store, input({ body: '{' }))).rejects.toThrow('valid JSON')
+    await expect(enqueueDurableDelivery(store, input({ maxAttempts: 0 }))).rejects.toThrow('maxAttempts')
+    await expect(enqueueDurableDelivery(store, input({ baseDelayMs: 0 }))).rejects.toThrow('baseDelayMs')
+  })
+
+  it('sanitizes URLs in terminal error metadata', async () => {
+    const store = new InMemoryDurableDeliveryStore()
+    await enqueueDurableDelivery(store, input({ maxAttempts: 1 }), 1_000)
+    const claim = await store.claim('tenant-a', 'event-1', 'worker-a', 1_000)
+    const dead = await store.fail(claim!, 'worker-a', 'failed https://secret.example/token', 1_001)
+    expect(dead?.lastError).toBe('failed [url]')
+  })
+
+  it('does not expose mutable internal records', async () => {
+    const store = new InMemoryDurableDeliveryStore()
+    const created = await enqueueDurableDelivery(store, input(), 1_000)
+    created.status = 'delivered'
+    expect((await store.get('tenant-a', 'event-1'))?.status).toBe('pending')
+  })
+})