@chamvi/channel-cron

The cron channel: fires cron.on() subscriptions once per tick across replicas, on hubwire. The public API of packages/channel-cron/src/index.ts.

npm install @chamvi/channel-cron @chamvi/core --save

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

cronChannel

function cronChannel(options?: CronChannelOptions): CronChannel; // IChannel with start/stop, plus on() bound to its name

const CRON_CHANNEL_NAME: 'cron';
const SCHEDULE_LOCK_TTL_MS: 600_000;

/*
 * 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 { IChannel } from '@chamvi/core';
import type { CronSubscriber } from '../subscription/types.ts';

export type CronChannelErrorContext = {
    agent: string;
    /** The subscription id. */
    subscription: string;
};

export type CronChannelOptions = {
    /** The channel name; default `cron`. A second instance needs its own. */
    name?: string;
    /** Owner of the tick locks. 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) and a
     * folder agent's subscription hubwire cannot schedule. Ticks another
     * replica won are not errors. Default console.error('[chamvi]', error, ctx).
     */
    onError?: (error: unknown, ctx: CronChannelErrorContext) => void;
};

/** The cron channel: background inbound only (no `in`, no `out`), with `on()` bound to its name. */
export type CronChannel = IChannel & CronSubscriber;

engine.start() starts the channel: it schedules the cron subscriptions of this channel (subscription.channel === name) of engine.listAgents() and locks over engine.store. A code agent's subscription the scheduler cannot use (expression, timezone) fails engine.start() with InvalidAgentDefinitionError; a folder agent's (one the engine holds at a revision) goes to onError instead and leaves that agent unscheduled. An invalid name (the agent-name rule), an empty owner or a lockTtlMs that is no positive integer throws InvalidAgentDefinitionError when cronChannel() is called.

It follows the engine's agents.changed events: an added agent is scheduled, a replaced one rescheduled, a removed one unscheduled. A replacement whose cron subscription hubwire cannot schedule goes to onError and leaves that agent unscheduled. stop() (called first by engine.drain() and engine.stop()) stops following and waits for the ticks in flight, also of replaced and removed agents.

function assertSchedulable(agent: Agent, channel?: string): void; // channel default 'cron'

Throws the same error as the start for an agent's subscriptions of channel, without scheduling anything; the app runs it on every folder agent before registering it.

cron.on

const cron: CronSubscriber; // cron.on(expression, options?): a subscription of the channel named 'cron'

/*
 * 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 { ChannelSubscription } from '@chamvi/core';

/** What a tick does while a run of the same subscription is active. */
export type OverlapPolicy = 'skip' | 'queue-one' | 'allow';

export type CronSubscriptionOptions = {
    /** Shared-store execution cap within the agent; omitted or false adds no cap. */
    concurrencyMaxRuns?: number | false;
    /** IANA zone; default the process's local zone. */
    timezone?: string;
    /** Default `skip`. */
    overlap?: OverlapPolicy;
    /** The run's payload; default `{ scheduledAt: '<ISO>' }`. */
    payload?: unknown;
    /** The subscription id, unique per agent; default `cron:<expression>`. */
    id?: string;
};

/** Makes cron subscriptions for one channel name. */
export type CronSubscriber = {
    on(expression: string, options?: CronSubscriptionOptions): ChannelSubscription;
};

Throws InvalidAgentDefinitionError for an expression cron-parser refuses, an unknown timezone, an unknown overlap or an empty id. The id defaults to cron:<expression>.

runJob

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

Starts the engine with start({ channels: false }). Rejects with InvalidAgentDefinitionError when the subscription is missing or ambiguous, or the engine holds no cron channel for it.

/*
 * 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,
    IEngine,
    RunStatusLiteral,
} from '@chamvi/core';

export type RunJobOptions = {
    /**
     * Not started yet; runJob starts it without channels and drains it, also
     * when the tick is skipped. It holds the subscription's cron channel. Its store is polled and its clock gives scheduledAt and drives
     * the poll.
     */
    engine: IEngine;
    agent: Agent;
    /**
     * Cron subscription id; optional when the agent has exactly one
     * subscription of a cron channel the engine holds.
     */
    subscription?: string;
    /** Fallback poll of engine.getRun, default 1_000. */
    pollMs?: number;
    /** Passed to engine.drain, default 30_000. */
    drainTimeoutMs?: number;
    /**
     * Stops the job: already aborted, nothing is fired and the engine is never
     * started (`skipped`); aborted while waiting, the wait ends and the engine
     * drains, and the result is the run as stored after the drain. The abort
     * reason is never thrown.
     */
    signal?: AbortSignal;
};

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, or a signal
     * aborted before the call). After an abort while waiting: the status the
     * drain left, usually `interrupted`, or `running` when the drain timed out
     * before the run was marked.
     */
    status: RunStatusLiteral | 'skipped';
    /** 1 for failed and cancelled; 0 otherwise (an interrupted run is resumed by a replica). */
    exitCode: 0 | 1;
};

Building blocks

function fireCronSubscription(ctx: CronTickContext): Promise<CronTickResult>;
function readCronOptions(agent: Agent, subscription: ChannelSubscription): CronOptions;
function createRepositoryScheduleLock(store: ILockRepository, owner: string): IScheduleLock;

fireCronSubscription is one tick (overlap check and dispatch) for an explicit scheduledAt; the channel and runJob both call it. createRepositoryScheduleLock adapts core's ILockRepository 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,
    ChannelContext,
    ChannelSubscription,
    IEngine,
} from '@chamvi/core';
import type { OverlapPolicy } from '../subscription/types.ts';

/** The options `cron.on()` stores on a subscription, read back and checked. */
export type CronOptions = {
    expression: string;
    timezone?: string;
    overlap: OverlapPolicy;
    payload?: unknown;
};

export type CronTickContext = {
    engine: IEngine;
    agent: Agent;
    subscription: ChannelSubscription;
    /** The tick's planned fire time; forms the idempotency key. */
    scheduledAt: Date;
    /**
     * The channel's dispatch (sets the channel name itself). Default
     * engine.dispatch with `channel` = the subscription's channel.
     */
    dispatch?: ChannelContext['dispatch'];
};

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

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

export type CronTickResult = CronTickDispatched | CronTickSkipped;