@boringnode/queue

A simple and efficient queue system for Node.js applications. Built for simplicity and ease of use, @boringnode/queue allows you to dispatch background jobs and process them asynchronously with support for multiple queue adapters.

Installation

npm install @boringnode/queue

Features

• Multiple Queue Adapters: Redis, Knex, Kysely (PostgreSQL, MySQL, SQLite), and Sync
• Type-Safe Jobs: TypeScript classes with typed payloads
• Delayed Jobs: Schedule jobs to run after a delay
• Priority Queues: Process high-priority jobs first
• Bulk Dispatch: Efficiently dispatch thousands of jobs at once
• Job Grouping: Organize related jobs for monitoring
• Job Deduplication: Prevent duplicate jobs with custom IDs
• Retry with Backoff: Exponential, linear, or fixed backoff strategies
• Job Timeout: Fail or retry jobs that exceed a time limit
• Job History: Retain completed/failed jobs for debugging
• Scheduled Jobs: Cron or interval-based recurring jobs
• Auto-Discovery: Automatically register jobs from specified locations
• Development Hot Reload: Run the latest job implementation without restarting workers

Quick Start

1. Define a Job

import{Job}from'@boringnode/queue'importtype{JobOptions}from'@boringnode/queue/types'interfaceSendEmailPayload{to: string}exportdefaultclassSendEmailJobextendsJob<SendEmailPayload>{staticoptions: JobOptions={queue: 'email',}asyncexecute(): Promise<void>{console.log(`Sending email to: ${this.payload.to}`)}}

Note

The job name defaults to the class name (SendEmailJob). You can override it with name: 'CustomName' in options.

Warning

If you minify your code in production, class names may be mangled. Always specify name explicitly in your job options.

2. Configure the Queue Manager

import{QueueManager}from'@boringnode/queue'import{redis}from'@boringnode/queue/drivers/redis_adapter'awaitQueueManager.init({default: 'redis',adapters: {redis: redis({host: 'localhost',port: 6379}),},locations: ['./app/jobs/**/*.ts'],})

3. Dispatch Jobs

// Simple dispatchawaitSendEmailJob.dispatch({to: 'user@example.com'})// With optionsawaitSendEmailJob.dispatch({to: 'user@example.com'}).toQueue('high-priority').priority(1).in('5m')

4. Start a Worker

import{Worker}from'@boringnode/queue'constworker=newWorker(config)awaitworker.start(['default','email'])

Hot Reloading Jobs

During development, jobs discovered from locations can be reloaded before each execution. This allows a long-running worker to use the latest saved job implementation without restarting.

Hot reload support integrates with Hot Hook. Install and initialize Hot Hook in your application, then enable hotReload on the queue manager.

npm install --save-dev hot-hook

awaitQueueManager.init({default: 'redis',adapters: {redis: redis({host: 'localhost',port: 6379}),},locations: ['./app/jobs/**/*.ts'],hotReload: process.env.NODE_ENV==='development',})

The process executing the jobs must be running with Hot Hook. For example, AdonisJS applications can start their development server with HMR enabled:

node ace serve --hmr

If the worker runs in a separate process, that worker process must also initialize Hot Hook. Enabling HMR only in the HTTP server does not affect jobs executed by another process. Follow the Hot Hook initialization guide when using it outside the AdonisJS development server.

How it works

Jobs loaded from locations are registered normally during QueueManager.init(). With hotReload: true, the worker dynamically imports the job module again before every execution. Hot Hook invalidates changed modules and their reloadable dependencies, allowing the dynamic import to return the latest job class. The queue marks these imports as Hot Hook boundaries, so jobs do not need to be repeated in Hot Hook's boundaries configuration. The queue does not install, initialize, or run Hot Hook itself.

Locator.registerFromGlob() can also enable this behavior directly:

import{Locator}from'@boringnode/queue'awaitLocator.registerFromGlob(['./app/jobs/**/*.ts'],{hotReload: true})constJobClass=awaitLocator.resolve('SendEmailJob')

Limitations

• Use hotReload during development only. Leave it disabled in production.
• Only jobs discovered from locations, or registered with Locator.registerFromGlob(), can be reloaded. Jobs registered manually with Locator.register() do not have a module path to reload.
• A running job instance keeps the version it started with. The latest version is used by the next execution.
• Adding, deleting, moving, or renaming a job requires a process restart so the job registry can be rebuilt.
• Changing a job's configured name also requires a restart. Already queued jobs continue to refer to the name stored when they were dispatched.
• Avoid import-time side effects in job modules, since their module code may execute again after an invalidation.

Bulk Dispatch

Efficiently dispatch thousands of jobs in a single batch operation:

const{ jobIds }=awaitSendEmailJob.dispatchMany([{to: 'user1@example.com'},{to: 'user2@example.com'},{to: 'user3@example.com'},]).group('newsletter-jan-2025').toQueue('emails').priority(3)console.log(`Dispatched ${jobIds.length} jobs`)

This uses Redis MULTI/EXEC or SQL batch insert for optimal performance.

Job Grouping

Organize related jobs together for monitoring and filtering:

// Group newsletter jobsawaitSendEmailJob.dispatch({to: 'user@example.com'}).group('newsletter-jan-2025')// Group with bulk dispatchawaitSendEmailJob.dispatchMany(recipients).group('newsletter-jan-2025')

The groupId is stored with job data and accessible via job.data.groupId.

Job Deduplication

Prevent the same job from being pushed multiple times. Four modes, all via .dedup():

Simple (skip while job exists)

// First dispatch - job is createdawaitSendInvoiceJob.dispatch({orderId: 123}).dedup({id: 'order-123'}).run()// Second dispatch with same dedup ID - silently skippedawaitSendInvoiceJob.dispatch({orderId: 123}).dedup({id: 'order-123'}).run()

Throttle (skip within TTL window)

// Within 5s, duplicates are skipped. After 5s, a new job is created.awaitSendEmailJob.dispatch({to: 'user@example.com'}).dedup({id: 'welcome-123',ttl: '5s'}).run()

Extend (reset TTL on duplicate)

// Each duplicate push resets the TTL timer.awaitRateLimitJob.dispatch({userId: 42}).dedup({id: 'rate-42',ttl: '1m',extend: true}).run()

Debounce (replace payload + reset TTL)

// Within the 2s window, the latest payload overwrites the previous pending job.awaitSaveDraftJob.dispatch({content: 'latest draft'}).dedup({id: 'draft-42',ttl: '2s',replace: true,extend: true}).run()

Inspecting the outcome

DispatchResult tells you what happened:

const{ jobId, deduped }=awaitSaveDraftJob.dispatch({content: '...'}).dedup({id: 'draft-42',ttl: '2s',replace: true}).run()// deduped: 'added' | 'skipped' | 'replaced' | 'extended'// jobId: the UUID of the job (the existing one when deduped)

How it works

• The dedup ID is automatically prefixed with the job name (SendInvoiceJob::order-123), so different job types can reuse the same key.
• The user-supplied id must be ≤ 400 characters, and the combined <jobName>::<id> key must be ≤ 510 characters (constrained by the Knex storage column). Both limits are validated at .dedup() time.
• ttl accepts a Duration ('5s', '1m') or milliseconds, and must be positive when provided. Use 0 or omit ttl if you want no expiry — ttl: 0 is rejected to avoid an ambiguous "expired immediately vs no-expiry" interpretation across engines.
• extend and replacerequirettl — calling them without ttl throws.
• replace only applies to jobs in pending or delayed state. Jobs that are active (executing) or retained in history (completed/failed with retention) are left alone; the dispatch returns { deduped: 'skipped' }.
• replace swaps the payload only — priority, queue, delay, groupId, and stored dedup options of the existing job are retained. To change those, use a different dedup id or wait for the TTL to expire.
• extend resets the TTL clock but never changes the window length. The window length is fixed to the ttl from the first dispatch that created the dedup slot. Later dispatches that pass a different ttl only reset the clock; their ttl value is ignored. To resize the window, let the slot expire and start over with a new dispatch.
• extend works in all states — even when the existing job is active (executing) or retained in history. Unlike replace (which is no-op on non-replaceable states), extend always refreshes the dedup TTL window. Use this when you want the dedup slot to keep blocking new dispatches for the lifetime of a long-running job.
• extend requires the first dispatch to have set a ttl. If the slot was created without a ttl, later extend dispatches have no window to refresh and return { deduped: 'skipped' } instead of 'extended'.
• retryJob does not touch the dedup entry — a retried job continues to occupy the dedup slot. TTL runs on wall-clock time, so long-running retries may outlive the TTL window. Use a generous TTL or no TTL if retries must stay deduped.
• Atomic and race-free:
  • Redis: a single Lua script per dispatch performs the dedup-key lookup, state check (pending/delayed ZSCORE), payload swap, and TTL refresh atomically.
  • Knex/Kysely: transactional SELECT ... FOR UPDATE + insert/update inside a transaction. A savepoint catches unique-constraint violations under concurrent inserts and returns { deduped: 'skipped' } pointing at the winner.
  • SyncAdapter: executes inline, no dedup support.

Caveats

• Without .dedup(), jobs use auto-generated UUIDs and are never deduplicated.
• The Sync adapter ignores .dedup() entirely — every dispatch executes inline and deduped is always undefined on the result. Use Redis, Knex, or Kysely if you need real deduplication.
• .dedup() is only available on single dispatch. dispatchMany / pushManyOn reject jobs with a dedup field.
• Scheduled jobs (.schedule()) do not support dedup — each cron/interval fire is an independent dispatch.
• With no ttl, dedup persists until the job is removed (completed/failed without retention). When retention keeps the record, re-dispatch stays blocked until the record is pruned.
• With ttl, dedup expires after the window — a new job (new UUID) is created. The old job still runs.
• Knex/Kysely MySQL concurrent race: MySQL does not support partial unique indexes, so two pushOn calls with the same dedup id firing at the exact same instant can both succeed. Serialize at the app layer if strict guarantees are required, or use Postgres / SQLite / Redis (all of which serialize correctly via the partial unique index or Lua atomicity).

Job History & Retention

Keep completed and failed jobs for debugging:

exportdefaultclassImportantJobextendsJob<Payload>{staticoptions: JobOptions={// Keep last 1000 completed jobsremoveOnComplete: {count: 1000},// Keep failed jobs for 7 daysremoveOnFail: {age: '7d'},}}

Retention options

Value	Behavior
true (default)	Remove immediately
false	Keep forever
{ count: n }	Keep last n jobs
{ age: '7d' }	Keep for duration
{ count: 100, age: '1d' }	Both limits apply

Query job history:

constjob=awaitadapter.getJob('job-id','queue-name')console.log(job.status)// 'completed' | 'failed'console.log(job.finishedAt)// timestampconsole.log(job.error)// error message (if failed)

Adapters

Redis (recommended for production)

import{redis}from'@boringnode/queue/drivers/redis_adapter'// With optionsconstadapter=redis({host: 'localhost',port: 6379})// With existing ioredis instanceimport{Redis}from'ioredis'constconnection=newRedis({host: 'localhost'})constadapter=redis(connection)

Knex (PostgreSQL, MySQL, SQLite)

import{knex}from'@boringnode/queue/drivers/knex_adapter'constadapter=knex({client: 'pg',connection: {host: 'localhost',database: 'myapp'},})

More Knex examples

// With existing Knex instanceimportKnexfrom'knex'constconnection=Knex({client: 'pg',connection: '...'})constadapter=knex(connection)// Custom table nameconstadapter=knex(config,'custom_jobs_table')

Database setup with QueueSchemaService

The Knex adapter requires tables to be created before use. Use QueueSchemaService to create them:

import{QueueSchemaService}from'@boringnode/queue'importKnexfrom'knex'constconnection=Knex({client: 'pg',connection: '...'})constschemaService=newQueueSchemaService(connection)// Create tables with default namesawaitschemaService.createJobsTable()awaitschemaService.createSchedulesTable()// Or extend with custom columnsawaitschemaService.createJobsTable('queue_jobs',(table)=>{table.string('tenant_id',255).nullable()})

AdonisJS migration example:

import{BaseSchema}from'@adonisjs/lucid/schema'import{QueueSchemaService}from'@boringnode/queue'exportdefaultclassextendsBaseSchema{asyncup(){constschemaService=newQueueSchemaService(this.db.connection().getWriteClient())awaitschemaService.createJobsTable()awaitschemaService.createSchedulesTable()}asyncdown(){constschemaService=newQueueSchemaService(this.db.connection().getWriteClient())awaitschemaService.dropSchedulesTable()awaitschemaService.dropJobsTable()}}

Kysely (PostgreSQL, MySQL, SQLite)

Pass the application-owned Kysely instance to the adapter factory.

import{kysely,KyselyQueueSchemaService,typeQueueDatabase,}from'@boringnode/queue/drivers/kysely_adapter'interfaceDatabaseextendsQueueDatabase{orders: OrderTable}constadapter=kysely<Database>(db,{dialect: 'postgres'})

The adapter never destroys the application-owned Kysely instance. Use tableName and schedulesTableName in the options when your migration uses custom names.

Database setup with KyselyQueueSchemaService

constschema=newKyselyQueueSchemaService(db,{dialect: 'postgres'})awaitschema.createJobsTable()awaitschema.createSchedulesTable()// downawaitschema.dropSchedulesTable()awaitschema.dropJobsTable()

Fake (testing + assertions)

import{QueueManager}from'@boringnode/queue'import{redis}from'@boringnode/queue/drivers/redis_adapter'awaitQueueManager.init({default: 'redis',adapters: {redis: redis({host: 'localhost'}),},locations: ['./app/jobs/**/*.ts'],})// The `using` keyword automatically restores the real adapters when// the variable goes out of scope (at the end of the test function).
using fake=QueueManager.fake()awaitSendEmailJob.dispatch({to: 'user@example.com'})fake.assertPushed(SendEmailJob)fake.assertPushed(SendEmailJob,{queue: 'default',payload: (payload)=>payload.to==='user@example.com',})fake.assertPushedCount(1)

You can also call QueueManager.restore() manually if you need more control over when the real adapters are restored.

Sync (for testing)

import{sync}from'@boringnode/queue/drivers/sync_adapter'constadapter=sync()// Jobs execute immediately

Use the sync adapter for tests and lightweight local development only.

• await MyJob.dispatch(payload).run() waits for the job to fully finish.
• Retries are executed inline, not by a background worker.
• If you configure backoff, the adapter will sleep between attempts.
• This means the caller can stay blocked for the full retry duration.

Example: with maxRetries: 3 and an exponential backoff of 1s, 2s, 4s, the request or command that dispatched the job can stay busy for about 7 seconds before the job exhausts its retries and runs failed().

Job Options

exportdefaultclassMyJobextendsJob<Payload>{staticoptions: JobOptions={queue: 'email',// Queue name (default: 'default')priority: 1,// Lower = higher priority (default: 5)maxRetries: 3,// Retry attempts before failingtimeout: '30s',// Max execution timefailOnTimeout: true,// Fail permanently on timeout (default: retry)removeOnComplete: {count: 100},// Keep last 100 completedremoveOnFail: {age: '7d'},// Keep failed for 7 days}}

Delayed Jobs

awaitSendEmailJob.dispatch(payload).in('30s')// 30 secondsawaitSendEmailJob.dispatch(payload).in('5m')// 5 minutesawaitSendEmailJob.dispatch(payload).in('2h')// 2 hoursawaitSendEmailJob.dispatch(payload).in('1d')// 1 day

Retry & Backoff

import{exponentialBackoff}from'@boringnode/queue'exportdefaultclassReliableJobextendsJob<Payload>{staticoptions: JobOptions={maxRetries: 5,retry: {backoff: ()=>exponentialBackoff({baseDelay: '1s',maxDelay: '1m',multiplier: 2,jitter: true,}),},}}

maxRetries can be defined directly on the job options, and retry.backoff controls the delay between attempts.

With the sync adapter, these delays happen inline in the caller via sleep. If a job fails repeatedly, dispatch().run() will take as long as the total backoff duration. Use a worker-backed adapter when you do not want retries to slow down the request/command that dispatched the job.

Available strategies

import{exponentialBackoff,linearBackoff,fixedBackoff}from'@boringnode/queue'// Exponential: 1s, 2s, 4s, 8s...exponentialBackoff({baseDelay: '1s',maxDelay: '1m',multiplier: 2})// Linear: 1s, 2s, 3s, 4s...linearBackoff({baseDelay: '1s',maxDelay: '30s',multiplier: 1})// Fixed: 5s, 5s, 5s...fixedBackoff({baseDelay: '5s',jitter: true})

Job Timeout

exportdefaultclassLongRunningJobextendsJob<Payload>{staticoptions: JobOptions={timeout: '30s',failOnTimeout: false,// Will retry (default)}asyncexecute(): Promise<void>{for(constitemofthis.payload.items){// Check abort signal for graceful timeout handlingif(this.signal?.aborted){thrownewError('Job timed out')}awaitthis.processItem(item)}}}

Job Context

Access execution metadata via this.context:

asyncexecute(): Promise<void>{console.log(this.context.jobId)// Unique job IDconsole.log(this.context.attempt)// 1, 2, 3...console.log(this.context.queue)// Queue nameconsole.log(this.context.priority)// Priority valueconsole.log(this.context.acquiredAt)// When acquiredconsole.log(this.context.stalledCount)// Stall recoveriesconsole.log(this.context.scheduleId)// Originating Schedule, when applicable}

Scheduled Jobs

Run jobs on a recurring basis:

// Every 10 secondsawaitMetricsJob.schedule({endpoint: '/health'}).every('10s')// Cron scheduleawaitCleanupJob.schedule({days: 30}).id('daily-cleanup').with('redis').cron('0 0 * * *')// Midnight daily.timezone('Europe/Paris')

Schedule management

import{Schedule}from'@boringnode/queue'// Find and manageconstschedule=awaitSchedule.find('daily-cleanup')awaitschedule.pause()awaitschedule.resume()awaitschedule.trigger()// Run nowawaitschedule.delete()// List schedulesconstall=awaitSchedule.list()constactive=awaitSchedule.list({status: 'active'})// Access schedules stored on a non-default AdapterconstredisSchedule=awaitSchedule.find('daily-cleanup',{adapter: 'redis'})constredisSchedules=awaitSchedule.list({},{adapter: 'redis'})

Schedule options:

Method	Description
.id(string)	Unique identifier
.every(duration)	Fixed interval ('5s', '1m', '1h')
.cron(expression)	Cron schedule
.timezone(tz)	Timezone (default: 'UTC')
.from(date)	Start boundary
.to(date)	End boundary
.limit(n)	Maximum runs
.with(adapter)	Adapter that owns the Schedule

A Schedule and every Job it dispatches stay on the same Adapter. Start a Worker for each Adapter that owns Schedules.

Dependency Injection

Integrate with IoC containers:

awaitQueueManager.init({// ...jobFactory: async(JobClass)=>{returnapp.container.make(JobClass)},})

Example with injected services

exportdefaultclassSendEmailJobextendsJob<SendEmailPayload>{constructor(privatemailer: MailerService,privatelogger: Logger){super()}asyncexecute(): Promise<void>{this.logger.info(`Sending email to ${this.payload.to}`)awaitthis.mailer.send(this.payload)}}

Worker Configuration

constconfig={worker: {adapter: 'redis',// Registered Adapter listened to by this Workerconcurrency: 5,// Parallel jobsidleDelay: '2s',// Poll interval when idletimeout: '1m',// Default job timeoutstalledThreshold: '30s',// When to consider job stalledstalledInterval: '30s',// How often to checkmaxStalledCount: 1,// Max recoveries before failinggracefulShutdown: true,// Wait for jobs on SIGTERM},}

Logging

import{pino}from'pino'awaitQueueManager.init({// ...logger: pino(),})

OpenTelemetry Instrumentation (experimental)

Warning

The OpenTelemetry instrumentation is experimental and its API may change in future releases.

@boringnode/queue ships with built-in OpenTelemetry instrumentation that creates PRODUCER spans for job dispatch and CONSUMER spans for job execution, following OTel messaging semantic conventions.

Quick Setup

import{QueueInstrumentation}from'@boringnode/queue/otel'import*asboringqueuefrom'@boringnode/queue'constinstrumentation=newQueueInstrumentation({messagingSystem: 'boringqueue',// defaultexecutionSpanLinkMode: 'link',// or 'parent'})instrumentation.enable()instrumentation.manuallyRegister(boringqueue)

The instrumentation patches QueueManager.init() to automatically inject its wrappers — no config changes needed in your queue setup.

Span Attributes

The instrumentation uses standard OTel messaging semantic conventions where they map cleanly, plus a few queue-specific custom attributes.

Attribute	Kind	Description
messaging.system	Semconv	'boringqueue' (configurable)
messaging.operation.name	Semconv	'publish' or 'process'
messaging.destination.name	Semconv	Queue name
messaging.message.id	Semconv	Job ID for single-message spans
messaging.batch.message_count	Semconv	Number of jobs in a batch dispatch
messaging.message.retry.count	Custom	Retry count (0-based) for a job attempt
messaging.job.name	Custom	Job class name (e.g. SendEmailJob)
messaging.job.status	Custom	'completed', 'failed', or 'retrying'
messaging.job.group_id	Custom	Queue-specific group identifier
messaging.job.priority	Custom	Queue-specific job priority
messaging.job.delay_ms	Custom	Delay before the job becomes available
messaging.job.queue_time_ms	Custom	Time spent waiting in queue before processing

Trace Context Propagation

The instrumentation automatically propagates trace context from dispatch to execution:

• Link mode (default): Each job execution is an independent trace, linked to the dispatch span
• Parent mode: Job execution is a child of the dispatch span (same trace)

Child spans created inside execute() (DB queries, HTTP calls, etc.) are automatically parented to the job consumer span.

diagnostics_channel

Raw telemetry events are available via diagnostics_channel for custom subscribers:

import{tracingChannels}from'@boringnode/queue'const{ executeChannel }=tracingChannelsexecuteChannel.subscribe({start(){},end(){},asyncStart(){},asyncEnd(message){console.log(`Job ${message.job.name}${message.status} in ${message.duration}ms`)},error(){},})

Benchmarks

Performance comparison with BullMQ (5ms simulated work per job):

Jobs	Concurrency	@boringnode/queue	BullMQ	Diff
1000	5	1096ms	1116ms	1.8% faster
1000	10	565ms	579ms	2.4% faster
100K	10	56.2s	57.5s	2.1% faster
100K	20	29.1s	29.6s	1.7% faster

npm run benchmark -- --realistic