fix(queue): enforce max 5 retries and make queues healthy

AGENTS.md

@@ -130,6 +130,17 @@ backfill code, design for this scale:
 - Queue consumers must be sized from measured throughput: batch size,
   concurrency, visibility timeout, retry count, provider/API limits, and caller
   timeout must all fit the same worst-case calculation.
+- **HARD RULE — max 5 retries for every queue:** `MAX_QUEUE_READS = 5` in
+  `supabase/functions/_backend/triggers/queue_consumer.ts` is the global ceiling
+  for pgmq `read_ct`. Never raise it. Never add a per-queue exception (including
+  `on_version_update`, webhook queues, or cron queues). Application-level
+  retries (e.g. `webhook_deliveries.max_attempts` / `WEBHOOK_MAX_ATTEMPTS`) must
+  also stay at **5 or less**. If work needs more passes (large deleted
+  manifests, delayed webhook delivery, etc.), commit progress and re-enqueue via
+  a sweeper/cron (`sweep_deleted_version_manifests`, delayed `pgmq.send`, etc.) —
+  do not burn queue reads as a fake progress budget. `/queue_health` treats
+  `read_ct > 5` as unhealthy (`stuck_high_read_ct`); `cleanup_queue_messages`
+  deletes those poison rows.
 - Do not assume a `202` HTTP response means queue work finished. If the handler
   uses background work, prove the work can finish inside the runtime limits and
   that successful queue messages are deleted before visibility timeout expires.
@@ -145,6 +156,28 @@ backfill code, design for this scale:
   HTTP/runtime logs. If the current logs cannot identify the cause, first add
   logging that will identify it in the next occurrence.
 
+### Queue Retry Budget (HARD RULE)
+
+**Retries must never exceed 5 for any queue or queue-backed delivery path.**
+
+| Layer | Cap | Source of truth |
+| --- | --- | --- |
+| pgmq consumer `read_ct` | **5** | `MAX_QUEUE_READS` in `queue_consumer.ts` |
+| Discord / skip-archive budget | **5** | `getQueueMaxReads()` → always `MAX_QUEUE_READS` |
+| `/queue_health` stuck threshold | **5** | `STUCK_READ_CT_THRESHOLD` |
+| `cleanup_queue_messages` poison delete | **`read_ct > 5`** | SQL cleanup cron |
+| Webhook delivery attempts | **5** | `WEBHOOK_MAX_ATTEMPTS` + `webhook_deliveries.max_attempts` default |
+
+Do **not**:
+- Add `VERSION_QUEUE_MAX_READS` (or any other per-queue) above 5
+- Lengthen webhook retry ladders past a few hours / past 5 attempts
+- Treat incomplete multi-pass work as “just raise retries”
+
+Do:
+- Size `cron_tasks.batch_size` / intervals so visible backlog drains
+- Re-enqueue unfinished work via sweepers (e.g. deleted-manifest cleanup)
+- Keep `/queue_health` green: no visible `read_ct=0` stale msgs, no `read_ct > 5`
+
 ### AI Workflow Notes
 
 - **Hono v4 HEAD routing:** do not add HEAD routes with `app.on`. Hono v4 removed `app.head()` because `GET` handlers implicitly serve `HEAD`; keep shared GET/HEAD logic in the `app.get(...)` handler and branch on `c.req.raw.method` only when the behavior must differ.

supabase/functions/_backend/public/queue_health.ts

@@ -193,12 +193,12 @@ export function evaluateQueueHealth(
 
   if (metrics.queue_table_exists && metrics.never_read_stale_count > 0) {
     reasons.push('never_read_stale')
-    reason_details.never_read_stale = `${metrics.never_read_stale_count} message(s) still have read_ct=0 after ${metrics.never_read_stale_seconds}s (oldest age ${metrics.oldest_message_age_seconds ?? 'unknown'}s). Consumers are not reading this queue in time.`
+    reason_details.never_read_stale = `${metrics.never_read_stale_count} visible message(s) still have read_ct=0 after ${metrics.never_read_stale_seconds}s (oldest age ${metrics.oldest_message_age_seconds ?? 'unknown'}s). Consumers are not reading this queue in time.`
   }
 
   if (metrics.queue_table_exists && metrics.stuck_count > 0) {
     reasons.push('stuck_high_read_ct')
-    reason_details.stuck_high_read_ct = `${metrics.stuck_count} message(s) have read_ct > ${thresholds.stuck_read_ct} (max ${metrics.max_read_ct ?? 'unknown'}). Messages are being retried without successful ack/delete.`
+    reason_details.stuck_high_read_ct = `${metrics.stuck_count} message(s) have read_ct > ${thresholds.stuck_read_ct} (max ${metrics.max_read_ct ?? 'unknown'}). Hard retry budget is 5 for every queue; poison messages must be archived/deleted, not retried forever.`
   }
 
   if (metrics.queue_table_exists && metrics.queue_count > thresholds.queue_depth_threshold) {
@@ -241,14 +241,14 @@ export function evaluateQueueHealth(
 export function buildQueueHealthCriteria(thresholds: QueueHealthThresholds) {
   return {
     never_read_stale: {
-      healthy_when: 'No queue messages remain with read_ct=0 longer than the queue stale threshold (derived from cron interval when known).',
-      unhealthy_when: 'Messages sit unread long enough that consumers are likely stuck or not scheduled.',
+      healthy_when: 'No visible (vt <= now()) queue messages remain with read_ct=0 longer than the queue stale threshold (derived from cron interval when known). Delayed retries with future vt are ignored.',
+      unhealthy_when: 'Visible messages sit unread long enough that consumers are likely stuck or not scheduled.',
       default_threshold_seconds: thresholds.default_never_read_stale_seconds,
       min_threshold_seconds: thresholds.min_never_read_stale_seconds,
       interval_multiplier: thresholds.never_read_interval_multiplier,
     },
     stuck_high_read_ct: {
-      healthy_when: `No queue messages have read_ct > ${thresholds.stuck_read_ct}.`,
+      healthy_when: `No queue messages have read_ct > ${thresholds.stuck_read_ct} (hard max retries = 5 for every queue).`,
       unhealthy_when: 'Messages keep retrying past the stuck threshold without successful processing.',
       threshold: thresholds.stuck_read_ct,
     },
@@ -373,14 +373,19 @@ async function fetchQueueMetrics(
       `
       SELECT
         COUNT(*)::bigint AS queue_count,
+        -- All unread rows (including delayed vt > now()). Public metric — keep meaning stable.
         COUNT(*) FILTER (WHERE read_ct = 0)::bigint AS never_read_count,
+        -- Staleness only cares about messages consumers can already read.
         COUNT(*) FILTER (
           WHERE read_ct = 0
+            AND vt <= now()
             AND enqueued_at < now() - ($1::text || ' seconds')::interval
         )::bigint AS never_read_stale_count,
         COUNT(*) FILTER (WHERE read_ct > $2)::bigint AS stuck_count,
         MAX(read_ct)::bigint AS max_read_ct,
-        EXTRACT(EPOCH FROM (now() - MIN(enqueued_at)))::numeric AS oldest_message_age_seconds
+        EXTRACT(EPOCH FROM (
+          now() - MIN(enqueued_at) FILTER (WHERE vt <= now())
+        ))::numeric AS oldest_message_age_seconds
       FROM pgmq.q_${queueName}
       `,
       [String(neverReadStaleSeconds), thresholds.stuck_read_ct],

supabase/functions/_backend/triggers/on_version_update.ts

@@ -278,7 +278,10 @@ type ManifestCleanupEntry = {
 /**
  * Trash unreferenced R2 objects first (exist → move to deleted-after-7-days/,
  * missing → ok), then delete that DB row. Never drop DB tracking before R2 is handled.
- * Incomplete work throws so the queue retries; already-trashed paths are idempotent.
+ * Per-file work is committed, so a timeout mid-pass is safe to retry. Leftover rows
+ * after the normal retry budget (MAX_QUEUE_READS=5) are reclaimed by
+ * sweep_deleted_version_manifests. Incomplete work throws so the queue retries;
+ * already-trashed paths are idempotent.
  */
 async function deleteManifest(c: Context, record: Database['public']['Tables']['app_versions']['Row']) {
   const readPgClient = getPgClient(c, true)

supabase/functions/_backend/triggers/queue_consumer.ts

@@ -25,8 +25,10 @@ const MANIFEST_QUEUE_VISIBILITY_TIMEOUT_SECONDS = 900
 const QUEUE_HTTP_TIMEOUT_MS = 15_000
 const VERSION_QUEUE_HTTP_TIMEOUT_MS = 300_000 // large deleted manifests: trash then DB delete
 const HEALTHCHECK_HTTP_TIMEOUT_MS = 8_000
+// HARD RULE: no pgmq queue may retry more than 5 times. Do not raise this, and
+// do not add per-queue exceptions. Leftover work must be re-enqueued by a
+// sweeper/cron (e.g. sweep_deleted_version_manifests), never by burning reads.
 export const MAX_QUEUE_READS = 5
-const VERSION_QUEUE_MAX_READS = 30 // deleted manifests can need many partial trash/delete passes
 const DISCORD_IGNORED_ERROR_CODES = new Set(['version_not_found', 'no_channel'])
 
 export const messageSchema = z.object({
@@ -281,9 +283,7 @@ function getQueueHttpTimeoutMs(functionName: string): number {
   return QUEUE_HTTP_TIMEOUT_MS
 }
 
-function getQueueMaxReads(queueName: string): number {
-  if (isVersionQueueFunction(queueName))
-    return VERSION_QUEUE_MAX_READS
+function getQueueMaxReads(_queueName: string): number {
   return MAX_QUEUE_READS
 }
 

supabase/functions/_backend/triggers/webhook_delivery.ts

@@ -20,6 +20,7 @@ import {
   queueWebhookDeliveryWithDelay,
   scheduleRetry,
   updateDeliveryResult,
+  WEBHOOK_MAX_ATTEMPTS,
 } from '../utils/webhook.ts'
 
 export const app = new Hono<MiddlewareKeyVariables>()
@@ -40,7 +41,7 @@ interface DeliveryMessage {
  * 1. Receive delivery data from queue
  * 2. Deliver the webhook to the user's endpoint
  * 3. On success: mark as success
- * 4. On failure: retry with exponential backoff (up to 3 attempts)
+ * 4. On failure: retry with backoff (hard max WEBHOOK_MAX_ATTEMPTS = 5)
  * 5. After max retries: mark as failed and send notification via Bento
  */
 app.post('/', middlewareAPISecret, async (c) => {
@@ -159,7 +160,8 @@ app.post('/', middlewareAPISecret, async (c) => {
     }
 
     // Handle failure
-    const maxAttempts = delivery.max_attempts || 10
+    // HARD RULE: never more than WEBHOOK_MAX_ATTEMPTS (matches MAX_QUEUE_READS).
+    const maxAttempts = Math.min(delivery.max_attempts || WEBHOOK_MAX_ATTEMPTS, WEBHOOK_MAX_ATTEMPTS)
 
     if (attemptCount < maxAttempts) {
       const retryDelaySeconds = await scheduleRetry(

supabase/functions/_backend/utils/webhook.ts

@@ -96,18 +96,17 @@ export type WebhookEventType = typeof WEBHOOK_EVENT_TYPES[number]
 
 const WEBHOOK_DELIVERY_TIMEOUT_MS = 20000
 const WEBHOOK_RESPONSE_BODY_LIMIT_BYTES = 10000
-const WEBHOOK_MAX_RETRY_AFTER_SECONDS = 24 * 60 * 60
+const WEBHOOK_MAX_RETRY_AFTER_SECONDS = 2 * 60 * 60
+// HARD RULE: same ceiling as MAX_QUEUE_READS — never more than 5 delivery attempts.
+export const WEBHOOK_MAX_ATTEMPTS = 5
 const WEBHOOK_RETRY_THROTTLE_STATUSES = new Set([429, 502, 504])
+// Delays for retries after attempts 1..4 (5th attempt is terminal). Every value
+// must stay <= WEBHOOK_MAX_RETRY_AFTER_SECONDS.
 const WEBHOOK_RETRY_DELAYS_SECONDS = [
   5,
   5 * 60,
   30 * 60,
   2 * 60 * 60,
-  5 * 60 * 60,
-  10 * 60 * 60,
-  14 * 60 * 60,
-  20 * 60 * 60,
-  24 * 60 * 60,
 ]
 
 interface WebhookLogUrlMetadata {

supabase/migrations/20260726134739_queue_health_retry_budget_5.sql

@@ -0,0 +1,88 @@
+-- Align all queue retry budgets to 5, raise org-stats drain rate, and purge
+-- already-stuck messages (read_ct > 5) so /queue_health can return healthy.
+
+-- 1) org_stats_queue was batch 10 / every 5 minutes (~100 msgs/5m) and could not
+-- keep up. Run every process_all_cron_tasks tick with a larger batch.
+UPDATE public.cron_tasks
+SET
+  batch_size = 100,
+  second_interval = 10,
+  minute_interval = NULL,
+  hour_interval = NULL,
+  run_at_hour = NULL,
+  updated_at = now()
+WHERE name = 'org_stats_queue';
+
+-- 2) Webhook delivery application retries must also obey the global max of 5.
+ALTER TABLE public.webhook_deliveries
+  ALTER COLUMN max_attempts SET DEFAULT 5;
+
+UPDATE public.webhook_deliveries
+SET max_attempts = 5
+WHERE max_attempts > 5;
+
+-- 3) Immediate purge of poison messages that already exceeded the hard retry
+-- budget. Bounded globally (batches + wall-clock) like cleanup_queue_messages.
+DO $$
+DECLARE
+  queue_name text;
+  deleted_batch integer;
+  deleted_total bigint := 0;
+  batch_size integer := 5000;
+  max_batches_total integer := 40;
+  batches_used integer := 0;
+  max_runtime_ms integer := 30000;
+  started_at timestamptz := pg_catalog.clock_timestamp();
+BEGIN
+  IF pg_catalog.to_regclass('pgmq.meta') IS NULL THEN
+    RAISE NOTICE 'queue_health_retry_budget_5: pgmq.meta missing, skip stuck purge';
+    RETURN;
+  END IF;
+
+  PERFORM pg_catalog.set_config('statement_timeout', '0', true);
+
+  FOR queue_name IN
+    SELECT q.queue_name
+    FROM pgmq.list_queues() q
+    ORDER BY q.queue_name
+  LOOP
+    EXIT WHEN batches_used >= max_batches_total;
+    EXIT WHEN (EXTRACT(EPOCH FROM (pg_catalog.clock_timestamp() - started_at)) * 1000) >= max_runtime_ms;
+
+    IF pg_catalog.to_regclass(pg_catalog.format('pgmq.q_%I', queue_name)) IS NULL THEN
+      CONTINUE;
+    END IF;
+
+    LOOP
+      EXIT WHEN batches_used >= max_batches_total;
+      EXIT WHEN (EXTRACT(EPOCH FROM (pg_catalog.clock_timestamp() - started_at)) * 1000) >= max_runtime_ms;
+
+      EXECUTE pg_catalog.format(
+        'DELETE FROM pgmq.q_%I
+         WHERE ctid IN (
+           SELECT ctid
+           FROM pgmq.q_%I
+           WHERE read_ct > 5
+           LIMIT $1
+         )',
+        queue_name,
+        queue_name
+      )
+      USING batch_size;
+
+      GET DIAGNOSTICS deleted_batch = ROW_COUNT;
+      EXIT WHEN deleted_batch = 0;
+
+      batches_used := batches_used + 1;
+      deleted_total := deleted_total + deleted_batch;
+    END LOOP;
+  END LOOP;
+
+  RAISE NOTICE
+    'queue_health_retry_budget_5: deleted_stuck_read_ct=% batches_used=%/% runtime_ms=%',
+    deleted_total,
+    batches_used,
+    max_batches_total,
+    (EXTRACT(EPOCH FROM (pg_catalog.clock_timestamp() - started_at)) * 1000)::bigint;
+END;
+$$;

tests/queue-consumer-message-shape.unit.test.ts

@@ -159,8 +159,8 @@ describe('queue_consumer legacy message compatibility', () => {
     expect(__queueConsumerTestUtils__.getQueueVisibilityTimeout('cron_email')).toBe(120)
     expect(__queueConsumerTestUtils__.getQueueVisibilityTimeout('on_version_update')).toBe(900)
     expect(__queueConsumerTestUtils__.getQueueHttpTimeoutMs('on_version_update')).toBe(300_000)
-    expect(__queueConsumerTestUtils__.getQueueMaxReads('on_version_update')).toBe(30)
-    expect(__queueConsumerTestUtils__.getQueueMaxReads('on_manifest_create')).toBe(5)
+    expect(__queueConsumerTestUtils__.getQueueMaxReads('on_version_update')).toBe(MAX_QUEUE_READS)
+    expect(__queueConsumerTestUtils__.getQueueMaxReads('on_manifest_create')).toBe(MAX_QUEUE_READS)
     expect(__queueConsumerTestUtils__.getQueueHttpTimeoutMs('cron_email')).toBe(15_000)
     expect(__queueConsumerTestUtils__.shouldRunQueueSyncInBackground('on_manifest_create')).toBe(false)
     expect(__queueConsumerTestUtils__.shouldRunQueueSyncInBackground('cron_email')).toBe(true)
@@ -237,29 +237,29 @@ describe('queue_consumer legacy message compatibility', () => {
     )).toBe('continue')
   })
 
-  it.concurrent('uses the version queue retry budget for Discord failure alerts', () => {
-    const versionRetryBudget = __queueConsumerTestUtils__.getQueueMaxReads('on_version_update')
+  it.concurrent('uses the shared queue retry budget for Discord failure alerts', () => {
+    const retryBudget = __queueConsumerTestUtils__.getQueueMaxReads('on_version_update')
     const midRetry = {
       cf_id: 'cf-version-mid',
       error_code: 'manifest_cleanup_incomplete',
       function_name: 'on_version_update',
       function_type: 'supabase',
       msg_id: 2,
       payload_size: 10,
-      read_count: MAX_QUEUE_READS,
+      read_count: MAX_QUEUE_READS - 1,
       status: 500,
       status_text: 'Internal Server Error',
     }
     const exhausted = {
       ...midRetry,
       cf_id: 'cf-version-done',
       msg_id: 3,
-      read_count: versionRetryBudget,
+      read_count: retryBudget,
     }
 
-    expect(versionRetryBudget).toBe(30)
-    expect(__queueConsumerTestUtils__.getActionableQueueFailures([midRetry], versionRetryBudget)).toEqual([])
-    expect(__queueConsumerTestUtils__.getActionableQueueFailures([exhausted], versionRetryBudget)).toEqual([exhausted])
+    expect(retryBudget).toBe(MAX_QUEUE_READS)
+    expect(__queueConsumerTestUtils__.getActionableQueueFailures([midRetry], retryBudget)).toEqual([])
+    expect(__queueConsumerTestUtils__.getActionableQueueFailures([exhausted], retryBudget)).toEqual([exhausted])
   })
 
   it.concurrent('alerts Discord after retry budget is exhausted', () => {

tests/queue-health.unit.test.ts

@@ -170,8 +170,10 @@ describe('evaluateQueueHealth', () => {
 
   it.concurrent('documents healthy and unhealthy criteria', () => {
     const criteria = buildQueueHealthCriteria(thresholds)
+    expect(criteria.never_read_stale.healthy_when).toContain('vt <= now()')
     expect(criteria.never_read_stale.healthy_when).toContain('read_ct=0')
     expect(criteria.archive_stale.unhealthy_when).toContain('ramping')
     expect(criteria.stuck_high_read_ct.threshold).toBe(STUCK_READ_CT_THRESHOLD)
+    expect(criteria.stuck_high_read_ct.healthy_when).toContain('hard max retries = 5')
   })
 })

tests/webhook-delivery-security.unit.test.ts

@@ -367,7 +367,8 @@ describe('webhook retry scheduling', () => {
     expect(getWebhookRetryDelaySeconds(1, null, 500, 0.5)).toBe(5)
     expect(getWebhookRetryDelaySeconds(2, null, 500, 0.5)).toBe(5 * 60)
     expect(getWebhookRetryDelaySeconds(3, null, 500, 0.5)).toBe(30 * 60)
-    expect(getWebhookRetryDelaySeconds(9, null, 500, 0.5)).toBe(24 * 60 * 60)
+    expect(getWebhookRetryDelaySeconds(4, null, 500, 0.5)).toBe(2 * 60 * 60)
+    expect(getWebhookRetryDelaySeconds(9, null, 500, 0.5)).toBe(2 * 60 * 60)
   })
 
   it('honors retry-after and throttles rate-limit responses', async () => {