Design a Distributed Job Scheduler (Airflow / Temporal / Distributed Cron)

TL;DR. A distributed job scheduler splits into three independent concerns: a scheduler tier that decides when to fire, a dispatch queue that decouples firing from execution, and a stateless worker fleet that runs user code with idempotency-key deduplication. At 100k registered jobs and 10k executions/sec, a single leader caps throughput, so hash-range sharding across etcd-leased scheduler instances scales horizontally. Exactly-once execution is a worker-side property: at-least-once delivery plus idempotent side effects keyed on {workflow_id, task_id, attempt}. Event-sourced workflow history (the Temporal model) eliminates mid-task crash bugs by replaying from an append-only log rather than snapshotting mutable state.

Learning Objectives#

• Design a scheduler-dispatcher-worker split that survives any single component failing
• Choose between leader-based and hash-range sharded scheduler topologies and justify the decision at 10k exec/sec
• Guarantee exactly-once task execution using idempotency keys and worker-side deduplication
• Apply late/missed-run policies (skip, catchup, single-most-recent) matching cron, Airflow, and Temporal semantics
• Reason about event-sourced workflow history as an alternative to DAG-state-in-a-database
• Estimate capacity for history writes at 50 events per workflow at 10k workflows/sec

Intuition#

A cron daemon on one machine is trivially correct. It fires jobs on schedule, and its failure domain is just that one box^[1]. At 10 users, this works. At 100,000 registered jobs firing 10,000 executions per second, it collapses for three reasons.

First, one box cannot evaluate 100k schedules per second and dispatch 10k tasks without falling behind. Second, when that box dies, every scheduled job stops until a human notices. Third, a worker crash mid-task either loses the work (at-most-once) or re-runs it (at-least-once with potential double-execution of non-idempotent side effects like charging a credit card).

The naive fix is "add a second scheduler." But now both might fire the same job at the same instant, producing double-execution. Or during failover, neither fires it, producing a miss. Google's production cron solves this with Paxos: "the most important state we keep in Paxos is information regarding which cron jobs are launched. We synchronously inform a quorum of replicas of the beginning and end of each scheduled launch"^[1:1].

The insight that unlocks the design: separate the decision to fire from the execution of user code. The scheduler decides when; a durable queue carries the decision; a stateless worker executes with an idempotency key as the final safety net. Each layer scales, upgrades, and fails independently.

Requirements#

Clarifying Questions#

• Q: Cron-style periodic only, or DAGs with dependencies and branching? Assume: Full DAGs up to 10k nodes with per-task retries and conditional branching.

• Q: Required execution semantics? Assume: Exactly-once observable effect. At-least-once delivery with worker-side idempotency.

• Q: How strict is schedule fidelity? Assume: p99 fire-delay < 5 seconds from the scheduled instant.

• Q: Missed-run policy when the scheduler was down? Assume: Configurable per job: skip, catch-up all missed intervals, or single-most-recent.

• Q: Single region or multi-region? Assume: Single region with multi-AZ redundancy. Cross-region is a follow-up.

• Q: Maximum task duration? Assume: From sub-second (event-triggered) to days (human-in-the-loop approval steps).

Functional Requirements#

• Register, update, delete, and pause job definitions (cron, interval, event-trigger, or DAG)^[2]
• Trigger manual executions and backfill closed time windows
• Execute DAGs up to 10k nodes with per-task retries, conditional branching, and fan-out/fan-in^[3]
• Emit per-task logs, per-workflow lineage, and completion/failure callbacks
• Support signals into waiting workflows (Temporal-style) and cancellation with optional compensation^[4]

Non-Functional Requirements#

• Load: 100k registered jobs, 10k executions/sec peak dispatch
• Latency: p99 fire-delay < 5s; p99 task-claim latency < 200 ms
• Availability: 99.99% control-plane; 99.9% execution path
• Consistency: exactly-once observable effect under worker crashes
• Durability: no triggered execution silently dropped; permanent failures land in DLQ

Capacity Estimation#

• Read:write ratio: 1:50 on the history path (writes dominate); 10:1 on the control-plane API (reads dominate)
• Hot path: history writes at 250 MB/s force Cassandra sharded on workflow_id; a single PostgreSQL instance cannot absorb this
• Metadata vs history: job definitions (1 GB) fit one PostgreSQL instance; the history is the big table

API and Data Model#

API Design#

POST /v1/jobs
  Body: { "name": "nightly-etl", "schedule": "0 2 * * *",
          "dag_spec": {...}, "retry_policy": {...}, "missed_run_policy": "skip" }
  Returns: 201 { "job_id": "j_abc", "next_fire_time": "..." }

POST /v1/executions
  Idempotency-Key: <uuid>
  Body: { "job_id": "j_abc", "trigger": "manual", "params": {...} }
  Returns: 202 { "execution_id": "e_xyz", "status": "pending" }

GET /v1/executions/{id}
  Returns: 200 { "execution_id": "e_xyz", "status": "running",
                  "tasks": [{ "task_id": "t1", "status": "completed" }, ...] }

POST /v1/executions/{id}/cancel
  Returns: 200 { "status": "cancelling" }

POST /v1/executions/{id}/signal
  Body: { "signal_name": "approval", "payload": {...} }
  Returns: 200

Pagination on list endpoints uses opaque cursor tokens. Rate limiting: 1,000 req/sec per caller at the API gateway. The missed_run_policy field accepts skip, catchup, or most_recent. Quartz supports integer trigger priorities where "if N Triggers are to fire at the same time, but there are only Z worker threads currently available, the first Z Triggers with the highest priority will be executed first"^[5]. Sidekiq supports both strict and weighted random queue ordering across named queues^[6].

Data Model#

-- Job definitions (PostgreSQL, single instance)
CREATE TABLE jobs (
  job_id        UUID PRIMARY KEY,
  name          TEXT NOT NULL,
  schedule      TEXT,            -- cron expression or NULL for event-triggered
  dag_spec      JSONB,           -- task graph definition
  retry_policy  JSONB,
  missed_run_policy TEXT DEFAULT 'skip',
  state         TEXT DEFAULT 'active',  -- active | paused | deleted
  shard_key     INT GENERATED ALWAYS AS (hashtext(job_id::text) % 256) STORED
);

-- Schedule index (Redis sorted set)
-- Score: next_fire_time as Unix epoch
-- Member: job_id
-- The scheduler's hot loop: ZRANGEBYSCORE schedules -inf <now> LIMIT 0 100

-- Execution history (Cassandra, partitioned by workflow_id)
CREATE TABLE execution_events (
  workflow_id   UUID,
  event_id      TIMEUUID,
  event_type    TEXT,    -- WorkflowStarted, ActivityScheduled, ActivityCompleted, ...
  payload       BLOB,
  PRIMARY KEY (workflow_id, event_id)
) WITH CLUSTERING ORDER BY (event_id ASC);

-- Task state (Cassandra, partitioned by workflow_id)
CREATE TABLE task_state (
  workflow_id     UUID,
  task_id         TEXT,
  attempt         INT,
  status          TEXT,    -- pending | running | completed | failed | dead_letter
  idempotency_key TEXT,   -- {workflow_id}.{task_id}.{attempt}
  worker_id       TEXT,
  heartbeat_at    TIMESTAMP,
  PRIMARY KEY (workflow_id, task_id, attempt)
);

High-Level Architecture#

Shard-owned schedulers poll Redis for due jobs, dispatch via Kafka, and stateless workers write execution events to Cassandra. etcd coordinates shard ownership.

Write path. The control-plane API registers a job in PostgreSQL and inserts its next_fire_time into a Redis sorted set^[7]. Each scheduler instance owns a range of shard keys (0-255) via etcd leases. On each tick, a scheduler runs ZRANGEBYSCORE on its shard's schedule entries, pops due jobs, computes the next fire time, and produces a dispatch message to Kafka keyed by workflow_id.

Execution path. Workers consume from Kafka, check the idempotency key against Cassandra's task_state table, execute user code, and write an ActivityTaskCompleted event to the history. On failure, the worker writes ActivityTaskFailed and the scheduler re-enqueues with incremented attempt and exponential backoff.

Coordination path. etcd leases expire after 10 seconds. If a scheduler dies, its shards are reassigned to surviving instances within one lease TTL. Workers are unaffected because they are decoupled via Kafka.

Deep Dives#

Scheduler topology: leader-based vs hash-range sharded#

A leader-based scheduler (one active instance via Paxos or etcd lease) is the simplest correct design. Google's distributed cron uses exactly this: "the leader is the only replica that actively launches cron jobs"^[1:2]. Failover completes within ~60 seconds^[1:3], which is acceptable for cron's one-minute granularity.

At 10k executions/sec, a single leader cannot keep up. The alternative: hash-range sharding. Each scheduler instance owns a contiguous range of a 256-slot hash ring. Job IDs hash to a slot; the slot's owner is the only process that may fire that job.

Leader-based caps at one-box throughput; hash-range sharding distributes load across N active schedulers with bounded blast radius per shard.

Split-brain risk. During lease renewal, two instances may briefly believe they own the same shard. The worker-side idempotency check ({workflow_id, task_id, attempt} dedup in Cassandra) is the final safety net. Google's cron addresses this by writing synchronously via Paxos before launching: "the actual cron job launch does not proceed until it receives confirmation that the Paxos quorum has received the launch notification"^[1:4].

Graceful upgrades. New instances join etcd. The coordinator reassigns shards off old instances. Drained schedulers flush in-flight writes and exit. Workers keep running because durable state lives outside the scheduler process.

Exactly-once execution via idempotency keys#

There is no exactly-once network delivery^[1:5][8]. A worker can always crash between running a side effect and acknowledging the work. The industry-standard resolution: at-least-once delivery plus idempotent side effects.

The scheduler stamps every dispatch message with an idempotency key: {workflow_id}.{task_id}.{attempt}. The worker checks this key against the task_state table before executing. If the key exists with status completed, the worker short-circuits and acknowledges without re-running.

Celery pattern. Without acks_late=True, Celery acknowledges on receive; a mid-task crash loses the work^[9]. With acks_late=True plus reject_on_worker_lost=True, a SIGKILL requeues the message^[9:1]. Sidekiq Pro's super_fetch uses Redis LMOVE to keep jobs in Redis until the worker confirms completion; orphan recovery runs when a heartbeat expires (60 seconds)^[10].

Step Functions split. AWS Step Functions Standard workflows persist state between transitions and return idempotent responses on same-named re-executions^[8:1]. Express workflows drop to at-least-once because "execution state doesn't persist between state transitions"^[8:2], trading durability for "up to 100,000 state transitions per second"^[11].

Worker heartbeat. Workers heartbeat every 10 seconds. If the history service sees no heartbeat for 3x the interval (30 seconds), it marks the task as timed out and the scheduler re-enqueues with attempt + 1. Sidekiq Pro's poison-pill detection kills a job automatically if it is recovered 3 times within 72 hours^[10:1].

Durable execution and event-sourced workflow history#

Airflow stores DAG state in PostgreSQL: task status rows updated in place. A crash between "task ran" and "status updated" leaves the system in an ambiguous state. Airflow's HA scheduler uses SELECT ... FOR UPDATE to serialize the critical section^[7:1], but this caps throughput at what one DB row lock can sustain.

Temporal takes a fundamentally different approach: event-sourced workflow history. "Each Workflow Execution emits a series of Commands and processes a sequence of Events, which are recorded in an Event History"^[12]. Events include ActivityTaskScheduled, ActivityTaskCompleted, TimerStarted, WorkflowExecutionSignaled^[13].

On worker restart, Temporal "doesn't restore memory from a snapshot. It starts the Workflow code from the beginning, replays the Event History step by step, and uses that history to guide the code back to the exact state as before"^[12:1]. Activities return their recorded result on replay instead of re-running.

A workflow execution from trigger through fan-out, per-task retry with idempotency check, history-event emission, and final completion.

Trade-off. Every activity writes at least two events (scheduled + completed). A 50-step workflow writes ~100 events. At 10k workflows/sec, that is ~500k event writes/sec^[13:1]. This is why the history store must be Cassandra or DynamoDB, not PostgreSQL.

Deterministic constraint. Replay forces workflow code to avoid non-deterministic constructs (direct clock reads, random, network calls). Temporal's SDK provides replay-safe wrappers^[12:2]. This is the mental-model cost of durable execution.

Late/missed-run policies and cron edge cases#

When a scheduler is down from 08:29 to 10:21, 112 minute-granularity runs are missed. On recovery, three policies apply:

• Skip: fire only the next scheduled time. Kubernetes CronJob's default behavior.

• Catchup: fire every missed interval. Airflow 2.x's catchup=True default (changed to False in Airflow 3.0), which triggers surprise storms after outages.

• Single-most-recent: fire only the last missed interval. A pragmatic middle ground.

Kubernetes CronJob caps missed schedules at 100: "if more than 100 schedules were missed, the CronJob is no longer scheduled"^[14]. The controller logs too many missed start times and stops, preventing the storm.

Per-task state machine: tasks transition through retry with exponential backoff until success or dead-letter.

Cron edge cases. DST transitions can cause a 2 AM job to fire twice (fall-back) or never (spring-forward). Kubernetes mandates .spec.timeZone (stable since v1.27) using Go's tz database; in-schedule CRON_TZ is rejected^[14:1]. Quartz handles misfires with per-trigger instructions: MISFIRE_INSTRUCTION_FIRE_NOW, MISFIRE_INSTRUCTION_DO_NOTHING, and type-specific policies^[15].

Thundering herd at midnight. Fifty thousand daily jobs configured as 0 0 * * * all fire at 00:00 UTC. Google's cron added a ? wildcard that hashes the job config to a deterministic value within the allowed range, spreading launches evenly^[1:6]. Marc Brooker's "Full Jitter" formula sleep = random(0, min(cap, base * 2^attempt)) reduces contention by more than half^[16].

Real-World Example#

Temporal and Uber Cadence at scale#

Temporal originated as a fork of Uber's Cadence^[17], which powers "over a thousand services" at Uber, processing over 12 billion executions and 270 billion actions per month^[18]. Stripe, Netflix, HashiCorp, and Datadog use Temporal for durable execution^[19]. As of February 2026, Temporal Cloud alone has processed 9.1 trillion lifetime action executions, with over 20 million installs per month^[20].

Architecture. A Temporal cluster decomposes into four server roles: Frontend (gRPC API), History (owns shards of workflow state), Matching (owns task queues), and Worker (internal system workflows)^[13:2]. The persistence store is pluggable: Cassandra, MySQL, or PostgreSQL. User-owned worker processes poll task queues, execute workflow code deterministically against the event history, and run activities (the actual side effects)^[12:3].

Key insight: replay, not snapshotting. "Temporal doesn't restore memory from a snapshot. It starts the Workflow code from the beginning, replays the Event History step by step"^[12:4]. This means a crashed worker resumes from the last consistent event without any checkpoint infrastructure. Activities return their recorded result on replay instead of re-executing, achieving exactly-once observable effect.

Contrast with Airflow. Airflow's HA scheduler uses database row-level locks: "we use database row-level locks (using SELECT ... FOR UPDATE)"^[7:2]. This rejects consensus algorithms explicitly: "by not using direct communication or consensus algorithm between schedulers (Raft, Paxos, etc.) nor another consensus tool (Apache Zookeeper, or Consul for instance) we have kept the 'operational surface area' to a minimum"^[7:3]. The trade-off: Airflow requires PostgreSQL 12+ or MySQL 8.0+ for SKIP LOCKED; MariaDB pre-10.6 is unsupported for HA^[7:4]. DAG parse times at scale are a known bottleneck. Airflow was started in October 2014 by Maxime Beauchemin at Airbnb and announced publicly in June 2015 to manage "complex data pipelines"^[21][3:1], and Netflix built Conductor as a JSON-blueprint alternative when cron-only systems devolved into ad hoc pub/sub choreography^[22].

Contrast with Step Functions. AWS Step Functions Standard Workflows offer exactly-once semantics at $25 per million state transitions, with a sustained StartExecution rate of 300/sec (burst 1,300) and a state-transition rate of 5,000/sec in the major regions (US East, US West, Europe Ireland)^[23]. Express Workflows reach "up to 100,000 state transitions per second" at $1.00 per million invocations (plus duration and memory) but drop to at-least-once^[11:1]. The dual product line splits long-durable from fast-throughput rather than forcing one runtime to do both^[8:3].

Trade-offs#

Cron is not a DAG engine, and a DAG engine is not a workflow engine. Pick the model that matches your real dependency shape. At 10k exec/sec with DAGs up to 10k nodes, the hash-range sharded scheduler with Kafka dispatch and Temporal-style event-sourced history is the right combination.

Scaling and Failure Modes#

At 10x (100k exec/sec): History writes hit 2.5 GB/sec. Mitigation: add Cassandra nodes, increase replication factor, and compact aggressively. Kafka partitions scale to 300. Scheduler shards scale to 30 instances.

At 100x (1M exec/sec): Single-region Kafka cannot absorb 1M/sec writes. Mitigation: multi-region Kafka clusters with geo-routing. History store moves to DynamoDB on-demand for elastic throughput. Schedule evaluation becomes the bottleneck; pre-compute next-fire-times in batch rather than per-tick evaluation.

At 1000x: The architecture shifts to a tiered model: edge schedulers handle cron evaluation locally, forwarding only dispatch decisions to a central history service. Workers auto-scale on queue depth with Kubernetes HPA.

Failure: Scheduler shard dies mid-tick. etcd lease expires in 10 seconds. Surviving instances acquire orphaned shards. Any jobs that were dispatched but not yet acknowledged by workers will be re-dispatched; the idempotency key prevents double-execution. Schedule fidelity degrades by at most one lease TTL (10 seconds), within the 5-second p99 target for most jobs. Quartz's clustered JDBC store detects failed nodes via clusterCheckinInterval (default 15 seconds) and reassigns their triggers^[26].

Failure: History database partition hot-spots. One trending workflow fires 10k instances/sec all keyed on the same entity. One Cassandra partition saturates. Mitigation: composite partition key {workflow_id, bucket} where bucket is hash(task_id) % 16. This is the same fix Kafka users apply for hot keys.

Failure: Kafka consumer lag spike. Workers fall behind during a cron storm. Transactional executions (payment settlements) are delayed. Detection: consumer lag > 10k. Mitigation: priority topics ensure high-priority jobs have dedicated consumer groups never starved by bulk batch work.

Common Pitfalls#

Follow-up Questions#

Exercise#

Exercise 1: Capacity for a payment settlement scheduler#

A fintech processes 500,000 payment settlements per day. Each settlement is a 3-task DAG (validate, transfer, reconcile). Tasks average 2 seconds each. The system must guarantee exactly-once execution. Estimate: peak dispatch QPS, history write throughput, and minimum worker count.

Key Takeaways#

• Scheduler, dispatcher, and worker are three distinct concerns. Separate them so each scales, upgrades, and fails independently.
• Exactly-once is a worker-side dedup property, not a network property. The idempotency key {workflow_id, task_id, attempt} is the contract.
• Hash-range sharding beats leader-based at scale. A single Paxos leader caps at one-box throughput; sharding distributes load with bounded blast radius.
• Event-sourced history (Temporal model) trades storage for durability. Replay from an append-only log eliminates mid-task crash bugs that plague mutable-state schedulers.
• Missed-run policy is a per-job configuration, not a system default. Skip, catchup, and single-most-recent serve different use cases; expose all three.
• Jitter is mandatory on every retry and every cron expression. Without it, synchronized retries and midnight storms are inevitable.

Further Reading#

• Google SRE Book, Chapter 24: Distributed Periodic Scheduling with Cron. The canonical public source on distributed cron with Paxos; the model most modern systems measure themselves against.
• Apache Airflow Scheduler docs. HA scheduler design with DB row-level locks; the most widely deployed DAG engine.
• Temporal Workflows encyclopedia. Deterministic replay and event history from the canonical durable-execution engine.
• AWS Step Functions: choosing workflow type. The cleanest public write-up of Standard vs Express trade-offs.
• Kubernetes CronJob. The minimal-viable periodic primitive and the pitfalls it bakes in (startingDeadlineSeconds, 100-missed-schedules cap).
• Marc Brooker: Exponential Backoff and Jitter. The definitive short read on retry-storm prevention with the Full Jitter formula.
• Sidekiq Reliability wiki. super_fetch, orphan recovery, poison-pill detection; Redis-backed queue reliability at the low-infra end.
• Netflix Conductor: A microservices orchestrator. DAG-based orchestration with JSON blueprints; useful contrast to workflow-as-code.

Flashcards#

References#