Design a distributed job scheduler
Design a distributed job scheduler for 300 million runs a day and a 10x spike at the top of every hour. Start with one loop and one table, grow it with leases and SKIP LOCKED, partitions by job ID, history off the hot table, owners with timing wheels and a paired region, then stress it with a paused owner, double runs and one team's flood, with code.
“Design a distributed job scheduler. Teams across the company register jobs: run this every weekday at 9 in New York, run that once in 20 minutes. We run them when they are due.”
On one machine, a scheduler is a heap of due times and a pool of threads. That is the LLD task scheduler, and it isn’t repeated here. This page is the distributed system around it. The loop becomes many machines that pause, crash and disagree about the time. And they must still agree on one question: has this run already been handed out?
So this page starts with one loop and one table, and lets the numbers grow it, up to the spike at the top of every hour. Then it tests the result against a paused owner, a run that happens twice, and one team’s flood.
The answer in one minute
Section titled “The answer in one minute”If you only have a minute, this is the design. The rest of the page explains each part.
- A job and its next run are different rows. Each job’s next occurrence is written ahead of time as a pending run. So “what is due now” is one index, never a scan of 80 million jobs.
- Find and lease in one statement. A worker gets a run on a lease that expires unless it sends heartbeats. If the worker dies, the run comes back.
- Split by job ID. 256 partitions on 16 Postgres shards, so a run’s whole life is on one shard.
- Owners read ahead. At ten times the load, one owner per partition loads the next 30 seconds into an in-memory timer, so nobody polls.
- Assume two owners will overlap. A paused process doesn’t know it paused. The database rejects its writes with a fencing number, and handlers use the run ID to make a repeat harmless.
Scope it
Section titled “Scope it”Four questions decide the design. Do we run the code? How late can a run be? Can a run happen twice? And what happens to runs missed during an outage?
Design a distributed job scheduler. Teams register jobs, and we run them when they’re due.
Whether you pin down what “run” means, the scale and its peak, how late a run may be, and duplicates. Drawing a leader and a cron table before asking whether a duplicate run is acceptable reads as a memorized answer.
These four questions matter most:
- Do we run the job’s code? No. Each team’s own workers pull runs from us and run their own code. Running other teams’ code would be a compute platform, a different system.
- How late can a run be? Within 1 second of due at p50, 5 seconds at p99. Never early.
- Can a run happen twice? Rarely, after a failure. Handlers must cope with a repeat.
- Missed runs? Each job chooses. The default is “run the latest missed run once”.
From these answers we make five promises:
- Never lost. An accepted job is never lost, and neither is a due run.
- On time. A run starts within 1 second of due at p50 and 5 seconds at p99, even at the top of the hour.
- Always up. Dispatch works 99.95% of the time, 4.4 hours of downtime a year.
- Rare duplicates. A run executes twice only after a failure, under 0.01% of runs.
- Fair. One team’s burst of 10 million runs can’t make another team’s runs late.
We leave out running tenant code, workflows where job B waits for job A (a layer above), and in-process timers under a second.
Go deeperMore questions, and what each changes1 min
| Question | We assume | If the answer were different |
|---|---|---|
| How many jobs? | 20 million recurring schedules, 60 million one-off jobs a day, 5,000 queues | 100 times fewer fits one Postgres with SKIP LOCKED for years |
| Sub-100 ms precision? | No, 1 s | In-memory timers from day one |
| “Never twice”? | No | Impossible with side effects outside our system; we’d say so |
| How many teams? | About 1,000 | One team removes quotas and fair claiming |
| How long do jobs run? | Median 2 s, some up to 6 hours | Only short jobs would allow a fixed timeout with no heartbeats |
| Run every missed run? | Per job; default “latest” | “All” needs the backlog throttled |
The numbers that decide it
Section titled “The numbers that decide it”Now we work out the numbers. The live data is small. The write rate is not.
- Recurring schedules
- 20 million
- Runs per schedule a day
- 12
- One-off jobs a day
- 60 million
- Top of the hour over average
- 10×
- Run store writes per run
- 3
- Seconds per run
- 2
- History bytes per run
- 500 bytes
- Days of history kept
- 30
20 million recurring schedules at 12 runs a day, plus 60 million one-off jobs.
People write 0 * * * * and 0 9 * * *, so runs bunch at the start of every hour.
So we design for a burst we can see coming.
Every run costs about 3 writes: create the pending run, lease it, finish it. One Postgres primary takes about 10,000 indexed row writes a second.
So we split the run store from day one: one primary is already full at the average.
35,000 runs a second, each taking 2 seconds. Heartbeats every 10 seconds add 6,900 writes a second.
So we send a heartbeat per run.
Jobs plus pending runs at about 1 KB each, and 30 days of history at 500 bytes per attempt.
So we keep finished runs out of the hot table.
Try the calculator: with 1 million schedules, peak writes fall to about 4,900 a second, and one Postgres primary is the whole design. At ten times our numbers, writes pass a million a second.
- Runs a day
- 300 million
- Due runs, average
- 3,500/s
- Due runs, top of the hour
- 35,000/s
- Dispatchers and workers scale out; the question is the store behind them.
- Run store writes, peak
- 100,000/s
- Past one Postgres primary (about 10,000 writes/s): partition the run store.
- Shards at 70% busy (10,000 writes/s each)
- 15
- Round up to a power of two so partitions split evenly.
- Runs executing at the peak
- 69,000
- Little's law: the worker slots tenants must provision.
- Heartbeat writes, peak (one per run every 10 s)
- 6,900/s
- Small next to claims. Per-run heartbeats are fine.
- Run history kept
- 4.5 TB
- Far bigger than the live set: stream finished runs to a separate history store with a TTL.
- Live jobs and pending runs (1 KB each)
- 80 GB
- The live set is small: shards are for write rate, not for size.
- Lookahead memory (30 s of peak, 200 bytes a run)
- 210 MB
- A 30 second window fits in memory across a few owners.
Build the design
Section titled “Build the design”We start with the simplest design that is correct. Then we let the numbers break it, one at a time. Each step fixes only what broke, adds only the boxes that fix needs, and says what the fix costs. Throughout, follow one scenario: a team registers a job for 9 am, and a worker takes it.
Step 1. One loop, one table
Section titled “Step 1. One loop, one table”The simplest version is one process. It writes jobs to Postgres, and every second it selects rows
with due_at <= now(), runs them on a thread pool, and works out the next run.
Stage 1: One loop, one table
Stage 1 of 6
One loop, one table
Why this stage
Nothing yet.
This is the smallest design that meets every functional requirement.
What we add
One scheduler process that takes API calls, polls a Postgres table every second for due jobs, and runs them on its own thread pool.
What it costs
Execution shares a process with dispatch, one crash loses every run in flight, and nothing retries.
At 35k due runs per second, Dispatchers is 11667% busy. Stage 2 is the fix.
The numbers behind the busy bars
What one run costs
- Each due run costs 3 run store writes (create, claim, finish) and 2 history events.
What one machine can do
- One Postgres primary sustains about 10,000 indexed row writes a second. Batching claims raises that to about 25,000.
- A dispatcher serves about 5,000 claims a second by polling, or 20,000 by run ID.
- A worker host runs about 500 two-second runs at once, so 250 a second.
- A Redis cluster takes about 1.5 million sorted set operations a second.
A schedule and a run are different rows. The jobs row holds the definition: a cron
expression and time zone, or a one-off time. Its next occurrence is written ahead of time as a
pending run row. So finding due work is one index over one shape, whatever the job, and a retry
is the same run row with a later due_at.
The next fire time is computed in the job’s own time zone, then stored in UTC. 0 9 * * * in New
York is 13:00 UTC in summer and 14:00 in winter.
This version is correct on a quiet day, but it breaks in four ways. A crash loses every run in flight. A slow job delays the next. Nothing retries. And at the peak, one process runs a few hundred 2-second jobs a second, against 35,000.
Step 2. Workers lease runs
Section titled “Step 2. Workers lease runs”Teams’ workers long-poll stateless dispatchers. Taking a run is a lease. The run is leased
until a time 30 seconds out, and the worker extends it with a heartbeat every 10 seconds. If the
worker dies, the lease runs out. Then a reaper puts the run back to pending, with a delay.
That is the SQS visibility timeout, with heartbeats. So a 6-hour job doesn’t need a 6-hour timeout.
Dispatchers find and lease due runs in one statement:
// What a claim returns to the worker. leaseToken goes up by one on every claim, so it// tells this attempt apart from any earlier or later one: it is the run's fencing token.record Run(UUID runId, UUID jobId, int attempt, long leaseToken, Instant leaseExpiresAt, String payload) {}
// The run store, reduced to the two calls the dispatcher needs.interface RunStore { List<Run> query(String sql, Object... args); int execute(String sql, Object... args);}
class Dispatcher { // Finding due runs and leasing them is one statement, so there is no moment where a // run is "found" but not yet "taken" for another dispatcher to also find. // SKIP LOCKED lets twenty dispatchers run this at once without queuing on each // other's rows: each takes rows nobody else holds and moves on. // now() is the database clock, the only clock every dispatcher shares. // ORDER BY priority is strict priority; deep dive 4 replaces it with weighted lanes. // No paused check: pausing a job moves its pending run to 'paused', out of the // due index, so paused work never fills the LIMIT and starves the queue. static final String CLAIM = """ WITH due AS ( SELECT run_id FROM runs WHERE queue = ? AND state = 'pending' AND due_at <= now() ORDER BY priority DESC, due_at LIMIT ? FOR UPDATE SKIP LOCKED ) UPDATE runs r SET state = 'leased', attempt = r.attempt + 1, lease_token = r.lease_token + 1, lease_owner = ?, lease_expires_at = now() + make_interval(secs => ?) FROM due, jobs j WHERE r.run_id = due.run_id AND j.job_id = r.job_id RETURNING r.run_id, r.job_id, r.attempt, r.lease_token, r.lease_expires_at, j.payload """;
// The visibility timeout in action: a lease nobody extended goes back to pending. // The attempt was counted at claim, so a run that keeps crashing its worker runs // out of attempts and goes dead instead of crash-looping the fleet. // Full jitter: a random delay up to 2^attempt seconds, capped at 10 minutes, so runs // that failed together do not all come back in the same second. static final String REAP = """ WITH expired AS ( SELECT run_id FROM runs WHERE state = 'leased' AND lease_expires_at < now() ORDER BY lease_expires_at LIMIT ? FOR UPDATE SKIP LOCKED ) UPDATE runs r SET state = CASE WHEN r.attempt >= r.max_attempts THEN 'dead' ELSE 'pending' END, due_at = now() + make_interval(secs => least(600, 2 ^ r.attempt) * random()), lease_owner = NULL FROM expired WHERE r.run_id = expired.run_id """;
private final RunStore db; private final Duration visibility; // 30 s: short, because heartbeats extend it
Dispatcher(RunStore db, Duration visibility) { this.db = db; this.visibility = visibility; }
// Called for a worker's long-poll. Taking up to `max` at once cuts round trips; the // worker asks for no more than it has free slots, so leases never sit idle. List<Run> claimDue(String queue, int max, String workerId) { return db.query(CLAIM, queue, max, workerId, visibility.toSeconds()); }
// Runs every second on every dispatcher. SKIP LOCKED makes that safe: two reapers // never reset the same row, and neither waits for the other. int reapExpired(int batch) { return db.execute(REAP, batch); }}# What a claim returns to the worker. lease_token goes up by one on every claim, so it# tells this attempt apart from any earlier or later one: it is the run's fencing token.@dataclass(frozen=True)class Run: run_id: UUID job_id: UUID attempt: int lease_token: int lease_expires_at: datetime payload: str
# The run store, reduced to the two calls the dispatcher needs.class RunStore(Protocol): def query(self, sql: str, *args: object) -> list[Run]: ... def execute(self, sql: str, *args: object) -> int: ...
# Finding due runs and leasing them is one statement, so there is no moment where a# run is "found" but not yet "taken" for another dispatcher to also find.# SKIP LOCKED lets twenty dispatchers run this at once without queuing on each# other's rows: each takes rows nobody else holds and moves on.# now() is the database clock, the only clock every dispatcher shares.# ORDER BY priority is strict priority; deep dive 4 replaces it with weighted lanes.# No paused check: pausing a job moves its pending run to 'paused', out of the# due index, so paused work never fills the LIMIT and starves the queue.CLAIM = """ WITH due AS ( SELECT run_id FROM runs WHERE queue = %s AND state = 'pending' AND due_at <= now() ORDER BY priority DESC, due_at LIMIT %s FOR UPDATE SKIP LOCKED ) UPDATE runs r SET state = 'leased', attempt = r.attempt + 1, lease_token = r.lease_token + 1, lease_owner = %s, lease_expires_at = now() + make_interval(secs => %s) FROM due, jobs j WHERE r.run_id = due.run_id AND j.job_id = r.job_id RETURNING r.run_id, r.job_id, r.attempt, r.lease_token, r.lease_expires_at, j.payload"""
# The visibility timeout in action: a lease nobody extended goes back to pending.# The attempt was counted at claim, so a run that keeps crashing its worker runs# out of attempts and goes dead instead of crash-looping the fleet.# Full jitter: a random delay up to 2^attempt seconds, capped at 10 minutes, so runs# that failed together do not all come back in the same second.REAP = """ WITH expired AS ( SELECT run_id FROM runs WHERE state = 'leased' AND lease_expires_at < now() ORDER BY lease_expires_at LIMIT %s FOR UPDATE SKIP LOCKED ) UPDATE runs r SET state = CASE WHEN r.attempt >= r.max_attempts THEN 'dead' ELSE 'pending' END, due_at = now() + make_interval(secs => least(600, 2 ^ r.attempt) * random()), lease_owner = NULL FROM expired WHERE r.run_id = expired.run_id"""
class Dispatcher: def __init__(self, db: RunStore, visibility: timedelta): self._db = db self._visibility = visibility # 30 s: short, because heartbeats extend it
# Called for a worker's long-poll. Taking up to max_runs at once cuts round trips; the # worker asks for no more than it has free slots, so leases never sit idle. def claim_due(self, queue: str, max_runs: int, worker_id: str) -> list[Run]: return self._db.query(CLAIM, queue, max_runs, worker_id, self._visibility.total_seconds())
# Runs every second on every dispatcher. SKIP LOCKED makes that safe: two reapers # never reset the same row, and neither waits for the other. def reap_expired(self, batch: int) -> int: return self._db.execute(REAP, batch)// What a claim returns to the worker. LeaseToken goes up by one on every claim, so it// tells this attempt apart from any earlier or later one: it is the run's fencing token.record Run(Guid RunId, Guid JobId, int Attempt, long LeaseToken, DateTime LeaseExpiresAt, string Payload);
// The run store, reduced to the two calls the dispatcher needs.interface IRunStore{ IReadOnlyList<Run> Query(string sql, params object[] args); int Execute(string sql, params object[] args);}
// visibility is 30 s: short, because heartbeats extend it.class Dispatcher(IRunStore db, TimeSpan visibility){ // Finding due runs and leasing them is one statement, so there is no moment where a // run is "found" but not yet "taken" for another dispatcher to also find. // SKIP LOCKED lets twenty dispatchers run this at once without queuing on each // other's rows: each takes rows nobody else holds and moves on. // now() is the database clock, the only clock every dispatcher shares. // ORDER BY priority is strict priority; deep dive 4 replaces it with weighted lanes. // No paused check: pausing a job moves its pending run to 'paused', out of the // due index, so paused work never fills the LIMIT and starves the queue. const string Claim = """ WITH due AS ( SELECT run_id FROM runs WHERE queue = @queue AND state = 'pending' AND due_at <= now() ORDER BY priority DESC, due_at LIMIT @max FOR UPDATE SKIP LOCKED ) UPDATE runs r SET state = 'leased', attempt = r.attempt + 1, lease_token = r.lease_token + 1, lease_owner = @worker, lease_expires_at = now() + make_interval(secs => @visibility) FROM due, jobs j WHERE r.run_id = due.run_id AND j.job_id = r.job_id RETURNING r.run_id, r.job_id, r.attempt, r.lease_token, r.lease_expires_at, j.payload """;
// The visibility timeout in action: a lease nobody extended goes back to pending. // The attempt was counted at claim, so a run that keeps crashing its worker runs // out of attempts and goes dead instead of crash-looping the fleet. // Full jitter: a random delay up to 2^attempt seconds, capped at 10 minutes, so runs // that failed together do not all come back in the same second. const string Reap = """ WITH expired AS ( SELECT run_id FROM runs WHERE state = 'leased' AND lease_expires_at < now() ORDER BY lease_expires_at LIMIT @batch FOR UPDATE SKIP LOCKED ) UPDATE runs r SET state = CASE WHEN r.attempt >= r.max_attempts THEN 'dead' ELSE 'pending' END, due_at = now() + make_interval(secs => least(600, 2 ^ r.attempt) * random()), lease_owner = NULL FROM expired WHERE r.run_id = expired.run_id """;
// Called for a worker's long-poll. Taking up to `max` at once cuts round trips; the // worker asks for no more than it has free slots, so leases never sit idle. public IReadOnlyList<Run> ClaimDue(string queue, int max, string workerId) => db.Query(Claim, queue, max, workerId, visibility.TotalSeconds);
// Runs every second on every dispatcher. SKIP LOCKED makes that safe: two reapers // never reset the same row, and neither waits for the other. public int ReapExpired(int batch) => db.Execute(Reap, batch);}// Run is what a claim returns to the worker. LeaseToken goes up by one on every claim, so it// tells this attempt apart from any earlier or later one: it is the run's fencing token.type Run struct { RunID, JobID string Attempt int LeaseToken int64 LeaseExpiresAt time.Time Payload string}
// RunStore is the run store, reduced to the two calls the dispatcher needs.type RunStore interface { Query(sql string, args ...any) ([]Run, error) Exec(sql string, args ...any) (int64, error)}
// Finding due runs and leasing them is one statement, so there is no moment where a// run is "found" but not yet "taken" for another dispatcher to also find.// SKIP LOCKED lets twenty dispatchers run this at once without queuing on each// other's rows: each takes rows nobody else holds and moves on.// now() is the database clock, the only clock every dispatcher shares.// ORDER BY priority is strict priority; deep dive 4 replaces it with weighted lanes.// No paused check: pausing a job moves its pending run to 'paused', out of the// due index, so paused work never fills the LIMIT and starves the queue.const claimSQL = `WITH due AS ( SELECT run_id FROM runs WHERE queue = $1 AND state = 'pending' AND due_at <= now() ORDER BY priority DESC, due_at LIMIT $2 FOR UPDATE SKIP LOCKED)UPDATE runs rSET state = 'leased', attempt = r.attempt + 1, lease_token = r.lease_token + 1, lease_owner = $3, lease_expires_at = now() + make_interval(secs => $4)FROM due, jobs jWHERE r.run_id = due.run_id AND j.job_id = r.job_idRETURNING r.run_id, r.job_id, r.attempt, r.lease_token, r.lease_expires_at, j.payload`
// The visibility timeout in action: a lease nobody extended goes back to pending.// The attempt was counted at claim, so a run that keeps crashing its worker runs// out of attempts and goes dead instead of crash-looping the fleet.// Full jitter: a random delay up to 2^attempt seconds, capped at 10 minutes, so runs// that failed together do not all come back in the same second.const reapSQL = `WITH expired AS ( SELECT run_id FROM runs WHERE state = 'leased' AND lease_expires_at < now() ORDER BY lease_expires_at LIMIT $1 FOR UPDATE SKIP LOCKED)UPDATE runs rSET state = CASE WHEN r.attempt >= r.max_attempts THEN 'dead' ELSE 'pending' END, due_at = now() + make_interval(secs => least(600, 2 ^ r.attempt) * random()), lease_owner = NULLFROM expiredWHERE r.run_id = expired.run_id`
type Dispatcher struct { db RunStore visibility time.Duration // 30 s: short, because heartbeats extend it}
// ClaimDue is called for a worker's long-poll. Taking up to max at once cuts round trips;// the worker asks for no more than it has free slots, so leases never sit idle.func (d *Dispatcher) ClaimDue(queue string, max int, workerID string) ([]Run, error) { return d.db.Query(claimSQL, queue, max, workerID, d.visibility.Seconds())}
// ReapExpired runs every second on every dispatcher. SKIP LOCKED makes that safe: two// reapers never reset the same row, and neither waits for the other.func (d *Dispatcher) ReapExpired(batch int) (int64, error) { return d.db.Exec(reapSQL, batch)}// What a claim returns to the worker. leaseToken goes up by one on every claim, so it// tells this attempt apart from any earlier or later one: it is the run's fencing token.interface Run { runId: string; jobId: string; attempt: number; leaseToken: bigint; leaseExpiresAt: Date; payload: string;}
// The run store, reduced to the two calls the dispatcher needs.interface RunStore { query(sql: string, ...args: unknown[]): Promise<Run[]>; execute(sql: string, ...args: unknown[]): Promise<number>;}
// Finding due runs and leasing them is one statement, so there is no moment where a// run is "found" but not yet "taken" for another dispatcher to also find.// SKIP LOCKED lets twenty dispatchers run this at once without queuing on each// other's rows: each takes rows nobody else holds and moves on.// now() is the database clock, the only clock every dispatcher shares.// ORDER BY priority is strict priority; deep dive 4 replaces it with weighted lanes.// No paused check: pausing a job moves its pending run to 'paused', out of the// due index, so paused work never fills the LIMIT and starves the queue.const CLAIM = ` WITH due AS ( SELECT run_id FROM runs WHERE queue = $1 AND state = 'pending' AND due_at <= now() ORDER BY priority DESC, due_at LIMIT $2 FOR UPDATE SKIP LOCKED ) UPDATE runs r SET state = 'leased', attempt = r.attempt + 1, lease_token = r.lease_token + 1, lease_owner = $3, lease_expires_at = now() + make_interval(secs => $4) FROM due, jobs j WHERE r.run_id = due.run_id AND j.job_id = r.job_id RETURNING r.run_id, r.job_id, r.attempt, r.lease_token, r.lease_expires_at, j.payload`;
// The visibility timeout in action: a lease nobody extended goes back to pending.// The attempt was counted at claim, so a run that keeps crashing its worker runs// out of attempts and goes dead instead of crash-looping the fleet.// Full jitter: a random delay up to 2^attempt seconds, capped at 10 minutes, so runs// that failed together do not all come back in the same second.const REAP = ` WITH expired AS ( SELECT run_id FROM runs WHERE state = 'leased' AND lease_expires_at < now() ORDER BY lease_expires_at LIMIT $1 FOR UPDATE SKIP LOCKED ) UPDATE runs r SET state = CASE WHEN r.attempt >= r.max_attempts THEN 'dead' ELSE 'pending' END, due_at = now() + make_interval(secs => least(600, 2 ^ r.attempt) * random()), lease_owner = NULL FROM expired WHERE r.run_id = expired.run_id`;
class Dispatcher { constructor( private readonly db: RunStore, private readonly visibilitySeconds: number, // 30: short, because heartbeats extend it ) {}
// Called for a worker's long-poll. Taking up to `max` at once cuts round trips; the // worker asks for no more than it has free slots, so leases never sit idle. claimDue(queue: string, max: number, workerId: string): Promise<Run[]> { return this.db.query(CLAIM, queue, max, workerId, this.visibilitySeconds); }
// Runs every second on every dispatcher. SKIP LOCKED makes that safe: two reapers // never reset the same row, and neither waits for the other. reapExpired(batch: number): Promise<number> { return this.db.execute(REAP, batch); }}FOR UPDATE locks the rows the statement picks. SKIP LOCKED makes other dispatchers skip those
rows instead of waiting, so twenty dispatchers take twenty different batches at full speed. Finding
and leasing in one statement leaves no gap between “I saw it was due” and “I took it”.
The lease token goes up by one on every lease. A heartbeat or completion must carry it, so a
message from an older attempt matches zero rows and gets 409. The worker then knows to stop.
Stage 2: Workers lease runs
Stage 2 of 6
Workers lease runs
Why this stage
At 35,000 due runs/s of about 2 seconds each, 70,000 runs execute at once; one process runs a few hundred.
A crash mid-run must not lose the run.
What we add
Tenant workers that long-poll for runs, a runs table with a lease per claim (a visibility timeout), heartbeats, retries with backoff, and stateless dispatchers that claim with SKIP LOCKED.
What it costs
At-least-once execution: a run whose worker stalls past its lease runs twice, so handlers must be idempotent.
One Postgres primary takes every write.
At 35k due runs per second, Run store is 1050% busy. Stage 3 is the fix.
The numbers behind the busy bars
What one run costs
- Each due run costs 3 run store writes (create, claim, finish) and 2 history events.
What one machine can do
- One Postgres primary sustains about 10,000 indexed row writes a second. Batching claims raises that to about 25,000.
- A dispatcher serves about 5,000 claims a second by polling, or 20,000 by run ID.
- A worker host runs about 500 two-second runs at once, so 250 a second.
- A Redis cluster takes about 1.5 million sorted set operations a second.
Now runs survive crashes and spread over many workers. But every run costs about three writes, and the peak needs about 100,000 writes a second on one Postgres primary.
Go deeperWhy not put due runs on a message queue1 min
A message queue such as SQS or Kafka already has leases, retries and many consumers. So why keep runs in a database?
Because a schedule needs things a queue doesn’t do well. A queue hands out messages in arrival order. But runs must come out in due-time order, and a run due next March must wait quietly until then. Some queues offer delayed delivery, but only for minutes, not months. A schedule also gets edited. A team pauses a job, changes its cron line, or deletes it. In a table, that is one update to a row. In a queue, the old message is already out there, and something must find it and cancel it.
So the table holds the truth about every run, and the lease statement acts as the queue. A queue can still help later, as a fast list of runs that are already due. Step 4 adds exactly that, as hints.
Step 3. Split by job ID
Section titled “Step 3. Split by job ID”One Postgres primary is already full at the average. So the run store splits into 256 partitions by a hash of the job ID, on 16 shards. A job and its runs share a partition, so every change to a run touches one shard. Dispatchers run the lease statement against each shard in turn.
Why not split by due time? Every 9:00 run would land on one shard at 9:00. By team? The biggest team would outgrow its shard. Hashing the job ID spreads load evenly, whatever the clock or the team mix.
Stage 3: Partition the run store
Stage 3 of 6
Partition the run store
Why this stage
Each run costs 3 writes, so 3,500 runs/s is 10,000 writes/s on average and 100,000 at the top of the hour.
One primary sustains about 10,000.
What we add
256 logical partitions by hash of job ID on 16 Postgres shards.
A job and all its runs live in one partition, so every claim and completion touches one shard.
What it costs
Dispatchers poll every shard for every queue, so idle polling grows with queues times shards.
Cross-tenant queries scatter.
At 35k due runs per second, every part has headroom. Raise the traffic until something turns red.
The numbers behind the busy bars
What one run costs
- Each due run costs 3 run store writes (create, claim, finish) and 2 history events.
What one machine can do
- One Postgres primary sustains about 10,000 indexed row writes a second. Batching claims raises that to about 25,000.
- A dispatcher serves about 5,000 claims a second by polling, or 20,000 by run ID.
- A worker host runs about 500 two-second runs at once, so 250 a second.
- A Redis cluster takes about 1.5 million sorted set operations a second.
Step 4. History off the hot table
Section titled “Step 4. History off the hot table”Finished runs pile up, and the lease statement scans the same table. So history moves out.
Each change to a run also writes an outbox row in the same transaction. Change data capture streams it to Kafka, and a writer stores attempts in Cassandra by job, kept 30 days. Finished runs are then deleted from the shard, so the table the lease statement scans stays near 80 GB.
Stage 4: History off the hot table
Stage 4 of 6
History off the hot table
Why this stage
Finished runs add up to 4.5 TB over 30 days, while the live set is about 80 GB.
Keeping both in one table bloats the indexes every claim scans.
What we add
An outbox row per state change, streamed by change data capture into Kafka, and a history store in Cassandra with a 30 day TTL.
Finished runs leave the hot table.
What it costs
Run history lags the truth by a few seconds, and one more pipeline to run.
At 35k due runs per second, every part has headroom. Raise the traffic until something turns red.
The numbers behind the busy bars
What one run costs
- Each due run costs 3 run store writes (create, claim, finish) and 2 history events.
What one machine can do
- One Postgres primary sustains about 10,000 indexed row writes a second. Batching claims raises that to about 25,000.
- A dispatcher serves about 5,000 claims a second by polling, or 20,000 by run ID.
- A worker host runs about 500 two-second runs at once, so 250 a second.
- A Redis cluster takes about 1.5 million sorted set operations a second.
Follow a create, then a finished run:
Trace a request: Create a cron job
Create a cron job: step 1 of 3
Tenant apps to Job API
POST /v1/jobs with "0 9 * * MON-FRI" in America/New_York and an Idempotency-Key. The API parses the cron expression and computes the next fire time in that zone, then in UTC.
- Tenant apps → Job API2 ms
- Job API → Run store8 ms
- Response
Trace a request: Heartbeat and finish
Heartbeat and finish: step 1 of 4
Workers to Dispatchers
Every 10 seconds while the handler runs, the worker sends (run ID, lease token). Finishing is the same call with a result.
- Workers → Dispatchers1 ms
- Dispatchers → Run store4 ms
- Run store → Run eventsasync
- Response
This is enough for a mid-level answer. Materialized runs. Leases with heartbeats, and one SKIP LOCKED statement. 256 partitions by job ID. And history off the hot table. The stress tests are what make it a senior answer.
Go deeperThe API2 min
POST /v1/jobsAuthorization: Bearer <service-token>Idempotency-Key: 4b7e0d1c-...
{ "name": "nightly-invoice-rollup", "queue": "billing", "schedule": { "cron": "0 9 * * MON-FRI", "timeZone": "America/New_York" }, "payload": { "region": "us" }, "retry": { "maxAttempts": 5, "backoff": "exponential" }, "catchUp": "latest", "overlap": "skip", "timeoutSeconds": 900}
201 Created{ "jobId": "job_8f2c...", "nextRunAt": "2026-10-29T13:00:00Z" }A one-off job sends { "runAt": "..." } or { "delaySeconds": 1200 }. A retried create with the
same key returns the first job instead of scheduling it twice.
POST /v1/queues/billing/lease long-poll, up to 20 s{ "workerId": "w-17", "max": 10, "visibilitySeconds": 30 }
200 OK{ "runs": [ { "runId": "run_51a...", "attempt": 1, "leaseToken": 7, "scheduledFor": "...", "payload": { "region": "us" } } ] }
POST /v1/runs/{runId}/heartbeat { "leaseToken": 7 } 200, or 409 lease lostPOST /v1/runs/{runId}/complete { "leaseToken": 7, "result": {...} } 200, or 409POST /v1/runs/{runId}/fail { "leaseToken": 7, "error": "...", "retryable": true }Workers pull, so they take only what they have room for, and a slow team hurts only its own queue. Teams that want a webhook run a small worker that pulls from us and calls their URL.
Go deeperThe tables2 min
CREATE TABLE jobs ( job_id UUID PRIMARY KEY, partition_id SMALLINT NOT NULL, -- hash(job_id) mod 256 tenant_id BIGINT NOT NULL, queue TEXT NOT NULL, cron_expr TEXT, -- null for one-off jobs time_zone TEXT, -- e.g. America/New_York payload JSONB NOT NULL, paused BOOLEAN NOT NULL DEFAULT false, next_fire_at TIMESTAMPTZ, -- the next occurrence to materialize version BIGINT NOT NULL -- bumped on every edit);
CREATE TABLE runs ( run_id UUID PRIMARY KEY, -- derived from (job_id, scheduled_for) job_id UUID, partition_id SMALLINT, queue TEXT, priority SMALLINT, scheduled_for TIMESTAMPTZ NOT NULL, -- the fire time, never changes due_at TIMESTAMPTZ NOT NULL, -- fire time, or the next retry time state TEXT NOT NULL, -- pending | leased | paused | succeeded | dead attempt SMALLINT NOT NULL DEFAULT 0, lease_token BIGINT NOT NULL DEFAULT 0, lease_expires_at TIMESTAMPTZ);CREATE INDEX runs_due ON runs (queue, priority, due_at) WHERE state = 'pending';CREATE INDEX runs_expired ON runs (lease_expires_at) WHERE state = 'leased';Two partial indexes do the work. runs_due holds only pending runs, and runs_expired only leased
ones, for the reaper. Creating an occurrence inserts its run, and moves the job’s next_fire_at
forward, in one statement. It only moves it from the value it read, so each occurrence is created
once.
A queue table churns: every lease and finish leaves a dead row version. Autovacuum must be aggressive, long transactions killed, and finished runs moved out, which the outbox does.
| Store | Costs |
|---|---|
| Postgres | One primary per shard; vacuum to tune |
| DynamoDB | No SKIP LOCKED; the due-time index is eventually consistent |
| Cassandra | Leases need lightweight transactions (about 4 round trips); no partial index |
Stress it
Section titled “Stress it”The design is complete. Now we test it where schedulers are hard. The top of the hour at ten times the load. An owner that wakes up from a pause. A run that happens twice. One team that floods the system. And a lost region. Only the first and the last add boxes.
The top of the hour, at ten times the load
Section titled “The top of the hour, at ten times the load”Traffic grows ten times, so 350,000 runs are due in the first second of each hour.
Two limits arrive together. Writes reach a million a second, which 64 shards and batching fix. Polling is worse. Ten times the queues on four times the shards is over 12 million index lookups a second. Almost all are empty, and all hit the same index pages at the top of the hour.
The fix: one owner per partition, reading ahead. Each partition has one owner at a time,
held by a lease row in its shard. The owner reads every run due in the next 30 seconds into an
in-memory timing wheel. That’s an array of slots, one per 10 ms, with a pointer that fires each
slot in turn. When a slot fires, the owner adds the due run IDs to a ready list per queue, a
Redis sorted set. Dispatchers stop scanning. They pop the oldest IDs, and lease each by ID, with the
same check (state = 'pending' AND due_at <= now()).
Stage 5: Owners and timing wheels
Stage 5 of 6
Owners and timing wheels
Why this stage
At 350,000 due runs/s the store takes 1 million writes/s, and idle polling (queues × shards × 4 polls a second) passes 10 million probes a second.
What we add
One owner per partition, held by a lease with an epoch (the fencing token).
Each owner reads the next 30 seconds into a timing wheel and adds due run IDs to a ready list per queue.
Dispatchers claim by run ID.
64 shards with batched writes.
What it costs
Ownership, leases and handoff to run and reason about.
Ready lists are hints, so a sweeper must catch any hint that gets lost.
At 35k due runs per second, every part has headroom. Raise the traffic until something turns red.
The numbers behind the busy bars
What one run costs
- Each due run costs 3 run store writes (create, claim, finish) and 2 history events.
What one machine can do
- One Postgres primary sustains about 10,000 indexed row writes a second. Batching claims raises that to about 25,000.
- A dispatcher serves about 5,000 claims a second by polling, or 20,000 by run ID.
- A worker host runs about 500 two-second runs at once, so 250 a second.
- A Redis cluster takes about 1.5 million sorted set operations a second.
Trace a request: Top of the hour, stage 5
Top of the hour, stage 5: step 1 of 5
Timer owners to Run store
At 09:59:30 the owner of partition 17 already read every run due before 10:00:00 into its timing wheel. The spike of reads happened spread over the previous 30 seconds.
- Timer owners → Run storeasync
- Timer owners → Ready lists2 ms
- Dispatchers → Ready lists1 ms
- Dispatchers → Run store8 ms
- Response
Ready lists are only hints. The run store is the truth. A stale hint finds the run already leased and is dropped. A lost hint leaves the run pending, and each second the owner re-adds every run overdue by 5 seconds. So losing all of Redis costs 5 seconds of lateness, never a run.
The 9:00 spike announces itself, and three things flatten it. Owners load 9:00’s runs from 8:59:30,
so only the leases happen at 9:00. A jittered schedule, like Jenkins’ H * * * * (“once an hour, at
a minute derived from the job”), spreads jobs that don’t care which minute. And workers lease only
what they have room for, so a million runs for a 1,000-slot queue drain at that queue’s speed.
Go deeperWays to find due work, compared1 min
| Approach | How it finds due work | Loses |
|---|---|---|
| Scan the jobs table | Evaluate every cron expression every second | 20 million evaluations a second |
| Indexed poll + SKIP LOCKED | Range scan on a partial index of pending runs | Cost grows with pollers × shards; 100s of ms late |
| Time-bucketed index | Key (minute, shard) in a wide-row store | No row locks; late inserts land in past buckets |
| Timing wheel with a lookahead | An owner loads the next 30 s into memory slots | Needs one owner per partition: leases and fencing |
| Broker delayed delivery | SQS delay (15 minutes at most), Redis sorted sets | Short horizon or memory-bound; not the truth |
We poll up to about 10,000 leases a second per shard. Then we move to owners, keeping the poll as
the fallback when Redis is down. Both read the same table. So the move changes who looks, not where
the truth lives. Runs created or retried inside the 30-second window are found within a second.
Every change to pending stamps a sequence number. The owner reads rows above its last one, minus a
few seconds, because sequence numbers can commit out of order.
Kafka uses timing wheels for delayed requests (Confluent). Inserting is O(1). Runs beyond one turn sit in a coarser wheel, and move down. Dispatch latency drops from about 130 ms to about 11 ms.
The owner that wakes up from a pause
Section titled “The owner that wakes up from a pause”An owner pauses for 40 seconds in garbage collection. Its lease expires, and another owner takes over. Then the first one wakes up. What does it do, and what stops it?
Whether you know a lease alone can’t prevent two owners, so the store must check a fencing token.
A process that checks “my lease is valid until 10:00:10”, pauses for 40 seconds, and acts at 10:00:50 can’t tell that time passed. Nothing inside the process fixes that. Only the thing being written can, by refusing writes from an older holder. That is a fencing token.
So taking a partition lease adds one to its epoch. Every owner transaction first reads the
partition’s epoch, with a share lock, and aborts if it isn’t its own. A takeover is an update of
that row, which waits for share locks. So the old owner’s transaction either commits before the new
epoch exists, or sees it and aborts. And lease expiry is judged by the shard’s own now(): one
clock.
// One transaction on the shard that holds this partition.interface Shard { <T> T inTransaction(Function<Tx, T> work);}
interface Tx { Optional<Long> queryLong(String sql, Object... args); int execute(String sql, Object... args);}
class LeaseLost extends RuntimeException { LeaseLost(int partition) { super("lost partition " + partition); }}
// Owns one partition: materializes its cron runs, reaps its expired leases and feeds its// timing wheel (not shown). Two owners WILL overlap for a while (a GC pause, a network// partition), so safety comes from the shard checking the epoch, not from the lease alone.class PartitionOwner { // The lease lives in the same shard as the partition's rows, and expiry is judged by // that shard's now(), so no two machines' clocks are ever compared. // Taking a lease bumps the epoch: the epoch is the fencing token. static final String ACQUIRE = """ UPDATE partition_leases SET owner = ?, epoch = epoch + 1, expires_at = now() + interval '10 seconds' WHERE partition_id = ? AND expires_at < now() RETURNING epoch """; static final String RENEW = """ UPDATE partition_leases SET expires_at = now() + interval '10 seconds' WHERE partition_id = ? AND owner = ? AND epoch = ? """; // FOR SHARE: a takeover's UPDATE of this row waits until our transaction commits. So // once a new epoch commits, no write made under the old epoch can commit after it. static final String FENCE = "SELECT epoch FROM partition_leases WHERE partition_id = ? FOR SHARE"; // The next run of a cron job. The job's next_fire_at is the cursor: advancing it only // from the value we read creates each occurrence once, even after its finished run row // was deleted. The version check turns an edit in between into a no-op instead of a run // on the old schedule, and the derived run ID makes a retried insert a key conflict. static final String MATERIALIZE = """ WITH job AS ( UPDATE jobs SET next_fire_at = ? WHERE job_id = ? AND version = ? AND next_fire_at = ? AND NOT paused RETURNING job_id, partition_id, queue, priority, max_attempts ) INSERT INTO runs (run_id, job_id, partition_id, queue, priority, max_attempts, scheduled_for, due_at, state) SELECT ?, job_id, partition_id, queue, priority, max_attempts, ?, ?, 'pending' FROM job ON CONFLICT (run_id) DO NOTHING """;
private final Shard shard; private final int partition; private final String me; private long epoch = -1; // -1: we do not own it
PartitionOwner(Shard shard, int partition, String me) { this.shard = shard; this.partition = partition; this.me = me; }
boolean tryAcquire() { Optional<Long> got = shard.inTransaction(tx -> tx.queryLong(ACQUIRE, me, partition)); got.ifPresent(e -> epoch = e); // then read the next 30 s of runs into the wheel return got.isPresent(); }
// Every 3 seconds, so two missed renewals still leave time before the 10 s expiry. boolean renew() { boolean ok = shard.inTransaction(tx -> tx.execute(RENEW, partition, me, epoch)) == 1; if (!ok) drop(); return ok; }
// Every write the owner makes goes through here. Row conditions (pending, lease token, // job version) stop each race someone thought of; the epoch is one check that also // covers the ones nobody did, and it tells a zombie to stop. <T> T fenced(Function<Tx, T> work) { long mine = epoch; return shard.inTransaction(tx -> { long current = tx.queryLong(FENCE, partition).orElseThrow(); if (current != mine) { drop(); // a zombie learns it here, before touching a single run throw new LeaseLost(partition); } return work.apply(tx); }); }
// nextAt: the occurrence after fireAt, computed in the job's time zone. void materialize(UUID jobId, long version, Instant fireAt, Instant nextAt) { UUID runId = UUID.nameUUIDFromBytes((jobId + "@" + fireAt).getBytes(StandardCharsets.UTF_8)); Timestamp at = Timestamp.from(fireAt); Timestamp next = Timestamp.from(nextAt); fenced(tx -> tx.execute(MATERIALIZE, next, jobId, version, at, runId, at, at)); }
// Forget everything held in memory: the new owner will rebuild it from the shard. void drop() { epoch = -1; }}class Tx(Protocol): def query_int(self, sql: str, *args: object) -> int | None: ... def execute(self, sql: str, *args: object) -> int: ...
# One transaction on the shard that holds this partition.class Shard(Protocol): def in_transaction(self, work: Callable[[Tx], T]) -> T: ...
class LeaseLost(Exception): pass
# The lease lives in the same shard as the partition's rows, and expiry is judged by# that shard's now(), so no two machines' clocks are ever compared.# Taking a lease bumps the epoch: the epoch is the fencing token.ACQUIRE = """ UPDATE partition_leases SET owner = %s, epoch = epoch + 1, expires_at = now() + interval '10 seconds' WHERE partition_id = %s AND expires_at < now() RETURNING epoch"""RENEW = """ UPDATE partition_leases SET expires_at = now() + interval '10 seconds' WHERE partition_id = %s AND owner = %s AND epoch = %s"""# FOR SHARE: a takeover's UPDATE of this row waits until our transaction commits. So# once a new epoch commits, no write made under the old epoch can commit after it.FENCE = "SELECT epoch FROM partition_leases WHERE partition_id = %s FOR SHARE"# The next run of a cron job. The job's next_fire_at is the cursor: advancing it only# from the value we read creates each occurrence once, even after its finished run row# was deleted. The version check turns an edit in between into a no-op instead of a run# on the old schedule, and the derived run ID makes a retried insert a key conflict.MATERIALIZE = """ WITH job AS ( UPDATE jobs SET next_fire_at = %s WHERE job_id = %s AND version = %s AND next_fire_at = %s AND NOT paused RETURNING job_id, partition_id, queue, priority, max_attempts ) INSERT INTO runs (run_id, job_id, partition_id, queue, priority, max_attempts, scheduled_for, due_at, state) SELECT %s, job_id, partition_id, queue, priority, max_attempts, %s, %s, 'pending' FROM job ON CONFLICT (run_id) DO NOTHING"""
# Owns one partition: materializes its cron runs, reaps its expired leases and feeds its# timing wheel (not shown). Two owners WILL overlap for a while (a GC pause, a network# partition), so safety comes from the shard checking the epoch, not from the lease alone.class PartitionOwner: def __init__(self, shard: Shard, partition: int, me: str): self._shard = shard self._partition = partition self._me = me self._epoch = -1 # -1: we do not own it
def try_acquire(self) -> bool: got = self._shard.in_transaction(lambda tx: tx.query_int(ACQUIRE, self._me, self._partition)) if got is None: return False self._epoch = got # then read the next 30 s of runs into the wheel return True
# Every 3 seconds, so two missed renewals still leave time before the 10 s expiry. def renew(self) -> bool: ok = self._shard.in_transaction( lambda tx: tx.execute(RENEW, self._partition, self._me, self._epoch) == 1 ) if not ok: self.drop() return ok
# Every write the owner makes goes through here. Row conditions (pending, lease token, # job version) stop each race someone thought of; the epoch is one check that also # covers the ones nobody did, and it tells a zombie to stop. def fenced(self, work: Callable[[Tx], T]) -> T: mine = self._epoch
def checked(tx: Tx) -> T: if tx.query_int(FENCE, self._partition) != mine: self.drop() # a zombie learns it here, before touching a single run raise LeaseLost(self._partition) return work(tx)
return self._shard.in_transaction(checked)
def materialize( self, job_id: UUID, version: int, fire_at: datetime, next_at: datetime ) -> None: # next_at: the occurrence after fire_at, computed in the job's time zone. run_id = uuid5(job_id, fire_at.isoformat()) self.fenced( lambda tx: tx.execute( MATERIALIZE, next_at, job_id, version, fire_at, run_id, fire_at, fire_at ) )
# Forget everything held in memory: the new owner will rebuild it from the shard. def drop(self) -> None: self._epoch = -1// One transaction on the shard that holds this partition.interface IShard{ T InTransaction<T>(Func<ITx, T> work);}
interface ITx{ long? QueryLong(string sql, params object[] args); int Execute(string sql, params object[] args);}
class LeaseLostException(int partition) : Exception($"lost partition {partition}");
// Owns one partition: materializes its cron runs, reaps its expired leases and feeds its// timing wheel (not shown). Two owners WILL overlap for a while (a GC pause, a network// partition), so safety comes from the shard checking the epoch, not from the lease alone.class PartitionOwner(IShard shard, int partition, string me){ // The lease lives in the same shard as the partition's rows, and expiry is judged by // that shard's now(), so no two machines' clocks are ever compared. // Taking a lease bumps the epoch: the epoch is the fencing token. const string Acquire = """ UPDATE partition_leases SET owner = @me, epoch = epoch + 1, expires_at = now() + interval '10 seconds' WHERE partition_id = @partition AND expires_at < now() RETURNING epoch """; const string Renew = """ UPDATE partition_leases SET expires_at = now() + interval '10 seconds' WHERE partition_id = @partition AND owner = @me AND epoch = @epoch """; // FOR SHARE: a takeover's UPDATE of this row waits until our transaction commits. So // once a new epoch commits, no write made under the old epoch can commit after it. const string Fence = "SELECT epoch FROM partition_leases WHERE partition_id = @partition FOR SHARE"; // The next run of a cron job. The job's next_fire_at is the cursor: advancing it only // from the value we read creates each occurrence once, even after its finished run row // was deleted. The version check turns an edit in between into a no-op instead of a run // on the old schedule, and the derived run ID makes a retried insert a key conflict. const string Materialize = """ WITH job AS ( UPDATE jobs SET next_fire_at = @next WHERE job_id = @job AND version = @version AND next_fire_at = @at AND NOT paused RETURNING job_id, partition_id, queue, priority, max_attempts ) INSERT INTO runs (run_id, job_id, partition_id, queue, priority, max_attempts, scheduled_for, due_at, state) SELECT @run, job_id, partition_id, queue, priority, max_attempts, @at, @at, 'pending' FROM job ON CONFLICT (run_id) DO NOTHING """;
long _epoch = -1; // -1: we do not own it
public bool TryAcquire() { long? got = shard.InTransaction(tx => tx.QueryLong(Acquire, me, partition)); if (got is not long e) return false; _epoch = e; // then read the next 30 s of runs into the wheel return true; }
// Every 3 seconds, so two missed renewals still leave time before the 10 s expiry. public bool RenewLease() { bool ok = shard.InTransaction(tx => tx.Execute(Renew, partition, me, _epoch)) == 1; if (!ok) Drop(); return ok; }
// Every write the owner makes goes through here. Row conditions (pending, lease token, // job version) stop each race someone thought of; the epoch is one check that also // covers the ones nobody did, and it tells a zombie to stop. public T Fenced<T>(Func<ITx, T> work) { long mine = _epoch; return shard.InTransaction(tx => { if (tx.QueryLong(Fence, partition) != mine) { Drop(); // a zombie learns it here, before touching a single run throw new LeaseLostException(partition); } return work(tx); }); }
// nextAtUtc: the occurrence after fireAtUtc, computed in the job's time zone. public void MaterializeRun(Guid jobId, long version, DateTime fireAtUtc, DateTime nextAtUtc) { // Deterministic: the same job and fire time always give the same run ID. var hash = System.Security.Cryptography.SHA256.HashData(Encoding.UTF8.GetBytes($"{jobId}@{fireAtUtc:O}")); var runId = new Guid(hash.AsSpan(0, 16)); Fenced(tx => tx.Execute(Materialize, runId, fireAtUtc, jobId, version, nextAtUtc)); }
// Forget everything held in memory: the new owner will rebuild it from the shard. void Drop() => _epoch = -1;}// Tx is one transaction on the shard that holds this partition.type Tx interface { QueryInt(sql string, args ...any) (int64, bool, error) // false: no row Exec(sql string, args ...any) (int64, error)}
type Shard interface { InTx(work func(tx Tx) error) error}
var ErrLeaseLost = errors.New("partition lease lost")
// The lease lives in the same shard as the partition's rows, and expiry is judged by// that shard's now(), so no two machines' clocks are ever compared.// Taking a lease bumps the epoch: the epoch is the fencing token.const acquireSQL = `UPDATE partition_leasesSET owner = $1, epoch = epoch + 1, expires_at = now() + interval '10 seconds'WHERE partition_id = $2 AND expires_at < now()RETURNING epoch`
const renewSQL = `UPDATE partition_leases SET expires_at = now() + interval '10 seconds'WHERE partition_id = $1 AND owner = $2 AND epoch = $3`
// FOR SHARE: a takeover's UPDATE of this row waits until our transaction commits. So// once a new epoch commits, no write made under the old epoch can commit after it.const fenceSQL = `SELECT epoch FROM partition_leases WHERE partition_id = $1 FOR SHARE`
// The next run of a cron job. The job's next_fire_at is the cursor: advancing it only// from the value we read creates each occurrence once, even after its finished run row// was deleted. The version check turns an edit in between into a no-op instead of a run// on the old schedule, and the derived run ID makes a retried insert a key conflict.const materializeSQL = `WITH job AS ( UPDATE jobs SET next_fire_at = $5 WHERE job_id = $3 AND version = $4 AND next_fire_at = $2 AND NOT paused RETURNING job_id, partition_id, queue, priority, max_attempts)INSERT INTO runs (run_id, job_id, partition_id, queue, priority, max_attempts, scheduled_for, due_at, state)SELECT $1, job_id, partition_id, queue, priority, max_attempts, $2, $2, 'pending'FROM jobON CONFLICT (run_id) DO NOTHING`
// PartitionOwner owns one partition: it materializes the partition's cron runs, reaps its// expired leases and feeds its timing wheel (not shown). Two owners WILL overlap for a while// (a GC pause, a network partition), so safety comes from the shard checking the epoch,// not from the lease alone.type PartitionOwner struct { shard Shard partition int me string epoch int64 // -1: we do not own it}
func (o *PartitionOwner) TryAcquire() (bool, error) { var got int64 var ok bool err := o.shard.InTx(func(tx Tx) (err error) { got, ok, err = tx.QueryInt(acquireSQL, o.me, o.partition) return err }) if err != nil || !ok { return false, err } o.epoch = got // then read the next 30 s of runs into the wheel return true, nil}
// Renew runs every 3 seconds, so two missed renewals still leave time before the 10 s expiry.func (o *PartitionOwner) Renew() (bool, error) { var n int64 err := o.shard.InTx(func(tx Tx) (err error) { n, err = tx.Exec(renewSQL, o.partition, o.me, o.epoch) return err }) if err == nil && n != 1 { o.drop() } return n == 1, err}
// Fenced wraps every write the owner makes. Row conditions (pending, lease token, job// version) stop each race someone thought of; the epoch is one check that also covers// the ones nobody did, and it tells a zombie to stop.func (o *PartitionOwner) Fenced(work func(tx Tx) error) error { mine := o.epoch return o.shard.InTx(func(tx Tx) error { current, ok, err := tx.QueryInt(fenceSQL, o.partition) if err != nil { return err } if !ok || current != mine { o.drop() // a zombie learns it here, before touching a single run return ErrLeaseLost } return work(tx) })}
// Materialize creates the run for fireAt. nextAt is the occurrence after it, computed in// the job's time zone.func (o *PartitionOwner) Materialize(jobID string, version int64, fireAt, nextAt time.Time) error { at := fireAt.UTC() sum := sha256.Sum256([]byte(jobID + "@" + at.Format(time.RFC3339Nano))) runID := hex.EncodeToString(sum[:16]) // deterministic: same job and fire time, same ID return o.Fenced(func(tx Tx) error { _, err := tx.Exec(materializeSQL, runID, at, jobID, version, nextAt.UTC()) return err })}
// drop forgets everything held in memory: the new owner will rebuild it from the shard.func (o *PartitionOwner) drop() { o.epoch = -1 }// One transaction on the shard that holds this partition.interface Tx { queryInt(sql: string, ...args: unknown[]): Promise<bigint | null>; execute(sql: string, ...args: unknown[]): Promise<number>;}
interface Shard { inTransaction<T>(work: (tx: Tx) => Promise<T>): Promise<T>;}
class LeaseLost extends Error {}
// The lease lives in the same shard as the partition's rows, and expiry is judged by// that shard's now(), so no two machines' clocks are ever compared.// Taking a lease bumps the epoch: the epoch is the fencing token.const ACQUIRE = ` UPDATE partition_leases SET owner = $1, epoch = epoch + 1, expires_at = now() + interval '10 seconds' WHERE partition_id = $2 AND expires_at < now() RETURNING epoch`;const RENEW = ` UPDATE partition_leases SET expires_at = now() + interval '10 seconds' WHERE partition_id = $1 AND owner = $2 AND epoch = $3`;// FOR SHARE: a takeover's UPDATE of this row waits until our transaction commits. So// once a new epoch commits, no write made under the old epoch can commit after it.const FENCE = `SELECT epoch FROM partition_leases WHERE partition_id = $1 FOR SHARE`;// The next run of a cron job. The job's next_fire_at is the cursor: advancing it only// from the value we read creates each occurrence once, even after its finished run row// was deleted. The version check turns an edit in between into a no-op instead of a run// on the old schedule, and the derived run ID makes a retried insert a key conflict.const MATERIALIZE = ` WITH job AS ( UPDATE jobs SET next_fire_at = $5 WHERE job_id = $3 AND version = $4 AND next_fire_at = $2 AND NOT paused RETURNING job_id, partition_id, queue, priority, max_attempts ) INSERT INTO runs (run_id, job_id, partition_id, queue, priority, max_attempts, scheduled_for, due_at, state) SELECT $1, job_id, partition_id, queue, priority, max_attempts, $2, $2, 'pending' FROM job ON CONFLICT (run_id) DO NOTHING`;
// Owns one partition: materializes its cron runs, reaps its expired leases and feeds its// timing wheel (not shown). Two owners WILL overlap for a while (a GC pause, a network// partition), so safety comes from the shard checking the epoch, not from the lease alone.class PartitionOwner { private epoch = -1n; // -1: we do not own it
constructor( private readonly shard: Shard, private readonly partition: number, private readonly me: string, ) {}
async tryAcquire(): Promise<boolean> { const got = await this.shard.inTransaction((tx) => tx.queryInt(ACQUIRE, this.me, this.partition), ); if (got === null) return false; this.epoch = got; // then read the next 30 s of runs into the wheel return true; }
// Every 3 seconds, so two missed renewals still leave time before the 10 s expiry. async renew(): Promise<boolean> { const n = await this.shard.inTransaction((tx) => tx.execute(RENEW, this.partition, this.me, this.epoch), ); if (n !== 1) this.drop(); return n === 1; }
// Every write the owner makes goes through here. Row conditions (pending, lease token, // job version) stop each race someone thought of; the epoch is one check that also // covers the ones nobody did, and it tells a zombie to stop. fenced<T>(work: (tx: Tx) => Promise<T>): Promise<T> { const mine = this.epoch; return this.shard.inTransaction(async (tx) => { if ((await tx.queryInt(FENCE, this.partition)) !== mine) { this.drop(); // a zombie learns it here, before touching a single run throw new LeaseLost(`lost partition ${this.partition}`); } return work(tx); }); }
// nextAt: the occurrence after fireAt, computed in the job's time zone. async materialize(jobId: string, version: number, fireAt: Date, nextAt: Date): Promise<void> { // Deterministic: the same job and fire time always give the same run ID. const digest = createHash('sha256').update(`${jobId}@${fireAt.toISOString()}`).digest('hex'); const runId = digest.slice(0, 32); await this.fenced((tx) => tx.execute(MATERIALIZE, runId, fireAt, jobId, version, nextAt)); }
// Forget everything held in memory: the new owner will rebuild it from the shard. private drop(): void { this.epoch = -1n; }}- Color key
- Service
- Database
The fence covers writes to the shards. A zombie can still add hints to Redis, but the lease check on the run row makes them harmless.
What this guarantees. A run has at most one live lease. The store accepts a heartbeat or completion only from the current attempt. No occurrence is created twice. What it doesn’t guarantee is that the job’s code runs once. A zombie worker keeps running until its next call to the store, and the store can’t undo a charge sent in between. That is the next stress test.
Go deeperOptions for stopping a zombie1 min
| Option | How a zombie is stopped | Loses |
|---|---|---|
| One leader, lock with a TTL | It isn’t: two leaders act during the pause | Double dispatch when a pause outlasts the TTL |
| Row compare-and-set on every write | state = 'pending', lease_token = n |
Every write path must remember its condition |
| Partition epoch, plus row checks | The shard rejects transactions under an old epoch | Leases and handoff to run |
| Consensus per partition (Raft) | A deposed leader can’t commit | You are building a database |
Row checks keep the run row correct under any zombie. The epoch covers owner decisions made from stale memory, such as skipping a missed run. It tells a zombie, on its first write, to drop everything. Temporal fences its history shards the same way. Martin Kleppmann’s essay on distributed locking explains why.
Running twice, and making it harmless
Section titled “Running twice, and making it harmless”A customer was charged twice by a billing job. How did that happen, and what does each layer do about it?
A run executes twice when the scheduler can’t know whether the first execution happened. A worker
pauses past its lease, and a second attempt starts. Or a worker finishes, and crashes before its
complete arrives. No scheduler can close that second window.
The worker side of the lease decides a lot:
// The worker's side of the lease. On the server each call is one conditional update:// heartbeat: UPDATE runs SET lease_expires_at = now() + interval '30 seconds'// WHERE run_id = ? AND lease_token = ? AND state = 'leased'// complete: UPDATE runs SET state = 'succeeded'// WHERE run_id = ? AND lease_token = ? AND state = 'leased'// Zero rows means the lease is gone: the reaper handed the run to another attempt.interface LeaseApi { boolean heartbeat(UUID runId, long leaseToken); // false: lease lost boolean complete(UUID runId, long leaseToken); // false: too late, another attempt owns it void fail(UUID runId, long leaseToken, String error, boolean retryable);}
record Run(UUID runId, long leaseToken, String payload) {}
// Thrown by a handler when retrying cannot help (bad input, a deleted account).class PermanentFailure extends Exception { PermanentFailure(String message) { super(message); }}
// A handler uses runId as the idempotency key for its side effects, and passes// leaseToken to any store that can reject writes from an older attempt.interface Handler { void run(Run run, AtomicBoolean cancelled) throws Exception;}
class LeasedExecution { static final Duration LEASE = Duration.ofSeconds(30); static final Duration EVERY = Duration.ofSeconds(10); // three chances before the lease ends static final Duration MARGIN = Duration.ofSeconds(5); // stop this long before it could end
private final LeaseApi api; private final ScheduledExecutorService timer = Executors.newSingleThreadScheduledExecutor();
LeasedExecution(LeaseApi api) { this.api = api; }
void execute(Run run, Handler handler) { AtomicBoolean cancelled = new AtomicBoolean(false); // When the last confirmed heartbeat was SENT, on the monotonic clock. The server // extended the lease after that instant, so the lease lasts at least LEASE from // here. No wall clock is compared with the server's, so skew cannot fool it. AtomicLong confirmedAt = new AtomicLong(System.nanoTime());
ScheduledFuture<?> beat = timer.scheduleAtFixedRate(() -> { long sent = System.nanoTime(); try { if (api.heartbeat(run.runId(), run.leaseToken())) confirmedAt.set(sent); else cancelled.set(true); // lost: another attempt may be running already } catch (RuntimeException networkBlip) { // Keep trying; the deadline below decides when to give up. } // Checked only on ticks, so stop now if the lease could end before the next one: // better to give up a live lease than to run alongside the next attempt. if (System.nanoTime() - confirmedAt.get() + EVERY.toNanos() > LEASE.minus(MARGIN).toNanos()) cancelled.set(true); }, EVERY.toMillis(), EVERY.toMillis(), TimeUnit.MILLISECONDS);
try { handler.run(run, cancelled); // the handler checks `cancelled` between steps if (cancelled.get()) return; // its result belongs to no live attempt: drop it // false: the lease ran out at the last moment. The run will execute again, // which is exactly why handlers must be idempotent. api.complete(run.runId(), run.leaseToken()); } catch (Exception e) { // Lease lost: the handler was told to stop, and its outcome belongs to no live attempt. if (cancelled.get()) return; // Retryable unless the handler said otherwise: the server backs off with // jitter, then dead-letters after the last attempt. api.fail(run.runId(), run.leaseToken(), e.toString(), !(e instanceof PermanentFailure)); } finally { beat.cancel(false); } }}# The worker's side of the lease. On the server each call is one conditional update:# heartbeat: UPDATE runs SET lease_expires_at = now() + interval '30 seconds'# WHERE run_id = %s AND lease_token = %s AND state = 'leased'# complete: UPDATE runs SET state = 'succeeded'# WHERE run_id = %s AND lease_token = %s AND state = 'leased'# Zero rows means the lease is gone: the reaper handed the run to another attempt.class LeaseApi(Protocol): def heartbeat(self, run_id: UUID, lease_token: int) -> bool: ... # False: lease lost def complete(self, run_id: UUID, lease_token: int) -> bool: ... # False: too late def fail(self, run_id: UUID, lease_token: int, error: str, retryable: bool) -> None: ...
@dataclass(frozen=True)class Run: run_id: UUID lease_token: int payload: str
# Raised by a handler when retrying cannot help (bad input, a deleted account).class PermanentFailure(Exception): ...
# A handler uses run_id as the idempotency key for its side effects, and passes# lease_token to any store that can reject writes from an older attempt.Handler = Callable[[Run, threading.Event], None]
LEASE = 30.0 # secondsEVERY = 10.0 # three chances before the lease endsMARGIN = 5.0 # stop this long before it could end
class LeasedExecution: def __init__(self, api: LeaseApi): self._api = api
def execute(self, run: Run, handler: Handler) -> None: cancelled = threading.Event() done = threading.Event() # When the last confirmed heartbeat was SENT, on the monotonic clock. The server # extended the lease after that instant, so the lease lasts at least LEASE from # here. No wall clock is compared with the server's, so skew cannot fool it. confirmed_at = time.monotonic()
def beat() -> None: nonlocal confirmed_at while not done.wait(EVERY): sent = time.monotonic() try: if self._api.heartbeat(run.run_id, run.lease_token): confirmed_at = sent else: cancelled.set() # lost: another attempt may be running already except OSError: pass # network blip: keep trying; the deadline below decides # Checked only on ticks, so stop now if the lease could end before the next one: # better to give up a live lease than to run alongside the next attempt. if time.monotonic() - confirmed_at + EVERY > LEASE - MARGIN: cancelled.set()
threading.Thread(target=beat, daemon=True).start() try: handler(run, cancelled) # the handler checks `cancelled` between steps if cancelled.is_set(): return # its result belongs to no live attempt: drop it # False: the lease ran out at the last moment. The run will execute again, # which is exactly why handlers must be idempotent. self._api.complete(run.run_id, run.lease_token) except Exception as e: # Lease lost: the handler was told to stop, and its outcome belongs to no live attempt. if cancelled.is_set(): return # Retryable unless the handler said otherwise: the server backs off with # jitter, then dead-letters after the last attempt. retryable = not isinstance(e, PermanentFailure) self._api.fail(run.run_id, run.lease_token, repr(e), retryable) finally: done.set()// The worker's side of the lease. On the server each call is one conditional update:// heartbeat: UPDATE runs SET lease_expires_at = now() + interval '30 seconds'// WHERE run_id = @id AND lease_token = @token AND state = 'leased'// complete: UPDATE runs SET state = 'succeeded'// WHERE run_id = @id AND lease_token = @token AND state = 'leased'// Zero rows means the lease is gone: the reaper handed the run to another attempt.interface ILeaseApi{ Task<bool> HeartbeatAsync(Guid runId, long leaseToken); // false: lease lost Task<bool> CompleteAsync(Guid runId, long leaseToken); // false: too late, another attempt owns it Task FailAsync(Guid runId, long leaseToken, string error, bool retryable);}
record Run(Guid RunId, long LeaseToken, string Payload);
// Thrown by a handler when retrying cannot help (bad input, a deleted account).class PermanentFailure(string message) : Exception(message);
// A handler uses RunId as the idempotency key for its side effects, and passes// LeaseToken to any store that can reject writes from an older attempt.delegate Task Handler(Run run, CancellationToken cancelled);
class LeasedExecution(ILeaseApi api){ static readonly TimeSpan Lease = TimeSpan.FromSeconds(30); static readonly TimeSpan Every = TimeSpan.FromSeconds(10); // three chances before the lease ends static readonly TimeSpan Margin = TimeSpan.FromSeconds(5); // stop this long before it could end
public async Task ExecuteAsync(Run run, Handler handler) { using var cancelled = new CancellationTokenSource(); using var done = new CancellationTokenSource(); // When the last confirmed heartbeat was SENT, on the monotonic clock. The server // extended the lease after that instant, so the lease lasts at least Lease from // here. No wall clock is compared with the server's, so skew cannot fool it. var clock = System.Diagnostics.Stopwatch.StartNew(); var confirmedAt = TimeSpan.Zero;
async Task Beat() { using var timer = new PeriodicTimer(Every); try { while (await timer.WaitForNextTickAsync(done.Token)) { var sent = clock.Elapsed; try { if (await api.HeartbeatAsync(run.RunId, run.LeaseToken)) confirmedAt = sent; else cancelled.Cancel(); // lost: another attempt may be running already } catch (Exception e) when (e is HttpRequestException or TaskCanceledException) { // Network blip: keep trying; the deadline below decides when to give up. } // Checked only on ticks, so stop now if the lease could end before the next one: // better to give up a live lease than to run alongside the next attempt. if (clock.Elapsed - confirmedAt + Every > Lease - Margin) cancelled.Cancel(); } } catch (OperationCanceledException) { } // the run finished }
var beat = Beat(); try { await handler(run, cancelled.Token); // the handler observes the token between steps if (cancelled.IsCancellationRequested) return; // its result belongs to no live attempt // false: the lease ran out at the last moment. The run will execute again, // which is exactly why handlers must be idempotent. await api.CompleteAsync(run.RunId, run.LeaseToken); } catch (Exception e) { // Lease lost: the handler was told to stop, and its outcome belongs to no live attempt. if (cancelled.IsCancellationRequested) return; // Retryable unless the handler said otherwise: the server backs off with // jitter, then dead-letters after the last attempt. await api.FailAsync(run.RunId, run.LeaseToken, e.Message, retryable: e is not PermanentFailure); } finally { done.Cancel(); await beat; } }}// LeaseAPI is the worker's side of the lease. On the server each call is one conditional update://// heartbeat: UPDATE runs SET lease_expires_at = now() + interval '30 seconds'// WHERE run_id = $1 AND lease_token = $2 AND state = 'leased'// complete: UPDATE runs SET state = 'succeeded'// WHERE run_id = $1 AND lease_token = $2 AND state = 'leased'//// Zero rows means the lease is gone: the reaper handed the run to another attempt.type LeaseAPI interface { Heartbeat(runID string, leaseToken int64) (bool, error) // false: lease lost Complete(runID string, leaseToken int64) (bool, error) // false: too late, another attempt owns it Fail(runID string, leaseToken int64, reason string, retryable bool) error}
type Run struct { RunID string LeaseToken int64 Payload string}
// ErrPermanent is wrapped by a handler when retrying cannot help (bad input, a deleted account).var ErrPermanent = errors.New("permanent failure")
// Handler uses RunID as the idempotency key for its side effects, and passes LeaseToken to// any store that can reject writes from an older attempt. It must return when ctx is done.type Handler func(ctx context.Context, run Run) error
const ( lease = 30 * time.Second every = 10 * time.Second // three chances before the lease ends margin = 5 * time.Second // stop this long before it could end)
func Execute(api LeaseAPI, run Run, handler Handler) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() done := make(chan struct{}) defer close(done)
go func() { // When the last confirmed heartbeat was SENT. time.Now carries a monotonic reading // and time.Since uses it, so no wall clock is compared with the server's and skew // cannot fool it. The server extended the lease after that instant. confirmedAt := time.Now() tick := time.NewTicker(every) defer tick.Stop() for { select { case <-done: return case <-tick.C: sent := time.Now() ok, err := api.Heartbeat(run.RunID, run.LeaseToken) switch { case err != nil: // Network blip: keep trying; the deadline below decides when to give up. case ok: confirmedAt = sent default: cancel() // lost: another attempt may be running already } // Checked only on ticks, so stop now if the lease could end before the next one: // better to give up a live lease than to run alongside the next attempt. if time.Since(confirmedAt)+every > lease-margin { cancel() } } } }()
err := handler(ctx, run) if ctx.Err() != nil { return // lease lost: the outcome belongs to no live attempt, so drop it } if err != nil { // Retryable unless the handler said otherwise: the server backs off with // jitter, then dead-letters after the last attempt. _ = api.Fail(run.RunID, run.LeaseToken, err.Error(), !errors.Is(err, ErrPermanent)) return } // false: the lease ran out at the last moment. The run will execute again, // which is exactly why handlers must be idempotent. _, _ = api.Complete(run.RunID, run.LeaseToken)}// The worker's side of the lease. On the server each call is one conditional update:// heartbeat: UPDATE runs SET lease_expires_at = now() + interval '30 seconds'// WHERE run_id = $1 AND lease_token = $2 AND state = 'leased'// complete: UPDATE runs SET state = 'succeeded'// WHERE run_id = $1 AND lease_token = $2 AND state = 'leased'// Zero rows means the lease is gone: the reaper handed the run to another attempt.interface LeaseApi { heartbeat(runId: string, leaseToken: bigint): Promise<boolean>; // false: lease lost complete(runId: string, leaseToken: bigint): Promise<boolean>; // false: too late fail(runId: string, leaseToken: bigint, error: string, retryable: boolean): Promise<void>;}
interface Run { runId: string; leaseToken: bigint; payload: string;}
// Thrown by a handler when retrying cannot help (bad input, a deleted account).class PermanentFailure extends Error {}
// A handler uses runId as the idempotency key for its side effects, and passes// leaseToken to any store that can reject writes from an older attempt.type Handler = (run: Run, signal: AbortSignal) => Promise<void>;
const LEASE_MS = 30_000;const EVERY_MS = 10_000; // three chances before the lease endsconst MARGIN_MS = 5_000; // stop this long before it could end
async function execute(api: LeaseApi, run: Run, handler: Handler): Promise<void> { const cancelled = new AbortController(); // When the last confirmed heartbeat was SENT, on the monotonic clock. The server // extended the lease after that instant, so the lease lasts at least LEASE_MS from // here. No wall clock is compared with the server's, so skew cannot fool it. let confirmedAt = performance.now();
const beat = setInterval(async () => { const sent = performance.now(); try { if (await api.heartbeat(run.runId, run.leaseToken)) confirmedAt = sent; else cancelled.abort(); // lost: another attempt may be running already } catch { // Network blip: keep trying; the deadline below decides when to give up. } // Checked only on ticks, so stop now if the lease could end before the next one: // better to give up a live lease than to run alongside the next attempt. if (performance.now() - confirmedAt + EVERY_MS > LEASE_MS - MARGIN_MS) cancelled.abort(); }, EVERY_MS);
try { await handler(run, cancelled.signal); // the handler checks the signal between steps if (cancelled.signal.aborted) return; // its result belongs to no live attempt: drop it // false: the lease ran out at the last moment. The run will execute again, // which is exactly why handlers must be idempotent. await api.complete(run.runId, run.leaseToken); } catch (e) { // Lease lost: the handler was told to stop, and its outcome belongs to no live attempt. if (cancelled.signal.aborted) return; // Retryable unless the handler said otherwise: the server backs off with // jitter, then dead-letters after the last attempt. await api.fail(run.runId, run.leaseToken, String(e), !(e instanceof PermanentFailure)); } finally { clearInterval(beat); }}A worker whose heartbeats fail can’t learn that its lease ended. So it measures the time since the last confirmed heartbeat, on its own monotonic clock. And it stops at least 5 seconds before the lease could expire. Clock differences can’t fool that. A long pause can, so the lease token also goes to the handler.
- Color key
- Database
- Background worker
Making a repeat harmless is the handler’s job, with keys we provide. The run ID stays the same across attempts. So it is the idempotency key for side effects, as in the payment system. The lease token fences stores that can compare: “write only if no newer token was seen”.
Retries. A failed attempt goes back to pending at a random delay, up to 2^attempt seconds
(capped at 10 minutes). So runs that failed together don’t return together. After the job’s maximum
attempts, the run goes dead. It’s visible, alerted to the team, and replayable once the bug is
fixed.
One team’s flood
Section titled “One team’s flood”One team schedules 10 million one-off jobs for 9:00, and another team’s 9:00 job runs 20 minutes late. Why, and what do we change?
Leases are per queue, so the 10 million runs sit in their own slice of the index and use their own team’s workers. What is shared is shard write capacity, owners, Redis and dispatchers. A team that scales to 20,000 workers can lease fast enough to fill the shards, and that is where the other team’s lateness came from.
So three limits protect shared capacity. A lease quota per team: a token bucket on leases per
second, split across dispatchers. So a team’s burst drains at its share. Weighted lanes inside a
queue: pop high, normal and low priority in a 70, 20, 10 ratio. So low priority is slower, but never
stopped. And a cap on pending runs per team, with 429 on create past it. That stops a bug that
schedules a billion runs before anyone notices.
Strict priority, “high always before normal”, is what we avoid: a flood of high-priority runs would starve everything else forever.
Go deeperFairness techniques compared1 min
| Technique | What it does | Cost |
|---|---|---|
| Strict priority | High always before normal | A flood of high starves normal forever |
| Lease quotas per team | Token bucket per team on leases per second, in each dispatcher | A team’s own burst drains slower |
| Weighted lanes | Ready lists per (queue, lane); pop high, normal, low at 70, 20, 10 | Low priority is slower, never stopped |
| Pending-run cap | A cap on pending runs per team; 429 on create past it |
A team must ask for a bigger quota |
Each team’s lease rate is split across dispatchers like the local shares in the rate limiter. An empty lane’s share passes down to the next, so no capacity sits idle. “Priority” no longer means “always first”, which has to be explained to every team that asks for high.
Runs missed while a region was down
Section titled “Runs missed while a region was down”Say a region goes down for three hours. What happens to the 9 am jobs, and what runs when it comes back?
A three-hour outage spends most of a year’s 4.4-hour budget. And copying to another region after the fact would lose the last seconds of accepted jobs. So each shard commits synchronously to a standby in a paired region, about 10 ms away. Dispatchers, owners and workers run in both.
Stage 6: A paired region
Stage 6 of 6
A paired region
Why this stage
Not load: a region outage of a few hours spends the whole 4.4 hour yearly budget of 99.95%, and an accepted job may never be lost.
What we add
Every shard commits synchronously to a standby in a paired region about 10 ms away.
Dispatchers, owners and tenant workers run in both regions.
What it costs
About 10 ms more per commit, which batching absorbs.
Runs in flight in the failed region are re-run after their leases expire.
At 35k due runs per second, every part has headroom. Raise the traffic until something turns red.
The numbers behind the busy bars
What one run costs
- Each due run costs 3 run store writes (create, claim, finish) and 2 history events.
What one machine can do
- One Postgres primary sustains about 10,000 indexed row writes a second. Batching claims raises that to about 25,000.
- A dispatcher serves about 5,000 claims a second by polling, or 20,000 by run ID.
- A worker host runs about 500 two-second runs at once, so 250 a second.
- A Redis cluster takes about 1.5 million sorted set operations a second.
On failure, the standbys are promoted, and owners in the surviving region take the expired leases, raising every epoch. An old primary that is cut off, not dead, commits nothing more, because every commit waits for the standby it can no longer reach. Runs leased in the failed region expire and run again, which handlers absorb.
Missed runs. An owner taking over after a long gap finds cron occurrences that were due while
nobody looked. Only the team knows what a missed run means, so the job’s catchUp setting decides:
| Setting | After a 2-hour outage of an hourly job | Right for |
|---|---|---|
latest (default) |
One run now; the other missed ones skipped | Reports and syncs that read current state |
all |
Two runs, oldest first, each with its fire time | Work per period: “bill the hour 13:00 to 14:00” |
none |
Nothing until the next scheduled time | Jobs where late is worse than never |
Skipped runs are recorded as skipped, and the backlog releases at each queue’s lease rate, so
recovery doesn’t stampede the workers.
Go deeperWhy synchronous, and deadlines for late runs1 min
Synchronous commits cost about 10 ms on every write, and the pair of regions must be close. A second pair far away, for an outage across a whole continent, could only be asynchronous, and would lose seconds of state.
A starting deadline, “run only if within N minutes of due”, combines with any catchUp setting.
These mirror Kubernetes CronJob’s startingDeadlineSeconds and Quartz’s misfire rules. A run older
than its deadline is recorded as skipped, with the reason, so the team can see what didn’t happen
and why.
When things break
Section titled “When things break”An owner process crashes at 8:59:50. What happens to its partitions’ 9:00 runs?
Whether a failure makes runs late but never lost, and you can say how late.
A failure should make runs late, never lost. The owner’s partition lease is 10 seconds. So by about 9:00:00, another owner takes it, raises the epoch, loads the next 30 seconds, and re-adds everything overdue. Those runs start up to 15 seconds late. If Redis is down too, dispatchers fall back to the SKIP LOCKED poll.
If the owner was only paused, not dead, it wakes up still believing it owns those partitions. The fencing token from the pause stress test stops it: its writes carry the old epoch, and the shard rejects them.
Every decision uses the shard’s clock: due times, lease expiry, partition leases. The timing wheel
fires on the owner’s clock, but the lease check re-tests due_at <= now() on the shard. So an owner
50 ms fast makes a hint wait 50 ms, never an early run.
Click any box to break it and see what users would notice:
Break it: what fails, who notices, what users see
Failure: click any box to break it
A timer owner goes down
What breaks
Its partitions stop pushing due runs, materializing cron runs and reaping leases.
How we notice
Its lease rows stop renewing; oldest-overdue-run age rises on those partitions.
What users see
Runs in those partitions are up to about 15 seconds late, then catch up.
Nothing is lost: the rows are still pending.
What we do
Another owner takes each expired lease, bumping its epoch, reads the window and pushes everything overdue.
A zombie owner’s writes fail the epoch check.
Go deeperEvery failure, and what users see1 min
| Part | What users see | What we do |
|---|---|---|
| Job API | Creates fail; schedules keep firing | Stateless; clients retry with idempotency keys |
| Dispatcher | Long-polls drop; workers reconnect | No state: leases live in the store |
| Worker | Its runs retry after 30 s; possible duplicates | The reaper re-queues with a delay; handlers absorb repeats |
| Timer owner | Its partitions up to 15 s late | A new owner takes the lease, raises the epoch |
| Redis ready lists | Runs late by seconds | Owners re-add; dispatchers fall back to polling |
| Run store shard | 1/64 of jobs up to 30 s late | Promote the synchronous standby; nothing accepted is lost |
| Kafka or history | Console history stale | Outbox rows wait in the shard |
| Region | In-flight runs retry; others about 1 min late | Promote paired standbys; owners there take the leases |
| Operation | Guarantee |
|---|---|
| Create or edit a job | Committed on the shard and its synchronous standby |
| Lease a run | One live lease at a time, by the row check |
| Execution | At-least-once |
| Due time | Never early by the shard’s clock; p99 5 s late |
| Pause | Immediate for runs not yet leased |
| Run history | Seconds behind |
Under stress, protect in this order: dispatching, heartbeats and completions, then creates, then history and the console.
Go deeperRunning it: alerts, safety, deploys and cost1 min
Wake someone up when:
- p99 lateness per partition passes 5 seconds, for 5 minutes.
- The oldest overdue run in any partition is over 30 seconds old. (It catches an unowned partition that averages hide.)
- Leases expiring pass 0.01% of leases, the duplicate budget.
- A partition has no owner for 30 seconds.
Safety. Services touch only their own queues. Payloads are capped at 64 KB and point to a secrets manager instead of carrying secrets, since they sit in history for 30 days. The webhook adapter reaches only allowed addresses, so a job can’t reach the cloud metadata service.
Time zones. On the spring-forward night, 02:30 doesn’t exist, so the run happens at 03:00. On the fall-back night, 01:30 happens twice, so the run happens once, at the first. Cron libraries disagree on both, so we pick one rule and test it on those dates.
Deploys. Owners release their leases on shutdown, so partitions move in milliseconds. Moving a partition to another shard copies it, then cuts over by raising its epoch, which fences anything still writing to the old shard.
Cost is shard primaries and their standbys, sized for write rate, then history. The teams’ worker fleets cost far more, and are theirs.
Wrap up
Section titled “Wrap up”The whole design
Section titled “The whole design”Here is everything we built, in the order we built it. Use the buttons to step through it.
The full design, stage by stage
Stage 6 of 6
A paired region
Why this stage
Not load: a region outage of a few hours spends the whole 4.4 hour yearly budget of 99.95%, and an accepted job may never be lost.
What we add
Every shard commits synchronously to a standby in a paired region about 10 ms away.
Dispatchers, owners and tenant workers run in both regions.
What it costs
About 10 ms more per commit, which batching absorbs.
Runs in flight in the failed region are re-run after their leases expire.
At 35k due runs per second, every part has headroom. Raise the traffic until something turns red.
The numbers behind the busy bars
What one run costs
- Each due run costs 3 run store writes (create, claim, finish) and 2 history events.
What one machine can do
- One Postgres primary sustains about 10,000 indexed row writes a second. Batching claims raises that to about 25,000.
- A dispatcher serves about 5,000 claims a second by polling, or 20,000 by run ID.
- A worker host runs about 500 two-second runs at once, so 250 a second.
- A Redis cluster takes about 1.5 million sorted set operations a second.
The decisions, and what each one costs
Section titled “The decisions, and what each one costs”| Decision | We chose | We gave up | We’d choose differently if |
|---|---|---|---|
| Execution | Teams’ workers pull with long-polls | Running code for teams | Teams wanted serverless execution |
| Delivery | At-least-once, leases with heartbeats | Any claim of exactly-once | Never; idempotency is the answer |
| Store | Postgres, 256 partitions by job ID | Effortless horizontal writes | Write rate dwarfed everything (bucketed Cassandra) |
| Finding due runs | SKIP LOCKED polling, then owners and wheels | Simplicity at ten times the load | Load stayed under 10,000 leases a second per shard |
| One lease per run | Row checks plus a fenced partition epoch | A design with no leases | One leader were enough (a few thousand runs a second) |
| Fairness | Quotas, weighted lanes, a pending cap | Strict priority | One team only |
| Regions | Synchronous standby in a paired region | 10 ms per commit | Losing seconds of state on a region loss were fine |
How to spend 45 minutes
Section titled “How to spend 45 minutes”- 0 to 4. Workers, scale, lateness, duplicates, missed runs.
- 4 to 8. 35,000 runs a second at the peak, 100,000 store writes, a small live set.
- 8 to 20. Build it: jobs and runs, the lease and the SKIP LOCKED statement, partitions by job ID, history. A mid-level answer can stop here.
- 20 to 40. The stress tests: the top of the hour with owners and timing wheels. The zombie owner and fencing. Running twice, and idempotency. Fairness. Then the paired region.
- 40 to 45. An owner crash, then your own questions.
Never skip the lease, the fencing token, or the answer to “exactly once”.
Follow-up questions
Section titled “Follow-up questions”Interviewer asksAn hourly job sometimes takes 90 minutes. What happens at the next hour?
The job’s overlap setting decides, like Kubernetes CronJob’s concurrencyPolicy. allow runs
both. skip records the new one as skipped. replace cancels the old run, through its next
heartbeat response.
Interviewer asksWhy not use a message queue with delayed delivery instead of building this?
For short delays it is fine: SQS delays up to 15 minutes. It stops at the horizon (next Tuesday), at cron (someone must create the next run) and at state (pause, edit, catch-up). Use a queue as the last hop, as our ready lists are, with a durable store as the truth.
Go deeperMore follow-up questions1 min
A team edits a schedule from hourly to daily at 10:59:59. Does the 11:00 run happen? The edit replaces the pending 11:00 run in the same transaction. If it was already leased, it runs. An owner that read the old version finds the version changed and writes nothing.
A team needs runs to start within 50 ms of their due time. Can you do it? With owners and wheels, dispatch takes about 11 ms, so yes, if their workers hold a long-poll open. Polling every 250 ms can’t, so the owners would come early for them.
Why not one heartbeat per worker instead of per run? It’s cheaper, and right at very high concurrency. But a live process keeps heartbeating while one handler is stuck. Per-run heartbeats, sent by the handler, catch that.
How do you test that a run is never handed to two workers? Fault injection. Run many dispatchers and owners against one shard, with pauses longer than the lease, killed processes and dropped heartbeats. Meanwhile, a checker records every run and lease token handed out. The pass rule: one accepted completion per run, and none starts early.
Further reading
Section titled “Further reading”- On this site: the LLD task scheduler (the in-process heap), partitioning, consensus, the notification system (dead letters) and the payment system (idempotency keys).
- How to do distributed locking, Martin Kleppmann: why a lease needs a fencing token checked by the resource.
- PostgreSQL SELECT, the locking
clause: what
FOR UPDATE,FOR SHAREandSKIP LOCKEDpromise. - Apache Kafka, Purgatory, and Hierarchical Timing Wheels, Confluent: timing wheels for millions of timers.
The designbeyondcode digest, every Saturday
New deep dives, interview tips and site news in one short email. Free. Leave with one click.
We never sell or share your email. We only use it to send this newsletter.