fix(webhooks): derive the stuck-in_flight window from the cycle bound and charge the stall an attempt (#1311)
* fix(webhooks): derive the stuck-in_flight window from the cycle bound and charge the stall an attempt recoverStuckInFlight re-armed any in_flight row older than 2x REQUEST_TIMEOUT_MS (20 s), but a cron cycle claims 50 rows and attempts them serially, stamping updated_at once at claim time. From row 3 onward every row was past the threshold before its own attempt started, so each cycle recovered and re-claimed the rows the previous cycle was still working through: duplicate POSTs of the same X-Gnubok-Delivery, and a terminal status decided by a race whose loser was swallowed by enforce_webhook_delivery_immutability as a log.warn. Both halves of #1257 are fixed: 1. The window is derived, not guessed. The attempt loop is now bounded by an explicit CYCLE_BUDGET_MS (120 s) instead of relying on the platform to kill it, and the sweep window is that bound plus one receiver timeout plus slack (160 s), floored at the cron's own batch size so the 5-row emit kick cannot re-arm rows the 50-row cron still owns. Each row is also re-stamped immediately before its own attempt, so a row's in_flight age measures the attempt rather than the claim. The same write doubles as an ownership check: a zero-row result means another cycle took the row, and the POST is dropped instead of duplicated. 2. The sweep charges an attempt, so MAX_ATTEMPTS is a real cap again. The predicate moves into a SECURITY DEFINER RPC because PostgREST can express neither `attempts = attempts + 1` nor the conditional flip at the cap, and a read-then-write loop would reopen a TOCTOU against the immutability trigger. A row recovered past the cap lands on exactly the terminal state the normal retry path produces: status 'dead', attempts = MAX_ATTEMPTS, error prefixed 'attempts_exhausted'. The trigger is neither weakened nor bypassed: the outer UPDATE keeps status = 'in_flight' in its own WHERE, so a row that raced to a terminal status fails re-evaluation under READ COMMITTED and is skipped rather than aborting the statement. Rows the cycle claimed but will not reach are handed back as re-claimable instead of being stranded in in_flight, without charging an attempt they never made. Adds the partial index the sweep needs (idx_webhook_deliveries_due is partial on pending/failed and structurally excludes in_flight). No retention or pruning cron: webhook_deliveries still has no cleanup path, which is a separate decision and stays a follow-up. Fixes #1257 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix(webhooks): back the cycle budget with maxDuration and give the stall the normal retry backoff Review follow-up on the #1257 fix. Two of the findings were blocking and compound each other: the fix made stranding likely and destructive at the same time. 1. The 160 s sweep window was derived from CYCLE_BUDGET_MS, but nothing granted a dispatch cycle 120 s: the cron route declared no maxDuration. If the platform killed the invocation before the budget check fired, releaseUnattempted never ran and the claimed-but- unattempted rows stayed in in_flight carrying their claim-time updated_at, which is exactly the invariant the window depends on. The route now declares maxDuration = 300, the way the stripe transactions and documents verify crons pair a budget with one, and a route test asserts both the literal and its relation to CYCLE_BUDGET_MS. The kick path can never be given a maxDuration (after() runs inside an arbitrary route), so dispatch-kick.ts now states why it does not need one: KICK_BATCH_SIZE x REQUEST_TIMEOUT_MS is 50 s, so the dispatcher's budget check never fires there. 2. The sweep charged an attempt but re-armed at p_now, i.e. no backoff, while the normal failure path waits RETRY_BACKOFF_SECONDS. A row that kept getting stranded (deploy, instance recycle, any cycle that outlives its invocation) was re-claimable on the next per-minute tick and could burn all 8 attempts in roughly 20 minutes, landing in the terminal, immutable 'dead' state without its receiver ever being contacted. Pre-fix that loop was infinite but harmless, so this was a net-new way to lose a delivery. recover_stuck_webhook_deliveries now takes p_backoff int[] (RETRY_BACKOFF_SECONDS, still single-sourced in TS) and sets next_attempt_at with the same clamped index lookup markFailedForRetry uses, so a stall costs an attempt AND the same wait a 500 costs. A non-positive or empty schedule is rejected rather than silently degrading to p_now. The migration has not been applied to any deployed environment, so it is amended in place rather than superseded; it drops the old 3-argument signature so no ambiguous overload can survive in a dev or CI database. Also from the review: - stuckInFlightAfterMs(batchSize) was dead code whose Math.min clamp made every input return 120_000, so the documented DEFAULT_BATCH_SIZE floor never fired and the test that pinned it (stuckInFlightAfterMs(5) === stuckInFlightAfterMs(50)) was a tautology. It is now the plain constant STUCK_IN_FLIGHT_AFTER_MS with a comment that credits the budget, and the test drives the window through dispatchDueDeliveries at batch sizes 5, 50 and 500, which fails if the window ever becomes batch-derived again. - The sweep's outcome reaches the operator: recovered / recoveredDead are on DispatchSummary and in the cron's structured log, so a tick that takes deliveries terminal is visible without grepping helper-level warn lines. - releaseUnattempted no longer writes 'failed' onto a never-attempted row. claim_due_webhook_deliveries does not return the pre-claim status, but it does return attempts, and every path that writes 'failed' also writes attempts >= 1, so attempts = 0 identifies a row that was 'pending' and it is restored as such. webhook_deliveries is customer-visible behandlingshistorik; a delivery that was claimed and handed back without a single POST must not read as a failure there. The two deferred hygiene items (no retention path for webhook_deliveries, and the sweep still being an unbounded tenant-global UPDATE) are reported as a comment on #1257 and noted in the migration. Fixes #1257 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Jakob Wennberg
co-authored by
Claude Opus 5
Claude Opus 5
parent
f3bf50d862
commit
1a5d205bd6
@@ -14,6 +14,22 @@ vi.mock('next/server', () => ({
|
||||
|
||||
import { kickWebhookDispatch, resetKickStateForTests, KICK_BATCH_SIZE } from '../dispatch-kick'
|
||||
|
||||
/**
|
||||
* A zero DispatchSummary. Spread rather than repeated so adding a counter to
|
||||
* the summary does not mean re-editing every case in this file; these tests
|
||||
* are about scheduling and coalescing, never about the counts.
|
||||
*/
|
||||
const EMPTY_SUMMARY = {
|
||||
picked: 0,
|
||||
delivered: 0,
|
||||
failed: 0,
|
||||
dead: 0,
|
||||
skipped: 0,
|
||||
released: 0,
|
||||
recovered: 0,
|
||||
recoveredDead: 0,
|
||||
}
|
||||
|
||||
/** Resolves after the microtask queue drains, so floating work has run. */
|
||||
const flush = () => new Promise((resolve) => setTimeout(resolve, 0))
|
||||
|
||||
@@ -24,7 +40,7 @@ beforeEach(() => {
|
||||
|
||||
describe('kickWebhookDispatch', () => {
|
||||
it('runs one dispatch cycle with the small kick batch size', async () => {
|
||||
const dispatch = vi.fn().mockResolvedValue({ picked: 1, delivered: 1, failed: 0, dead: 0 })
|
||||
const dispatch = vi.fn().mockResolvedValue({ ...EMPTY_SUMMARY, picked: 1, delivered: 1 })
|
||||
|
||||
kickWebhookDispatch(dispatch)
|
||||
await flush()
|
||||
@@ -46,7 +62,7 @@ describe('kickWebhookDispatch', () => {
|
||||
new Promise((resolve) => {
|
||||
setTimeout(() => {
|
||||
settled = true
|
||||
resolve({ picked: 0, delivered: 0, failed: 0, dead: 0 })
|
||||
resolve({ ...EMPTY_SUMMARY })
|
||||
}, 20)
|
||||
}),
|
||||
)
|
||||
@@ -66,7 +82,7 @@ describe('kickWebhookDispatch', () => {
|
||||
it('coalesces a burst of kicks into a single cycle', async () => {
|
||||
// A bulk booking emits once per row. Without coalescing, 100 rows would
|
||||
// mean 100 claim round trips against the same handful of due deliveries.
|
||||
const dispatch = vi.fn().mockResolvedValue({ picked: 0, delivered: 0, failed: 0, dead: 0 })
|
||||
const dispatch = vi.fn().mockResolvedValue({ ...EMPTY_SUMMARY })
|
||||
|
||||
for (let i = 0; i < 100; i++) kickWebhookDispatch(dispatch)
|
||||
await flush()
|
||||
@@ -75,7 +91,7 @@ describe('kickWebhookDispatch', () => {
|
||||
})
|
||||
|
||||
it('accepts a new kick once the previous cycle has started', async () => {
|
||||
const dispatch = vi.fn().mockResolvedValue({ picked: 0, delivered: 0, failed: 0, dead: 0 })
|
||||
const dispatch = vi.fn().mockResolvedValue({ ...EMPTY_SUMMARY })
|
||||
|
||||
kickWebhookDispatch(dispatch)
|
||||
await flush()
|
||||
@@ -101,7 +117,7 @@ describe('kickWebhookDispatch', () => {
|
||||
kickWebhookDispatch(failing)
|
||||
await flush()
|
||||
|
||||
const ok = vi.fn().mockResolvedValue({ picked: 0, delivered: 0, failed: 0, dead: 0 })
|
||||
const ok = vi.fn().mockResolvedValue({ ...EMPTY_SUMMARY })
|
||||
kickWebhookDispatch(ok)
|
||||
await flush()
|
||||
|
||||
|
||||
@@ -76,12 +76,17 @@ function makeSupabase(opts: {
|
||||
}
|
||||
|
||||
if (table === 'webhook_deliveries') {
|
||||
// update(...).eq(...) -> markDead / markDelivered
|
||||
// update(...).eq(...).lt(...).select('id') -> recoverStuckInFlight
|
||||
// update(...).eq(...) -> markDead / markDelivered
|
||||
// update(...).eq('id').eq('status').select('id') -> touchInFlight
|
||||
//
|
||||
// The touch must resolve to a NON-empty row set: an empty result means
|
||||
// "this delivery is no longer ours" and the dispatcher skips the
|
||||
// attempt, which would take the 410 auto-disable path out of reach.
|
||||
// The stuck sweep no longer runs through this chain at all; it goes
|
||||
// through rpc('recover_stuck_webhook_deliveries').
|
||||
const thenable = {
|
||||
eq: () => thenable,
|
||||
lt: () => thenable,
|
||||
select: async () => ({ data: [], error: null }),
|
||||
select: async () => ({ data: [{ id: DELIVERY_ID }], error: null }),
|
||||
then: (resolve: (v: { error: null }) => unknown) => resolve({ error: null }),
|
||||
}
|
||||
return { update: () => thenable }
|
||||
|
||||
@@ -0,0 +1,470 @@
|
||||
/**
|
||||
* Stuck-in_flight recovery and per-row re-stamping (#1257).
|
||||
*
|
||||
* Two defects are pinned here:
|
||||
*
|
||||
* 1. the sweep window used to be a fixed 2x REQUEST_TIMEOUT_MS (20 s), which
|
||||
* is shorter than a single serial cycle, so cycle N recovered rows cycle
|
||||
* N-1 was still working through;
|
||||
* 2. recovery reset rows to 'failed' without charging an attempt, so
|
||||
* MAX_ATTEMPTS stopped being a real cap.
|
||||
*
|
||||
* The window is now derived from the bounded cycle and the cap is enforced
|
||||
* server-side by recover_stuck_webhook_deliveries, so the assertions here are
|
||||
* about what the dispatcher hands the RPC and about the per-row re-stamp that
|
||||
* makes a row's in_flight age measure its own attempt.
|
||||
*/
|
||||
|
||||
import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'
|
||||
import type { SupabaseClient } from '@supabase/supabase-js'
|
||||
import type { PinnedFetchResult } from '../pinned-fetch'
|
||||
|
||||
vi.mock('@/lib/logger', () => ({
|
||||
createLogger: () => ({
|
||||
info: vi.fn(),
|
||||
warn: vi.fn(),
|
||||
error: vi.fn(),
|
||||
child: vi.fn(),
|
||||
}),
|
||||
}))
|
||||
|
||||
// Import after mocks
|
||||
import { dispatchDueDeliveries, __TESTING__ } from '../dispatcher'
|
||||
|
||||
const COMPANY_ID = '11111111-1111-4111-8111-111111111111'
|
||||
const WEBHOOK_ID = '22222222-2222-4222-8222-222222222222'
|
||||
const DELIVERY_IDS = [
|
||||
'aaaaaaaa-0001-4000-8000-000000000001',
|
||||
'aaaaaaaa-0002-4000-8000-000000000002',
|
||||
'aaaaaaaa-0003-4000-8000-000000000003',
|
||||
]
|
||||
|
||||
/** Fixed clock so the derived window is exactly assertable. */
|
||||
const NOW = new Date('2026-07-30T10:00:00.000Z')
|
||||
|
||||
interface RecordedUpdate {
|
||||
payload: Record<string, unknown>
|
||||
filters: Record<string, unknown>
|
||||
}
|
||||
|
||||
interface Recorded {
|
||||
rpcCalls: { fn: string; args: Record<string, unknown> }[]
|
||||
deliveryUpdates: RecordedUpdate[]
|
||||
}
|
||||
|
||||
/**
|
||||
* Ordered-log Supabase stub in the same house style as
|
||||
* dispatcher-auto-disable.test.ts. Every write against webhook_deliveries is
|
||||
* appended to `deliveryUpdates` in the order it resolves, so the interleaving
|
||||
* of touch writes and terminal writes is directly assertable.
|
||||
*
|
||||
* `lostOwnership` makes a specific delivery's touch resolve to zero rows,
|
||||
* which is how the real table reports "another cycle owns this row now".
|
||||
*/
|
||||
function makeSupabase(opts: {
|
||||
deliveryCount?: number
|
||||
attempts?: number
|
||||
lostOwnership?: Set<string>
|
||||
/** Rows the sweep RPC reports as recovered this tick. */
|
||||
recovered?: Array<{ id: string; status: string; attempts: number }>
|
||||
} = {}): { client: SupabaseClient; recorded: Recorded } {
|
||||
const recorded: Recorded = { rpcCalls: [], deliveryUpdates: [] }
|
||||
const count = opts.deliveryCount ?? 1
|
||||
|
||||
const client = {
|
||||
rpc: vi.fn(async (fn: string, args: Record<string, unknown>) => {
|
||||
recorded.rpcCalls.push({ fn, args })
|
||||
if (fn === 'claim_due_webhook_deliveries') {
|
||||
return {
|
||||
data: DELIVERY_IDS.slice(0, count).map((id) => ({
|
||||
id,
|
||||
webhook_id: WEBHOOK_ID,
|
||||
company_id: COMPANY_ID,
|
||||
event_type: 'invoice.paid',
|
||||
payload: { id: 'inv-1' },
|
||||
previous_attributes: null,
|
||||
api_version: '2026-05-12',
|
||||
attempts: opts.attempts ?? 0,
|
||||
})),
|
||||
error: null,
|
||||
}
|
||||
}
|
||||
// recover_stuck_webhook_deliveries: nothing stuck unless a case says so.
|
||||
return { data: opts.recovered ?? [], error: null }
|
||||
}),
|
||||
from: vi.fn((table: string) => {
|
||||
if (table === 'webhook_deliveries') {
|
||||
return {
|
||||
update: (payload: Record<string, unknown>) => {
|
||||
const filters: Record<string, unknown> = {}
|
||||
const rec: RecordedUpdate = { payload, filters }
|
||||
const chain: Record<string, unknown> = {
|
||||
eq: (col: string, val: unknown) => {
|
||||
filters[col] = val
|
||||
return chain
|
||||
},
|
||||
in: (col: string, val: unknown) => {
|
||||
filters[col] = val
|
||||
return chain
|
||||
},
|
||||
// touchInFlight terminates with .select('id')
|
||||
select: async () => {
|
||||
recorded.deliveryUpdates.push(rec)
|
||||
const id = filters.id as string
|
||||
const lost = opts.lostOwnership?.has(id) ?? false
|
||||
return { data: lost ? [] : [{ id }], error: null }
|
||||
},
|
||||
// markDelivered / markDead / markFailedForRetry await the chain
|
||||
then: (resolve: (v: { error: null }) => unknown) => {
|
||||
recorded.deliveryUpdates.push(rec)
|
||||
return resolve({ error: null })
|
||||
},
|
||||
}
|
||||
return chain
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
if (table === 'webhooks') {
|
||||
return {
|
||||
select: () => ({
|
||||
in: async () => ({
|
||||
data: [
|
||||
{
|
||||
id: WEBHOOK_ID,
|
||||
company_id: COMPANY_ID,
|
||||
webhook_url: 'https://receiver.example.com/hook',
|
||||
secret: 'whsec_test',
|
||||
},
|
||||
],
|
||||
error: null,
|
||||
}),
|
||||
}),
|
||||
}
|
||||
}
|
||||
|
||||
throw new Error(`unexpected table: ${table}`)
|
||||
}),
|
||||
} as unknown as SupabaseClient
|
||||
|
||||
return { client, recorded }
|
||||
}
|
||||
|
||||
/** Receiver answers 200 and records which delivery ids were actually POSTed. */
|
||||
function makeOkFetch(postedIds: string[]) {
|
||||
return async (
|
||||
_url: string,
|
||||
init: { headers: Record<string, string> },
|
||||
): Promise<PinnedFetchResult> => {
|
||||
postedIds.push(init.headers['X-Gnubok-Delivery'])
|
||||
return {
|
||||
kind: 'ok',
|
||||
status: 200,
|
||||
headers: { 'content-type': 'application/json' },
|
||||
body: '{"ok":true}',
|
||||
bodyTruncated: false,
|
||||
pinnedAddress: '93.184.216.34',
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function recoverArgs(recorded: Recorded): Record<string, unknown> {
|
||||
const call = recorded.rpcCalls.find((c) => c.fn === 'recover_stuck_webhook_deliveries')
|
||||
if (!call) throw new Error('recover_stuck_webhook_deliveries was never called')
|
||||
return call.args
|
||||
}
|
||||
|
||||
describe('dispatcher stuck-in_flight recovery window', () => {
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks()
|
||||
})
|
||||
|
||||
it('derives a window wider than one bounded cycle, so a mid-cycle row cannot be swept', async () => {
|
||||
const { client, recorded } = makeSupabase()
|
||||
|
||||
await dispatchDueDeliveries({
|
||||
supabase: client,
|
||||
now: NOW,
|
||||
pinnedFetchImpl: makeOkFetch([]) as never,
|
||||
})
|
||||
|
||||
const args = recoverArgs(recorded)
|
||||
const windowMs = NOW.getTime() - Date.parse(args.p_stuck_before as string)
|
||||
|
||||
// The cycle budget (what actually bounds a cycle, whatever its batch size)
|
||||
// plus one in-flight request plus slack. Spelled out as a literal as well
|
||||
// as a derivation so a change to any constant has to be deliberate.
|
||||
expect(windowMs).toBe(160_000)
|
||||
expect(windowMs).toBe(
|
||||
__TESTING__.CYCLE_BUDGET_MS +
|
||||
__TESTING__.REQUEST_TIMEOUT_MS +
|
||||
__TESTING__.STUCK_RECOVERY_SLACK_MS,
|
||||
)
|
||||
|
||||
// The old threshold. A concurrent tick must no longer be able to re-arm a
|
||||
// row that a live cycle claimed only a couple of attempts ago.
|
||||
expect(windowMs).toBeGreaterThan(__TESTING__.REQUEST_TIMEOUT_MS * 2)
|
||||
// And it must cover the whole bounded cycle, not just part of it.
|
||||
expect(windowMs).toBeGreaterThan(__TESTING__.CYCLE_BUDGET_MS)
|
||||
|
||||
expect(args.p_now).toBe(NOW.toISOString())
|
||||
})
|
||||
|
||||
it('uses the same window for the 5-row kick as for the 50-row cron', async () => {
|
||||
const windows: number[] = []
|
||||
for (const batchSize of [5, __TESTING__.DEFAULT_BATCH_SIZE, 500]) {
|
||||
const { client, recorded } = makeSupabase()
|
||||
await dispatchDueDeliveries({
|
||||
supabase: client,
|
||||
batchSize,
|
||||
now: NOW,
|
||||
pinnedFetchImpl: makeOkFetch([]) as never,
|
||||
})
|
||||
windows.push(NOW.getTime() - Date.parse(recoverArgs(recorded).p_stuck_before as string))
|
||||
}
|
||||
|
||||
// The sweep is tenant-global, so a small batch must not shrink the window
|
||||
// and a large one must not stretch it: the cycle budget bounds every
|
||||
// caller alike. A batch-derived window (the pre-#1257 shape, and the shape
|
||||
// a "floor at DEFAULT_BATCH_SIZE" would silently allow again above 50)
|
||||
// would produce 50_000 here for the kick and 160_000 for the cron.
|
||||
expect(windows).toEqual([160_000, 160_000, 160_000])
|
||||
})
|
||||
|
||||
it('hands the attempts cap and the retry schedule to the RPC', async () => {
|
||||
const { client, recorded } = makeSupabase()
|
||||
|
||||
await dispatchDueDeliveries({
|
||||
supabase: client,
|
||||
now: NOW,
|
||||
pinnedFetchImpl: makeOkFetch([]) as never,
|
||||
})
|
||||
|
||||
const args = recoverArgs(recorded)
|
||||
// The cap stays single-sourced in TS; the increment happens in SQL, which
|
||||
// is the only place `attempts = attempts + 1` can be expressed atomically.
|
||||
expect(args.p_max_attempts).toBe(__TESTING__.MAX_ATTEMPTS)
|
||||
expect(__TESTING__.MAX_ATTEMPTS).toBe(8)
|
||||
|
||||
// And so does the backoff: a swept row must wait exactly as long as a row
|
||||
// whose receiver answered 500. Re-arming at p_now let a repeatedly
|
||||
// stranded delivery spend all 8 attempts inside half an hour and land in
|
||||
// the terminal, immutable 'dead' state without ever being contacted.
|
||||
expect(args.p_backoff).toEqual([...__TESTING__.RETRY_BACKOFF_SECONDS])
|
||||
expect(args.p_backoff).toEqual([60, 300, 1800, 7200, 43200, 86400, 172800])
|
||||
})
|
||||
|
||||
it('surfaces the sweep outcome in the summary, dead rows separately', async () => {
|
||||
const { client } = makeSupabase({
|
||||
recovered: [
|
||||
{ id: DELIVERY_IDS[0], status: 'failed', attempts: 2 },
|
||||
{ id: DELIVERY_IDS[1], status: 'dead', attempts: 8 },
|
||||
{ id: DELIVERY_IDS[2], status: 'dead', attempts: 8 },
|
||||
],
|
||||
})
|
||||
|
||||
const summary = await dispatchDueDeliveries({
|
||||
supabase: client,
|
||||
now: NOW,
|
||||
pinnedFetchImpl: makeOkFetch([]) as never,
|
||||
})
|
||||
|
||||
// A tick that takes deliveries terminal has to be visible in the cron's
|
||||
// own summary line, not only in a helper-level log.warn.
|
||||
expect(summary.recovered).toBe(3)
|
||||
expect(summary.recoveredDead).toBe(2)
|
||||
})
|
||||
})
|
||||
|
||||
describe('dispatcher per-row in_flight re-stamp', () => {
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks()
|
||||
})
|
||||
|
||||
it('re-stamps each row immediately before its own attempt, not once at claim time', async () => {
|
||||
const { client, recorded } = makeSupabase({ deliveryCount: 3 })
|
||||
const posted: string[] = []
|
||||
|
||||
const summary = await dispatchDueDeliveries({
|
||||
supabase: client,
|
||||
now: NOW,
|
||||
pinnedFetchImpl: makeOkFetch(posted) as never,
|
||||
})
|
||||
|
||||
expect(summary.delivered).toBe(3)
|
||||
expect(posted).toEqual(DELIVERY_IDS)
|
||||
|
||||
// touch, terminal, touch, terminal, touch, terminal. A single re-stamp at
|
||||
// claim time (the pre-fix behaviour) would leave only the three terminal
|
||||
// writes here, and rows 2 and 3 would carry the claim's updated_at into
|
||||
// the next cycle's sweep window.
|
||||
expect(recorded.deliveryUpdates).toHaveLength(6)
|
||||
for (let i = 0; i < 3; i++) {
|
||||
const touch = recorded.deliveryUpdates[i * 2]
|
||||
expect(touch.payload).toEqual({ status: 'in_flight' })
|
||||
expect(touch.filters).toEqual({ id: DELIVERY_IDS[i], status: 'in_flight' })
|
||||
|
||||
const terminal = recorded.deliveryUpdates[i * 2 + 1]
|
||||
expect(terminal.payload).toMatchObject({ status: 'delivered' })
|
||||
expect(terminal.filters).toEqual({ id: DELIVERY_IDS[i] })
|
||||
}
|
||||
})
|
||||
|
||||
it('drops the POST when the touch shows the row is no longer ours', async () => {
|
||||
const { client, recorded } = makeSupabase({
|
||||
deliveryCount: 3,
|
||||
lostOwnership: new Set([DELIVERY_IDS[1]]),
|
||||
})
|
||||
const posted: string[] = []
|
||||
|
||||
const summary = await dispatchDueDeliveries({
|
||||
supabase: client,
|
||||
now: NOW,
|
||||
pinnedFetchImpl: makeOkFetch(posted) as never,
|
||||
})
|
||||
|
||||
// The duplicate POST is what the issue calls avoidable load: the loser of
|
||||
// the race would have had its terminal write swallowed by the immutability
|
||||
// trigger anyway.
|
||||
expect(posted).toEqual([DELIVERY_IDS[0], DELIVERY_IDS[2]])
|
||||
expect(posted).not.toContain(DELIVERY_IDS[1])
|
||||
expect(summary.skipped).toBe(1)
|
||||
expect(summary.delivered).toBe(2)
|
||||
|
||||
// No terminal write for the row we no longer own.
|
||||
const terminalForRow2 = recorded.deliveryUpdates.filter(
|
||||
(u) => u.filters.id === DELIVERY_IDS[1] && u.payload.status !== 'in_flight',
|
||||
)
|
||||
expect(terminalForRow2).toHaveLength(0)
|
||||
})
|
||||
|
||||
it('reconciles the summary: picked === delivered + failed + dead + skipped + released', async () => {
|
||||
const { client } = makeSupabase({
|
||||
deliveryCount: 3,
|
||||
lostOwnership: new Set([DELIVERY_IDS[2]]),
|
||||
})
|
||||
|
||||
const summary = await dispatchDueDeliveries({
|
||||
supabase: client,
|
||||
now: NOW,
|
||||
pinnedFetchImpl: makeOkFetch([]) as never,
|
||||
})
|
||||
|
||||
expect(summary.picked).toBe(3)
|
||||
expect(
|
||||
summary.delivered + summary.failed + summary.dead + summary.skipped + summary.released,
|
||||
).toBe(summary.picked)
|
||||
})
|
||||
})
|
||||
|
||||
describe('dispatcher cycle budget', () => {
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks()
|
||||
})
|
||||
|
||||
afterEach(() => {
|
||||
vi.restoreAllMocks()
|
||||
})
|
||||
|
||||
it('hands back unattempted claims instead of stranding them in in_flight', async () => {
|
||||
const { client, recorded } = makeSupabase({ deliveryCount: 3 })
|
||||
const posted: string[] = []
|
||||
|
||||
// `Date.now` is the dispatcher's only real-clock read (the injected `now`
|
||||
// covers everything else), so driving it directly is enough to simulate a
|
||||
// receiver that ate the whole budget on the first row.
|
||||
let clock = 1_000_000
|
||||
vi.spyOn(Date, 'now').mockImplementation(() => clock)
|
||||
|
||||
const slowFetch = async (
|
||||
_url: string,
|
||||
init: { headers: Record<string, string> },
|
||||
): Promise<PinnedFetchResult> => {
|
||||
posted.push(init.headers['X-Gnubok-Delivery'])
|
||||
clock += __TESTING__.CYCLE_BUDGET_MS
|
||||
return {
|
||||
kind: 'ok',
|
||||
status: 200,
|
||||
headers: { 'content-type': 'application/json' },
|
||||
body: '{"ok":true}',
|
||||
bodyTruncated: false,
|
||||
pinnedAddress: '93.184.216.34',
|
||||
}
|
||||
}
|
||||
|
||||
const summary = await dispatchDueDeliveries({
|
||||
supabase: client,
|
||||
now: NOW,
|
||||
pinnedFetchImpl: slowFetch as never,
|
||||
})
|
||||
|
||||
// Row 1 fits, rows 2 and 3 do not: they are released, not POSTed.
|
||||
expect(posted).toEqual([DELIVERY_IDS[0]])
|
||||
expect(summary.delivered).toBe(1)
|
||||
expect(summary.released).toBe(2)
|
||||
expect(summary.picked).toBe(3)
|
||||
|
||||
// Released rows go back to a re-claimable state immediately. Without this
|
||||
// they would sit in in_flight for the full 160 s window before any sweep
|
||||
// could re-arm them, and the sweep would then charge them an attempt they
|
||||
// never made.
|
||||
//
|
||||
// 'pending', not 'failed': these rows carry attempts = 0, so they had
|
||||
// never been attempted and their pre-claim status was necessarily
|
||||
// 'pending'. webhook_deliveries is customer-visible behandlingshistorik;
|
||||
// a delivery that was claimed and handed back without a single POST must
|
||||
// not read as a failure there.
|
||||
const release = recorded.deliveryUpdates.at(-1)
|
||||
expect(release?.payload).toEqual({
|
||||
status: 'pending',
|
||||
next_attempt_at: NOW.toISOString(),
|
||||
})
|
||||
expect(release?.filters).toEqual({
|
||||
id: [DELIVERY_IDS[1], DELIVERY_IDS[2]],
|
||||
status: 'in_flight',
|
||||
})
|
||||
// Not an attempt: `attempts` must not be bumped and the previous attempt's
|
||||
// diagnostic must survive.
|
||||
expect(release?.payload).not.toHaveProperty('attempts')
|
||||
expect(release?.payload).not.toHaveProperty('error')
|
||||
})
|
||||
|
||||
it('releases a previously attempted row back to failed, not pending', async () => {
|
||||
const { client, recorded } = makeSupabase({ deliveryCount: 3, attempts: 2 })
|
||||
const posted: string[] = []
|
||||
|
||||
let clock = 1_000_000
|
||||
vi.spyOn(Date, 'now').mockImplementation(() => clock)
|
||||
|
||||
const slowFetch = async (
|
||||
_url: string,
|
||||
init: { headers: Record<string, string> },
|
||||
): Promise<PinnedFetchResult> => {
|
||||
posted.push(init.headers['X-Gnubok-Delivery'])
|
||||
clock += __TESTING__.CYCLE_BUDGET_MS
|
||||
return {
|
||||
kind: 'ok',
|
||||
status: 200,
|
||||
headers: { 'content-type': 'application/json' },
|
||||
body: '{"ok":true}',
|
||||
bodyTruncated: false,
|
||||
pinnedAddress: '93.184.216.34',
|
||||
}
|
||||
}
|
||||
|
||||
const summary = await dispatchDueDeliveries({
|
||||
supabase: client,
|
||||
now: NOW,
|
||||
pinnedFetchImpl: slowFetch as never,
|
||||
})
|
||||
|
||||
expect(summary.released).toBe(2)
|
||||
// attempts > 0 means the row was already in the retry loop before this
|
||||
// cycle claimed it, so 'failed' is its real pre-claim status.
|
||||
const release = recorded.deliveryUpdates.at(-1)
|
||||
expect(release?.payload).toEqual({
|
||||
status: 'failed',
|
||||
next_attempt_at: NOW.toISOString(),
|
||||
})
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,547 @@
|
||||
import { randomUUID } from 'node:crypto'
|
||||
import { afterAll, describe, expect, it } from 'vitest'
|
||||
import { getClient, getPool, runAsServiceRole } from '@/tests/pg/setup'
|
||||
import { seedCompany } from '@/tests/pg/fixtures'
|
||||
|
||||
/**
|
||||
* recover_stuck_webhook_deliveries (migration 20260730123000, issue #1257).
|
||||
*
|
||||
* The sweep that pulls webhook deliveries out of in_flight moved from a
|
||||
* PostgREST chain into SQL so it can charge an attempt atomically and land a
|
||||
* row past MAX_ATTEMPTS on the same terminal state the normal retry path
|
||||
* produces. The properties that need real Postgres:
|
||||
*
|
||||
* - `attempts = attempts + 1` actually happens (PostgREST cannot express it)
|
||||
* - the re-armed row waits out the SAME backoff a normal failed attempt
|
||||
* waits, so a repeatedly stranded row cannot burn its 8 attempts in
|
||||
* minutes and land in the immutable 'dead' state uncontacted
|
||||
* - a row at the cap becomes 'dead' with attempts = p_max_attempts, and the
|
||||
* claim function then never picks it up again
|
||||
* - terminal rows are skipped by the outer WHERE, so
|
||||
* enforce_webhook_delivery_immutability never fires
|
||||
* - the service-role gate rejects everyone else
|
||||
*/
|
||||
|
||||
// ──────────────────────────────────────────────────────────────────────
|
||||
// Fixtures: parent webhook + child delivery (copied from
|
||||
// claim-due-webhook-deliveries.pg.test.ts, plus explicit updated_at)
|
||||
// ──────────────────────────────────────────────────────────────────────
|
||||
|
||||
const MAX_ATTEMPTS = 8
|
||||
|
||||
/**
|
||||
* RETRY_BACKOFF_SECONDS from lib/webhooks/dispatcher.ts, spelled out rather
|
||||
* than imported: this file asserts what the SQL does with the array it is
|
||||
* handed, and importing the constant would make a schedule change silently
|
||||
* rewrite the expectations too.
|
||||
*/
|
||||
const BACKOFF = [60, 5 * 60, 30 * 60, 2 * 60 * 60, 12 * 60 * 60, 24 * 60 * 60, 48 * 60 * 60]
|
||||
|
||||
/**
|
||||
* Every delivery this file seeds, so the non-terminal ones can be removed
|
||||
* again. The suite is tenant-global by nature (the sweep has no company
|
||||
* filter) and a leftover row that becomes due later shows up as a phantom
|
||||
* extra claim in the sibling claim-due suite, which reads the whole table.
|
||||
* Terminal rows are left alone: block_webhook_delivery_terminal_delete
|
||||
* forbids deleting them, and the claim function never picks them up anyway.
|
||||
*/
|
||||
const seededDeliveryIds: string[] = []
|
||||
|
||||
afterAll(async () => {
|
||||
if (seededDeliveryIds.length === 0) return
|
||||
await getPool().query(
|
||||
`DELETE FROM public.webhook_deliveries
|
||||
WHERE id = ANY($1::uuid[]) AND status NOT IN ('delivered', 'dead')`,
|
||||
[seededDeliveryIds],
|
||||
)
|
||||
})
|
||||
|
||||
async function insertWebhook(params: { companyId: string }): Promise<string> {
|
||||
const id = randomUUID()
|
||||
await getPool().query(
|
||||
`INSERT INTO public.webhooks
|
||||
(id, company_id, name, event_type, webhook_url, secret, active)
|
||||
VALUES ($1, $2, 'pg-test', 'invoice.paid', 'https://example.com/hook', $3, true)`,
|
||||
[id, params.companyId, `whsec_${randomUUID().replace(/-/g, '')}`],
|
||||
)
|
||||
return id
|
||||
}
|
||||
|
||||
/**
|
||||
* updated_at is only auto-stamped by the BEFORE UPDATE trigger, so a value
|
||||
* supplied at INSERT time sticks: that is what lets a row be seeded as
|
||||
* "abandoned N seconds ago".
|
||||
*/
|
||||
async function insertDelivery(params: {
|
||||
webhookId: string | null
|
||||
companyId: string
|
||||
status?: 'pending' | 'in_flight' | 'delivered' | 'failed' | 'dead'
|
||||
attempts?: number
|
||||
updatedAt?: string
|
||||
nextAttemptAt?: string
|
||||
responseStatus?: number | null
|
||||
}): Promise<string> {
|
||||
const id = randomUUID()
|
||||
await getPool().query(
|
||||
`INSERT INTO public.webhook_deliveries
|
||||
(id, webhook_id, company_id, event_type, payload, api_version,
|
||||
status, next_attempt_at, attempts, updated_at, response_status)
|
||||
VALUES ($1, $2, $3, 'invoice.paid', '{"hello":"world"}'::jsonb, '2026-05-12',
|
||||
$4, $5, $6, $7, $8)`,
|
||||
[
|
||||
id,
|
||||
params.webhookId,
|
||||
params.companyId,
|
||||
params.status ?? 'in_flight',
|
||||
params.nextAttemptAt ?? new Date().toISOString(),
|
||||
params.attempts ?? 0,
|
||||
params.updatedAt ?? new Date().toISOString(),
|
||||
params.responseStatus ?? null,
|
||||
],
|
||||
)
|
||||
seededDeliveryIds.push(id)
|
||||
return id
|
||||
}
|
||||
|
||||
interface DeliveryRow {
|
||||
status: string
|
||||
attempts: number
|
||||
error: string | null
|
||||
next_attempt_at: Date
|
||||
response_status: number | null
|
||||
}
|
||||
|
||||
async function readDelivery(id: string): Promise<DeliveryRow> {
|
||||
const r = await getPool().query<DeliveryRow>(
|
||||
`SELECT status, attempts, error, next_attempt_at, response_status
|
||||
FROM public.webhook_deliveries WHERE id = $1`,
|
||||
[id],
|
||||
)
|
||||
const row = r.rows[0]
|
||||
if (!row) throw new Error(`delivery ${id} not found`)
|
||||
return row
|
||||
}
|
||||
|
||||
/** A timestamp far enough in the past to be inside any sane sweep window. */
|
||||
function longAgo(seconds: number): string {
|
||||
return new Date(Date.now() - seconds * 1000).toISOString()
|
||||
}
|
||||
|
||||
/** Sweep boundary: rows older than this are abandoned. */
|
||||
function stuckBefore(seconds = 160): string {
|
||||
return new Date(Date.now() - seconds * 1000).toISOString()
|
||||
}
|
||||
|
||||
describe('recover_stuck_webhook_deliveries.pg', () => {
|
||||
it('charges an attempt and re-arms an abandoned in_flight row', async () => {
|
||||
const { companyId } = await seedCompany()
|
||||
const webhookId = await insertWebhook({ companyId })
|
||||
const deliveryId = await insertDelivery({
|
||||
webhookId,
|
||||
companyId,
|
||||
status: 'in_flight',
|
||||
attempts: 0,
|
||||
updatedAt: longAgo(600),
|
||||
})
|
||||
|
||||
const pNow = new Date()
|
||||
const returned = await runAsServiceRole(async (client) => {
|
||||
const r = await client.query<{ id: string; status: string; attempts: number }>(
|
||||
`SELECT * FROM public.recover_stuck_webhook_deliveries($1, $2, $3::int[], $4)`,
|
||||
[stuckBefore(), MAX_ATTEMPTS, BACKOFF, pNow.toISOString()],
|
||||
)
|
||||
return r.rows
|
||||
})
|
||||
|
||||
expect(returned.find((r) => r.id === deliveryId)).toMatchObject({
|
||||
status: 'failed',
|
||||
attempts: 1,
|
||||
})
|
||||
|
||||
const row = await readDelivery(deliveryId)
|
||||
// The undercount is the second half of #1257: without the increment a row
|
||||
// caught in the recover/re-claim loop is re-POSTed past MAX_ATTEMPTS.
|
||||
expect(row.attempts).toBe(1)
|
||||
expect(row.status).toBe('failed')
|
||||
expect(row.error).toBe('recovered_from_in_flight_timeout')
|
||||
// First backoff step, exactly like markFailedForRetry on a 0-attempt row.
|
||||
// p_now would make the row re-claimable on the next per-minute tick, which
|
||||
// is how a repeatedly stranded delivery burned all 8 attempts in minutes.
|
||||
expect(row.next_attempt_at.getTime()).toBe(pNow.getTime() + BACKOFF[0] * 1000)
|
||||
})
|
||||
|
||||
it('re-arms a swept row far enough out that the next tick cannot claim it', async () => {
|
||||
const { companyId } = await seedCompany()
|
||||
const webhookId = await insertWebhook({ companyId })
|
||||
const deliveryId = await insertDelivery({
|
||||
webhookId,
|
||||
companyId,
|
||||
status: 'in_flight',
|
||||
attempts: 0,
|
||||
updatedAt: longAgo(600),
|
||||
})
|
||||
|
||||
const pNow = new Date()
|
||||
await runAsServiceRole(async (client) => {
|
||||
await client.query(
|
||||
`SELECT * FROM public.recover_stuck_webhook_deliveries($1, $2, $3::int[], $4)`,
|
||||
[stuckBefore(), MAX_ATTEMPTS, BACKOFF, pNow.toISOString()],
|
||||
)
|
||||
})
|
||||
|
||||
// The claim function is the real predicate: a swept row must NOT be due
|
||||
// at the moment it was swept, nor a tick later. Rolled back so the claim's
|
||||
// in_flight flips do not leak into the sibling claim-due suite.
|
||||
const client = await getClient()
|
||||
try {
|
||||
await client.query('BEGIN')
|
||||
const { rows } = await client.query<{ id: string }>(
|
||||
`SELECT id FROM public.claim_due_webhook_deliveries($1, $2)`,
|
||||
[100, new Date(pNow.getTime() + 59_000).toISOString()],
|
||||
)
|
||||
expect(rows.map((r) => r.id)).not.toContain(deliveryId)
|
||||
} finally {
|
||||
await client.query('ROLLBACK').catch(() => {})
|
||||
client.release()
|
||||
}
|
||||
})
|
||||
|
||||
it('picks the backoff step for the attempt it just charged', async () => {
|
||||
const { companyId } = await seedCompany()
|
||||
const webhookId = await insertWebhook({ companyId })
|
||||
// attempts = 3 before the sweep, so the charged attempt is the 4th and the
|
||||
// wait is BACKOFF[3] (2 h): index = attempts BEFORE this one, exactly the
|
||||
// lookup markFailedForRetry does in TS, shifted by one for 1-indexed
|
||||
// Postgres arrays.
|
||||
const deliveryId = await insertDelivery({
|
||||
webhookId,
|
||||
companyId,
|
||||
status: 'in_flight',
|
||||
attempts: 3,
|
||||
updatedAt: longAgo(600),
|
||||
})
|
||||
|
||||
const pNow = new Date()
|
||||
await runAsServiceRole(async (client) => {
|
||||
await client.query(
|
||||
`SELECT * FROM public.recover_stuck_webhook_deliveries($1, $2, $3::int[], $4)`,
|
||||
[stuckBefore(), MAX_ATTEMPTS, BACKOFF, pNow.toISOString()],
|
||||
)
|
||||
})
|
||||
|
||||
const row = await readDelivery(deliveryId)
|
||||
expect(row.attempts).toBe(4)
|
||||
expect(row.next_attempt_at.getTime()).toBe(pNow.getTime() + BACKOFF[3] * 1000)
|
||||
})
|
||||
|
||||
it('clamps to the last backoff step instead of subscripting past the array', async () => {
|
||||
const { companyId } = await seedCompany()
|
||||
const webhookId = await insertWebhook({ companyId })
|
||||
const deliveryId = await insertDelivery({
|
||||
webhookId,
|
||||
companyId,
|
||||
status: 'in_flight',
|
||||
attempts: 4,
|
||||
updatedAt: longAgo(600),
|
||||
})
|
||||
|
||||
// A two-step schedule with a cap of 8: index 5 is past the end. Without
|
||||
// the least() clamp the subscript yields NULL and next_attempt_at would go
|
||||
// NULL, which the claim function reads as "not due, ever".
|
||||
const shortBackoff = [60, 300]
|
||||
const pNow = new Date()
|
||||
await runAsServiceRole(async (client) => {
|
||||
await client.query(
|
||||
`SELECT * FROM public.recover_stuck_webhook_deliveries($1, $2, $3::int[], $4)`,
|
||||
[stuckBefore(), MAX_ATTEMPTS, shortBackoff, pNow.toISOString()],
|
||||
)
|
||||
})
|
||||
|
||||
const row = await readDelivery(deliveryId)
|
||||
expect(row.attempts).toBe(5)
|
||||
expect(row.next_attempt_at).not.toBeNull()
|
||||
expect(row.next_attempt_at.getTime()).toBe(pNow.getTime() + 300 * 1000)
|
||||
})
|
||||
|
||||
it('lands a row at the cap on the same dead state the normal path produces', async () => {
|
||||
const { companyId } = await seedCompany()
|
||||
const webhookId = await insertWebhook({ companyId })
|
||||
const deliveryId = await insertDelivery({
|
||||
webhookId,
|
||||
companyId,
|
||||
status: 'in_flight',
|
||||
attempts: MAX_ATTEMPTS - 1,
|
||||
updatedAt: longAgo(600),
|
||||
responseStatus: 503,
|
||||
})
|
||||
|
||||
await runAsServiceRole(async (client) => {
|
||||
await client.query(
|
||||
`SELECT * FROM public.recover_stuck_webhook_deliveries($1, $2, $3::int[], now())`,
|
||||
[stuckBefore(), MAX_ATTEMPTS, BACKOFF],
|
||||
)
|
||||
})
|
||||
|
||||
const row = await readDelivery(deliveryId)
|
||||
expect(row.status).toBe('dead')
|
||||
expect(row.attempts).toBe(MAX_ATTEMPTS)
|
||||
expect(row.error).toBe('attempts_exhausted:in_flight_timeout')
|
||||
// The last recorded response is the only diagnostic left on an abandoned
|
||||
// attempt: recovery must not null it out the way markDead-without-outcome
|
||||
// does.
|
||||
expect(row.response_status).toBe(503)
|
||||
})
|
||||
|
||||
it('does not retry a recovered-at-cap row forever: the claim function skips it', async () => {
|
||||
const { companyId } = await seedCompany()
|
||||
const webhookId = await insertWebhook({ companyId })
|
||||
const deliveryId = await insertDelivery({
|
||||
webhookId,
|
||||
companyId,
|
||||
status: 'in_flight',
|
||||
attempts: MAX_ATTEMPTS - 1,
|
||||
updatedAt: longAgo(600),
|
||||
})
|
||||
|
||||
await runAsServiceRole(async (client) => {
|
||||
await client.query(
|
||||
`SELECT * FROM public.recover_stuck_webhook_deliveries($1, $2, $3::int[], now())`,
|
||||
[stuckBefore(), MAX_ATTEMPTS, BACKOFF],
|
||||
)
|
||||
})
|
||||
expect((await readDelivery(deliveryId)).status).toBe('dead')
|
||||
|
||||
// Inside a rolled-back transaction: the claim function flips every due row
|
||||
// in the table to in_flight, and this suite must not leave that behind for
|
||||
// the sibling claim-due file.
|
||||
const client = await getClient()
|
||||
try {
|
||||
await client.query('BEGIN')
|
||||
const { rows } = await client.query<{ id: string }>(
|
||||
`SELECT id FROM public.claim_due_webhook_deliveries($1, now())`,
|
||||
[100],
|
||||
)
|
||||
expect(rows.map((r) => r.id)).not.toContain(deliveryId)
|
||||
} finally {
|
||||
await client.query('ROLLBACK').catch(() => {})
|
||||
client.release()
|
||||
}
|
||||
})
|
||||
|
||||
it('a second sweep over the same rows does not trip the immutability trigger', async () => {
|
||||
const { companyId } = await seedCompany()
|
||||
const webhookId = await insertWebhook({ companyId })
|
||||
const deliveryId = await insertDelivery({
|
||||
webhookId,
|
||||
companyId,
|
||||
status: 'in_flight',
|
||||
attempts: MAX_ATTEMPTS - 1,
|
||||
updatedAt: longAgo(600),
|
||||
})
|
||||
|
||||
const sweep = () =>
|
||||
runAsServiceRole(async (client) => {
|
||||
await client.query(
|
||||
`SELECT * FROM public.recover_stuck_webhook_deliveries($1, $2, $3::int[], now())`,
|
||||
[stuckBefore(), MAX_ATTEMPTS, BACKOFF],
|
||||
)
|
||||
})
|
||||
|
||||
await sweep()
|
||||
// The row is now 'dead', so the WHERE excludes it and
|
||||
// enforce_webhook_delivery_immutability is never reached. A sweep that
|
||||
// matched terminal rows would abort the whole statement with a
|
||||
// check_violation.
|
||||
await expect(sweep()).resolves.toBeUndefined()
|
||||
|
||||
const row = await readDelivery(deliveryId)
|
||||
expect(row.status).toBe('dead')
|
||||
expect(row.attempts).toBe(MAX_ATTEMPTS)
|
||||
})
|
||||
|
||||
it('leaves a row a live cycle is still working through alone', async () => {
|
||||
const { companyId } = await seedCompany()
|
||||
const webhookId = await insertWebhook({ companyId })
|
||||
// Freshly re-stamped by touchInFlight right before its own attempt.
|
||||
const deliveryId = await insertDelivery({
|
||||
webhookId,
|
||||
companyId,
|
||||
status: 'in_flight',
|
||||
attempts: 2,
|
||||
updatedAt: new Date().toISOString(),
|
||||
})
|
||||
|
||||
const returned = await runAsServiceRole(async (client) => {
|
||||
const r = await client.query<{ id: string }>(
|
||||
`SELECT * FROM public.recover_stuck_webhook_deliveries($1, $2, $3::int[], now())`,
|
||||
[stuckBefore(), MAX_ATTEMPTS, BACKOFF],
|
||||
)
|
||||
return r.rows
|
||||
})
|
||||
|
||||
expect(returned.map((r) => r.id)).not.toContain(deliveryId)
|
||||
const row = await readDelivery(deliveryId)
|
||||
expect(row.status).toBe('in_flight')
|
||||
expect(row.attempts).toBe(2)
|
||||
})
|
||||
|
||||
it('never touches terminal rows, however old their updated_at is', async () => {
|
||||
const { companyId } = await seedCompany()
|
||||
const webhookId = await insertWebhook({ companyId })
|
||||
const deliveredId = await insertDelivery({
|
||||
webhookId,
|
||||
companyId,
|
||||
status: 'delivered',
|
||||
attempts: 1,
|
||||
updatedAt: longAgo(6000),
|
||||
})
|
||||
const deadId = await insertDelivery({
|
||||
webhookId,
|
||||
companyId,
|
||||
status: 'dead',
|
||||
attempts: MAX_ATTEMPTS,
|
||||
updatedAt: longAgo(6000),
|
||||
})
|
||||
|
||||
// No check_violation: the statement must complete, not abort.
|
||||
await expect(
|
||||
runAsServiceRole(async (client) => {
|
||||
await client.query(
|
||||
`SELECT * FROM public.recover_stuck_webhook_deliveries($1, $2, $3::int[], now())`,
|
||||
[stuckBefore(), MAX_ATTEMPTS, BACKOFF],
|
||||
)
|
||||
}),
|
||||
).resolves.toBeUndefined()
|
||||
|
||||
expect(await readDelivery(deliveredId)).toMatchObject({ status: 'delivered', attempts: 1 })
|
||||
expect(await readDelivery(deadId)).toMatchObject({ status: 'dead', attempts: MAX_ATTEMPTS })
|
||||
})
|
||||
|
||||
it('requires the service role', async () => {
|
||||
// Plain pool: superuser connection, auth.role() NULL. Nothing reachable by
|
||||
// anon or authenticated may re-arm another tenant's deliveries.
|
||||
await expect(
|
||||
getPool().query(
|
||||
`SELECT * FROM public.recover_stuck_webhook_deliveries($1, $2, $3::int[], now())`,
|
||||
[stuckBefore(), MAX_ATTEMPTS, BACKOFF],
|
||||
),
|
||||
).rejects.toThrow(/server-controlled service role/i)
|
||||
})
|
||||
|
||||
it('rejects a missing stuck boundary and a non-positive cap', async () => {
|
||||
await expect(
|
||||
runAsServiceRole(async (client) => {
|
||||
await client.query(
|
||||
`SELECT * FROM public.recover_stuck_webhook_deliveries($1, $2, $3::int[], now())`,
|
||||
[null, MAX_ATTEMPTS, BACKOFF],
|
||||
)
|
||||
}),
|
||||
).rejects.toThrow(/p_stuck_before is required/i)
|
||||
|
||||
await expect(
|
||||
runAsServiceRole(async (client) => {
|
||||
await client.query(
|
||||
`SELECT * FROM public.recover_stuck_webhook_deliveries($1, $2, $3::int[], now())`,
|
||||
[stuckBefore(), 0, BACKOFF],
|
||||
)
|
||||
}),
|
||||
).rejects.toThrow(/p_max_attempts must be > 0/i)
|
||||
|
||||
await expect(
|
||||
runAsServiceRole(async (client) => {
|
||||
await client.query(
|
||||
`SELECT * FROM public.recover_stuck_webhook_deliveries($1, $2, $3::int[], now())`,
|
||||
[stuckBefore(), null, BACKOFF],
|
||||
)
|
||||
}),
|
||||
).rejects.toThrow(/p_max_attempts must be > 0/i)
|
||||
})
|
||||
|
||||
it('rejects a missing or non-positive backoff schedule', async () => {
|
||||
// A NULL, empty or zero/negative schedule would re-arm swept rows for
|
||||
// immediate re-claim, which is the strand loop the backoff exists to stop.
|
||||
// Failing loudly beats silently reverting to next_attempt_at = p_now.
|
||||
for (const backoff of [null, [], [60, 0], [60, -5]]) {
|
||||
await expect(
|
||||
runAsServiceRole(async (client) => {
|
||||
await client.query(
|
||||
`SELECT * FROM public.recover_stuck_webhook_deliveries($1, $2, $3::int[], now())`,
|
||||
[stuckBefore(), MAX_ATTEMPTS, backoff],
|
||||
)
|
||||
}),
|
||||
).rejects.toThrow(/p_backoff/i)
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
/**
|
||||
* The other half of #1257 lives in TypeScript (touchInFlight), but it rests
|
||||
* entirely on two DB behaviours that no mock can prove. Pinned here because
|
||||
* if either stops holding, the fix silently reverts to the defect: a row
|
||||
* carrying its claim-time updated_at into the next cycle's sweep.
|
||||
*/
|
||||
describe('in_flight re-stamp: the DB behaviour touchInFlight relies on', () => {
|
||||
/** Byte-for-byte what PostgREST issues for the touchInFlight chain. */
|
||||
const TOUCH_SQL = `UPDATE public.webhook_deliveries
|
||||
SET status = 'in_flight'
|
||||
WHERE id = $1 AND status = 'in_flight'
|
||||
RETURNING id`
|
||||
|
||||
it('re-stamps updated_at on a no-op status write, pulling the row out of the sweep window', async () => {
|
||||
const { companyId } = await seedCompany()
|
||||
const webhookId = await insertWebhook({ companyId })
|
||||
// Claimed 10 minutes ago: deep inside any sweep window.
|
||||
const deliveryId = await insertDelivery({
|
||||
webhookId,
|
||||
companyId,
|
||||
status: 'in_flight',
|
||||
attempts: 1,
|
||||
updatedAt: longAgo(600),
|
||||
})
|
||||
|
||||
const touched = await getPool().query<{ id: string }>(TOUCH_SQL, [deliveryId])
|
||||
expect(touched.rowCount).toBe(1)
|
||||
|
||||
// Postgres runs the UPDATE even though no column value changed, so the
|
||||
// BEFORE UPDATE update_updated_at_column trigger (migration
|
||||
// 20260515200000) fires and re-stamps updated_at. Without that, writing
|
||||
// the status back verbatim would be an expensive no-op and the row's
|
||||
// in_flight age would still measure the claim, not the attempt.
|
||||
const { rows } = await getPool().query<{ updated_at: Date }>(
|
||||
`SELECT updated_at FROM public.webhook_deliveries WHERE id = $1`,
|
||||
[deliveryId],
|
||||
)
|
||||
expect(Date.now() - rows[0].updated_at.getTime()).toBeLessThan(60_000)
|
||||
|
||||
// And that is the property the sweep reads: the row was swept-eligible a
|
||||
// moment ago and is not any more.
|
||||
const swept = await runAsServiceRole(async (client) => {
|
||||
const r = await client.query<{ id: string }>(
|
||||
`SELECT * FROM public.recover_stuck_webhook_deliveries($1, $2, $3::int[], now())`,
|
||||
[stuckBefore(), MAX_ATTEMPTS, BACKOFF],
|
||||
)
|
||||
return r.rows
|
||||
})
|
||||
expect(swept.map((r) => r.id)).not.toContain(deliveryId)
|
||||
expect(await readDelivery(deliveryId)).toMatchObject({ status: 'in_flight', attempts: 1 })
|
||||
})
|
||||
|
||||
it('matches no row on a terminal delivery, so the immutability trigger never fires', async () => {
|
||||
const { companyId } = await seedCompany()
|
||||
const webhookId = await insertWebhook({ companyId })
|
||||
const deliveryId = await insertDelivery({
|
||||
webhookId,
|
||||
companyId,
|
||||
status: 'delivered',
|
||||
attempts: 1,
|
||||
updatedAt: longAgo(600),
|
||||
})
|
||||
|
||||
// The status filter is what keeps the touch off terminal rows. If it were
|
||||
// dropped, this would abort with a check_violation instead of returning
|
||||
// zero rows, and the dispatcher would read that as "row still ours".
|
||||
const touched = await getPool().query<{ id: string }>(TOUCH_SQL, [deliveryId])
|
||||
expect(touched.rowCount).toBe(0)
|
||||
expect(await readDelivery(deliveryId)).toMatchObject({ status: 'delivered', attempts: 1 })
|
||||
})
|
||||
})
|
||||
@@ -34,13 +34,27 @@
|
||||
* the same row at the same moment. That is a claim-time guarantee only, and it
|
||||
* is worth being precise about what it does NOT buy: the RPC autocommits, so
|
||||
* its locks are gone before any POST is issued, and from then on ownership is
|
||||
* just status='in_flight'. A later cycle's recoverStuckInFlight sweep can
|
||||
* re-arm a row that is still queued behind an earlier cycle's serial loop.
|
||||
* Delivery therefore stays at-least-once, exactly as the public docs promise
|
||||
* ("the same delivery id may arrive more than once ... idempotency is on
|
||||
* you"). The kick does not change that contract: the cron already claims 50
|
||||
* rows serially against the same 20 s stuck threshold, which is a wider window
|
||||
* than this batch of 5 can open.
|
||||
* just status='in_flight'. Delivery therefore stays at-least-once, exactly as
|
||||
* the public docs promise ("the same delivery id may arrive more than once ...
|
||||
* idempotency is on you").
|
||||
*
|
||||
* The kick does not widen that. Since #1257 the stuck sweep's window is a
|
||||
* constant derived from the dispatcher's cycle budget rather than from the
|
||||
* caller's batch size, so it is identical for both callers and always wider
|
||||
* than any live cycle: neither the cron nor this 5-row kick can re-arm a row
|
||||
* the other still owns. What remains is the genuine crash case, where a
|
||||
* hard-killed invocation's rows are recovered while its POSTs may still be in
|
||||
* flight; the ownership check in touchInFlight drops the loser's POST before
|
||||
* it is issued.
|
||||
*
|
||||
* One asymmetry is worth stating plainly: the cron route backs the dispatcher's
|
||||
* CYCLE_BUDGET_MS with `export const maxDuration = 300`, but this path cannot.
|
||||
* after() runs inside whatever arbitrary API route emitted the event, and that
|
||||
* route's maxDuration is not ours to set. It does not need to be: a kick claims
|
||||
* at most KICK_BATCH_SIZE (5) rows and each attempt is capped at
|
||||
* REQUEST_TIMEOUT_MS (10 s), so a kick cycle cannot exceed ~50 s of attempt
|
||||
* time and the dispatcher's 110 s budget check never fires here. The bound on
|
||||
* the kick is its batch size, not a clock.
|
||||
*/
|
||||
|
||||
import { after } from 'next/server'
|
||||
|
||||
+248
-32
@@ -45,6 +45,48 @@ const MAX_ATTEMPTS = RETRY_BACKOFF_SECONDS.length + 1 // initial + 7 retries = 8
|
||||
const REQUEST_TIMEOUT_MS = 10_000
|
||||
const MAX_RESPONSE_BODY_BYTES = 4096
|
||||
|
||||
/** Rows claimed per cycle when the caller passes nothing: the cron. */
|
||||
const DEFAULT_BATCH_SIZE = 50
|
||||
|
||||
/**
|
||||
* Wall-clock ceiling on one cycle's serial attempt loop. 50 rows at a 10 s
|
||||
* receiver timeout is 500 s of worst case, which no serverless invocation
|
||||
* survives, so bound the cycle explicitly instead of letting the platform
|
||||
* kill it mid-loop and leaving the remainder to the sweep.
|
||||
*
|
||||
* The cron route backs this with `export const maxDuration = 300`
|
||||
* (app/api/webhooks/dispatch/cron/route.ts): the budget is only a real bound
|
||||
* if the platform actually grants the invocation more wall time than the
|
||||
* budget spends, and the release path below only runs if the loop is still
|
||||
* alive when the budget expires.
|
||||
*/
|
||||
const CYCLE_BUDGET_MS = 120_000
|
||||
|
||||
/** Slack over the cycle bound: claim round trip, re-stamp, terminal write, clock skew. */
|
||||
const STUCK_RECOVERY_SLACK_MS = 30_000
|
||||
|
||||
/**
|
||||
* How long an in_flight row may legitimately sit before the sweep may treat
|
||||
* it as abandoned: 160 s. Derived, not guessed (#1257), and deliberately
|
||||
* independent of batch size.
|
||||
*
|
||||
* A cycle attempts its rows serially at up to REQUEST_TIMEOUT_MS each, but
|
||||
* the loop is cut off by CYCLE_BUDGET_MS regardless of how many rows were
|
||||
* claimed, so the BUDGET, not the row count, is what bounds a cycle's life:
|
||||
* budget + one in-flight request + slack. A 5-row kick and a 50-row cron
|
||||
* therefore use the same window, which is what the tenant-global sweep needs
|
||||
* (neither caller may re-arm a row the other still owns). Anything
|
||||
* batch-derived would let the kick pick a window narrower than the cron's
|
||||
* live cycle occupies, which is the shape #1257 was reported against.
|
||||
*
|
||||
* A plain constant rather than a function of batchSize on purpose: the
|
||||
* previous shape took a batchSize it could not act on (the clamp swallowed
|
||||
* every value) and documented a floor that never fired. If CYCLE_BUDGET_MS
|
||||
* changes this moves with it, and the cron's maxDuration must stay
|
||||
* comfortably above it.
|
||||
*/
|
||||
const STUCK_IN_FLIGHT_AFTER_MS = CYCLE_BUDGET_MS + REQUEST_TIMEOUT_MS + STUCK_RECOVERY_SLACK_MS
|
||||
|
||||
interface DueDelivery {
|
||||
id: string
|
||||
webhook_id: string
|
||||
@@ -68,6 +110,14 @@ export interface DispatchSummary {
|
||||
delivered: number
|
||||
failed: number
|
||||
dead: number
|
||||
/** Claimed but no longer ours by the time the attempt was due to start. */
|
||||
skipped: number
|
||||
/** Claimed but handed back unattempted because the cycle budget ran out. */
|
||||
released: number
|
||||
/** Rows this tick's sweep pulled out of an abandoned in_flight state. */
|
||||
recovered: number
|
||||
/** Subset of `recovered` the sweep took terminal (attempts exhausted). */
|
||||
recoveredDead: number
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -85,21 +135,41 @@ export async function dispatchDueDeliveries(args: {
|
||||
/** Override for tests; injected pinned-fetch implementation. */
|
||||
pinnedFetchImpl?: typeof pinnedHttpsFetch
|
||||
}): Promise<DispatchSummary> {
|
||||
const batchSize = args.batchSize ?? 50
|
||||
const batchSize = args.batchSize ?? DEFAULT_BATCH_SIZE
|
||||
const now = args.now ?? new Date()
|
||||
const pinnedFetchImpl = args.pinnedFetchImpl ?? pinnedHttpsFetch
|
||||
|
||||
const summary: DispatchSummary = { picked: 0, delivered: 0, failed: 0, dead: 0 }
|
||||
const summary: DispatchSummary = {
|
||||
picked: 0,
|
||||
delivered: 0,
|
||||
failed: 0,
|
||||
dead: 0,
|
||||
skipped: 0,
|
||||
released: 0,
|
||||
recovered: 0,
|
||||
recoveredDead: 0,
|
||||
}
|
||||
|
||||
// Recover stuck in_flight rows: a previous tick that was killed mid-flight
|
||||
// (Vercel function timeout, hard crash, manual termination) leaves rows
|
||||
// marked in_flight forever otherwise. Sweep them back to 'failed' so the
|
||||
// retry loop picks them up at next_attempt_at.
|
||||
// retry loop picks them up at next_attempt_at, on the same backoff schedule
|
||||
// a normal failed attempt gets.
|
||||
//
|
||||
// Threshold = 2× REQUEST_TIMEOUT_MS. A live attempt takes at most
|
||||
// REQUEST_TIMEOUT_MS plus the body read; doubling that gives an
|
||||
// unambiguous "this is stuck, not in-flight" boundary.
|
||||
await recoverStuckInFlight(args.supabase, now)
|
||||
// The window is STUCK_IN_FLIGHT_AFTER_MS: the bounded cycle
|
||||
// (CYCLE_BUDGET_MS, backed by the route's maxDuration) plus one receiver
|
||||
// timeout plus slack. It is deliberately WIDER than any live cycle can be,
|
||||
// so a row still queued behind an earlier cycle's serial loop is never
|
||||
// re-armed under it (#1257). The old fixed 2x REQUEST_TIMEOUT_MS was
|
||||
// narrower than a single cycle, which made every cycle recover the rows
|
||||
// the previous cycle was still working through.
|
||||
const recovery = await recoverStuckInFlight(args.supabase, now)
|
||||
summary.recovered = recovery.recovered
|
||||
summary.recoveredDead = recovery.dead
|
||||
|
||||
// Real clock, not the injectable `now`: the budget bounds this invocation's
|
||||
// wall time, so a fixed test `now` must not be able to trip it.
|
||||
const cycleStartedAt = Date.now()
|
||||
|
||||
const due = await claimDueDeliveries(args.supabase, batchSize, now)
|
||||
summary.picked = due.length
|
||||
@@ -109,7 +179,24 @@ export async function dispatchDueDeliveries(args: {
|
||||
const webhookIds = Array.from(new Set(due.map((d) => d.webhook_id)))
|
||||
const webhookMap = await loadWebhooksByIds(args.supabase, webhookIds)
|
||||
|
||||
for (const delivery of due) {
|
||||
for (const [index, delivery] of due.entries()) {
|
||||
// Out of budget: hand back everything we will not reach rather than
|
||||
// letting the platform kill the invocation mid-loop and leaving the
|
||||
// remainder stranded in in_flight until a sweep re-arms it.
|
||||
if (Date.now() - cycleStartedAt >= CYCLE_BUDGET_MS - REQUEST_TIMEOUT_MS) {
|
||||
const remaining = due.slice(index)
|
||||
await releaseUnattempted(
|
||||
args.supabase,
|
||||
remaining.map((d) => ({ id: d.id, attempts: d.attempts })),
|
||||
now,
|
||||
)
|
||||
summary.released = remaining.length
|
||||
log.warn('cycle budget exhausted: unattempted claims released', {
|
||||
count: remaining.length,
|
||||
})
|
||||
break
|
||||
}
|
||||
|
||||
const webhook = webhookMap.get(delivery.webhook_id)
|
||||
if (!webhook) {
|
||||
// The webhook was deleted between enqueue and dispatch. Mark dead;
|
||||
@@ -137,6 +224,19 @@ export async function dispatchDueDeliveries(args: {
|
||||
continue
|
||||
}
|
||||
|
||||
// Re-stamp updated_at immediately before this row's own attempt, so the
|
||||
// row's in_flight age measures the attempt rather than the claim, and use
|
||||
// the same write as an ownership check (#1257).
|
||||
if (!(await touchInFlight(args.supabase, delivery.id))) {
|
||||
log.info('delivery skipped: no longer in_flight', {
|
||||
deliveryId: delivery.id,
|
||||
webhookId: webhook.id,
|
||||
companyId: delivery.company_id,
|
||||
})
|
||||
summary.skipped++
|
||||
continue
|
||||
}
|
||||
|
||||
const outcome = await attemptDelivery({
|
||||
delivery,
|
||||
webhook,
|
||||
@@ -193,37 +293,64 @@ export async function dispatchDueDeliveries(args: {
|
||||
// ──────────────────────────────────────────────────────────────────────
|
||||
|
||||
/**
|
||||
* Mark in_flight rows whose updated_at is older than the stuck-threshold
|
||||
* back to 'failed' with next_attempt_at = now so they re-enter the
|
||||
* dispatch queue. Best-effort: a write failure here is logged but
|
||||
* Sweep in_flight rows whose updated_at is older than the stuck-threshold
|
||||
* back into the retry queue, charging an attempt for the stall so
|
||||
* MAX_ATTEMPTS stays a real cap. Best-effort: a failure here is logged but
|
||||
* doesn't block the rest of the cycle.
|
||||
*
|
||||
* The predicate now lives in SQL (recover_stuck_webhook_deliveries,
|
||||
* migration 20260730123000) rather than in a PostgREST chain, because
|
||||
* PostgREST can express neither `attempts = attempts + 1` nor the
|
||||
* conditional flip to 'dead' at the cap, and a read-then-write loop would
|
||||
* reopen a TOCTOU against enforce_webhook_delivery_immutability. MAX_ATTEMPTS
|
||||
* and RETRY_BACKOFF_SECONDS are passed in so the cap AND the schedule stay
|
||||
* single-sourced here in TS.
|
||||
*
|
||||
* A swept row is re-armed on the SAME backoff the normal failure path uses
|
||||
* (markFailedForRetry), not immediately: a row that keeps getting stranded
|
||||
* (deploy, instance recycle, a cycle that outlives its invocation) would
|
||||
* otherwise be re-claimed on the very next per-minute tick, and could burn
|
||||
* all MAX_ATTEMPTS in under half an hour and land in the terminal, immutable
|
||||
* 'dead' state without its receiver ever having been contacted. The ~87 h
|
||||
* retry schedule is the delivery guarantee; the sweep must not shorten it.
|
||||
*
|
||||
* Under READ COMMITTED (Postgres default), UPDATE re-evaluates the WHERE
|
||||
* clause against each row's current value when it acquires the row lock.
|
||||
* The function keeps `status = 'in_flight'` in the outer UPDATE's WHERE for
|
||||
* exactly that reason: a row that raced from 'in_flight' to
|
||||
* 'delivered'/'dead' between scan and lock fails re-evaluation and is
|
||||
* skipped entirely, so the immutability trigger never fires and a mid-flight
|
||||
* terminal flip cannot abort the sweep.
|
||||
*/
|
||||
async function recoverStuckInFlight(supabase: SupabaseClient, now: Date): Promise<void> {
|
||||
const stuckBefore = new Date(now.getTime() - 2 * REQUEST_TIMEOUT_MS)
|
||||
// Under READ COMMITTED (Postgres default), UPDATE re-evaluates the WHERE
|
||||
// clause against each row's current value when it acquires the row lock.
|
||||
// A row that raced from 'in_flight' to 'delivered'/'dead' between scan
|
||||
// and lock will fail status='in_flight' on re-evaluation and be skipped
|
||||
// entirely: the immutability trigger never fires, so a mid-flight
|
||||
// terminal flip cannot abort the bulk update.
|
||||
const { data, error } = await supabase
|
||||
.from('webhook_deliveries')
|
||||
.update({
|
||||
status: 'failed',
|
||||
next_attempt_at: now.toISOString(),
|
||||
error: 'recovered_from_in_flight_timeout',
|
||||
})
|
||||
.eq('status', 'in_flight')
|
||||
.lt('updated_at', stuckBefore.toISOString())
|
||||
.select('id')
|
||||
async function recoverStuckInFlight(
|
||||
supabase: SupabaseClient,
|
||||
now: Date,
|
||||
): Promise<{ recovered: number; dead: number }> {
|
||||
const stuckBefore = new Date(now.getTime() - STUCK_IN_FLIGHT_AFTER_MS)
|
||||
const { data, error } = await supabase.rpc('recover_stuck_webhook_deliveries', {
|
||||
p_stuck_before: stuckBefore.toISOString(),
|
||||
p_max_attempts: MAX_ATTEMPTS,
|
||||
p_backoff: [...RETRY_BACKOFF_SECONDS],
|
||||
p_now: now.toISOString(),
|
||||
})
|
||||
|
||||
if (error) {
|
||||
log.warn('stuck in_flight recovery failed', { code: error.code })
|
||||
return
|
||||
return { recovered: 0, dead: 0 }
|
||||
}
|
||||
if (data && data.length > 0) {
|
||||
log.warn('recovered stuck in_flight rows', { count: data.length })
|
||||
const rows = (data ?? []) as Array<{ id: string; status: string; attempts: number }>
|
||||
const dead = rows.filter((r) => r.status === 'dead').length
|
||||
if (rows.length > 0) {
|
||||
log.warn('recovered stuck in_flight rows', {
|
||||
count: rows.length,
|
||||
dead,
|
||||
retrying: rows.length - dead,
|
||||
})
|
||||
}
|
||||
// Counts are returned as well as logged: a tick that takes deliveries
|
||||
// terminal is the event worth alerting on, and it must be visible in the
|
||||
// cron's own summary line rather than only in a helper-level warn.
|
||||
return { recovered: rows.length, dead }
|
||||
}
|
||||
|
||||
async function claimDueDeliveries(
|
||||
@@ -321,6 +448,91 @@ async function markFailedForRetry(
|
||||
if (error) log.warn('mark failed-for-retry update failed', { id, code: error.code })
|
||||
}
|
||||
|
||||
/**
|
||||
* Re-stamp a claimed row's updated_at right before its own attempt starts,
|
||||
* and report whether the row is still ours.
|
||||
*
|
||||
* `status` is written back verbatim: Postgres runs the UPDATE regardless of
|
||||
* whether any value changed, so the table's BEFORE UPDATE
|
||||
* update_updated_at_column trigger (migration 20260515200000) re-stamps
|
||||
* updated_at. That is what makes a row's in_flight age measure its attempt
|
||||
* instead of the moment the whole batch was claimed, which is the property
|
||||
* the stuck sweep reads.
|
||||
*
|
||||
* The `.eq('status', 'in_flight')` filter keeps the write off terminal rows,
|
||||
* so enforce_webhook_delivery_immutability never fires. A zero-row result
|
||||
* means the delivery is no longer ours (another cycle recovered and re-claimed
|
||||
* it, or it already reached a terminal state): skip it rather than POSTing a
|
||||
* duplicate whose terminal write would lose the race anyway.
|
||||
*/
|
||||
async function touchInFlight(supabase: SupabaseClient, id: string): Promise<boolean> {
|
||||
const { data, error } = await supabase
|
||||
.from('webhook_deliveries')
|
||||
.update({ status: 'in_flight' })
|
||||
.eq('id', id)
|
||||
.eq('status', 'in_flight')
|
||||
.select('id')
|
||||
|
||||
if (error) {
|
||||
// Infrastructure hiccup, not lost ownership: proceed. Worst case the row
|
||||
// looks older than it is and a later sweep re-arms it, which is exactly
|
||||
// the pre-existing behaviour.
|
||||
log.warn('in_flight touch failed', { id, code: error.code })
|
||||
return true
|
||||
}
|
||||
return (data?.length ?? 0) > 0
|
||||
}
|
||||
|
||||
/**
|
||||
* Hand back rows this cycle claimed but will not attempt, so they are
|
||||
* re-claimable immediately instead of waiting out the stuck window.
|
||||
*
|
||||
* Deliberately does NOT bump attempts (these rows were never attempted) and
|
||||
* deliberately does NOT write `error` (a release is not a failure; overwriting
|
||||
* would destroy the previous attempt's diagnostic).
|
||||
*
|
||||
* The status written back is the row's PRE-CLAIM status, reconstructed from
|
||||
* `attempts` rather than guessed: claim_due_webhook_deliveries accepts
|
||||
* status IN ('pending','failed'), and every path that writes 'failed' also
|
||||
* writes attempts >= 1 (markFailedForRetry, the recovery sweep), while rows
|
||||
* enter the table as 'pending' with attempts = 0. So attempts = 0 means the
|
||||
* row was 'pending' and must go back to 'pending'. This matters because
|
||||
* webhook_deliveries is behandlingshistorik (BFNAR 2013:2 kap 8 §) exposed on
|
||||
* GET /webhooks/{id}/deliveries: a brand-new delivery that was claimed and
|
||||
* handed back without a single POST must not read as 'failed' to the
|
||||
* customer. Restoring the status verbatim would need the claim RPC to return
|
||||
* it, which it does not; the attempts inference gets the same answer without
|
||||
* changing a shipped function.
|
||||
*/
|
||||
async function releaseUnattempted(
|
||||
supabase: SupabaseClient,
|
||||
rows: Array<{ id: string; attempts: number }>,
|
||||
now: Date,
|
||||
): Promise<void> {
|
||||
if (rows.length === 0) return
|
||||
|
||||
const byPriorStatus: Array<{ status: 'pending' | 'failed'; ids: string[] }> = [
|
||||
{ status: 'pending', ids: rows.filter((r) => r.attempts === 0).map((r) => r.id) },
|
||||
{ status: 'failed', ids: rows.filter((r) => r.attempts > 0).map((r) => r.id) },
|
||||
]
|
||||
|
||||
for (const group of byPriorStatus) {
|
||||
if (group.ids.length === 0) continue
|
||||
const { error } = await supabase
|
||||
.from('webhook_deliveries')
|
||||
.update({ status: group.status, next_attempt_at: now.toISOString() })
|
||||
.in('id', group.ids)
|
||||
.eq('status', 'in_flight')
|
||||
if (error) {
|
||||
log.warn('release of unattempted claims failed', {
|
||||
count: group.ids.length,
|
||||
status: group.status,
|
||||
code: error.code,
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async function markDead(
|
||||
supabase: SupabaseClient,
|
||||
id: string,
|
||||
@@ -651,4 +863,8 @@ export const __TESTING__ = {
|
||||
MAX_ATTEMPTS,
|
||||
REQUEST_TIMEOUT_MS,
|
||||
MAX_RESPONSE_BODY_BYTES,
|
||||
DEFAULT_BATCH_SIZE,
|
||||
CYCLE_BUDGET_MS,
|
||||
STUCK_RECOVERY_SLACK_MS,
|
||||
STUCK_IN_FLIGHT_AFTER_MS,
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user