Lesson 6 — Metrics, Stalled Jobs & Cleanup

Tied to mission: "monitor queue health" is an explicit success criterion — and the 3am page usually comes from a stalled job nobody noticed or a Redis OOM nobody prevented. Builds on Lesson 2 (stalled basics) and Lesson 1.

Part A — Metrics: throughput over time

BullMQ keeps lightweight time-series of completed and failed jobs, bucketed per one-minute interval. Off by default; you enable it on the worker and read it from the queue.

Enable

import { Worker, MetricsTime } from 'bullmq';

new Worker('Paint', processor, {
  connection,
  metrics: { maxDataPoints: MetricsTime.ONE_WEEK * 2 },  // keep ~2 weeks
});

maxDataPoints = how many 1-minute buckets to retain. MetricsTime.ONE_WEEK is a convenience constant. The storage cost is tiny — a counter per bucket.

Read

const completed = await queue.getMetrics('completed', 0, MetricsTime.ONE_WEEK * 2);
// { meta: {...}, data: { values: number[], processedCount: number }, count, start, end }

const failed = await queue.getMetrics('failed');   // default start=0, end=-1 (all)

Snapshot counts — the monitoring bread-and-butter

const counts = await queue.getJobCounts();
// { waiting, active, completed, failed, delayed,
//   prioritized, 'waiting-children', paused, repeat }

This is what you wire into a /health endpoint or a dashboard: a backlog climbing in waiting while completed flattens = workers are stuck. All available states: completed, failed, delayed, active, wait/waiting, waiting-children, prioritized, paused, repeat.

// list jobs in a state, paginated
const failed = await queue.getJobs(['failed'], 0, 50);
const job = await queue.getJob(id);

Part B — Stalled jobs: when a worker vanishes

While a job is active, the worker renews a lock by heartbeat. If the worker process crashes, gets OOM-killed, or blocks the Node event loop too long, the lock lapses. BullMQ then moves the job back to waiting so another worker can grab it — or, if it's stalled too many times, fails it.

The two knobs (on the Worker)

new Worker('Paint', processor, {
  connection,
  stalledInterval: 30_000,
  maxStalledCount: 1,
});

Listen for it

worker.on('stalled', (jobId) => {
  console.warn(`job ${jobId} stalled and was re-queued`);
});

Alert on this in prod. A stalled job means a worker died mid-flight — if you see a steady stream of stalls, you have CPU-bound work blocking the event loop.

Part C — Cleanup: don't let Redis grow forever

Every completed/failed job is a Redis key. Without bounds, Redis grows until it OOMs — and BullMQ won't save you. There are two layers: automatic (per-job options) and manual (queue methods).

Layer 1 — automatic retention (set this always)

new Queue('email', {
  connection,
  defaultJobOptions: {
    removeOnComplete: 1000,   // keep last 1000 completed
    removeOnFail: 5000,       // keep last 5000 failed (for debugging)
  },
});

Layer 2 — manual cleanup methods

// surgical: remove failed jobs older than 1 hour, max 1000
await queue.clean(3_600_000, 1000, 'failed');

// nuke from orbit (CI / test only)
await queue.obliterate();
await queue.obliterate({ force: true });  // even if jobs are active

Full picture — an observable, self-cleaning worker

import { Queue, Worker, MetricsTime } from 'bullmq';
import IORedis from 'ioredis';
const connection = new IORedis({ maxRetriesPerRequest: null });

const email = new Queue('email', {
  connection,
  defaultJobOptions: {
    removeOnComplete: { age: 86_400, count: 2000 },  // ≤2000, none older than 24h
    removeOnFail:    { age: 604_800, count: 5000 },  // ≤5000, none older than 7d
  },
});

const worker = new Worker('email', processor, {
  connection,
  concurrency: 5,
  stalledInterval: 30_000,
  maxStalledCount: 1,
  metrics: { maxDataPoints: MetricsTime.ONE_WEEK * 2 },
});

worker.on('stalled', (id) => metrics.increment('bullmq.stalled', { queue: 'email' }));

// health endpoint / cron:
//   const c = await email.getJobCounts();
//   const m = await email.getMetrics('completed');

What you just learned

• Metrics: enable on the worker (metrics.maxDataPoints), read with queue.getMetrics('completed'|'failed'); per-minute buckets.
• queue.getJobCounts() is the snapshot — one call, all states incl. waiting-children and repeat.
• Stalled = lapsed lock, not a state. Worker renews via heartbeat; stalledInterval/maxStalledCount govern detection.
• Chronic stalls mean CPU-bound work — move to sandboxed processors, don't just raise the knobs.
• Always set removeOnComplete/removeOnFail (number, age-object, or true).
• Manual: clean() (surgical), drain() (waiting/delayed), obliterate() (everything, no undo).
• Active jobs are locked — remove()/clean() throw; only obliterate({force}) overrides.

🧠 Quiz — 10 questions

📚 Primary sources

📎 Reference

💬 Ask me next

• "Show me a /healthz endpoint backed by getJobCounts."
• "How do I export BullMQ metrics to Prometheus?"
• "When do I reach for a sandboxed processor — and how is it different?"
• "My queue's 'waiting' keeps climbing — how do I diagnose?"

Or say "next" for Lesson 7: production — graceful shutdown, connection pooling, scaling workers.