--- id: "titan/modules/scheduler" title: "titan-scheduler" --- # titan-scheduler official · @omnitron-dev/titan-scheduler · stable 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. ```bash 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`](/docs/titan/modules/lock) or built-in distributed coordination). ## Quickstart ```typescript 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 ```typescript 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 ```typescript 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](#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 | — | During shutdown the scheduler stops issuing ticks and signals in-flight jobs. Queued jobs and pending retries settle without starting a new handler. Handlers that need to unwind receive the remainder of `shutdownTimeout`. Choose zero for immediate return. A handler that ignores cancellation may still be executing when that deadline expires. Both `maxConcurrent` and `queueSize` bound admission; a full queue rejects new work. ## Decorators ### `@Cron(expression, options?)` ```typescript @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?)` ```typescript @Interval(60_000, { immediate: true }) async heartbeat() { /* runs immediately then every 60s */ } ``` ### `@Timeout(ms, options?)` ```typescript @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` | | `onSuccess?` | `(result: any) => void \| Promise` | | `disabled?` | `boolean` — don't auto-start | | `timeout?` | `number` (ms) — max execution time | | `persist?` | `boolean` — save to persistence provider | | `metadata?` | `Record` | | `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: ` — 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 `name`s, 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 ```typescript 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: ```typescript 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`: ```typescript interface IPersistenceProvider { saveJob(job: IScheduledJob): Promise; loadJob(id: string): Promise; loadAllJobs(): Promise; deleteJob(id: string): Promise; saveExecutionResult(result: IJobExecutionResult): Promise; loadExecutionHistory(jobId: string, limit?: number): Promise; clear(): Promise; } ``` ## 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. ```typescript 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` ```typescript 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 ```typescript 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 ```typescript // 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. ```mermaid stateDiagram-v2 [*] --> PENDING: register / addCronJob() PENDING --> RUNNING: trigger fires RUNNING --> COMPLETED: success RUNNING --> FAILED: error & no retries left RUNNING --> RETRYING: error & retries remain RETRYING --> RUNNING: backoff elapsed RUNNING --> CANCELLED: cancel PENDING --> PAUSED: stopJob() PAUSED --> PENDING: startJob() COMPLETED --> PENDING: next cron tick FAILED --> PENDING: startJob() ``` ## 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`](/docs/titan/modules/lock) — pair with `@WithDistributedLock` per job - [`titan-metrics`](/docs/titan/modules/metrics) — picks up scheduler metrics if both modules loaded - [`titan-health`](/docs/titan/modules/health) — `SchedulerHealthIndicator` exposed