Design a distributed job scheduler

Questions: one-off jobs, recurring jobs, or both? How precise must the start time be (within a second, or within a minute)? What does a job do: call an HTTP endpoint, or run code? How long do jobs run? What if the system is down at the scheduled moment: run late, skip, or catch up every missed run? Must a job never run twice? Do owners need to cancel, pause and see history? Are there time zones in the schedules?

Assumed scope: register, update and cancel jobs; fire them on time by calling a webhook or putting a message on a queue; retries; run history. Out of scope: executing arbitrary user code, dependencies between jobs, and a UI.

Planning figures for the exercise, not facts about any real scheduler.

• Registry: 100 million jobs x 300 bytes = 30 GB. A single volume holds it.
• Average load: 10,000,000 / 86,400 = about 116 executions a second. At three row writes each (claim, outbox insert, complete), about 350 writes a second. Trivial.
• The burst is the number: suppose 1 million jobs are due at midnight and each must start within 60 seconds. 1,000,000 / 60 = about 16,700 starts a second, and 50,000 row writes a second with claim, outbox and complete.
• Primary limit: a single primary commits about 5,000 writes a second. 3,000,000 writes / 5,000 = 600 seconds, ten minutes late. For the 60-second target you need at least 50,000 / 5,000 = 10 partitions of writes. Use 16 for headroom: 16 x 5,000 = 80,000 writes a second, and 3,000,000 / 80,000 = about 38 seconds to drain.
• Workers: at 16,700 starts a second and an assumed 200 ms per callback, in-flight calls = 16,700 x 0.2 = about 3,300. Spread over 20 worker processes, that is about 170 concurrent calls each.

What the numbers force: the average says one machine, the burst says partition the schedule table and put a queue between "due" and "running". Round-time spikes are also worth smoothing by letting owners tolerate a small random delay.

1. API and registry. POST /jobs takes a target (webhook URL or queue topic), a payload, either run_at or a cron expression with a time zone id, a retry policy, and a client Idempotency-Key. Rows live in jobs(job_id, shard, owner, schedule, next_run_at, state, lease_token, leased_until, attempt).
2. Sharding. shard = hash(job_id) mod 16, each shard its own relational primary with replicas. The index that matters is (state, next_run_at) on each shard, so "what is due" is a short range scan, not a table scan. Keep the shard count a multiple so you can split later by moving whole logical shards.
3. Pollers. Several pollers per shard run a short loop: select up to 500 rows with state = 'scheduled' AND next_run_at <= now(), ordered by next_run_at, using a row-locking read that skips rows other pollers hold (FOR UPDATE SKIP LOCKED in Postgres). In the same transaction set state = 'claimed', leased_until = now + 60 s, and a new lease_token, and write the due runs to an outbox. Because claims skip locked rows, no leader is needed for correctness, and more pollers add throughput.
4. Dispatch queue. A relay moves claimed runs to a partitioned queue (one partition set per shard). Workers pull from it, call the target with the header Idempotency-Key: job_id:scheduled_time, and heartbeat to extend the lease every 20 seconds while the call is in flight.
5. Completion. On success the worker runs UPDATE ... WHERE job_id = ? AND lease_token = ? to mark the run done. For a recurring job the same transaction computes the next next_run_at from the scheduled time, not from now, so jitter never accumulates as drift.
6. Failure. If a worker dies, the lease expires and a janitor poller finds rows with state = 'claimed' AND leased_until < now() and re-queues them with an incremented attempt. A late completion from the old worker fails the token check. Retries use exponential backoff with jitter, a cap on attempts, and a dead-letter state the owner can inspect and replay.
7. Time zones. Store the schedule as expression plus zone id and compute the next instant in UTC with a maintained time zone library. A daylight-saving gap or repeat needs an explicit rule: skip, run once, or run at the first valid instant.
8. Missed runs. Per job policy: run once on recovery (the default), run every missed occurrence, or skip. Choose per job because billing wants "run once" and a report might want "skip".

Hard problem one: exactly once is not available. The system gives at-least-once start and relies on the target to de-duplicate with the idempotency key. What it can promise is that state transitions are fenced by the lease token, and that a run is started twice only when a lease expires before completion is recorded: the first attempt was slow, or the worker died after the target had already succeeded. Say this plainly; it is what the interviewer is waiting for.

Hard problem two: the burst. Partitioned claims, batch claiming and an outbox remove the single hot table. Because the burst is a thundering herd on the targets as well, add a per-target rate limit and a concurrency cap in the workers, and retry in only one layer so three attempts in each of three layers does not become 27 calls.

• Database polling vs a delay queue or timer wheel in memory. Polling an indexed table is durable, easy to inspect and survives restarts, but adds up to the poll interval in lateness. In-memory timers are precise but lose state on a crash unless rebuilt from storage. Pick polling every one to two seconds. Flip to a hybrid (poll the next minute into memory timers) when second-level precision is required.
• Leaderless claiming vs a single leader per shard. Row-level claims need no election and scale by adding pollers, but contend on hot index ranges. A leader per shard avoids contention but needs election and a fencing story. Pick claims. Flip to a leader only when claim contention measurably limits throughput.
• At-least-once with idempotent targets vs trying for exactly once. At-least-once is achievable and honest. Exactly once only holds inside one transactional boundary. Pick at-least-once, and make the idempotency key part of the contract. Flip to a transactional outbox into the target's own database for targets that cannot de-duplicate.
• Relational shards vs a distributed key-value store. Relational shards give transactions for the claim-and-enqueue step and simple range scans. A key-value store scales further with less operations effort per shard but makes the due-time range scan and conditional updates harder. Pick relational at this size. Flip when the registry passes what 16 primaries can hold.

• Sizing from the average (116 a second) and never looking at the round-time burst.
• One cron-like process on one machine, with no answer for its failure, or two copies that both fire.
• Scanning the whole jobs table for due work instead of using an index on the next run time.
• Promising exactly-once execution without saying what the target must do.
• Computing the next run from the time the last run finished, so schedules drift.
• Using local time without a zone id, then being surprised by daylight-saving changes.
• Letting a stale worker mark a job complete because there is no lease token or fencing check.
• Retrying at the poller, the queue and the worker all at once.