@chamvi/scheduler

Fires chamvi cron triggers once per tick across replicas, on hubwire. The public API of packages/scheduler/src/index.ts.

npm install @chamvi/scheduler @chamvi/core --save

@chamvi/core is a peer dependency. See Cron Scheduling for the behaviour.

createScheduler

function createScheduler(opts: SchedulerOptions): IScheduler;

const SCHEDULE_LOCK_TTL_MS: 600_000;

Schedules engine.listAgents() (or only the agents named in agents) and locks over engine.store. Throws InvalidAgentDefinitionError for a cron trigger the scheduler cannot use (expression, timezone, lockTtlMs) and for a name in agents the engine lacks or that is given twice.

/*
 * Copyright (c) 2026.
 * Author Peter Placzek (tada5hi)
 * For the full copyright and license information,
 * view the LICENSE file that was distributed with this source code.
 */

import type { Engine } from '@chamvi/core';

export type SchedulerErrorContext = {
    agent: string;
    trigger: string;
};

export type SchedulerOptions = {
    /** Its agents' cron triggers are scheduled; its store holds the tick locks. */
    engine: Engine;
    /**
     * Names of the engine's agents to schedule, e.g. to split cron triggers
     * across deployments. Default: every agent of the engine.
     */
    agents?: string[];
    /** Lock owner. Default `${hostname()}-${pid}`. */
    owner?: string;
    /**
     * How long a tick's lock key is held; a tick whose timer fires this late or
     * later is skipped. Default SCHEDULE_LOCK_TTL_MS.
     */
    lockTtlMs?: number;
    /**
     * Receives a failing tick (lock store, overlap check or dispatch). Ticks
     * another replica won are not errors and are not reported.
     * Default console.error('[chamvi]', error, ctx).
     */
    onError?: (error: unknown, ctx: SchedulerErrorContext) => void;
};

export interface IScheduler {
    /** Schedules every cron trigger; never fires on start. Idempotent. */
    start(): Promise<void>;
    /** Unschedules every trigger, then waits for ticks in flight. */
    stop(): Promise<void>;
}

runJob

function runJob(opts: RunJobOptions): Promise<RunJobResult>;

Rejects with InvalidAgentDefinitionError when the trigger is missing or ambiguous.

/*
 * Copyright (c) 2026.
 * Author Peter Placzek (tada5hi)
 * For the full copyright and license information,
 * view the LICENSE file that was distributed with this source code.
 */

import type {
    Agent,
    Engine,
    RunStatusLiteral,
} from '@chamvi/core';

export type RunJobOptions = {
    /**
     * Not started yet; runJob starts and drains it, also when the tick is
     * skipped. Its store is polled and its clock gives scheduledAt and drives
     * the poll.
     */
    engine: Engine;
    agent: Agent;
    /** Cron trigger id; optional when the agent has exactly one cron trigger. */
    trigger?: string;
    /** Fallback poll of engine.getRun, default 1_000. */
    pollMs?: number;
    /** Passed to engine.drain, default 30_000. */
    drainTimeoutMs?: number;
};

export type RunJobResult = {
    runId?: string;
    /**
     * The run's status when the job returned: `completed`, `failed`,
     * `cancelled`, `waiting`, `queued` (behind a run of its concurrency key
     * that waits for an approval), or 'skipped' (overlap).
     */
    status: RunStatusLiteral | 'skipped';
    /** 0 for completed, waiting, queued and skipped; 1 for failed and cancelled. */
    exitCode: 0 | 1;
};

Building blocks

function fireCronTrigger(ctx: CronTickContext): Promise<CronTickResult>;
function readCronOptions(agent: Agent, trigger: TriggerDefinition): CronTriggerOptions;
function createStoreScheduleLock(store: ILockStore, owner: string): IScheduleLock;

fireCronTrigger is one tick (overlap check and dispatch) for an explicit scheduledAt; createScheduler and runJob both call it. createStoreScheduleLock adapts core's ILockStore to hubwire's IScheduleLock.

/*
 * Copyright (c) 2026.
 * Author Peter Placzek (tada5hi)
 * For the full copyright and license information,
 * view the LICENSE file that was distributed with this source code.
 */

import type {
    Agent,
    Engine,
    OverlapPolicy,
    TriggerDefinition,
} from '@chamvi/core';

export type CronTriggerOptions = {
    expression: string;
    timezone?: string;
    overlap: OverlapPolicy;
    payload?: unknown;
};

export type CronTickContext = {
    engine: Engine;
    agent: Agent;
    trigger: TriggerDefinition;
    /** The tick's planned fire time; forms the idempotency key. */
    scheduledAt: Date;
};

export type CronTickDispatched = {
    dispatched: true;
    runId: string;
    deduplicated: boolean;
};

export type CronTickSkipped = {
    dispatched: false;
    reason: 'overlap';
};

export type CronTickResult = CronTickDispatched | CronTickSkipped;