Worker Patterns & Production

Graceful Shutdown

Workers must finish their current jobs before stopping. An abrupt shutdown (SIGKILL) mid-job leaves your data in an inconsistent state.

import { Worker } from 'bullmq';

const worker = new Worker('jobs', jobHandler, { connection, concurrency: 10 });

// Graceful shutdown handler
async function shutdown(): Promise<void> {
	logger.info('Shutting down worker...');

	// Stop picking up new jobs
	await worker.pause();

	// Wait for in-progress jobs to complete (up to 30s)
	const timeout = setTimeout(async () => {
		logger.warn('Shutdown timeout — forcing close');
		await worker.close(true); // force close
		process.exit(1);
	}, 30_000);

	await worker.close(); // waits for active jobs to finish
	clearTimeout(timeout);

	logger.info('Worker shutdown complete');
	process.exit(0);
}

process.on('SIGTERM', shutdown);
process.on('SIGINT', shutdown);

Container deployment: Set terminationGracePeriodSeconds in Kubernetes to be longer than your longest expected job:

spec:
  containers:
    - name: worker
      lifecycle:
        preStop:
          exec:
            command: ['sleep', '5'] # give SIGTERM time to propagate
  terminationGracePeriodSeconds: 60 # matches your 30s worker timeout + buffer

Priority Queues

Some jobs are more urgent than others. Implement multiple queues with dedicated workers per priority tier, or use BullMQ’s built-in priority:

Multiple queues (explicit control):

const criticalQueue = new Queue('critical', { connection });
const defaultQueue = new Queue('default', { connection });
const bulkQueue = new Queue('bulk', { connection });

// More workers on critical queue
const criticalWorker = new Worker('critical', handler, { connection, concurrency: 20 });
const defaultWorker = new Worker('default', handler, { connection, concurrency: 5 });
const bulkWorker = new Worker('bulk', handler, { connection, concurrency: 2 });

BullMQ priority (single queue, ordered by priority value):

// Lower number = higher priority
await queue.add('send-alert', { userId }, { priority: 1 }); // picked first
await queue.add('send-report', { userId }, { priority: 10 });
await queue.add('sync-data', { userId }, { priority: 100 }); // picked last

BullMQ priority uses sorted sets — workers always pick the lowest-priority-number job next. This works well but can starve lower-priority jobs during sustained high load. For that, use weighted round-robin across multiple queues instead.

Fan-Out Pattern

One job spawns many child jobs. Useful for bulk operations where you want per-item retries and concurrency:

// Parent job: dispatch work to children
async function processOrderBatch(job: Job<{ orderIds: string[] }>): Promise<void> {
	const { orderIds } = job.data;

	// Fan out — one child per order
	await Promise.all(
		orderIds.map((orderId) =>
			childQueue.add(
				'process-order',
				{ orderId },
				{
					attempts: 3,
					backoff: { type: 'exponential', delay: 1000 }
				}
			)
		)
	);

	logger.info({ count: orderIds.length }, 'Dispatched order processing jobs');
}

// Child job: handles one order, retried independently if it fails
async function processOrder(job: Job<{ orderId: string }>): Promise<void> {
	const order = await db.orders.findById(job.data.orderId);
	if (!order) return; // already deleted — skip

	await fulfillOrder(order);
	await sendConfirmationEmail(order);
}

Fan-out gives you:

• Independent retry per item (one bad order doesn’t block others)
• Parallelism (many workers handle children simultaneously)
• Progress visibility (see completed/failed counts per child)

Flow Control: Job Dependencies

BullMQ Flows let you define parent-child job trees with automatic progression:

import { FlowProducer } from 'bullmq';

const flowProducer = new FlowProducer({ connection });

// Parent runs only after all children complete
await flowProducer.add({
	name: 'generate-invoice',
	queueName: 'invoicing',
	data: { invoiceId: 'inv_123' },
	children: [
		{
			name: 'fetch-line-items',
			queueName: 'data',
			data: { invoiceId: 'inv_123' }
		},
		{
			name: 'calculate-tax',
			queueName: 'data',
			data: { invoiceId: 'inv_123' }
		},
		{
			name: 'apply-discounts',
			queueName: 'data',
			data: { invoiceId: 'inv_123' }
		}
	]
});

// Parent handler receives results from all children
const invoicingWorker = new Worker('invoicing', async (job) => {
	const childResults = await job.getChildrenValues();
	// childResults: { 'fetch-line-items': [...], 'calculate-tax': {...}, ... }

	const invoice = buildInvoice(childResults);
	await db.invoices.update(job.data.invoiceId, invoice);
});

Rate Limiting Workers

Prevent hammering external APIs:

import { RateLimiter } from 'limiter';

// 10 requests per second to external API
const rateLimiter = new RateLimiter({ tokensPerInterval: 10, interval: 'second' });

const worker = new Worker(
	'api-sync',
	async (job) => {
		await rateLimiter.removeTokens(1); // blocks until token available
		await externalApi.sync(job.data);
	},
	{ connection, concurrency: 20 }
); // 20 concurrent, but rate-limited to 10/s

BullMQ also supports queue-level rate limiting:

const worker = new Worker('api-sync', handler, {
	connection,
	limiter: {
		max: 10, // max 10 jobs
		duration: 1000 // per 1000ms
	}
});

Worker Health Monitoring

// Emit metrics for each job
worker.on('completed', (job, result) => {
	metrics.histogram('job.duration', Date.now() - job.processedOn!, {
		type: job.name
	});
	metrics.increment('job.completed', { type: job.name });
});

worker.on('failed', (job, err) => {
	metrics.increment('job.failed', { type: job?.name ?? 'unknown' });
	logger.error({ jobId: job?.id, error: err.message, type: job?.name }, 'Job failed');
});

// Stalled job detection (BullMQ auto-detects these)
worker.on('stalled', (jobId) => {
	logger.warn({ jobId }, 'Job stalled — worker may have crashed mid-job');
	metrics.increment('job.stalled');
});

Key production metrics:

• job.duration p50/p95/p99 per job type
• job.completed and job.failed rates
• Queue depth (waiting count) per queue
• Worker active count vs concurrency limit
• Stalled job count

Production Checklist

□ Graceful shutdown on SIGTERM — drain active jobs before exit
□ terminationGracePeriodSeconds >= max job duration + buffer
□ Concurrency tuned per job type (I/O vs CPU bound)
□ Priority queues for time-sensitive vs bulk jobs
□ DLQ configured — failed jobs held for inspection, not silently dropped
□ Job payloads small — store large data in S3/DB, pass ID in job
□ Sensitive data not stored in job payloads (logs and UIs expose them)
□ Metrics exported: duration, throughput, error rate, queue depth
□ Alert on: queue depth spike, sustained failure rate, stalled jobs
□ Worker restarts don't lose jobs (queue is the source of truth)
□ Idempotency tested for all job handlers
□ Job timeouts set (don't let a job hang forever)

Job payload size discipline:

// WRONG — large payload in queue
await queue.add('process-upload', {
	fileContents: Buffer.from(file).toString('base64') // MB of data
});

// RIGHT — store large data separately, pass reference
const s3Key = await s3.upload(file);
await queue.add('process-upload', { s3Key }); // tiny payload

Job payloads live in Redis/Postgres — keep them small. Aim for under 1KB. If you need more, store it in S3 or a DB table and reference it by ID.