Job Scheduler at Scale

Step 1: Job Scheduler Requirements

Job Scheduler Functional Requirements

• Schedule one-time jobs (run at a specific timestamp) and recurring jobs (cron expressions)
• Execute jobs within a target window of their scheduled time
• Retry failed jobs with configurable retry policies
• Support job dependencies (DAGs, Directed Acyclic Graphs, meaning jobs form a dependency chain with no circular references). Job B runs only after job A succeeds
• Priority levels (critical, high, normal, low) to control execution order
• Job status tracking: pending → queued → running → success/failed/dead-lettered
• APIs: create job, cancel job, pause/resume, get status, list executions

Job Scheduler Non-Functional Requirements

• Scale: 10 million jobs/day
• Timeliness: execute within 5 seconds of scheduled time
• Reliability: at-least-once execution guarantee
• No duplicates: exactly-once semantics via distributed locking
• Fault tolerance: handle worker crashes, network partitions, scheduler failover
• Horizontal scaling: add capacity without downtime

Step 2: Job Scheduler Estimation

Job Scheduler Throughput

• 10M jobs/day → ~115 jobs/sec average
• Peak (3x average) → ~350 jobs/sec

Job Scheduler Storage

• Job metadata: ~2 KB per job (ID, schedule, payload, status, history)
• 10M jobs × 2 KB = 20 GB/day of new job records
• With 30-day retention: ~600 GB total

Worker Pool

• Average job duration: 30 seconds
• At peak: 350 jobs/sec × 30s = 10,500 concurrent jobs
• With 10 jobs per worker instance: ~1,050 worker instances at peak
• Provision ~1,200 workers for headroom

Queue Depth

• If workers lag by 60 seconds at peak: 350 × 60 = ~21,000 messages in queue
• Design queue to handle 100K+ messages for burst absorption

Step 3: Job Scheduler High-Level Design

┌─────────────────────────────────────────────────────────────────────┐
│                         API Gateway / CLI                           │
└──────────────────────────────┬──────────────────────────────────────┘
                               │ create/cancel/status
                               ▼
┌──────────────────────────────────────────────────────────────────────┐
│                        Job Service (CRUD)                            │
└──────────────────────────────┬───────────────────────────────────────┘
                               │ writes
                               ▼
┌──────────────────────────────────────────────────────────────────────┐
│                     Job Store (PostgreSQL / Vitess)                  │
│  [job_id, schedule, next_run_at, status, priority, payload, ...]     │
└──────────────────────────────┬───────────────────────────────────────┘
                               │ polls next_run_at <= now()
                               ▼
┌──────────────────────────────────────────────────────────────────────┐
│              Scheduler Service (leader-elected or partitioned)       │
│  - Finds due jobs, acquires lock, enqueues to priority queue         │
└──────────────────────────────┬───────────────────────────────────────┘
                               │ enqueue
                               ▼
┌──────────────────────────────────────────────────────────────────────┐
│              Priority Job Queue (Redis / Kafka / SQS)                │
│  [critical] [high] [normal] [low]                                    │
└──────────────────────────────┬───────────────────────────────────────┘
                               │ dequeue
                               ▼
┌──────────────────────────────────────────────────────────────────────┐
│                     Worker Pool (auto-scaled)                        │
│  - Execute job, report result, send heartbeat                        │
└──────────────────────────────┬───────────────────────────────────────┘
                               │ results
                               ▼
┌─────────────────┐    ┌─────────────────────┐    ┌─────────────────────┐
│  Result Store   │    │  Dead Letter Queue  │    │  Monitoring/Alerting│
│  (S3 / DB)      │    │  (exhausted retries)│    │  (Prometheus/PD)    │
└─────────────────┘    └─────────────────────┘    └─────────────────────┘

Step 4: Job Scheduler Deep Dive

Job Storage Schema

CREATE TABLE jobs (
    job_id          UUID PRIMARY KEY,
    name            VARCHAR(255),
    schedule        VARCHAR(100),       -- cron expression or NULL for one-time
    next_run_at     TIMESTAMP NOT NULL, -- indexed, used for polling
    status          ENUM('pending','queued','running','success','failed','dead'),
    priority        SMALLINT DEFAULT 2, -- 0=critical, 1=high, 2=normal, 3=low
    payload         JSONB,              -- execution context
    max_retries     SMALLINT DEFAULT 3,
    retry_count     SMALLINT DEFAULT 0,
    locked_by       VARCHAR(255),       -- worker ID holding the lease
    lock_expires_at TIMESTAMP,
    dag_id          UUID,               -- NULL if standalone
    depends_on      UUID[],             -- upstream job IDs
    created_at      TIMESTAMP,
    updated_at      TIMESTAMP
);

CREATE INDEX idx_due_jobs ON jobs (next_run_at) WHERE status = 'pending';
CREATE INDEX idx_dag ON jobs (dag_id) WHERE dag_id IS NOT NULL;

Scheduler Design

Option A: Leader-elected single scheduler:

• One active scheduler polls SELECT ... WHERE next_run_at <= now() AND status = 'pending' ORDER BY priority, next_run_at LIMIT 500 FOR UPDATE SKIP LOCKED (SKIP LOCKED means: lock these rows for me, but if another process already locked some, skip them instead of waiting, which prevents contention between concurrent schedulers)
• Standby replicas take over via leader election (ZooKeeper / etcd lease, a time-limited lock that expires if the holder dies)
• Simple but becomes a bottleneck at extreme scale

Option B: Partitioned schedulers (preferred at scale):

• Partition jobs by hash(job_id) % N across N scheduler instances
• Each scheduler owns a partition range and only polls its subset
• Rebalance partitions when instances join/leave (consistent hashing)

Exactly-Once Execution

1. Scheduler atomically sets status = 'queued' and locked_by = scheduler_id using FOR UPDATE SKIP LOCKED
2. Worker acquires a lease (lock_expires_at = now + 2× expected duration)
3. If worker completes, it releases the lock and writes result
4. If lock expires (worker died), scheduler reclaims the job and re-enqueues
5. Workers use idempotency keys: the job payload includes a unique execution_id; downstream systems deduplicate on this key

Worker Management

• Workers send heartbeats every 10 seconds to a registry (Redis hash)
• Scheduler runs a reaper process every 30 seconds: any worker missing 3 heartbeats is declared dead
• Dead worker's in-progress jobs (where lock_expires_at < now()) are reset to pending
• Workers pull jobs (not pushed), providing natural backpressure when workers are busy

Job Scheduler Retry Strategy

retry_delay = base_delay × 2^(retry_count) + random_jitter

• Base delay: 10 seconds
• Max retries: configurable per job (default 3)
• After max retries exhausted → move to Dead Letter Queue, set status = dead
• DLQ jobs can be manually inspected and replayed via admin API
• Alert fires when DLQ depth exceeds threshold

DAG Execution

• Jobs within a DAG store dag_id and depends_on (list of upstream job IDs)
• When a job completes successfully, the DAG executor queries: "Are all of job X's dependencies now in success status?"
• If yes → transition job X from blocked to pending (makes it eligible for scheduling)
• Topological sort at DAG creation time validates no cycles
• If any upstream fails and exhausts retries → downstream jobs are marked skipped

Priority Queues

• Separate physical queues per priority level (or weighted fair queuing)
• Workers always drain critical first, then high, then normal, then low
• Starvation prevention: promote jobs that have waited longer than a threshold (aging)
• Critical queue has dedicated worker capacity that cannot be borrowed

Scaling Strategy

Key Takeaways

• Poll with FOR UPDATE SKIP LOCKED to avoid contention between multiple scheduler instances picking the same job
• Lease-based locking with expiry handles worker crashes without manual intervention. No job is permanently lost
• Idempotency keys in job payloads are essential for exactly-once semantics even with at-least-once delivery
• DAG dependencies are resolved at completion time by checking upstream statuses, not by pre-scheduling the entire graph
• Priority + aging prevents starvation while ensuring critical jobs always execute first
• Partition the scheduler rather than relying on a single leader to achieve horizontal scale beyond ~1K jobs/sec