js/jobs
js/jobs.ts
fino:jobs — durable background jobs and cron schedules over sqlite.
Work is described by fino:task tasks and delivered by name: a job row
stores {task, input}, so anything pushed survives a restart and runs
wherever that task name is registered. Task instances are worker
definitions, registered separately from pushes — inline (process()) to
run on this realm's loop, or as an exclusive worker pool (workers())
whose entry module default-exports a Task.
Delivery is at-least-once: a crash after side effects or an expired
lease reruns the job, so handlers must be idempotent. fino:task/durable
tasks are the built-in idempotency tool — a retried durable job resumes
its workflow run from the last checkpoint instead of starting over, and
its ctx.sleep() / ctx.waitForSignal() parks are woken by this module's
scheduler.
Two homes, one behavior
Under fino run, the orchestrator hosts the jobs service and this module
becomes a thin client over the injected fino:jobs/control facade — the
scheduler and its single database connection outlive the app realm. In
any other realm (tests, embedded library use), Jobs.open() hosts the
service locally. The sqlite file must have exactly one service per
process, and one process per file.
Cron expressions evaluate in UTC (see internal:jobs/cron); @daily
means midnight UTC.
import { task } from 'fino:task';
import { Jobs } from 'fino:jobs';
const greet = task({
name: 'greet',
run: async (input: { name: string }) => `hello ${input.name}`,
});
await using jobs = await Jobs.open({ path: './.fino/jobs.db', tasks: [greet] });
const job = await jobs.push('greet', { name: 'Ada' });
const done = await jobs.wait(job.id);
console.log(done.result);Types
type JobStatus = 'pending' | 'claimed' | 'running' | 'waiting' | 'done' | 'error' | 'dead' | 'cancelled'
Job lifecycle states stored in JobRecord.status.
Terminal states are done, error, dead, and cancelled; wait()
resolves when a job reaches one of those states.
Interfaces
interface JobRetryPolicy {
Retry/backoff policy copied onto each job.
Attempts are delayed by baseMs * factor ** (attempt - 1), capped at
maxMs. When jitter is true, the computed delay is randomized between
zero and the capped delay to avoid retry bursts.
Every field is required here; JobsPushOptions.retry and
JobsScheduleOptions.retry accept a Partial<JobRetryPolicy> and merge it
over the service defaults.
import type { JobRetryPolicy } from 'fino:jobs';
const policy: JobRetryPolicy = {
maxAttempts: 5,
baseMs: 1000,
factor: 2,
maxMs: 60_000,
jitter: true,
};Properties
maxAttempts: number
Total attempts before the job is dead-lettered, including the first run.
baseMs: number
Delay of the first retry, and the base of the exponential backoff.
factor: number
Multiplier applied per attempt: attempt n waits baseMs * factor ** (n - 1).
maxMs: number
Ceiling on the computed delay so backoff does not grow without bound.
jitter: boolean
When true, randomize each delay in [0, computed] to spread out retry bursts.
interface JobRecord {
Persisted job row returned by push(), get(), list(), job(), and
wait().
Timestamps are epoch milliseconds. input, result, waitingOn, and
error are JSON-compatible values stored in the jobs database. Active jobs
may have claimedBy / claimedUntil set while a worker owns their lease;
terminal jobs set finishedAt.
import { Jobs } from 'fino:jobs';
const jobs = await Jobs.open({ path: './.fino/jobs.db' });
const record = await jobs.get('job-id');
if (record !== null && record.status === 'error') {
console.error(`${record.task} failed after ${record.attempts} attempts:`, record.error?.message);
}Properties
id: string
Unique job id assigned at push time.
queue: string
Queue the job belongs to; scopes dedupe and stats.
task: string
Registered task name that will run this job.
input: unknown
JSON-compatible input passed to the task handler.
status: JobStatus
Current lifecycle state.
priority: number
Sort priority among jobs due at the same time; higher runs first.
runAt: number
Earliest epoch-ms time the job may be claimed.
attempts: number
Number of attempts made so far.
maxAttempts: number
Maximum attempts before dead-lettering, copied from the retry policy.
backoff: JobRetryPolicy | null
Retry/backoff policy for this job, or null to use service defaults.
timeoutMs: number | null
Per-attempt timeout in milliseconds, or null for no timeout.
dedupeKey: string | null
Dedupe key that reserves the (queue, key) slot while active, or null.
claimedBy: string | null
Worker id holding the current lease, or null when unclaimed.
claimedUntil: number | null
Epoch-ms expiry of the current lease, or null when unclaimed.
workflowRunId: string | null
Durable workflow run id backing this job, or null for a plain job.
waitingOn: unknown
Signal or condition a parked durable job is waiting on, if any.
scheduleId: string | null
Id of the schedule that enqueued this job, or null if pushed directly.
result: unknown
Return value of a done job; otherwise null/undefined.
error: {
message: string;
stack?: string;
} | null
Failure details for an error or dead job, or null.
createdAt: number
Epoch-ms creation time.
updatedAt: number
Epoch-ms time of the last state change.
finishedAt: number | null
Epoch-ms time the job reached a terminal state, or null while active.
interface QueueStats {
Aggregate counts for one queue or all queues.
Returned by the live Jobs.stats() signal. oldestPendingAt is the oldest
due pending job timestamp, or null when there is no pending work.
import { Jobs } from 'fino:jobs';
const jobs = await Jobs.open({ path: './.fino/jobs.db' });
const stats = jobs.stats('emails');
stats.subscribe((s) => {
if (s.oldestPendingAt !== null && Date.now() - s.oldestPendingAt > 60_000) {
console.warn(`emails queue is backing up: ${s.pending} pending`);
}
});Properties
pending: number
Jobs that are due and awaiting a claim.
running: number
Jobs currently executing on a worker.
waiting: number
Durable jobs parked on a signal or sleep.
done: number
Jobs that completed successfully.
error: number
Jobs that failed their last attempt and may still retry.
dead: number
Jobs that exhausted retries or were dead-lettered.
cancelled: number
Jobs cancelled before completion.
oldestPendingAt: number | null
Epoch-ms timestamp of the oldest due pending job, or null when idle.
interface ScheduleRecord {
Persisted schedule row returned by schedule() and schedules().
spec is the normalized cron or interval expression. nextRunAt and
lastRunAt are epoch milliseconds; lastJobId links to the most recently
enqueued job when one exists.
import { Jobs } from 'fino:jobs';
const jobs = await Jobs.open({ path: './.fino/jobs.db' });
for (const s of await jobs.schedules()) {
console.log(`${s.id} (${s.spec}) next runs at ${new Date(s.nextRunAt).toISOString()}`);
}Properties
id: string
Schedule name, unique per service; passed to schedule()/unschedule().
task: string
Task name enqueued on each occurrence.
input: unknown
JSON-compatible input passed to every enqueued job.
queue: string
Queue that jobs from this schedule are pushed to.
spec: string
Normalized cron or interval expression driving the cadence.
enabled: boolean
Whether the schedule is currently active.
overlap: 'skip' | 'allow'
Overlap policy when a prior scheduled job is still active.
catchup: 'skip' | 'one'
Catch-up policy for occurrences missed during downtime.
retry: JobRetryPolicy | null
Retry policy copied onto each enqueued job, or null for defaults.
nextRunAt: number
Epoch-ms time of the next due occurrence.
lastRunAt: number | null
Epoch-ms time the schedule last fired, or null if it never has.
lastJobId: string | null
Id of the most recently enqueued job, or null if none exists.
createdAt: number
Epoch-ms creation time.
updatedAt: number
Epoch-ms time of the last modification.
interface JobsOptions {
Options for Jobs.open().
path is the only required field. Supplying tasks registers inline
workers in the same call, equivalent to a follow-up process(); the
remaining fields tune the locally-hosted service and are ignored when a
runtime orchestrator already hosts the jobs service.
import { task } from 'fino:task';
import { Jobs } from 'fino:jobs';
const resize = task({ name: 'resize', run: async () => null });
await using jobs = await Jobs.open({
path: './.fino/jobs.db',
tasks: [resize],
concurrency: 4,
leaseMs: 30_000,
});Properties
path: string
Path to the sqlite database file backing jobs, schedules, and durable runs.
tasks?: Task[]
Inline worker definitions: task trees this realm executes directly.
concurrency?: number
Concurrency for the inline processor (default 1).
leaseMs?: number
Worker lease duration; expired leases are requeued or dead-lettered.
pollIntervalMs?: number
Fallback poll interval for work created outside this process.
closeTimeout?: number
How long stop() waits for in-flight jobs before letting leases recover them.
interface JobsPushOptions {
Options for Jobs.push().
All fields are optional; an empty object enqueues an immediately-due job on
the default queue. Combine delay with key to schedule debounced work,
or priority with retry to control ordering and failure handling.
import { Jobs } from 'fino:jobs';
const jobs = await Jobs.open({ path: './.fino/jobs.db' });
await jobs.push('send-digest', { userId: 42 }, {
queue: 'emails',
delay: '1h',
key: 'digest:42',
priority: 10,
retry: { maxAttempts: 3 },
timeoutMs: 15_000,
});Properties
queue?: string
Queue name used for ordering, stats, and worker selection.
Defaults to 'default'. Jobs in different queues are independent for
dedupe and queue-level stats, but all queues share the same backing store.
delay?: number | string | Date
Delay before the job becomes claimable.
A number is milliseconds from now, a string accepts <n><ms|s|m|h|d>, and
a Date is treated as an absolute run time. Omit it to make the job due
immediately.
priority?: number
Sort priority among jobs that are due at the same time.
Higher numbers are claimed first. Defaults to 0.
key?: string
Dedupe key for active work in this queue.
When set, at most one non-terminal job may exist for (queue, key).
Finished, dead, cancelled, or errored jobs release the key.
retry?: Partial<JobRetryPolicy>
Retry policy overrides for this job.
Values are merged with the service defaults. Throw
NonRetryableJobError from a handler to bypass retries and dead-letter
the job immediately.
timeoutMs?: number
Per-attempt timeout in milliseconds.
When set, a running attempt that exceeds this duration is treated as a failed attempt and follows the retry policy.
interface JobsScheduleOptions {
Options for Jobs.schedule().
Exactly one of cron or every sets the cadence; the rest tune overlap,
catch-up after downtime, the target queue, and the retry policy copied onto
each enqueued job. Cron cadences evaluate in UTC.
import { Jobs } from 'fino:jobs';
const jobs = await Jobs.open({ path: './.fino/jobs.db' });
await jobs.schedule('nightly-report', 'report', {}, {
cron: '0 3 * * *',
queue: 'reports',
overlap: 'skip',
catchup: 'one',
retry: { maxAttempts: 2 },
});Properties
cron?: string
Five-field UTC cron expression or supported alias such as @daily.
Use either cron or every, not both. Cron schedules are evaluated in
UTC.
every?: string
Fixed interval schedule using <n><ms|s|m|h|d> syntax.
Use either every or cron, not both.
queue?: string
Queue used for jobs created by this schedule.
Defaults to 'default'.
overlap?: 'skip' | 'allow'
Overlap behavior when a previous scheduled job is still active.
'skip' avoids enqueueing another job while one from this schedule is
pending, running, waiting, claimed, or errored. 'allow' always enqueues
the due occurrence.
catchup?: 'skip' | 'one'
Catch-up behavior after downtime or delayed scheduler ticks.
'skip' advances to the next future occurrence without backfilling.
'one' enqueues at most one missed occurrence.
retry?: Partial<JobRetryPolicy>
Retry policy applied to jobs created by the schedule.
Values are copied onto each enqueued job when it is created.
Classes
class NonRetryableJobError extends Error {
A job that could not complete and should not be retried.
Throw from a task handler to send the job straight to the dead-letter state regardless of remaining attempts. Use it for failures that cannot succeed on a retry — malformed input, a permanent 4xx from an upstream service, or a business-rule rejection.
import { task } from 'fino:task';
import { NonRetryableJobError } from 'fino:jobs';
const charge = task({
name: 'charge',
run: async (input: { amount: number }) => {
if (input.amount <= 0) {
throw new NonRetryableJobError(`invalid amount: ${input.amount}`);
}
return input.amount;
},
});Constructors
constructor(message: string)
Construct the error with a human-readable failure reason.
class Jobs {
Handle to the jobs system: pushes, schedules, worker registration, and
job lifecycle operations. Create with Jobs.open().
Static Methods
static async open(opts: JobsOptions): Promise<Jobs>
Open the jobs system.
Detects the orchestrator's fino:jobs/control facade and becomes a
client of the runtime-hosted service when present; otherwise hosts the
service in this realm.
import { Jobs } from 'fino:jobs';
await using jobs = await Jobs.open({ path: './.fino/jobs.db' });Methods
push(task: string, input: unknown, opts: JobsPushOptions = {}): Promise<JobRecord>
Enqueue a job by task name.
Task instances are never pushed — a job row must be restorable from the database alone, so work requests carry only the name and input.
import { Jobs } from 'fino:jobs';
const jobs = await Jobs.open({ path: './.fino/jobs.db' });
await jobs.push('ingest', { url: 'https://example.com' }, { delay: '5s' });schedule(
name: string,
task: string,
input: unknown,
opts: JobsScheduleOptions
): Promise<ScheduleRecord>
Create or replace a named schedule that enqueues task on a cron or
interval cadence (UTC).
import { Jobs } from 'fino:jobs';
const jobs = await Jobs.open({ path: './.fino/jobs.db' });
await jobs.schedule('nightly-compact', 'compact', {}, { cron: '@daily' });unschedule(name: string): Promise<boolean>
Remove a named schedule. Returns whether it existed.
get(id: string): Promise<JobRecord | null>
Load one job by id.
list(filter: {
queue?: string;
status?: JobStatus;
task?: string;
scheduleId?: string;
limit?: number;
offset?: number;
} = {}): Promise<JobRecord[]>
List jobs, newest first.
job(id: string): ReadonlySignal<JobRecord | null>
Watch one job by id.
The signal is seeded from get(id) while subscribed and refreshes when
local jobs runtime events mention the same job. A periodic reconcile also
runs while hot so missed external writes eventually converge.
stats(queue?: string): ReadonlySignal<QueueStats>
Watch aggregate queue statistics.
The signal starts when subscribed, refreshes from the jobs store on runtime job events, and reconciles every five seconds while hot.
schedules(): Promise<ScheduleRecord[]>
List schedules.
cancel(id: string): Promise<boolean>
Cancel a job. Pending and parked jobs cancel immediately; a running job is marked cancelled but its in-flight execution is not interrupted.
retry(id: string): Promise<boolean>
Requeue a dead, errored, or cancelled job from attempt zero. Durable jobs keep their workflow run and resume from the last checkpoint.
signal(id: string, name: string, payload?: unknown): Promise<void>
Deliver an external signal to a parked durable job and make it claimable.
import { Jobs } from 'fino:jobs';
const jobs = await Jobs.open({ path: './.fino/jobs.db' });
await jobs.signal('job-id', 'approved', { by: 'ada' });wait(id: string, opts: {
timeoutMs?: number;
} = {}): Promise<JobRecord>
Wait for a job to reach a terminal state.
async process(opts: {
tasks: Task[];
concurrency?: number;
}): Promise<void>
Register this realm as an inline processor for tasks: claimed jobs for
those task names execute on this realm's event loop.
import { task } from 'fino:task';
import { Jobs } from 'fino:jobs';
const jobs = await Jobs.open({ path: './.fino/jobs.db' });
await jobs.process({ tasks: [task({ name: 'noop', run: async () => null })] });async workers(opts: {
entry: string;
size?: number;
realm?: Omit<RealmOptions, 'entry' | 'thread'>;
}): Promise<void>
Register a pool processor: an exclusive worker pool whose entry module
default-exports a Task (children included). Each job runs in a fresh
worker realm.
import { Jobs } from 'fino:jobs';
const jobs = await Jobs.open({ path: './.fino/jobs.db' });
await jobs.workers({ entry: './handlers/ingest.ts', size: 4 });async stop(): Promise<void>
Stop the locally-hosted service (drain, then close). In client mode the runtime owns the service lifecycle and this is a no-op.