titan-scheduler

Official@omnitron-dev/titan-scheduler

Maintained by the Omnitron team. Independent npm package.

Cron, interval, and timeout job scheduling with auto-discovery of decorated methods, pluggable persistence (memory / Redis / database), optional distributed coordination, per-job retry/backoff, concurrency limits, listener pattern, and lifecycle-aware graceful shutdown.

pnpm add @omnitron-dev/titan-scheduler

When you need it

• Periodic background work. Cron-style cleanup, hourly reports, nightly aggregations.
• Interval polling. Poll an external system every N seconds.
• Delayed actions. Run something exactly once, N ms after registration.
• Distributed jobs. Multi-pod fleet where exactly-one execution is required (pair with titan-lock or built-in distributed coordination).

Quickstart

import { SchedulerModule } from '@omnitron-dev/titan-scheduler';

@Module({
  imports: [
    SchedulerModule.forRoot({
      enabled:         true,
      timezone:        'UTC',
      persistence:     { enabled: true },           // in-memory default
      metrics:         { enabled: true },
      maxConcurrent:   10,
      shutdownTimeout: 30_000,
    }),
  ],
  providers: [TasksService],
})
class AppModule {}

Decorate methods

import {
  Schedulable,
  Cron, CronExpression,
  Interval,
  Timeout,
} from '@omnitron-dev/titan-scheduler';

@Injectable()
@Schedulable()
class TasksService {
  @Cron(CronExpression.EVERY_HOUR, { timezone: 'UTC' })
  async hourlySync() { /* … */ }

  @Interval(30_000, { name: 'queue-poll' })
  async pollQueue() { /* … */ }

  @Timeout(5_000)
  async runOnceAfterBoot() { /* … */ }
}

The discovery service finds decorated methods at onInit and registers them with the scheduler.

Async configuration

SchedulerModule.forRootAsync({
  imports: [ConfigModule],
  useFactory: (config: ConfigService) => ({
    enabled:       config.get('scheduler.enabled'),
    persistence:   { enabled: true, provider: new RedisPersistenceProvider(redis) },
    maxConcurrent: config.get('scheduler.concurrency'),
  }),
  inject: [ConfigService],
})

For distributed (multi-node) scheduling, also supply a lockProvider — see Distributed coordination below.

ISchedulerModuleOptions

Option	Type	Default
enabled	boolean	true
timezone	string	—
persistence	{ enabled, provider?, options? }	—
metrics	{ enabled, interval?, includeDetails? }	—
distributed	{ enabled, lockProvider?: 'redis' | 'database', lockTTL?, nodeId? }	—
retry	IRetryOptions — global retry policy	—
maxConcurrent	number	—
queueSize	number	—
queueTimeout	number (ms)	—
debug	boolean	—
shutdownTimeout	number (ms) — grace window	30_000
healthCheck	{ enabled, path? }	—
listeners	IJobListener[] — global listeners	—
lockProvider	ISchedulerLockProvider — required when distributed.enabled (SC-1)	—
executor / persistenceProvider / metricsProvider	InjectionToken — override the built-in providers	—

Decorators

@Cron(expression, options?)

@Cron('0 3 * * *', { timezone: 'UTC', immediate: false })
async nightly() { /* runs at 03:00 UTC daily */ }

@Cron(CronExpression.EVERY_5_MINUTES)
async refresh() { /* … */ }

Options:

Field	Type
timezone?	string
utcOffset?	number
startTime?	Date | string
endTime?	Date | string
immediate?	boolean — run once at start
(and IBaseJobOptions — see below)

@Interval(ms, options?)

@Interval(60_000, { immediate: true })
async heartbeat() { /* runs immediately then every 60s */ }

@Timeout(ms, options?)

@Timeout(10_000)
async runOnceAfterBoot() { /* fires 10s after registration */ }

@Schedulable()

Class-level marker; optional but recommended for clarity. The discovery service finds @Cron/@Interval/@Timeout methods on any @Injectable() class.

IBaseJobOptions — shared across all job types

Field	Type
name?	string — explicit job name
priority?	JobPriority — CRITICAL | HIGH | NORMAL | LOW | IDLE
retry?	{ maxAttempts?, delay?, backoff?, maxDelay?, retryIf? }
onError?	(error: Error) => void | Promise<void>
onSuccess?	(result: any) => void | Promise<void>
disabled?	boolean — don't auto-start
timeout?	number (ms) — max execution time
persist?	boolean — save to persistence provider
metadata?	Record<string, any>
preventOverlap?	boolean — reject concurrent executions of the same job

A job that fails to register stops the boot

disabled: true is the only way a job is silently absent, and it is silent because you asked for it.

Every other registration failure now throws at startup rather than dropping the job. The realistic one is a name collision: two definitions claiming the same name make registerJob throw a conflict — Job with this name already exists: <name> — and the loser used to be discarded with nothing logged — the application reported a successful start and the first evidence was a nightly task that had not run. Give jobs distinct names, or leave name off and let it be derived.

This matches what the module already does elsewhere with configuration that would otherwise be silently wrong: onStart refuses distributed mode without a lock provider rather than let every instance fire every job.

CronExpression enum

import { CronExpression } from '@omnitron-dev/titan-scheduler';

CronExpression.EVERY_SECOND          // '* * * * * *'
CronExpression.EVERY_5_SECONDS       // '*/5 * * * * *'
CronExpression.EVERY_30_SECONDS      // '*/30 * * * * *'
CronExpression.EVERY_MINUTE          // '*/1 * * * *'
CronExpression.EVERY_5_MINUTES       // '*/5 * * * *'
CronExpression.EVERY_HOUR            // '0 * * * *'
CronExpression.EVERY_DAY_AT_1AM      // '0 1 * * *'
// …through EVERY_YEAR = '0 0 1 1 *'

SchedulerService

Most jobs are registered declaratively with the decorators above. To register a job dynamically at runtime, inject the service and use addCronJob / addInterval / addTimeout — each takes a name, the pattern, a handler, and optional job options:

import { SchedulerService, SCHEDULER_SERVICE_TOKEN } from '@omnitron-dev/titan-scheduler';

@Service({ name: 'jobs' })
class JobsService {
  constructor(@Inject(SCHEDULER_SERVICE_TOKEN) private readonly scheduler: SchedulerService) {}

  @Public()
  async runAfter(ms: number, payload: unknown) {
    this.scheduler.addTimeout(
      `once:${crypto.randomUUID()}`,
      ms,
      async () => { /* … fires once after `ms` … */ },
      { metadata: { payload } },
    );
  }
}

Method	Purpose
addCronJob(name, expression, handler, options?)	Register + schedule a cron job dynamically
addInterval(name, ms, handler, options?)	Register + schedule an interval job
addTimeout(name, ms, handler, options?)	Register + schedule a one-shot timeout job
triggerJob(name)	Run a job immediately (bypasses distributed lock)
stopJob(name) / startJob(name)	Stop (pause) / re-schedule a job by name
deleteJob(name)	Stop + remove a job (and its persisted record)
getJob(name)	Look up a single job by name
getAllJobs()	All registered jobs
findJobs(filter)	Filter jobs (IJobFilterOptions)
getMetrics()	Scheduler metrics (ISchedulerMetrics | null)
isRunning()	Whether the scheduler has started

Jobs are keyed by name, not an opaque id. Lifecycle (onInit / onStart / onStop) is driven by the framework — you rarely call these methods yourself for decorator-defined jobs.

Persistence providers

Provider class	Persistence
InMemoryPersistenceProvider	Default — keeps last 100 executions per job in memory
RedisPersistenceProvider	Redis-backed job + execution storage; survives restart
DatabasePersistenceProvider	SQL-backed; durable, transactional history

Implement IPersistenceProvider:

interface IPersistenceProvider {
  saveJob(job: IScheduledJob): Promise<void>;
  loadJob(id: string): Promise<IScheduledJob | null>;
  loadAllJobs(): Promise<IScheduledJob[]>;
  deleteJob(id: string): Promise<void>;
  saveExecutionResult(result: IJobExecutionResult): Promise<void>;
  loadExecutionHistory(jobId: string, limit?: number): Promise<IJobExecutionResult[]>;
  clear(): Promise<void>;
}

Distributed coordination

For multi-pod deployments, two paths:

A) Module-level distributed config (SC-1)

When distributed.enabled is true, before each scheduled fire the scheduler acquires a per-fire-window lock so a given fire runs on exactly one node; the other nodes skip it. This requires a lock provider — an object implementing ISchedulerLockProvider (acquireLock(key, ttlMs) / releaseLock(key, lockId)). Pass it as the lockProvider module option (or register it at SCHEDULER_LOCK_TOKEN). titan-lock's DistributedLockService is structurally compatible and can be used directly.

Fail-fast. If distributed.enabled is true but no lock provider is wired, the scheduler throws on start rather than silently running every job on every node.

import { LOCK_SERVICE_TOKEN } from '@omnitron-dev/titan-lock';

SchedulerModule.forRootAsync({
  imports: [TitanLockModule.forRoot({ /* … */ })],
  useFactory: (locks) => ({
    distributed: {
      enabled: true,
      lockTTL: 60_000,
      nodeId:  process.env.HOSTNAME,
    },
    lockProvider: locks,    // ISchedulerLockProvider (DistributedLockService is compatible)
  }),
  inject: [LOCK_SERVICE_TOKEN],
})

The distributed.lockProvider?: 'redis' | 'database' field is only an advisory label describing the kind of backing store; the actual coordination is done by the lockProvider instance you supply. The scheduler holds each per-fire lock for its full lockTTL and does not release it early (so a clock-skewed-late node can't re-run the same fire).

B) Per-job @WithDistributedLock

import { WithDistributedLock } from '@omnitron-dev/titan-lock';

@Cron(CronExpression.EVERY_HOUR)
@WithDistributedLock('jobs:hourly-report', 10 * 60_000)
async hourlyReport() { /* exactly one pod runs this per hour */ }

Per-job locking gives you finer control (only the decorated jobs are coordinated, and it releases the lock as soon as the method returns); module-level coordination (A) covers every job at once and holds each fire's lock for the full lockTTL.

Listeners

const auditListener: IJobListener = {
  onJobStart:    async (job, ctx) => { /* … */ },
  onJobComplete: async (job, result) => { /* … */ },
  onJobError:    async (job, error, ctx) => { /* … */ },
  onJobRetry:    async (job, attempt, error) => { /* … */ },
  onJobCancelled: async (job, reason) => { /* … */ },
};

SchedulerModule.forRoot({ listeners: [auditListener] });

Use for unified observability (metrics, alerting, audit trails) without coupling each job's body to telemetry concerns.

Status / priority enums

// JobStatus is a const-object string union:
//   'pending' | 'running' | 'completed' | 'failed' | 'cancelled' | 'paused' | 'retrying'
const JobStatus = {
  PENDING: 'pending', RUNNING: 'running', COMPLETED: 'completed',
  FAILED: 'failed', CANCELLED: 'cancelled', PAUSED: 'paused', RETRYING: 'retrying',
} as const;

enum JobPriority {
  CRITICAL = 0,
  HIGH     = 1,
  NORMAL   = 2,
  LOW      = 3,
  IDLE     = 4,
}

Lifecycle

SchedulerService implements:

• async onInit() — load persisted jobs, run discovery on decorated methods, auto-start if enabled: true.
• async onStart() — activate all registered jobs, emit SCHEDULER_STARTED.
• async onStop() — stop intervals / cron, wait up to shutdownTimeout for running jobs to finish, persist state.

Tokens

Token
SCHEDULER_SERVICE_TOKEN
SCHEDULER_CONFIG_TOKEN
SCHEDULER_REGISTRY_TOKEN
SCHEDULER_EXECUTOR_TOKEN
SCHEDULER_PERSISTENCE_TOKEN
SCHEDULER_METRICS_TOKEN
SCHEDULER_DISCOVERY_TOKEN
SCHEDULER_LISTENERS_TOKEN
SCHEDULER_LOCK_TOKEN

Anti-patterns

• Long-running jobs with short intervals. @Interval(1_000) with a 5s body queues up. Set preventOverlap: true or shorten the body.
• Cron-style polling for low-latency triggers. Cron has ≥1s resolution. For sub-second reactions, use titan-events or a message queue.
• Forgetting persistence in production. Without persistence, jobs disappear on restart. Use RedisPersistenceProvider or DatabasePersistenceProvider.
• Distributed coord without lock TTL tuning. Lock TTL must exceed the job's worst-case duration, or another pod picks it up mid-execution.

See also

• titan-lock — pair with @WithDistributedLock per job
• titan-metrics — picks up scheduler metrics if both modules loaded
• titan-health — SchedulerHealthIndicator exposed