Design a Distributed Job Scheduler

The single server's poll loop has hit the wall. This is the heart of the whole board, and it is worth being precise about what breaks, because the fix follows straight from it. The problem is not that tasks arrive too fast — accepting them is cheap. The problem is that finding the due ones means looking through the pending ones, and the pending ones are a vast, mostly-far-future set that grows without bound. Hundreds of millions of tasks are pending, scheduled for every imaginable future instant, and a few thousand of them come due every second.

The obvious first answer is a database table indexed by fire time that you poll every tick, and it is exactly what the single server did. It is familiar and durable, and an index on the fire time sounds right. But the workload kills it: Airbnb's own Quartz-on-a-relational-database saturated one big instance at about a thousand queries a second, each enqueue amplified into four SELECTs and three INSERTs, and more read replicas do not help because the bottleneck is the constant write-and-lock churn on the due-soon rows the poll is sweeping, and replicas do not take writes. A table you poll is fighting a workload that scans the whole future-dated set every second.

So the requirement sharpens: finding what's due has to be a cheap pop off the front, not a scan. A durable time-bucketed index gives that — group the pending tasks into buckets by when they are due, front the bucketing with a named hierarchical timing wheel (Kafka's Purgatory), and finding what's due is reading the bucket under the clock hand, a cost that does not grow with the pending set. The Redis sorted-set queue is a real in-production alternative that also beats the table; the bucketed timing wheel wins here on the cleanliness of the due-now pop at this scale. And the timing wheel is not invented at the whiteboard — it is a named, benchmarked primitive, so the design commits to it and does not re-derive its bucket arithmetic.

A durable, time-bucketed index fronted by a hierarchical timing wheel. The obvious first answer is the one the single server used, scaled up: keep the tasks in a database table indexed by fire time, and poll it. Add read replicas, buy a bigger box. It is familiar and durable, and the attraction is real — you already run databases, and an index on the fire time sounds like exactly the right tool. But watch what the workload does to it. Airbnb ran precisely this design — Quartz on a large relational database — and measured where it broke: one big instance saturated at about a thousand queries a second, with each task enqueue amplified into four SELECTs and three INSERTs. A thousand a second is less than a quarter of the derived average due-rate of forty-six hundred, before any peak multiplier. And here is why more replicas do not save it: the bottleneck is not read capacity, it is the constant write-and-lock churn on the due-soon rows — marking tasks fired, contending on the rows the poll is sweeping — and read replicas do not take writes. You cannot replica your way out of a write-and-lock bottleneck. A table you poll is fighting the workload.

So the requirement sharpens: the structure has to make what's due now a cheap pop off the front, not a scan of everything. That is what a durable time-bucketed index gives you. Instead of one big table sorted by fire time, group the pending tasks into buckets by when they are due — a bucket for each small slice of time — so that finding what's due now means looking only at the bucket for now, never at the buckets for next week. Fronting that bucketing is a named, decades-old structure built for exactly this: a hierarchical timing wheel, the same one Kafka uses inside its Purgatory to track millions of in-flight timeouts. Picture a clock hand sweeping a ring of buckets, one bucket per tick; a task scheduled for a given instant is dropped into the bucket the hand will reach at that instant, and when the hand arrives, everything in that bucket is due. Inserting a task is dropping it in a bucket, and popping the due ones is reading the bucket under the hand — both cheap, and neither depends on how many tasks are pending elsewhere. For due times too far out for one ring, higher-level wheels at coarser resolution cover the longer range, and a task cascades down to a finer wheel as its time approaches. And it is durable: the buckets are backed by a key-value store at about a kilobyte per task, so a crash does not lose the pending set.

There is a genuine alternative worth a real hearing, because a named company ships it: a Redis sorted-set delayed queue, the shape Netflix built on Dynomite. Keep the pending tasks in a sorted set scored by fire time, poll the due ones up to now, and claim each atomically. It works, it is in production, and its command-level mechanics are clean — one small script closes the poll-and-claim race, and instead of busy-polling you peek the next-lowest score and sleep until that instant. Its cost is in the constant factor: Netflix's design needs three coordinated structures per queue — a sorted set of queued items, a hash of payloads, and a second sorted set of in-flight items — so it carries more index entries per task than a single-record store, and it shards by availability zone rather than by the timing wheel's bucketing. It is a real, defensible choice, and the reason it is not the pick here is only that the bucketed timing wheel makes the what's-due-now pop cleaner at the pending-set sizes this board is built for. Both beat the table.

Two things are worth pinning under the hood. First, notice which part is cheap and which is the wall. The timing wheel's own throughput is enormous — a single node handles a hundred and five thousand requests a second at a million outstanding timers, roughly twenty times the derived average due-rate — so the algorithm for finding what's due is not where this board strains. The wall is everything durable behind it: holding four hundred million tasks on disk, replicated, and split across machines, which is the next use case. Naming which wall bites is half the skill here, and it is not the timing wheel. Second, there is a shortcut worth knowing: a task due very soon — Dynein uses fifteen minutes — does not need to go through the durable time-bucketed index at all. It can skip straight to the dispatch path, because it will fire before the index would even finish settling it. So the index only ever holds the genuinely-future tasks, and the short-fuse jobs take a fast lane.

The index answers what's due now cheaply now. But it still lives on one machine, and the pending set is about to tell us that cannot hold. The pending set is hundreds of millions of future-dated tasks, each about a kilobyte, held durably. That does not fit on one machine's disk, and if it did, losing that machine would lose every pending task on it. This use case is short and it is arithmetic, because the number does all the arguing.

The pick is not really in doubt — a durable set that cannot afford to lose a single task cannot sit on one machine, so you shard. The real choice is the shard key. Shard by a hash of the task id, with consistent hashing, and the pending set spreads evenly and a node addition moves only a small slice; each shard runs its own timing wheel over its slice and replicates it, so a node loss costs neither availability nor a task. Airbnb's rebuilt index gets virtually unlimited linear scalability precisely because the store shards under the hood by partition key — a single-node store gets none of that for free, which is why sharding is a decision, not an afterthought.

The shard key that reads natural and is wrong is fire time: shard by when the task is due, and every task due at the same instant lands on the same node, so the midnight bucket concentrates onto one shard — the exact hot bucket the last use case is about. Hashing by task id spreads that load; sharding by fire time gathers it. And the number does the arguing on why to shard at all: not because a bit over a terabyte is a lot of disk — it isn't — but because a durable set that cannot afford to lose a single task cannot sit on a single machine.

Shard the durable index by a hash of the task id, with consistent hashing and replication. Do the arithmetic out loud, from the bytes up, and flag every derived figure as derived. Start from the one cited anchor: Uber's Cadence runs over twelve billion executions a month, which spread across a thirty-day month is about forty-six hundred dispatches a second on average — derived, not a sourced point-in-time figure. Now ask how many tasks are resident in the pending set at once. By Little's Law, that is the due-rate times how long a task sits pending before it fires. Take a one-day average dwell as a stated planning assumption — some tasks fire in seconds, many are recurring or scheduled weeks out, and one day is a reasonable middle — and the pending count is about forty-six hundred a second times eighty-six thousand four hundred seconds, roughly four hundred million tasks resident at once, derived. At Dynein's measured one kilobyte per task, one full copy of that set is about four hundred gigabytes. And because the index is durable and must survive a machine loss, it is replicated — a standard threefold replication gives a bit over a terabyte fleet-wide, derived. That is small in absolute disk. But it is more than one machine should hold as a single point of loss, and the write-and-index rate across it is more than one machine should serve, so the index is split.

The pick is not really in doubt — you shard — but how you shard is a real choice, and there is a clean way. Split the index by a hash of the task id, using consistent hashing so the pending set spreads evenly across nodes and so adding or removing a node moves only a small slice of tasks rather than reshuffling everything. This is exactly Airbnb's experience: rebuilt on a key-value store that shards under the hood by partition key, the index gets virtually unlimited, linear scalability — but note why that phrase holds. It holds because the store shards automatically by partition key; a single-node store does not get linear scaling for free, which is the whole reason sharding is a decision here and not an afterthought. Each shard runs its own timing wheel over its slice of the buckets, pops its own due tasks, and holds a replicated copy of its slice, so losing one node loses neither availability nor any task. And there is a shard key that reads natural and is wrong: shard by fire time, and every task due in the same instant lands on the same node, so the midnight bucket concentrates onto one shard — the very hot bucket the last use case is about. Hashing by task id spreads that load; sharding by fire time gathers it.

This is the both-sizings habit, and this board is unusual in that two independent walls bite. The first sizing is capacity: the pending set is a count of tasks times bytes per task times replication — the derived bit-over-a-terabyte — and it is split across nodes for durability, not for raw disk. The second sizing is throughput, and it lives on the execution side, not here: the peak due-rate has to be dispatched to workers, and as the next use case shows, a single dispatch queue has a hard ceiling on how many tasks it can hold in flight at once, which forces fanning out across many queues. So you size the index by capacity and the dispatch tier by throughput — separate calculations in separate units — and you provision each for its own wall. The timing-wheel algorithm, notably, is the ceiling on neither — at a hundred and five thousand a second per node it is roughly twenty times the average due-rate. The walls are the durable storage here and the dispatch-queue depth next, never the algorithm.

The index holds the pending set and surfaces the due tasks. Now a task is due, and it has to reach a worker and actually run — and the worker might crash halfway through. This is the execution half, and it is where the reliability story is decided. It is deliberately reached after the time-index wall is paid: finding what's due was the signature; running it safely is the supporting mechanism. You need to get a due task to a worker, have the worker run its payload, and be certain the task actually ran — even if the worker it went to crashes mid-run, or the acknowledgement is lost.

The obvious answer is fire-and-forget — pop the due task, send it to a worker, done — and it would be correct if workers never died. But workers die mid-run, before acknowledging, all the time at fleet scale, and fire-and-forget silently drops a task whose worker crashes, trading away the one thing a scheduler cannot trade: never dropping a task. So the hand-off has to be one you can take back. Lease-and-ack with a visibility timeout — a queue like SQS — leases the task with a timer, deletes it on ack, and redelivers it to another worker if the timer expires with no ack. The task is never lost because it is not done until it is acked; the worst case is it runs again. That is at-least-once, what every named production scheduler ships.

At-least-once hands you a bill — a task can run twice — and the clean way to make that safe is a client-supplied idempotency key: the worker records the first run's result under the key and replays it for any redelivered duplicate within the safety window (Stripe honors it for twenty-four hours), instead of re-running the payload. The reliability story is not exactly-once, which no named system promises across arbitrary side effects; it is never dropped, and safe to repeat. Server-side dedup is a defensible variant, but a client key covers the effect's own idempotency where a server table only covers the delivery.

Lease-and-ack with a visibility timeout, at-least-once, made safe by an idempotency key. The obvious answer is fire-and-forget: pop the due task, send it to a worker, done. It is simple, and if workers never died it would be correct. But workers die — mid-run, before acknowledging, all the time at fleet scale — and with fire-and-forget a task handed to a worker that then crashes is simply gone. No retry, no error, a payment silently never retried. Fire-and-forget trades away the one thing a scheduler cannot trade: never dropping a task. So the hand-off has to be one you can take back. That is lease-and-ack with a visibility timeout, exactly what a queue like SQS provides. The dispatcher pops the due task and leases it to a worker — hands it over but keeps it, invisible to other workers, with a timer running (SQS defaults to thirty seconds, up to a twelve-hour maximum). If the worker finishes and acknowledges before the timer expires, the task is deleted, done. But if the worker crashes and the timer expires with no acknowledgement, the task becomes visible again and is redelivered to another worker. The task is never lost, because it is not considered done until it is acked; the worst case is that it runs again. That is at-least-once delivery, and it is what every named production scheduler ships — Dynein, SQS, Netflix's queue — none of them claims exactly-once.

At-least-once buys safety and hands you a bill: sometimes a task runs twice. The worker that seemed to die may have actually finished just as its lease expired, and the redelivered copy runs the payload a second time. So the worker's handler has to be idempotent, and the clean way to guarantee that is an idempotency key. The client supplies a unique key with the task; the worker records the result of the first run under that key, and if the same task arrives again within the safety window — Stripe honors the key for twenty-four hours — it replays the stored first result instead of re-running the payload. A charge that already happened is not charged again; the second run returns the first run's answer. The key is what makes runs-at-least-once safe to live with.

Trace the due-dispatch flow, and note it is authored as hop pairs on the dispatcher, not one long chain. When a task comes due, the Due-time index pops the due task and hands it to the Dispatcher. The Dispatcher, separately, leases the task to a worker in the Worker fleet, starting the visibility timer. The worker runs the payload behind its idempotency key, and acknowledges. If the ack arrives in time the dispatcher deletes the task; if the timer expires first, the dispatcher redelivers to a different worker. The popping and the leasing are two separate hops on the Dispatcher; the running and the idempotency check happen inside the worker and are not hops.

Two facts shape the dispatch tier under the hood. First, a single queue cannot hold unlimited leased tasks: a standard SQS queue caps at about a hundred and twenty thousand in-flight — leased-but-unacked — messages. At a peak due-rate in the tens of thousands a second with even a few seconds of average run time, the in-flight count on one queue approaches that ceiling, so the dispatcher fans out across many queues rather than funneling every lease through one. This is the second independent wall the board sizes for — a throughput wall on the execution side, separate from the index's storage wall. Second, a task whose run fails — as opposed to whose worker died — should be retried, but not instantly and not forever: retry it on an exponential backoff with a little jitter so retries do not synchronize, extend the lease with a heartbeat call for a legitimately long-running task so it is not redelivered underneath itself, and after a bounded number of failed attempts move it to a dead-letter queue for a human to look at.

Step back and look at what is built. The durable time-bucketed index answers what's due now as a cheap pop, sharded so the pending set fits and survives a node loss; the dispatcher leases each due task to a worker at-least-once and redelivers on a crash; the worker's idempotency key makes a re-run safe. It is complete for the happy path. Now the premise arrives already broken, the way a real incident does — and the twist is that this is not an off-to-the-side failure. This is the signature's own structure, the time-bucketed index, hit on its hot axis in production. Google's own SRE book tells this story: across a datacenter, many independent teams each schedule a daily cron job, and asked when to run it, engineers overwhelmingly pick a round instant — midnight, the top of the hour — because that is the obvious thing to type.

The instinct, in the room and in the moment, is to read a synchronized load spike as a capacity shortfall and reach for the obvious lever — the midnight bucket overwhelms us, so add machines, give it headroom. And it is exactly the wrong move. The load is not evenly distributed and under-provisioned; it is pathologically concentrated on specific instants and idle between them. Provisioning for the spike pays for a fleet that sits nearly idle the rest of the day, and the spike can always grow past whatever you provision as more teams pick midnight. A multiplicative concentration is not bought out with a linear addition of capacity.

The fix that actually worked de-concentrates the load at its source: a hashing operator that spreads each job's launch time across a window instead of the dot of midnight, so the clock hand meets an even trickle instead of a wall, with the index's own sharding spreading any residual hot bucket across nodes. The reframe the whole board rewards is that the hot bucket lives on the due-time axis the index is built on, so the fix lives on that axis too — and jitter is load-bearing infrastructure, not a tuning knob, because Google's own data shows the residual spikes are real even with the spread in place.

Spread the synchronized due-times off the round instant — jitter as load-bearing infrastructure. So thirty teams' unrelated daily jobs all come due at exactly the same second. On the time-bucketed index, that means one bucket — the midnight bucket — holds a massive pile of tasks while its neighbors are nearly empty, and when the clock hand reaches it, all of that load fires at once. It is a hot time bucket: a thundering herd of synchronized due-times landing on a single slice of the index. And Google's own data shows it is not a one-time fluke — the launch spikes cluster at fixed schedule times and persist even after the first mitigations, a recurring production pattern, not a single dated glitch. The instinct, in the room and in the moment, is to read a synchronized load spike as a capacity shortfall and reach for the obvious lever: the midnight bucket overwhelms the system, so add more machines, give it headroom for the spike. And it is the wrong move, for a reason that goes to the shape of the problem. The load is not evenly distributed and under-provisioned; it is pathologically concentrated on specific instants and idle between them. Provisioning for the midnight spike means paying for a fleet that sits nearly idle the rest of the day, and the spike can always grow past whatever you provision as more teams pick midnight. You cannot buy your way out of a concentration problem with capacity; you have to de-concentrate the load.

The fix that actually worked was to stop pinning jobs to the literal round instant. Google added a hashing operator to the schedule that spreads each job's actual launch time across a configured window instead of landing it on the dot of midnight. The midnight pile smears out across many buckets, and the clock hand meets an even trickle instead of a wall. Where a bucket is still hot, sharding the index by task id spreads even that bucket's tasks across nodes. The reframe to carry out of this incident is the one the whole board rewards: the hot bucket is a property of the due-time axis, the same axis the index is built on, so the fix lives on that axis too. Jitter here is not a nice-to-have you sprinkle on at the end; it is load-bearing infrastructure — even with the spread in place the residual spikes are real, so the spreading is a permanent, necessary part of the design, not a one-time cleanup.

There is a second, distinct reliability layer here under the hood, and it is worth separating cleanly from the worker's idempotency key. That key protects against a worker running a task twice. But what protects against the dispatcher — the thing that leases tasks — dying mid-launch and its replacement re-firing a task that already went out? Google's cron service answers with a leader that synchronously notifies a quorum both before and after each launch: if it skipped the after notice, a leader that died right after launching would let its successor re-launch the same job. And every launched job's identity embeds its scheduled fire time, so a newly elected leader can look at the record of completed launches and skip the ones already done. Leader failover completes within about a minute. This is a different layer from the worker idempotency key — one dedupes at the dispatcher on leader change, the other dedupes at the worker on redelivery — and a strong answer names both without conflating them.