Design a Distributed Job Scheduler & Task Execution Queue

1. Problem Statement & Scope Clarification

System Mission

Design an enterprise-grade, distributed, fault-tolerant Job Scheduling and Task Execution Platform (equivalent to Temporal, Quartz Scheduler, or AWS EventBridge Scheduler) capable of scheduling, orchestrating, and executing billions of delayed, recurring (cron), and one-off asynchronous workloads with at-least-once dispatch and fenced, effectively-once execution (a task can be dispatched twice after a crash, but fencing tokens guarantee only one worker's side effects are ever committed), dynamic worker auto-scaling, conditional write lease fencing, and end-to-end execution observability.

Functional Requirements

1. Diverse Schedule Types (POST /v1/jobs):
   • One-Off Delayed Tasks: Execute once at an exact future epoch timestamp (e.g., in 45 minutes).
   • Recurring Cron Workloads: Execute repeatedly based on standard 5-to-6-field cron expressions (e.g., 0 0 * * *).
   • Instant Asynchronous Tasks: Enqueue immediate background executions with priority routing.
2. Prioritized Workload Dispatch: Segregate jobs into priority tiers (CRITICAL, HIGH, DEFAULT, LOW) to guarantee that critical payment workflows never suffer head-of-line blocking behind bulk report generation.
3. Task Lifecycle Tracking: Real-time tracking of task states (SCHEDULED, CLAIMED, DISPATCHED, RUNNING, SUCCEEDED, FAILED, CANCELLED).
4. Resilient Retries & Dead-Letter Queues (DLQ): Configurable exponential backoff with full randomized jitter and automatic DLQ routing after max attempts.
5. Worker Heartbeating & Lease Fencing: Long-running jobs renew heartbeats every 10 seconds. If a worker dies mid-execution, its lease expires, and the task is safely reclaimed by another node without split-brain duplicate execution.

Non-Functional Requirements (SLAs & SLOs)

• High Availability: $99.999\%$ ("five nines") uptime SLA for the scheduling ingestion and dispatching plane.
• Timing Accuracy: Dispatched within $< 500\text{ ms}$ of the target execution time ($P99 < 1000\text{ ms}$).
• Task Durability: $99.999999999\%$ (11 9s) zero task loss guarantee prior to execution.
• Throughput & Scale: Support $100\text{ Million}$ active scheduled jobs, $500\text{ Million}$ daily completed task runs, and peak dispatch bursts of $50,000\text{ tasks/sec}$.

2. Capacity & Scale Estimation (Back-of-the-Envelope Math)

Ingestion & Dispatch Scale

• Total Active Scheduled Tasks: $100,000,000$ ($100\text{M tasks}$).
• Daily Completed Workloads: $500,000,000\text{ tasks/day}$ ($500\text{M tasks/day}$).
• Average Dispatch Throughput: $\text{Average Dispatch QPS} = \frac{500,000,000}{86,400\text{ sec}} \approx \mathbf{5,787\text{ tasks/sec}}$
• Peak Dispatch Throughput ($5\times$ Top-of-Hour Multiplier): Recurring cron jobs spike massively at the start of hours (00:00, 01:00): $\text{Peak Dispatch QPS} = 5,787 \times 5 \approx \mathbf{28,900\text{ tasks/sec}}$

Storage Footprint (Active State vs Historical Lake)

• Task Definition Size:
  • job_id ($16\text{B}$), tenant_id ($16\text{B}$), target_url ($128\text{B}$), payload_json ($512\text{B}$), cron_expression ($32\text{B}$), next_run_epoch_ms ($8\text{B}$), retry_policy ($64\text{B}$), metadata pointers ($124\text{B}$) $\approx \mathbf{1\text{ KB/task}}$.
• Active Task Store (Amazon DynamoDB): $\text{Active Storage} = 100 \times 10^6\text{ tasks} \times 1\text{ KB} = \mathbf{100\text{ GB}}$
• Monthly Execution History Lake (Amazon S3 via Kinesis Firehose): $\text{Monthly History} = 500\text{M tasks/day} \times 1\text{ KB} \times 30\text{ days} = \mathbf{15\text{ TB/month}}$

Redis Time-Bucket Memory Sizing

To prevent a single giant Redis Sorted Set (ZSET) from bottlenecking memory and CPU, scheduled tasks are partitioned into 1-Minute Time Buckets (schedule_bucket:YYYYMMDDHHMM):

• Jobs due within the next 2 hours are buffered in Redis: $120\text{ minutes} \times (5,787\text{ tasks/sec} \times 60\text{ sec}) \approx 41.6 \times 10^6\text{ tasks}$
• Each ZSET entry stores member: job_id ($16\text{ bytes}$) and score: epoch_ms ($8\text{ bytes}$).
• With Redis jemalloc and skip-list pointer overhead $\approx 64\text{ bytes/entry}$: $\text{Active Redis Buffer RAM} = 41.6 \times 10^6 \times 64\text{ bytes} \approx \mathbf{2.66\text{ GB RAM}}$ An Amazon ElastiCache Redis Cluster with $3\times$ replicas easily hosts this dataset in memory.

3. High-Level Architecture & AWS Component Mapping

Synthesizing vector architecture diagram...

Data Flow Walkthrough

1. Job Scheduling Ingress: A client issues POST /v1/jobs with a schedule and target payload. The Job Management Service generates a unique 128-bit job_id, persists the master record to DynamoDB, and (if the execution time is within the next 2 hours) inserts (score=run_at_epoch_ms, member=job_id) into the appropriate 1-minute Redis Sorted Set.
2. Time-Slice Scanning: The Scanner Fleet uses consistent hashing to assign 1-minute time buckets across scanner nodes. Every $250\text{ ms}$, scanners execute ZRANGEBYSCORE for tasks with score <= now_ms.
3. Two-Phase Claim & Fencing: For each due job, the scanner executes a conditional DynamoDB update (status = 'DISPATCHED', version = version + 1). Only the scanner that successfully increments the version token claims the job, preventing duplicate dispatches.
4. Priority Queueing: Claimed jobs are dispatched to Amazon SQS queues segmented by priority tier (CRITICAL, HIGH, DEFAULT).
5. Execution & Heartbeat Loop: Workers pull messages from SQS, transition the job state to RUNNING in DynamoDB, and start a background heartbeat thread that renews the 30-second execution lease every 10 seconds. Upon successful completion, the task status transitions to SUCCEEDED.

Concrete Step-by-Step Request Walkthrough: Tracing a Scheduled Delayed Task

4. API Interface Design & Wire Protocol

1. Schedule a Job (POST /v1/jobs)

httpCopy
POST /v1/jobs HTTP/1.1
Host: scheduler.production.aws.internal
Authorization: Bearer eyJhbGciOi...
Idempotency-Key: idemp_job_8492019481
Content-Type: application/json

{
  "tenant_id": "tenant_enterprise_01",
  "job_name": "monthly_billing_aggregation",
  "schedule_type": "CRON",
  "cron_expression": "0 2 1 * *",
  "timezone": "America/New_York",
  "priority": "HIGH",
  "target": {
    "type": "HTTP_WEBHOOK",
    "endpoint": "https://billing.internal.aws/invoices/generate",
    "method": "POST",
    "timeout_seconds": 120,
    "headers": {
      "Content-Type": "application/json"
    }
  },
  "payload": {
    "billing_cycle": "2026-09",
    "dry_run": false
  },
  "retry_policy": {
    "max_retries": 3,
    "initial_interval_sec": 10,
    "backoff_multiplier": 2.0,
    "max_interval_sec": 300
  }
}

Response: 201 Created
Content-Type: application/json

{
  "job_id": "job_9847291048",
  "tenant_id": "tenant_enterprise_01",
  "status": "SCHEDULED",
  "schedule_type": "CRON",
  "next_run_time_epoch_ms": 1767225600000,
  "next_run_time_iso": "2026-10-01T02:00:00.000Z",
  "created_at": "2026-09-16T12:00:00.000Z"
}

2. Inspect Job Execution Status (GET /v1/jobs/{job_id})

httpCopy
GET /v1/jobs/job_9847291048 HTTP/1.1
Host: scheduler.production.aws.internal
Accept: application/json

Response: 200 OK
Content-Type: application/json

{
  "job_id": "job_9847291048",
  "status": "RUNNING",
  "last_heartbeat_at": "2026-09-16T12:00:10.000Z",
  "current_lease_worker_id": "worker-ecs-task-04",
  "current_attempt": 1,
  "fencing_token": 104
}

5. Data Models & Storage Architecture

1. Amazon DynamoDB Single-Table Design (JobSchedulerMasterTable)

Global Secondary Indexes (GSIs)

• GSI1 (Reaper Index):
  • GSI1-PK: status#shard where shard = hash(job_id) % 16 (e.g. RUNNING#07)
  • GSI1-SK: lease_expiry_ts
  • Allows background reaper daemons to discover abandoned tasks where lease_expiry_ts < now_ms with 16 targeted range queries (one per shard, run in parallel).
  • Why the shard suffix: a GSI keyed on bare status would put every RUNNING job (millions of items, tens of thousands of writes per second as leases renew) into a single GSI partition, and DynamoDB caps any one partition at $1{,}000\text{ WCU}$ / $3{,}000\text{ RCU}$. Write-sharding the key spreads the heartbeat load across 16 partitions; the reaper simply fans out its query.

2. Redis Time-Bucket Sorted Sets

• Key Pattern: schedule:<epoch_minute> (e.g., schedule:202609161205)
• Data Structure: ZSET
  • Score: epoch_ms (exact execution timestamp)
  • Member: job_id
• Expiry: Key TTL set to 4 hours (EXPIRE 14400) after the bucket minute completes, ensuring automatic memory cleanup.