Design a Distributed Job Scheduler

Internal teams need to run work on a schedule or on demand: nightly billing closes, hourly report builds, a cleanup every five minutes, a one-off email batch at 9 a.m. tomorrow. Today each team runs its own cron box, and when that box dies jobs silently stop or, after a messy failover, run twice.

Design a shared, multi-tenant scheduler. Users register jobs with a cron expression or a one-time timestamp, a priority, a timeout and a retry policy. The platform must start every run close to its scheduled time, survive machine failures, and give jobs as close to exactly-once execution as is honestly possible.

In 45 minutes, cover the data model, how due jobs are found without scanning everything, how runs are handed to workers, and what happens when a worker or a scheduler node dies mid-run.

Sketch on the HLD board, write your answer in the Design doc, and size it with the Estimator. Start the mock interview from the timer for a 45-minute run.

Design Path · Data & Infrastructure · problem 40 of 70

Ask before you design. Try answering each yourself, then compare.

1. How many jobs and runs?
   Typical answer

   About 10M registered jobs and 100M runs per day, with sharp spikes at the top of each hour when many cron jobs share a schedule.

2. How precise must start times be?
   Typical answer

   Runs should start within 5 seconds of their scheduled time at p99. Sub-second precision is not needed.

3. What does the platform execute?
   Typical answer

   A job is a container image or an HTTP callback with arguments. Runs last from seconds to a few hours.

4. What delivery guarantee do users get?
   Typical answer

   At-least-once execution with deduplicated triggering. We pass an idempotency key so handlers can make side effects exactly-once.

5. How do priorities work?
   Typical answer

   Three tiers. Under contention, high-priority runs are started first, and each tenant has a concurrency quota so one team cannot hog workers.

6. Do we need dependencies between jobs (DAGs)?
   Typical answer

   Not in the core design; mention it as a follow-up.

7. What about missed runs after an outage?
   Typical answer

   Per-job policy: run once to catch up, run every missed slot, or skip.

Requirements

Functional

• Create, update, pause and delete jobs with cron or one-time schedules, payload, priority, timeout and retry policy.
• Trigger runs at their scheduled time and allow manual triggers.
• Execute runs on a worker fleet, retrying failures with backoff and moving exhausted runs to a dead-letter list.
• Show run history, status and logs per job.

Non-functional

• 100M runs/day, peaks of ~30k run starts per second at the top of the hour.
• Start latency under 5 s p99 after the scheduled time.
• No scheduled run silently skipped; duplicate triggers for the same slot are suppressed.
• 99.99% scheduler availability with no single node whose loss stops scheduling.

Out of scope

• Workflow DAGs and data dependencies between jobs.
• Building the container runtime itself (assume a cluster manager exists).
• Billing per job.

Back-of-the-envelope estimates

• Runs: 100M/day ≈ 1,200 starts/s on average. Cron schedules cluster on round minutes and hours, so the top of the hour can see 25× that, ~30k/s, for a few seconds.
• Job store: 10M definitions × ~2 KB ≈ 20 GB, small enough for a sharded SQL database with a few shards.
• Run ledger: 100M runs/day × ~2 KB ≈ 200 GB/day; keeping 90 days of history is ~18 TB, ~54 TB replicated, so old runs move to cheaper storage.
• State updates: each run writes about 4 times (queued, running, heartbeats, finished): ~5k writes/s average, well over 100k/s at the hourly spike.
• Reads: status pages and APIs read ~5× more than runs start, mostly recent runs.

The hourly spike and the run-ledger write rate drive the design, not the size of job definitions: triggering must be spread over shards and ahead of time, and ledger writes must be sharded by job.

API

POST   /v1/jobs            {name, schedule:"0 * * * *" | runAt, target, payload,
                            priority, timeoutS, retry:{max, backoff}, catchUp}
POST   /v1/jobs/{id}:trigger  {idempotencyKey}        -> runId
GET    /v1/jobs/{id}/runs  ?status&cursor
// worker RPCs
Lease(workerId, pools, max)        -> [{runId, attempt, fencingToken, payload}]
Heartbeat(runId, fencingToken)     -> extendUntil | CANCELLED
Complete(runId, fencingToken, ok, output)

Run IDs are derived from the job ID and the scheduled slot, so any number of trigger attempts for the same slot map to one run row and an insert-if-absent removes duplicates.

Data model

jobs (job_id PK, tenant, schedule, next_run_at, priority, timeout, retry_policy, paused, version) in sharded SQL, partitioned by job_id, with an index on (shard, next_run_at) so each scheduler can read just its due jobs. Transactions make “insert run and advance next_run_at” atomic.

runs (run_id = job_id@slot PK, state, attempt, worker, lease_until, fencing_token, not_before, started_at, finished_at) in the same shard as its job, so trigger and run creation are one local transaction. Old runs are archived to object storage.

Ready queues: per priority and per worker pool, holding run IDs only; the run row stays the source of truth. Shard leases live in a small consensus store.

High-level design

Registering. Owners call the Job API, which validates the cron expression, checks tenant quotas, computes the first next_run_at and writes the job row to its shard in the job store.

Triggering. The job table is split into a few thousand logical shards. Scheduler nodes take leases on shards from the coordination service, and each lease comes with an increasing fencing token. Every second, a scheduler queries its shards for jobs with next_run_at up to a short lookahead (say 30 seconds). For each due job it runs one transaction: insert the run row with ID job@slot if absent, and advance next_run_at to the following slot. It then pushes the run ID onto the ready queue for its priority, with a not-before time equal to the slot.

Executing. Executors lease runs from the queues, highest priority first and within tenant concurrency limits. Each lease sets lease_until and a fencing token on the run row; the executor heartbeats to extend it while the job runs, and reports success or failure with the token. Failures are rescheduled with exponential backoff; runs that exhaust retries go to a dead-letter queue and alert the owner. A monitor scans for runs whose lease expired or that never started and requeues or flags them. The API reads the run ledger for history and status.

Open the HLD board and press Reference to see the architecture diagram.

Deep dives

Finding due jobs cheaply

Store next_run_at on every job and index it per shard. A scheduler that owns a shard runs a range query for the next 30 seconds every second and caches the results in an in-memory timer wheel, firing each at its slot. Looking ahead turns the hourly spike into a gentle ramp: runs for 01:00:00 are created and queued during the preceding half minute and only released at the exact second. Sharding by job ID spreads one popular cron expression evenly across schedulers, and optional per-job jitter of a few seconds smooths it further when exact timing is not needed.

Exactly-once-ish: dedupe triggers, fence executions

Two things can duplicate work. First, two schedulers may both trigger a slot during a lease handover; deriving the run ID from job and slot and inserting with a uniqueness constraint makes the second trigger a no-op. Second, a slow worker may lose its lease while still running; the run is retried elsewhere and the old worker later reports. Every lease increments a fencing token stored on the run, and completions with an older token are rejected. Truly exactly-once side effects still need the handler's cooperation: we pass run_id as an idempotency key for its writes.

Leases, heartbeats and timeouts

Executors hold a lease of, say, 30 seconds and heartbeat every 10. Missing three heartbeats means the run is presumed lost and is requeued with its attempt count increased. Lease length trades detection speed against false requeues during long garbage-collection pauses or network blips. The job's own timeout is separate: when exceeded, the executor kills the process and reports failure. On the scheduler side, shard leases work the same way, and a scheduler that loses its lease must stop triggering immediately; the fencing token on its writes enforces that even if it does not notice.

Priorities and fairness

Keep separate ready queues per priority and let executors drain them with weighted selection, for example 70/20/10, so low priority still progresses instead of starving. Within a priority, enforce per-tenant concurrency limits at lease time: if a tenant already has its quota running, skip its runs and pick the next. Keep long batch jobs in a separate worker pool so a backlog of hour-long runs never delays a 5-second job that must start on time. Admission control rejects new one-off triggers when a tenant's backlog exceeds its limit.

Failure modes

• If a scheduler node dies, its shard leases expire within seconds and other nodes take them over; the lookahead overlap plus deterministic run IDs mean no slot is missed or doubled.
• If an executor dies mid-run, heartbeats stop, the lease expires and the run is retried with a new fencing token; the dead worker's late result is rejected.
• If the job store shard is unavailable, scheduling for that shard pauses and resumes from next_run_at afterwards, applying each job's catch-up policy.
• If the ready queue loses messages, the monitor finds runs stuck in 'queued' past their slot and re-enqueues them, because the run ledger is the source of truth.
• If a bad job fails every time, retries back off exponentially and then dead-letter it so it cannot consume the worker fleet.

Core algorithm in code

import heapq


class Scheduler:
    """Runs with leases, retries with backoff, priorities and idempotent run IDs.
    Time is simulated so the demo is deterministic."""

    def __init__(self, lease_s=30, max_attempts=3):
        self.lease_s, self.max_attempts = lease_s, max_attempts
        self.runs = {}            # run_id -> record (the run ledger)
        self.ready = []           # heap of (not_before, priority, run_id)
        self.fence = 0            # increases on every lease grant
        self.dead = []

    def schedule(self, job_id, slot, priority=5):
        run_id = f"{job_id}@{slot}"                    # same job + slot -> same run
        if run_id in self.runs:
            return run_id, False                       # duplicate trigger ignored
        self.runs[run_id] = {"state": "queued", "attempt": 0, "lease_until": 0, "token": 0}
        heapq.heappush(self.ready, (slot, priority, run_id))
        return run_id, True

    def lease(self, worker, now):
        while self.ready and self.ready[0][0] <= now:
            _, prio, run_id = heapq.heappop(self.ready)
            r = self.runs[run_id]
            if r["state"] in ("done", "dead"):
                continue
            self.fence += 1
            r.update(state="running", worker=worker, lease_until=now + self.lease_s,
                     token=self.fence, attempt=r["attempt"] + 1, prio=prio)
            return run_id, self.fence
        return None, None

    def complete(self, run_id, token, ok, now):
        r = self.runs[run_id]
        if token != r["token"]:
            return "rejected: stale fencing token"     # a zombie worker's late result
        if ok:
            r["state"] = "done"
            return "done"
        return self._retry(run_id, now)

    def _retry(self, run_id, now):
        r = self.runs[run_id]
        if r["attempt"] >= self.max_attempts:
            r["state"] = "dead"
            self.dead.append(run_id)
            return "dead-lettered"
        backoff = 10 * 2 ** (r["attempt"] - 1)         # 10 s, 20 s, 40 s ...
        r["state"] = "queued"
        heapq.heappush(self.ready, (now + backoff, r["prio"], run_id))
        return f"retry in {backoff}s"

    def reap_expired(self, now):
        """Requeue runs whose worker stopped heartbeating."""
        for run_id, r in self.runs.items():
            if r["state"] == "running" and r["lease_until"] <= now:
                print(f"  t={now}: lease on {run_id} expired ({r['worker']} silent) ->", self._retry(run_id, now))


s = Scheduler()
print(s.schedule("nightly-report", slot=0, priority=5))
print(s.schedule("billing-close", slot=0, priority=1))
print(s.schedule("nightly-report", slot=0, priority=5), "<- second scheduler fired the same slot")

run, tok = s.lease("w1", now=0)
print(f"t=0: w1 leased {run} (priority first), token {tok}")
print("  fails ->", s.complete(run, tok, ok=False, now=5))

run2, tok2 = s.lease("w2", now=5)
print(f"t=5: w2 leased {run2}, token {tok2}, then goes silent")

run, tok = s.lease("w1", now=15)
print(f"t=15: w1 leased {run} attempt {s.runs[run]['attempt']} ->", s.complete(run, tok, ok=True, now=16))

s.reap_expired(now=40)
run3, tok3 = s.lease("w3", now=50)
print(f"t=50: w3 leased {run3}, token {tok3}")
print("  late result from w2:", s.complete(run2, tok2, ok=True, now=51))
print("  result from w3:     ", s.complete(run3, tok3, ok=True, now=52))

s.schedule("flaky-sync", slot=100, priority=3)
now = 100
while True:
    run, tok = s.lease("w1", now=now)
    if run:
        print(f"t={now}: {run} attempt {s.runs[run]['attempt']} fails ->", s.complete(run, tok, ok=False, now=now))
    if s.dead:
        break
    now += 10
print("states:", {k: v["state"] for k, v in s.runs.items()})

Trade-offs

• A SQL job store gives transactions for 'create run + advance next_run_at' but needs careful sharding; a key-value store scales writes more easily but makes that atomic step harder.
• Short leases detect dead workers quickly but cause duplicate executions during pauses; long leases avoid duplicates but delay recovery.
• A lookahead window smooths spikes and hides scheduler latency but means pausing or editing a job close to its slot may not take effect for that run.
• Weighted priority queues prevent starvation but let some low-priority work run while high-priority work waits; strict priority is simpler but can starve.
• Catching up every missed slot after an outage is faithful but can flood workers; running only the latest slot is safer for most jobs.

What a strong answer shows

• Uses an indexed next_run_at per shard and a lookahead instead of scanning all jobs.
• Makes triggering idempotent with run IDs derived from job and slot.
• Uses leases, heartbeats and fencing tokens to handle dead and zombie workers.
• Is precise that execution is at-least-once and explains how handlers achieve effectively-once with idempotency keys.
• Handles the top-of-the-hour spike, priorities and tenant fairness.
• Treats the run ledger as the source of truth and the queue as a disposable index.

Mistakes to avoid

• A single leader scheduler with no plan for failover or for its throughput ceiling.
• Claiming exactly-once execution without fencing tokens or idempotent handlers.
• Polling the whole job table every second.
• Letting a lost queue message mean a lost run because the queue is the only record.
• Ignoring that most cron jobs fire on the same round minutes.

Follow-ups to expect

• How would you add job dependencies? (Store a DAG; a run becomes ready when all upstream runs for the same logical date succeed.)
• How would you support time zones and daylight saving? (Store the zone with the schedule, compute next_run_at in that zone, define behaviour for skipped and repeated hours.)
• How would you let users cancel a running job? (Mark the run cancelled; the next heartbeat response tells the executor to stop.)
• How would you run the scheduler across regions? (Shard ownership per region with jobs pinned to a home region; fail over shards on region loss.)

Practice next