Design a Distributed Job Scheduler: Cron at Scale, Leases and At-Least-Once Runs

Listen

Listen. Teams submit jobs that run once at a set time, or on a cron schedule in a given time zone. A job is a container image or an HTTP callback with arguments. Each job has a retry policy, a timeout and a priority. Teams can pause, resume and cancel, and see run history and logs. Multi-step workflows with dependencies are out of scope.

Qualities. A job should start within about two seconds of its due time, and a scheduled tick must never be silently skipped. Exactly-once execution is impossible when machines can die mid-run, so promise at least once and ask teams to make jobs idempotent. Survive any crash, and be fair, so one team's ten thousand jobs cannot starve everyone else.

Ask scale

Scale. Ten million job definitions producing a hundred million runs a day is only about a thousand runs a second on average. But people love round times. A large share of cron jobs say midnight or the top of the hour, so a hundred thousand runs can be due in the same second. Run records are about fifty gigabytes a day; keep ninety days.

10M job definitions
100M runs/day
  100M / 10⁵ s ≈ 1,000 runs/s

but cron loves round times:
  if 0.1% of a day's runs are
    due at exactly 00:00:00
  → 100,000 in one second

run record ≈ 500 B
  100M × 500 B = 50 GB/day

Define API

The API. Creating a job takes either a cron schedule with a time zone or a single run time, the target, arguments, timeout, retry limit and priority, plus an idempotency key so a retried create does not make two jobs. It returns the job id and the next run time. Teams can list runs, pause, resume, cancel a run and delete the job.

POST /v1/jobs
  { name, schedule: "0 2 * * *",
    tz, run_at?, target, args,
    timeout_s, max_retries,
    priority, idempotency_key }
  → 201 { job_id, next_run_at }
GET  /v1/jobs/{id}/runs?cursor=
POST /v1/jobs/{id}:pause|resume
POST /v1/runs/{id}:cancel
DELETE /v1/jobs/{id}

Data model

Data model. Use a relational store, sharded later, because we need transactions and an index on time. Jobs carry their next run time, indexed, plus a bucket number from a hash of the job id. Each run is its own row with a unique constraint on job id and due time, so one tick can never create two runs. Runs track state, attempt, worker, lease and a fencing token.

jobs
  job_id PK, owner, schedule, tz
  next_run_at  (INDEX)
  bucket = hash(job_id) % 1024

runs
  run_id PK, job_id, due_at
  UNIQUE (job_id, due_at)
  state, attempt, worker_id
  lease_until, fence_token
  started_at, finished_at

High-level design

The high-level design. The API writes the job with its next run time. Scheduler instances keep asking the database what is due, create a run row for each, advance the job's next run time, and push the run onto a queue. Workers take runs from the queue, execute them, and report the result back to the runs table.

Why split it this way? Deciding when is cheap and must be precise. Doing the work is slow and unpredictable, from one second to one hour. A queue between them absorbs bursts and orders work by priority, and workers scale on queue depth. Logs stream to object storage, with a link on the run row.

Deep dive: due jobs

Deep dive one: finding due jobs. Each scheduler polls about once a second for jobs whose next run time is within the next five seconds, using the index. FOR UPDATE SKIP LOCKED lets several schedulers work at once without blocking or grabbing the same rows. In one transaction per job, insert the run and compute the next run time from the cron expression.

To scale out, split jobs into a thousand and twenty-four buckets and have each scheduler own a share through a lease in a coordinator like etcd, renewed every few seconds. If a scheduler dies, its lease expires and the others pick up its buckets within about ten seconds. If two ever overlap, the unique run key turns the second insert into a harmless conflict.

every ~1 s, per owned bucket:
SELECT job_id FROM jobs
 WHERE bucket IN (:mine)
   AND next_run_at <= now() + 5s
 ORDER BY next_run_at LIMIT 500
 FOR UPDATE SKIP LOCKED

per job, one transaction:
  INSERT run (job_id, due_at)
  UPDATE next_run_at = next(cron)

Deep dive: execution

Deep dive two: execution. When a worker takes a run, it marks it running with a lease thirty seconds out and a new fencing token. It sends a heartbeat every ten seconds to extend the lease. If the worker dies, heartbeats stop, the lease expires, and a reaper puts the run back on the queue with the attempt count increased.

The tricky case is a zombie. Worker A pauses for a minute, maybe a long garbage collection. Its lease expires, worker B takes the run, then A wakes up and writes its result. Fencing tokens fix this: each claim gets a higher number, and the database rejects writes with an older one, as in Concepts episode 31. Side effects can still repeat, which is why jobs must be idempotent.

Failures retry with exponential backoff and jitter: ten seconds, twenty, forty, and so on, until the retry limit. Then the run is marked dead and the owner is alerted. A run that exceeds its timeout is killed and counts as a failure. And decide what happens to ticks missed during a pause or an outage: skip them, or run once to catch up.

on failure:
  attempt < max_retries →
    due_at = now + 2^attempt × 10s
    (+ jitter)
  else → state = dead, alert

timeout_s reached →
  kill container, count as fail

cron missed while paused:
  policy = skip | run once

Deep dive: midnight

Deep dive three: midnight. A hundred thousand runs due at once overwhelms the schedulers and the workers. Look ahead and enqueue runs a few seconds early into a delay queue that releases them on time. Encourage jitter: hourly can mean a random but stable minute per job. Use priority queues, and cap concurrent runs per team so one team cannot take every worker.

Trade-offs

Failure modes. A dead scheduler means its buckets run about ten seconds late, but nothing is skipped because due jobs stay due in the database. A dead worker means a retry. If the queue loses a message, a reaper finds runs stuck in queued and re-enqueues them. If the database primary fails, schedulers pause during failover, then catch up.

Monitor start delay, from due time to actual start, at p ninety-nine, the number of runs that are due but not yet queued, queue depth per priority, and failure, retry and dead counts per team. At ten times the load, shard the jobs table by bucket, add schedulers, and autoscale workers on queue depth.

Say it like this

In the interview: jobs are indexed by next run time and hashed into buckets. Schedulers own buckets through leases and claim due rows with skip locked, and a unique key on job and due time guarantees one run per tick. Workers hold leases with heartbeats and fencing tokens. Failures retry with backoff, and look-ahead, jitter and quotas tame the midnight rush.