How would you design a distributed job scheduler?

// Distributed job scheduler, simulated in one process with a fake clock.
// The Map "runs" stands for a database table with a UNIQUE run_id column.
const fmt = (n) => Math.round(n).toString().replace(/\B(?=(\d{3})+(?!\d))/g, ",");

console.log("== 1. Capacity (assumptions: 5M recurring jobs, each runs hourly) ==");
const jobs = 5_000_000, runsPerDay = jobs * 24;
console.log(`runs/day ${fmt(runsPerDay)}; average ${fmt(runsPerDay / 86400)} runs/s`);
console.log(`if every job fires at minute 0: ${fmt(jobs)} runs in one second`);
console.log(`run rows/day at 300 bytes each: ${fmt(runsPerDay * 300 / 1e9)} GB`);

console.log("\n== 2. Top-of-hour herd: 100,000 jobs, with and without jitter ==");
function fnv1a(s) {
  let h = 0x811c9dc5;
  for (let i = 0; i < s.length; i++) { h ^= s.charCodeAt(i); h = Math.imul(h, 0x01000193) >>> 0; }
  return h;
}
const perSec = new Array(60).fill(0);
for (let i = 0; i < 100000; i++) perSec[fnv1a("job-" + i) % 60]++;
console.log("no jitter: second 0 gets 100,000, seconds 1-59 get 0");
console.log(`jitter = hash(job_id) % 60: min ${fmt(Math.min(...perSec))}/s, max ${fmt(Math.max(...perSec))}/s`);

console.log("\n== 3. Two scheduler instances fire the same tick ==");
const runs = new Map();
function createRun(who, job, at) {
  const id = `${job}@${at}`;
  if (runs.has(id)) { console.log(`${who}: INSERT ${id} -> already exists, ignored`); return; }
  runs.set(id, { id, state: "PENDING", owner: null, leaseUntil: 0, fence: 0, attempt: 0 });
  console.log(`${who}: INSERT ${id} -> created`);
}
createRun("scheduler-A", "report", 60);
createRun("scheduler-B", "report", 60);

console.log("\n== 4. Lease, heartbeat, crash-like pause, fencing token ==");
const LEASE = 30;
const effects = new Set();
function claim(t, worker) {
  for (const r of runs.values()) {
    const free = r.state === "PENDING" || (r.state === "RUNNING" && r.leaseUntil <= t);
    if (!free) continue;
    r.state = "RUNNING"; r.owner = worker; r.leaseUntil = t + LEASE; r.fence++; r.attempt++;
    console.log(`t=${t} ${worker} claims ${r.id} attempt ${r.attempt} fence ${r.fence} lease until ${r.leaseUntil}`);
    return { id: r.id, fence: r.fence };
  }
  console.log(`t=${t} ${worker} finds nothing to claim`);
  return null;
}
function heartbeat(t, worker, c) {
  const r = runs.get(c.id);
  if (r.fence !== c.fence) { console.log(`t=${t} ${worker} heartbeat rejected (fence ${c.fence} != ${r.fence})`); return; }
  r.leaseUntil = t + LEASE;
  console.log(`t=${t} ${worker} heartbeat -> lease until ${r.leaseUntil}`);
}
function complete(t, worker, c) {
  const r = runs.get(c.id);
  if (r.fence !== c.fence) { console.log(`t=${t} ${worker} commit fence ${c.fence} REJECTED (current fence ${r.fence})`); return; }
  if (!effects.has(r.id)) effects.add(r.id); // side effect keyed by run_id: idempotent
  r.state = "DONE";
  console.log(`t=${t} ${worker} commit fence ${c.fence} accepted -> ${r.id} DONE`);
}
const c1 = claim(60, "W1");
heartbeat(70, "W1", c1);
console.log("t=71 W1 stalls (long GC pause / network partition), no more heartbeats");
claim(85, "W2");
const c2 = claim(101, "W2");
complete(105, "W2", c2);
complete(110, "W1", c1);
console.log(`side effects applied for report@60: ${[...effects].filter((e) => e === "report@60").length}`);

console.log("\n== 5. Misfire policies: job every 60s, scheduler down from t=130 to t=370 ==");
const lastFired = 120, every = 60, now = 370;
const missed = [];
for (let t = lastFired + every; t <= now; t += every) missed.push(t);
console.log(`missed fire times: ${missed.join(",")}`);
const policies = {
  "fire-all": missed,
  "fire-once": missed.slice(-1),
  "skip": [],
  "deadline-30s": missed.filter((t) => now - t <= 30),
};
for (const [name, list] of Object.entries(policies))
  console.log(`${name.padEnd(13)} -> runs ${list.length ? list.join(",") : "none"}; next fire ${missed[missed.length - 1] + every}`);