@setu-ts/scheduler-plugin

Job scheduling. Registers an IScheduler under CAPABILITIES.SCHEDULER ('scheduler').

Cron, fixed-interval, and one-shot delayed jobs, with retry/backoff, pause/resume, and optional distributed locking. The 5-field UTC cron parser is implemented here — no dependency.

Installation

import { SchedulerPlugin } from '@setu-ts/scheduler-plugin';

Usage

import { createApplication } from '@setu-ts/kernel';
import { RuntimePlugin } from '@setu-ts/runtime';
import { SchedulerPlugin } from '@setu-ts/scheduler-plugin';
import { CAPABILITIES, type IScheduler } from '@setu-ts/common';

// Your application's own work — stand-ins so this example compiles as written.
declare function buildReport(): Promise<void>;
declare function pollInbox(): Promise<void>;
declare function warmCaches(): Promise<void>;

const app = createApplication({
  plugins: [
    RuntimePlugin(),
    SchedulerPlugin({ distributedLock: { enabled: true, storage: 'redis' } }),
  ],
});
await app.start({ port: 3000 });

const scheduler = app.services.get<IScheduler>(CAPABILITIES.SCHEDULER);

// 5-field UTC cron
await scheduler.cron('nightly-report', '0 2 * * *', async () => {
  await buildReport();
});

// Fixed interval, and a one-shot delay
await scheduler.every('poll-inbox', 30_000, async () => await pollInbox());
await scheduler.delay('warmup', 5_000, async () => await warmCaches());

await scheduler.pause('poll-inbox');
await scheduler.resume('poll-inbox');

Timing and registration refusals

delay, every, cron and retry backoff honour long waits in timer chunks of at most 2147483647 ms, checking the wall clock on every wake. A job never runs before its due time, even after an early wake or a clock step backwards. There is no upper limit on a delay or interval: delay(30 days) waits the full 30 days, and monthly cron jobs wait until their next due instant. Previously, Node, Deno and Bun clamped an over-limit timer to about 1 ms, so these jobs could fire early.

Invalid time values now make the async registration call reject with a RangeError:

Each message names the job, option and rule without echoing the rejected value. Zero delayMs and zero retry.delay are accepted; large finite values are accepted. Fix invalid values before registering jobs.

An exponential backoff that overflows to Infinity with an enormous retry limit now waits forever instead of retrying after about 1 ms.

Disconnect, remove and pause cancel pending retry backoff without another attempt. A one-shot that already succeeded or exhausted its retries is removed even while paused and is not re-run. A one-shot paused mid-retry stays registered; resume starts it from attempt 1 using the original delay. Recurring jobs resume at the next cron/interval boundary. Remaining backoff is discarded. A cancelled fire is not reported as success: diagnostics settle it with false, counting it in failed because there is no separate cancellation counter. An in-flight handler may still reach a final success or failure after cancellation. If a job is resumed while that handler is still running, it is armed only once the handler settles, so the two never overlap. Disconnect releases every one-shot :once slot this service holds, each release bounded by acquireTimeoutMs; resume() rejects on a disconnected service, and after reconnecting a one-shot re-claims its slot, so a replica that claimed it meanwhile runs the job instead. A pause, remove or disconnect during that re-claim wins over the resume, and a second resume() waits for the first.

Options

Declare jobs where the plugin is composed, instead of resolving the scheduler after start():

import { createApplication } from '@setu-ts/kernel';
import { RuntimePlugin } from '@setu-ts/runtime';
import { SchedulerPlugin } from '@setu-ts/scheduler-plugin';
import type { IIngressBehavior } from '@setu-ts/common';

/** Runs ahead of every job handler; `next()` continues the chain. */
const auditEveryRun: IIngressBehavior = {
  handle: (ctx, next) => {
    console.log(`${ctx.kind} ${ctx.name} attempt ${ctx.attempt ?? 1}`);
    return next();
  },
};

const app = createApplication({
  plugins: [
    RuntimePlugin(),
    SchedulerPlugin({
      behaviors: [auditEveryRun],
      jobs: [
        {
          trigger: 'every',
          name: 'send-email',
          intervalMs: 60_000,
          data: { to: 'ada@example.com' },
          handler: (job) => {
            console.log('sending to', job.data);
          },
        },
      ],
    }),
  ],
});

// Declared jobs register during startup, so nothing is scheduled until now.
await app.start({ port: 3000 });

Distributed locking

Without distributedLock, a process-local MemoryLock is used — fine for a single instance, but every replica will run every job. Set { enabled: true, storage: 'redis' } to use RedisLock (over ioredis, lazily imported or injected) so only one replica executes each firing.

Every lock acquire is bounded, whichever lock is in use. An acquire still unsettled after acquireTimeoutMs is a skipped fire: it is logged, counted as lockFailed in the execution observations, and the job is re-armed for its next slot, so a paused or unreachable lock backend skips fires instead of stopping the schedule. Keep the bound below the job’s interval. An acquire that returns a token after the bound has its lock released, so the abandoned token does not block the next fire until its TTL expires.

Connection errors of the ioredis client the lock builds go to the application logger — the first error of an outage at warn, identical repeats at debug, the recovery at info — instead of ioredis printing every reconnect failure to console.error. An injected client gets no listener: it belongs to the caller.

node-redis. To use node-redis instead of ioredis, pass distributedLock: { enabled: true, storage: 'redis', client: fromNodeRedis(createClient({ url })) } (fromNodeRedis from @setu-ts/common, createClient from redis, installed in your application). See Using node-redis.

Behaviours run inside that lock. A replica that does not acquire it runs neither the behaviour chain nor the job handler; a behaviour throw follows the job’s existing retry policy. With no behaviours, the handler receives its original job directly and no chain is allocated.

Health

The scheduler indicator reports up while the scheduler is connected and down after disconnect. Its data includes connected and lock: { reachable }, where reachable is true, false, or 'unknown'. A shared lock outage stops job fires while request serving can continue, so reachability does not change lifecycle status or remove the replica from readiness. Alert on data.lock.reachable to detect lock outages.

IDistributedLock.isHealthy?(): Promise<boolean | undefined> is optional: an absent probe, an undefined answer, or a probe that does not settle within 2 s reports 'unknown'. The indicator caches each answer for 5 s, uses runtime clocks and timers, and probes only while connected. MemoryLock always answers true. RedisLock answers false before connect and after disconnect; while connected it calls the optional IRedisLockClient.ping?(): Promise<unknown>. A resolved PING means true, a rejected PING means false, and a client without PING means 'unknown'. A built client’s PING also uses its configured commandTimeout.

Execution observations

SchedulerPlugin({ diagnostics: { enabled: true, alias, jobs } }) registers one ISchedulerDiagnosticsSource under CAPABILITIES.SCHEDULER_DIAGNOSTICS (as a multi provider, one per instance) that the local diagnostics connector reads to serve GET /v1/scheduler. Only exact job names listed in jobs are observed, each under its approved alias; counters distinguish contended fires (a lock held elsewhere), lock-failed fires, dispatched fires and handler attempts (including retries; a behaviors entry that declines a dispatch invokes no handler, so it records no attempt), plus the wall-clock lateness each fire started late. A skipped local fire is never reported as a globally missed execution. Job names, cron expressions, payloads, job ids, lock keys and thrown errors are never captured; without the option nothing is observed at all.

Exports

Generated from the package barrel by pnpm run docs:exports; pnpm run check:docs fails when it drifts.

Full API

Every export and option is documented in PUBLIC_API.md.

API reference

Scheduler (@setu-ts/scheduler-plugin)

Provides cron jobs, fixed-interval recurring jobs, and one-shot delayed jobs, with retry and distributed locking.

Registers IScheduler under CAPABILITIES.SCHEDULER ('scheduler').

Execution is in-process and time-driven — jobs are not durably persisted, so a restart drops the schedule until the registering plugin re-creates it. For durable background work, use Queue instead.

Exports

• SchedulerPlugin — Plugin factory for registering the scheduler service; throws SchedulerUnavailableError at register() on Cloudflare Workers, where its timers cannot fire — use WorkersCron with [triggers] crons there instead
• SchedulerUnavailableError — Thrown by register() when the runtime platform cannot run the scheduler (Cloudflare Workers); catch it by identity to branch on the refusal
• SchedulerPluginOptions — Plugin configuration options (timezone?, distributedLock?)
• DistributedLockOptions — Lock configuration (enabled?, storage?, url?, client?, lock?, ttlMs?, acquireTimeoutMs?, commandTimeoutMs?). Since M101a every lock acquire is bounded by acquireTimeoutMs (default 5000, 0 waits indefinitely), whichever lock is in use: an acquire still unsettled at the bound is a skipped fire — logged, counted lockFailed in the execution observations, and re-armed for the next slot — and a token it returns later is released. commandTimeoutMs is the ioredis commandTimeout on the client RedisLock builds (default acquireTimeoutMs, or 15000 when that is 0; never applied to an injected client or lock); a timed-out SET may still apply on the server, so RedisLock releases that exact token before rethrowing. Either value outside 0–2147483647, or a commandTimeoutMs above a non-zero acquireTimeoutMs, throws RangeError when SchedulerPlugin(...) is called
• IDistributedLock — Lock seam (acquire/release) for a custom lock implementation; optional isHealthy?(): Promise<boolean | undefined> reports reachable, unreachable, or unknown
• IRedisLockClient — Structural ioredis shape accepted by distributedLock.client; optional ping?(): Promise<unknown> supports lock reachability probes; eval(...) returns Promise<unknown> (M117, widened from Promise<number | string | null> so a fromNodeRedis adapter satisfies it; the lock ignores the result)
• IScheduler — Scheduler service interface (re-exported from @setu-ts/common)
• ScheduledJob<T> — Job instance handed to the handler (re-exported)
• SchedulerJobHandler<T> — Handler callback type (re-exported)
• ScheduleOptions<T> — Options for cron()/every()/delay() (re-exported)
• RetryOptions — Retry configuration (re-exported)
• SchedulerBackoff — 'fixed' | 'exponential' (re-exported)

Registration

import { SchedulerPlugin } from '@setu-ts/scheduler-plugin';

// Process-local locking (default)
app.register(SchedulerPlugin());

// Distributed locking via Redis, for multi-instance deployments
app.register(
  SchedulerPlugin({
    timezone: 'UTC',
    distributedLock: {
      enabled: true,
      storage: 'redis',
      url: config.get('REDIS_URL'),
      ttlMs: 30000,
    },
  }),
);

timezone defaults to 'UTC', and 'UTC' is the only supported value — any other value throws from the factory. Cron expressions are always evaluated in UTC.

When distributedLock is absent or enabled: false, a process-local memory lock is used. With enabled: true and storage: 'redis', an ioredis client is lazily imported (ioredis) unless you inject one via client, or supply your own IDistributedLock via lock (which takes priority over storage). ttlMs defaults to 30000 and must exceed the job’s worst-case runtime — if the lock expires mid-run, another instance may start the same job.

A Redis lock client the plugin builds routes the 'error' events of the ioredis client it builds to the application logger through createConnectionErrorReporter (common): the first error of an outage at warn, identical repeats at debug, the recovery at info. Without a listener ioredis prints every reconnect failure to console.error, bypassing the logger and redaction. An injected client gets no listener — it belongs to the caller. A lock that cannot reach Redis still fails its acquire; the health indicator exposes that outage in data.lock.reachable.

Multi-instance deduplication works across replicas sharing one lock backend. cron and every fires each claim a never-released slot lock keyed on the fire’s intended time (scheduler:job:<name>:<slot>), so two replicas whose timers land anywhere in the same intended slot run the handler exactly once between them; ttlMs bounds how long a claimed slot is remembered, so it must exceed the maximum skew between replicas. delay jobs claim their slot at REGISTRATION, keyed on the job name plus a :once suffix (scheduler:job:<name>:once), because a delay’s intended fire time is now + delayMs and carries per-replica startup skew — the name is what identifies “the same one-shot job” across replicas. The suffix is load-bearing: the bare scheduler:job:<name> is the per-handler mutex, and a slot holding that key would make the mutex acquire at fire time always lose to the slot’s own claim. A delay’s slot is released when the job leaves the registry (on final settlement or remove), on disconnect, or by TTL expiry, so re-registering the same name after it fired gets a fresh slot and fires again.

A delay job is lost if the replica that claimed it leaves before firing. The claim is decided at registration, so a replica that already found the slot held never re-attempts: if the claiming replica crashes or shuts down gracefully between registering the delay and firing it, the original due fire is lost on every replica and nothing reports it. A graceful shutdown releases the slot, so a replica that registers the same name afterwards can claim it and schedule a new fire. The exposure window is the delay itself, so it matters for a long delayMs on a replica set that may lose a pod in that window. cron and every are unaffected: they claim at fire time, so the next fire re-contends. A separate per-handler mutex preserves overlap protection — a second fire of a job whose previous fire is still running is skipped locally. every jobs arm on an absolute epoch grid ((floor(now / interval) + 1) * interval), so replicas registered at different instants agree on slot keys; the first fire may come sooner than one full interval after registration.

Declarative jobs and behaviours

The chain runs inside the distributed lock: a replica that does not acquire the lock runs neither a behaviour nor the handler. With no behaviours configured, the handler receives its job directly and no chain is allocated.

Scheduling Jobs

import type { IScheduler } from '@setu-ts/common';

app.register({
  name: 'scheduled-jobs',
  version: '1.0.0',
  dependencies: ['scheduler'],
  async register(ctx) {
    const scheduler = ctx.services.get<IScheduler>('scheduler');

    // Cron expression (5-field, UTC)
    await scheduler.cron('cleanup-temp-files', '0 2 * * *', async () => {
      await cleanupTempFiles();
    });

    // Every 5 minutes
    await scheduler.every('health-check', 300000, async () => {
      await runHealthCheck();
    });

    // One-time delayed job — auto-removed once it fires
    await scheduler.delay(
      'send-followup',
      86400000,
      async (job) => {
        await sendFollowupEmail(job.data.userId);
      },
      { data: { userId: '123' } },
    );

    // With retry
    await scheduler.cron(
      'sync-external-api',
      '*/30 * * * *',
      async () => {
        await syncFromExternalApi();
      },
      {
        retry: {
          limit: 3,
          delay: 10000,
          backoff: 'exponential',
        },
      },
    );
  },
});

Every scheduling call is async and throws if name is already scheduled. Job names are unique per scheduler instance.

Cron expressions use the standard 5 fields (minute hour day-of-month month day-of-week) and support *, lists (1,15), ranges (1-5), and steps (*/30, 1-59/2). An invalid field or expression throws.

Retry Behavior

Without retry, a handler that throws runs once and the failure is logged. With retry, the job is re-attempted up to limit total attempts (limit: 3 means at most 3 runs, not 3 retries after the first). delay is the base delay in milliseconds before the first retry; backoff: 'fixed' reuses it unchanged, while backoff: 'exponential' doubles it per attempt. job.attempts is 1-based and increments on each attempt. A job that exhausts its retries is logged and does not crash the scheduler; recurring jobs still fire on their next scheduled tick.

Managing Jobs

const scheduler = ctx.services.get<IScheduler>('scheduler');

// Pause (idempotent — pausing a paused job is a no-op)
await scheduler.pause('cleanup-temp-files');

// Resume (idempotent — resuming a running job is a no-op)
await scheduler.resume('cleanup-temp-files');

// Remove
await scheduler.remove('cleanup-temp-files');

// Next run time, as epoch milliseconds
const nextRun = await scheduler.getNextRun('cleanup-temp-files');

All four throw if no job with that name exists — including after a delay job has fired and auto-removed itself. getNextRun additionally throws if the job is currently paused.

resume re-arms from the current time rather than resuming the original countdown: cron jobs compute the next fire from now, delay jobs re-arm the full original delayMs from now, and every jobs resume on the next epoch grid boundary of their interval ((floor(now / intervalMs) + 1) * intervalMs) — not a full interval after now. This is breaking versus 0.1.0-alpha.8, whose contract stated the interval “restarts from now”: the fire may now come sooner than one full interval after resume (never later). Grid alignment at resume is deliberate — it is the same alignment that makes replicas agree on fire times and slot keys, and a resume that restarted the phase would let one replica’s resumed job drift out of slot agreement with the others.

Timing and registration refusals (M113m)

delay, every, cron and retry backoff wait in chunks of at most 2147483647 ms and check the wall clock on every wake. A job never runs before its due time, including after an early wake or a clock step backwards. There is no upper limit on a delay or interval: long delays and monthly cron waits are honoured instead of being clamped by Node, Deno or Bun to about 1 ms.

The async registration call rejects with a RangeError for these values:

The message names the job, option and rule without echoing the rejected value. Zero delayMs and zero retry.delay are accepted; large finite values are accepted. These refusals are breaking: fix the value before registering the job.

Disconnect, remove and pause cancel pending retry backoff without another attempt. A one-shot that already succeeded or exhausted its retries is removed even while paused and is not re-run. A one-shot paused mid-retry stays registered; resume starts it from attempt 1 using the original delay. Recurring jobs resume at the next cron/interval boundary. Remaining backoff is discarded. A cancelled fire is not reported as success: diagnostics settle it with false, counting it in failed because there is no separate cancellation counter. An in-flight handler may still reach a final success or failure after cancellation. If a job is resumed while that handler is still running, it is armed only once the handler settles, so the two never overlap. Disconnect releases every one-shot :once slot this service holds, each release bounded by acquireTimeoutMs; resume() rejects on a disconnected service, and after reconnecting a one-shot re-claims its slot, so a replica that claimed it meanwhile runs the job instead. A pause, remove or disconnect during that re-claim wins over the resume, and a second resume() waits for the first.

IScheduler Interface

interface IScheduler {
  cron<T>(
    name: string,
    expression: string,
    handler: SchedulerJobHandler<T>,
    options?: ScheduleOptions<T>,
  ): Promise<void>;
  every<T>(
    name: string,
    intervalMs: number,
    handler: SchedulerJobHandler<T>,
    options?: ScheduleOptions<T>,
  ): Promise<void>;
  delay<T>(
    name: string,
    delayMs: number,
    handler: SchedulerJobHandler<T>,
    options?: ScheduleOptions<T>,
  ): Promise<void>;
  pause(name: string): Promise<void>;
  resume(name: string): Promise<void>;
  remove(name: string): Promise<void>;
  getNextRun(name: string): Promise<number>;
}

interface ScheduledJob<T = unknown> {
  readonly id: string;
  readonly name: string;
  readonly data: T;
  readonly attempts: number;
}

Health status

The plugin registers a 'scheduler' health indicator with data: { connected, lock: { reachable: true | false | 'unknown' } }. Status stays lifecycle: connected → up, disconnected → down. Lock reachability is diagnostic because a shared lock outage stops scheduled fires while request serving can continue; alert on data.lock.reachable.

The indicator probes only while connected, caches answers for 5 s and bounds each probe at 2 s using runtime clocks and timers. A missing IDistributedLock.isHealthy, an undefined answer, or a timed-out/rejecting custom probe reports 'unknown'. Disconnected checks report down with connected: false and lock.reachable: 'unknown' without probing, even if an answer was cached.

MemoryLock.isHealthy() answers true. RedisLock.isHealthy() answers false before connect and after disconnect, undefined for a client without optional ping, and otherwise true when PING resolves or false when it rejects. A built client’s PING is bounded by its configured commandTimeout; injected clients keep their own configuration.

An onClose hook clears all armed timers and disconnects the Redis lock on shutdown.