·QueueHub Team·20 min read

BullMQ Diagnostic Toolbox: Health Checks, Redis Introspection, and Automated Recovery

BullMQ health checksBullMQ diagnosticsBullMQ Redis introspectionBullMQ worker monitoringBullMQ automated recoveryBullMQ liveness probesBullMQ production monitoring

When a BullMQ queue goes silent in production, the first instinct is to search for error logs — but the real insight often lives at a different layer. Redis itself holds a complete picture of every queue's state, every job's position in its lifecycle, and every worker's heartbeat. The challenge is that most teams only interact with BullMQ through the SDK's high-level API, never learning to read the raw data structures underneath.

This post takes a different approach to troubleshooting. Instead of explaining why jobs stall or connections drop, we build a programmable diagnostic system — a toolkit you can drop into any project to inspect queues at the Redis level, surface health metrics via HTTP endpoints, detect and recover stalled jobs automatically, and wire up container-level health probes that actually verify your workers are processing work.

Every section includes complete, copy-paste-ready TypeScript code. By the end you'll have six production-grade tools you can deploy today.

Section 1: Redis Keys Under the Hood — Introspecting BullMQ Data Structures

BullMQ maps queue states to specific Redis data types. Knowing this mapping lets you debug queue issues directly from redis-cli without waiting for a dashboard to load. More importantly, it unlocks programmatic introspection via ioredis that works even when the BullMQ SDK itself is unavailable — for example, from a separate monitoring service that only has Redis credentials.

1.1 Mapping Redis Key Patterns to Queue State

Every BullMQ queue named email-notifications with the default prefix bull creates the following Redis keys:

Key Type Purpose
bull:email-notifications:wait LIST Jobs waiting for a worker
bull:email-notifications:active LIST Jobs currently being processed
bull:email-notifications:delayed ZSET Jobs waiting for a delay timeout (score = epoch ms)
bull:email-notifications:prioritized ZSET Jobs ordered by priority (score = priority value)
bull:email-notifications:completed ZSET Completed jobs (score = completion timestamp)
bull:email-notifications:failed ZSET Failed jobs (score = failure timestamp)
bull:email-notifications:paused ZSET Jobs that were moved to paused list
bull:email-notifications:waiting-children ZSET Jobs in flow producer waiting for children
bull:email-notifications:stalled SET Job IDs currently flagged as stalled
bull:email-notifications:meta HASH Queue metadata (version, concurrency, etc.)
bull:email-notifications:id STRING Monotonically increasing job ID counter
bull:email-notifications:events STREAM BullMQ event stream

The individual job data itself lives at bull:email-notifications:{jobId} as a HASH with fields like data, opts, name, timestamp, processedOn, finishedOn, returnvalue, and attemptsMade.

1.2 Building a Redis-Side Queue Inspector with SCAN and TYPE

The most useful raw diagnostic tool is a Redis-side inspector that walks an entire queue's keyspace and reports live state. This snippet uses ioredis directly — not the BullMQ SDK — so it works from any process that has Redis credentials:

import Redis from 'ioredis';

interface QueueKeyReport {
  key: string;
  type: string;
  cardinality: number;
  ttlSeconds: number;
}

interface QueueReport {
  queueName: string;
  prefix: string;
  keys: QueueKeyReport[];
  summary: Record<string, number>;
}

async function inspectQueueKeys(
  connection: Redis,
  queueName: string,
  prefix: string = 'bull',
): Promise<QueueReport> {
  const pattern = `${prefix}:${queueName}:*`;
  const reports: QueueKeyReport[] = [];
  const summary: Record<string, number> = {};
  let cursor = '0';

  do {
    const [nextCursor, keys] = await connection.scan(
      cursor,
      'MATCH',
      pattern,
      'COUNT',
      100,
    );
    cursor = nextCursor;

    for (const key of keys) {
      const type = await connection.type(key);
      const ttl = await connection.ttl(key);
      let cardinality = 0;

      switch (type) {
        case 'list':
          cardinality = await connection.llen(key);
          break;
        case 'zset':
          cardinality = await connection.zcard(key);
          break;
        case 'set':
          cardinality = await connection.scard(key);
          break;
        case 'hash':
          cardinality = await connection.hlen(key);
          break;
        case 'stream':
          cardinality = await connection.xlen(key);
          break;
        case 'string':
          cardinality = 1;
          break;
      }

      const shortKey = key.replace(`${prefix}:${queueName}:`, '');
      reports.push({ key, type, cardinality, ttlSeconds: ttl });
      summary[shortKey] = cardinality;
    }
  } while (cursor !== '0');

  return { queueName, prefix, keys: reports, summary };
}

// Usage
async function main() {
  const redis = new Redis({ host: 'localhost', port: 6379 });
  const report = await inspectQueueKeys(redis, 'email-notifications');
  console.log('Queue state summary:', report.summary);

  // Flag excessive failed jobs
  if ((report.summary.failed ?? 0) > 100) {
    console.warn(
      `Failed job set has ${report.summary.failed} entries — check for systemic failure`,
    );
  }

  await redis.quit();
}

main().catch(console.error);

This function is invaluable during incident response because it works even when BullMQ's own SDK methods are timing out due to Redis congestion — raw ioredis commands have a much lower overhead.

1.3 Creating a Comprehensive Queue State Report

For a higher-level view that incorporates both Redis introspection and BullMQ's SDK, combine both approaches:

import { Queue } from 'bullmq';
import Redis from 'ioredis';

interface ComprehensiveReport {
  queue: string;
  jobCounts: Record<string, number>;
  stalledCount: number;
  oldestWaitingAgeMs: number;
  redisMemoryUsage: string;
}

async function generateQueueReport(
  queue: Queue,
  connection: Redis,
): Promise<ComprehensiveReport> {
  const queueName = queue.name;

  const [jobCounts, waitingJobs] = await Promise.all([
    queue.getJobCounts('wait', 'active', 'completed', 'failed', 'delayed'),
    queue.getWaiting(0, 1),
  ]);

  const stalledKey = `bull:${queueName}:stalled`;
  const stalledCount = await connection.scard(stalledKey);

  let oldestWaitingAgeMs = 0;
  if ((jobCounts.wait ?? 0) > 0 && waitingJobs[0]) {
    oldestWaitingAgeMs = Date.now() - waitingJobs[0].timestamp;
  }

  const info = await connection.info('memory');
  const match = info.match(/used_memory_human:(\S+)/);
  const redisMemoryUsage = match ? match[1] : 'unknown';

  return {
    queue: queueName,
    jobCounts: {
      wait: jobCounts.wait ?? 0,
      active: jobCounts.active ?? 0,
      completed: jobCounts.completed ?? 0,
      failed: jobCounts.failed ?? 0,
      delayed: jobCounts.delayed ?? 0,
    },
    stalledCount,
    oldestWaitingAgeMs,
    redisMemoryUsage,
  };
}

Section 2: Building a Programmatic Health Check for BullMQ

A health check endpoint that returns { status: 'ok' } is nearly useless for a queue-based system. What you need is an endpoint that tells you whether workers are actually consuming jobs, whether stalls are accumulating, and whether backlog duration exceeds your SLO.

2.1 What to Check: Workers, Stalls, Throughput, and Lag

The most actionable signals for a BullMQ health check are:

  • Worker presence — At least one worker is registered on each critical queue
  • Stall accumulation — Jobs entering the stalled state faster than they are recovered
  • Wait duration — The oldest waiting job's age relative to your SLO
  • Throughput — Completed job rate over the last 60 seconds
  • Redis connectivity — The underlying Redis connection is responsive
  • Event loop lag — The health check itself is not blocking

2.2 Express API Route for Queue Health

import express, { Request, Response } from 'express';
import { Queue } from 'bullmq';
import Redis from 'ioredis';

interface HealthCheckConfig {
  queues: { name: string }[];
  redis: Redis;
  staleJobThresholdMs: number;
  minimumThroughputPerMinute: number;
}

interface QueueHealth {
  name: string;
  healthy: boolean;
  issues: string[];
}

interface HealthReport {
  status: 'healthy' | 'degraded' | 'unhealthy';
  timestamp: string;
  redisLatencyMs: number;
  queues: QueueHealth[];
  eventLoopLagMs: number;
}

function createBullMQHealthRouter(config: HealthCheckConfig): express.Router {
  const router = express.Router();

  router.get('/health/bullmq', async (_req: Request, res: Response) => {
    const start = performance.now();

    // Measure event loop lag
    const eventLoopStart = Date.now();
    await new Promise<void>((resolve) => setImmediate(() => resolve()));
    const eventLoopLagMs = Date.now() - eventLoopStart;

    // Measure Redis latency
    const redisPingStart = Date.now();
    await config.redis.ping();
    const redisLatencyMs = Date.now() - redisPingStart;

    const queueHealthResults: QueueHealth[] = [];
    let overallStatus: 'healthy' | 'degraded' | 'unhealthy' = 'healthy';

    for (const qConfig of config.queues) {
      const queue = new Queue(qConfig.name, { connection: config.redis });
      const issues: string[] = [];
      let healthy = true;

      try {
        const [jobCounts, activeJobs, waitingJobs] = await Promise.all([
          queue.getJobCounts('wait', 'active', 'failed', 'completed', 'delayed'),
          queue.getActive(0, 100),
          queue.getWaiting(0, 1),
        ]);

        // 1. Worker presence check
        const activeWorkers = activeJobs.length;
        if (activeWorkers === 0 && (jobCounts.wait ?? 0) > 0) {
          issues.push('No active workers but jobs are waiting');
          healthy = false;
        }

        // 2. Stalled job check
        const stalledSetKey = `bull:${qConfig.name}:stalled`;
        const stalledCount = await config.redis.scard(stalledSetKey);
        if (stalledCount > 0) {
          issues.push(`${stalledCount} jobs in stalled set`);
          healthy = false;
        }

        // 3. Oldest waiting job age
        let oldestWaitingAgeMs = 0;
        if ((jobCounts.wait ?? 0) > 0 && waitingJobs[0]) {
          oldestWaitingAgeMs = Date.now() - waitingJobs[0].timestamp;
          if (oldestWaitingAgeMs > config.staleJobThresholdMs) {
            issues.push(
              `Oldest waiting job is ${(oldestWaitingAgeMs / 1000).toFixed(0)}s old`,
            );
            healthy = false;
          }
        }

        // 4. Throughput check
        const completedSetKey = `bull:${qConfig.name}:completed`;
        const now = Date.now();
        const oneMinuteAgo = now - 60_000;
        const recentCompletions = await config.redis.zcount(
          completedSetKey,
          oneMinuteAgo,
          now,
        );
        if (
          recentCompletions < config.minimumThroughputPerMinute &&
          (jobCounts.wait ?? 0) > 0
        ) {
          issues.push(
            `Low throughput: ${recentCompletions} jobs/min`,
          );
          healthy = false;
        }

        // 5. Failed job accumulation warning
        if ((jobCounts.failed ?? 0) > 1000) {
          issues.push(`${jobCounts.failed} failed jobs — possible systemic failure`);
        }
      } catch (err) {
        const message = err instanceof Error ? err.message : 'Unknown error';
        issues.push(`BullMQ SDK error: ${message}`);
        healthy = false;
      } finally {
        await queue.close();
      }

      if (!healthy && overallStatus === 'healthy') {
        overallStatus = 'degraded';
      }

      queueHealthResults.push({
        name: qConfig.name,
        healthy,
        issues,
      });
    }

    const totalDuration = performance.now() - start;
    if (totalDuration > 5000) {
      overallStatus = 'unhealthy';
    }

    const healthReport: HealthReport = {
      status: overallStatus,
      timestamp: new Date().toISOString(),
      redisLatencyMs,
      queues: queueHealthResults,
      eventLoopLagMs,
    };

    const httpStatus = overallStatus === 'healthy' ? 200 : overallStatus === 'degraded' ? 200 : 503;
    res.status(httpStatus).json(healthReport);
  });

  return router;
}

// Setup
const redisConnection = new Redis({
  host: process.env.REDIS_HOST || 'localhost',
  port: Number(process.env.REDIS_PORT) || 6379,
  maxRetriesPerRequest: null,
});

const config: HealthCheckConfig = {
  queues: [{ name: 'email-notifications' }, { name: 'report-generation' }],
  redis: redisConnection,
  staleJobThresholdMs: 300_000,
  minimumThroughputPerMinute: 1,
};

const app = express();
app.use(createBullMQHealthRouter(config));
app.listen(8080, () => console.log('BullMQ health endpoint at /health/bullmq'));

This endpoint returns HTTP 200 for healthy and degraded states (so load balancers don't drop degraded pods), and HTTP 503 when truly unhealthy. The degraded status gives you operational visibility without causing unnecessary pod restarts.

Section 3: Automated Stalled Job Detection and Recovery

BullMQ's built-in stall detection works at the individual worker level — it re-queues jobs whose lock has expired. But when the entire worker fleet goes down, stalled jobs stay stranded in the active state until a new worker comes online. Programmatic recovery from a separate monitoring process gives you more control.

3.1 Finding Stalled Candidates by Age

import { Queue, Job } from 'bullmq';
import Redis from 'ioredis';

interface RecoveryConfig {
  connection: Redis;
  queueName: string;
  lockDurationMs: number;
  maxRecoveriesPerRun: number;
  onRecovered?: (jobs: Job[]) => Promise<void>;
  onError?: (error: Error, jobId: string) => void;
}

async function findStalledActiveJobs(
  queue: Queue,
  lockDurationMs: number,
): Promise<Job[]> {
  const activeJobs = await queue.getActive(0, 1000);
  const now = Date.now();
  const stalled: Job[] = [];

  for (const job of activeJobs) {
    const activeSince = job.processedOn ?? job.timestamp ?? now;
    const activeDuration = now - activeSince;

    if (activeDuration > lockDurationMs * 1.5) {
      stalled.push(job);
    }
  }

  return stalled;
}

3.2 Recovering Stalled Jobs Programmatically

async function recoverStalledJobs(
  config: RecoveryConfig,
): Promise<{ recovered: number; failed: string[] }> {
  const queue = new Queue(config.queueName, { connection: config.connection });
  const failedIds: string[] = [];
  let recoveredCount = 0;

  try {
    const stalledCandidates = await findStalledActiveJobs(queue, config.lockDurationMs);

    if (stalledCandidates.length === 0) {
      return { recovered: 0, failed: [] };
    }

    const candidates = stalledCandidates.slice(0, config.maxRecoveriesPerRun);
    const stalledKey = `bull:${config.queueName}:stalled`;
    const activeKey = `bull:${config.queueName}:active`;

    for (const job of candidates) {
      try {
        // Add to stalled set — triggers BullMQ's internal recovery
        await config.connection.sadd(stalledKey, job.id!);
        // Remove from active set
        await config.connection.lrem(activeKey, 1, job.id!;

        recoveredCount++;

        if (config.onRecovered) {
          await config.onRecovered([job]);
        }
      } catch (err) {
        const message = err instanceof Error ? err.message : 'Unknown error';
        failedIds.push(job.id!);
        if (config.onError) {
          config.onError(err instanceof Error ? err : new Error(message), job.id!);
        }
      }
    }
  } finally {
    await queue.close();
  }

  return { recovered: recoveredCount, failed: failedIds };
}

// Scheduled recovery
async function startRecoveryScheduler(
  config: RecoveryConfig,
  intervalMs: number = 60_000,
): Promise<NodeJS.Timeout> {
  const run = async () => {
    try {
      const result = await recoverStalledJobs(config);
      if (result.recovered > 0 || result.failed.length > 0) {
        console.log(
          `[Recovery] Recovered ${result.recovered} stalled jobs. Failed: ${result.failed.length}`,
        );
      }
    } catch (err) {
      console.error('[Recovery] Scheduler error:', err);
    }
  };

  await run();
  return setInterval(run, intervalMs);
}

// Usage
const redis = new Redis({ host: 'localhost', port: 6379 });

startRecoveryScheduler(
  {
    connection: redis,
    queueName: 'video-transcoding',
    lockDurationMs: 30_000,
    maxRecoveriesPerRun: 50,
    onRecovered: async (jobs) => {
      console.log(`Recovered ${jobs.length} stalled jobs: ${jobs.map((j) => j.id).join(', ')}`);
    },
  },
  30_000,
);

3.3 Handling Edge Cases in Automated Recovery

Three edge cases require attention:

False positives — A genuinely slow job (e.g., a 45-second video encode with lockDuration: 30000) will be flagged. Always set lockDuration per queue based on the maximum expected processing time.

Recovery loops — A job that immediately re-stalls after recovery indicates a bug in the worker itself. Track recovery attempts and stop after N tries:

async function safeRecoverJob(
  connection: Redis,
  queueName: string,
  jobId: string,
): Promise<boolean> {
  const lockKey = `bull:${queueName}:${jobId}:lock`;
  const lockTtl = await connection.ttl(lockKey);

  // Only recover if the lock has expired
  if (lockTtl > 1) {
    return false; // Worker is still holding the lock
  }

  await connection.sadd(`bull:${queueName}:stalled`, jobId);

  const trackingKey = `recovery:tracking:${queueName}:${jobId}`;
  const attempts = await connection.incr(trackingKey);
  await connection.expire(trackingKey, 3600);

  if (attempts > 3) {
    console.warn(`Job ${jobId} recovered ${attempts} times — moving to failed`);
    await connection.zadd(`bull:${queueName}:failed`, Date.now(), jobId);
    await connection.srem(`bull:${queueName}:stalled`, jobId);
    return false;
  }

  return true;
}

Race conditions — Between identifying a stalled job and recovering it, a legitimate worker might renew the lock. Check that the job's lock key TTL is at most 1 second before moving it.

Section 4: Worker Heartbeat and Self-Healing

When a worker dies silently — because of an unhandled exception, OOM killer, or a stalled event loop — jobs pile up in the waiting state. BullMQ's default stall detection can take up to 30 seconds to react. A heartbeat-based monitoring system can detect failures in seconds.

4.1 The Heartbeat Pattern

Each worker writes a timestamp to a Redis hash keyed by worker ID at a regular interval:

import { Worker, Job } from 'bullmq';
import Redis from 'ioredis';
import os from 'os';

interface HeartbeatPayload {
  workerId: string;
  hostname: string;
  queueName: string;
  pid: number;
  jobsProcessed: number;
  currentJobs: string[];
  eventLoopLag: number;
  memoryMb: number;
  timestamp: number;
}

class HeartbeatWorker {
  private readonly worker: Worker;
  private readonly connection: Redis;
  private readonly heartbeatKey: string;
  private readonly intervalMs: number;
  private heartbeatTimer: NodeJS.Timeout | null = null;
  private jobsProcessed = 0;
  private currentJobs: Set<string> = new Set();
  private readonly workerId: string;

  constructor(
    queueName: string,
    processor: (job: Job) => Promise<unknown>,
    connection: Redis,
    intervalMs: number = 5000,
    workerId: string = `${os.hostname()}-${process.pid}`,
  ) {
    this.worker = new Worker(queueName, processor, {
      connection,
      maxRetriesPerRequest: null,
      name: workerId,
    });
    this.connection = connection;
    this.heartbeatKey = `bull:heartbeats:${queueName}`;
    this.intervalMs = intervalMs;
    this.workerId = workerId;

    this.worker.on('completed', () => { this.jobsProcessed++; });
    this.worker.on('active', (job) => { this.currentJobs.add(job.id!); });
    this.worker.on('failed', (job) => { this.currentJobs.delete(job.id!); });
  }

  private async emitHeartbeat(): Promise<void> {
    const payload: HeartbeatPayload = {
      workerId: this.workerId,
      hostname: os.hostname(),
      queueName: this.worker.name,
      pid: process.pid,
      jobsProcessed: this.jobsProcessed,
      currentJobs: Array.from(this.currentJobs),
      eventLoopLag: await this.measureEventLoopLag(),
      memoryMb: Math.round(process.memoryUsage().heapUsed / 1024 / 1024),
      timestamp: Date.now(),
    };

    await this.connection.hset(this.heartbeatKey, this.workerId, JSON.stringify(payload));
    await this.connection.pexpire(this.heartbeatKey, this.intervalMs * 3);
  }

  private measureEventLoopLag(): Promise<number> {
    return new Promise((resolve) => {
      const start = Date.now();
      setImmediate(() => resolve(Date.now() - start));
    });
  }

  async start(): Promise<void> {
    await this.worker.waitUntilReady();
    console.log(`[HeartbeatWorker] ${this.workerId} starting heartbeats`);
    await this.emitHeartbeat();
    this.heartbeatTimer = setInterval(
      () => this.emitHeartbeat().catch((err) => console.error('[Heartbeat] Error:', err)),
      this.intervalMs,
    );
  }

  async close(): Promise<void> {
    if (this.heartbeatTimer) {
      clearInterval(this.heartbeatTimer);
      this.heartbeatTimer = null;
    }
    await this.connection.hdel(this.heartbeatKey, this.workerId);
    await this.worker.close();
  }
}

4.2 Supervisor Process for Dead Worker Detection

The supervisor runs as a separate process and reads all heartbeats to determine which workers have gone silent:

interface WorkerStatus {
  workerId: string;
  queueName: string;
  hostname: string;
  pid: number;
  jobsProcessed: number;
  lastHeartbeatMs: number;
  ageMs: number;
  alive: boolean;
  memoryMb: number;
}

async function getWorkerStatuses(
  connection: Redis,
  queueName: string,
  staleThresholdMs: number = 15_000,
): Promise<WorkerStatus[]> {
  const heartbeatKey = `bull:heartbeats:${queueName}`;
  const heartbeats = await connection.hgetall(heartbeatKey);
  const now = Date.now();
  const statuses: WorkerStatus[] = [];

  for (const [workerId, payloadStr] of Object.entries(heartbeats)) {
    let payload: HeartbeatPayload;
    try {
      payload = JSON.parse(payloadStr) as HeartbeatPayload;
    } catch {
      continue;
    }

    statuses.push({
      workerId,
      queueName: payload.queueName,
      hostname: payload.hostname,
      pid: payload.pid,
      jobsProcessed: payload.jobsProcessed,
      lastHeartbeatMs: payload.timestamp,
      ageMs: now - payload.timestamp,
      alive: (now - payload.timestamp) < staleThresholdMs,
      memoryMb: payload.memoryMb,
    });
  }

  return statuses;
}

async function supervisorCheck(connection: Redis, queueName: string): Promise<void> {
  const statuses = await getWorkerStatuses(connection, queueName);
  const deadWorkers = statuses.filter((s) => !s.alive);

  if (deadWorkers.length > 0) {
    for (const w of deadWorkers) {
      console.warn(
        `[Supervisor] Dead worker: ${w.workerId} (last heartbeat ${(w.ageMs / 1000).toFixed(1)}s ago)`,
      );
    }
    // Clean up stale entries
    const multi = connection.multi();
    for (const w of deadWorkers) {
      multi.hdel(`bull:heartbeats:${queueName}`, w.workerId);
    }
    await multi.exec();
  }
}

// Run every 10 seconds
const supervisorRedis = new Redis({ host: 'localhost', port: 6379 });
setInterval(
  () => supervisorCheck(supervisorRedis, 'video-transcoding').catch(console.error),
  10_000,
);

When the supervisor detects sustained absence, it can trigger process exit (which systemd or Kubernetes translates into a restart):

async function supervisorWithAutoRestart(
  connection: Redis,
  queueName: string,
  maxMissedHeartbeats: number = 3,
): Promise<void> {
  const statuses = await getWorkerStatuses(connection, queueName);
  const trackingKey = `bull:missed-checkins:${queueName}`;

  for (const worker of statuses) {
    if (worker.alive) {
      await connection.hdel(trackingKey, worker.workerId);
      continue;
    }

    const missCount = await connection.hincrby(trackingKey, worker.workerId, 1);

    if (missCount >= maxMissedHeartbeats) {
      console.error(
        `[Supervisor] Worker ${worker.workerId} missed ${missCount} heartbeats. Restarting.`,
      );
      process.exit(1);
    }
  }
}

Section 5: Production Health Checks for Containerized Workers

Container orchestration platforms need health probes that tell them not just "is the process running?" but "is the worker actually processing jobs?" A worker connected to Redis but stalled on a synchronous CPU-bound operation will pass a basic TCP liveness probe but fail to make progress on its queue.

5.1 Docker HEALTHCHECK for BullMQ Workers

Docker's HEALTHCHECK instruction runs a command inside the container at intervals. For BullMQ workers, this command should verify that the worker has processed at least one job recently:

FROM node:20-alpine

WORKDIR /app
COPY package*.json ./
RUN npm ci --omit=dev
COPY . .

HEALTHCHECK --interval=15s --timeout=5s --start-period=30s --retries=3 \
  CMD ["docker-healthcheck.sh"]

CMD ["node", "dist/worker.js"]

The health check script checks either a timestamp file (written by the worker) or the /health/bullmq endpoint:

#!/bin/sh
# docker-healthcheck.sh
set -e

REDIS_CLI="redis-cli -u ${REDIS_URL:-redis://localhost:6379}"
QUEUE_NAME="${QUEUE_NAME:-default}"
LAST_JOB_FILE="/tmp/worker-last-job-timestamp"

# Option A: Check via timestamp file (lightweight, no Redis call)
if [ -f "$LAST_JOB_FILE" ]; then
  LAST_JOB_TS=$(cat "$LAST_JOB_FILE")
  NOW_MS=$(date +%s%N | cut -b1-13)
  AGE_MS=$((NOW_MS - LAST_JOB_TS))

  if [ "$AGE_MS" -gt 120000 ]; then
    echo "UNHEALTHY: No job processed in ${AGE_MS}ms"
    exit 1
  fi
  echo "HEALTHY: Last job processed ${AGE_MS}ms ago"
  exit 0
fi

# Option B: Check via BullMQ health endpoint
HEALTH_URL="${HEALTH_URL:-http://localhost:8080/health/bullmq}"
STATUS=$(wget -q -O - "$HEALTH_URL" 2>/dev/null | grep -o '"status":"[^"]*"' | cut -d'"' -f4)

if [ "$STATUS" = "healthy" ] || [ "$STATUS" = "degraded" ]; then
  echo "HEALTHY: Queue status is $STATUS"
  exit 0
fi

echo "UNHEALTHY: Queue status is $STATUS"
exit 1

To use Option A, have your worker write the timestamp:

import { writeFileSync } from 'fs';

// Inside your job processor, after each job completes
writeFileSync('/tmp/worker-last-job-timestamp', Date.now().toString());

5.2 Kubernetes Liveness and Readiness Probes

Kubernetes differentiates between liveness (is the process alive?) and readiness (can it accept work?). For BullMQ workers, readiness should be more aggressive:

apiVersion: apps/v1
kind: Deployment
metadata:
  name: bullmq-worker
spec:
  replicas: 3
  selector:
    matchLabels:
      app: bullmq-worker
  template:
    metadata:
      labels:
        app: bullmq-worker
    spec:
      containers:
        - name: worker
          image: myregistry/bullmq-worker:latest
          ports:
            - containerPort: 8080
              name: health
          env:
            - name: REDIS_HOST
              value: "redis-service"
            - name: QUEUE_NAME
              value: "email-notifications"
          livenessProbe:
            httpGet:
              path: /health/bullmq
              port: health
            initialDelaySeconds: 30
            periodSeconds: 15
            timeoutSeconds: 5
            failureThreshold: 3
          readinessProbe:
            httpGet:
              path: /health/bullmq
              port: health
              httpHeaders:
                - name: X-Readiness-Check
                  value: "strict"
            initialDelaySeconds: 10
            periodSeconds: 10
            timeoutSeconds: 3
            failureThreshold: 2
          resources:
            requests:
              memory: "256Mi"
              cpu: "250m"
            limits:
              memory: "512Mi"
              cpu: "500m"

To make the readiness probe stricter, add header-aware threshold switching:

// Inside the /health/bullmq handler
router.get('/health/bullmq', async (req: Request, res: Response) => {
  const isReadiness = req.headers['x-readiness-check'] === 'strict';

  // Existing health check logic...
  // For readiness: stricter thresholds
  const staleJobThresholdMs = isReadiness ? 60_000 : 300_000;
  const minimumThroughput = isReadiness ? 5 : 1;

  // Readiness: 503 on any queue issue
  // Liveness: 200 as long as Redis is reachable
  const httpStatus = isReadiness
    ? health.status === 'healthy' ? 200 : 503
    : health.redisLatencyMs < 2000 ? 200 : 503;

  res.status(httpStatus).json(health);
});

5.3 Systemd Service Monitoring for VM Deployments

For VM-based deployments, use systemd's watchdog feature combined with the worker's heartbeat via sd_notify:

// worker-systemd.ts — Run as a systemd Type=notify service
import { Worker } from 'bullmq';
import Redis from 'ioredis';
import dgram from 'dgram';

function systemdNotify(state: string): void {
  const notifySocket = process.env.NOTIFY_SOCKET;
  if (!notifySocket) return;
  try {
    const sock = dgram.createSocket('unix_dgram');
    sock.send(Buffer.from(state), 0, state.length, notifySocket);
    sock.close();
  } catch {
    // systemd notify not available
  }
}

async function runSystemdWorker(): Promise<void> {
  const connection = new Redis({
    host: process.env.REDIS_HOST || 'localhost',
    port: Number(process.env.REDIS_PORT) || 6379,
    maxRetriesPerRequest: null,
  });

  const worker = new Worker(
    process.env.QUEUE_NAME || 'default',
    async (job) => {
      systemdNotify('WATCHDOG=1');
      return await processJob(job);
    },
    { connection },
  );

  systemdNotify('READY=1');

  const shutdown = async () => {
    systemdNotify('STOPPING=1');
    await worker.close();
    await connection.quit();
  };

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

The corresponding systemd unit:

[Unit]
Description=BullMQ Worker — email-notifications
After=redis.service
Requires=redis.service

[Service]
Type=notify
NotifyAccess=all
WatchdogSec=30
ExecStart=/usr/bin/node /opt/app/dist/worker-systemd.js
Restart=on-failure
RestartSec=5
User=node
Group=node
Environment=NODE_ENV=production
Environment=QUEUE_NAME=email-notifications
Environment=REDIS_HOST=127.0.0.1

[Install]
WantedBy=multi-user.target

With WatchdogSec=30, systemd kills and restarts the worker if it doesn't receive WATCHDOG=1 within 30 seconds — even if the Node.js process is still running and accepting signals.

Bringing It All Together with Queue Hub

The tools in this post — Redis key introspection, programmatic health checks, automated stalled job recovery, worker heartbeats, and container health probes — form a complete diagnostic stack. But maintaining them across multiple services and environments is its own operational burden.

This is where Queue Hub fills the gap. Queue Hub provides a centralized view of all the signals these tools generate:

  • Live queue state shows you wait, active, failed, completed counts across all queues in one dashboard — the same data the Redis-side inspector surfaces, but without running a SCAN command.
  • Worker visibility displays which workers are connected, their heartbeat timing, and their current job load — replicating what the supervisor process monitors.
  • Automated stall alerts notify you when jobs exceed configurable age thresholds, equivalent to the programmatic recovery scheduler but with a UI and configurable notification channels.
  • Multi-backend support means you can monitor queues across development, staging, and production Redis instances from a single interface.
  • Agent tunneling for private Redis instances lets you monitor queues that aren't publicly accessible.

By combining the DIY diagnostic tools from this guide with Queue Hub's centralized dashboard, you get both the raw programmatic access for automation and the visual overview for day-to-day operations. The health check endpoint from Section 2 integrates natively with your monitoring stack (Prometheus, Datadog, Grafana), while Queue Hub gives your whole team at-a-glance queue health without each member needing Redis credentials.

Start with the code in this guide to build your foundation. Then evaluate where manual inspection turns into toil — and let Queue Hub handle the rest.

← All Articles
Share on XShare on LinkedIn