Design a Distributed Job Scheduler

• Schedule recurring jobs — Cron expressions with a time zone, because "09:00 every weekday" means different instants in March depending on where the customer is.
• Schedule one-off jobs at a future time — Send this reminder in 40 days, expire this hold in 10 minutes. The same machinery, with a schedule of one.
• Run each due execution once — The hard requirement. Charging a card twice because a worker retried is the failure this system exists to prevent.
• Retry failures with backoff — A bounded number of attempts with exponential backoff and jitter, then a dead-letter queue and an alert rather than an infinite loop.
• Respect concurrency and ordering per job — A job that overruns its interval must not start a second copy unless it says it may, and some jobs must not run concurrently with each other.
• Observability and control — History per job, why the last run failed, and the ability to pause, resume, trigger now and backfill a missed window.
• Out of scope — The execution environment itself (containers, capacity), workflow orchestration with dependencies between jobs, and data pipeline lineage.

• Scheduling accuracy (within 1 s of due time at p99) — Late is usually fine, early never is. "Run at 09:00" that fires at 08:59 breaks business logic that assumes the day has started.
• Scale (10 M scheduled jobs, 100 k executions/s peak) — The peak is at the top of the minute and especially the top of the hour — a self-inflicted thundering herd.
• Delivery (at least once, with idempotent execution) — Exactly-once execution is not achievable across a network. Say so, then make duplicates harmless.
• Durability (no scheduled job lost) — A job accepted and then forgotten is invisible until someone notices a report never arrived. Silent loss is the worst failure mode here.
• Availability (99.99 %) — A scheduler outage is not a crash, it is a delay — but a delay long enough becomes a missed window, which for some jobs is a lost day.
• Isolation (no tenant starves another) — One customer scheduling a million jobs at midnight must not delay everyone else's.
• Long jobs (up to 6 hours) — The execution model must tolerate a task that outlives any single lease renewal interval.

• Scheduled jobs: ~10 M — 100 k tenants × ~100 jobs each = 10 M definitions. At ~1 KB each that is 10 GB of definitions: small. The volume is in executions, not definitions.
• Executions per day: ~1 B — 10 M jobs averaging ~100 runs a day (a mix of per-minute, hourly and daily schedules) = ~1 B executions/day ≈ 12 k/s average.
• Peak burst: ~100 k/s — Everything cron lands on the minute, and a large share on the hour. If 30 % of daily executions are hourly jobs clustered in the first second of the hour, the instantaneous rate is ~100 k/s — roughly eight times the average.
• Due-index scan size: ~100 k rows/s — A scan every second for jobs due in the next window returns the burst-sized set: ~100 k rows at the top of the hour, a few hundred otherwise. The index must make that a range scan, never a table scan.
• Execution record storage: ~300 GB/day — 1 B executions × ~300 B (job id, attempt, start, end, status, error) = ~300 GB/day, retained 30 days hot then rolled up. Execution history is far larger than the jobs themselves.
• Workers needed: ~2 000 — Average execution 200 ms of work; 12 k/s × 0.2 s = 2 400 concurrent executions. At ~20 concurrent per worker that is ~120 workers for the average and ~1 000 for the peak — so the queue absorbs the burst rather than the fleet.
• Lease renewal traffic: ~2 k/s — ~2 400 concurrent executions renewing a lease every 30 s = ~80/s normally, rising with long jobs. Trivial, which is what makes leases affordable as the ownership mechanism.

• Tenant service — The application that registers jobs and receives their executions, usually as an HTTP callback or a message. It owns the idempotency of its own handler, which is the other half of the correctness story.
• Schedule API — Create, update, pause, delete, trigger now and backfill. Validates the cron expression and time zone, computes the first due time, and writes the definition.
• Job store (sharded SQL, index on next_run_at) — One row per job: schedule, payload, target, retry policy, concurrency policy, and next_run_at. The index on (shard, next_run_at) is the whole design — every dispatcher query is a range scan on it.
• Dispatcher (one owner per partition) — Wakes each second, claims jobs due in the next window from its partitions, enqueues an execution for each, and advances next_run_at. Partition ownership is leased, so exactly one dispatcher claims any given job.
• Near-term timer wheel (in-memory, next 60 s) — Jobs due within the next minute held in a hierarchical timer wheel so firing is O(1) per tick rather than a database query per second. The store remains the authority; this is a latency optimisation that can be rebuilt at any time.
• Execution queue (partitioned by tenant) — Absorbs the top-of-the-hour burst and decouples deciding from doing. Partitioned per tenant with weighted draining so one tenant's midnight flood cannot starve everyone else.
• Worker pool — Takes an execution, acquires a lease, invokes the target, renews the lease while it runs, and reports the outcome. Stateless and autoscaled on queue depth.
• Lease store (Redis / etcd with TTL) — execution_id → worker, with a short TTL renewed by a heartbeat. A worker that dies stops renewing and its lease expires, which is how work is recovered without anyone having to detect the death directly.
• Job target — The thing actually run: an HTTP endpoint, a queue message, a container. Given an idempotency key derived from the execution so a duplicate delivery is recognisable as one.
• Execution history (time-partitioned) — One row per attempt with status, timing and error. The answer to "did it run and what happened", and the source of the per-job dashboards.
• Retry scheduler — Turns a failure into a future execution with exponential backoff and jitter, up to the policy limit, then dead-letters. Reuses the same due-time mechanism rather than inventing a second one.
• Reaper — Finds executions whose lease expired without an outcome and requeues them. The only thing standing between a worker dying and a job silently never running.
• Monitoring & alerts — Lateness distribution, failure rate per tenant, dead-letter depth, and missed-window alerts. A scheduler that is quietly an hour late looks healthy from the inside unless lateness is measured explicitly.

1. A daily job fires and runs — The happy path, and the shape of the whole system: decide, enqueue, lease, invoke, record, reschedule.
   1. A tenant registers "every weekday at 09:00 Europe/London".
   2. The dispatcher that owns this partition scans for jobs due in the next window.
   3. Jobs due within the minute are loaded into the in-memory timer wheel.
   4. At 08:00:00 UTC the timer fires and an execution is enqueued.
   5. next_run_at is advanced to the next occurrence in the same transaction as the claim.
   6. A worker takes it, acquires a lease, and invokes the target.
   7. The outcome is recorded and the lease released.
2. Midnight: everything is due at once — The scale-breaking case, and it is self-inflicted: cron expressions cluster on round numbers.
   1. At 00:00:00 several hundred thousand jobs become due in the same second.
   2. Dispatchers claim in bounded batches rather than in one enormous scan.
   3. Executions are enqueued as fast as they are claimed; the queue absorbs the rest.
   4. Workers drain at their sustainable rate; jobs run late, not lost.
   5. Per-tenant queue partitions stop one tenant's flood from starving the rest.
   6. New schedules are jittered at creation to flatten future peaks.
3. A worker dies halfway through a job — The failure path. Nobody detects the death directly; the lease simply stops being renewed.
   1. A worker acquires a 60-second lease and starts a job that takes minutes.
   2. The worker is terminated mid-execution.
   3. The lease expires because nothing renews it.
   4. The reaper finds an expired lease with no recorded outcome and requeues the execution.
   5. A new worker picks it up and invokes the target with the same key.
   6. Repeated reaping is capped and the execution is dead-lettered with an alert.
4. A job fails and is retried — Retries reuse the scheduling machinery rather than inventing a parallel one.
   1. The target returns a 500 and the worker records the failure.
   2. The retry scheduler computes the next attempt time with exponential backoff and jitter.
   3. The attempt is written as a future execution in the same due-time index.
   4. The attempt runs with the same execution id and an incremented attempt number.
   5. After the policy limit the execution is dead-lettered and the tenant is alerted.
   6. A tenant-wide failure rate above a threshold pauses that tenant's retries.
5. The clocks go back — The correctness case people forget. "09:00 every day" is ambiguous twice a year and the answer must be deliberate.
   1. A job is defined as 01:30 daily in a zone that moves its clocks.
   2. The next occurrence is computed in local time, then converted to UTC.
   3. A skipped local time is resolved by a documented rule.
   4. A repeated local time fires once, not twice.
   5. A time-zone database update shifts future occurrences and they are recomputed.

Ask: How do you find, every second, which of ten million jobs should run now?

Good answers name: Indexed due-time column, partitioned, with an in-memory timer wheel for the near term, Poll the database every second, no timer wheel, Delayed-delivery queue (SQS-style) as the scheduler, A timer wheel holding every job in memory.

Our pick: A next_run_at column indexed with the partition key, scanned by the owning dispatcher once a second for a 60-second look-ahead with a bounded batch size, feeding a hierarchical timer wheel that fires with millisecond accuracy. The claim and the advance of next_run_at happen in one transaction, so a crash loses an enqueue rather than duplicating an occurrence. Creation-time jitter spreads jobs within their minute where the expression allows.

1. A dispatcher dies holding a loaded wheel. What happens?
   Its partition lease expires and another dispatcher takes over, scans the store and reloads the next 60 seconds. Jobs due during the gap are found immediately because next_run_at is already in the past, so they fire late rather than not at all. The store being authoritative is what makes that safe.
2. Why claim and advance in one transaction?
   So that the occurrence can never fire twice. If you enqueued first and advanced afterwards, a crash between the two re-fires it on the next scan. Advancing first means a crash loses an execution — detectable by the missed-window check and recoverable by backfill — and losing is a far better failure than duplicating.
3. How do you handle a job scheduled 40 days out?
   Identically. It sits in the store with a far-future next_run_at and is invisible to the scan until its day arrives, costing nothing until then. That is the advantage of an indexed due time over any in-memory or delay-queue approach, both of which struggle with long horizons.
4. The index is hot at midnight. How do you spread it?
   Partition on a hash of the job id so the midnight rows are spread across partitions rather than adjacent in one index. Add creation-time jitter within the minute so the due times themselves are spread. The remaining concentration is absorbed by the queue rather than by the database.

Ask: Is exactly-once achievable, and if not, what do you promise instead?

Good answers name: At-least-once delivery with a deterministic idempotency key, At-most-once, Two-phase commit between scheduler and target, Transactional outbox inside the target.

Our pick: At-least-once delivery with an execution id of job_id:scheduled_time:attempt, passed to the target as an idempotency key and stable across reaps and requeues. Leases prevent concurrent duplicates in the common case; the key handles the uncommon one. Documentation says at-least-once plainly and tells tenants how to deduplicate, and the duplicate-delivery rate is monitored so a regression in the lease path is visible.

1. Why does the key include the scheduled time rather than a random id?
   So it is derivable. Both the original delivery and any reap or retry of the same occurrence produce the same key without anyone having to remember one, which is what lets the target deduplicate. A random id would make every delivery look new, which defeats the whole mechanism.
2. Should the attempt number be in the key?
   It is in the payload but not in the deduplication key, and the distinction matters. A reap of attempt 1 must carry attempt 1's key, or the target sees a new key and runs again. A deliberate retry after a genuine failure is a new attempt and should run — it is a different occurrence of work, not a duplicate delivery.
3. A tenant's target is not idempotent. What do you do?
   Tell them, loudly, in the documentation and at registration time — and offer a concurrency policy of "forbid" plus a "no automatic retry" option so the exposure is bounded to reap duplicates only. You cannot make an arbitrary endpoint safe from the outside; the honest thing is to make the guarantee explicit rather than to imply one you cannot provide.
4. How would you measure duplicate deliveries?
   Count deliveries per execution id in the history: anything above one is a duplicate. A low background rate from reaps is expected; a rise means leases are expiring too aggressively or workers are being killed, and it is the earliest signal that the correctness story is degrading.

Ask: How does the system know a worker is still alive and still owns a job?

Good answers name: Short TTL leases with heartbeat renewal, and a reaper for expiries, Queue visibility timeout (SQS-style), Distributed lock via consensus (etcd, ZooKeeper), Worker heartbeats to a central registry, with a supervisor detecting death.

Our pick: Per-execution leases with a 60-second TTL renewed every 20 seconds while the job runs, held in a small fast store. A reaper requeues executions whose lease expired without a recorded outcome, keeping the execution id so the delivery is recognisably a repeat. Leases are advisory rather than authoritative — the target's idempotency is what makes a stale-lease duplicate harmless — and the number of reaps per execution is capped so a poison job cannot cycle forever.

1. A worker is paused by garbage collection for 90 seconds, then resumes. What happens?
   Its lease expired, the reaper requeued the work, and another worker may already be running it. The resumed worker must check its lease before doing anything with an external effect, and ultimately the idempotency key is what prevents the duplicate. This is the classic stale-lease scenario and it is why the guarantee is at-least-once rather than exactly-once.
2. How do you choose the TTL?
   A trade between recovery time and false expiry. Too short and ordinary pauses cause spurious reaps and duplicate deliveries; too long and a dead worker's job sits idle. A minute with 20-second renewals gives three chances to renew before expiry, which survives a transient blip while keeping recovery inside a minute.
3. Can a job run longer than any reasonable TTL — six hours?
   Yes, because renewal decouples job duration from lease duration. The job holds a 60-second lease renewed for six hours. What does break is a job that blocks its own heartbeat thread, which is why renewal must be on a separate thread or process from the work.
4. The lease store goes down. What happens?
   No new leases can be acquired, so execution stops and the queue backs up — the scheduler is unavailable rather than incorrect, which is the right way round. Continuing without leases would mean duplicate execution, and that is a worse failure than a delay. The lease store is replicated for exactly this reason.

Ask: How do you keep a customer with a million midnight jobs from delaying everyone else?

Good answers name: Per-tenant queue partitions with weighted fair draining and concurrency caps, One shared FIFO queue, Dedicated worker pools per tenant, Rate limiting at the API instead of at execution.

Our pick: Queue partitions per tenant, drained by workers with weighted fair queueing and a per-tenant cap on concurrent executions. Dispatcher partitions are also assigned by a hash of tenant and job id so one tenant cannot dominate a single dispatcher's scan. Quotas on total jobs and creation rate limit the blast radius at the source, and very large tenants can be given dedicated worker pools as a paid tier.

1. How do you set the per-tenant concurrency cap?
   From the plan, with a floor generous enough that a small tenant never notices it. It should be visible to the tenant and reported as a metric — "you were capped for 4 minutes at midnight" — because a silent cap looks like an outage from their side, and that is a support ticket you have caused.
2. Fair queueing across a hundred thousand partitions sounds expensive.
   You do not poll every partition. Keep a ready set of partitions with work and draw from it with weights, so cost scales with active tenants rather than with total tenants. At midnight the active set is large, but it is still the set with work, not the whole tenant list.
3. A tenant's jobs are slow and hold workers for minutes each. Is the cap enough?
   A concurrency cap bounds workers held, which is the resource that matters, so yes for isolation. Separately, a maximum execution duration should be enforced and the job terminated and marked failed beyond it, or one tenant's hung endpoint permanently occupies its full concurrency allowance.
4. What is the metric that tells you isolation is working?
   Lateness percentiles per tenant, correlated against other tenants' load. If tenant B's p99 lateness rises when tenant A schedules a million jobs, isolation has failed regardless of what the design says. That correlation is the test worth alerting on.

Ask: A job scheduled every five minutes takes seven. What should happen?

Good answers name: A per-job concurrency policy: forbid, allow, or replace, defaulting to forbid, Always allow concurrent runs, Always forbid, with no option, Queue the occurrence and run it after the current one.

Our pick: A concurrency policy per job, defaulting to forbid: a due occurrence whose previous run is still active is skipped and recorded as skipped rather than queued. Allow permits overlap for jobs that declare themselves safe, and replace cancels the running attempt and starts the new one for jobs where only the latest matters. Enforcement uses a per-job lock in the lease store, and a run that exceeds its maximum duration is terminated so a hang does not silently stop the job forever.

1. Under "forbid", how does the tenant know runs are being skipped?
   Skips are first-class rows in the execution history with a reason, and a run of consecutive skips raises an alert. A skip that is only visible as an absence in a list is the kind of failure nobody notices for a month — which is exactly what this system exists to prevent.
2. What if the job is down for two hours? Do you backfill?
   A per-job policy again, and the default is no: firing 120 accumulated occurrences at once is usually worse than skipping them, and for something like a report it is meaningless. Jobs that genuinely need every window offer an explicit backfill with a cap on how many are replayed, and an operator can trigger one manually.
3. Does "replace" risk the same duplicate problem as a reap?
   Yes, and it has to be honest about it. Cancelling a running job does not guarantee the target stopped, so the replacement may run concurrently with work that is still finishing. Replace is only appropriate for idempotent jobs where the latest result wins, and the documentation has to say so.
4. How is "forbid" enforced across workers?
   A per-job lock in the lease store, acquired before invoking and released on completion, with a TTL so a dead worker does not block the job forever. It is the same mechanism as the execution lease at a different granularity — and reusing it rather than inventing a second locking scheme is what keeps the failure modes understandable.