Scheduler
Decide when a daily task is due, claim its occurrence and hand expensive work to Queue. Inspect what happened without hiding schedule definitions in a database.
On this page
Source-backed MarkdownSet up schedules and the queue
Scheduler needs ScheduledOccurrence in the application database. The database queue also needs QueuedJob and FailedJob. The lab installs these in a disposable database; application deployments must use migrations.
Create one App and register schedules once at process boot. The same bootstrap must register queue job classes in every worker. Resolving app().scheduler also attaches the recorder that updates occurrence history from queue lifecycle events.
Copy the daily-summary example
The example writes a small summary file through a queued job. It uses a fixed test minute and temporary storage, so it does not wait until tomorrow or dispatch real application work.
mkdir -p examples
cp -R node_modules/@db3.ai/app/src/scheduler/examples/. examples/Run it
Expect one dispatched occurrence, one skipped duplicate, successful queue/occurrence states, and no due task in the following minute. The file contains Daily summary ready.
npx tsx examples/runDailySummary.tsPut the work in a job
The job below has a stable durable name and an empty JSON payload. It replaces a summary file rather than appending duplicate output on retry. Real jobs still need an application-specific idempotency policy, particularly when sending messages or charging credits.
import { QueueableJob } from '@db3.ai/app/queue';
import { app } from '@db3.ai/app/server';
/** A small application job showing scheduled work performed by the queue. */
export class WriteDailySummaryJob extends QueueableJob {
static readonly jobName = 'write-daily-summary';
/** Creates a zero-argument job with a serializable empty payload. */
constructor() {
super({});
}
/** Writes a replaceable summary; repeated attempts do not append duplicates. */
async handle(): Promise<void> {
await app().storage.write('reports/latest.txt', 'Daily summary ready');
}
}
Register a daily schedule
A zero-argument QueueableJob class supplies its durable name automatically. For a job with constructor data, pass a fresh-job factory and give the schedule an explicit stable name(). Inline call() definitions also need a name.
Use daily() for midnight or dailyAt("HH:mm"). Names must be unique. Times use UTC unless you set an IANA timezone(). The current API does not support weekly/monthly schedules, arbitrary cron expressions or every-minute frequencies.
import { app } from '@db3.ai/app/server';
import { WriteDailySummaryJob } from './WriteDailySummaryJob';
/**
* Registers one daily report in each scheduler and queue-worker process.
*
* Call once at boot. Resolving scheduler also attaches the queue lifecycle
* recorder, so workers update occurrence history after processing the job.
*/
export function registerDailySummary(): void {
app().queue.registerJob(WriteDailySummaryJob);
app().scheduler.job(WriteDailySummaryJob).dailyAt('09:00').timezone('UTC');
}
Evaluate, then process
runDue() evaluates the current minute. The lab passes a fixed Date explicitly, evaluates it twice, then calls workNextJob() to perform the queued work. Dispatch success is not job completion.
The unique occurrence claim is per schedule name and UTC minute. That prevents duplicate dispatch for the same claimed minute; it is not an exactly-once guarantee for external side effects.
import { mkdtemp, rm } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { pathToFileURL } from 'node:url';
import { createGeneratedTestDatabase } from '@db3.ai/app/db/test/db';
import { FailedJob, QueuedJob } from '@db3.ai/app/queue';
import { ScheduledOccurrence } from '@db3.ai/app/scheduler';
import { App } from '@db3.ai/app/server';
import { registerDailySummary } from './registerDailySummary';
/**
* Evaluates a fixed minute, processes its SQL-backed job and checks deduplication.
*
* This isolated lab owns its schema and files. It does not run historical
* application schedules or wait for wall-clock time to pass.
*
* @returns Stable scheduling, processing and history observations.
*/
export async function runDailySummary() {
const root = await mkdtemp(join(tmpdir(), 'db3-scheduler-guide-'));
try {
const database = await createGeneratedTestDatabase('scheduler_guide');
const application = new App({ db: database.db, storage: { disks: { local: { driver: 'local', root } } } });
try {
await application.db.install(QueuedJob, FailedJob, ScheduledOccurrence);
registerDailySummary();
const minute = new Date('2026-01-01T09:00:00Z');
const first = await application.scheduler.runDue(minute);
const repeated = await application.scheduler.runDue(minute);
const processed = await application.queue.workNextJob();
if (!processed) throw new Error('The scheduled job was not available to the worker.');
const occurrence = await ScheduledOccurrence.where('name', 'write-daily-summary').firstOrFail();
const text = await application.storage.readToString('reports/latest.txt');
const notDue = await application.scheduler.runDue(new Date('2026-01-01T09:01:00Z'));
return { dispatched: first.dispatched, duplicateSkipped: repeated.skipped, processed: processed.status, occurrenceStatus: occurrence.status, text, nextMinuteDue: notDue.due };
} finally {
try { await application.close(); } finally { await database.destroy(); }
}
} finally {
await rm(root, { recursive: true, force: true });
}
}
if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) {
console.log(JSON.stringify(await runDailySummary(), null, 2));
}
Understand missed minutes and time zones
There is no automatic catch-up after downtime. A task that was due while the scheduler was stopped is not replayed later by the normal worker.
Local wall-clock schedules follow their selected time zone. A skipped daylight-saving time will not occur; a repeated local time can correspond to two UTC minutes. Choose UTC or add an application business-date policy if the task must run once per business day.
Run a dedicated scheduler process
Use SchedulerWorker for a long-running process, or call runDue() from a supervised once-per-minute runner. Keep slow work in Queue; call() runs directly in the scheduler and is appropriate only for short operations.
The framework exports runSchedulerConsole() for an application-owned CLI entry point. It accepts an App resolver and bootstrap hook. After wiring that entry point, it supports scheduler:work, scheduler:run, scheduler:list and scheduler:history --limit=25 --status=failed.
Those are command names for your adapter, not globally installed executables. The runnable lab does not supply a production supervisor or CLI bootstrap. On shutdown, stop the SchedulerWorker before closing the App. Queue workers must use the same code/configuration and initialize the scheduler recorder.
Inspect history and failures
ScheduledOccurrence records the name, due minute, queue IDs, attempts, status, timestamps and errors. Queued work can move through queued, running, retrying/deferred, succeeded or failed states.
Use queue.retryFailed() for a terminal queued failure. It creates a replacement with a new identity and a retryOf link. The original failed queue record and ScheduledOccurrence remain failed for audit, even when that replacement succeeds. Observe the replacement’s queue lifecycle/result and application output; Queue does not retain a general success-history table. Persist recovery on an application-owned report when needed. A claim and queue dispatch are separate transitions: a crash between them can leave a claimed occurrence without a queued job. Monitor and reconcile those cases deliberately; there is no automatic historical catch-up or exactly-once delivery promise.
Repair work without rewriting history
Run npx tsx examples/runScheduledReplay.ts after copying the Scheduler examples. It deliberately fails three immediate attempts because a local source is missing, supplies the source, then replays once. No historical application schedule is touched.
Expect originalStatus: "failed", replacementStatus: "succeeded", linked: true, newIdentity: true, one retained failure, zero active jobs and the recovered summary. Re-evaluating the original minute still skips its existing claim. Immediate backoff is for this deterministic lab, not production polling.
import { mkdtemp, rm } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { pathToFileURL } from 'node:url';
import { createGeneratedTestDatabase } from '@db3.ai/app/db/test/db';
import { FailedJob, QueuedJob } from '@db3.ai/app/queue';
import { ScheduledOccurrence } from '@db3.ai/app/scheduler';
import { App } from '@db3.ai/app/server';
import { RecoverableSummaryJob } from './RecoverableSummaryJob';
/**
* Recovers a terminal scheduled failure without rewriting its original audit row.
*
* The replacement job has a new identity and links to the retained failure.
* The lab uses immediate retries only to avoid waiting; production workers
* should use deliberate backoff. All SQL and files belong to this invocation.
*
* @returns Original audit state, replacement outcome, linkage and cleaned queue.
*/
export async function runScheduledReplay() {
const root = await mkdtemp(join(tmpdir(), 'db3-scheduled-replay-'));
try {
const database = await createGeneratedTestDatabase('scheduled_replay');
const application = new App({ db: database.db, queue: { retryDelaySeconds: 0, queueMonitor: false }, storage: { disks: { local: { driver: 'local', root } } } });
try {
await application.db.install(QueuedJob, FailedJob, ScheduledOccurrence);
application.queue.registerJob(RecoverableSummaryJob);
application.scheduler.job(RecoverableSummaryJob).dailyAt('09:00').timezone('UTC');
const minute = new Date('2026-01-01T09:00:00Z');
await application.scheduler.runDue(minute);
// Default maxTries is three. Bound the drain so a regression cannot hang.
for (let attempt = 0; attempt < 3; attempt++) await application.queue.workNextJob();
const original = await ScheduledOccurrence.where('name', RecoverableSummaryJob.jobName).firstOrFail();
const [failure] = await application.queue.failedJobs();
if (!failure || original.status !== 'failed') throw new Error('Expected the missing source to reach terminal failure.');
await application.storage.write('summary-source.txt', 'Recovered');
const replacementId = await application.queue.retryFailed(failure.id);
const replacement = await QueuedJob.findOrFail(replacementId);
const linked = replacement.payload?.retryOf?.failedJobId === failure.id;
const result = await application.queue.workNextJob();
const retained = await ScheduledOccurrence.findOrFail(original.id!);
return {
originalStatus: retained.status, replacementStatus: result?.status, linked,
newIdentity: replacement.payload?.uuid !== failure.payload.uuid,
text: await application.storage.readToString('reports/recovered.txt'),
duplicateSkipped: (await application.scheduler.runDue(minute)).skipped,
retainedFailures: (await application.queue.failedJobs()).length,
activeJobs: await QueuedJob.query().count(),
};
} finally {
try { await application.close(); } finally { await database.destroy(); }
}
} finally {
await rm(root, { recursive: true, force: true });
}
}
if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) {
console.log(JSON.stringify(await runScheduledReplay(), null, 2));
}
Read the source before replacing output
The repairable job reads its local source before writing the report. A missing source therefore cannot overwrite the last output. Replay performs the same normal job contract rather than changing the scheduled audit row.
import { QueueableJob } from '@db3.ai/app/queue';
import { app } from '@db3.ai/app/server';
/** A scheduled report whose missing local input can be repaired before replay. */
export class RecoverableSummaryJob extends QueueableJob {
static readonly jobName = 'recoverable-summary';
/** Creates a durable empty payload; the app owns the source location. */
constructor() {
super({});
}
/** Reads the source before replacing output so failed reads cannot overwrite it. */
async handle(): Promise<void> {
const source = await app().storage.readToString('summary-source.txt');
await app().storage.write('reports/recovered.txt', `Summary: ${source.trim()}`);
}
}
Testing
Create a tests/scheduler directory and save the test below as runDailySummary.test.ts. It imports the example you copied into examples/. Keep the same folder layout so the relative import resolves.
Run from the application root with the development dependencies from Installation. These are consumer tests, not commands that assume a framework checkout. This database lab needs the same test-only SQL credentials when run through Vitest.
import { expect, it } from 'vitest';
import { runDailySummary } from '../../examples/runDailySummary';
import { runScheduledReplay } from '../../examples/runScheduledReplay';
it('runs the scheduler guide with real SQL queue claims and occurrence history', async () => {
expect(await runDailySummary()).toEqual({
dispatched: 1, duplicateSkipped: 1, processed: 'succeeded', occurrenceStatus: 'succeeded',
text: 'Daily summary ready', nextMinuteDue: 0,
});
});
it('recovers a scheduled job while retaining the original terminal occurrence and failure', async () => {
expect(await runScheduledReplay()).toEqual({
originalStatus: 'failed', replacementStatus: 'succeeded', linked: true, newIdentity: true,
text: 'Summary: Recovered', duplicateSkipped: 1, retainedFailures: 1, activeJobs: 0,
});
});
Run and extend the test
Use a fixed Date in tests instead of sleeping. Try a non-due minute, repeat an already-claimed minute, or make the job fail and inspect the occurrence. The copied test proves the simple successful database-queue path and duplicate claim rejection.
npx vitest run tests/scheduler/runDailySummary.test.ts
npx tsc --noEmit --target ES2022 --module ESNext --moduleResolution Bundler --types node --skipLibCheck examples/*.tsCoverage and reference
Tested here: registration, daily matching, duplicate claim rejection, SQL queue dispatch/execution and successful history recording. The second copied test repairs a failed scheduled job, checks its replay link/output and confirms the terminal occurrence stays failed. The service-owned Scheduler suite separately covers validation, workers, console behaviour and more lifecycle transitions.
Still to write: a complete production CLI/supervisor walkthrough, DST/business-date examples, failed-dispatch reconciliation and an application-owned recovery dashboard. The Source-backed Markdown version includes the service reference.
Evaluates a fixed minute, deduplicates a repeat, processes a real database-queued job and reads its successful occurrence.
The guide test passes against the real framework components.packages/app/src/scheduler/tests/examples/runDailySummary.test.tsThis test command requires the framework repository. Use the walkthrough commands in an installed application.
Environment: A dedicated MariaDB/MySQL test account and temporary local storage. No clock sleeps or remote providers.