Design: Distributed Job Scheduler (cron/Airflow at scale)¶

Difficulty: Senior → Staff | Time: 60–75 minutes

1. Problem Statement¶

Build a scheduler like cron, but distributed: users register recurring jobs (0 * * * *, "every hour") and one-off jobs ("run at 2026-08-20T03:00Z"), a fleet of workers executes them, and the system must give a specific correctness guarantee: each scheduled firing runs exactly once — not zero times because a coordinator died at the wrong instant, and not twice because two schedulers both believed they owned that trigger.

Three sub-problems, not one:

1. Trigger correctness. A single process with an in-memory timer is trivially correct until it crashes — then every job whose fire time passed during the outage is silently gone, not late. That's worse than a slow response; nothing indicates the job should have run at all.
2. Execution correctness. A worker can crash mid-job. Does the job retry? On what basis do you decide a retry is safe — does retry risk running the job's side effects twice (charging a customer twice, sending a duplicate email)?
3. Dependency correctness. Real workloads aren't independent cron lines — they're DAGs (Airflow's shape): extract → transform → load, where transform must not start until extract succeeds, and must not run at all if extract failed and there's no meaningful data to transform.

Do not reach for "just use a cron library and a queue." First decide where the durable record of "what should have fired" lives, because that record — not the timer — is what makes exactly-once possible at all.

2. Clarifying Questions¶

3. Functional Requirements¶

• Register a recurring job (cron expression) or a one-off job (specific timestamp)
• Register a DAG: a set of jobs plus edges (job_b depends on job_a)
• Trigger each job at (or as close as SLA allows to) its scheduled time
• Execute the job on a worker, capture success/failure and output/logs
• Retry failed jobs per a configured policy; dead-letter after exhausting retries
• Query the status of a specific run and the run history of a job
• Pause/resume/delete a job definition without losing history

4. Non-Functional Requirements¶

5. Capacity Estimation¶

Job definitions:
  500K registered jobs (cron + DAG nodes)
  Average job definition: ~1 KB (cron expr, command, retry policy, DAG edges)
  → 500 MB of definitions, trivially cacheable in the leader's memory

Trigger volume:
  50K triggers/minute at peak ≈ 830 triggers/second
  Most jobs are hourly/daily — peak clusters at :00 and midnight boundaries
  Minute-boundary spike: jobs scheduled "every minute" (~5% of fleet) all fire
    within the same second → 25K of the 50K can land in one wall-clock second

Job execution duration (long-tailed):
  p50: 2 seconds (cache warmers, health pings)
  p90: 45 seconds (report generation, small ETL steps)
  p99: 20 minutes (large batch transforms)
  max: unbounded — must have a hard timeout, not "unbounded" in practice

Worker fleet:
  10K workers, each handling ~5 concurrent job slots → 50K concurrent job capacity
  At 830 triggers/s and avg 2s duration → ~1,660 concurrent jobs steady state
  Peak minute-boundary burst (25K jobs in ~5s window) needs burst capacity,
    not steady-state sizing — this is the number that decides queue depth, not fleet size

Run history:
  50K triggers/min × 1440 min/day = 72M run records/day
  ~500 bytes/record (job_id, run_id, state transitions, timestamps) → 36 GB/day
  Retain hot (queryable) for 30 days ≈ 1 TB; archive older to cold storage

6. API Design¶

# Job registration
POST /api/v1/jobs
Request: {
  "name": "nightly-report",
  "schedule": { "type": "cron", "expression": "0 3 * * *" },
  "command": "run_report.sh",
  "timeout_s": 1200,
  "retry_policy": { "max_attempts": 3, "backoff": "exponential", "base_s": 30 },
  "depends_on": []              -- non-empty means this job is a DAG node
}
Response: { "job_id": "job_a1b2", "next_run_at": "2026-08-18T03:00:00Z" }

POST /api/v1/jobs/{job_id}/trigger        -- manual/one-off trigger, bypasses schedule
GET  /api/v1/jobs/{job_id}                -- definition + next_run_at
PATCH /api/v1/jobs/{job_id}                -- pause, resume, update schedule
DELETE /api/v1/jobs/{job_id}

# Run status
GET /api/v1/jobs/{job_id}/runs?limit=50
GET /api/v1/runs/{run_id}
Response: {
  "run_id": "run_9f3e",
  "job_id": "job_a1b2",
  "scheduled_for": "2026-08-18T03:00:00Z",
  "state": "running",           -- scheduled|claimed|running|succeeded|failed|retrying|dead_letter
  "attempt": 1,
  "claimed_by": "worker-042",
  "lease_expires_at": "2026-08-18T03:05:00Z"
}

# Worker protocol (internal)
POST /internal/workers/{worker_id}/heartbeat  { "capacity_free": 3 }
POST /internal/runs/claim                     { "worker_id": "worker-042" }
  Response: { "run_id": "run_9f3e", "command": "...", "lease_s": 300 }
POST /internal/runs/{run_id}/lease/renew       { "worker_id": "worker-042" }
POST /internal/runs/{run_id}/complete          { "worker_id": "worker-042", "status": "success", "output_ref": "..." }

7. Data Model¶

-- Job definitions: cron jobs and DAG nodes share one table.
-- depends_on makes this a DAG; empty array is plain independent cron.
CREATE TABLE jobs (
    job_id        VARCHAR(32) PRIMARY KEY,
    name          VARCHAR(200) NOT NULL,
    schedule_type VARCHAR(16) NOT NULL,     -- cron | one_off | dag_child
    cron_expr     VARCHAR(64),              -- null for one_off / dag_child
    run_at        TIMESTAMPTZ,              -- for one_off
    dag_id        VARCHAR(32),              -- groups jobs that form one DAG
    depends_on    TEXT[],                   -- job_ids that must succeed first
    command       TEXT NOT NULL,
    timeout_s     INT NOT NULL,
    retry_policy  JSONB NOT NULL,
    status        VARCHAR(16) NOT NULL,     -- active | paused | deleted
    next_run_at   TIMESTAMPTZ,              -- computed, indexed — the scheduler's work queue
    INDEX idx_next_run (next_run_at) WHERE status = 'active'
);

-- The durable run ledger. Every trigger decision is a row here BEFORE dispatch.
-- This table IS the exactly-once mechanism: it's what a crashed scheduler replays.
CREATE TABLE job_runs (
    run_id          VARCHAR(32) PRIMARY KEY,
    job_id          VARCHAR(32) NOT NULL REFERENCES jobs(job_id),
    dag_run_id      VARCHAR(32),            -- groups all runs of one DAG instance
    scheduled_for   TIMESTAMPTZ NOT NULL,   -- the tick this run corresponds to
    state           VARCHAR(16) NOT NULL,   -- scheduled|claimed|running|succeeded|failed|retrying|dead_letter
    attempt         INT NOT NULL DEFAULT 1,
    claimed_by      VARCHAR(64),            -- worker_id
    lease_expires_at TIMESTAMPTZ,
    created_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
    UNIQUE (job_id, scheduled_for, attempt),  -- dedupe key: same tick can't be logged twice
    INDEX idx_lease (state, lease_expires_at) WHERE state IN ('claimed', 'running')
);

-- Worker registry: a lease, exactly like service-discovery registration
-- (see distributed-systems/fundamentals.md#service-discovery)
CREATE TABLE workers (
    worker_id     VARCHAR(64) PRIMARY KEY,
    capacity      INT NOT NULL,
    capacity_free INT NOT NULL,
    last_heartbeat TIMESTAMPTZ NOT NULL,
    lease_expires_at TIMESTAMPTZ NOT NULL
);

State machine for job_runs.state:

scheduled → claimed → running → succeeded
                             ↘ failed → retrying → claimed (next attempt)
                                     ↘ dead_letter (retries exhausted)
claimed → scheduled            (lease expired before work started — reclaim)
running → scheduled            (lease expired mid-execution — reclaim)

The UNIQUE (job_id, scheduled_for, attempt) constraint is what makes replaying the ledger safe: if the scheduler recomputes "job X should have fired at 03:00" twice (once before a crash, once after recovery), the second insert is a no-op conflict, not a duplicate trigger.

8. Version 1 — simplest thing that works¶

Single scheduler process. In-memory timer loop scans job definitions, computes next-fire times, and dispatches directly to a fixed worker pool over HTTP when a time arrives.

graph LR
    Sched[Scheduler process<br/>in-memory timer loop] -->|poll every 1s| Jobs[(Job definitions)]
    Sched -->|HTTP dispatch| W1[Worker 1]
    Sched -->|HTTP dispatch| W2[Worker 2]
    Sched -->|HTTP dispatch| W3[Worker 3]

# V1 — single process, in-memory, no durability
import time, heapq

class Scheduler:
    def __init__(self, jobs, workers):
        self.jobs = jobs          # list of Job(cron_expr, command, next_run_at)
        self.workers = workers    # round-robin pool
        self.heap = [(j.next_run_at, j) for j in jobs]
        heapq.heapify(self.heap)

    def run_forever(self):
        while True:
            now = time.time()
            while self.heap and self.heap[0][0] <= now:
                _, job = heapq.heappop(self.heap)
                worker = self.workers.pick()
                worker.dispatch(job.command)              # fire and forget
                job.next_run_at = job.compute_next(now)
                heapq.heappush(self.heap, (job.next_run_at, job))
            time.sleep(1)

This handles the demo case fine — one process, predictable ticks, no coordination overhead. Do not add leader election or a ledger yet — find the actual failure mode first.

9. Identify the bottleneck¶

10. Version 2 — durable ledger, leader election, lease-based claiming¶

Three changes, each closing one specific hole from V1:

1. Durable run ledger. Before dispatching anything, the scheduler writes a job_runs row with state = scheduled. Dispatch reads from this table, not from the in-memory heap. On restart, the scheduler's first act is: "find every active job whose next_run_at has passed and has no corresponding job_runs row yet, and insert one." That's the replay — it turns "what should have fired" from a fact that lived only in RAM into a fact that lived in a table.
2. Leader election among multiple scheduler instances, so there is always exactly one process deciding triggers, and a crash triggers automatic failover instead of total silence. Use Raft-style leader election (see Raft and leader election in Distributed Fundamentals) — a term-based election with randomized timeouts backed by etcd/Consul in practice, not hand-rolled.
3. Worker claim via lease, replacing fire-and-forget dispatch. A worker claims a job_runs row (scheduled → claimed, sets claimed_by and lease_expires_at), must renew the lease periodically while running, and if it doesn't renew or complete in time, the run's lease simply expires and any other worker can reclaim it — exactly the lease mechanics in Distributed Fundamentals — Leases: nobody has to detect the crash, the clock bounds the damage.

graph TD
    Sched[Scheduler leader] -->|1: write scheduled row| Ledger[(job_runs ledger)]
    Sched -->|2: notify| Queue[Dispatch queue]
    W1[Worker] -->|3: claim + lease| Ledger
    W1 -->|4: heartbeat / renew lease| Ledger
    W1 -->|5: complete| Ledger
    Standby[Standby scheduler] -.->|watches leader lease| LeaderLock[(Leader election lock)]
    Sched -->|holds| LeaderLock

# V2 — leader writes durably before dispatching; worker claims via lease
def leader_tick(db, now):
    due = db.query("""
        SELECT job_id, next_run_at FROM jobs
        WHERE status = 'active' AND next_run_at <= %s
    """, now)
    for job_id, scheduled_for in due:
        db.execute("""
            INSERT INTO job_runs (run_id, job_id, scheduled_for, state, attempt)
            VALUES (%s, %s, %s, 'scheduled', 1)
            ON CONFLICT (job_id, scheduled_for, attempt) DO NOTHING
        """, gen_run_id(job_id, scheduled_for), job_id, scheduled_for)
        db.execute("UPDATE jobs SET next_run_at = %s WHERE job_id = %s",
                    compute_next(job_id, scheduled_for), job_id)

def worker_claim(db, worker_id, lease_s=300):
    row = db.query_one("""
        UPDATE job_runs SET state='claimed', claimed_by=%s,
               lease_expires_at = now() + interval '%s seconds'
        WHERE run_id = (
            SELECT run_id FROM job_runs
            WHERE state = 'scheduled' ORDER BY scheduled_for LIMIT 1 FOR UPDATE SKIP LOCKED
        )
        RETURNING *
    """, worker_id, lease_s)
    return row

def reclaim_expired_leases(db):
    # Runs on the leader on a timer. Any claimed/running run whose lease lapsed
    # goes back to 'scheduled' — no crash detection needed, just a clock check.
    db.execute("""
        UPDATE job_runs SET state = 'scheduled', claimed_by = NULL
        WHERE state IN ('claimed', 'running') AND lease_expires_at < now()
    """)

The ON CONFLICT DO NOTHING on (job_id, scheduled_for, attempt) is what makes replay idempotent at the ledger layer: a new leader recomputing "what should have fired" after failover will attempt the same insert the old leader already made, and it's a safe no-op.

11. Identify the next bottleneck¶

12. Version 3 — production architecture¶

graph TD
    subgraph Control["Scheduler control plane"]
        Leader[Scheduler leader<br/>writes ledger before dispatch]
        Standby1[Standby scheduler]
        Standby2[Standby scheduler]
        LE[(Leader election<br/>etcd/Consul, Raft)]
        Leader --- LE
        Standby1 -.watch.- LE
        Standby2 -.watch.- LE
    end

    Leader -->|scheduled rows| Ledger[(Job run ledger<br/>Postgres, sharded by job_id hash)]
    DAGRes[DAG dependency resolver] -->|releases downstream on upstream success| Ledger

    Ledger -->|claim via lease| WFleet[Worker fleet]
    WFleet -->|heartbeat + lease renew| Ledger
    WFleet -->|complete/fail| Ledger

    Ledger -->|failed, retries exhausted| DLQ[Dead-letter queue]
    Ledger -->|retrying, backoff| RetryQ[Retry scheduler]
    RetryQ -->|re-insert as new attempt| Ledger

    Ledger --> Mon[Monitoring: missed schedule,<br/>overrunning job, lease reclaim rate]
    Mon --> Alert[Alerting]

Key production decisions:

• Sharded ledger by job_id hash, not one Postgres primary. At 50K triggers/minute the ledger write path, not the timer logic, is the throughput ceiling — shard the same way the rate limiter shards its counter store, because this is again "many independent writers hitting one integer-shaped resource."
• DAG resolver as a separate consumer of the ledger, not baked into the leader's dispatch loop. It subscribes to succeeded/failed state transitions and only then inserts scheduled rows for downstream jobs in the same dag_run_id — this keeps "when does X fire" (the scheduler's job) separate from "is X allowed to fire yet" (the resolver's job), so DAG complexity doesn't bloat the leader's hot path.
• Retry with backoff is a re-insertion, not a mutation. A failed run's next attempt is a new job_runs row (attempt = attempt + 1) inserted after a backoff delay, not an in-place retry of the same row — this keeps the full attempt history intact for debugging and keeps the lease semantics simple (each attempt has its own lease).
• Dead-letter after exhausting retries, visible via the same status API, with an explicit alert — a job silently dead-lettering with nobody watching is the DAG-shaped version of V1's silent loss.
• Monitoring is first-class, not bolted on — see Section 16. A missed schedule or an overrunning job is exactly as bad as a wrong answer; both need paging, not a log line.

13. Failure analysis¶

14. Consistency considerations — what "exactly-once" actually means here¶

Be honest about this the way distributed-message-queue.md is honest about queue delivery: true exactly-once trigger-and-execute, end to end, across an arbitrary crash, is not achievable without idempotent jobs. What this design actually delivers is:

• At-least-once dispatch, durably. The ledger write happens before dispatch and is the atomic commit point (Section 13). A crash before that write means the trigger simply hasn't happened yet and gets replayed; a crash after means it's already durable and will be claimed. This guarantees the trigger is never silently lost — but a worker crash after claiming (before completing, before the platform can observe the outcome) can still produce a second full execution once the lease expires and another worker reclaims the row.
• Idempotent execution turns "at least once" into "effectively exactly once." The platform's contribution is giving every attempt a stable run_id so the job can dedupe its side effects (attempt-scoped idempotency keys for payments, INSERT ... ON CONFLICT for data writes, checking "have I already sent this notification for this run_id" before sending). The platform cannot force a job's business logic to be idempotent — that's a contract with the job author, spelled out in the clarifying questions (Section 2) precisely because it changes what's achievable.
• Read-your-writes on the status API: after a worker calls complete, the next GET /runs/{run_id} must reflect the terminal state — write to the ledger's primary shard and read from the same shard for status queries immediately after a state change; replicas can lag for historical/analytics queries.
• The DAG resolver's dependency check must be strongly consistent within one dag_run_id. Two downstream jobs racing to read "has my dependency succeeded yet" from a stale replica is the DAG-shaped version of a lost-update bug — route dependency checks to the primary, not a replica, the same way burn-after-read in pastebin.md needed an atomic check rather than a cached one.

15. Observability¶

Metrics:
  scheduler_ledger_write_latency_p99
  scheduler_leader_term (gauge; alerts on unexpected increments)
  jobs_missed_schedule_total{job_id}      -- scheduled_for passed, no job_runs row exists yet
  jobs_overrunning{job_id}                -- running longer than 2x historical p99 duration
  lease_reclaim_rate                      -- runs reclaimed after lease expiry, per minute
  dag_stalled_total{dag_id}               -- downstream nodes stuck in 'scheduled' beyond SLA
  dead_letter_rate{job_id}
  worker_claim_latency_p99                -- time between 'scheduled' and 'claimed'

Alerts:
  missed_schedule > 0 for any job tagged critical         -- page immediately, this is data loss
  overrunning_job exceeds 3x p99 duration                  -- likely stuck, investigate before it blocks downstream
  lease_reclaim_rate spike                                 -- worker fleet instability, not just one bad node
  leader_term increments > 2/hour                          -- election storm, check network/GC per raft.md
  dag_stalled > SLA                                        -- resolver bug or a silently-skipped dependency

Traces:
  span per ledger write, per claim, per lease renewal — attribute run_id, job_id, dag_run_id

"Missed schedule" and "overrunning job" are first-class, independently alertable conditions — not derived after the fact from a dashboard someone has to remember to check. A missed schedule is silent data loss; an overrunning job is a leading indicator of a stuck DAG blocking everything downstream of it.

16. Cost analysis¶

Ledger store (Postgres, sharded, 36 GB/day hot, 30-day retention ~1 TB):
  Primary + replica per shard, 4 shards:                    ~$1,600/mo
Leader election (etcd/Consul, 3-5 node cluster):             ~$300/mo
Worker fleet (10K workers, mostly reused compute
  from existing job workloads — marginal scheduler cost is
  the claim/heartbeat traffic, not the compute itself):      ~negligible incremental
Dead-letter / retry queue infra:                              ~$150/mo
Monitoring/alerting pipeline:                                 ~$200/mo
Total (excluding worker compute, which is workload-specific): ~$2,250/mo

Cost lever: cold-archive job_runs older than 30 days to object storage
  (same pattern as pastebin's lifecycle policy) — hot ledger stays small,
  cuts primary shard storage ~70%

17. Alternative architectures¶

18. Staff Engineer Extensions¶

19. Interview follow-ups¶

1. "Why is at-least-once dispatch plus idempotent execution the honest answer, instead of just saying exactly-once?" — Because the crash window between "decided to fire" and "confirmed execution completed" cannot be eliminated in an async system; the ledger write closes the trigger-loss half of that gap, but a worker dying mid-execution after claiming still requires either a job that tolerates re-execution or accepting occasional duplicate side effects. Naming both halves is what separates this from a queue-delivery-semantics trivia answer.
2. "How is the lease duration chosen, and what happens if it's wrong?" — Too short: healthy long-running jobs get reclaimed and duplicated because a worker under transient load misses a renewal. Too long: a genuinely crashed worker's job sits unclaimed for the full lease window before anyone else can pick it up, inflating missed-schedule-adjacent latency. Base it on the job's own p99 duration plus renewal margin, per-job if durations vary widely (a 2-second health check and a 20-minute batch job should not share a 5-minute lease).
3. "How would you support 'catch up' semantics for a job that missed several ticks during an outage?" — Depends on the clarifying answer in Section 2: catch-up-once means the replay logic inserts a single scheduled row for the latest missed tick and skips the rest (common for reports); catch-up-all means inserting one row per missed tick (common for time-series backfills) — the ledger's UNIQUE (job_id, scheduled_for, attempt) key naturally supports either, since scheduled_for values differ per missed tick.
4. "What's the difference between this and a message queue with delayed delivery?" — A queue's delayed-delivery feature gets you the timer, but not the DAG dependency resolution, not the "recompute next cron tick" logic, and not a run-history/status model keyed by scheduled time rather than message offset. You could build the ledger on top of a queue's storage layer, but the scheduling and dependency logic is a distinct layer above it — see distributed-message-queue.md for what the queue layer alone guarantees.

Self-Assessment¶

• [ ] I can explain why "exactly-once" is really "at-least-once dispatch + idempotent execution," and why the crash window makes true exactly-once unreachable without the job's cooperation
• [ ] I can walk through the ledger-write-before-dispatch mechanism and say precisely which side of that write a crash "never happened" vs. "will happen, possibly late"
• [ ] I can explain lease-based worker claiming and why it needs no explicit crash detection
• [ ] I can describe how split-brain during a partition is bounded by quorum election plus a unique constraint on the ledger, not by either mechanism alone
• [ ] I can explain why a DAG resolver must distinguish "succeeded" from "dead_letter" when deciding whether to release downstream jobs
• [ ] I can name the number that forces sharding the scheduler's own leader responsibility, not just the worker fleet, at 100x scale