Design a Distributed Job Scheduler

A job scheduler runs tasks at a specified time or on a recurring schedule (for example, "send this report every day at 06:00" or "run this in 10 minutes"). A single-machine scheduler such as cron is a single point of failure and limited to one host. A distributed job scheduler stores jobs durably, dispatches them to a fleet of workers, and keeps running when nodes fail. Examples include Quartz clustered mode, Airflow, Kubernetes CronJobs, and cloud schedulers.

Requirements

Functional

• Create, update, pause, and delete jobs.
• One-off jobs (run at time T) and recurring jobs (cron expressions or fixed intervals).
• Jobs have a payload and a target (an HTTP endpoint, a queue topic, or a registered handler).
• Automatic retries with backoff on failure; a maximum attempt count.
• Query job status and execution history.
• Optional: job priorities, dependencies between jobs (DAGs), per-job timeouts.

Non-functional

• Reliability. A job that is due must run; no job is silently lost. Execution guarantee is at-least-once, with idempotent handlers.
• Timeliness. Jobs start within a small delay (for example, seconds) of their scheduled time.
• Scalability. Millions of scheduled jobs and thousands of executions per second at peak (for example, many jobs scheduled at the top of the hour).
• High availability. No single point of failure; scheduler node failure must not delay jobs beyond seconds.
• Isolation. A misbehaving job or tenant must not starve others.

Estimation

Assumptions: 10 million jobs stored, of which 5 million are hourly recurring jobs and the rest are one-off or infrequent; this gives about 5 million executions per hour at steady state.

• Average execution rate: 5,000,000 / 3,600 ≈ 1,400 executions per second.
• Bursts: jobs cluster at minute and hour boundaries. If 20 percent of the hourly executions are scheduled at minute 0: 0.2 × 5,000,000 = 1,000,000 executions in the first minute, about 16,700 per second at peak. Design for at least 20,000 per second dispatch.
• Job record: about 1 KB (schedule, payload, status, metadata). Storage: 10,000,000 × 1 KB = 10 GB of definitions.
• Execution history: 5,000,000 per hour × 24 = 120 million rows per day × 200 B = 24 GB per day. Retain for days, archive or expire beyond.
• Workers: if the average job runs 10 seconds and 1,400 start per second, concurrent executions ≈ 14,000. With 50 concurrent slots per worker, about 280 workers.

What the numbers imply

• The 16,700 per second burst, not the 1,400 average, sizes the system: dispatch and queue capacity are provisioned for 20,000 per second, and the burst is absorbed by the queue, with workers autoscaled before round hours.
• At 20,000 claims per second, claim in batches of 500 to 1,000 rows (20 to 40 batches per second) instead of row by row; a single relational primary can do this, but the history writes (120 million rows per day, about 1,400 per second average) push you to a separate store from the job table.
• 10 GB of definitions fits in memory of one database node; sharding is for write and claim throughput, not capacity.

API design

POST   /v1/jobs                    { "schedule": "0 6 * * *" | "run_at": "2026-10-01T12:00:00Z",
                                     "target": "https://svc/report", "payload": {...},
                                     "retry": {"max_attempts": 5, "backoff": "exponential"},
                                     "timeout_sec": 300, "idempotency_key": "..." }
GET    /v1/jobs/{job_id}
PATCH  /v1/jobs/{job_id}           (pause, resume, change schedule)
DELETE /v1/jobs/{job_id}
GET    /v1/jobs/{job_id}/executions?limit=50&cursor=...

idempotency_key on creation prevents duplicate jobs when a client retries the create call. Worker-facing interface (internal): poll/lease, heartbeat(execution_id), complete(execution_id, status).

Data model

Two entities with different lifecycles.

jobs(job_id PK, tenant_id, schedule, next_run_time, target, payload, retry_policy, status, version)
executions(execution_id PK, job_id, scheduled_time, attempt, state, worker_id, lease_expires_at, started_at, finished_at, result)

Execution states: pending, running, succeeded, failed, retry_wait.

Storage reasoning:

• The scheduler's core query is "find jobs with next_run_time <= now that are not yet claimed", ordered by time. This needs an index on next_run_time and atomic claim of rows.
• A relational database (PostgreSQL, MySQL) with a (status, next_run_time) index supports transactional claiming, and 10 GB of definitions fits comfortably. At larger scale, shard by job_id hash so each shard has its own due-query (see sharding).
• A wide-column or key-value store can work if the partition key encodes a time bucket (for example, bucket = minute(next_run_time)) with the job id as clustering key; this makes "due jobs in this minute" a single-partition read. The cost is weaker transactions, so claiming needs conditional writes.
• The execution history is append-heavy and can go to a separate table, or store, partitioned by date so old partitions are dropped cheaply.

High-level design

flowchart LR
    U["Clients / API"] --> API["Job API service"]
    API --> DB[("Job store")]
    subgraph SC["Scheduler cluster"]
        S1["Scheduler node 1"]
        S2["Scheduler node 2"]
    end
    DB <--> S1
    DB <--> S2
    S1 --> Q[("Ready queue")]
    S2 --> Q
    Q --> W["Worker pool"]
    W -->|"heartbeat, complete"| DB
    W --> T["Target services"]
    MON["Watchdog: expired leases"] --> DB
    MON --> Q

Component responsibilities:

• Job API service. Validates requests, computes the first next_run_time from the schedule, and writes the job.
• Job store. Source of truth for definitions and execution state.
• Scheduler nodes. Periodically find due jobs, create an execution record, and put it on the ready queue. Several nodes run in parallel, each owning a partition of the job space.
• Ready queue (message queue). Buffers due work between the scheduler and workers, provides load leveling and priority lanes.
• Workers. Pull executions, run the target, send heartbeats, and report the result.
• Watchdog. Finds executions whose lease expired (worker died) and requeues them.

Dispatch flow

sequenceDiagram
    participant S as Scheduler
    participant D as Job store
    participant Q as Ready queue
    participant W as Worker
    S->>D: claim jobs whose next_run_time has passed
    D-->>S: jobs, marked claimed
    S->>D: create execution, set next_run_time to next occurrence
    S->>Q: enqueue execution_id
    W->>Q: pull execution_id
    W->>D: acquire lease (running, lease_expires_at)
    W->>W: run job, extend lease by heartbeat
    W->>D: complete (succeeded or failed)
    Note over D: on failure, schedule retry with backoff

Deep dive

1. Finding due jobs efficiently

Options for time-based dispatch:

A common design combines the first two: the database is the durable store; each scheduler pre-loads jobs due in the next minute into an in-memory min-heap and pops them at their exact time.

import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.PriorityQueue;

public class Main {
    record Due(long runAt, int jobId) {} // runAt: seconds since epoch

    static final class DueQueue {
        // PriorityQueue is a min-heap: the head is the smallest element under the comparator.
        private final PriorityQueue<Due> heap =
                new PriorityQueue<>(Comparator.comparingLong(Due::runAt).thenComparingInt(Due::jobId));

        void schedule(long runAt, int jobId) {
            heap.add(new Due(runAt, jobId));
        }

        /** Removes and returns every job due at or before now, earliest first. */
        List<Due> pollDue(long now) {
            var due = new ArrayList<Due>();
            while (!heap.isEmpty() && heap.peek().runAt() <= now) {
                due.add(heap.poll());
            }
            return due;
        }

        int waiting() {
            return heap.size();
        }
    }

    public static void main(String[] args) {
        var queue = new DueQueue();
        queue.schedule(120, 1);
        queue.schedule(60, 2);
        queue.schedule(90, 3);
        queue.schedule(60, 4);

        long now = 100;
        for (var due : queue.pollDue(now)) {
            System.out.println("dispatch job " + due.jobId() + " scheduled for t=" + due.runAt());
        }
        System.out.println("still waiting: " + queue.waiting());
    }
}

dispatch job 2 scheduled for t=60
dispatch job 4 scheduled for t=60
dispatch job 3 scheduled for t=90
still waiting: 1

java.util.PriorityQueue is a binary min-heap ordered by the comparator: due time first, job id as the tie-break. It gives O(log n) insert and pop and O(1) peek at the next due job.

2. Preventing duplicate dispatch across scheduler nodes

Two scheduler nodes must not both dispatch the same due job. Alternatives:

• Row-level claim. Use SELECT ... FOR UPDATE SKIP LOCKED (PostgreSQL, MySQL 8) or a conditional update (UPDATE ... SET status='claimed', owner=me WHERE job_id=? AND status='scheduled') so only one node succeeds. This needs no coordination beyond the database.
• Static partitioning. Assign each scheduler node a range of the job-id hash space via a coordination service, so nodes never overlap; when a node dies, its ranges are reassigned (see consistent hashing and leader election).
• Single leader. One elected scheduler dispatches, others stand by. Simple, but the leader is a throughput ceiling. Use fencing tokens so a paused old leader cannot dispatch after a new one is elected.

Even with these, duplicates can still happen on failures, so handlers must be idempotent (below).

Concrete settings. Poll interval 1 second; pre-load window 1 minute; claim batch 500 to 1,000 with SKIP LOCKED. Heartbeat every 10 seconds and lease of 30 seconds (three missed heartbeats), with a watchdog scan every 10 seconds, so a dead worker's job restarts within roughly 40 seconds; shorten the lease for jobs that need faster recovery, at the cost of more false expirations from GC pauses. Shard the job table by hash(job_id) into, say, 16 shards, each with its own due-index; in a time-bucketed store, split the hot minute-0 bucket into (bucket, hash mod N) sub-partitions so 1 million jobs do not land on one partition.

3. Execution guarantees, leases, and failure recovery

Exactly-once execution across a network is not achievable in general (see delivery semantics). The scheduler provides at-least-once:

1. A worker takes a lease on an execution: state running, lease_expires_at = now + T.
2. The worker extends the lease with heartbeats while running.
3. If the worker crashes, heartbeats stop and the lease expires. The watchdog resets the execution to pending and requeues it, so another worker retries it.
4. A worker that was only slow (not dead) may finish after its lease expired and another worker restarted the job, so the job may run twice. To prevent double side effects, the worker completes with a conditional update on (execution_id, attempt), and the target applies work idempotently using the execution_id or a per-job idempotency key as a deduplication key.

At-most-once (mark as running before the run, never retry) is right only when a duplicate is worse than a miss, for example some payment triggers; the choice is per job type.

4. Retries and failure handling

• Retry with exponential backoff and jitter (for example, delay = min(cap, base × 2^attempt), randomized) to avoid synchronized retry storms.
• Distinguish retryable failures (timeout, 5xx) from permanent ones (validation error, 4xx). Do not retry the latter.
• After the maximum attempts, move the execution to a dead-letter state, alert, and allow manual re-run (see DLQs).
• Missed runs. If the scheduler was down at the scheduled time, define a per-job policy: run immediately once on recovery, run all missed occurrences, or skip.
• Overlap policy for recurring jobs. If the previous run is still going when the next is due: skip, queue, or allow concurrent runs.

5. Recurring jobs and time zones

The scheduler computes the next occurrence from the schedule when it dispatches the current one and updates next_run_time in the same transaction that creates the execution. This keeps a recurring job continuously scheduled after a crash: either both changes committed or neither. Cron expressions are evaluated in the job's configured time zone, and daylight-saving transitions need an explicit policy, since a local time can occur twice or not at all. Store times in UTC.

Consistency model. The job store is strongly consistent per job (single row transaction for claim and next_run_time update); cross-job views such as dashboards may lag. Backoff example: delay = min(1 hour, 5 s × 2^attempt) with full jitter, so attempt 5 waits up to 160 seconds and at most 5 attempts.

Bottlenecks and scaling

• Thundering herd at round times. Add optional jitter to schedules that do not need exact times, spread the load by pre-loading and dispatching early into the queue, and scale workers ahead of the hour.
• Job store. Index on (status, next_run_time); shard by job id; archive old executions.
• Scheduler capacity. Run several nodes on disjoint partitions so dispatch scales horizontally.
• Worker pool. Autoscale on queue depth; separate queues per priority and per tenant to prevent one tenant's flood from starving others (weighted fair scheduling and per-tenant concurrency caps).
• Long-running jobs. Require heartbeats; enforce timeouts; support cancellation.
• Observability. Track schedule lag (actual start time minus scheduled time), queue depth, failure rate, and lease expirations.

Failure modes

• Scheduler node dies. Its claimed-but-undispatched jobs stay in claimed state; the watchdog releases claims older than a claim timeout (for example 60 seconds), and other nodes pick them up. Lateness is bounded by the timeout.
• Database primary fails. Dispatch stops until failover (typically tens of seconds); queued executions and running workers continue. Because dispatch is due-time based, jobs run late and the missed-run policy decides whether to run once, all, or skip.
• Queue outage. Executions stay pending in the store; a reconciler re-enqueues pending executions older than a threshold. Enqueue is idempotent by execution_id.
• Worker pool overloaded. Schedule lag grows; per-tenant concurrency caps keep a noisy tenant from consuming all slots. Shed or delay low-priority lanes first.
• At 10x (about 200,000 dispatches per second at peak, 140,000 concurrent executions). One store cannot claim that fast: shard to 64 or more partitions, run a scheduler per shard, use partitioned queues, and push jitter onto schedules that do not need exact times. Execution history at 240 GB per day needs date-partitioned storage with cheap partition drops.

Trade-offs and follow-up questions

• Polling versus push timers. Polling is simple and durable but coarse; in-memory timers are precise but need reloading after restarts.
• Exactly-once. Not achievable end to end; the design gives at-least-once plus idempotent handlers.
• Central database versus sharded store. A single relational store is simplest and handles millions of jobs; sharding is needed when dispatch rate exceeds what one primary can handle.
• What if the scheduler leader pauses (long GC or network delay) and comes back? Fencing tokens and conditional claims reject its stale actions.
• How would you support job dependencies (DAGs)? Store edges, mark a job runnable when all upstream executions succeed, and evaluate readiness on each completion event (as Airflow does).
• Follow-ups: multi-region failover, rate limiting per target service, and cancellation of running jobs.
• What if a job takes longer than its lease and is restarted elsewhere? Two copies may run; the completion write is conditional on (execution_id, attempt) and the target dedups on execution_id.
• What if 1 million jobs are due at 00:00? They queue; workers drain at capacity. Pre-scale, and give exact-time-critical jobs a higher-priority lane.
• Why not use a broker delay queue? Delays are capped (SQS at 15 minutes), and you cannot easily update or cancel a scheduled message.
• What if the clock on a scheduler node is wrong? Use the database's clock for due checks (now() in the query) so all nodes agree.