diff --git a/apps/web/src/app/api/webhooks/deliver/route.ts b/apps/web/src/app/api/webhooks/deliver/route.ts index a90f75f..94c0662 100644 --- a/apps/web/src/app/api/webhooks/deliver/route.ts +++ b/apps/web/src/app/api/webhooks/deliver/route.ts @@ -1,6 +1,6 @@ import { NextResponse } from 'next/server'; import { withClient, ensureSchema } from '@/lib/db'; -import { deliverDue } from '@/lib/webhooks'; +import { deliverDue, pendingDue } from '@/lib/webhooks'; export const dynamic = 'force-dynamic'; export const maxDuration = 30; @@ -23,7 +23,11 @@ export async function GET(request: Request) { try { const result = await withClient(async (client) => { await ensureSchema(client); - return deliverDue(client); + const outcome = await deliverDue(client); + // Remaining lag after this run: the signal a scheduler uses to scale + // consumer frequency to backlog (#165). + const lag = await pendingDue(client); + return { ...outcome, lag }; }); return NextResponse.json({ success: true, ...result }); } catch (error: unknown) { diff --git a/apps/web/src/app/api/webhooks/route.ts b/apps/web/src/app/api/webhooks/route.ts index 1d18bf5..8599e88 100644 --- a/apps/web/src/app/api/webhooks/route.ts +++ b/apps/web/src/app/api/webhooks/route.ts @@ -14,7 +14,9 @@ export async function GET() { configured: false, pending: 0, failed: 0, + deadLetter: 0, delivered: 0, + lag: 0, recentFailed: [], }); } diff --git a/apps/web/src/lib/webhooks.test.ts b/apps/web/src/lib/webhooks.test.ts index 7bafeef..5835626 100644 --- a/apps/web/src/lib/webhooks.test.ts +++ b/apps/web/src/lib/webhooks.test.ts @@ -10,6 +10,8 @@ import { deliverDue, enqueueWebhookDelivery, payloadFromRow, + pendingDue, + webhookSummary, } from './webhooks'; describe('shouldRetry', () => { @@ -176,3 +178,78 @@ describe('deliverDue — a sleeping host cannot stall the caller past the budget expect(elapsed).toBeLessThan(5_000); }); }); + +describe('pendingDue — the lag signal a consumer fleet scales on (#165)', () => { + it('counts only pending rows whose retry time has passed (or never set)', async () => { + const queries: string[] = []; + const query = vi.fn(async (sql: string) => { + queries.push(sql); + return { rows: [{ count: '7' }] }; + }); + + const lag = await pendingDue({ query } as never, { now: new Date('2026-08-01T00:00:00Z') }); + + expect(lag).toBe(7); + const sql = queries[0]; + expect(sql).toContain("status = 'pending'"); + // Null next_retry_at (never attempted) is always due. + expect(sql).toContain('next_retry_at IS NULL OR next_retry_at <= $1::timestamptz'); + }); +}); + +describe('webhookSummary', () => { + it('reports lag and the dead-letter count alongside the existing tallies', async () => { + const query = vi.fn(async (sql: string) => { + if (/^SELECT status, count/.test(sql)) { + return { + rows: [ + { status: 'pending', n: '2' }, + { status: 'delivered', n: '5' }, + { status: 'dead_letter', n: '3' }, + ], + }; + } + if (/^SELECT count\(\*\)::text AS count/.test(sql)) { + return { rows: [{ count: '2' }] }; + } + return { rows: [] }; + }); + + const summary = await webhookSummary({ query } as never); + + expect(summary.pending).toBe(2); + expect(summary.delivered).toBe(5); + expect(summary.deadLetter).toBe(3); + expect(summary.lag).toBe(2); + expect(summary.recentFailed).toEqual([]); + }); + + it('lists dead-lettered deliveries in recentFailed for operator inspection', async () => { + const query = vi.fn(async (sql: string) => { + if (/^SELECT status, count/.test(sql)) return { rows: [{ status: 'dead_letter', n: '1' }] }; + if (/^SELECT count\(\*\)::text AS count/.test(sql)) return { rows: [{ count: '0' }] }; + if (/^SELECT id, payment_tx_hash/.test(sql)) { + return { + rows: [ + { + id: '9', + payment_tx_hash: 'a'.repeat(64), + status: 'dead_letter', + attempts: 8, + last_status_code: 503, + last_error: 'HTTP 503', + updated_at: new Date('2026-08-01T00:00:00.000Z'), + }, + ], + }; + } + return { rows: [] }; + }); + + const summary = await webhookSummary({ query } as never); + + expect(summary.deadLetter).toBe(1); + expect(summary.recentFailed[0].status).toBe('dead_letter'); + expect(summary.recentFailed[0].attempts).toBe(8); + }); +}); diff --git a/apps/web/src/lib/webhooks.ts b/apps/web/src/lib/webhooks.ts index dde3c39..9afd462 100644 --- a/apps/web/src/lib/webhooks.ts +++ b/apps/web/src/lib/webhooks.ts @@ -15,7 +15,12 @@ export const DELIVERY_WINDOW_MS = 24 * 60 * 60 * 1000; export const ATTEMPT_TIMEOUT_MS = 2_000; export const MAX_BACKOFF_MS = 60 * 60 * 1000; -export type DeliveryStatus = 'pending' | 'delivering' | 'delivered' | 'failed'; +export type DeliveryStatus = + | 'pending' + | 'delivering' + | 'delivered' + | 'failed' + | 'dead_letter'; export interface PaymentPayload { tx_hash: string; @@ -274,7 +279,9 @@ export async function deliverDue( transportError, }); if (terminal.status === 'delivered') delivered++; - else if (terminal.status === 'failed') failed++; + // A dead-lettered row is terminal, not a retry — count it as failed here + // so the run's tallies reflect deliveries that gave up (#165). + else if (terminal.status === 'failed' || terminal.status === 'dead_letter') failed++; else retried++; } @@ -308,7 +315,11 @@ async function recordAttempt( let status: DeliveryStatus; if (ok) status = 'delivered'; else if (next) status = 'pending'; - else status = 'failed'; + // A delivery that exhausts its attempt budget (or its 24h delivery window) + // is dead-lettered: it is kept for operator inspection and never retried + // again (#165). This is the queue's explicit dead-letter state, distinct + // from a transient 'failed' row. + else status = 'dead_letter'; await client.query( `INSERT INTO webhook_attempts (delivery_id, attempt_number, status_code, error) @@ -332,10 +343,37 @@ async function recordAttempt( return { id: input.id, status, statusCode: input.statusCode, error: input.error }; } +/** + * The queue's lag: deliveries that are due for (re)delivery right now. + * + * This is the signal a consumer fleet auto-scales on (#165) — when lag stays + * high, schedule more frequent or overlapping `/api/webhooks/deliver` runs; + * when it is zero, the queue is drained. Rows with a null `next_retry_at` + * (never attempted) are always due; the rest are due once their retry time + * has passed. + */ +export async function pendingDue( + client: Client, + opts: { now?: Date } = {}, +): Promise { + const now = (opts.now ?? new Date()).toISOString(); + const res = await client.query<{ count: string }>( + `SELECT count(*)::text AS count + FROM webhook_deliveries + WHERE status = 'pending' + AND (next_retry_at IS NULL OR next_retry_at <= $1::timestamptz)`, + [now], + ); + return Number(res.rows[0]?.count ?? 0); +} + export async function webhookSummary(client: Client): Promise<{ pending: number; failed: number; + deadLetter: number; delivered: number; + /** Deliveries due right now — the lag a consumer fleet scales on (#165). */ + lag: number; recentFailed: Array<{ id: number; paymentTxHash: string; @@ -349,7 +387,12 @@ export async function webhookSummary(client: Client): Promise<{ const counts = await client.query<{ status: string; n: string }>( `SELECT status, count(*)::text AS n FROM webhook_deliveries GROUP BY status`, ); - const byStatus: Record = { pending: 0, failed: 0, delivered: 0 }; + const byStatus: Record = { + pending: 0, + failed: 0, + delivered: 0, + dead_letter: 0, + }; for (const row of counts.rows) byStatus[row.status] = Number(row.n); const recent = await client.query<{ @@ -363,7 +406,7 @@ export async function webhookSummary(client: Client): Promise<{ }>( `SELECT id, payment_tx_hash, status, attempts, last_status_code, last_error, updated_at FROM webhook_deliveries - WHERE status = 'failed' + WHERE status IN ('failed', 'dead_letter') ORDER BY updated_at DESC LIMIT 20`, ); @@ -371,7 +414,9 @@ export async function webhookSummary(client: Client): Promise<{ return { pending: (byStatus.pending ?? 0) + (byStatus.delivering ?? 0), failed: byStatus.failed ?? 0, + deadLetter: byStatus.dead_letter ?? 0, delivered: byStatus.delivered ?? 0, + lag: await pendingDue(client), recentFailed: recent.rows.map((row) => ({ id: Number(row.id), paymentTxHash: row.payment_tx_hash,