Design a distributed job scheduler

In one minute

• Separate when a job should run from who runs it. A scheduler tier turns "due at time T" into "ready now" by pushing into a work queue; a worker tier pulls ready work. They scale and fail independently.
• Find due jobs with a time-ordered index partitioned across scheduler shards, each holding the next few minutes in a min-heap. Never scan the whole job table.
• Execution is at-least-once with leases. A worker holds a lease with a deadline and heartbeats; an expired lease makes the job runnable again. Duplicate runs are expected, so jobs carry an idempotency key and a fencing token.
• Cron jobs materialise one future occurrence at a time: when an occurrence finishes, compute the next fire time and insert it.
• Plan for the top-of-the-hour herd and for tenant fairness, because users schedule work at round numbers.

Requirements

Functional:

• Submit a one-off job to run at a time, or after a delay; submit a recurring job with a cron expression and time zone.
• Run the job's payload (an HTTP callback or a message to a named worker pool) close to its due time.
• Retry failures with exponential backoff up to a limit, then move the job to a dead-letter state.
• Cancel, pause, and inspect jobs and their execution history.

Non-functional:

• Never lose an accepted job. Durability beats latency.
• Start a job within 2 seconds of its due time at p99 under normal load.
• At-least-once execution; no two workers believe they own the same attempt at once.
• Multi-tenant isolation: one tenant's burst must not delay everyone else.

Out of scope: DAG dependencies between jobs (that is a workflow engine like Airflow or Temporal), running user code in sandboxes (we dispatch to the tenant's own workers or endpoints), and sub-second precision.

Back-of-envelope

Planning assumptions for a shared internal platform at a large company.

• Active jobs: 20M, of which 2M are recurring. Executions per day: 60M (one-off jobs plus recurring jobs averaging about 20 runs per day).
• Average execution rate: 60M / 86,400 ≈ 700 per second.
• Peak: cron schedules cluster on round minutes. If 10% of a day's recurring runs land on the 24 top-of-the-hour seconds, that is 0.1 × 40M / 24 ≈ 170,000 jobs due in the same second. Even if the real figure is a few times smaller, the scheduler must dispatch tens of thousands per second for a few seconds and spread the rest.
• Storage: a job definition is about 1 KB (payload reference, schedule, retry policy, metadata), so 20M × 1 KB = 20 GB. An execution record is about 300 B, so 60M × 300 B = 18 GB per day, or 18 GB × 365 × 5 ≈ 33 TB over five years. Keep 30 days hot (about 540 GB) and archive the rest to object storage.
• Bandwidth: dispatch messages carry a payload reference of about 1 KB, so 700/s × 1 KB ≈ 0.7 MB/s on average and about 170 MB/s at a top-of-hour spike, briefly.
• Scheduler memory: holding the next 5 minutes of due jobs at peak density, perhaps 2M entries, at about 64 B per heap entry is 128 MB spread over the scheduler shards.

API

POST /v1/jobs                      Idempotency-Key: <uuid>
  {name, target:{type:"http"|"queue", ref}, payload,
   schedule:{runAt:"2026-10-16T09:00:00Z"} | {cron:"0 9 * * MON-FRI", tz:"Europe/London"},
   retry:{maxAttempts:5, backoffSeconds:10}, timeoutSeconds:300}
  -> 201 {jobId, nextRunAt}

DELETE /v1/jobs/{jobId}            -> 204
POST   /v1/jobs/{jobId}:pause      -> 200
GET    /v1/jobs/{jobId}/executions?cursor=...
  -> [{executionId, scheduledFor, attempt, state, startedAt, finishedAt, error}]

// worker protocol (gRPC)
Lease(pool, workerId, max, leaseSeconds) -> [{executionId, jobId, attempt, payload, leaseUntil}]
Heartbeat(executionId, attempt)          -> {leaseUntil} | LEASE_LOST
Complete(executionId, attempt, result)   -> OK | LEASE_LOST
Fail(executionId, attempt, error)        -> {retryAt} | DEAD

The attempt number is the fencing token. Every call from a worker includes it, and the server rejects calls whose attempt is no longer current.

Data model

• jobs: job_id PK, tenant_id, schedule, target, payload_ref, retry_policy, state, next_run_at, version. Partition key job_id (hash). Large payloads go to object storage; the row holds a reference.
• due_index: (shard_id, due_minute, next_run_at, job_id) as the primary key. shard_id = hash(job_id) mod 256; due_minute buckets time so one scheduler shard reads "my jobs due in the next minute" as a single contiguous range scan. This index is what makes finding due work cheap. Rows are deleted when the job is dispatched.
• executions: execution_id PK, job_id, scheduled_for, attempt, state, lease_owner, lease_until, started_at, finished_at, error. Unique on (job_id, scheduled_for) so an occurrence is dispatched once, however many times the scheduler retries. Partition by job_id, with time-based table partitions so 30-day retention is a cheap partition drop.
• shard_leases (in etcd or ZooKeeper): which scheduler instance owns which of the 256 shards, with a lease and a revision number used as a fencing token.

A sharded relational store (Postgres, MySQL or a distributed SQL database) works well: the access patterns are point lookups and short range scans, and unique constraints do the deduplication. Cassandra also fits due_index well, but would need lightweight transactions for the dedup, which are slow.

High-level design

flowchart LR
  CL["Clients"] --> API["Job API"]
  API --> DB[("Jobs + due_index")]
  ETCD["etcd shard leases"] --> SCH["Scheduler shards"]
  SCH -- "scan next minute" --> DB
  SCH --> HEAP["In-memory min-heap"]
  HEAP -- "due now" --> RQ["Ready queues per pool"]
  RQ --> W["Workers"]
  W -- "lease, heartbeat, complete" --> EX["Execution service"]
  EX --> DB
  EX -- "retry or next cron run" --> DB
  EX --> DLQ["Dead letters"]

Write path. A client submits a job. The API validates the cron expression, computes the first next_run_at in the job's time zone, and in one transaction inserts the jobs row and its due_index row. If the due time is within the current minute, it also notifies the owning scheduler shard directly so short delays are not missed.

Dispatch path. Each scheduler instance owns a set of shards through etcd leases. For each shard, a loader runs every few seconds and reads due_index rows for the current and next minute into a min-heap keyed by next_run_at. A dispatcher loop peeks at the heap top, sleeps until it is due, then in one transaction inserts the executions row (the unique constraint deduplicates) and deletes the due_index row, and publishes the execution to the ready queue for the job's worker pool. Workers lease executions from the queue, heartbeat while running, and call complete or fail. Complete on a recurring job computes the next fire time and inserts a new due_index row; fail either inserts a retry row at now + backoff × 2^(attempt-1) or marks the execution dead.

Read path. Job and execution history reads go to replicas, paginated by time. Operators get per-tenant dashboards of lag (now minus scheduled_for at dispatch), failure rate and queue depth.

Deep dives

Finding due jobs

Options considered. First, poll the database: SELECT ... WHERE next_run_at <= now() ORDER BY next_run_at LIMIT 100 FOR UPDATE SKIP LOCKED from many dispatchers. It is simple and correct, and SKIP LOCKED lets dispatchers share a table without blocking each other; Postgres documents it as unsuitable for general queries, since skipped rows give an inconsistent view, but meant for exactly this case of many consumers on a queue-like table (Postgres docs). It works well up to perhaps a few thousand dispatches per second, then the index's hot head and constant polling dominate. Second, a Redis sorted set scored by due time with ZRANGEBYSCORE and a Lua script to claim. It is fast but holds durable state in an asynchronously replicated store. Third, a hierarchical timing wheel in memory, as Kafka uses for its delayed operations: Kafka replaced a DelayQueue with O(log n) insert and delete by timing wheels with O(1) cancellation, and in its benchmark the saturation point rose from about 25,000 to about 105,000 requests per second (Confluent). It is very cheap per timer, but has to be rebuilt from durable state after a crash.

Choice: durable due_index partitioned by shard and minute, plus an in-memory min-heap per shard holding only the next one to two minutes. The database does what it is good at (durable, range-scannable storage) and memory does what it is good at (precise ordering of a small, hot set). If a scheduler dies, another acquires its shards' leases and reloads from due_index; anything already dispatched is protected by the unique (job_id, scheduled_for) constraint. Dropbox's Async Task Framework is a production version of the same split: tasks sit in a store with a next-trigger time, a consumer polls for due ones every two seconds and pushes them onto per-callback SQS queues, and the target was 95% of tasks starting within five seconds of their scheduled time (Dropbox).

Leases, retries and duplicate execution

Exactly-once execution of side effects is not achievable when workers can stall: a worker pauses for a 40-second GC, its lease expires, the job is re-leased to another worker, and then the first worker wakes and finishes. So the contract is at-least-once, and the design makes duplicates harmless. This is the fencing-token argument: the lease alone cannot stop a paused process, so the resource it writes to must reject stale tokens (Kleppmann). Each attempt gets an incrementing attempt number. Re-leasing bumps it, and complete or heartbeat with an old attempt returns LEASE_LOST, so the stale worker's result is discarded. Downstream side effects get an idempotency key of job_id + scheduled_for, so a duplicate HTTP callback can be recognised by the receiver. Lease length is a trade-off: short leases recover crashed work quickly but force long jobs to heartbeat often; we default to 30 seconds with heartbeats every 10.

Retries use exponential backoff with jitter, capped at the job's maxAttempts, after which the execution moves to dead letters with its last error. A poison job that crashes its worker every time shows up as repeated lease expiries; counting those as attempts keeps it from looping forever.

Cron, time zones and the top-of-the-hour herd

Only the next occurrence of a recurring job exists in due_index. On completion (or on dispatch, for jobs that allow overlap) we compute the next fire time from the cron expression in the job's IANA time zone. Overlap is a per-job policy worth naming, as Kubernetes does with concurrencyPolicy: Allow (the default), Forbid (skip the new run while the old one is still going) or Replace (stop the old one); Kubernetes also only made .spec.timeZone stable in v1.27 and warns that a CronJob may create two Jobs or none for one slot, so jobs must be idempotent (Kubernetes docs). Daylight saving needs explicit rules: a 02:30 job on a spring-forward day has no 02:30, so we run it at the next valid instant; on fall-back day 01:30 happens twice, and we run it once. If the scheduler was down and several occurrences were missed, the default policy is to run once and skip the rest, with an opt-in "catch up" policy. Google's cron service leans the same way: when it cannot tell whether a launch happened, it prefers skipping a run to risking a double launch (Google SRE book, ch. 24). Kubernetes caps catch-up outright: a CronJob that has missed more than 100 schedules logs an error and does not start the Job.

For the herd, we offer and encourage a jitter window on cron jobs ("run within 5 minutes after 09:00"), hash the job ID into that window (Google added a ? to its crontab syntax that does this hashing, and EventBridge Scheduler exposes it as a flexible time window (AWS docs)), and enforce per-tenant dispatch quotas in the ready queues so one tenant with 500,000 jobs at midnight does not starve others. Ready queues are per worker pool, so a slow pool backs up only its own queue.

Down the stack

Failure modes

• 10x load. Dispatch lag grows. Add scheduler shards (256 logical shards can move between more instances without re-hashing) and scale worker pools on queue depth. If the database insert rate is the bottleneck, batch dispatch inserts per shard per tick.
• Scheduler instance dies. Its etcd leases expire in about 10 seconds; another instance takes the shards, reloads the next minutes from due_index, and dispatches anything overdue. The unique execution key prevents double dispatch for jobs the old owner already handled.
• Region outage. Run active-passive with the job store replicated to a second region. On failover, jobs due during the gap dispatch late but are not lost, subject to the replication lag at the moment of failure (seconds with asynchronous replication). Critical tenants can opt into synchronous cross-region commit at higher submit latency.
• Hot key or hot tenant. A tenant creates a million jobs due at midnight. Per-tenant quotas in ready queues and a submit-time warning when a tenant's due-time histogram is spiky contain it. A single job hammered by updates is rare; version checks keep updates safe.
• Downstream outage. A tenant's HTTP endpoint is down, so all its jobs fail and retry. A per-target circuit breaker pauses dispatch to that target instead of burning attempts and filling dead letters.

In the wild

• Slack, about 2016: the job queue that could not drain. Database contention slowed job execution until Slack's Redis-backed job queue hit its memory limit. New jobs were refused, and because dequeuing also needed spare memory to move a job onto an in-flight list, the queue stayed stuck after the database recovered. At the time Slack ran over 1.4 billion jobs on busy days, peaking at 33,000 per second, and the fix put Kafka in front of Redis as a durable buffer (Slack Engineering).
• GitHub, 2022: retrying jobs that can never succeed. Automated accounts created and deleted repositories so fast that many webhook jobs could not build their payloads, because the data was already gone. Workers kept retrying them, which blocked new events for 31 minutes. GitHub changed the workers to stop retrying jobs whose data no longer exists, the same "is this failure retryable?" question a dead-letter policy has to answer (GitHub availability report).
• Datadog, 2023: one schedule, every region at once. A legacy OS update channel applied a systemd security update in its default 06:00 to 07:00 UTC window. Restarting systemd-networkd deleted the routes Cilium manages, and tens of thousands of nodes in five otherwise isolated regions dropped off the network within an hour; recovery took about two days. A fixed wall-clock window with no per-region spread is exactly the herd this lesson's jitter window prevents (Datadog).

Trade-offs

Leadership lens

• Build first: the execution state machine (leased, succeeded, failed, dead) with attempt fencing, on top of a single Postgres table with SKIP LOCKED. That serves the first year for most companies; introduce the sharded heap when measured dispatch lag says so.
• Rollout: migrate existing crontabs tenant by tenant, running old and new side by side with the new one in dry-run mode, comparing fire times, then flipping.
• Ownership: the platform team owns scheduling and dispatch; tenant teams own idempotency of their handlers. Write that split into the onboarding doc, because duplicates will happen.
• Cost drivers: execution history storage (archive aggressively), the database write rate at peaks, and idle worker capacity reserved for midnight spikes.
• Design review question: what does a job owner see when their job runs twice, and how do they find out?

Check yourself

1. A worker pauses for a 40-second GC, its lease expires, and the job is re-leased to another worker. What stops the first worker's late result from being recorded, and why can't the design promise exactly-once?

   Show answer

   Re-leasing bumps the attempt number, which acts as a fencing token, so the stale worker's complete or heartbeat returns LEASE_LOST and its result is discarded. Its code may already have produced side effects, though, so the contract is at-least-once and handlers dedupe with the job_id + scheduled_for idempotency key.

2. Why does the design pair a durable due_index with a small in-memory min-heap per shard instead of using a Redis sorted set or polling the jobs table with SKIP LOCKED?

   Show answer

   Polling hits a ceiling of a few thousand dispatches per second as the hot index head and constant polling dominate, and Redis holds durable state in an asynchronously replicated store. The due index keeps truth durable and range-scannable by shard and minute, while the heap orders only the next minute or two and is rebuilt from the index after failover.

3. Using the lesson's numbers, roughly how many jobs could be due in the same second at the top of the hour, and what does the design do about it?

   Show answer

   If 10% of the 40M daily recurring runs land on the 24 top-of-the-hour seconds, about 0.1 x 40M / 24, or roughly 170,000 jobs are due in one second. The design encourages a cron jitter window with the job ID hashed into it and enforces per-tenant dispatch quotas in per-pool ready queues, turning spikes into plateaus.

4. A scheduler instance dies after dispatching some jobs. How does another instance take over without double-dispatching them?

   Show answer

   Its etcd shard leases expire in about 10 seconds, another instance acquires the shards and reloads upcoming minutes from due_index. Dispatch inserts the executions row under a unique (job_id, scheduled_for) constraint, so any occurrence the old owner already dispatched cannot be inserted again.

Sources and further study

Primary

• Varghese and Lauck, "Hashed and Hierarchical Timing Wheels" (SOSP 1987): the O(1) start and stop timer structure behind Kafka's purgatory and many kernel timers.
• Kleppmann, "How to do distributed locking": why leases need fencing tokens checked by the resource, with the GC-pause example.

Systems docs

• Kubernetes CronJob: concurrency policies, starting deadlines, time zones and the idempotency warning.
• etcd API: leases: LeaseGrant, LeaseKeepAlive and key deletion on expiry.
• Amazon EventBridge Scheduler: a managed one-time and cron scheduler, the service AWS points to beyond SQS's 15-minute delay.

Books

• Designing Data-Intensive Applications (1st ed.), Martin Kleppmann, Ch 8 "The Trouble with Distributed Systems": process pauses, unreliable clocks, leases and fencing tokens.
• Designing Data-Intensive Applications (1st ed.), Martin Kleppmann, Ch 11 "Stream Processing": message brokers, acknowledgments and redelivery.

Other curricula

• Azure Architecture Center: Scheduler Agent Supervisor pattern: frames scheduling as a durable state store with complete-by deadlines, a periodic supervisor that recovers expired steps, and idempotent agents.
• donnemartin/system-design-primer: Asynchronism: treats task queues as background workers with back pressure, returning 503 when the queue is full.

Watch

• Distributed Job Scheduler Design Deep Dive with Google SWE! (Jordan has no life): requirements, estimates and a detailed design for the same question.
• Job Scheduler: System Design Interview with a senior FAANG Engineer (interviewing.io): a realistic mock interview with interviewer probes.

Free resources

• Distributed periodic scheduling with Cron (Google SRE book): how Google runs cron at scale, including Paxos-backed leaders and handling missed runs.
• Asynchronous task scheduling at Dropbox (Dropbox Tech): a production delayed-task system with at-least-once semantics and lease handling.
• Scaling Slack's job queue (Slack Engineering): moving a Redis-backed job queue behind Kafka to survive spikes.
• Avoiding insurmountable queue backlogs (AWS Builders' Library): tenant fairness, backlog recovery and why queues need shedding.
• SELECT and the locking clause (PostgreSQL docs): the semantics of FOR UPDATE SKIP LOCKED for database-backed queues.