Design a distributed job scheduler with exactly-once semantics

"Design a scheduler that runs a million scheduled jobs a day across a fleet, and guarantees each job runs exactly once."

Step 1: clarify, and correct the premise (4 minutes)

The first move is to challenge "exactly once", carefully and without being pedantic, because the entire design follows from what it actually means here.

"I want to be precise about exactly-once, because there are two versions of it and only one is achievable. Exactly-once delivery is impossible in a distributed system: if I dispatch a job and the worker dies before acknowledging, I cannot distinguish 'it never ran' from 'it ran and the ack was lost', so I must either re-dispatch (risking a second run) or not (risking zero runs). What I can build is exactly-once effect: at-least-once dispatch, plus idempotent execution, so re-running produces the same outcome. Is that what you need, or is there a job whose side effect genuinely cannot be made idempotent?"

That question is the answer to the drill. Everything after it is engineering.

The other clarifications:

Scale         1M jobs/day = ~12/sec average, but scheduling is BURSTY:
              a large fraction fire at :00 of the hour and at midnight.
              Assume 50,000 jobs firing within one second at peak.

Job types     Cron-style recurring, one-shot delayed, and immediate.
              Assume all three.

Timing SLA    "Within 5 seconds of scheduled time" for p99.
              This rules out a polling interval of 60 seconds.

Duration      Seconds to hours. Long jobs change the lease design.

Failure       Retries with backoff, a maximum attempt count, and a
              dead-letter destination.

Step 2: capacity math (3 minutes)

Steady state:  1M/day       = 12 jobs/sec
Peak burst:    50,000 in 1 second at the top of the hour

Scheduler read pattern
  "Which jobs are due in the next N seconds?"
  A range scan on (next_run_at) with a covering index.
  At 50k due in one second, that is one scan returning 50k rows.

Storage
  10M job definitions x ~1 KB = 10 GB. Trivial.
  Execution history: 1M/day x 90 days x 500 bytes = 45 GB. Also fine.
  -> One Postgres instance handles this. Do NOT reach for a
     distributed database; say so and say why.

Workers
  Average job 10 s, 12 jobs/sec -> 120 concurrent. 
  Peak burst 50,000 jobs -> queue drains over ~40 s at 1,200 workers,
  which meets a 5-second SLA only if we PRE-SCALE for known bursts.
  This is a real design consequence of hourly cron alignment.

The observation worth volunteering: at one million jobs a day, this is a single-Postgres problem, not a distributed-database problem. Reaching for Cassandra or DynamoDB here adds partition-key design, eventual consistency and a much harder locking story to solve a problem that fits on one machine with room to spare. Saying that demonstrates more judgement than designing something impressive.

Step 3: architecture

                  ┌──────────────────────┐
   API ─────────► │  job definitions     │  Postgres
                  │  (schedule, payload) │
                  └──────────┬───────────┘
                             │
                  ┌──────────▼───────────┐
                  │  SCHEDULER (N nodes) │  leader-elected or
                  │  poll for due jobs   │  partitioned by hash
                  └──────────┬───────────┘
                             │  enqueue with idempotency key
                  ┌──────────▼───────────┐
                  │  QUEUE (SQS / Kafka) │
                  └──────────┬───────────┘
                             │
                  ┌──────────▼───────────┐
                  │  WORKERS (autoscaled)│
                  │  lease -> run -> ack │
                  └──────────┬───────────┘
                             │
                  ┌──────────▼───────────┐
                  │  execution records   │  Postgres, unique on
                  │  (the dedupe table)  │  (job_id, scheduled_for)
                  └──────────────────────┘

The separation is deliberate: the scheduler decides what should run and the workers decide how. A scheduler that also executes cannot scale the two independently, and execution is the part with variable resource needs.

Step 4: the scheduler, and how it avoids double-dispatch

The naive loop double-dispatches whenever two scheduler nodes poll simultaneously. Two mechanisms fix it, and the second is the one that makes it robust.

Mechanism 1: claim rows atomically with SKIP LOCKED.

-- Each scheduler node claims a distinct batch. SKIP LOCKED is what
-- makes concurrent pollers not block each other and not collide.
WITH due AS (
  SELECT id
  FROM jobs
  WHERE state = 'scheduled'
    AND next_run_at <= now() + interval '5 seconds'
  ORDER BY next_run_at
  LIMIT 500
  FOR UPDATE SKIP LOCKED          -- the whole trick
)
UPDATE jobs j
SET state = 'dispatching', claimed_at = now(), claimed_by = $1
FROM due
WHERE j.id = due.id
RETURNING j.id, j.payload, j.next_run_at;

FOR UPDATE SKIP LOCKED is the single most useful primitive in queue-on-a-database designs: a concurrent transaction skips locked rows instead of waiting, so N pollers get N disjoint batches with no coordination.

Mechanism 2: an idempotency key derived from identity plus scheduled time.

# NOT a uuid4. The key must be derivable from the job's identity and
# its scheduled instant, so a re-dispatch of the SAME occurrence
# produces the SAME key.
idempotency_key = f"{job_id}:{scheduled_for.isoformat()}"

This is the load-bearing decision. A retry of the 09:00 occurrence generates the same key; the 10:00 occurrence generates a different one. Deduplication then works across the scheduler crashing, the queue redelivering, and a worker retrying.

Why both are needed: SKIP LOCKED prevents two schedulers dispatching simultaneously. It does not prevent a scheduler crashing after enqueueing but before marking the row dispatched, so the next poll re-dispatches. The idempotency key covers that case. Together they give at-least-once dispatch with exactly-once effect, which is the achievable target from step 1.

Step 5: the worker, and the deduplication table

def execute(message) -> None:
    key = message.idempotency_key
    try:
        # INSERT is the dedupe. The unique constraint is the entire
        # mechanism; nothing else is trusted.
        with db.transaction():
            db.execute(
                "INSERT INTO executions (idempotency_key, job_id, "
                "scheduled_for, state, worker_id, lease_until) "
                "VALUES (%s, %s, %s, 'running', %s, now() + interval '60 s')",
                (key, message.job_id, message.scheduled_for, WORKER_ID),
            )
    except UniqueViolation:
        # Someone else has this occurrence. Two sub-cases, and they
        # differ: a healthy holder means we ack and move on; an expired
        # lease means the holder died and we may take over.
        row = db.fetch_one(
            "SELECT state, lease_until FROM executions "
            "WHERE idempotency_key = %s", (key,))
        if row.state == 'succeeded':
            ack(message); return
        if row.state == 'running' and row.lease_until > now():
            ack(message); return          # healthy holder, not our work
        if not try_steal_lease(key):
            ack(message); return
        # else: we now own it, fall through and run

    try:
        run_job(message.payload, heartbeat=lambda: extend_lease(key))
        db.execute("UPDATE executions SET state='succeeded', "
                   "finished_at=now() WHERE idempotency_key=%s", (key,))
        ack(message)
    except Exception as exc:
        db.execute("UPDATE executions SET state='failed', "
                   "error=%s WHERE idempotency_key=%s", (str(exc), key))
        nack(message)      # let the queue's retry policy handle backoff

The unique constraint on idempotency_key is the whole mechanism. Everything else is bookkeeping. If two workers race, the database picks one, deterministically, and the loser learns it lost by catching the violation.

The lease, and why heartbeats matter

A job that takes 40 minutes cannot hold a 60-second lease, and it cannot hold a 40-minute lease either, because then a worker that dies at minute 2 blocks the job for 38 more.

# Short lease, extended by a heartbeat from the running job.
# Dead worker -> heartbeats stop -> lease expires in 60 s -> takeover.
# Live long job -> heartbeats continue -> lease never expires.
def extend_lease(key: str) -> None:
    updated = db.execute(
        "UPDATE executions SET lease_until = now() + interval '60 s' "
        "WHERE idempotency_key = %s AND worker_id = %s AND state='running'",
        (key, WORKER_ID))
    if updated == 0:
        # We LOST the lease: someone took over while we were working.
        # Abort immediately rather than finishing and double-writing.
        raise LeaseLost(key)

The LeaseLost branch is the part people omit and it is where correctness actually lives. A worker that was paused (GC, a long syscall, a VM migration) can wake up believing it still owns the job while a takeover has already happened. Checking the lease on every heartbeat and aborting on loss is what prevents two concurrent executions.

This is a fencing problem, and the fully-correct version adds a monotonically increasing fence token that the downstream side-effect target checks, which is Kleppmann's argument about why leases alone are insufficient. Worth naming; usually not built, because it requires the downstream system to participate.

Step 6: making effects idempotent

Deduplication protects against re-dispatch. It cannot protect against a job that succeeded, wrote its side effect, and then failed to record success. The side effect itself must be idempotent, and there are three ways.

-- 1. Natural idempotency: upsert on a key derived from the occurrence.
INSERT INTO daily_report (report_date, content)
VALUES ($1, $2)
ON CONFLICT (report_date) DO UPDATE SET content = EXCLUDED.content;

# 2. Pass the idempotency key downstream. Most payment and messaging
#    APIs accept one, and this is the correct use of that feature.
stripe.Charge.create(amount=..., idempotency_key=key)
sendgrid.send(msg, custom_args={"dedup_key": key})

-- 3. Transactional outbox: side effect and completion in ONE
--    transaction, with a relay publishing from the outbox.
BEGIN;
  UPDATE executions SET state='succeeded' WHERE idempotency_key=$1;
  INSERT INTO outbox (topic, payload, dedup_key) VALUES (...);
COMMIT;

Option 3 is the one that closes the gap, because it removes the window between "the effect happened" and "we recorded that it happened". If both are in one local transaction, there is no window.

And the honest limitation to state: if the side effect is a call to a third-party system with no idempotency key and no way to query whether the effect already occurred, exactly-once effect is not achievable and the correct answer is to say so and choose which failure you prefer. Most such jobs should be at-most-once (do not retry) with an alert, because a duplicate charge is usually worse than a missing one that a human resolves.

Step 7: failure modes

Scheduler crashes after enqueue, before marking dispatched
  -> Next poll re-dispatches. Same idempotency key. Worker dedupes.
     This is the designed-for case.

Worker dies mid-job
  -> Heartbeats stop, lease expires in 60 s, another worker takes over.
     Job re-runs from the start unless it has internal checkpointing.

Worker pauses (GC, VM migration) and resumes after takeover
  -> Next heartbeat returns 0 rows updated, LeaseLost raised, worker
     aborts. This is the subtle one.

Queue redelivers a message
  -> Idempotency key already in executions. Ack and move on.

Postgres failover
  -> Schedulers and workers reconnect. In-flight claims roll back
     (they were uncommitted), so those jobs are re-polled. Correct.

Clock skew between scheduler nodes
  -> Use the DATABASE clock (now()) for all time comparisons, never
     the node's clock. This eliminates skew as a variable entirely.

Thundering herd at :00
  -> Jitter: store next_run_at with a deterministic per-job offset
     within the schedule's tolerance, so a million hourly jobs spread
     over 60 seconds instead of arriving in one.

A job whose previous run has not finished
  -> Explicit policy per job: skip, queue, or run concurrently.
     Defaulting to "run concurrently" surprises people; default to
     skip and make it configurable.

Using the database clock is the cheapest correctness win in the whole design. Clock skew between scheduler nodes is a real source of double-dispatch and near-misses, and routing every time comparison through now() on one database removes it as a category.

Step 8: what changes at ten times the scale

At 10 million jobs a day (roughly 120/sec, peaks near 500,000 in a burst):

Polling one table stops working. The fix is partitioning: shard job rows by hash(job_id) % N, and assign partitions to scheduler nodes via a consistent-hash ring with leader election (etcd, ZooKeeper, or a database advisory lock per partition). Each node polls only its partitions, so the scan cost per node stays flat.

Timer wheels replace polling for near-term jobs. Polling every second for jobs due in the next 5 seconds is wasteful at scale. A hierarchical timing wheel in memory, hydrated from the database for the next few minutes, gives sub-second precision with no polling, and the database becomes the durable backing store rather than the hot path. This is what Kafka's DelayedOperationPurgatory and Netflix's Timer service do.

The executions table needs partitioning and aggressive retention. 10M rows a day is 3.6 billion a year. Partition by day, drop partitions past the retention window, and keep the unique index on idempotency_key scoped to the recent partitions, since dedupe only matters within the retry window.

The burst problem gets worse, not better. Hourly-aligned cron means 500,000 jobs firing in one second. The answer is scheduling jitter as a first-class feature plus predictive pre-scaling of the worker fleet, because reactive autoscaling has minutes of lag and the burst is over in seconds.

Production evidence

Quartz Scheduler (JVM) uses database row locking for cluster coordination, which is the same SKIP LOCKED pattern, and its documentation on misfire policies is a good catalogue of the "previous run has not finished" question.

Airflow's scheduler moved to SELECT ... FOR UPDATE SKIP LOCKED for exactly this reason (multiple schedulers claiming disjoint task batches without blocking each other), and its scheduler HA design documentation describes the resulting guarantees.

Temporal and its predecessor Cadence are the reference systems for durable execution: they persist the workflow's event history so a worker crash resumes from the last completed step rather than from the beginning, which is the stronger version of what the lease-and-retry design here achieves.

AWS EventBridge Scheduler publishes an at-least-once delivery guarantee explicitly, which is direct vendor evidence for the framing in step 1: a managed service operating at enormous scale does not claim exactly-once delivery, because it is not available.

Kleppmann's "How to do distributed locking" (2016) is the canonical argument that a lease alone does not prevent two concurrent holders (a paused process can wake up believing it still holds the lock), and that fencing tokens checked by the downstream resource are the complete fix.

Chris Richardson's transactional outbox pattern (microservices.io) is the standard answer to the effect-recorded-but-not-marked-complete window.

The debate

The case for a database-backed scheduler: transactions, unique constraints and SKIP LOCKED give you correctness primitives for free, the operational story is one system you already run, and at a million jobs a day it fits comfortably on one instance. Debugging is a SELECT.

The case for a dedicated system (Temporal, Cadence): durable execution is qualitatively stronger. A crashed worker resumes from the last completed step rather than re-running a two-hour job from the start, and multi-step workflows with compensation get first-class support instead of being hand-rolled.

The case for a managed service (EventBridge Scheduler, Cloud Scheduler): no scheduler to operate at all. The trade is less control over dispatch semantics and a hard dependency on the provider.

My position: Postgres with SKIP LOCKED and an idempotency-keyed executions table, until the jobs themselves become multi-step workflows. At a million jobs a day the database approach is simpler, cheaper and easier to debug, and the correctness comes from a unique constraint rather than from a protocol I have to reason about. The moment jobs become "call three services and compensate if the third fails", that is a workflow engine's problem and hand-rolling it is how teams end up with a worse Temporal.

The framing I would not compromise on is correcting "exactly once" at the start. It is not pedantry: if the interviewer believes exactly-once delivery is achievable, every subsequent design decision is being evaluated against an impossible standard, and the candidate who quietly designs at-least-once-plus-idempotency without naming it looks like they missed the requirement rather than met it. The correction takes fifteen seconds and it reframes the entire conversation.

The design decision I would defend hardest is the idempotency key being derived from job identity plus scheduled instant rather than generated per dispatch. A uuid4 per dispatch attempt provides no deduplication at all, which is a mistake that looks correct in code review and fails only under the exact conditions the system was built for.

And the LeaseLost check on every heartbeat. A lease that is only checked at acquisition does not prevent two concurrent executions, because a paused worker wakes up believing it still owns the job. That is the failure that produces duplicate charges and double emails, and it is the one most implementations miss.

Follow-up Q&A

"Can you actually guarantee exactly-once?" Not delivery, and I would say so first. If I dispatch and the worker dies before acknowledging, I cannot distinguish "never ran" from "ran and the ack was lost", so I must either re-dispatch and risk a duplicate or not and risk zero runs. What I can build is exactly-once effect: at-least-once dispatch plus idempotent execution, so re-running produces the same outcome. The mechanism is an idempotency key derived from job identity plus the scheduled instant, and a unique constraint that makes the database arbitrate races.

"Why derive the key rather than generate one?" Because a uuid4 per dispatch attempt deduplicates nothing: the retry gets a different key and runs again. The key has to be the same for the same occurrence and different across occurrences, so job_id:scheduled_for is exactly right. The 09:00 run and the 10:00 run differ; a retry of the 09:00 run does not. This looks like a detail and it is the entire mechanism.

"What stops two schedulers dispatching the same job?" SELECT ... FOR UPDATE SKIP LOCKED, which lets concurrent pollers claim disjoint batches without blocking each other. But that alone is not sufficient, because a scheduler can crash after enqueueing and before marking the row dispatched, and the next poll will re-dispatch. That is covered by the idempotency key at the worker. Both mechanisms exist because they cover different failures.

"How do you handle a job that runs for two hours?" A short lease, sixty seconds, extended by a heartbeat from the running job. A dead worker stops heartbeating and the lease expires, so takeover happens in about a minute. A healthy long job keeps heartbeating and the lease never expires. The critical part is that the heartbeat is a conditional update on worker ownership, and if it updates zero rows the worker has lost the lease and must abort immediately rather than finishing, because a takeover has already happened.

"Why would a worker lose a lease while it's still alive?" Because it was paused: a long garbage collection, a blocked syscall, a VM migration. It stops heartbeating for long enough that the lease expires, another worker takes over, and then the original wakes up believing it still owns the job. This is the case Kleppmann's distributed locking article is about, and it is why the check has to be on every heartbeat rather than only at acquisition. The complete fix is a fencing token that the downstream resource validates, which requires that resource to participate.

"A million hourly jobs all fire at :00. What happens?" A thundering herd, and reactive autoscaling cannot help because it has minutes of lag and the burst lasts seconds. Two answers. Deterministic jitter, storing next_run_at with a per-job offset inside the schedule's tolerance so the burst spreads over a minute. And predictive pre-scaling for known peaks, since the schedule is knowable in advance, which is unusual and worth exploiting.

"What if the side effect can't be made idempotent?" Then exactly-once effect is not achievable and I would say so rather than pretend. The choices are at-least-once with possible duplicates or at-most-once with possible misses, and it is a product decision. For a payment or an email I would default to at-most-once with an alert, because a duplicate charge is usually worse than a missing one a human resolves. Where the downstream accepts an idempotency key, which most payment and messaging APIs do, passing it through is the correct use of that feature.

"Would you use Postgres or a distributed database?" Postgres, and I would say why unprompted: a million jobs a day is twelve per second and ten gigabytes of definitions. That fits one instance with enormous headroom, and it buys transactions, unique constraints and SKIP LOCKED, which are exactly the primitives this problem needs. Reaching for Cassandra or DynamoDB adds partition-key design, eventual consistency and a much harder locking story to solve a problem that does not exist yet. At ten million a day I would partition the job table by hash and assign partitions to scheduler nodes, which is still Postgres.

"When would you use Temporal instead?" When the jobs become multi-step workflows. The design here re-runs a failed job from the start; durable execution resumes from the last completed step, which matters a lot for a two-hour job that failed at minute ninety. And once you need compensation logic across several service calls, hand-rolling it produces a worse version of a workflow engine. Single-step scheduled jobs do not need that and the database approach is simpler to operate and debug.

Common misconceptions

"Exactly-once delivery is achievable with enough care." It is not, and the useful target is exactly-once effect via at-least-once delivery plus idempotency.

"A UUID is an idempotency key." Only if it is derived from the occurrence. A fresh UUID per attempt deduplicates nothing.

"A lease prevents concurrent execution." It prevents concurrent acquisition. A paused process can wake up believing it still holds one, which is why the heartbeat must verify ownership and abort on loss.

"Kafka's exactly-once solves this." Kafka's transactional guarantees cover read-process-write within Kafka. A job whose effect is an HTTP call or a database write outside Kafka is not covered.

"You need a distributed database for a million jobs a day." Twelve per second fits on one Postgres instance with room to spare, and the relational primitives are what make the correctness easy.

Interview delivery note

Open by correcting the premise, in a way that is collaborative rather than pedantic, because it reframes everything that follows: "Before I design this I want to be precise about exactly-once, because there are two versions and only one is achievable. Exactly-once delivery is impossible: if a worker dies before acknowledging I can't tell 'never ran' from 'ran and the ack was lost'. What I can build is exactly-once effect, which is at-least-once dispatch plus idempotent execution."

Then give the two mechanisms and be explicit that they cover different failures: "SELECT FOR UPDATE SKIP LOCKED so concurrent schedulers claim disjoint batches without blocking. And an idempotency key at the worker, derived from job id plus scheduled instant, with a unique constraint. Both, because SKIP LOCKED doesn't cover a scheduler crashing after enqueue and before marking dispatched."

Volunteer the lease subtlety, because it is the depth signal here: "The part that's easy to get wrong is the lease. A short lease extended by a heartbeat handles both a dead worker and a two-hour job. But the heartbeat has to be a conditional update on ownership, and if it updates zero rows the worker has lost the lease and must abort immediately. Otherwise a worker that was paused by a long GC wakes up and finishes a job someone else has already taken over."

And show scale judgement, which is often what actually separates candidates here: "I'd also say that a million jobs a day is twelve per second and about ten gigabytes. That's one Postgres instance with enormous headroom, and it gives me transactions and unique constraints, which are exactly the primitives this problem needs. I'd resist reaching for a distributed database to solve a problem I don't have."

The Staff & Team Lead Interview Reference