diff --git a/.gitignore b/.gitignore
index 5f9c756..30cd7ee 100644
--- a/.gitignore
+++ b/.gitignore
@@ -11,6 +11,7 @@ AGENTS.md
 agents/
 issue.md
 node_modules/
+coverage/
 
 # Windows build artifacts
 *.exe
diff --git a/README.md b/README.md
index f75dc12..b0ca350 100644
--- a/README.md
+++ b/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
diff --git a/docs/WEBHOOK_ARCHITECTURE.md b/docs/WEBHOOK_ARCHITECTURE.md
index 8652feb..7d92a6d 100644
--- a/docs/WEBHOOK_ARCHITECTURE.md
+++ b/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).
diff --git a/docs/WEBHOOK_DEPLOYMENT.md b/docs/WEBHOOK_DEPLOYMENT.md
index fd19f5d..684c4c4 100644
--- a/docs/WEBHOOK_DEPLOYMENT.md
+++ b/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
diff --git a/docs/WEBHOOK_RUNBOOK.md b/docs/WEBHOOK_RUNBOOK.md
index db4b737..2279cc6 100644
--- a/docs/WEBHOOK_RUNBOOK.md
+++ b/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`).
diff --git a/webhook-delivery-service/src/delivery.ts b/webhook-delivery-service/src/delivery.ts
index 87be392..192686f 100644
--- a/webhook-delivery-service/src/delivery.ts
+++ b/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,
diff --git a/webhook-delivery-service/src/index.ts b/webhook-delivery-service/src/index.ts
index 849c6cb..bd428a9 100644
--- a/webhook-delivery-service/src/index.ts
+++ b/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(),
   });
 });
 
diff --git a/webhook-delivery-service/src/jobScheduler.ts b/webhook-delivery-service/src/jobScheduler.ts
new file mode 100644
index 0000000..bf611f3
--- /dev/null
+++ b/webhook-delivery-service/src/jobScheduler.ts
@@ -0,0 +1,343 @@
+/**
+ * Distributed job scheduler with lease-based worker claiming.
+ *
+ * Multiple stateless workers (each running a `runWorker` loop) compete to claim
+ * due jobs from a shared scheduler. A worker claims a job by acquiring a short
+ * lived *lease* from a shared lease registry (modelled here by `LeaseStore`; in
+ * production the same protocol maps to Redis/etcd). While a lease is valid only
+ * the holding worker may process the job, so concurrent workers can never
+ * double-deliver the same unit of work.
+ *
+ * - **Fencing**: `LeaseStore.claim` is synchronous (therefore atomic within the
+ *   Node event loop, and serialisable against a shared store across processes),
+ *   so at most one worker can hold a given job's lease at a time.
+ * - **Heartbeat / lease renewal**: while a job is executing, the scheduler
+ *   renews the lease on an interval so a long-running, healthy job is not
+ *   stolen by another worker.
+ * - **Crash recovery**: if a worker fails to renew (crashed or unresponsive),
+ *   its lease expires and another worker may reclaim the job, giving
+ *   exactly-once-under-load / at-least-once-on-failure delivery semantics.
+ * - **Retry with reschedule**: a handler may call `ctx.reschedule(runAt)` to
+ *   requeue the job at a future time (e.g. exponential backoff). Until then the
+ *   job is not claimable by any worker.
+ */
+import {
+  trackActiveLeases,
+  trackJobSubmitted,
+  trackJobProcessed,
+  trackJobFailed,
+  trackSchedulerWorkers,
+  trackLeaseReclaimed,
+} from './metrics';
+
+export interface ExecuteContext {
+  /** Id of the claimed job. */
+  jobId: string;
+  /** Worker that currently holds this job's lease. */
+  workerId: string;
+  /** Extend the current lease (heartbeat) for this job. */
+  renewLease: () => void;
+  /** Requeue the job to run again at the given timestamp (ms). */
+  reschedule: (runAt: number) => void;
+}
+
+export interface ScheduledTask {
+  id: string;
+  /** Earliest timestamp (ms) at which the job becomes claimable. */
+  runAt: number;
+  /** Performs the job. Rescheduling is done through `ctx`. */
+  execute: (ctx: ExecuteContext) => Promise<void>;
+  /** Called (and swallowed) if `execute` throws. */
+  onError?: (error: unknown) => void;
+}
+
+export interface Lease {
+  jobId: string;
+  workerId: string;
+  expiresAt: number;
+  fencingToken: number;
+}
+
+export interface ClaimAttempt {
+  granted: boolean;
+  reason: 'TAKEN' | 'GRANTED';
+  reclaimed: boolean;
+  lease?: Lease;
+}
+
+/**
+ * Shared lease registry. Within a single process `claim` is synchronous and
+ * thus atomic; when backed by a shared key/value store (Redis, etc.) the same
+ * read-compare-write (with a fencing token) provides cross-process mutual
+ * exclusion.
+ */
+export class LeaseStore {
+  private readonly leases = new Map<string, Lease>();
+  private tokenCounter = 0;
+
+  private now(): number {
+    return Date.now();
+  }
+
+  /**
+   * Attempt to claim the lease for a job. A lease that is still unexpired is
+   * held and the claim fails; an expired lease may be reclaimed. Returns
+   * `false` if the lease is currently held by (possibly another) worker.
+   */
+  claim(jobId: string, workerId: string, leaseDurationMs: number): ClaimAttempt {
+    const existing = this.leases.get(jobId);
+    if (existing && existing.expiresAt > this.now()) {
+      return { granted: false, reason: 'TAKEN', reclaimed: false };
+    }
+
+    const reclaimed = existing != null;
+    const lease: Lease = {
+      jobId,
+      workerId,
+      expiresAt: this.now() + leaseDurationMs,
+      fencingToken: ++this.tokenCounter,
+    };
+    this.leases.set(jobId, lease);
+    return { granted: true, reason: 'GRANTED', reclaimed, lease };
+  }
+
+  /** Extend the lease if still held by the requesting worker. */
+  renew(jobId: string, workerId: string, leaseDurationMs: number): boolean {
+    const lease = this.leases.get(jobId);
+    if (lease && lease.workerId === workerId) {
+      lease.expiresAt = this.now() + leaseDurationMs;
+      return true;
+    }
+    return false;
+  }
+
+  /** Release a lease; only succeeds for the current holder. */
+  release(jobId: string, workerId: string): boolean {
+    const lease = this.leases.get(jobId);
+    if (lease && lease.workerId === workerId) {
+      this.leases.delete(jobId);
+      return true;
+    }
+    return false;
+  }
+
+  /** Whether a job currently has an unexpired lease. */
+  isActive(jobId: string): boolean {
+    const lease = this.leases.get(jobId);
+    return lease != null && lease.expiresAt > this.now();
+  }
+
+  /** Number of currently active (unexpired) leases. */
+  activeCount(): number {
+    const now = this.now();
+    let count = 0;
+    for (const lease of this.leases.values()) {
+      if (lease.expiresAt > now) {
+        count++;
+      }
+    }
+    return count;
+  }
+}
+
+export interface JobSchedulerOptions {
+  /**
+   * How long a worker's claim on a job lasts before it expires and may be
+   * reclaimed by another worker.
+   * @default 30000
+   */
+  leaseDurationMs?: number;
+  /**
+   * How often the running worker renews (heartbeats) its lease while a job is
+   * executing.
+   * @default 5000
+   */
+  leaseRenewIntervalMs?: number;
+  /**
+   * How often idle workers poll for newly due jobs.
+   * @default 50
+   */
+  pollIntervalMs?: number;
+}
+
+export class JobScheduler {
+  private readonly leaseStore = new LeaseStore();
+  private readonly pending = new Map<string, ScheduledTask>();
+  private readonly leaseDurationMs: number;
+  private readonly leaseRenewIntervalMs: number;
+  private readonly pollIntervalMs: number;
+  private readonly workers = new Set<string>();
+  private workerCounter = 0;
+  private running = false;
+
+  constructor(options: JobSchedulerOptions = {}) {
+    this.leaseDurationMs = options.leaseDurationMs ?? 30000;
+    this.leaseRenewIntervalMs = options.leaseRenewIntervalMs ?? 5000;
+    this.pollIntervalMs = options.pollIntervalMs ?? 50;
+  }
+
+  /**
+   * Register a unit of work. Jobs are only executed once their `runAt` has
+   * elapsed and a worker acquires the lease.
+   */
+  submit(task: ScheduledTask): void {
+    this.pending.set(task.id, task);
+    trackJobSubmitted();
+    trackActiveLeases(this.leaseStore.activeCount());
+  }
+
+  /**
+   * Start worker loops. Idempotent: repeated calls return the existing workers
+   * rather than spawning duplicates. Returns the worker ids.
+   */
+  start(workerCount: number): string[] {
+    if (this.running) {
+      return [...this.workers];
+    }
+    this.running = true;
+    for (let i = 0; i < workerCount; i++) {
+      const workerId = `worker-${this.workerCounter++}`;
+      this.workers.add(workerId);
+      setImmediate(() => {
+        void this.runWorker(workerId);
+      });
+    }
+    trackSchedulerWorkers(this.workers.size);
+    return [...this.workers];
+  }
+
+  /** Stop all worker loops and wait for them to exit. */
+  async stop(): Promise<string[]> {
+    this.running = false;
+    const ids = [...this.workers];
+    this.workers.clear();
+    trackSchedulerWorkers(0);
+    trackActiveLeases(this.leaseStore.activeCount());
+    // Give running iterations a moment to observe the stop flag.
+    await new Promise((resolve) => setTimeout(resolve, this.pollIntervalMs * 2));
+    return ids;
+  }
+
+  /** Drop every queued job that is waiting for execution. */
+  clear(): void {
+    this.pending.clear();
+    trackActiveLeases(this.leaseStore.activeCount());
+  }
+
+  /** Number of jobs currently waiting (or due) to be claimed. */
+  getPendingCount(): number {
+    return this.pending.size;
+  }
+
+  /** Number of jobs currently being executed under an active lease. */
+  getActiveLeaseCount(): number {
+    return this.leaseStore.activeCount();
+  }
+
+  getWorkerIds(): string[] {
+    return [...this.workers];
+  }
+
+  getLeaseStore(): LeaseStore {
+    return this.leaseStore;
+  }
+
+  /** Status summary for dashboards / `/health`. */
+  getStatus(): { workers: string[]; pendingCount: number; activeLeases: number } {
+    return {
+      workers: this.getWorkerIds(),
+      pendingCount: this.getPendingCount(),
+      activeLeases: this.getActiveLeaseCount(),
+    };
+  }
+
+  /** Claim the next due job not already under an active lease. */
+  private pickAndClaim(workerId: string): ScheduledTask | null {
+    const now = Date.now();
+    for (const task of this.pending.values()) {
+      if (task.runAt > now) {
+        continue;
+      }
+      const attempt = this.leaseStore.claim(task.id, workerId, this.leaseDurationMs);
+      if (attempt.granted) {
+        if (attempt.reclaimed) {
+          trackLeaseReclaimed();
+        }
+        trackActiveLeases(this.leaseStore.activeCount());
+        return task;
+      }
+    }
+    return null;
+  }
+
+  /** Polling loop for a single worker. */
+  private async runWorker(workerId: string): Promise<void> {
+    while (this.running) {
+      const claimed = this.pickAndClaim(workerId);
+      if (!claimed) {
+        // No due job available; back off before polling again.
+        await delay(this.pollIntervalMs);
+        continue;
+      }
+      await this.execute(workerId, claimed);
+    }
+  }
+
+  private async execute(workerId: string, task: ScheduledTask): Promise<void> {
+    // The job stays in the pending pool while in flight. Its active lease
+    // prevents any other worker from claiming it; if the lease expires without
+    // renewal (worker crash), another worker may reclaim it (at-least-once
+    // recovery). On completion the job is removed from the pool entirely.
+    let rescheduleAt: number | null = null;
+
+    // Heartbeat the lease while the job is in flight so a healthy execution is
+    // never stolen by a competing worker.
+    const heartbeat = setInterval(() => {
+      this.leaseStore.renew(task.id, workerId, this.leaseDurationMs);
+    }, this.leaseRenewIntervalMs);
+
+    const ctx: ExecuteContext = {
+      jobId: task.id,
+      workerId,
+      renewLease: () => this.leaseStore.renew(task.id, workerId, this.leaseDurationMs),
+      reschedule: (at: number) => {
+        rescheduleAt = at;
+      },
+    };
+
+    try {
+      await task.execute(ctx);
+      trackJobProcessed();
+    } catch (error) {
+      trackJobProcessed();
+      trackJobFailed();
+      if (task.onError) {
+        try {
+          task.onError(error);
+        } catch {
+          // Never let error handling mask a delivery failure.
+        }
+      }
+    } finally {
+      clearInterval(heartbeat);
+      this.leaseStore.release(task.id, workerId);
+      trackActiveLeases(this.leaseStore.activeCount());
+    }
+
+    if (rescheduleAt !== null) {
+      task.runAt = rescheduleAt;
+      // Job remains in the pending pool, claimable again at the new runAt.
+    } else {
+      this.pending.delete(task.id);
+      trackActiveLeases(this.leaseStore.activeCount());
+    }
+  }
+}
+
+function delay(ms: number): Promise<void> {
+  return new Promise((resolve) => setTimeout(resolve, ms));
+}
+
+export function createJobScheduler(options?: JobSchedulerOptions): JobScheduler {
+  return new JobScheduler(options);
+}
\ No newline at end of file
diff --git a/webhook-delivery-service/src/metrics.ts b/webhook-delivery-service/src/metrics.ts
index b15d37a..a7339ea 100644
--- a/webhook-delivery-service/src/metrics.ts
+++ b/webhook-delivery-service/src/metrics.ts
@@ -32,6 +32,42 @@ const queueSize = new Gauge({
   registers: [registry],
 });
 
+const schedulerWorkers = new Gauge({
+  name: 'webhook_scheduler_workers_current',
+  help: 'Current number of active scheduler worker loops',
+  registers: [registry],
+});
+
+const schedulerActiveLeases = new Gauge({
+  name: 'webhook_scheduler_active_leases_current',
+  help: 'Current number of jobs being executed under an active worker lease',
+  registers: [registry],
+});
+
+const schedulerJobsSubmitted = new Counter({
+  name: 'webhook_scheduler_jobs_submitted_total',
+  help: 'Total number of jobs submitted to the distributed scheduler',
+  registers: [registry],
+});
+
+const schedulerJobsProcessed = new Counter({
+  name: 'webhook_scheduler_jobs_processed_total',
+  help: 'Total number of jobs executed by scheduler workers',
+  registers: [registry],
+});
+
+const schedulerJobsFailed = new Counter({
+  name: 'webhook_scheduler_jobs_failed_total',
+  help: 'Total number of jobs whose execution threw an error',
+  registers: [registry],
+});
+
+const schedulerLeaseReclaimed = new Counter({
+  name: 'webhook_scheduler_lease_reclaimed_total',
+  help: 'Total number of expired leases reclaimed by another worker (crash recovery)',
+  registers: [registry],
+});
+
 const totalFailures = new Counter({
   name: 'webhook_failures_total',
   help: 'Total number of webhooks that completely failed after all retries or SSRF drops',
@@ -112,6 +148,72 @@ export function trackQueueSize(size: number): void {
   }
 }
 
+/**
+ * Tracks the number of active scheduler worker loops
+ */
+export function trackSchedulerWorkers(count: number): void {
+  try {
+    schedulerWorkers.set(count);
+  } catch {
+    // Ignored
+  }
+}
+
+/**
+ * Tracks the number of active (unexpired) worker leases
+ */
+export function trackActiveLeases(count: number): void {
+  try {
+    schedulerActiveLeases.set(count);
+  } catch {
+    // Ignored
+  }
+}
+
+/**
+ * Tracks a job being submitted to the scheduler
+ */
+export function trackJobSubmitted(): void {
+  try {
+    schedulerJobsSubmitted.inc();
+  } catch {
+    // Ignored
+  }
+}
+
+/**
+ * Tracks a job being executed (successfully or not)
+ */
+export function trackJobProcessed(): void {
+  try {
+    schedulerJobsProcessed.inc();
+  } catch {
+    // Ignored
+  }
+}
+
+/**
+ * Tracks a job whose execution threw
+ */
+export function trackJobFailed(): void {
+  try {
+    schedulerJobsFailed.inc();
+  } catch {
+    // Ignored
+  }
+}
+
+/**
+ * Tracks a lease that expired and was reclaimed by another worker
+ */
+export function trackLeaseReclaimed(): void {
+  try {
+    schedulerLeaseReclaimed.inc();
+  } catch {
+    // Ignored
+  }
+}
+
 /**
  * Tracks absolute failure / dropped webhook
  */
diff --git a/webhook-delivery-service/tests/jobScheduler.test.ts b/webhook-delivery-service/tests/jobScheduler.test.ts
new file mode 100644
index 0000000..7932593
--- /dev/null
+++ b/webhook-delivery-service/tests/jobScheduler.test.ts
@@ -0,0 +1,308 @@
+import { JobScheduler, LeaseStore } from '../src/jobScheduler';
+
+/** Poll until `cond` is true or the timeout elapses. */
+async function waitFor(cond: () => boolean, timeoutMs = 3000, intervalMs = 10): Promise<void> {
+  const start = Date.now();
+  while (!cond()) {
+    if (Date.now() - start > timeoutMs) {
+      throw new Error(`waitFor timed out after ${timeoutMs}ms`);
+    }
+    await new Promise((resolve) => setTimeout(resolve, intervalMs));
+  }
+}
+
+function sleep(ms: number): Promise<void> {
+  return new Promise((resolve) => setTimeout(resolve, ms));
+}
+
+describe('LeaseStore (shared lease registry)', () => {
+  test('claim grants an unexpired lease to the requesting worker', () => {
+    const store = new LeaseStore();
+    const attempt = store.claim('job-1', 'worker-a', 500);
+
+    expect(attempt.granted).toBe(true);
+    expect(attempt.reclaimed).toBe(false);
+    expect(attempt.lease!.workerId).toBe('worker-a');
+    expect(store.isActive('job-1')).toBe(true);
+    expect(store.activeCount()).toBe(1);
+  });
+
+  test('claim is denied while another worker holds an active lease', () => {
+    const store = new LeaseStore();
+    store.claim('job-1', 'worker-a', 500);
+
+    const attempt = store.claim('job-1', 'worker-b', 500);
+
+    expect(attempt.granted).toBe(false);
+    expect(attempt.reason).toBe('TAKEN');
+    expect(store.isActive('job-1')).toBe(true);
+  });
+
+  test('an expired lease can be reclaimed by another worker', async () => {
+    const store = new LeaseStore();
+    store.claim('job-1', 'worker-a', 20);
+    expect(store.isActive('job-1')).toBe(true);
+
+    await sleep(40); // let the lease expire
+
+    const attempt = store.claim('job-1', 'worker-b', 500);
+
+    expect(attempt.granted).toBe(true);
+    expect(attempt.reclaimed).toBe(true);
+    expect(attempt.lease!.workerId).toBe('worker-b');
+  });
+
+  test('renew only extends a lease owned by the requesting worker', () => {
+    const store = new LeaseStore();
+    store.claim('job-1', 'worker-a', 50);
+
+    expect(store.renew('job-1', 'worker-b', 5000)).toBe(false); // not the holder
+    expect(store.renew('job-1', 'worker-a', 5000)).toBe(true); // holder heartbeats
+    expect(store.isActive('job-1')).toBe(true);
+  });
+
+  test('release only succeeds for the current lease holder', () => {
+    const store = new LeaseStore();
+    store.claim('job-1', 'worker-a', 500);
+
+    expect(store.release('job-1', 'worker-b')).toBe(false);
+    expect(store.release('job-1', 'worker-a')).toBe(true);
+    expect(store.isActive('job-1')).toBe(false);
+    expect(store.activeCount()).toBe(0);
+  });
+});
+
+describe('JobScheduler (lease-based worker claiming)', () => {
+  test('a job is executed exactly once even with multiple competing workers', async () => {
+    const scheduler = new JobScheduler({ pollIntervalMs: 10 });
+    scheduler.start(3);
+    let executions = 0;
+    let maxConcurrent = 0;
+    let active = 0;
+
+    scheduler.submit({
+      id: 'job-once',
+      runAt: Date.now(),
+      execute: async () => {
+        active++;
+        maxConcurrent = Math.max(maxConcurrent, active);
+        await sleep(30);
+        active--;
+        executions++;
+      },
+    });
+
+    await waitFor(() => executions === 1);
+    await sleep(100); // give competing workers a chance to (incorrectly) re-run
+
+    expect(executions).toBe(1);
+    expect(maxConcurrent).toBe(1); // never two workers on the same job
+    expect(scheduler.getPendingCount()).toBe(0);
+    expect(scheduler.getActiveLeaseCount()).toBe(0);
+    await scheduler.stop();
+  });
+
+  test('independent jobs are each processed exactly once across workers', async () => {
+    const scheduler = new JobScheduler({ pollIntervalMs: 5 });
+    scheduler.start(4);
+    const executed = new Set<string>();
+    const workerUsed = new Set<string>();
+
+    for (let i = 0; i < 8; i++) {
+      const id = `job-${i}`;
+      scheduler.submit({
+        id,
+        runAt: Date.now(),
+        execute: async (ctx) => {
+          executed.add(id);
+          workerUsed.add(ctx.workerId);
+          await sleep(10);
+        },
+      });
+    }
+
+    await waitFor(() => executed.size === 8);
+    await sleep(80);
+
+    expect(executed.size).toBe(8);
+    expect(workerUsed.size).toBeGreaterThan(1); // work was distributed
+    expect(scheduler.getPendingCount()).toBe(0);
+    await scheduler.stop();
+  });
+
+  test('a job is not executed before its scheduled runAt', async () => {
+    const scheduler = new JobScheduler({ pollIntervalMs: 10 });
+    scheduler.start(2);
+    let executions = 0;
+
+    scheduler.submit({
+      id: 'delayed',
+      runAt: Date.now() + 200,
+      execute: async () => {
+        executions++;
+      },
+    });
+
+    await sleep(80);
+    expect(executions).toBe(0); // not due yet
+
+    await waitFor(() => executions === 1);
+    await scheduler.stop();
+  });
+
+  test('reschedule() re-queues a job at a future time (retry pattern)', async () => {
+    const scheduler = new JobScheduler({ pollIntervalMs: 10 });
+    scheduler.start(2);
+    let executions = 0;
+    let scheduledAgain = false;
+
+    scheduler.submit({
+      id: 'retrying',
+      runAt: Date.now(),
+      execute: async (ctx) => {
+        executions++;
+        if (!scheduledAgain) {
+          scheduledAgain = true;
+          ctx.reschedule(Date.now() + 30); // like a backoff retry
+        }
+      },
+    });
+
+    await waitFor(() => executions === 2);
+    await sleep(80);
+    expect(executions).toBe(2); // ran twice, then done
+    expect(scheduler.getPendingCount()).toBe(0);
+    await scheduler.stop();
+  });
+
+  test('heartbeat renewal prevents a healthy long-running job from being stolen', async () => {
+    const scheduler = new JobScheduler({
+      leaseDurationMs: 30, // lease would expire quickly...
+      leaseRenewIntervalMs: 5, // ...but the worker heartbeats faster than expiry
+      pollIntervalMs: 5,
+    });
+    scheduler.start(2);
+    let executions = 0;
+
+    scheduler.submit({
+      id: 'long-running',
+      runAt: Date.now(),
+      execute: async () => {
+        executions++;
+        await sleep(120); // far longer than the 30ms lease
+      },
+    });
+
+    await waitFor(() => executions === 1);
+    await waitFor(() => scheduler.getActiveLeaseCount() === 0);
+    await sleep(100); // any stolen re-execution would have happened by now
+
+    expect(executions).toBe(1); // no double processing while healthy
+    await scheduler.stop();
+  });
+
+  test('an expired lease is reclaimed by another worker (crash recovery, at-least-once)', async () => {
+    const scheduler = new JobScheduler({
+      leaseDurationMs: 20, // short lease
+      leaseRenewIntervalMs: 60000, // effectively never heartbeats (simulates a dead worker)
+      pollIntervalMs: 5,
+    });
+    scheduler.start(3);
+    let executions = 0;
+
+    scheduler.submit({
+      id: 'crashed-worker',
+      runAt: Date.now(),
+      execute: async () => {
+        executions++;
+        await sleep(40); // lease expires mid-execution because there is no heartbeat
+      },
+    });
+
+    await waitFor(() => executions >= 2); // a second worker reclaimed the job
+    await scheduler.stop();
+
+    expect(executions).toBeGreaterThanOrEqual(2);
+  });
+
+  test('clear() drops queued jobs without executing them', async () => {
+    const scheduler = new JobScheduler({ pollIntervalMs: 10 });
+    scheduler.start(2);
+    let executions = 0;
+
+    scheduler.submit({
+      id: 'to-clear',
+      runAt: Date.now(),
+      execute: async () => {
+        executions++;
+      },
+    });
+
+    scheduler.clear();
+    expect(scheduler.getPendingCount()).toBe(0);
+    await sleep(120);
+    expect(executions).toBe(0);
+    await scheduler.stop();
+  });
+
+  test('onError is invoked when a job handler throws', async () => {
+    const scheduler = new JobScheduler({ pollIntervalMs: 10 });
+    scheduler.start(2);
+    const onError = jest.fn();
+
+    scheduler.submit({
+      id: 'throwing',
+      runAt: Date.now(),
+      execute: async () => {
+        throw new Error('boom');
+      },
+      onError,
+    });
+
+    await waitFor(() => onError.mock.calls.length === 1);
+    expect(onError.mock.calls[0][0]).toEqual(new Error('boom'));
+    expect(scheduler.getPendingCount()).toBe(0);
+    await scheduler.stop();
+  });
+
+  test('stop() halts workers so queued jobs are no longer processed', async () => {
+    const scheduler = new JobScheduler({ pollIntervalMs: 10 });
+    const workers = scheduler.start(2);
+    expect(workers).toHaveLength(2);
+    expect(scheduler.getWorkerIds()).toHaveLength(2);
+
+    await scheduler.stop();
+    expect(scheduler.getWorkerIds()).toHaveLength(0);
+
+    let executions = 0;
+    scheduler.submit({
+      id: 'after-stop',
+      runAt: Date.now(),
+      execute: async () => {
+        executions++;
+      },
+    });
+
+    await sleep(150);
+    expect(executions).toBe(0);
+  });
+
+  test('submit() with a future runAt is claimable only after runAt', async () => {
+    const scheduler = new JobScheduler({ pollIntervalMs: 10 });
+    scheduler.start(1);
+    let executions = 0;
+
+    scheduler.submit({
+      id: 'future',
+      runAt: Date.now() + 150,
+      execute: async () => {
+        executions++;
+      },
+    });
+
+    await sleep(60);
+    expect(executions).toBe(0);
+    await waitFor(() => executions === 1);
+    await scheduler.stop();
+  });
+});
\ No newline at end of file
diff --git a/webhook-delivery-service/tests/webhook.test.ts b/webhook-delivery-service/tests/webhook.test.ts
index efcf5de..7d1cd84 100644
--- a/webhook-delivery-service/tests/webhook.test.ts
+++ b/webhook-delivery-service/tests/webhook.test.ts
@@ -3,6 +3,7 @@ import request from 'supertest';
 import express from 'express';
 import app from '../src/index';
 import * as delivery from '../src/delivery';
+import { JobScheduler, LeaseStore } from '../src/jobScheduler';
 import { validateUrlForSsrf, generateSignatures, verifyHmacSignature, verifyEd25519Signature } from '../src/security';
 import { resetMetricCache, getStatsSummary } from '../src/metrics';
 import axios from 'axios';
@@ -207,6 +208,63 @@ describe('Webhook Delivery Service Suite', () => {
     });
   });
 
+  describe('6.5 Distributed Job Scheduler with Lease-based Worker Claiming', () => {
+    test('runs multiple scheduler workers that claim jobs under leases', async () => {
+      const status = delivery.getSchedulerStatus();
+      expect(status.workers.length).toBeGreaterThanOrEqual(2);
+      expect(status.pendingCount).toBe(0);
+      expect(status.activeLeases).toBe(0);
+    });
+
+    test('a delivery completes exactly once despite multiple competing workers', async () => {
+      mockedAxios.post.mockResolvedValue({ status: 200, data: {} });
+
+      const jobId = delivery.enqueueWebhook(
+        { event: 'scheduler_test', timestamp: Date.now(), data: {} },
+        'https://webhook.receiver.com/hook',
+        'secret'
+      );
+
+      await new Promise((resolve) => setTimeout(resolve, 150));
+
+      const logs = delivery.getDeliveryLogs();
+      const log = logs.find((l) => l.id === jobId);
+      expect(log).toBeDefined();
+      expect(log!.status).toBe('SUCCESS');
+      expect(log!.attempts).toBe(1); // never double-delivered
+      expect(delivery.getSchedulerStatus().pendingCount).toBe(0);
+      expect(delivery.getSchedulerStatus().activeLeases).toBe(0);
+    });
+
+    test('lease fencing prevents two workers from executing the same task', async () => {
+      const scheduler = new JobScheduler({ pollIntervalMs: 5, leaseDurationMs: 30000 });
+      scheduler.start(4);
+      let executions = 0;
+
+      scheduler.submit({
+        id: 'fenced-job',
+        runAt: Date.now(),
+        execute: async () => {
+          executions++;
+          await new Promise((resolve) => setTimeout(resolve, 20));
+        },
+      });
+
+      await new Promise((resolve) => setTimeout(resolve, 150));
+      await scheduler.stop();
+      expect(executions).toBe(1);
+    });
+
+    test('expired leases are reclaimed by another worker (crash recovery)', async () => {
+      const store = new LeaseStore();
+      store.claim('j1', 'worker-a', 10);
+      await new Promise((resolve) => setTimeout(resolve, 30));
+      const attempt = store.claim('j1', 'worker-b', 1000);
+      expect(attempt.granted).toBe(true);
+      expect(attempt.reclaimed).toBe(true);
+    });
+  });
+
   describe('6. Monitoring endpoints', () => {
     test('GET /health should return system status UP', async () => {
       const response = await request(app).get('/health');