Design a Distributed Job Scheduler

1Clarifying the requirements#

Pin down what "job scheduler" means here before drawing a box, because the answer changes the whole design.

• A cron engine and a "charge this card in 7 days" engine look identical on a napkin.
• They diverge hard the moment durability enters the picture.
• So I spend the first two minutes narrowing scope, not sketching.

Candidate: When you say scheduler, do you mean cron-style recurring jobs, one-shot delayed tasks like "charge this card in 7 days," or both?

Interviewer: Both. Users submit one-time jobs with a run-at timestamp and recurring jobs with a cron expression.

Candidate: What does a job actually do when it fires? Are we executing arbitrary user code inside our fleet, or are we just emitting a trigger to some downstream worker?

Interviewer: Assume the scheduler's responsibility ends at reliably invoking a handler: an HTTP callback or dropping a message on a queue. The user's business logic runs in their own workers.

Candidate: Good, that keeps us out of the sandboxing business. Execution guarantee: is at-least-once acceptable, or do we need exactly-once?

Interviewer: At-least-once is the floor. We want strong tools to make duplicate runs safe, but you cannot promise exactly-once over a network. Good.

Candidate: Scale and timing precision? Are we okay with a job firing a few seconds late, or is this sub-second?

Interviewer: Hundreds of millions of scheduled jobs. Second-level precision is fine for most. Flag anything that needs tighter.

What those answers just told me:

• "Handler invocation only" keeps me out of sandboxing and makes the queue the real product.
• "At-least-once is the floor" means the whole game is making duplicates safe, not avoiding them.
• "Second-level precision" plus "hundreds of millions of jobs" is the interviewer quietly handing me the thundering herd to design around.

Functional requirements

• Submit a one-time job with a run-at time, or a recurring job with a cron schedule.
• Cancel or update a job before it fires.
• Execute by calling an HTTP endpoint or enqueuing a message, with configurable retries and backoff.
• At-least-once delivery, with an idempotency key passed to the handler so the user can dedupe.
• Status and run history per job.

Non-functional requirements

• Durability first. A scheduled job must survive node and datacenter loss. Losing a "charge the customer" job is a business incident, not a blip.
• Horizontally scalable to hundreds of millions of pending jobs and tens of thousands of fires per second at peak.
• No single point of failure. A dead scheduler node should cause a brief delay, not dropped jobs.
• Bounded lateness. P99 fire latency within a few seconds of the target time.
• Multi-tenant fairness. One customer's million-job burst should not starve everyone else.

2Back-of-the-envelope estimation#

Get the scale in your head, because one number decides the architecture. I will flag it when I reach it.

Pending set and average rate.

• Say 200M pending jobs at any moment, mostly recurring with fewer one-shots.
• If recurring jobs run roughly hourly and one-shots are a minority, call it 2 billion fires per day.
• 2B over 86,400s is about 23,000 fires/second on average, which is comfortable.

The spike (this is the one that matters).

• Humans love round numbers: everyone schedules for :00 of the hour, midnight UTC, the top of the minute.
• Assume peak is 10x the average inside a one-second window: about 230,000 fires/second at the worst tick.
• Store jobs and poll "what is due now," and at midnight a thundering herd slams a single tick.
• The architecture has to smear that load across time, not swallow it in one gulp.

Storage.

• 200M jobs at roughly 1 KB each (metadata, cron string, payload pointer, status) is about 200 GB for the live set, which fits a sharded store.
• Run history at 2B fires/day, 7-day retention, ~300 bytes per record is 2B x 7 x 300 bytes, roughly 4.2 TB.
• History goes to cheap append-only storage, never the hot path.

Bandwidth on fire.

• If each fire sends a 2 KB payload callback, 230k/s x 2 KB is about 460 MB/s out of the dispatch tier at peak.
• Fine for a horizontally scaled fleet of dispatchers behind connection pools.

3Core APIs#

Keep these small and boring on purpose. The interesting design hides in two fields, and I will point at them.

POST /v1/jobs
Authorization: Bearer <token>
Content-Type: application/json

{
  "type": "recurring",
  "schedule": "0 */1 * * *",
  "timezone": "America/New_York",
  "target": { "kind": "http", "url": "https://api.tenant.com/hook", "method": "POST" },
  "payload": { "report": "daily" },
  "retry": { "max_attempts": 5, "backoff": "exponential" },
  "idempotency_key_template": "job-{job_id}-run-{scheduled_ts}"
}

# Response
201 Created
{ "job_id": "jb_8fh2", "next_run_at": "2026-11-03T15:00:00Z", "status": "scheduled" }

GET    /v1/jobs/{job_id}          # current state + next_run_at
DELETE /v1/jobs/{job_id}          # cancel; soft-delete, tombstone
PATCH  /v1/jobs/{job_id}          # update schedule or payload
GET    /v1/jobs/{job_id}/runs     # paginated run history

Two design choices worth defending:

• The idempotency key template is load-bearing, not decoration. Because we are at-least-once, the key the handler sees has to be stable for a given scheduled instant. I compute it from job_id plus scheduled_ts, never a per-attempt UUID, so a retry or an accidental duplicate carries the exact same key and the user dedupes cleanly. A per-attempt UUID would defeat the whole point.
• Cancel is a soft delete (tombstone), not a hard delete. A fire already in flight when the cancel lands can check the tombstone and bail, instead of racing a hard delete.

4High-level design#

The spine is two planes with a queue between them, so draw it fast:

• A scheduling plane decides what should fire and when.
• An execution plane actually fires it and tracks the result.
• A durable queue sits between them so a slow or failing handler never backs up the clock.
• Jobs are partitioned into shards; scheduler nodes claim shards via leader election so exactly one node owns a shard at a time.

Write path.

• The client hits the Job API, which validates the request.
• It computes the first next_run_at from the cron expression plus timezone.
• It writes the row to the durable Job Store (the source of truth) and returns.
• Nothing fires yet, and that is the point. The write path is boring and durable on purpose.

Read and trigger path.

• Scheduler nodes continuously look ahead a short window (say the next 60 seconds) inside their owned shards.
• They pull jobs whose next_run_at falls in that window and enqueue them into the ready queue, keyed by second.
• Dispatchers pull from the current second's bucket, call the handler, and record the result.
• For recurring jobs the dispatcher computes the next occurrence and writes it back.
• If a handler fails, the dispatcher re-enqueues into a delayed bucket with backoff: retries loop back through the same durable path.

Why split it this way.

• Scheduling is a steady scan; dispatching is spiky and latency-sensitive. Different shapes, different scaling.
• The queue decouples "the job is due" from "the handler responded," so a slow handler never stalls the clock.
• It gives me the one lever that tames the spike: schedulers enqueue fast, dispatchers consume at a pace I control.

5Data model#

Use a few stores, each picked for its access pattern. Jobs and run history are different beasts, so I do not force them into one layout.

The due-scan is the access pattern that decides everything.

• Two patterns dominate: point lookups by job_id, and a range scan that says "give me jobs in shard S due before time T."
• That second one is a hot, latency-sensitive range query running constantly alongside next_run_at updates.
• Pick the wrong primary key layout and the scan becomes a full-table crawl that falls over at the spike.
• So the partition key is the shard and the clustering key is next_run_at: the due-scan is a cheap ordered range read, not a scan-and-filter.

Store choice: a horizontally partitioned SQL store or a strongly consistent wide-column store (sharded Postgres, CockroachDB, or Cassandra and DynamoDB used carefully). I lean toward strong per-key consistency, because the next_run_at update after a fire must not race a second dispatcher.

jobs (source of truth, partitioned by shard, clustered by next_run_at):

runs (append-only, cheap storage, not the hot path):

• Jobs need transactional updates and ordered range scans; runs are write-heavy, read-rarely, and perfect for a log-structured or columnar store.
• Big payloads go to object storage behind payload_ref, so the hot row stays near 1 KB and the due-scan stays fast.
• I keep latency_ms on every run row because it is my single best health signal: it is literally fire time minus scheduled time, and it is the thing that spikes first when the design is under stress.

6Deep dives#

6.1 At-least-once execution and exactly-once effects (the signature hard part)#

This is the heart of the problem. You cannot get exactly-once execution across a network, full stop.

• A dispatcher calls the handler, the handler runs, then the dispatcher crashes before recording success.
• On recovery we genuinely do not know whether it ran.
• Skip it and that is at-most-once: we might drop a payment.
• Retry it and that is at-least-once: we might double-charge.
• At-least-once is the only honest choice, so the real work is making duplicates harmless. I say this to the interviewer directly: the goal is exactly-once effects, not exactly-once execution.

Three mechanisms working together get us to safe duplicates.

• 1. Lease with fencing. When a dispatcher claims a due job it writes lease_owner and lease_until = now + 30s under optimistic concurrency using the version column. Only the current holder may mark the run complete. If it dies, the lease expires and another dispatcher reclaims it. The version is the fence: a dispatcher that paused on a long GC and woke up still thinking it owns the job cannot write a stale result, because its version is behind and the CAS fails.
• 2. Idempotency key handed to the handler. Derived from job_id plus scheduled_ts, so every attempt for the same scheduled instant carries an identical key. The handler is contractually expected to dedupe on it. This is how "charge card" becomes safe: the payment service sees the same key on the duplicate and refuses the second charge. We do not make the effect exactly-once ourselves, we hand the downstream the tool to do it.
• 3. Transactional state transition. Marking a run done and computing the next occurrence for a recurring job happen in one atomic write. Either the run is recorded and next_run_at advances, or neither does. There is no window where a job is both "done" and "still due," which would otherwise produce a phantom second fire on the next scan.

Missed and overlapping runs are the two failure modes interviewers love to probe.

• Missed runs. If the whole scheduler fleet was down for 10 minutes, on recovery a shard's scan finds jobs whose next_run_at is in the past. For a recurring job I usually fire once for the most recent missed slot and skip the backlog ("catch up at most one"), because replaying 10 minutes of hourly jobs is almost never what the user wants. Configurable per job.
• Overlapping runs. A long-running recurring job can have its next fire arrive before the previous one finished. I guard with a per-job concurrency flag: allow_overlap: false means the scheduler will not enqueue the next instance while a lease on the current run is still live.

Trade-off I am stating on purpose.

• I chose at-least-once with idempotency over building a two-phase-commit "exactly-once" fire.
• 2PC across our scheduler and an arbitrary tenant handler is fragile and slow.
• Most tenant handlers simply cannot participate in a distributed transaction we control.
• Pushing dedupe to a stable idempotency key is cheaper, degrades gracefully, and actually works in production.

6.2 Leader election, sharding, and absorbing the spike#

The scheduling plane needs exactly one owner per shard, and it has to rebalance when nodes join or die.

• I keep a coordination store (etcd, ZooKeeper, or a Raft group) holding a lease per shard.
• A scheduler node acquires a shard lease, renews it every few seconds, and if it fails to renew, the lease expires and another node grabs it.
• This is leader election at shard granularity, not one global leader, so I get real parallelism and no single scheduling bottleneck.

Why shard by hash(job_id).

• It spreads load evenly and makes ownership a pure function of the key: any node computes which shard owns a job without asking anyone.
• The downside: a tenant's jobs scatter across shards. That is good for fairness (no shard becomes a per-tenant hotspot) but means per-tenant rate limiting has to be coordinated separately, it does not fall out of the sharding for free.

Now the spike, which is the reason this whole plane exists.

• At 00:00:00 every shard's look-ahead scan surfaces a pile of due jobs at once.
• Fire them all in the same second and you melt the handlers and the queue together.

• Two defenses, used together:

• Time-bucketed queue with jitter. The ready queue is partitioned by second-bucket. Schedulers enqueue into the exact target second and dispatchers drain at a per-tenant rate limit. For jobs that only need "approximately on the hour," I add a small deterministic jitter derived from job_id (say 0 to 5 seconds), computed at scheduling time so the herd self-spreads across six buckets instead of slamming one. Jobs that genuinely need exact timing opt out and pay for a reserved-rate lane.

• Look-ahead window. Schedulers scan the next 60 seconds, not just "now." This pre-loads the queue so the dispatch tier has a steady runway instead of hitting a cliff at each tick. Scanning ahead also hides Job Store read latency, since the read happens well before the deadline rather than on the critical path at T.

The combination turns 230k-in-one-tick into a roughly flat drain: jitter spreads the herd, buckets sequence it, and the per-tenant rate limit caps what any one customer can inflict.

6.3 Reliable triggering without losing or duplicating the enqueue#

There is a subtle gap between "the scheduler decided this job is due" and "it is safely sitting in the ready queue."

• Enqueue, then crash before recording the enqueue: on restart it re-scans and enqueues again, a duplicate.
• Record the hand-off, then crash before enqueuing: the job is marked handed-off but never reached the queue, a lost fire.
• This is the classic dual-write problem, and losing a fire is the one outcome my durability-first requirement forbids.

The clean fix is the transactional outbox.

• The scheduler, in one transaction against the Job Store, advances a handoff_ts and appends a row to an outbox table. Both land or neither does.
• A relay process tails the outbox, pushes to the ready queue, and marks outbox rows done only after a confirmed enqueue.
• The queue hand-off is itself at-least-once, so a duplicate can still reach the dispatcher. That is fine by design: the lease CAS means the second copy loses the race to claim and gets dropped.
• So the invariant holds end to end: never lose a fire, collapse any duplicate at the lease.

This is the posture of the whole design: layered redundancy.

• Every hand-off is at-least-once, and every consumer is idempotent.
• Correctness survives a crash at any single step.
• I would rather have three cheap at-least-once hops that each dedupe than one clever "exactly-once" hop that nobody can actually build.

7Bottlenecks and trade-offs#

The due-scan on the Job Store breaks first.

• At 230k due jobs in a tick spread over many shards, the range reads plus the next_run_at write-back are a write-amplification problem, because every fired recurring job is also a write.
• Keep the hot index on (shard_id, next_run_at) tight, batch the look-ahead reads, and batch the next_run_at updates into grouped writes.
• If a single shard still gets hot, split it. Shard count should be generous from day one (in the thousands), so rebalancing moves a thin slice and stays cheap.

The coordination store is a risk if you overuse it.

• etcd and ZooKeeper are great but not infinitely scalable, and people kill them by routing too much through them.
• I keep traffic tiny: only shard leases and renewals, never per-job state. Per-job coordination lives in the Job Store, not the consensus layer.
• If the coordination store blips, running leases keep their grace period, so a short outage is a brief rebalance pause, not dropped jobs.

Single points of failure: none by design.

• Schedulers are a pool, dispatchers are a pool, the Job Store is replicated with a quorum, and the coordination store is a Raft or ZAB quorum.
• The closest thing to a SPOF is the ready queue, so I run it as a replicated, partitioned log (Kafka-style), where a broker loss does not lose buckets.

Consistency call I made on purpose.

• Strong consistency on the fire path (lease CAS, transactional state transitions), eventual on run history and status reads.
• A dashboard showing a run as "running" for a second after it finished is a perfectly acceptable lie.
• I will not trade correctness on the fire path to make a dashboard snappier.

Clock skew.

• Schedulers decide "due" on their own clocks, so skew means a job could fire a second early or late.
• I rely on NTP-synced clocks and treat the 60-second look-ahead window as the skew buffer.
• For tighter bounds I would pin a job to a single time-authority, but I would not pay that cost for everyone when second-level precision is the stated norm.

Fairness and noisy neighbor.

• Per-tenant token buckets at the dispatcher, so one tenant's million-job midnight burst drains at their own rate limit while everyone else keeps flowing.
• The queue is tenant-aware, so a backed-up tenant cannot head-of-line block others sharing a partition.

Rank durability above latency on the fire path, then spend the consistency budget in one narrow place.

8Summary#

The decisions that matter, the ones I would lead with on a recap:

• Separate scheduling from execution with a durable time-bucketed queue, so a slow handler never stalls the clock and the midnight spike smears across buckets instead of landing in one tick.
• Shard by job_id with per-shard leader election, for parallelism, even load, and no global bottleneck.
• At-least-once plus idempotency keys and lease fencing for exactly-once effects, because exactly-once execution over a network to arbitrary handlers is a fantasy and saying so is the senior move.
• Outbox hand-off to kill the dual-write gap between "decided due" and "safely queued," with the lease collapsing any duplicate that slips through.
• Strong consistency on the fire path, eventual on reads, a deliberate and narrow place to spend the consistency budget.
• If I had more time: adaptive sharding that splits hot shards from live metrics, a dead-letter queue with alerting for jobs that exhaust retries, exact-timing priority lanes with reserved dispatcher capacity, and a what-if simulator so tenants can preview their next 24 hours before scheduling a herd into 00:00.

If you want to see how a real company frames the coordination and durability pieces, check our walkthrough on building the message queue that sits underneath systems like this.