14 KiB
14 KiB
@suaveplan/job-scheduler - Universal in-process job scheduler with cron + queued runs, retries, timeouts, concurrency control, and OTel observability
Summary
Universal in-process job scheduler with cron + queued runs, retries, timeouts, concurrency control, and OpenTelemetry observability. Built on @suaveplan/cron for expression parsing and next-fire computation.
Documentation tree
- errors — (no submodules)
- job-scheduler-error — (no submodules)
- events — (no submodules)
- events — (no submodules)
- execution — (no submodules)
- job-execution — (no submodules)
- otel — (no submodules)
- job-scheduler-otel — (no submodules)
- queue — (no submodules)
- job-queue — (no submodules)
- registry — (no submodules)
- job-registry — (no submodules)
- cron-rescheduler — Houses the two cron-related decisions the scheduler makes after a job settles or resumes: (1) compute the next fire fro…
- job-dispatcher — Encapsulates the "hand a job to the executor" step of the scheduler's tick loop. The scheduler core decides which job…
- job-scheduler — The
JobSchedulerclass is the package's main entry point: it accepts both cron-scheduled and one-shot queued jobs, ru… - job-snapshot — Projects a mutable
ScheduledJobrecord (the scheduler's internal data structure) to an immutableJobStatussnapshot…
- job-dispatcher — Encapsulates the "hand a job to the executor" step of the scheduler's tick loop. The scheduler core decides which job…
- job-scheduler — The
JobSchedulerclass is the package's main entry point: it accepts both cron-scheduled and one-shot queued jobs, ru… - job-snapshot — Projects a mutable
ScheduledJobrecord (the scheduler's internal data structure) to an immutableJobStatussnapshot… - job-scheduler-error — (no submodules)- events — (no submodules)- events — (no submodules)- execution — (no submodules)- job-execution — (no submodules)- otel — (no submodules)- job-scheduler-otel — (no submodules)- queue — (no submodules)- job-queue — (no submodules)- registry — (no submodules)- job-registry — (no submodules)- cron-rescheduler — Houses the two cron-related decisions the scheduler makes after a job settles or resumes: (1) compute the next fire fro…
- job-dispatcher — Encapsulates the "hand a job to the executor" step of the scheduler's tick loop. The scheduler core decides which job…
- job-scheduler — The
JobSchedulerclass is the package's main entry point: it accepts both cron-scheduled and one-shot queued jobs, ru… - job-snapshot — Projects a mutable
ScheduledJobrecord (the scheduler's internal data structure) to an immutableJobStatussnapshot… - job-scheduler-error — (no submodules)- events — (no submodules)- events — (no submodules)- execution — (no submodules)- job-execution — (no submodules)- otel — (no submodules)- job-scheduler-otel — (no submodules)- queue — (no submodules)- job-queue — (no submodules)- registry — (no submodules)- job-registry — (no submodules)- cron-rescheduler — Houses the two cron-related decisions the scheduler makes after a job settles or resumes: (1) compute the next fire fro…
- job-dispatcher — Encapsulates the "hand a job to the executor" step of the scheduler's tick loop. The scheduler core decides which job…
- job-scheduler — The
JobSchedulerclass is the package's main entry point: it accepts both cron-scheduled and one-shot queued jobs, ru… - job-snapshot — Projects a mutable
ScheduledJobrecord (the scheduler's internal data structure) to an immutableJobStatussnapshot… - job-scheduler-error — (no submodules)- events — (no submodules) - events — (no submodules)- execution — (no submodules) - job-execution — (no submodules)- otel — (no submodules) - job-scheduler-otel — (no submodules)- queue — (no submodules) - job-queue — (no submodules)- registry — (no submodules) - job-registry — (no submodules)- cron-rescheduler — Houses the two cron-related decisions the scheduler makes after a job settles or resumes: (1) compute the next fire fro…
- job-dispatcher — Encapsulates the "hand a job to the executor" step of the scheduler's tick loop. The scheduler core decides which job…
- job-scheduler — The
JobSchedulerclass is the package's main entry point: it accepts both cron-scheduled and one-shot queued jobs, ru… - job-snapshot — Projects a mutable
ScheduledJobrecord (the scheduler's internal data structure) to an immutableJobStatussnapshot…
Install
bun add @suaveplan/job-scheduler
Quick start
import { createJobScheduler } from "@suaveplan/job-scheduler";
const scheduler = createJobScheduler();
scheduler.schedule("@hourly", async ({ logger }) => {
logger.info("hourly heartbeat");
});
scheduler.queue(async () => {
await flushQueue();
}, { delayMs: 5_000 });
scheduler.start();
// On graceful shutdown:
await scheduler.stop();
Retries + timeout
scheduler.schedule(
"*/15 * * * *",
async ({ signal, attempt }) => {
const res = await fetch("/api/sync", { signal });
if (!res.ok) throw new Error("upstream failed");
if (attempt > 1) console.warn("recovered after retry");
},
{ retries: 3, retryDelayMs: 1_000, timeout: 30_000, tz: "UTC" }
);
Deterministic testing with FakeClock
import { createJobScheduler } from "@suaveplan/job-scheduler";
interface PendingTimer { fireAt: number; cb: () => void }
const state = { now: 0, nextId: 1, pending: new Map<number, PendingTimer>() };
const timers = {
setTimer: (cb: () => void, ms: number) => {
const id = state.nextId++;
state.pending.set(id, { fireAt: state.now + ms, cb });
return id;
},
clearTimer: (h: unknown) => state.pending.delete(h as number),
};
async function advance(target: number) {
while (true) {
let id: number | undefined;
let at = Infinity;
for (const [k, v] of state.pending) {
if (v.fireAt <= target && v.fireAt < at) { id = k; at = v.fireAt; }
}
if (id === undefined) break;
const t = state.pending.get(id)!;
state.pending.delete(id);
state.now = t.fireAt;
t.cb();
await Promise.resolve();
}
state.now = target;
}
const scheduler = createJobScheduler({
clock: () => state.now,
timers,
});
API summary
| Method | Purpose |
|---|---|
schedule(cronExpr, fn, opts?) |
Run fn on every cron fire; returns jobId. |
queue(fn, opts?) |
One-shot run with delayMs or absolute runAt. |
cancel(jobId) |
Remove from queue + abort in-flight attempt. |
pause(jobId?) / resume(jobId?) |
Per-job or global pause/resume. |
jobs() |
Frozen JobStatus[] snapshot for observability. |
start() |
Begin the tick loop (idempotent). |
stop() |
Drain in-flight + stop ticking (idempotent). |
Relationship to @suaveplan/cron
This package consumes @suaveplan/cron's parseCron + CronExpression and is the recommended entry point for any application that needs both cron-style and ad-hoc scheduling. The cron parser stays narrow and easy to audit; this scheduler adds retries, timeouts, concurrency, lifecycle, and observability on top.
Events
The scheduler emits the following topics on the bus passed via JobSchedulerOptions.events:
job:start— fired before each attempt.job:complete— successful settle.job:fail— terminal failure for an attempt (retries-permitting, the next attempt fires anotherjob:start).job:scheduling_exhausted— cron expression has no future fire; job is auto-cancelled.job:paused/job:resumed/job:cancelled— lifecycle transitions.
Use the JOB_SCHEDULER_EVENTS constants instead of literal strings to get compile-time guarantees that subscribers match emitters.
Observability
The package records into the @suaveplan/job-scheduler OTel meter:
job_scheduler.duration_ms(histogram, labelledoutcome+kind).job_scheduler.runs_total(counter, labelledoutcome+kind).job_scheduler.failures_total(counter, labelledkind).job_scheduler.cron_exhausted_total(counter).job_scheduler.queue_depth(histogram, sampled per tick).
outcome ∈ { success, error, timeout } and kind ∈ { cron, queued } — both closed enums to keep label cardinality bounded.