@chamvi/channel-cron

The chamvi cron channel: fires cron subscriptions once per tick across replicas, on hubwire.
Every replica registers the same channel. A lock in the store elects one per tick,
and the same package fires a subscription one-shot for a Kubernetes CronJob.

Table of Contents

• Installation
• Usage
• How a tick works
• Overlap policies
• Timezones
• Shutdown
• One-shot jobs
• Options
• API
• License

Installation

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

ESM only, Node.js >=26.0.0. @chamvi/core is a peer dependency; hubwire (croner inside) and cron-parser are installed with the package.

Usage

cron.on() subscribes an agent to the cron channel; cronChannel() is the channel the engine starts:

import type { LanguageModel } from 'ai';
import {
    Engine,
    createMemoryStore,
    defineModelAgent,
} from '@chamvi/core';
import { cron, cronChannel } from '@chamvi/channel-cron';

export async function startScheduling(model: LanguageModel) {
    const sweep = defineModelAgent({
        name: 'sweep',
        model,
        instructions: 'Close stale issues older than 30 days.',
        // every 30 minutes, Berlin time; skipped while the previous run is active
        channels: [cron.on('*/30 * * * *', { timezone: 'Europe/Berlin' })],
    });

    // createMemoryStore is per process; with several replicas, pass one shared store
    // (e.g. createDatabaseStore) so its lock fires each tick once across them
    const engine = new Engine({
        store: createMemoryStore(),
        agents: [sweep],
        // every replica registers the same channel; it reads agents and store from the engine
        channels: [cronChannel()],
    });

    // starts the cron channel after recovering runs
    await engine.start();

    // shutdown: drain() stops the channel first, then lets the runs in flight finish
    // await engine.drain();
    return { engine };
}

cron.on(expression, options?) validates the expression (cron-parser) and the timezone and throws InvalidAgentDefinitionError for one it refuses. engine.start() starts the channel after recovering runs: the subscription of an agent passed in code that the scheduler cannot use (an expression croner refuses) fails engine.start() with InvalidAgentDefinitionError naming the agent and subscription, an agent folder's goes to onError and leaves that agent unscheduled. Subscriptions of other channels are ignored. The channel follows agents registered, replaced and removed at runtime.

A second cron channel (e.g. one per region) needs its own name; its on() makes subscriptions only it fires:

const eu = cronChannel({ name: 'cron-eu' });
defineModelAgent({ name: 'digest', channels: [eu.on('0 8 * * *', { timezone: 'Europe/Berlin' })] /* ... */ });
new Engine({ channels: [cronChannel(), eu] /* ... */ });

How a tick works

• One component per subscription. Each becomes a hubwire scheduled component named cron:<agent>:<subscriptionId>.
• One replica per tick. Before a tick runs, the replica acquires the store lock cron:<agent>:<subscriptionId>:<scheduledAt ISO> (through ILockRepository) for lockTtlMs (default 10 minutes). The key is never released: a replica that fires the same tick late cannot win it again.
• Deduplicated dispatch. The winner dispatches through the channel (envelope channel = the channel's name, trigger = the subscription id) with payload options.payload ?? { scheduledAt: '<ISO>' } and the idempotency key <subscriptionId>@<scheduledAt ISO>, so even a second firing of the same tick returns the existing run. A subscription's own idempotencyKey() is ignored for cron.
• Never on start. Starting or restarting a replica fires nothing; the first dispatch is at the next scheduled tick.
• No catch-up. A tick that falls into a restart, or whose timer fires lockTtlMs or more late (a suspended process), is not fired later.
• Failures are reported, not thrown. A failing lock store, overlap check or dispatch reaches onError(error, { agent, subscription }); the next tick runs normally.

Overlap policies

cron.on(expression, { overlap }) sets the policy; the channel enforces it before dispatching. "Active" means a run of the same subscription in queued, running, waiting or interrupted.

A concurrencyKey() on the subscription wins over the default. The default key is what makes queue-one wait: the engine starts one run per agent and key at a time, so the queued run starts when the active one settles. It also makes a race harmless: a tick that checks while the previous tick's dispatch is still in flight may dispatch a second run, which then waits instead of running alongside.

Timezones

cron.on(expression, { timezone: 'Europe/Berlin' }) takes an IANA zone; without one, the process's local zone applies. At DST changes:

• A local time that occurs twice (fall-back) fires once, at its first occurrence.
• A local time the spring-forward change skips fires once, an hour later in local time (30 2 * * * fires at 03:30).

Shutdown

engine.drain() (and engine.stop()) stop the channel first (no new ticks; ticks in flight finish), then drain the runs:

await engine.drain(); // stops the cron channel, waits for running runs, then interrupts the rest

One-shot jobs

runJob() fires one cron subscription once and waits for its run, for a Kubernetes CronJob instead of a long-lived replica:

const { exitCode } = await runJob({ engine, agent: sweep });
process.exitCode = exitCode;

It dispatches the tick at the current time (the overlap policy applies), starts the engine with start({ channels: false }) (so a job never starts cron schedules or other channels) and runs one approval tick (also when the tick is skipped), waits until the run is completed, failed, cancelled or waiting, or queued behind a run of the same subscription that waits for an approval, then calls engine.drain(). The engine must hold the subscription's cron channel (it refuses the agent otherwise).

A run that pauses for an approval does not hold the pod: the job returns, the approval is answered (or times out), and the answered run is executed by a long-lived replica or by the next job pod, which starts the engine even when its own tick is skipped. That pod also recovers a run whose pod was killed and denies approvals past their timeout. Caveats:

• engine.start() recovers and executes every queued run of every agent in the shared store, not only the cron run.
• drain() interrupts whatever still runs after drainTimeoutMs. runJob waits only for the run it dispatched, so a leftover run is finished by a long-lived replica or by a later job pod, within that pod's drainTimeoutMs.
• Size drainTimeoutMs to the job's runs, and prefer a long-lived deployment with cronChannel() when runs are long.

Options

cronChannel(options):

cron.on(expression, options):

runJob(options):

API

Types: CronChannel, CronChannelOptions, CronChannelErrorContext, CronSubscriber, CronSubscriptionOptions, OverlapPolicy, RunJobOptions, RunJobResult, CronOptions, CronTickContext, CronTickResult, CronTickDispatched, CronTickSkipped.

License

Published under Apache License 2.0.