@ballistix.digital/task-scheduler

A NestJS module that runs background work as durable task rows behind a pluggable queue engine. A task is a database row with a processing status, and the queue only transports the message that names that row. One engine ships, on pg-boss, behind its own entry, so an application installs pg-boss only when it uses that engine.

This file is the consumer guide. The pages that describe how the package is built live in the Ballistix wiki.

The model

The processingStatus column of the row is the lifecycle.

stateDiagram-v2
    classDef waiting fill:#FFD700,stroke:#333,color:#000
    classDef active fill:#87CEEB,stroke:#333,color:#00008B
    classDef done fill:#90EE90,stroke:#333,color:#006400
    classDef failed fill:#FFB6C1,stroke:#DC143C,color:#000

    [*] --> PENDING: trigger() saves the row
    PENDING --> PROCESSING: claim (only when PENDING)
    PROCESSING --> COMPLETED: handle() returns
    PROCESSING --> PENDING: handle() throws, status >= 500, attempts left
    PROCESSING --> FAILED: handle() throws, status < 500 or attempts exhausted
    COMPLETED --> [*]
    FAILED --> [*]

    class PENDING waiting
    class PROCESSING active
    class COMPLETED done
    class FAILED failed

Read the two exits of PROCESSING: a retry returns the row to PENDING, and every other failure settles it FAILED.

Three mechanisms carry the guarantees:

• Transactional outbox. The task row and its task message commit in one transaction, or neither commits.
• Claim. The processor sets PROCESSING with a conditional update on PENDING. A repeated delivery loses the claim and stops.
• Retry guard. A thrown error with status >= 500 sends the row back to PENDING while attempts remain. Any other error settles the row FAILED. The queue owns the retry policy: the guard reads the retryLimit of the message and the processingMaxRetries of the row, and the lower one wins.

One run, step by step

sequenceDiagram
    box rgb(230,230,250) Application
        participant App as Application service
        participant P as ReportTaskProcessor
    end
    box rgb(144,238,144) Engine
        participant E as TaskEngine
    end
    box rgb(211,211,211) Postgres
        participant DB as Postgres
    end

    rect rgb(255,250,205)
        Note over App,DB: 1. trigger: row and message in one transaction
        App->>P: trigger({ reportId }, transaction)
        P->>DB: INSERT report_task (PENDING)
        P->>E: enqueue('report', { id }, transaction)
        E->>DB: INSERT task message (same transaction)
        P-->>App: the saved row
    end

    rect rgb(224,255,224)
        Note over App,DB: 2. delivery: claim, handle, land
        E->>P: deliver { id }, attempt 0
        P->>DB: UPDATE status = PROCESSING WHERE id AND status = PENDING
        P->>P: handle(task)
        P->>DB: UPDATE status = COMPLETED
        P->>P: afterCompleted(task)
    end

Look at the first block: the row and the message travel on one transaction, so a rollback leaves neither.

Two processors

The difference between the two base classes is who creates the task row.

| | Task processor | Scheduled task processor | | --- | --- | --- | | Base class | AbstractTaskProcessor | AbstractScheduledTaskProcessor | | Who creates the row | Your code, with trigger() | The processor itself, on every tick, with the same trigger() | | When it runs | When a row is triggered | On the cron expression, and when a row is triggered | | Typical use | An upload to parse, an invitation to send | A nightly sync, an hourly cleanup | | Extra members | none | cronExpression, a queueOptions default that adds the singleton policy |

A tick is a message without a task id. The scheduled processor answers it with its own trigger(), and the { id } message that enqueues runs the work like a manual run. The queue of a scheduled processor runs one message at a time across every replica, so a tick never overlaps a manual run.

Install

npm install @ballistix.digital/task-scheduler pg-boss

A browser build that reads the ./types entry alone installs the package by itself:

npm install @ballistix.digital/task-scheduler

The package needs Node 22.12 or later and has no dependencies of its own. Every peer dependency is optional, so npm installs none of them for you: install in the application what the entries it imports need.

| Package | Range | Needed by | | --- | --- | --- | | @nestjs/common | ^11 | the root entry | | @nestjs/core | ^11 | the root entry | | typeorm | >=0.3.0 <2.0.0 | the root entry | | reflect-metadata | ^0.2 | the root entry | | @ballistix.digital/exception-mapper | ^0.3.0 \|\| ^0.4.0 | the root entry | | @ballistix.digital/exception-types | ^0.4.0 | the root entry | | pg-boss | ^12 | the ./pg-boss entry |

| Entry | Peers it needs | | --- | --- | | @ballistix.digital/task-scheduler | all of them except pg-boss | | @ballistix.digital/task-scheduler/pg-boss | pg-boss, next to the peers of the root entry | | @ballistix.digital/task-scheduler/testing | the peers of the root entry | | @ballistix.digital/task-scheduler/types | none |

Only the ./pg-boss entry loads pg-boss, so an application with an engine of its own, or a package that only tests against TaskEngineMock, leaves pg-boss out. The ./types entry imports no framework at all: it carries ProcessingStatusEnum alone, so a browser application that shares DTO types with the backend re-exports the enum from that entry instead of copying it.

The two exception packages carry the error envelope and the mappers that build it. A processor stores that envelope in processingError and reads its status to decide the retry. A worker thread maps what its worker function threw through the same mappers and sends the envelope back.

A worker file that runs from its TypeScript source, which is what a Jest suite does, loads in its thread through ts-node/register. Install ts-node as a dev dependency for that, pass other loader flags through WorkerModuleOptions.execArgv, or test with WorkerRunnerMock, which needs no loader. See Run CPU-bound work in a worker thread.

Register the module

import { ExceptionModule } from '@ballistix.digital/exception-mapper';
import { Module } from '@nestjs/common';
import { TaskModule } from '@ballistix.digital/task-scheduler';
import { PgBossEngineService } from '@ballistix.digital/task-scheduler/pg-boss';

@Module({
	imports: [
		ExceptionModule.forRoot(),
		TaskModule.forRoot({
			engine: new PgBossEngineService({
				connection: { host: 'localhost', port: 5432, user: 'app', password: 'secret', database: 'app' },
				schema: 'pgboss',
				poolSize: 5,
			}),
		}),
	],
})
export class AppModule {}

If the options need injection, use TaskModule.forRootAsync({ imports, inject, useFactory }). The factory returns the same options object.

Both modules are global. A processor and a service alike inject TaskRuntime without a further import; nothing else of the package is injectable. TaskModule starts the engine in onModuleInit and stops it in onApplicationShutdown. Nest runs that hook after the HTTP server has drained its in-flight requests, so a request that is still running can enqueue until the end. Call app.enableShutdownHooks() in the bootstrap of the application: without it a SIGTERM ends the process at once and neither the drain nor the stop happens. A test of the application swaps the engine by overriding the TASK_ENGINE token the module provides it under; see Swap the engine in tests.

The engine keeps its own pool and its own schema, apart from the TypeORM connection of the application. It reads no environment variables. The pg-boss engine page gives the other options.

Give the task a row

Compose ProcessableTaskMixin onto the entity base class of the application:

import { ProcessableTaskMixin } from '@ballistix.digital/task-scheduler';
import { BaseEntity, Column, Entity, PrimaryGeneratedColumn } from 'typeorm';

@Entity({ name: 'report_task' })
export class ReportTask extends ProcessableTaskMixin(BaseEntity) {
	@PrimaryGeneratedColumn('uuid')
	id: string;

	@Column({ type: 'uuid', nullable: false })
	reportId: string;
}

The mixin contributes five columns:

| Column | Type | Default | | --- | --- | --- | | processingStatus | enum processing_status, indexed | PENDING | | processingError | jsonb, nullable | null | | processingRetryCount | int | 0 | | processingMaxRetries | int | 5 (DEFAULT_RETRY_LIMIT) | | processingDuration | int, nullable, milliseconds | null |

processingError holds an ExceptionDto. Import that type from @ballistix.digital/exception-types: this package does not re-export it.

Every entity on the mixin shares the one Postgres enum type processing_status. Generate a migration for the new table.

Write a task processor

import { GenericBadRequestException } from '@ballistix.digital/exception-types';
import { Injectable } from '@nestjs/common';
import { InjectRepository } from '@nestjs/typeorm';
import { AbstractTaskProcessor, TaskRuntime } from '@ballistix.digital/task-scheduler';
import { Repository } from 'typeorm';
import { ReportTask } from './report.task.entity';
import { RenderService } from './render.service';

@Injectable()
export class ReportTaskProcessor extends AbstractTaskProcessor<ReportTask> {
	protected readonly queue = 'report';

	constructor(
		taskRuntime: TaskRuntime,
		@InjectRepository(ReportTask) reportTaskRepository: Repository<ReportTask>,
		private readonly renderService: RenderService,
	) {
		super(taskRuntime, reportTaskRepository);
	}

	/** A 400 settles the task on the first attempt: the input cannot become valid on a retry. */
	protected async handle(task: ReportTask): Promise<void> {
		if (!task.reportId) {
			throw new GenericBadRequestException('Task carries no reportId');
		}

		await this.renderService.render(task.reportId);
	}

	/** Runs after COMPLETED has committed. A failure here never reverts COMPLETED. */
	protected async afterCompleted(task: ReportTask): Promise<void> {
		await this.renderService.publish(task.reportId);
	}
}

Register the class as a provider of the owning module. On bootstrap the base creates the queue with queueOptions and subscribes to it.

Queue options

The queue is the one place that holds the retry policy. A processor declares retryLimit, retryDelaySeconds and retryBackoff in queueOptions, the engine applies them on bootstrap, and every message on the queue inherits them: the message of trigger(), the message of a schedule tick, and the message a service enqueues for a row it saved itself. No enqueue of the package passes retry options of its own, so every run on one queue retries the same way.

A processor that declares nothing keeps the default of the base, DEFAULT_QUEUE_OPTIONS: retryLimit: 5 (DEFAULT_RETRY_LIMIT) with retryBackoff: true. A processor that must not repeat a run declares retryLimit: 0: the first throw settles the row FAILED. A processor whose work needs a pause before the next attempt adds retryDelaySeconds:

protected readonly queueOptions: QueueOptions = {
	...DEFAULT_QUEUE_OPTIONS,
	retryDelaySeconds: 30,
};

Every start reconciles the queue with queueOptions. A changed expireInSeconds, retryLimit, retryDelaySeconds or retryBackoff reaches a queue an earlier release created. A changed policy does not: a policy is fixed when the queue is created, so the engine keeps the queue and logs a warning that names both policies. Give a processor that needs another policy a new queue name.

super() takes two arguments. The first is the TaskRuntime the module provides. Pass it up and forget it: the type carries enqueue() and nothing else, the class behind it is not exported, and a new dependency of the base changes that class and no processor. The second is the repository. Name the parameter after the entity, such as reportTaskRepository. If the subclass needs the repository itself, keep it in a private field with that name.

You implement handle(). You can also override:

| Member | Default | Override when | | --- | --- | --- | | afterCompleted(task) | nothing | work must follow a committed COMPLETED, such as a notification | | buildTask(input) | repository.create(input ?? {}) | the row needs values the entity defaults cannot give | | onMessage(message) | process the id, warn when there is none | you need another message shape (rare) |

handle() runs outside any held transaction, so a slow run holds no row lock. It opens its own transactions when it needs them. The queue is a plain string. Keep the queue names of the application in one enum.

Write a scheduled task processor

The constructor follows the task processor above.

@Injectable()
export class AfasSyncTaskProcessor extends AbstractScheduledTaskProcessor<AfasSyncTask> {
	protected readonly queue = 'afas-sync';
	protected readonly cronExpression = '5 */6 * * *';

	/** This sync must not repeat a partial run, so the queue allows no retry. */
	protected readonly queueOptions: QueueOptions = { policy: 'singleton', expireInSeconds: 1800, retryLimit: 0 };

	protected async handle(task: AfasSyncTask): Promise<void> {
		await this.afasService.syncCustomers();
		await this.afasService.syncProjects();
	}
}

A scheduled processor is a task processor plus cronExpression. Cron expressions are UTC. The base creates the queue with the singleton policy, so one run is active at a time across every replica.

queueOptions holds the time budget of one run (expireInSeconds, default 600) and the retry policy of the queue, which starts from DEFAULT_QUEUE_OPTIONS. A tick and a manual run inherit the same policy. The example gives the queue a budget of 30 minutes and no retry: retryLimit: 0 settles every run FAILED on its first throw, whatever processingMaxRetries the row carries.

A tick creates the row with trigger() and no input. If the entity has a column the tick cannot leave empty, override buildTask().

Schedules tick in every environment. Keep a processor safe to run from staging through its own configuration, not through the absence of a schedule.

Trigger a run from a service

@Injectable()
export class ReportService {
	constructor(
		private readonly dataSource: DataSource,
		private readonly processor: ReportTaskProcessor,
	) {}

	public async requestReport(reportId: string): Promise<ReportTask> {
		return this.dataSource.transaction(async (transaction) => {
			await transaction.update(Report, reportId, { requestedAt: new Date() });

			return this.processor.trigger({ reportId }, transaction);
		});
	}
}

trigger(input, transaction) saves the row and enqueues { id } on the same transaction, with no enqueue options: the message inherits the retry policy of the queue. Without a transaction the processor opens one. A rolled-back transaction leaves no row and no task message.

trigger() returns the saved row, so the caller can poll processingStatus and processingError. A manual trigger() on a scheduled processor works the same way, because the schedule and the manual runs share the one subscriber.

A processor never reads an application authentication context. Put audit values in the trigger() input.

Enqueue a row you saved yourself

A service that owns the row, such as an upload flow that marks an attachment PENDING, enqueues the message itself through TaskRuntime:

@Injectable()
export class UploadService {
	constructor(private readonly taskRuntime: TaskRuntime) {}

	public async onUploadReady(upload: Upload, transaction: EntityManager): Promise<void> {
		await transaction.update(Upload, upload.id, { processingStatus: ProcessingStatusEnum.PENDING });
		await this.taskRuntime.enqueue('upload', { id: upload.id }, transaction, { dedupeKey: upload.id });
	}
}

enqueue() is the one call TaskRuntime exposes. The message rides the transaction you pass, so the row and the message commit together. Pass a dedupeKey when the policy of the queue needs one, and no retry options: the message inherits the retry policy of the queue, the same as a trigger(). The consumer surface diagram shows what sits behind the token and why you cannot reach it.

Errors and retries

When handle() throws, the ExceptionMapperRegistry normalizes the thrown value into an ExceptionDto. The processor stores that envelope in processingError, and the status of the envelope decides what happens next.

flowchart TD
    T["handle() throws"] --> N["error = exceptionMapperRegistry.toExceptionDto(thrown)"]
    N --> S{"error.status >= 500?"}
    S -- no --> F["FAILED, processingError stored, no rethrow, telemetry line"]
    S -- yes --> A{"attempt < min(message.retryLimit, processingMaxRetries)?"}
    A -- no --> F
    A -- yes --> R["PENDING, processingError stored, rethrow so the engine retries with backoff"]

    classDef decision fill:#FFD700,stroke:#333,stroke-width:2px,color:#000
    classDef retry fill:#87CEEB,stroke:#333,stroke-width:2px,color:#00008B
    classDef terminal fill:#FFB6C1,stroke:#DC143C,stroke-width:2px,color:#000
    classDef step fill:#F5F5F5,stroke:#333,stroke-width:1px,color:#000
    class S,A decision
    class R retry
    class F terminal
    class T,N step

Look at the first decision: the registry normalizes every thrown value, so the guard reads one field whatever the code threw.

If the code matters, throw a BaseException of the types package. Otherwise throw any value that carries a status:

| Thrown value | Stored ExceptionDto | Retried? | | --- | --- | --- | | new GenericBadRequestException('bad input') | title: 'bad input', status: 400, code: 'GENERIC_ERROR' | no, the input is wrong | | new QueryFailedError(...) with SQLSTATE 23505 | status: 400, code: 'DATABASE_ERROR' | no, the statement is wrong | | new QueryFailedError(...) with SQLSTATE 40P01 | status: 500, code: 'DATABASE_ERROR' | yes, while attempts remain | | { message, status: 400 } | title: message, status: 400, code: 'GENERIC_ERROR' | no, the input is wrong | | { message, status: 503 } | title: message, status: 503, code: 'GENERIC_ERROR' | yes, while attempts remain | | new Error('boom') | title: 'boom', status: 500, code: 'GENERIC_ERROR' | yes, while attempts remain | | 'a string' | title: 'a string', status: 500 | yes, while attempts remain |

The registry tries its mappers in order and the generic mapper is last. A thrown value that names no code therefore becomes GENERIC_ERROR, and one errors detail carries the message.

The queue decides the retry budget through its retryLimit, and the engine stamps that limit on every message it delivers. processingMaxRetries on the row is a row-level budget that can only lower the budget of the queue: the guard retries while attempt is below both numbers. A row that asks for more retries than its queue allows gets the retries of the queue, because the engine stops redelivering at its own limit and a row sent back to PENDING past that limit would wait for a message that never comes. Both numbers default to 5 (DEFAULT_RETRY_LIMIT).

processingRetryCount records how many attempts ran. processingDuration records how long the last attempt of handle() ran, in milliseconds, whatever its outcome. The engine schedules the retries with the delay and the backoff of the queue. The row is the durable record you read.

Run CPU-bound work in a worker thread

Node runs JavaScript on one thread. A render of a document or an email blocks the event loop for its full duration, and while it runs the API answers nothing. WorkerRunnerService runs one function in a thread of its own, and the caller awaits the result the way it awaits any promise.

flowchart LR
    S["service or handle()<br/>workerRunner.run(__dirname, 'render.worker', input)"] --> R["WorkerRunnerService<br/>one thread per run"]
    R --> B["thread<br/>loads the worker file,<br/>calls its default export"]
    B -->|"result"| R
    B -->|"error envelope"| R
    R -->|"resolve"| OK["the result"]
    R -->|"reject"| E["WorkerException<br/>status, code, details kept"]

    classDef main fill:#87CEEB,stroke:#333,stroke-width:2px,color:#00008B
    classDef thread fill:#FFD700,stroke:#333,stroke-width:2px,color:#000
    classDef done fill:#90EE90,stroke:#333,color:#006400
    classDef failed fill:#FFB6C1,stroke:#DC143C,color:#000
    class S,R main
    class B thread
    class OK done
    class E failed

Register the module once, at the root. It is global, so any feature module injects the runner without importing anything.

import { WorkerModule } from '@ballistix.digital/task-scheduler';

WorkerModule.forRootAsync({
	inject: [ConfigService],
	useFactory: (config: ConfigService<EnvModel>) => ({
		timeoutSeconds: config.get('WORKER_TIMEOUT_SECONDS'),
	}),
});

timeoutSeconds is the timeout of every run that passes none of its own. The default is 300. execArgv is the list of flags a thread starts with when it loads a .ts worker file; the default is ['-r', 'ts-node/register'].

Write a worker file

A worker file is a plain module next to the service that calls it, named <name>.worker.ts. It exports the worker function as its default export, and it imports nothing from this package. The function takes one input and returns one result. Both cross the thread as structured clones, so a class instance arrives as a plain object and a Node Buffer arrives as a Uint8Array.

// render.worker.ts
export interface RenderInput {
	template: Uint8Array;
	data: Record<string, unknown>;
}

export interface RenderOutput {
	file: Uint8Array;
}

export default async function render(input: RenderInput): Promise<RenderOutput> {
	const file = renderWithSomeLibrary(input.template, input.data);

	return { file };
}

// Optional: the buffers of the result to move instead of copy.
export const transfer = (output: RenderOutput): ArrayBuffer[] => [output.file.buffer as ArrayBuffer];

The worker function is a pure function of its input: no Nest injection, no TypeORM, no request context. Everything else, the database reads, the blob storage and the task lifecycle, stays on the main thread. Export the input and the output types, so the caller imports the types and nothing else from the file.

Call the runner

import { WorkerRunnerService } from '@ballistix.digital/task-scheduler';
import type { RenderInput, RenderOutput } from './render.worker';

@Injectable()
export class ReportService {
	constructor(private readonly workerRunner: WorkerRunnerService) {}

	async render(template: Uint8Array, data: Record<string, unknown>): Promise<Uint8Array> {
		const output = await this.workerRunner.run<RenderInput, RenderOutput>(
			__dirname,
			'render.worker',
			{ template, data },
			{ timeoutSeconds: 120, transfer: [template.buffer as ArrayBuffer] },
		);

		return output.file;
	}
}

run(dir, workerName, input, options?) finds <workerName>.ts in dir when the source exists on disk and <workerName>.js otherwise, so the same call works under a test runner and from dist in production. Pass __dirname, and the worker file sits next to the caller in both trees.

A run that passes timeoutSeconds wins over the value of the module. A run that lists buffers in transfer moves them to the thread instead of copying them: the buffer has byte length 0 on the calling side afterwards. Only a buffer that owns its ArrayBuffer can move. A small Node Buffer shares a pooled slab with others, so copy it out first, with Buffer.allocUnsafeSlow or new Uint8Array(buffer).

A task processor calls the runner from handle() the same way. The row, the claim and the retry stay on the main thread; only the CPU-bound call moves.

What a failure becomes

The thread maps what the worker function threw through the exception mapper registry, the same mappers a processor uses, and posts the envelope. The run rejects with a WorkerException, a BaseException restored from that envelope, with the stack the error had in the thread. So the retry guard reads a worker failure like any other: a 400 lands FAILED, a 500 retries.

| What happened in the thread | The run rejects with | Status, code | | --- | --- | --- | | The worker function threw a BaseException | WorkerException | its own status and code, its details kept | | The worker function threw a class-validator ValidationException | WorkerException | 400, VALIDATION_FAILED | | The worker function threw anything else | WorkerException | 500, GENERIC_ERROR, the message as title | | An error escaped the promise chain, from a timer for example | WorkerException | as above, by what was thrown | | The thread could not start, for example on a buffer that cannot move | GenericErrorException | 500, GENERIC_ERROR, Worker <name> could not start: <cause> | | The run timed out | GenericErrorException | 500, GENERIC_ERROR, Worker <name> timed out after <n> s | | The thread exited before it posted a result | GenericErrorException | 500, GENERIC_ERROR, Worker <name> exited with code <code> before it posted a result | | The application shut down during the run | GenericErrorException | 500, GENERIC_ERROR, Worker <name> was terminated because the runner is shut down |

The run settles once, on the first of those events, and the runner terminates the thread after every settle, also after a result. A worker function that leaves a timer or a socket open cannot keep a thread alive.

On shutdown the runner terminates every thread still running and rejects its run, and it refuses every run that comes after. A task processor that was inside a run settles that attempt as a 500 and retries it after the restart. app.enableShutdownHooks() is what makes a signal reach the hook, the same as for the engine.

Telemetry

The base processor writes one structured line per attempt through the Nest Logger, with a flat context object as the second argument. Dashboards and monitors build on these fields.

| Field | Value | | --- | --- | | message | Task <id> on <queue> executed with status <status> in <n> ms | | level | info for completed, warn for pending, error for failed | | context | task_queue, task_id, task_status, task_duration_ms, task_scheduled, ddtags | | context, conditional | task_error_code when the attempt failed |

The status of the line is the ProcessingStatusEnum member the landing wrote on the row. The line lower-cases it, so the message, task_status and the task.status tag read completed, pending for a retry, or failed for a terminal failure. task_duration_ms is the processingDuration of the row, so the line and the row never disagree.

task_scheduled is true on every run of a processor that carries a cronExpression, whether a tick or a manual trigger() created the row, and false on every other run. The tag task.scheduled:true joins ddtags on the true runs only, because a tag exists to filter on. Query the field and the queue together: @task_scheduled:true @task_queue:<QUEUE>.

The architecture page gives the ddtags format.

Test a processor

The ./testing entry gives TaskEngineMock. It performs no I/O, records every call, and delivers a message on demand.

import { ExceptionModule } from '@ballistix.digital/exception-mapper';
import { Test } from '@nestjs/testing';
import { getRepositoryToken } from '@nestjs/typeorm';
import { ProcessingStatusEnum, TaskModule } from '@ballistix.digital/task-scheduler';
import { TaskEngineMock } from '@ballistix.digital/task-scheduler/testing';

it('completes the task the engine delivers', async () => {
	const engine = new TaskEngineMock();
	const repository = testDataSource.getRepository(ReportTask);

	const moduleRef = await Test.createTestingModule({
		imports: [ExceptionModule.forRoot(), TaskModule.forRoot({ engine })],
		providers: [ReportTaskProcessor, { provide: getRepositoryToken(ReportTask), useValue: repository }],
	}).compile();
	await moduleRef.init();

	const processor = moduleRef.get(ReportTaskProcessor);
	const task = await processor.trigger({ reportId: 'r-1' });
	expect(engine.enqueued[0].data).toEqual({ id: task.id });

	await engine.deliver('report', { id: task.id });

	const settled = await repository.findOneByOrFail({ id: task.id });
	expect(settled.processingStatus).toBe(ProcessingStatusEnum.COMPLETED);
});

moduleRef.init() runs the bootstrap hook, so the processor subscribes before the first deliver. Read these members of the mock:

| Member | Holds | | --- | --- | | enqueued | every enqueue call: queue, data, options, transaction | | queues | every createQueue call: queue, options | | subscriptions | the handler per queue | | schedules | every schedule call: queue, cron | | unscheduled | every queue that was unscheduled | | deliver(queue, data, attempt, retryLimit?) | hands a message to the subscribed handler | | reset() | empties the recordings and keeps the subscriptions |

To test a scheduled processor, deliver an empty object. engine.deliver('afas-sync', {}) is a tick.

deliver() stamps a retryLimit on the message the way an engine does. Left out, it is the retryLimit the queue was created with, so a processor under test sees its own policy. A queue created without one gives infinity, so a spec delivers as many attempts as it asks for.

A spec that boots the module once and runs several cases on it empties the recordings between them, and keeps the handlers the processors subscribed on bootstrap:

beforeEach(() => engine.reset());

Run a worker function without a thread

The ./testing entry gives WorkerRunnerMock. It has the run() signature of the service and runs the worker function on the calling thread: the test runner transforms the .ts worker file the way it transforms any other module, so no loader is installed, and a thrown value becomes the same WorkerException a thread would send. It records every run in runs and empties them on reset(). There is no timeout, and a buffer listed in transfer stays readable.

import { WorkerRunnerService } from '@ballistix.digital/task-scheduler';
import { WorkerRunnerMock } from '@ballistix.digital/task-scheduler/testing';

const moduleRef = await Test.createTestingModule({ imports: [AppModule] })
	.overrideProvider(WorkerRunnerService)
	.useClass(WorkerRunnerMock)
	.compile();

const runner = moduleRef.get<WorkerRunnerMock>(WorkerRunnerService);
expect(runner.runs).toEqual([{ dir: expect.any(String), workerName: 'render.worker', input, options: undefined }]);

A service built by hand takes new WorkerRunnerMock() where it takes the runner. To run the real thread in a spec instead, install ts-node and use the service as it is.

Swap the engine in tests

TaskEngine is a type, not a provider token, so overrideProvider(TaskEngine) finds nothing to replace. TASK_ENGINE is the token the module holds the engine under, and it is exported for this: a test that boots the whole application overrides it and needs no token of its own.

import { TASK_ENGINE, TaskModule } from '@ballistix.digital/task-scheduler';
import { PgBossEngineService } from '@ballistix.digital/task-scheduler/pg-boss';
import { TaskEngineMock } from '@ballistix.digital/task-scheduler/testing';

// AppModule
TaskModule.forRootAsync({
	inject: [ConfigService],
	useFactory: (config: ConfigService) => ({ engine: new PgBossEngineService({ ... }) }),
});

// A test
const moduleRef = await Test.createTestingModule({ imports: [AppModule] })
	.overrideProvider(TASK_ENGINE)
	.useClass(TaskEngineMock)
	.compile();

The override replaces the provider the registration built, so the factory of forRootAsync never runs its engine and the application reaches the mock everywhere: TaskModule starts and stops it, every processor subscribes on it, and TaskRuntime.enqueue() records on it. Read it back with moduleRef.get<TaskEngineMock>(TASK_ENGINE), and drive the processors with engine.deliver() the same way as above. useValue(new TaskEngineMock()) works as well, when the spec wants the instance before it compiles.

API reference

Everything below comes from the package root, except PgBossEngineService and PgBossEngineOptions, which come from @ballistix.digital/task-scheduler/pg-boss, TaskEngineMock, RecordedQueue, RecordedEnqueue, RecordedSchedule, WorkerRunnerMock and RecordedRun, which come from @ballistix.digital/task-scheduler/testing, and ExceptionDto, which comes from @ballistix.digital/exception-types. ProcessingStatusEnum also comes from @ballistix.digital/task-scheduler/types, the entry that imports no framework.

| Entry | Holds | | --- | --- | | @ballistix.digital/task-scheduler | The modules, the processors, TaskRuntime, the worker runner, the TASK_ENGINE token and the domain types | | @ballistix.digital/task-scheduler/pg-boss | PgBossEngineService and PgBossEngineOptions | | @ballistix.digital/task-scheduler/testing | TaskEngineMock, WorkerRunnerMock and their recorded-call types | | @ballistix.digital/task-scheduler/types | ProcessingStatusEnum, and nothing that imports a framework |

Every emitted module is also reachable by its path under dist/, because the Nest Swagger CLI plugin rebuilds an enum-typed DTO property as a require() of the declaration file the enum came from. Import through an entry in code you write.

Module

| Export | Purpose | | --- | --- | | TaskModule.forRoot(options) | Registers the module with an engine instance | | TaskModule.forRootAsync(options) | The same, with imports, inject and a useFactory | | TaskModuleOptions | { engine: TaskEngine } | | TaskModuleAsyncOptions | { imports?, inject?, useFactory } | | TASK_ENGINE | The token the module holds the engine under. A test overrides it to swap the engine | | PgBossEngineService | The pg-boss engine, constructed with PgBossEngineOptions; from the ./pg-boss entry | | PgBossEngineOptions | { connection, schema, poolSize?, superviseIntervalSeconds?, monitorIntervalSeconds?, queueCacheIntervalSeconds? } |

Processors

| Export | Purpose | | --- | --- | | AbstractTaskProcessor<T> | Base class of a task processor | | AbstractScheduledTaskProcessor<T> | Base class of a scheduled task processor | | TaskRuntime | The injectable of the task layer: the first argument of super() in a processor, and enqueue() for a service | | TaskEngine | The engine contract an application passes to TaskModule; a type, not something you inject |

AbstractTaskProcessor<T> declares two abstract members, queue and handle(task). trigger(input?, transaction?) is its only public method. The protected members a subclass reads or overrides are queueOptions, afterCompleted(task), buildTask(input), onMessage(message), repository and logger. The engine, the telemetry and the exception mapper registry sit behind TaskRuntime, and only the base reads them.

AbstractTaskProcessor<T> sets queueOptions to DEFAULT_QUEUE_OPTIONS. AbstractScheduledTaskProcessor<T> adds the abstract cronExpression and sets queueOptions to { ...DEFAULT_QUEUE_OPTIONS, policy: 'singleton', expireInSeconds: 600 }.

Domain

| Export | Purpose | | --- | --- | | ProcessableTaskMixin(Base) | Adds the five processing columns to an entity class | | ProcessableTask | The interface of a row with those columns | | ProcessingStatusEnum | PENDING, PROCESSING, COMPLETED, FAILED. Also on the ./types entry, for a browser build that shares the status | | ExceptionDto | { title, status, code, errors }, stored in processingError. Import it from @ballistix.digital/exception-types | | TaskMessage<T> | { data, attempt, retryLimit }, one delivery. retryLimit is how many redeliveries the engine allows | | EnqueueOptions | dedupeKey, retryLimit, retryBackoff, retryDelaySeconds, expireInSeconds. The retry fields are for a message that must differ from its queue | | QueueOptions | policy, expireInSeconds, retryLimit, retryDelaySeconds, retryBackoff | | DEFAULT_QUEUE_OPTIONS | { retryLimit: DEFAULT_RETRY_LIMIT, retryBackoff: true }, the queueOptions a processor starts from | | DEFAULT_RETRY_LIMIT | 5, the default of processingMaxRetries and of the retryLimit of a queue | | QueuePolicy | standard, short, singleton, stately | | Constructor<T> | The constructor type the mixin accepts |

Worker

| Export | Purpose | | --- | --- | | WorkerModule.forRoot(options?) | Registers the worker runner with its options | | WorkerModule.forRootAsync(options) | The same, with imports, inject and a useFactory | | WorkerModuleOptions | { timeoutSeconds?, execArgv? }: the timeout of a run without one, default 300 s, and the flags of a thread that loads a .ts worker file, default -r ts-node/register | | WorkerModuleAsyncOptions | { imports?, inject?, useFactory } | | WorkerRunnerService | The injectable: run<TIn, TOut>(dir, workerName, input, options?) resolves with the result of the worker function | | WorkerRunOptions | { timeoutSeconds?, transfer? }: the timeout of this run, and the buffers of the input to move | | WorkerException | What a run rejects with when the worker function threw: a BaseException restored from the envelope the thread posted, with the stack of the thread | | DEFAULT_WORKER_TIMEOUT_SECONDS | 300 | | DEFAULT_TYPESCRIPT_EXEC_ARGV | ['-r', 'ts-node/register'] | | WorkerRunnerMock | The runner that runs the worker function on the calling thread and records each run; from the ./testing entry | | RecordedRun | { dir, workerName, input, options }, one recorded run |

A worker file exports its worker function as default and, when it moves result buffers, transfer(output): ArrayBuffer[]. It imports nothing from this package.

Engine contract

TaskEngine is abstract. An application constructs one implementation and hands it to TaskModule; the module and the processors call it. It is not injectable: a service enqueues through TaskRuntime.

| Method | Purpose | | --- | --- | | start() | Opens the connection and starts delivering | | stop() | Closes the connection. Safe to call more than once | | createQueue(queue, options?) | Creates the queue with these options, and reconciles the options an existing queue still accepts. The policy is not one of them | | enqueue(queue, data, transaction?, options?) | Puts one message on a queue. Returns the message id, or null when the policy of the queue collapsed it | | subscribe(queue, handler) | Registers the handler of a queue | | schedule(queue, cron) | Registers a recurring tick that inherits the options of the queue | | unschedule(queue) | Removes a schedule. Never throws for an unknown name |

Guarantees

• The task row and its task message commit together. A fire-and-forget enqueue does not exist.
• Every run is a row. "Last run" is a query, not a log search.
• The claim is atomic. A second delivery of the same task stops at the claim.
• One queue per processor. A tick and a manual run never overlap on one replica.
• The queue owns the retry policy. processingMaxRetries on the row can lower the budget of the queue and never raise it. An error with status below 500 settles the task FAILED on the first attempt.
• afterCompleted() is best effort. Its failure is logged and never reverts COMPLETED.
• A processor never reads an application authentication context. Audit values travel in the trigger() input.
• A worker run settles once, and its thread is terminated after every settle. A failure of the worker function keeps its status, code and details across the thread.