Design a Job Scheduler

# CronJob reached batch/v1 in Kubernetes 1.21. .spec.timeZone went beta and on
# by default in 1.25 and stable in 1.27, so 1.25 and 1.26 honour it as well. On
# 1.24 it is alpha and needs the CronJobTimeZone feature gate; on anything older
# the field does not exist and is dropped, and the schedule then runs in the
# controller manager's zone, usually UTC - silently.
apiVersion: batch/v1
kind: CronJob
metadata:
  name: nightly-invoices
spec:
  schedule: "30 2 * * *"
  timeZone: "Europe/London"      # 02:30 does not exist on the spring-forward day
  concurrencyPolicy: Forbid      # default is Allow, which lets a slow run
                                 # overlap itself; Replace kills the running one
  startingDeadlineSeconds: 300   # also caps how many missed starts accumulate
  successfulJobsHistoryLimit: 3
  failedJobsHistoryLimit: 5
  jobTemplate:
    spec:
      backoffLimit: 2             # up to 3 pod attempts: at-least-once, again
      activeDeadlineSeconds: 3600 # a hung job is failed rather than left running
      template:
        spec:
          restartPolicy: Never
          terminationGracePeriodSeconds: 120   # time to checkpoint after SIGTERM
          containers:
            - name: invoices
              image: registry.example.com/invoices:1.8.3
              args: ["run", "nightly-invoices"]
---
# The lease the control plane itself uses for leader election, for comparison.
apiVersion: coordination.k8s.io/v1
kind: Lease
metadata:
  name: invoices-scheduler
  namespace: billing
spec:
  holderIdentity: invoices-scheduler-7d9f-abc12
  leaseDurationSeconds: 15       # a candidate waits this long past renewTime
  renewTime: "2026-03-01T02:29:58.000000Z"

// The job runs in a worker rather than in the scheduler, so restarting the
// scheduler cannot kill it. Restarting the *worker* still can, so: claim with a
// heartbeat, checkpoint progress, and handle SIGTERM.
const STALE = "interval '60 seconds'";

export async function runJob(db, { jobId, scheduledFor }, work) {
  // Claim it, or take it over if whoever had it stopped heartbeating. This
  // reclaim is exactly what makes the system at-least-once: a worker that is
  // merely slow and partitioned can have its job taken.
  const claimed = await db.query(
    `UPDATE job_runs
        SET state = 'running', worker = $3, heartbeat_at = now()
      WHERE job_id = $1 AND scheduled_for = $2
        AND (state = 'queued'
             OR (state = 'running' AND heartbeat_at < now() - ${STALE}))
      RETURNING checkpoint`,
    [jobId, scheduledFor, process.env.HOSTNAME],
  );
  if (claimed.rowCount === 0) return { skipped: true };   // someone else has it

  const beat = setInterval(() => {
    db.query(`UPDATE job_runs SET heartbeat_at = now()
               WHERE job_id = $1 AND scheduled_for = $2`, [jobId, scheduledFor])
      .catch((err) => console.error({ msg: 'heartbeat failed', err }));
  }, 15_000);

  const stopping = new AbortController();
  // A deploy sends SIGTERM and then waits terminationGracePeriodSeconds. Use
  // that window to finish the current batch and checkpoint, not to start one.
  process.once('SIGTERM', () => stopping.abort());

  try {
    await work({
      from: claimed.rows[0].checkpoint ?? null,
      signal: stopping.signal,
      // Called by the job after each batch. The checkpoint shortens the replay
      // after a crash; it does not prevent one, so each batch must still be
      // idempotent on its own.
      checkpoint: (c) => db.query(
        `UPDATE job_runs SET checkpoint = $3
          WHERE job_id = $1 AND scheduled_for = $2`,
        [jobId, scheduledFor, c],
      ),
    });
    await db.query(`UPDATE job_runs SET state = 'done'
                     WHERE job_id = $1 AND scheduled_for = $2`, [jobId, scheduledFor]);
    return { done: true };
  } finally {
    clearInterval(beat);   // runs on the abort path too, so the row goes stale
  }                        // and another worker can resume from the checkpoint
}

// The scheduler was down from 01:00 to 05:00 and four hourly ticks passed.
// There is no correct general answer, only three policies. Set one per job;
// do not inherit whichever one your framework happens to implement.
import { CronExpressionParser } from 'cron-parser';

export function missedTicks(job, lastDispatched, now = new Date()) {
  const it = CronExpressionParser.parse(job.schedule, {
    tz: job.timeZone,
    currentDate: lastDispatched,
    endDate: now,
  });

  const ticks = [];
  while (it.hasNext() && ticks.length <= job.maxCatchUp) {
    ticks.push(new Date(it.next().getTime()));
  }

  switch (job.catchUp) {
    case 'skip':
      // Hourly cache warm, metrics rollup: the next run supersedes all of them.
      return [];
    case 'latest':
      // Nightly report: run it once, now, for the most recent window only.
      return ticks.slice(-1);
    case 'all':
      // Billing, ledger postings: every window must be processed exactly once,
      // in order. Bound it - replaying 2,000 ticks at once is its own incident,
      // which is the same hazard behind Kubernetes refusing to schedule after
      // 100 missed start times.
      if (ticks.length > job.maxCatchUp) {
        throw new Error(
          `${job.id}: ${ticks.length} missed ticks exceeds maxCatchUp=${job.maxCatchUp};` +
          ' refusing to replay automatically',
        );
      }
      return ticks;
    default:
      throw new Error(`${job.id}: no catchUp policy set`);
  }
}

// Note what this needs to work at all: a persisted lastDispatched per job, and
// a scheduled_for on every run. A scheduler that only knows "now" cannot have
// a catch-up policy, because it cannot name the ticks it missed.