feat: distributed job scheduler with lease-based worker claiming (Closes #123)

.gitignore

@@ -11,6 +11,7 @@ AGENTS.md
 agents/
 issue.md
 node_modules/
+coverage/
 
 # Windows build artifacts
 *.exe

README.md

@@ -50,6 +50,7 @@ An enterprise-grade, high-performance off-chain delivery daemon for real-time So
 - **Performance**: `< 100ms` P99 ingestion latency target via an asynchronous event-driven memory queue.
 - **Robust Security**: Includes HMAC-SHA256 and Ed25519 signature headers, strict replay protection windowing, and thorough SSRF IP/DNS blacklisting.
 - **Resiliency**: Built-in exponential backoff retry schedules with full randomized jitter to survive downstream subscriber downtimes and network drops.
+- **Distributed Scheduling**: Lease-based worker claiming (`WEBHOOK_WORKER_COUNT`) prevents duplicate deliveries across concurrent workers and replicas, with heartbeat renewal and crash-recovery reclaim.
 - **Operational Guides**: See [WEBHOOK_ARCHITECTURE.md](docs/WEBHOOK_ARCHITECTURE.md), [WEBHOOK_DEPLOYMENT.md](docs/WEBHOOK_DEPLOYMENT.md), and [WEBHOOK_RUNBOOK.md](docs/WEBHOOK_RUNBOOK.md).
 
 ## Architecture

docs/WEBHOOK_ARCHITECTURE.md

@@ -92,6 +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
+
+Queue processing runs on a **distributed job scheduler** (`jobScheduler.ts`) where multiple worker loops compete to claim due webhook jobs under short-lived **leases**:
+
+- **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.
+
 ---
 
 ## 4. Monitoring & Metrics
@@ -101,3 +112,9 @@ 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).

docs/WEBHOOK_DEPLOYMENT.md

@@ -57,6 +57,12 @@ Once the Green environment passes all sanity tests:
 2. Monitor active connections on Blue and allow a **5-minute graceful drain window** to complete any outstanding retry attempts or delivery backlogs.
 3. Shut down or idle the Blue infrastructure.
 
+### Delivery Concurrency & Scheduler Workers
+Queue processing is performed by a configurable pool of scheduler workers that claim jobs under short-lived leases, so multiple replicas can run concurrently without double-delivering webhooks:
+- Set `WEBHOOK_WORKER_COUNT` (default `3`) per deployment to control in-process delivery concurrency.
+- To scale out, increase the replica count of the webhook containers; each replica contributes its worker pool and lease-based claiming prevents duplicate deliveries across replicas.
+- After scaling, verify `webhook_scheduler_workers_current` and `webhook_scheduler_active_leases_current` in `/metrics` and confirm `webhook_scheduler_lease_reclaimed_total` stays near zero (reclaimed leases indicate workers expiring mid-delivery and warrant investigation).
+
 ---
 
 ## 3. Canary Analysis Strategy

docs/WEBHOOK_RUNBOOK.md

@@ -79,3 +79,23 @@ To prevent breaking integrations during rotation:
 To ensure the safety of the off-chain system, the SSRF (Server-Side Request Forgery) engine must be audited after any networking or DNS upgrades:
 1. Verify that the URL parser correctly flags subnets by running integration tests.
 2. Inspect server firewalls, ensuring egress traffic is strictly barred from routing to cloud provider private IP ranges and internal Kubernetes API service accounts.
+
+---
+
+## 5. Scheduler & Worker 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`.
+
+### 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.
+
+### 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.
+
+### Tuning
+- Delivery concurrency per instance: `WEBHOOK_WORKER_COUNT` (default `3`).
+- Lease duration and heartbeat are configurable in `jobScheduler.ts` (`leaseDurationMs`, `leaseRenewIntervalMs`, `pollIntervalMs`).

webhook-delivery-service/src/delivery.ts

@@ -1,6 +1,7 @@
 import axios from 'axios';
 import { generateSignatures, validateUrlForSsrf } from './security';
 import { trackDeliveryAttempt, trackQueueSize, trackFailure } from './metrics';
+import { JobScheduler, ExecuteContext } from './jobScheduler';
 import { logger, LogAttributes } from './logger';
 
 // Structured logging is skipped in tests: Jest's console interception adds
@@ -40,12 +41,16 @@ export interface WebhookDeliveryLog {
   lastAttemptTime: number;
 }
 
-// In-memory job queue and log storage
-const queue: WebhookJob[] = [];
+// Delivery log storage (bounded)
 const deliveryLogs: WebhookDeliveryLog[] = [];
 const MAX_LOGS = 100;
 
-let isProcessing = false;
+// Distributed job scheduler: multiple workers claim due webhook jobs under
+// short-lived leases so concurrent replicas/workers never double-deliver the
+// same webhook. Worker count is configurable for horizontal scaling.
+const WORKER_COUNT = parseInt(process.env.WEBHOOK_WORKER_COUNT || '3', 10);
+const scheduler = new JobScheduler();
+scheduler.start(WORKER_COUNT);
 
 /**
  * Enqueue a new webhook delivery job
@@ -70,8 +75,16 @@ export function enqueueWebhook(
     nextAttemptTime: Date.now(),
   };
 
-  queue.push(job);
-  trackQueueSize(queue.length);
+  // Submit the job to the distributed scheduler; a worker will claim it under
+  // a lease as soon as it is due.
+  scheduler.submit({
+    id,
+    runAt: job.nextAttemptTime,
+    execute: async (ctx: ExecuteContext) => {
+      await deliverWebhook(job, ctx);
+    },
+  });
+  trackQueueSize(scheduler.getPendingCount());
 
   // Initialize delivery log
   addLog({
@@ -84,11 +97,6 @@ export function enqueueWebhook(
     lastAttemptTime: Date.now(),
   });
 
-  // Process queue asynchronously
-  setImmediate(() => {
-    processQueue();
-  });
-
   return id;
 }
 
@@ -100,17 +108,24 @@ export function getDeliveryLogs(): WebhookDeliveryLog[] {
 }
 
 /**
- * Retrieve queue size
+ * Retrieve queue size (jobs waiting or due to be claimed by workers)
  */
 export function getQueueSize(): number {
-  return queue.length;
+  return scheduler.getPendingCount();
+}
+
+/**
+ * Retrieve scheduler status (worker ids, pending jobs, active leases)
+ */
+export function getSchedulerStatus() {
+  return scheduler.getStatus();
 }
 
 /**
  * Clear queue and logs (primarily for testing)
  */
 export function clearQueueAndLogs(): void {
-  queue.length = 0;
+  scheduler.clear();
   deliveryLogs.length = 0;
   trackQueueSize(0);
 }
@@ -143,46 +158,10 @@ export function calculateRetryDelay(attempt: number, baseDelay = 1000, maxDelay
 }
 
 /**
- * Background queue processor
- */
-async function processQueue() {
-  if (isProcessing) return;
-  isProcessing = true;
-
-  try {
-    while (queue.length > 0) {
-      // Find jobs ready for processing (nextAttemptTime <= now)
-      const now = Date.now();
-      const jobIndex = queue.findIndex((job) => job.nextAttemptTime <= now);
-
-      if (jobIndex === -1) {
-        // No jobs are ready right now, wait or break
-        break;
-      }
-
-      // Extract the job
-      const [job] = queue.splice(jobIndex, 1);
-      trackQueueSize(queue.length);
-
-      // Process the job
-      await deliverWebhook(job);
-    }
-  } finally {
-    isProcessing = false;
-
-    // If there are still items in the queue, schedule the next check
-    if (queue.length > 0) {
-      setTimeout(() => {
-        processQueue();
-      }, 200); // Check every 200ms
-    }
-  }
-}
-
-/**
- * Deliver a single webhook job
+ * Deliver a single webhook job. Runs inside a scheduler worker that holds the
+ * job's lease; retries are handled by rescheduling the job at a future time.
  */
-async function deliverWebhook(job: WebhookJob) {
+async function deliverWebhook(job: WebhookJob, ctx: ExecuteContext) {
   job.attempts++;
   const startTime = Date.now();
 
@@ -259,11 +238,12 @@ async function deliverWebhook(job: WebhookJob) {
     trackDeliveryAttempt(statusCode || 0, duration, job.attempts);
 
     if (job.attempts < job.maxAttempts) {
-      // Re-queue for retry
+      // Schedule a retry with exponential backoff; the job is released back to
+      // the scheduler pool and becomes claimable again at nextAttemptTime.
       const delay = calculateRetryDelay(job.attempts);
       job.nextAttemptTime = Date.now() + delay;
-      queue.push(job);
-      trackQueueSize(queue.length);
+      ctx.reschedule(job.nextAttemptTime);
+      trackQueueSize(scheduler.getPendingCount());
 
       logDelivery('warn', 'webhook delivery failed, retrying', {
         'webhook.id': job.id,

webhook-delivery-service/src/index.ts

@@ -1,6 +1,6 @@
 import express, { Request, Response } from 'express';
-import { enqueueWebhook, getDeliveryLogs, getQueueSize } from './delivery';
-import { getPrometheusMetrics, getStatsSummary, trackIngestionDuration } from './metrics';
+import { enqueueWebhook, getDeliveryLogs, getQueueSize, getSchedulerStatus } from './delivery';
+import { getPrometheusMetrics, getStatsSummary } from './metrics';
 import { logger } from './logger';
 
 const app = express();
@@ -104,6 +104,7 @@ app.get('/health', (req: Request, res: Response) => {
     status: 'UP',
     timestamp: Date.now(),
     queueSize: getQueueSize(),
+    scheduler: getSchedulerStatus(),
   });
 });