BeeQueue Production Patterns and Redis Internals: Beyond the Basics

You've built a BeeQueue worker. Jobs are flowing. Your queue.process() callback fires reliably and your application handles background work gracefully. But then production happens:

A job mysteriously vanishes — it was dispatched but never processed. Your Redis memory climbs and you're not sure which keys are safe to delete. A SIGTERM from your orchestrator kills a job mid-flight, leaving it stuck in the active list. A payment webhook keeps failing but sits silently in the failed set with no alert.

BeeQueue's minimalism is a feature during development but a gap in production. In this post, we'll go beyond the basics and cover the internal architecture and production hardening patterns that turn BeeQueue from a simple queue library into a production-grade worker framework.

1. Peeking Under the Hood: BeeQueue's Redis Key Structure

Every queue library is just a set of conventions on top of Redis data structures. When a job mysteriously vanishes, stalls, or duplicates, knowing exactly which Redis keys BeeQueue reads and writes is your fastest debugging path.

The Complete Key Layout

BeeQueue creates exactly 6 key types per queue, all namespaced under a configurable prefix (default bq):

Key Pattern	Redis Type	Purpose
bq:{name}:jobs	HASH	Job ID → JSON blob (data + options + state + result + error)
bq:{name}:waiting	LIST	Ordered IDs of jobs awaiting processing
bq:{name}:active	LIST	IDs of jobs currently being processed (via BRPOPLPUSH)
bq:{name}:succeeded	SET	IDs of completed jobs
bq:{name}:failed	SET	IDs of permanently failed jobs
bq:{name}:stall	HASH	Worker check-in timestamps for stall detection
bq:{name}:delayed	ZSET	Scores are timestamps; delayed jobs moved to waiting by scheduler

For comparison, BullMQ creates 10+ key types per queue — separate keys for job data vs metadata, an events list, a paused key, queue metadata, and repeatable job tracking. BeeQueue's simplicity means you can understand the entire storage model in a few minutes.

Inspecting Keys with Redis CLI

import { createClient } from 'redis';

async function inspectQueueKeys(prefix: string, queueName: string) {
  const client = createClient({ url: 'process.env.REDIS_URL' });
  await client.connect();

  const pattern = `${prefix}:${queueName}:*`;
  let cursor = 0;
  const keyInfo: Record<string, string> = {};

  do {
    const result = await client.scan(cursor, { MATCH: pattern, COUNT: 100 });
    cursor = result.cursor;
    for (const key of result.keys) {
      const type = await client.type(key);
      let encoding = '';
      if (type === 'hash' || type === 'string') {
        const debug = await client.objectEncoding(key);
        encoding = debug || 'unknown';
      }
      keyInfo[key] = `${type} (encoding: ${encoding})`;
    }
  } while (cursor !== 0);

  console.table(keyInfo);
  await client.quit();
}

// Usage:
await inspectQueueKeys('bq', 'email-queue');

Inside a BeeQueue Job Entry

Each job stored in the jobs HASH is a JSON blob containing everything BeeQueue knows about that job. Here's the schema:

interface BeeQueueJobInternal {
  id: string;
  data: Record<string, unknown>;
  options: {
    retries: number;
    timeout: number;
    backoff: { type: 'immediate' | 'fixed' | 'exponential'; delay: number };
    removeOnSuccess: boolean;
    removeOnFailure: boolean;
  };
  status: 'created' | 'waiting' | 'active' | 'succeeded' | 'failed' | 'delayed';
  result: unknown | null;
  error: { message: string; stack?: string } | null;
  progress: number;
}

Key Structural Difference from BullMQ

• Single HASH vs split keys: BeeQueue uses one jobs HASH for ALL job metadata. BullMQ splits this across separate keys for data, options, and state, which increases Redis operations per job lifecycle event.
• No events list: BeeQueue uses Redis Pub/Sub channels instead of a dedicated events list.
• No paused state: BeeQueue has no paused key — it relies on the active list emptiness as an implicit signal.
• No meta key: BeeQueue doesn't track queue-level metadata (creation time, version, etc.).

When to inspect manually:

• Stalled jobs: Check bq:{name}:stall for worker check-in timestamps. If the oldest check-in is more than stallInterval seconds old, the worker may have died.
• Zombie jobs: Look for job IDs in active whose worker has been gone for more than stallInterval — these will eventually be re-enqueued.
• Cleanup validation: Check succeeded and failed set sizes to verify your removeOnSuccess/removeOnFailure settings are working as expected.
• Capacity planning: Monitor the jobs HASH size to estimate Redis memory consumption.

2. Custom Health Checks and Prometheus Monitoring

BeeQueue deliberately ships without built-in monitoring — no QueueEvents equivalent, no lifecycle listeners, no built-in metrics. In production you need queue depth alerts, stalled-job detection, and processing-lag visibility. Here's how to build it.

The HealthChecker Class

BeeQueue's queue.checkHealth() returns counts per queue state. We'll poll this on a timer and export the results to Prometheus:

import Queue from 'bee-queue';
import client from 'prom-client';
import { Gauge, Counter, Histogram } from 'prom-client';
import express from 'express';

const register = new client.Registry();
client.collectDefaultMetrics({ register });

interface HealthResult {
  waiting: number;
  active: number;
  succeeded: number;
  failed: number;
  delayed: number;
  newestJob: number | null;
}

export class BeeQueueMetrics {
  private readonly queueDepth: Gauge<string>;
  private readonly activeJobs: Gauge<string>;
  private readonly failedJobs: Counter<string>;
  private readonly succeededJobs: Counter<string>;
  private readonly processingLatency: Histogram<string>;
  private readonly stallAgeSeconds: Gauge<string>;
  private readonly errorRate: Gauge<string>;
  private consecutiveErrors = 0;
  private totalChecks = 0;

  constructor(queueName: string, prefix = 'beequeue') {
    this.queueDepth = new Gauge({
      name: `${prefix}_queue_depth`,
      help: 'Number of jobs waiting in the queue',
      labelNames: ['queue'],
      registers: [register],
    });

    this.activeJobs = new Gauge({
      name: `${prefix}_active_jobs`,
      help: 'Number of jobs currently being processed',
      labelNames: ['queue'],
      registers: [register],
    });

    this.failedJobs = new Counter({
      name: `${prefix}_failed_jobs_total`,
      help: 'Total number of jobs that failed permanently',
      labelNames: ['queue'],
      registers: [register],
    });

    this.succeededJobs = new Counter({
      name: `${prefix}_succeeded_jobs_total`,
      help: 'Total number of jobs that completed successfully',
      labelNames: ['queue'],
      registers: [register],
    });

    this.processingLatency = new Histogram({
      name: `${prefix}_job_processing_duration_seconds`,
      help: 'Duration of job processing in seconds',
      labelNames: ['queue', 'status'],
      buckets: [0.01, 0.05, 0.1, 0.5, 1, 2, 5, 10, 30],
      registers: [register],
    });

    this.stallAgeSeconds = new Gauge({
      name: `${prefix}_stall_age_seconds`,
      help: 'Age of the oldest stalled worker check-in in seconds',
      labelNames: ['queue'],
      registers: [register],
    });

    this.errorRate = new Gauge({
      name: `${prefix}_health_check_error_ratio`,
      help: 'Ratio of failed health checks in the last window',
      labelNames: ['queue'],
      registers: [register],
    });
  }

  recordCheck(health: HealthResult, error: boolean): void {
    this.totalChecks++;
    this.consecutiveErrors = error
      ? this.consecutiveErrors + 1
      : Math.max(0, this.consecutiveErrors - 1);

    this.queueDepth.set({ queue: 'default' }, health.waiting);
    this.activeJobs.set({ queue: 'default' }, health.active);
    this.errorRate.set(
      { queue: 'default' },
      this.consecutiveErrors / Math.max(1, this.totalChecks)
    );
  }

  recordJobCompletion(status: 'succeeded' | 'failed', durationMs: number): void {
    this.processingLatency.observe({ queue: 'default', status }, durationMs / 1000);
    if (status === 'succeeded') {
      this.succeededJobs.inc({ queue: 'default' });
    } else {
      this.failedJobs.inc({ queue: 'default' });
    }
  }

  recordStallAge(ageSeconds: number): void {
    this.stallAgeSeconds.set({ queue: 'default' }, ageSeconds);
  }
}

The Health Monitor Poller

export class BeeQueueHealthMonitor {
  private readonly queue: Queue;
  private readonly metrics: BeeQueueMetrics;
  private intervalHandle: ReturnType<typeof setInterval> | null = null;

  constructor(queue: Queue, metrics: BeeQueueMetrics) {
    this.queue = queue;
    this.metrics = metrics;
  }

  start(pollIntervalMs = 15_000): void {
    this.intervalHandle = setInterval(async () => {
      try {
        const health = await this.queue.checkHealth() as unknown as HealthResult;
        this.metrics.recordCheck(health, false);
        const stallAge = await this.computeStallAge();
        this.metrics.recordStallAge(stallAge);

        if (health.waiting > 1000) {
          console.warn(`Queue depth alert: ${health.waiting} waiting jobs`);
        }
        if (health.failed > 50) {
          console.warn(`High failure count: ${health.failed} failed jobs`);
        }
      } catch (err) {
        this.metrics.recordCheck(
          { waiting: 0, active: 0, succeeded: 0, failed: 0, delayed: 0, newestJob: null },
          true
        );
        console.error('Health check failed:', err);
      }
    }, pollIntervalMs);
  }

  stop(): void {
    if (this.intervalHandle) {
      clearInterval(this.intervalHandle);
      this.intervalHandle = null;
    }
  }

  private async computeStallAge(): Promise<number> {
    try {
      const redisClient = (this.queue as any).redis;
      if (!redisClient) return 0;
      const prefix = (this.queue as any).keyPrefix || 'bq';
      const stallKey = `${prefix}:${this.queue.name}:stall`;
      const now = Date.now();
      const entries = await redisClient.hGetAll(stallKey);
      let oldest = now;
      for (const workerId of Object.keys(entries)) {
        const checkin = parseInt(entries[workerId], 10);
        if (checkin < oldest) oldest = checkin;
      }
      return oldest < now ? (now - oldest) / 1000 : 0;
    } catch {
      return 0;
    }
  }
}

Wiring It All Together

import Queue from 'bee-queue';
import express from 'express';

const queue = new Queue('email-queue', { prefix: 'bq' });

const metrics = new BeeQueueMetrics('email-queue');
const monitor = new BeeQueueHealthMonitor(queue, metrics);
monitor.start(15_000);

queue.process(async (job) => {
  const start = Date.now();
  try {
    const result = await sendEmail(job.data);
    metrics.recordJobCompletion('succeeded', Date.now() - start);
    return result;
  } catch (err) {
    metrics.recordJobCompletion('failed', Date.now() - start);
    throw err;
  }
});

const app = express();
app.get('/metrics', async (_req, res) => {
  try {
    res.set('Content-Type', register.contentType);
    res.end(await register.metrics());
  } catch (err) {
    res.status(500).end(String(err));
  }
});
app.listen(9090, () => console.log('Metrics at http://localhost:9090/metrics'));

With this setup, Prometheus scrapes /metrics every 15 seconds, and you get alerts for queue depth, stall age, and error rates — all without any built-in monitoring from BeeQueue.

3. Batch Processing Patterns

BeeQueue shines at high throughput, but processing jobs one-at-a-time leaves throughput on the table. Here are four batch and fan-out patterns to saturate your Redis connection and aggregate results efficiently.

Pattern A: Fan-Out with Concurrent Workers

Multiple workers on the same queue is the simplest fan-out. BeeQueue's brpoplpush mechanism means idle workers get jobs immediately via blocking pop.

import Queue from 'bee-queue';
import { cpus } from 'os';

const imageQueue = new Queue('image-processing', {
  prefix: 'bq',
  stallInterval: 10_000,
});

const WORKER_CONCURRENCY = Math.max(1, cpus().length - 1);

imageQueue.process(WORKER_CONCURRENCY, async (job) => {
  const result = await processImage(job.data.imagePath, job.data.operations);
  return { jobId: job.id, processed: true, outputPath: result };
});

Deploy this across multiple containers or servers — each runs the same code. BeeQueue distributes jobs via BRPOPLPUSH across all workers.

Pattern B: Batch Consumption with getJobs()

When processing one job at a time has too much Redis overhead (e.g., inserting analytics records in bulk), pull multiple jobs at once:

import Queue from 'bee-queue';

interface AnalyticsEvent {
  userId: string;
  event: string;
  timestamp: number;
}

const analyticsQueue = new Queue('analytics-ingest', { prefix: 'bq' });
const BATCH_SIZE = 50;

analyticsQueue.process(1, async (job, done) => {
  const batch: AnalyticsEvent[] = [job.data as AnalyticsEvent];

  const pendingJobs = await analyticsQueue.getJobs('waiting', { size: BATCH_SIZE - 1 });

  for (const pending of pendingJobs) {
    batch.push(pending.data as AnalyticsEvent);
    await analyticsQueue.removeJob(pending.id);
  }

  await bulkInsertAnalytics(batch);
  done(null, { batched: batch.length });
});

Pattern C: Worker Pool with Rate Limiting

Control exactly how many concurrent jobs run with an external rate limiter:

import Queue from 'bee-queue';
import Bottleneck from 'bottleneck';

const smsQueue = new Queue('sms-delivery', { prefix: 'bq' });

const limiter = new Bottleneck({
  maxConcurrent: 5,
  minTime: 100, // 10 per second (Twilio API limit)
});

smsQueue.process(5, async (job) => {
  const payload = job.data as { phone: string; message: string };
  return limiter.schedule(() => sendSms(payload.phone, payload.message));
});

Pattern D: Aggregate Results with saveAll()

When the producer needs to create many jobs efficiently and collect all results:

import Queue from 'bee-queue';

interface CheckResult {
  url: string;
  statusCode: number;
  responseTimeMs: number;
}

const healthCheckQueue = new Queue('health-check', { prefix: 'bq' });

async function batchHealthCheck(urls: string[]): Promise<void> {
  const jobs = urls.map((url) =>
    healthCheckQueue.createJob({ url, checkTime: Date.now() })
  );
  const errors = await healthCheckQueue.saveAll(jobs);
  for (const [job, err] of errors) {
    if (err) {
      console.error(`Failed to save job for ${job.data.url}:`, err);
    }
  }
}

healthCheckQueue.process(10, async (job) => {
  return performHttpCheck(job.data.url) as Promise<CheckResult>;
});

4. Implementing Dead Letter Queues and Failure Escalation

BeeQueue has retries and a failed set, but failed jobs sit there indefinitely with no escalation path. Without a dead letter pattern, a failing job can silently rot in Redis or pile up until memory is exhausted.

Building a DeadLetterHandler

We'll store failed job data in a separate Redis HASH with TTL and offer a replay mechanism:

import Queue from 'bee-queue';
import { createClient, RedisClientType } from 'redis';

export interface DeadLetterEntry {
  originalJobId: string;
  queueName: string;
  data: unknown;
  options: Record<string, unknown>;
  error: { message: string; stack?: string };
  failedAt: string;
  failureCount: number;
}

interface DeadLetterHandlerOpts {
  redisUrl?: string;
  ttlSeconds?: number;
  maxRetries?: number;
  onEscalation?: (entry: DeadLetterEntry) => Promise<void>;
}

export class DeadLetterHandler {
  private readonly dlqClient: RedisClientType;
  private readonly dlqKey: string;
  private readonly ttlSeconds: number;
  private readonly maxRetries: number;
  private readonly onEscalation?: (entry: DeadLetterEntry) => Promise<void>;

  constructor(queueName: string, opts?: DeadLetterHandlerOpts) {
    this.dlqKey = `dlq:${queueName}`;
    this.ttlSeconds = opts?.ttlSeconds ?? 604_800;
    this.maxRetries = opts?.maxRetries ?? 3;
    this.onEscalation = opts?.onEscalation;
    this.dlqClient = createClient({ url: opts?.redisUrl ?? 'redis://localhost:6379' });
  }

  async connect(): Promise<void> {
    await this.dlqClient.connect();
  }

  async disconnect(): Promise<void> {
    await this.dlqClient.quit();
  }

  async store(entry: DeadLetterEntry): Promise<boolean> {
    const exists = await this.dlqClient.hExists(this.dlqKey, entry.originalJobId);
    await this.dlqClient.hSet(this.dlqKey, entry.originalJobId, JSON.stringify(entry));
    await this.dlqClient.expire(this.dlqKey, this.ttlSeconds);
    return !exists;
  }

  async get(jobId: string): Promise<DeadLetterEntry | null> {
    const raw = await this.dlqClient.hGet(this.dlqKey, jobId);
    return raw ? JSON.parse(raw) as DeadLetterEntry : null;
  }

  async list(): Promise<DeadLetterEntry[]> {
    const map = await this.dlqClient.hGetAll(this.dlqKey);
    return Object.values(map).map((raw) => JSON.parse(raw) as DeadLetterEntry);
  }

  async replay(queue: Queue, jobId: string): Promise<void> {
    const entry = await this.get(jobId);
    if (!entry) throw new Error(`Dead letter entry ${jobId} not found`);

    const newJob = queue.createJob(entry.data);
    newJob.retries(entry.options.retries as number ?? 3);
    if (entry.options.backoff) {
      const backoff = entry.options.backoff as { type: string; delay: number };
      newJob.backoff(backoff.type as 'immediate' | 'fixed' | 'exponential', backoff.delay);
    }
    await newJob.save();
    await this.dlqClient.hDel(this.dlqKey, jobId);
    console.log(`Replayed job ${jobId} as new job ${newJob.id}`);
  }

  async replayAll(queue: Queue): Promise<number> {
    const entries = await this.list();
    let count = 0;
    for (const entry of entries) {
      try {
        await this.replay(queue, entry.originalJobId);
        count++;
      } catch (err) {
        console.error(`Failed to replay ${entry.originalJobId}:`, err);
      }
    }
    return count;
  }

  async count(): Promise<number> {
    return this.dlqClient.hLen(this.dlqKey);
  }
}

Integration with BeeQueue Processor

async function createDLQProtectedQueue(
  queueName: string,
  handler: (job: Queue.Job) => Promise<unknown>,
  dlqHandler: DeadLetterHandler
): Promise<Queue> {
  const queue = new Queue(queueName, { prefix: 'bq' });

  queue.process(async (job) => {
    try {
      return await handler(job);
    } catch (err) {
      const failureCount =
        (job.options.retries ?? 0) - (job.options.remainingRetries ?? 0);
      const entry: DeadLetterEntry = {
        originalJobId: job.id,
        queueName,
        data: job.data,
        options: job.options as Record<string, unknown>,
        error: {
          message: err instanceof Error ? err.message : String(err),
          stack: err instanceof Error ? err.stack : undefined,
        },
        failedAt: new Date().toISOString(),
        failureCount,
      };

      await dlqHandler.store(entry);

      if (failureCount >= 3 && dlqHandler['onEscalation']) {
        await dlqHandler['onEscalation'](entry);
      }

      throw err;
    }
  });

  return queue;
}

// Usage
const dlq = new DeadLetterHandler('payment-webhook', {
  ttlSeconds: 86400 * 7,
  maxRetries: 3,
  onEscalation: async (entry) => {
    await sendSlackAlert({
      channel: '#queue-alerts',
      text: `Payment webhook ${entry.originalJobId} failed ${entry.failureCount} times. Error: ${entry.error.message}`,
    });
  },
});
await dlq.connect();

const queue = await createDLQProtectedQueue(
  'payment-webhook',
  async (job) => processPaymentWebhook(job.data),
  dlq
);

5. Graceful Shutdown — Taming BeeQueue's Worker Lifecycle

Unlike BullMQ where worker.close() waits for active jobs, BeeQueue's queue.close() only closes Redis connections — it does not wait for in-flight jobs. This means a SIGTERM can kill jobs mid-flight, leaving them stuck in the active list until stallInterval expires.

Building a ShutdownManager

import Queue from 'bee-queue';

interface ActiveJobTracker {
  jobId: string;
  startedAt: number;
  abort?: () => void;
}

interface ShutdownSummary {
  completedJobs: number;
  killedJobs: number;
  totalDuration: number;
  timedOut: boolean;
}

export class BeeQueueShutdownManager {
  private readonly queue: Queue;
  private readonly activeJobs: Map<string, ActiveJobTracker> = new Map();
  private readonly gracefulTimeoutMs: number;
  private shuttingDown = false;
  private jobCounter = 0;

  constructor(queue: Queue, gracefulTimeoutMs = 30_000) {
    this.queue = queue;
    this.gracefulTimeoutMs = gracefulTimeoutMs;
  }

  trackJob(jobId: string, abort?: () => void): void {
    if (this.shuttingDown) {
      throw new Error('Worker is shutting down, not accepting new jobs');
    }
    this.activeJobs.set(jobId, { jobId, startedAt: Date.now(), abort });
    this.jobCounter++;
  }

  completeJob(jobId: string): void {
    this.activeJobs.delete(jobId);
  }

  installHandlers(): void {
    const handleSignal = async (signal: string) => {
      console.log(`Received ${signal}, starting graceful shutdown...`);
      const summary = await this.shutdown();
      console.log('Shutdown complete:', summary);
      process.exit(0);
    };
    process.on('SIGTERM', handleSignal);
    process.on('SIGINT', handleSignal);
  }

  async shutdown(): Promise<ShutdownSummary> {
    this.shuttingDown = true;
    const startedAt = Date.now();

    if (this.activeJobs.size === 0) {
      await this.queue.close();
      return { completedJobs: 0, killedJobs: 0, totalDuration: Date.now() - startedAt, timedOut: false };
    }

    const timedOut = await this.waitForActiveJobs(this.gracefulTimeoutMs);

    let killedCount = 0;
    for (const [jobId, tracker] of this.activeJobs) {
      try { tracker.abort?.(); } catch { /* ignore */ }
      killedCount++;
    }
    this.activeJobs.clear();
    await this.queue.close();

    return {
      completedJobs: this.jobCounter - killedCount,
      killedJobs: killedCount,
      totalDuration: Date.now() - startedAt,
      timedOut: timedOut === 'timeout',
    };
  }

  private async waitForActiveJobs(timeoutMs: number): Promise<'completed' | 'timeout'> {
    return new Promise((resolve) => {
      const checkInterval = 100;
      let elapsed = 0;
      const interval = setInterval(() => {
        elapsed += checkInterval;
        if (this.activeJobs.size === 0) {
          clearInterval(interval);
          resolve('completed');
        } else if (elapsed >= timeoutMs) {
          clearInterval(interval);
          resolve('timeout');
        }
      }, checkInterval);
    });
  }
}

Usage with Kubernetes

const queue = new Queue('video-transcode', { prefix: 'bq' });
const shutdown = new BeeQueueShutdownManager(queue, 30_000);
shutdown.installHandlers();

queue.process(3, async (job) => {
  let aborted = false;
  shutdown.trackJob(job.id, () => { aborted = true; });

  try {
    for (let i = 0; i < 100; i++) {
      if (aborted) return { aborted: true };
      await job.reportProgress(i);
      await new Promise((r) => setTimeout(r, 100));
    }
    return await transcodeVideo(job.data);
  } finally {
    shutdown.completeJob(job.id);
  }
});

With the K8s preStop hook and a 40-second pod termination grace period, your BeeQueue worker drains active jobs gracefully before the pod exits.

6. Performance Benchmarking Methodology

"BeeQueue is fast" is a claim, not data. To make informed trade-offs between throughput and reliability, you need a reproducible benchmarking methodology.

The Benchmark Runner

import Queue from 'bee-queue';
import { performance, PerformanceObserver } from 'perf_hooks';
import { createClient } from 'redis';

interface BenchmarkConfig {
  queueName: string;
  concurrency: number;
  payloadSize: number;
  jobDurationMs: number;
  totalJobs: number;
  prefix?: string;
}

interface BenchmarkResult {
  config: BenchmarkConfig;
  jobsPerSecond: number;
  p50LatencyMs: number;
  p95LatencyMs: number;
  p99LatencyMs: number;
  errorCount: number;
}

export class BeeQueueBenchmark {
  private latencies: number[] = [];
  private completed = 0;
  private errors = 0;

  async run(config: BenchmarkConfig): Promise<BenchmarkResult> {
    this.latencies = [];
    this.errors = 0;
    this.completed = 0;

    const queue = new Queue(config.queueName, {
      prefix: config.prefix ?? 'bench',
      isWorker: true,
      storeJobs: false,
      getEvents: false,
      sendEvents: false,
      removeOnSuccess: true,
      removeOnFailure: true,
    });
    await queue.ready();

    const payload = { field: 'x'.repeat(Math.max(1, config.payloadSize - 50)) };

    const obs = new PerformanceObserver((list) => {
      for (const entry of list.getEntries()) {
        if (entry.name === 'job-process') this.latencies.push(entry.duration);
      }
    });
    obs.observe({ type: 'measure', buffered: false });

    // Warm-up
    await this.warmUp(queue, payload, 100);

    // Process
    const processPromise = new Promise<void>((resolve) => {
      queue.process(config.concurrency, async (job) => {
        performance.mark('job-start');
        await new Promise((r) => setTimeout(r, config.jobDurationMs));
        performance.mark('job-end');
        performance.measure('job-process', 'job-start', 'job-end');
        return { processed: true };
      });

      queue.on('succeeded', () => {
        this.completed++;
        if (this.completed >= config.totalJobs) resolve();
      });
      queue.on('failed', () => {
        this.errors++;
        this.completed++;
        if (this.completed >= config.totalJobs) resolve();
      });
    });

    // Enqueue
    for (let i = 0; i < config.totalJobs; i += 100) {
      const batch = Array.from(
        { length: Math.min(100, config.totalJobs - i) },
        () => queue.createJob(payload)
      );
      await queue.saveAll(batch);
    }

    await processPromise;
    obs.disconnect();
    await queue.destroy();
    await queue.close();

    const sorted = [...this.latencies].sort((a, b) => a - b);
    const p50 = this.percentile(sorted, 50);
    const p95 = this.percentile(sorted, 95);
    const p99 = this.percentile(sorted, 99);

    return {
      config,
      jobsPerSecond: (config.totalJobs / (sorted.reduce((a, b) => a + b, 0) / sorted.length)) * 1000,
      p50LatencyMs: p50,
      p95LatencyMs: p95,
      p99LatencyMs: p99,
      errorCount: this.errors,
    };
  }

  private async warmUp(queue: Queue, payload: Record<string, unknown>, count: number): Promise<void> {
    return new Promise((resolve) => {
      let warmed = 0;
      queue.process(10, async () => {
        await new Promise((r) => setTimeout(r, 5));
        return { done: true };
      });
      queue.on('succeeded', () => {
        warmed++;
        if (warmed >= count) {
          queue.removeAllListeners('succeeded');
          resolve();
        }
      });
      for (let i = 0; i < count; i++) queue.createJob(payload).save();
    });
  }

  private percentile(sorted: number[], p: number): number {
    if (sorted.length === 0) return 0;
    const index = Math.ceil((p / 100) * sorted.length) - 1;
    return sorted[Math.max(0, index)];
  }
}

Tuning Takeaways

• BeeQueue throughput scales almost linearly with concurrency until you saturate Redis CPU.
• Large payloads (>50KB) bottleneck on Redis network bandwidth and serialization.
• Job duration has minimal impact on latency variance — BeeQueue's overhead is ~1-2ms per job.
• Memory overhead per job: ~400 bytes for the jobs hash entry, plus payload size.
• Enabling storeJobs, getEvents, and sendEvents adds ~15% throughput overhead — disable these on worker-only queues.

Beyond These Patterns

BeeQueue's minimalism is a double-edged sword: it delivers raw throughput and a tiny dependency footprint, but the lack of built-in production features means you have to build your own health monitoring, dead letter handling, graceful shutdown, and batch processing patterns. The Redis key structure is refreshingly simple — just a handful of lists, sets, and hashes — making it easy to debug with redis-cli and integrate with Prometheus.

As your queue infrastructure grows across multiple backends and teams, maintaining these patterns yourself becomes a full-time job. When you need unified visibility and management for both BeeQueue and BullMQ queues from a single dashboard — without building your own monitoring platform — QueueHub has you covered.

QueueHub gives you production-grade monitoring, live worker views, queue management, and real-time alerts for all your Redis-backed queues. Zero configuration — point it at your Redis and go.

Try QueueHub free →