@studnicky/scheduler
Scheduler primitives: real-time (setTimeout/setInterval) and virtual (min-heap, deterministic) implementations.
Install
pnpm add @studnicky/schedulerUsage
Virtual scheduler (deterministic)
A stock-refresh job scheduled to run at a specific future moment shouldn't actually have to wait for that moment in a test — VirtualScheduler lets Northstar schedule one-shot tasks against virtual timestamps and fast-forward time in controlled steps instead. The example schedules tasks at virtual ms 100 and 200, then advances to 150 — firing only the first — before a second advance crosses 200 and fires the rest, proving tasks fire exactly when their due time is reached and not a moment sooner.
const counter = VirtualTimeCounter.create({ 'startMs': 0 });
const scheduler = VirtualScheduler.create({ 'counter': counter });
const fireOrder: number[] = [];
scheduler.scheduleAt(100, () => { fireOrder.push(100); });
scheduler.scheduleAt(200, () => { fireOrder.push(200); });
// Advance to 150 — only the task at 100 should fire.
scheduler.advance(150);
console.log('Fire order after advance(150):', fireOrder);
// Advance another 100 (total 250) — the task at 200 should now fire.
scheduler.advance(100);
console.log('Fire order after advance(100) more:', fireOrder);Interval tasks and cancellation
A repeating stock-refresh check needs to fire on a fixed cadence, and when Northstar shuts it down, it needs to actually stop — not keep firing from some orphaned timer. scheduleEvery sets up the repeating side of that story; cancelAll handles the shutdown. The example proves both: a 50ms interval advanced 200ms fires exactly four times, while an identical interval cancelled before any time advances never fires at all.
// --- Part 1: interval fires the expected number of times ---
const counter = VirtualTimeCounter.create({ 'startMs': 0 });
const scheduler = VirtualScheduler.create({ 'counter': counter });
let count = 0;
// Counter starts at 0; first fire at 0+50=50. Subsequent fires at 100, 150, 200.
scheduler.scheduleEvery(50, () => { count++; });
scheduler.advance(200);
console.log('Interval fire count:', count);
// --- Part 2: cancelAll() prevents further fires ---
const counter2 = VirtualTimeCounter.create({ 'startMs': 0 });
const scheduler2 = VirtualScheduler.create({ 'counter': counter2 });
let countAfterCancel = 0;
scheduler2.scheduleEvery(50, () => { countAfterCancel++; });
// Cancel before any advance — no tasks should fire.
scheduler2.cancelAll();
scheduler2.advance(200);
console.log('Fires after cancelAll():', countAfterCancel);Scheduler-aware sleep
Code that needs to pause for a moment shouldn't force its tests to actually sit and wait that moment out. Delay.sleep(ms, { clock?, scheduler?, signal? }) resolves through whichever scheduler you hand it, so a VirtualScheduler paired with a VirtualClockProvider sharing one counter makes the wait deterministic — no wall-clock timers, no flakiness. The example sleeps for a real 10ms first, then schedules a virtual 1000ms sleep, confirms it hasn't resolved yet, advances the counter by exactly 1000ms, and watches the promise resolve the instant that threshold is crossed.
// Real-time: resolves after ~10ms of wall-clock time.
await Delay.sleep(10);
console.log('Real-time sleep resolved');
// Virtual-time: resolves as soon as advance() crosses the requested delay,
// with no real wall-clock wait.
const counter = VirtualTimeCounter.create({ 'startMs': 0 });
const scheduler = VirtualScheduler.create({ 'counter': counter });
const clock = VirtualClockProvider.create(counter);
let resolved = false;
const sleepPromise = Delay.sleep(1_000, { 'clock': clock, 'scheduler': scheduler }).then(() => {
resolved = true;
});
console.log('Resolved before advance:', resolved);
scheduler.advance(1_000);
await sleepPromise;
console.log('Resolved after advance:', resolved);Reducer-with-effects process composition
A background job process sometimes needs more than a scheduler alone can give it — a state machine to track where the job is, effects to trigger scheduled work, and a cancellation signal to cut it short. The example below composes a StateMachine, EffectInterpreter, VirtualScheduler, and Signal directly into one job process: it distinguishes an acknowledgment that dispatches within the same drain cycle from a scheduled advance that arrives through interpreter.send() after the cycle ends, cancels pending scheduled work through an AbortSignal, and exercises both a deliberately rejected transition and an attempt to act on an already-terminated machine. It runs node-only, since it relies on node:assert.
import type { EffectHandlerInterface, FsmStepInterface } from '@studnicky/fsm/node';
/** processKitComposition — directly composes fsm's StateMachine + EffectInterpreter,
* scheduler's VirtualScheduler, and signal's Signal into a reducer-with-effects process.
* The `requestAck` effect uses the handler's same-cycle `dispatch` capability, while the
* delayed `scheduleAdvance` effect uses the interpreter's public `send()` path after its
* drain cycle ends. Virtual time keeps scheduling deterministic, and Signal supplies the
* cancellation boundary. Run: npx tsx examples/processKitComposition.ts */
import { VirtualTimeCounter } from '@studnicky/clock/node';
import { RuntimeError } from '@studnicky/errors/node';
// #region usage
import { EffectInterpreter, MachineTerminatedError, StateMachine, TransitionRejectedError } from '@studnicky/fsm/node';
import { Signal } from '@studnicky/signal/node';
import assert from 'node:assert/strict';
import type { ScheduledTaskInterface } from '../src/interfaces/index.js';
import type { JobEffectEntity } from './entities/JobEffectEntity.js';
import type { JobEventEntity } from './entities/JobEventEntity.js';
import type { JobStateEntity } from './entities/JobStateEntity.js';
import { VirtualScheduler } from '../src/index.js';
// --- Domain: a job that starts, self-acknowledges in the same cycle, waits for a scheduled
// advance, then settles. Two terminal outcomes (completed, cancelled) — reduce() stays a
// pure function of (state, event) throughout. ---
class JobProcess extends StateMachine<JobStateEntity.Type, JobEventEntity.Type, JobEffectEntity.Type> {
static make(): JobProcess { return new JobProcess(); }
getInitialState(): JobStateEntity.Type { return { 'variant': 'idle' }; }
reduce(
state: JobStateEntity.Type,
event: JobEventEntity.Type
): FsmStepInterface<JobStateEntity.Type, JobEffectEntity.Type> {
if (state.variant === 'idle' && event.type === 'start') {
return { 'effects': [{ 'variant': 'requestAck' }], 'state': { 'variant': 'waiting' } };
}
if (state.variant === 'waiting' && event.type === 'acknowledge') {
return { 'effects': [{ 'delayMs': 50, 'variant': 'scheduleAdvance' }], 'state': { 'variant': 'acknowledged' } };
}
if (state.variant === 'acknowledged' && event.type === 'advance') {
return { 'effects': [], 'state': { 'variant': 'completed' } };
}
if ((state.variant === 'waiting' || state.variant === 'acknowledged') && event.type === 'cancel') {
return { 'effects': [], 'state': { 'variant': 'cancelled' } };
}
// Deliberate rejection (e.g. cancelling before the job ever started) — distinct from a
// reducer defect, so callers can `instanceof`-check TransitionRejectedError.
throw new TransitionRejectedError({
'eventType': event.type,
'reason': `no transition defined for state '${state.variant}'`,
'stateVariant': state.variant
});
}
// Once settled, further transitions are rejected outright — reduce() is never called.
protected override isTerminated(state: JobStateEntity.Type): boolean {
const result = state.variant === 'completed' || state.variant === 'cancelled';
return result;
}
}
// `VirtualScheduler` gives this example a deterministic, fast clock — no real timers.
const counter = VirtualTimeCounter.create({ 'startMs': 0 });
const scheduler = VirtualScheduler.create({ 'counter': counter });
class Job {
static make(): {
'interpreter': EffectInterpreter<JobStateEntity.Type, JobEventEntity.Type, JobEffectEntity.Type>;
'scheduledTaskReference': { 'current': ScheduledTaskInterface | undefined };
'waitForScheduledDispatch': () => Promise<void>;
} {
const scheduledTaskReference: { 'current': ScheduledTaskInterface | undefined } = { 'current': undefined };
const scheduledDispatchReference: { 'current': Promise<void> | undefined } = { 'current': undefined };
const handler: EffectHandlerInterface<JobEffectEntity.Type, JobEventEntity.Type> = (effect, dispatch) => {
if (effect.variant === 'requestAck') {
dispatch({ 'type': 'acknowledge' });
return;
}
scheduledTaskReference.current = scheduler.scheduleAt(counter.nowMs() + effect.delayMs, () => {
scheduledTaskReference.current = undefined;
const scheduledDispatch = interpreter.send({ 'type': 'advance' });
scheduledDispatchReference.current = scheduledDispatch;
return scheduledDispatch;
});
};
const interpreter = EffectInterpreter.create<JobStateEntity.Type, JobEventEntity.Type, JobEffectEntity.Type>(JobProcess.make(), {
'handler': handler
});
const waitForScheduledDispatch = (): Promise<void> => {
const scheduledDispatch = scheduledDispatchReference.current;
if (scheduledDispatch === undefined) {
throw RuntimeError.create('Scheduled dispatch has not started');
}
return scheduledDispatch;
};
return {
'interpreter': interpreter,
'scheduledTaskReference': scheduledTaskReference,
'waitForScheduledDispatch': waitForScheduledDispatch
};
}
}
// Cancellation composed via Signal: an AbortSignal drives a 'cancel'
// event into the interpreter and cancels any still-pending scheduled advance.
class CancellationWiring {
static wire(
interpreter: EffectInterpreter<JobStateEntity.Type, JobEventEntity.Type, JobEffectEntity.Type>,
scheduledTaskReference: { 'current': ScheduledTaskInterface | undefined },
abortSignal: AbortSignal
): Promise<void> {
const result = new Promise<void>((resolve, reject) => {
abortSignal.addEventListener('abort', () => {
const pending = scheduledTaskReference.current;
if (pending !== undefined) { pending.cancel(); }
interpreter.send({ 'type': 'cancel' }).then(resolve, reject);
}, { 'once': true });
});
return result;
}
}
// #endregion usage
// --- Scenario A: start -> same-cycle self-advance to 'acknowledged' -> scheduled advance
// fires -> completed. ---
const signalSource = Signal.create();
const jobA = Job.make();
jobA.interpreter.start();
await jobA.interpreter.send({ 'type': 'start' });
// dispatch() already carried the machine from 'waiting' to 'acknowledged' within this one
// send() call — no second send() was needed for that leg.
assert.equal(jobA.interpreter.getState().variant, 'acknowledged');
scheduler.advance(50);
await jobA.waitForScheduledDispatch();
assert.equal(jobA.interpreter.getState().variant, 'completed');
console.log('Job A final state:', jobA.interpreter.getState().variant);
// --- Scenario B: start -> cancelled via Signal before the scheduled advance fires; the
// pending scheduled task is cancelled too, so advancing time afterward has no effect. ---
const jobB = Job.make();
const controllerB = new AbortController();
using composedSignalB = await signalSource.compose({ 'signal': controllerB.signal });
const cancellationB = CancellationWiring.wire(jobB.interpreter, jobB.scheduledTaskReference, composedSignalB.signal);
jobB.interpreter.start();
await jobB.interpreter.send({ 'type': 'start' });
assert.equal(jobB.interpreter.getState().variant, 'acknowledged');
assert.notEqual(jobB.scheduledTaskReference.current, undefined, 'a scheduled advance must be pending while acknowledged');
controllerB.abort();
await cancellationB;
assert.equal(jobB.interpreter.getState().variant, 'cancelled');
scheduler.advance(50);
assert.equal(jobB.interpreter.getState().variant, 'cancelled', 'cancelling must also cancel the pending scheduled task');
console.log('Job B final state:', jobB.interpreter.getState().variant);
// --- Scenario C: a deliberate rejection (TransitionRejectedError) is distinguishable from
// attempting a transition on an already-terminated machine (MachineTerminatedError). ---
const jobRejected = Job.make();
jobRejected.interpreter.start();
let rejectedByReducer = false;
try {
// 'cancel' has no defined transition from 'idle' — reduce() throws TransitionRejectedError.
await jobRejected.interpreter.send({ 'type': 'cancel' });
} catch (error) {
rejectedByReducer = error instanceof TransitionRejectedError;
}
assert.equal(rejectedByReducer, true);
// A fresh instance, driven all the way to its 'completed' terminal state, demonstrates the
// distinct MachineTerminatedError for an event sent after settlement.
const jobTerminated = Job.make();
jobTerminated.interpreter.start();
await jobTerminated.interpreter.send({ 'type': 'start' });
scheduler.advance(50);
await jobTerminated.waitForScheduledDispatch();
assert.equal(jobTerminated.interpreter.getState().variant, 'completed');
let rejectedByTermination = false;
try {
// The machine is terminated — send() surfaces MachineTerminatedError before reduce() ever
// runs again, distinguishing "process already settled" from "reducer rejected the event".
await jobTerminated.interpreter.send({ 'type': 'advance' });
} catch (error) {
rejectedByTermination = error instanceof MachineTerminatedError;
}
assert.equal(rejectedByTermination, true);
console.log('processKitComposition: all assertions passed');Public API
Import Delay, RealTimeScheduler, VirtualScheduler, and SchedulerError from @studnicky/scheduler/node; import PendingTaskInterface, ScheduledTaskInterface, and SchedulerProviderInterface from @studnicky/scheduler/interfaces. Construct schedulers through RealTimeScheduler.create() or VirtualScheduler.create({ counter }).
Extending
A WorkQueue that hard-codes VirtualScheduler can never be swapped to real timers in production, or back to virtual time in a test, without editing its internals. Depend on the injectable SchedulerProviderInterface instead, and any scheduler implementation can be passed in from outside. The example below wires a LoggingScheduler subclass — one that records every schedule and fire event — into a WorkQueue that only knows about the interface, then enqueues two labeled tasks and advances time to watch both the scheduler's log and the queue's processed output land in the right order.
/** VirtualScheduler subclass that appends lifecycle events to a log array. */
class LoggingScheduler extends VirtualScheduler {
public readonly log: SchedulerLogEntryEntity.Type[] = [];
public constructor(counter: Readonly<VirtualTimeCounter>) { super(counter); }
protected override onSchedule(id: string, _atMs: number, _variant: 'interval' | 'timeout'): void {
this.log.push({ 'event': 'schedule', 'id': id });
}
protected override onFire(id: string): void {
this.log.push({ 'event': 'fire', 'id': id });
}
}
/** Accepts any SchedulerProviderInterface — injectable for production/test swap. */
class WorkQueue {
readonly #scheduler: SchedulerProviderInterface;
public readonly processed: string[] = [];
public constructor(scheduler: SchedulerProviderInterface) {
this.#scheduler = scheduler;
}
public enqueue(atMs: number, label: string): void {
this.#scheduler.scheduleAt(atMs, () => { this.processed.push(label); });
}
}
const counter = VirtualTimeCounter.create({ 'startMs': 0 });
const loggingScheduler = new LoggingScheduler(counter);
const queue = new WorkQueue(loggingScheduler);
queue.enqueue(100, 'alpha');
queue.enqueue(200, 'beta');
loggingScheduler.advance(250);
console.log('Scheduler log:', loggingScheduler.log);
console.log('Processed labels:', queue.processed);Observability hooks
VirtualScheduler and RealTimeScheduler expose the same named set of protected lifecycle hooks, so a subclass written against one carries over to the other. One hook's declared return type differs between the two: VirtualScheduler.onFire is (id: string): void | Promise<void>, while RealTimeScheduler.onFire remains (id: string): void. Override any of them to add logging, metrics, or alerting without wiring the scheduler itself to any particular library.
VirtualScheduler hooks
| Hook | When it fires | Args |
|---|---|---|
onSchedule(id, atMs, variant) | After a task is inserted into the heap via scheduleAt or scheduleEvery | id: string, atMs: number, variant: 'timeout' | 'interval' |
onAdvance(deltaMs) | At the start of advance(), before the counter is incremented | deltaMs: number |
onRunUntil(atMs) | At the start of runUntil() | atMs: number |
onFire(id) | Immediately before a task's fire callback is invoked | id: string |
onFireError(id, error) | When a task's fire callback throws synchronously or returns a rejected Promise | id: string, error: unknown |
onReschedule(id, atMs) | After an interval task is re-inserted into the heap following a successful fire | id: string, atMs: number (next scheduled time) |
onCancel(id) | When a task's cancel() method is invoked | id: string |
onCancelAll() | At the end of cancelAll() | — |
onIdle() | After runUntil / runAll drains the heap, or after cancelAll | — |
RealTimeScheduler hooks
| Hook | When it fires | Args |
|---|---|---|
onSchedule(id, atMs, variant) | After a task is registered via scheduleAt or scheduleEvery | id: string, atMs: number, variant: 'timeout' | 'interval' |
onFire(id) | Inside the timer callback, immediately before fire is invoked | id: string |
onFireError(id, error) | When a task's fire callback throws synchronously or returns a rejected Promise | id: string, error: unknown |
onDrift(id, dueMs, actualMs, driftMs) | When a one-shot task fires later than its scheduled atMs | id: string, dueMs: number, actualMs: number, driftMs: number |
onMiss(id, atMs, nowMs) | When scheduleAt receives an atMs already in the past | id: string, atMs: number, nowMs: number |
onCancel(id) | When a task's cancel() method is invoked | id: string |
onCancelAll() | At the end of cancelAll(), after all timers are cleared | — |
onIdle() | After cancelAll fully drains all tracked tasks | — |
Demo trace (virtual scheduler)
class ObservedScheduler extends VirtualScheduler {
readonly events: string[] = [];
public constructor(counter: Readonly<VirtualTimeCounter>) {
super(counter);
}
protected override onSchedule(id: string, atMs: number, variant: 'interval' | 'timeout'): void {
const line = `[scheduler] schedule id=${id} atMs=${atMs.toString()} variant=${variant}`;
console.log(line);
this.events.push(line);
}
protected override onAdvance(deltaMs: number): void {
const line = `[scheduler] advance deltaMs=${deltaMs.toString()}`;
console.log(line);
this.events.push(line);
}
protected override onRunUntil(atMs: number): void {
const line = `[scheduler] runUntil atMs=${atMs.toString()}`;
console.log(line);
this.events.push(line);
}
protected override onFire(id: string): void {
const line = `[scheduler] fire id=${id}`;
console.log(line);
this.events.push(line);
}
protected override onFireError(id: string, error: Error): void {
const message = error.message;
const line = `[scheduler] fireError id=${id} error="${message}"`;
console.log(line);
this.events.push(line);
}
protected override onReschedule(id: string, atMs: number): void {
const line = `[scheduler] reschedule id=${id} nextAtMs=${atMs.toString()}`;
console.log(line);
this.events.push(line);
}
protected override onCancel(id: string): void {
const line = `[scheduler] cancel id=${id}`;
console.log(line);
this.events.push(line);
}
protected override onCancelAll(): void {
const line = '[scheduler] cancelAll';
console.log(line);
this.events.push(line);
}
protected override onIdle(): void {
const line = '[scheduler] idle';
console.log(line);
this.events.push(line);
}
}
const counter = VirtualTimeCounter.create({ 'startMs': 0 });
const scheduler = new ObservedScheduler(counter);
// Schedule a one-shot task at t=100
scheduler.scheduleAt(100, () => {
console.log('[task:one-shot] fired at t=100');
});
// Schedule an interval task every 150 ms
scheduler.scheduleEvery(150, () => {
console.log('[task:interval] fired');
});
// Schedule a task that throws — exercises onFireError
scheduler.scheduleAt(200, () => {
console.log('[task:failing] about to throw');
throw RuntimeError.create('task failure');
});
// Advance to t=300 — fires the one-shot at 100, the interval at 150, the failing at 200,
// and reschedules the interval to 300.
scheduler.advance(300);
// Cancel everything to exercise onCancelAll + onIdle
scheduler.cancelAll();The base class never calls any logger or metrics library. All hooks are no-ops by default.
Try it
This demo subclasses VirtualScheduler and overrides all nine protected lifecycle methods to print a full trace as three tasks run their course — a one-shot, a repeating interval, and one that deliberately throws. Watch every scheduleAt/scheduleEvery call emit schedule, each advance() emit advance then runUntil, the failing task trigger both fire and fireError, the interval reschedule itself after every fire, and cancelAll followed by idle close out the run.
Entities
@studnicky/scheduler/entities exports every schema namespace in src/entities.
import { SchedulerTaskDataEntity } from "@studnicky/scheduler/entities";Interfaces
@studnicky/scheduler/interfaces exports every TypeScript interface in src/interfaces, including configuration and state contracts.
import type { PendingTaskInterface } from "@studnicky/scheduler/interfaces";What it is
@studnicky/scheduler is a real-time and deterministic virtual scheduling primitive. It schedules and cancels local work against a selected clock; it does not own a job service, persistence layer, business calendar, or workflow.
What it is for
Northstar Books uses a scheduler when a local process needs a delay, a follow-up, or repeatable time-driven behaviour in a test. A Node or browser runtime entrypoint selects the safe platform surface, while entities validate scheduled-task data and interfaces let Northstar supply a scheduler without coupling its domain process to a timer implementation.
Northstar Books examples
- Scheduler lifecycle hooks solves the “observe scheduled stock-refresh work without embedding monitoring in the scheduler” problem. It traces one-shot and interval tasks, failures, rescheduling, cancellation, and idle state, proving that Northstar can attach telemetry while the scheduling result remains unchanged.
Public entrypoints
| Import path | Use it when |
|---|---|
@studnicky/scheduler/node | Northstar schedules local server-side catalogue refreshes, delayed work, or deterministic test tasks. |
@studnicky/scheduler/browser | Northstar schedules browser-side reader or bookseller interface work with the same primitive contract. |
@studnicky/scheduler/entities | Northstar validates scheduler task data at an application boundary. |
@studnicky/scheduler/interfaces | Northstar accepts an injectable scheduler port or types pending and scheduled tasks without selecting an implementation. |
Exports
| Symbol | Purpose | Import path |
|---|---|---|
Delay | Provides delay functionality. | @studnicky/scheduler/node |
RealTimeScheduler | Provides real time scheduler functionality. | @studnicky/scheduler/node |
SchedulerError | Represents scheduler failures. | @studnicky/scheduler/node |
SchedulerProviderInterface | Defines the scheduler provider contract. | @studnicky/scheduler/interfaces |
VirtualScheduler | Provides virtual scheduler functionality. | @studnicky/scheduler/node |