feat: dead letter queue for failed message processing (Closes #124)

docs/WEBHOOK_ARCHITECTURE.md

@@ -92,16 +92,17 @@ Transient network drops, rate limits (HTTP 429), and short-term receiver outages
 - **Full Jitter**: Prevents "thundering herd" issues by introducing randomized delay ($t_{jitter} = \text{random}(0, t_{backoff})$).
 - **Max Retries**: Defaulted to **5 attempts** before a webhook is classified as failed.
 
-### 3.6 Distributed Job Scheduler with Lease-based Worker Claiming
+### 3.6 Dead Letter Queue (`/deadletter`)
 
-Queue processing runs on a **distributed job scheduler** (`jobScheduler.ts`) where multiple worker loops compete to claim due webhook jobs under short-lived **leases**:
+Rather than silently dropping a message that can never be delivered, the service routes permanently failed jobs into a bounded **Dead Letter Queue (DLQ)** for inspection and operator-driven redelivery:
 
-- **Claim protocol**: A job becomes claimable once its `runAt` time has elapsed. A worker acquires the job by claiming a lease from a shared lease registry (`LeaseStore`); while that lease is valid, no other worker can claim the same job, so concurrent replicas/workers can never double-deliver the same webhook.
-- **Fencing across processes**: `LeaseStore.claim` is atomic (synchronous within the Node event loop and serialisable against a shared store such as Redis/etcd in production), which gives cross-process mutual exclusion. Each lease carries a monotonic fencing token.
-- **Heartbeat / lease renewal**: while a job is executing, the owning worker renews its lease on a configurable interval, so a healthy long-running delivery is never stolen by a competing worker.
-- **Crash recovery**: if a worker dies without renewing, its lease expires and another worker reclaims the job — exactly-once under normal operation, at-least-once on worker failure.
-- **Retry via rescheduling**: a failed attempt with retries remaining calls `ctx.reschedule(nextAttemptTime)` (exponential backoff + jitter), returning the job to the claimable pool at a future time.
-- **Horizontal scaling**: worker count is controlled by `WEBHOOK_WORKER_COUNT` (default `3`). Scaling replicas or raising the worker count increases delivery concurrency without risking duplicate deliveries.
+- **What is dead-lettered**: Deliveries that exhaust their maximum retry budget (`MAX_ATTEMPTS_EXHAUSTED`) and jobs rejected by the SSRF shield (`SSRF_BLOCKED`).
+- **Retention**: A bounded, in-memory store (default **1,000 entries**, FIFO). When full, the oldest dead letter is evicted and counted as discarded for alerting.
+- **Inspection**: `GET /deadletter` lists entries (newest first); `GET /deadletter/:id` fetches a single entry including the failed payload, attempt history, reason, and last error.
+- **Redelivery**: `POST /deadletter/:id/requeue` reconstructs a fresh delivery job from the stored entry (with a fresh retry budget) and pushes it back onto the active queue. Requeued deliveries are re-signed and pass through the full security + retry pipeline again.
+- **Removal**: `DELETE /deadletter/:id` removes a single entry; `DELETE /deadletter?confirm=true` purges the whole queue. Purging requires an explicit confirmation query parameter to prevent accidental data loss.
+
+The DLQ is intentionally in-memory to mirror the service's ingestion pipeline and keep inspection/redelivery on the fast path. For deployments requiring cross-restart durability, the out-of-process persistent queue pattern (Redis/RabbitMQ) noted in the runbook should be substituted.
 
 ---
 
@@ -112,9 +113,7 @@ The service registers Prometheus counters and histograms to measure health indic
 - `webhook_delivery_duration_seconds`: Histogram of endpoint response latency.
 - `webhook_queue_size_current`: Gauge representing current queue occupancy.
 - `webhook_failures_total`: Total dropped or exhausted delivery alerts.
-- `webhook_scheduler_workers_current`: Gauge of active worker loops.
-- `webhook_scheduler_active_leases_current`: Gauge of jobs currently executing under a worker lease.
-- `webhook_scheduler_jobs_submitted_total`: Counter of jobs submitted to the scheduler.
-- `webhook_scheduler_jobs_processed_total`: Counter of jobs executed by workers.
-- `webhook_scheduler_jobs_failed_total`: Counter of jobs whose execution threw.
-- `webhook_scheduler_lease_reclaimed_total`: Counter of expired leases reclaimed by another worker (crash recovery events).
+- `webhook_dead_letter_queue_size_current`: Gauge of the current DLQ occupancy.
+- `webhook_dead_letter_enqueued_total`: Counter of messages entering the DLQ, labeled by `reason` (`MAX_ATTEMPTS_EXHAUSTED` | `SSRF_BLOCKED`).
+- `webhook_dead_letter_requeued_total`: Counter of dead letters pushed back onto the active queue.
+- `webhook_dead_letter_discarded_total`: Counter of dead letters evicted, purged, or manually removed.

docs/WEBHOOK_RUNBOOK.md

@@ -82,20 +82,39 @@ To ensure the safety of the off-chain system, the SSRF (Server-Side Request Forg
 
 ---
 
-## 5. Scheduler & Worker Operations
+## 5. Dead Letter Queue (DLQ) Operations
 
-Queue processing runs on the distributed job scheduler, where workers claim jobs under short-lived leases. Diagnose scheduler health through the `webhook_scheduler_*` metrics on `/metrics` and the scheduler block on `/health`.
+When a webhook permanently fails after exhausting its retry budget, or is rejected by the SSRF shield, it is moved to the **dead letter queue** instead of being dropped. `webhook_dead_letter_queue_size_current` climbing or `webhook_dead_letter_enqueued_total` increasing indicates persistent downstream failures.
 
-### Health Indicators
-- `webhook_scheduler_workers_current` **0** → worker loops are not running; the scheduler cannot drain the queue. Restart the service.
-- `webhook_scheduler_active_leases_current` sustained at the worker count → all workers are blocked on slow deliveries; inspect downstream endpoint latency and consider scaling out.
-- `webhook_scheduler_lease_reclaimed_total` climbing → workers are expiring mid-delivery (lease not renewed). Investigate event-loop blocking / GC pauses, or increase the lease duration.
+### Step 1: Inspect the dead letter queue
+```bash
+curl -s http://webhook-service.internal/deadletter
+# {"count": 3, "deadLetters": [ { "id": "...", "reason": "MAX_ATTEMPTS_EXHAUSTED", ... } ]}
+
+# Inspect a single entry to see the failure reason and last error
+curl -s http://webhook-service.internal/deadletter/<job-id>
+```
 
-### Diagnosing duplicate or missed deliveries
-1. Confirm workers are healthy: `curl -s http://webhook-service.internal/health` and check `scheduler.workers` is non-empty and `scheduler.pendingCount` is not climbing.
-2. Confirm no lease thrash: `curl -s http://webhook-service.internal/metrics | grep webhook_scheduler_lease_reclaimed_total`.
-3. If `pendingCount` climbs while `active_leases` stays low, a worker crash loop is likely; scale the deployment and inspect container restart counts.
+### Step 2: Confirm the root cause before redelivering
+1. Verify the downstream endpoint is healthy (`curl` / `GET /health` on the receiver).
+2. Confirm the stored `errorMessage` is a transient failure (5xx, timeout) and not a request you should not re-send (e.g. 4xx contract violations).
+
+### Step 3: Redeliver the message
+```bash
+# Push a single dead letter back onto the active queue with a fresh retry budget
+curl -X POST http://webhook-service.internal/deadletter/<job-id>/requeue
+# { "status": "REQUEUED", "jobId": "<new-job-id>" }
+```
+The requeued message is re-signed and passes through the full security + retry pipeline again.
+
+### Step 4: Discard dead letters
+```bash
+# Remove a single entry
+curl -X DELETE http://webhook-service.internal/deadletter/<job-id>
+# Purge the entire queue (requires explicit confirmation)
+curl -X DELETE http://webhook-service.internal/deadletter?confirm=true
+```
 
-### Tuning
-- Delivery concurrency per instance: `WEBHOOK_WORKER_COUNT` (default `3`).
-- Lease duration and heartbeat are configurable in `jobScheduler.ts` (`leaseDurationMs`, `leaseRenewIntervalMs`, `pollIntervalMs`).
+### Operational Notes
+- **Bounded retention**: The DLQ holds up to 1,000 entries in memory; the oldest entry is evicted (and counted as `webhook_dead_letter_discarded_total`) when capacity is exceeded.
+- **Durability**: The DLQ is in-memory. For transactions that must survive service restarts, deploy the out-of-process persistent queue pattern (Redis/RabbitMQ) as the backing store.

webhook-delivery-service/src/deadLetterQueue.ts

@@ -0,0 +1,160 @@
+/**
+ * Dead Letter Queue (DLQ) for the Webhook Delivery Service.
+ *
+ * Webhook deliveries that permanently fail (max attempts exhausted) or that are
+ * rejected by the SSRF security shield are moved into the dead letter queue
+ * instead of being silently dropped. This preserves the failed messages for
+ * inspection, offline retry ("requeue"), and operational forensics, so that a
+ * downstream subscriber outage or a mis-configured endpoint is never the cause
+ * of silent data loss.
+ *
+ * The queue is backed by an in-memory bounded store (mirroring the rest of the
+ * service's in-memory delivery pipeline) and is drained through the `/deadletter`
+ * HTTP endpoints. A pluggable requeue handler is registered by the delivery
+ * module so that dead letters can be pushed back onto the active delivery queue
+ * with a fresh retry budget.
+ */
+import type { WebhookPayload } from './delivery';
+import {
+  trackDeadLetterCount,
+  trackDeadLetterEnqueued,
+  trackDeadLetterDiscarded,
+} from './metrics';
+
+export type DeadLetterReason = 'MAX_ATTEMPTS_EXHAUSTED' | 'SSRF_BLOCKED';
+
+export interface DeadLetterEntry {
+  /** Identifier of the original webhook job. */
+  id: string;
+  /** Destination endpoint that the webhook was targeting. */
+  url: string;
+  /** Event payload that could not be delivered. */
+  payload: WebhookPayload;
+  /** Shared secret used for HMAC signing (required for requeue). */
+  secret: string;
+  /** Optional Ed25519 private key used for signing (required for requeue). */
+  privateKey?: string;
+  /** Number of delivery attempts made before dead-lettering. */
+  attempts: number;
+  /** Maximum number of attempts permitted for redelivery. */
+  maxAttempts: number;
+  /** Why the message was dead-lettered. */
+  reason: DeadLetterReason;
+  /** Human readable error description captured at failure time. */
+  errorMessage: string;
+  /** HTTP status code observed on the last failed attempt, if any. */
+  statusCode?: number;
+  /** Timestamp (ms) at which the message entered the dead letter queue. */
+  deadLetteredAt: number;
+}
+
+const DEFAULT_MAX_DEAD_LETTERS = 1000;
+
+const store = new Map<string, DeadLetterEntry>();
+let maxDeadLetters = DEFAULT_MAX_DEAD_LETTERS;
+
+/*
+ * The requeue orchestration (dead letter -> active delivery queue) lives in the
+ * delivery module, which owns both the queue and this store. This module only
+ * exposes the primitive data operations; requeueing pops an entry and the
+ * delivery module reconstructs a fresh job from it.
+ */
+
+/**
+ * Insert a permanently failed delivery into the dead letter queue. If the
+ * queue is at capacity the oldest entry is evicted (FIFO) so the queue stays
+ * bounded; evicted entries are tracked as discarded for alerting purposes.
+ */
+export function reportDeadLetter(entry: DeadLetterEntry): void {
+  if (!store.has(entry.id) && store.size >= maxDeadLetters) {
+    const oldest = oldestEntryId();
+    if (oldest) {
+      store.delete(oldest);
+      trackDeadLetterDiscarded();
+    }
+  }
+  store.set(entry.id, entry);
+  trackDeadLetterEnqueued(entry.reason);
+  trackDeadLetterCount(store.size);
+}
+
+/** Return a snapshot of all dead letters, newest first. */
+export function getDeadLetters(): DeadLetterEntry[] {
+  return [...store.values()].sort((a, b) => b.deadLetteredAt - a.deadLetteredAt);
+}
+
+/** Return a single dead letter by id. */
+export function getDeadLetter(id: string): DeadLetterEntry | undefined {
+  return store.get(id);
+}
+
+/** Return the current number of dead letters. */
+export function getDeadLetterCount(): number {
+  return store.size;
+}
+
+/**
+ * Atomically remove a dead letter from the queue and return it so the caller
+ * can re-enqueue it as a fresh delivery job. Returns `undefined` if the entry
+ * does not exist.
+ */
+export function popDeadLetter(id: string): DeadLetterEntry | undefined {
+  const entry = store.get(id);
+  if (!entry) {
+    return undefined;
+  }
+  store.delete(id);
+  trackDeadLetterCount(store.size);
+  return entry;
+}
+
+/**
+ * Remove a single dead letter from the queue without requeueing it.
+ * Returns `true` if an entry was present and removed.
+ */
+export function removeDeadLetter(id: string): boolean {
+  const existed = store.delete(id);
+  if (existed) {
+    trackDeadLetterCount(store.size);
+    trackDeadLetterDiscarded();
+  }
+  return existed;
+}
+
+/** Remove all dead letters from the queue. Returns the number removed. */
+export function purgeDeadLetters(): number {
+  const count = store.size;
+  if (count > 0) {
+    store.clear();
+    trackDeadLetterCount(0);
+    for (let i = 0; i < count; i++) {
+      trackDeadLetterDiscarded();
+    }
+  }
+  return count;
+}
+
+/** Clear the DLQ and reset internal state (primarily for tests). */
+export function resetDeadLetterQueue(): void {
+  store.clear();
+  maxDeadLetters = DEFAULT_MAX_DEAD_LETTERS;
+  trackDeadLetterCount(0);
+}
+
+/** Configure the maximum size of the dead letter queue (primarily for tests). */
+export function setMaxDeadLetters(size: number): void {
+  maxDeadLetters = size;
+}
+
+/** Return the id of the oldest entry (the FIFO eviction candidate). */
+function oldestEntryId(): string | undefined {
+  let oldestId: string | undefined;
+  let oldestTime = Infinity;
+  for (const entry of store.values()) {
+    if (entry.deadLetteredAt < oldestTime) {
+      oldestTime = entry.deadLetteredAt;
+      oldestId = entry.id;
+    }
+  }
+  return oldestId;
+}

webhook-delivery-service/src/delivery.ts

@@ -1,7 +1,7 @@
 import axios from 'axios';
 import { generateSignatures, validateUrlForSsrf } from './security';
-import { trackDeliveryAttempt, trackQueueSize, trackFailure } from './metrics';
-import { JobScheduler, ExecuteContext } from './jobScheduler';
+import { trackDeliveryAttempt, trackQueueSize, trackFailure, trackDeadLetterRequeued } from './metrics';
+import { reportDeadLetter, popDeadLetter, resetDeadLetterQueue, DeadLetterEntry } from './deadLetterQueue';
 import { logger, LogAttributes } from './logger';
 
 // Structured logging is skipped in tests: Jest's console interception adds
@@ -121,12 +121,28 @@ export function getSchedulerStatus() {
   return scheduler.getStatus();
 }
 
+/**
+ * Push a dead letter back onto the active delivery queue as a fresh job with a
+ * fresh retry budget. Returns the new webhook job id, or `null` if the dead
+ * letter does not exist.
+ */
+export function requeueDeadLetter(id: string): string | null {
+  const entry = popDeadLetter(id);
+  if (!entry) {
+    return null;
+  }
+  const newJobId = enqueueWebhook(entry.payload, entry.url, entry.secret, entry.privateKey, entry.maxAttempts);
+  trackDeadLetterRequeued();
+  return newJobId;
+}
+
 /**
  * Clear queue and logs (primarily for testing)
  */
 export function clearQueueAndLogs(): void {
   scheduler.clear();
   deliveryLogs.length = 0;
+  resetDeadLetterQueue();
   trackQueueSize(0);
 }
 
@@ -170,6 +186,18 @@ async function deliverWebhook(job: WebhookJob, ctx: ExecuteContext) {
   if (!ssrfCheck.valid) {
     const errorMsg = `SSRF Prevention: ${ssrfCheck.reason}`;
     trackFailure();
+    reportDeadLetter({
+      id: job.id,
+      url: job.url,
+      payload: job.payload,
+      secret: job.secret,
+      privateKey: job.privateKey,
+      attempts: job.attempts,
+      maxAttempts: job.maxAttempts,
+      reason: 'SSRF_BLOCKED',
+      errorMessage: errorMsg,
+      deadLetteredAt: Date.now(),
+    });
     logDelivery('warn', 'webhook delivery dropped by SSRF check', {
       'webhook.id': job.id,
       'webhook.event': job.payload.event,
@@ -266,8 +294,21 @@ async function deliverWebhook(job: WebhookJob, ctx: ExecuteContext) {
         lastAttemptTime: Date.now(),
       });
     } else {
-      // Max attempts exhausted
+      // Max attempts exhausted -> move to the dead letter queue
       trackFailure();
+      reportDeadLetter({
+        id: job.id,
+        url: job.url,
+        payload: job.payload,
+        secret: job.secret,
+        privateKey: job.privateKey,
+        attempts: job.attempts,
+        maxAttempts: job.maxAttempts,
+        reason: 'MAX_ATTEMPTS_EXHAUSTED',
+        errorMessage: `Max attempts (${job.maxAttempts}) exhausted. Last error: ${errorMessage}`,
+        statusCode,
+        deadLetteredAt: Date.now(),
+      });
       logDelivery('error', 'webhook delivery failed permanently', {
         'webhook.id': job.id,
         'webhook.event': job.payload.event,