Design a Distributed Job Scheduler (Cron at Scale)

Every company eventually outgrows crontab on a single box. The box dies and the nightly settlement doesn't run; two copies of the box run it twice; somebody needs "send a reminder 24 hours after signup" for 10 million users, which cron can't express. A distributed job scheduler fixes all three. It is a popular interview question because it tests time-based data modelling, coordination (leases, leader election) and an honest understanding of delivery guarantees, all in one problem.

What We Are Building

Scope it as a multi-tenant internal platform, like a simplified mix of Quartz clustering, Airflow and a cloud scheduler service. Three job types cover almost every use case:

The scheduler does not care what the job does. A job's payload is either an HTTP callback, a message to publish, or a container/command to run on a worker fleet. That abstraction keeps the core small.

Requirements

Functional

• Create, update, pause, resume and delete jobs (cron, one-off, DAG).
• Trigger each run within a tolerance of its scheduled time (target: within a few seconds).
• Retry failed runs with a configurable policy; mark permanently failed runs and alert.
• Per-job concurrency policy: allow overlap, skip if previous still running, or queue.
• Execution history and logs per run; manual "run now" and backfill.

Non-functional

• No lost jobs: once a create call returns, the job will run (durability).
• No duplicate triggers under normal operation; at-least-once under failure.
• Horizontally scalable to hundreds of millions of scheduled one-off jobs.
• Highly available: losing any single node delays jobs by seconds, not hours.
• Tenant isolation: one team's 1M jobs at 00:00 must not delay another team's critical job.

Capacity Estimates

Assumptions for a large company's shared platform:

• 1 million recurring job definitions; median frequency hourly, some every minute. Say an average of 30 runs/day each → 30M recurring runs/day.
• 200 million one-off delayed jobs created per day (order timeouts, reminders, retries for other systems) → ~2,300 creates/s average.
• Total executions: ~230M/day → ~2,700/s average.

The average hides the real problem: clock alignment. Humans schedule things at 0 * * * * and 0 0 * * *. If even 20% of the day's 30M recurring runs fall on the 24 top-of-hour seconds, that is 6M / 24 ≈ 250,000 triggers in a single second on the hour. Design for bursty triggers, not average load.

Storage: a run record of ~500 bytes × 230M/day ≈ 115 GB/day of history; keep 30 days hot (~3.5 TB) and archive the rest. Pending one-off jobs at any time: if the average delay is 1 day, ~200M pending rows × ~300 bytes ≈ 60 GB, which comfortably fits in a partitioned database.

API and Data Model

POST /v1/jobs
{
  "name": "nightly-settlement",
  "schedule": { "cron": "0 2 * * *", "timezone": "Asia/Kolkata" },
  "target": { "type": "http", "url": "https://settle.internal/run", "timeout_s": 900 },
  "retry": { "max_attempts": 5, "backoff": "exponential", "base_s": 30, "max_s": 1800 },
  "concurrency": "skip_if_running",
  "idempotency_key": "settlement-v1"
}
→ 201 { "job_id": "job_01H..." , "next_run_at": "2026-10-11T02:00:00+05:30" }

POST /v1/jobs   { "run_at": "2026-10-10T18:15:00Z", "target": {...} }   # one-off
POST /v1/jobs/{id}:pause | :resume | :trigger
GET  /v1/jobs/{id}/runs?limit=50

jobs(job_id PK, tenant_id, schedule_cron, timezone, run_at, target JSON,
     retry_policy JSON, concurrency, status, next_run_at, version)

-- the trigger index: what is due, organised by time
schedule_index(bucket_minute, shard, run_at, job_id, scheduled_for,
               PRIMARY KEY ((bucket_minute, shard), run_at, job_id))

runs(run_id PK, job_id, scheduled_for, attempt, state, worker_id,
     lease_expires_at, fencing_token, started_at, finished_at, error)
UNIQUE (job_id, scheduled_for, attempt)

The (job_id, scheduled_for) pair is the natural identity of an execution: the 02:00 run on 11 October is one logical run no matter how many times we try to trigger it. That uniqueness is your first line of defence against duplicates.

Architecture

Request flow for a recurring job: the API validates the cron expression, computes next_run_at in the job's timezone, writes the job and one schedule-index row. A trigger node that owns that row's shard sees it when its bucket comes due, creates a runs row and enqueues it. A worker executes it and reports success or failure. On completion (or at trigger time, depending on concurrency policy) the next occurrence is computed and inserted into the index. Only one future occurrence per recurring job lives in the index at a time; you never materialise a year of runs.

Deep Dive: Time-Bucketed Storage

The naive query is SELECT * FROM jobs WHERE next_run_at <= now() AND status='active', every second, against 200M rows. With an index on next_run_at it works on a single database for a while, then the index becomes a write hotspot (every insert lands at the "future" end) and the scan becomes a contention point across trigger nodes.

Instead, organise pending runs by time bucket and shard:

• bucket_minute = the run time truncated to the minute (e.g. 2026-10-11T02:00).
• shard = hash(job_id) % S, with S around 256–1024, so a single minute's jobs are spread across many partitions and many trigger nodes.
• Within a partition, rows are sorted by exact run_at.

A trigger node owning shard 17 runs, every second: "read partition (current_minute, 17) where run_at ≤ now, not yet triggered". That is a bounded, sorted range read on one partition, which Cassandra/ScyllaDB, DynamoDB (bucket+shard as partition key, run_at as sort key) or a hash-partitioned SQL table all do efficiently.

Pre-loading for precision

To hit "within one second" for big bursts, trigger nodes read the next minute's partitions ~30 seconds early into an in-memory min-heap or a hierarchical timing wheel, then fire from memory. Timing wheels (used by Kafka's delayed operations and Netty's HashedWheelTimer) give O(1) insert and expiry, which matters when a node holds hundreds of thousands of pending timers.

import heapq, time

class Trigger:
    def __init__(self, enqueue):
        self.heap = []            # (run_at_epoch, job_id, scheduled_for)
        self.enqueue = enqueue

    def preload(self, rows):
        for r in rows:
            heapq.heappush(self.heap, (r.run_at, r.job_id, r.scheduled_for))

    def tick(self, now=None):
        now = now or time.time()
        fired = 0
        while self.heap and self.heap[0][0] <= now:
            run_at, job_id, scheduled_for = heapq.heappop(self.heap)
            self.enqueue(job_id, scheduled_for)   # idempotent: dedup on (job, time)
            fired += 1
        return fired

Late and missed buckets

If a trigger node was down for 3 minutes, its successor must process old buckets too. Each shard keeps a checkpoint ("fully processed up to 02:03:00"); the new owner resumes from the checkpoint, not from "now". Jobs then apply a misfire policy: fire once late (most jobs), skip if more than X late (a "refresh cache every minute" job shouldn't run 180 times to catch up), or fire all missed occurrences (rarely what you want).

Deep Dive: Leader Election and Partition Ownership

If two trigger nodes process the same shard, every job in it fires twice. We need each shard owned by exactly one node, with fast failover.

Option A: one leader assigns shards

Nodes run leader election through ZooKeeper (ephemeral sequential znodes) or etcd (a lease-backed key with a compare-and-swap). The leader assigns shards to live nodes and reassigns when a node's session expires. Simple to reason about; the leader is only a coordinator, not on the hot path.

Option B: per-shard leases

Every shard has a lease row/key: owner, expires_at, token. Nodes try to acquire unowned or expired leases with an atomic compare-and-set, renew theirs every few seconds, and balance by giving up leases when they hold more than their fair share. No single leader at all.

UPDATE shard_leases
   SET owner = :me,
       expires_at = now() + interval '15 seconds',
       token = token + 1
 WHERE shard = :s
   AND (owner = :me OR expires_at < now())
RETURNING token;   -- no row returned => someone else holds it

Why fencing tokens matter

Node A holds shard 17, then pauses for 20 seconds in a long GC or VM freeze. Its lease expires; node B takes shard 17 with token 8. Node A wakes up, still believes it is the owner (token 7) and enqueues a batch. Without protection, those jobs fire twice. With fencing, every write the trigger node makes (creating the run row, advancing the checkpoint) includes its token, and the store rejects tokens lower than the latest seen for that shard. This is the standard fix for "leases alone are not safe" and a point that distinguishes strong candidates.

Consensus systems like etcd and ZooKeeper choose consistency over availability during partitions, which is what you want for ownership decisions; see CAP theorem for interviews for how to articulate that trade-off.

Deep Dive: At-Least-Once Execution on Workers

Once a run is enqueued, a worker picks it up. The execution protocol must survive workers dying mid-job:

1. Claim. The worker atomically sets state=RUNNING, worker_id, lease_expires_at = now + 60s where state=QUEUED. If the update affects zero rows, another worker has it; drop the message.
2. Heartbeat. While running, extend the lease every ~20 seconds. Long jobs (a 2-hour report) heartbeat for 2 hours.
3. Complete. Set state=SUCCEEDED/FAILED conditioned on still owning the run (worker_id = me), then ack the queue message.
4. Reaper. A background sweeper finds RUNNING rows whose lease expired (worker crashed or partitioned) and moves them to RETRY_PENDING.

If you use a database table as the queue (fine up to thousands of runs/second), PostgreSQL's FOR UPDATE SKIP LOCKED gives you safe concurrent claiming without workers blocking on each other:

WITH next AS (
  SELECT run_id FROM runs
   WHERE state = 'QUEUED' AND queue = 'default'
   ORDER BY priority DESC, scheduled_for
   LIMIT 10
   FOR UPDATE SKIP LOCKED
)
UPDATE runs r
   SET state = 'RUNNING', worker_id = :me,
       lease_expires_at = now() + interval '60 seconds'
  FROM next WHERE r.run_id = next.run_id
RETURNING r.*;

At higher throughput, use Kafka or SQS for dispatch and keep the runs table as the state record. With Kafka, partition by tenant or job ID so one tenant's flood doesn't block others, and remember that a consumer committing offsets before finishing work gives at-most-once, while committing after gives at-least-once. The broker trade-offs are covered in Kafka vs RabbitMQ vs SQS.

Concurrency policies

For skip_if_running, the trigger checks for an existing RUNNING run of the same job and records the new occurrence as SKIPPED. For "only one instance ever", use a per-job lock key (job_lock:{job_id}) acquired by the worker with a lease. Kubernetes CronJob exposes the same three choices as concurrencyPolicy: Allow | Forbid | Replace, a nice real-world reference to mention.

Deep Dive: Retries and Backoff

Failures are normal: the downstream API is down, a deploy restarted the worker, a database deadlocked. The retry policy decides whether a transient failure becomes an incident.

• Exponential backoff: delay = min(max_delay, base × 2^attempt).
• Jitter: randomise the delay so 50,000 runs that failed together don't retry together. "Full jitter" picks uniformly in [0, delay].
• Max attempts then a terminal FAILED/dead-letter state that pages the job owner.
• Classify errors: 4xx-style errors (bad payload) should not be retried; 5xx/timeouts should.
• Retries go back through the scheduler as a one-off run at now + delay, so they are durable and visible, not a sleep loop in the worker.

import java.time.Duration;
import java.time.Instant;
import java.util.concurrent.ThreadLocalRandom;

public final class Backoff {
    public static Instant nextAttemptAt(Instant now, int attempt,
                                        Duration base, Duration max) {
        long cap = max.toMillis();
        long exp = base.toMillis() * (1L << Math.min(attempt, 30));
        long ceiling = Math.min(cap, exp);
        long delay = ThreadLocalRandom.current().nextLong(ceiling + 1); // full jitter
        return now.plusMillis(delay);
    }

    public static void main(String[] args) {
        Instant now = Instant.parse("2026-10-11T02:00:00Z");
        for (int a = 0; a < 5; a++) {
            System.out.println(a + " -> " +
                nextAttemptAt(now, a, Duration.ofSeconds(30), Duration.ofMinutes(30)));
        }
    }
}

With base 30 s, the ceilings for attempts 0–5 are 30 s, 60 s, 2 min, 4 min, 8 min, 16 min, then capped at 30 min. Also put a per-target circuit breaker in the worker: if 90% of runs calling one endpoint fail, pause dispatch to it instead of burning retries.

Deep Dive: DAG Dependencies

Workflows like "ingest → reconcile → settle → report" add dependencies. Model a workflow as a DAG; each DAG run gets its own task-instance rows. Validate the DAG at creation time (reject cycles) and release tasks as their parents finish, which is Kahn's topological sort executed incrementally.

from collections import defaultdict, deque

def validate_dag(edges, tasks):
    indeg = {t: 0 for t in tasks}
    adj = defaultdict(list)
    for u, v in edges:            # u must finish before v
        adj[u].append(v); indeg[v] += 1
    q = deque(t for t in tasks if indeg[t] == 0)
    seen = 0
    while q:
        u = q.popleft(); seen += 1
        for v in adj[u]:
            indeg[v] -= 1
            if indeg[v] == 0:
                q.append(v)
    if seen != len(tasks):
        raise ValueError("cycle detected")
    return adj

class DagRun:
    def __init__(self, tasks, edges):
        self.adj = validate_dag(edges, tasks)
        self.remaining = {t: 0 for t in tasks}
        for _, v in edges:
            self.remaining[v] += 1

    def initial(self):
        return [t for t, d in self.remaining.items() if d == 0]

    def on_success(self, task):
        ready = []
        for v in self.adj[task]:
            self.remaining[v] -= 1
            if self.remaining[v] == 0:
                ready.append(v)
        return ready

run = DagRun(["ingest", "fx", "reconcile", "settle", "report"],
             [("ingest", "reconcile"), ("fx", "reconcile"),
              ("reconcile", "settle"), ("settle", "report")])
print(run.initial())               # ['ingest', 'fx']
print(run.on_success("ingest"))    # []  (reconcile still waits for fx)
print(run.on_success("fx"))        # ['reconcile']

Validation is O(V + E). In production, the remaining counter lives in the database and is decremented with an atomic update (UPDATE task_instances SET pending_parents = pending_parents - 1 WHERE ... RETURNING pending_parents) so two parents finishing concurrently can't both miss the zero. Real systems also add trigger rules (run if all succeeded, if any failed, always for cleanup tasks), upstream failure propagation (mark downstream as UPSTREAM_FAILED), and backfills (create DAG runs for past dates). Apache Airflow's scheduler, metadata DB and executors map almost one-to-one onto this design, which makes it a handy reference in interviews.

Exactly-Once: What You Can and Cannot Promise

Interviewers often ask, "Can you guarantee each job runs exactly once?" The precise answer:

• Triggering exactly once is achievable in steady state: one owner per shard, fenced writes, unique (job_id, scheduled_for).
• Executing exactly once is impossible in general. A worker can perform the side effect (send the SMS, call the payment API) and crash before reporting success. The scheduler sees an expired lease and cannot tell "never ran" from "ran but didn't report". If it doesn't retry, it risks never running; if it retries, it risks running twice. It must pick at-least-once.
• Effectively-once is achievable when the job's side effects are idempotent or transactional:
  • Pass a stable execution_id = hash(job_id, scheduled_for) to the job; downstream APIs dedupe on it (payment gateways, for instance, accept idempotency keys).
  • If the job writes to a database, write its result and a "processed execution_id" marker in the same transaction.
  • For Kafka-to-Kafka jobs, Kafka transactions give exactly-once within Kafka only.

Trade-offs and Failure Modes

Failure scenarios to walk through

• Trigger node dies: lease expires in ~15 s; new owner resumes from checkpoint; some runs are seconds late.
• Queue is down: triggers can't enqueue; checkpoints don't advance; on recovery, the backlog fires with misfire policies applied.
• Clock skew: trigger nodes should use time from a synchronised source (NTP/chrony) and tolerate small skew; never let a worker's clock decide whether a job is due.
• Timezones and DST: store the job's IANA timezone and compute next runs in it. India has no DST, but a global platform must handle "02:30 doesn't exist today" and "01:30 happens twice".
• Noisy tenant: per-tenant quotas on creation rate and concurrent runs; separate queues for priority tiers. The same token-bucket ideas from rate limiter design apply.

Interview Playbook for "Design a Job Scheduler"

Red flags interviewers note

• Every scheduler node scanning the whole table with no ownership story.
• Materialising all future occurrences of recurring jobs.
• Claiming exactly-once execution with a lock.
• Retries implemented as sleep() in the worker, lost on crash.
• No answer for a worker that dies mid-job.

If you present the scheduler as "a durable time index, single-owner triggering, at-least-once execution with idempotency", you will have the right skeleton for nearly every follow-up. The same lease-and-fencing pattern shows up in distributed key-value stores and Kafka's partition leadership, so it is worth learning well once.

Frequently Asked Questions

Pranjal Jain

Ex-Microsoft SDE · IIT Kanpur · Founder of Prepflix. Helps engineers crack startup and product company interviews across India.