Design a Distributed Job Scheduler

1Requirements

2Back-of-envelope estimation

3API design

4High-level design

Three planes: an API service that validates and persists jobs, a scheduling plane that decides what is due, and a worker plane that executes. The single most important design decision is that the source of truth is a durable jobs store (a relational DB works well into the tens of thousands of QPS), and everything else (queues, in-memory timing wheels) is a rebuildable acceleration layer on top of it.

The scheduling plane handles two shapes of time. One-off jobs are simply rows with run_at; a due-job query is an index scan on (status, run_at). Cron jobs are templates: when a firing is claimed, the scheduler materializes the execution and immediately computes and writes the next run_at from the cron expression, so a cron job is just a self-replenishing one-off. This unifies the model and means missed windows (scheduler down for 5 minutes) are naturally caught up because due rows are still due.

Dispatch uses worker leases rather than fire-and-forget queues. A worker claims due jobs with an atomic UPDATE ... RETURNING that flips status to running, stamps the worker id, and sets lease_expires_at = now + lease_duration. If the worker dies, it stops heartbeating, the lease expires, and a reaper flips the row back to pending for another worker. This gives at-least-once execution with no lost jobs and no zookeeper-style external lock service.

Exactly-once semantics are layered on with fencing and idempotency. Every claim gets a monotonically increasing attempt number that acts as a fencing token: a zombie worker that wakes up after its lease expired will have its completion report rejected because its token is stale. For side effects outside the system (send an email, charge a card), the job payload carries an idempotency key that the downstream effect must honor; the scheduler can only guarantee exactly-once state transitions, not exactly-once side effects.

Priority and load shedding: the claim query orders by (priority DESC, run_at ASC), so high-priority jobs are always claimed first, and under sustained overload low-priority jobs simply age in the queue rather than causing failures. Retries reinsert the job with run_at = now + backoff(attempt), where backoff is exponential with jitter, and after max_attempts the job moves to a dead-letter state for human inspection.

5Data model

6Deep dives

7Rapid implementation: build the MVP

Theory is table stakes. Here is how you would stand up a working version fast, on a budget, with the core algorithm in real code.

1. 01Create the jobs table with the partial index: CREATE INDEX ON jobs (priority DESC, run_at) WHERE status = 'pending'
2. 02Write a claimJobs(workerId, n) function using UPDATE ... FROM a SKIP LOCKED subquery, RETURNING the claimed rows
3. 03Write the worker loop: claim up to 10 jobs, execute each with a try/catch, report success or failure with the fencing predicate (AND attempt = claimedAttempt)
4. 04Add a heartbeat interval per running job that pushes lease_expires_at forward every 20 seconds
5. 05Add the reaper: every 30 seconds, UPDATE jobs SET status='pending' WHERE status='running' AND lease_expires_at < now()
6. 06Add cron support with the cron-parser npm package: on claiming a cron job, insert the next execution row before running the current one
7. 07Implement failure handling: retryable errors set run_at = now + jittered backoff and status='pending'; attempt >= max_attempts sets status='dead'
8. 08Kill -9 a worker mid-job and watch the reaper recover it; run 3 workers against 10k seeded jobs to verify no job runs with two overlapping leases

-- Claim up to 10 due jobs atomically; concurrent workers get disjoint sets.
UPDATE jobs j
SET    status           = 'running',
       worker_id        = $1,
       attempt          = j.attempt + 1,
       lease_expires_at = now() + interval '60 seconds'
FROM (
  SELECT job_id
  FROM   jobs
  WHERE  status = 'pending'
    AND  run_at <= now()
  ORDER  BY priority DESC, run_at ASC
  LIMIT  10
  FOR UPDATE SKIP LOCKED
) due
WHERE j.job_id = due.job_id
RETURNING j.job_id, j.type, j.payload_json, j.attempt, j.cron_expr;

import { sql } from "./db";
import { handlers } from "./handlers";

const WORKER_ID = process.env.WORKER_ID ?? "worker-" + process.pid;

async function runOne(job: { job_id: string; type: string; payload_json: any; attempt: number }) {
  const hb = setInterval(() => {
    sql("UPDATE jobs SET lease_expires_at = now() + interval '60 seconds' " +
        "WHERE job_id = $1 AND worker_id = $2 AND attempt = $3",
        [job.job_id, WORKER_ID, job.attempt]).catch(() => {});
  }, 20_000);

  try {
    await handlers[job.type](job.payload_json);
    // Fencing: attempt must still match, or a zombie is trying to complete.
    await sql(
      "UPDATE jobs SET status = 'succeeded' " +
      "WHERE job_id = $1 AND worker_id = $2 AND attempt = $3",
      [job.job_id, WORKER_ID, job.attempt]
    );
  } catch (err) {
    const backoffSec = Math.random() * Math.min(900, 5 * 2 ** job.attempt);
    await sql(
      "UPDATE jobs SET " +
      "  status = CASE WHEN attempt >= max_attempts THEN 'dead' ELSE 'pending' END, " +
      "  run_at = now() + ($4 || ' seconds')::interval " +
      "WHERE job_id = $1 AND worker_id = $2 AND attempt = $3",
      [job.job_id, WORKER_ID, job.attempt, backoffSec.toFixed(0)]
    );
  } finally {
    clearInterval(hb);
  }
}

import parser from "cron-parser";
import { sql } from "./db";

// Called right after claiming a job that has a cron expression:
// schedule the next firing before running this one (fixed-rate semantics).
export async function materializeNext(job: {
  job_id: string; type: string; payload_json: any; cron_expr: string;
}) {
  const next = parser.parseExpression(job.cron_expr).next().toDate();
  await sql(
    "INSERT INTO jobs (type, payload_json, cron_expr, run_at, status, priority, max_attempts) " +
    "SELECT type, payload_json, cron_expr, $2, 'pending', priority, max_attempts " +
    "FROM jobs WHERE job_id = $1 " +
    "AND NOT EXISTS (" +
    "  SELECT 1 FROM jobs WHERE cron_expr = $3 AND type = $4 " +
    "  AND status = 'pending' AND run_at = $2" +
    ")",
    [job.job_id, next.toISOString(), job.cron_expr, job.type]
  );
}