Event-Driven and Tiered Cleanup: Advanced Redis Expiration Strategies for BullMQ

Your queue is silently consuming Redis memory right now. Every completed job, every failed attempt, every log entry occupies a Redis key that lives in RAM. If your queue processes tens of thousands of jobs daily, that RAM adds up — fast.

BullMQ's built-in clean() API is the standard go-to, and for small queues it works perfectly. But clean() is an O(N) blocking operation: it calls LREM, ZREMRANGEBYSCORE, or SREM on Redis lists and sorted sets. At 500K completed jobs, a single clean() call can block the Redis event loop for several seconds, causing latency spikes across all queue operations.

This post explores four alternative strategies that bypass clean() entirely:

1. Event-driven cleanup — QueueEvents listeners trigger immediate per-job removal
2. SCAN-based gradual cleanup — cursor-based, non-blocking iteration over job keys
3. EXPIRE at the individual key level — Redis-managed TTL for zero-code data expiration
4. Dead letter queue lifecycle management — separate cleanup for permanently failed jobs

Plus: tiered archival pipelines, a benchmark framework to compare strategies, and multi-queue cleanup orchestration with dependency ordering.

---

Section 1: Event-Driven Cleanup with QueueEvents

BullMQ's QueueEvents class subscribes to Redis streams for global job lifecycle events. Unlike Worker.on('completed'), which only fires in the worker process that handled the job, QueueEvents receives events from all workers — one listener covers your entire fleet.

This makes QueueEvents the ideal foundation for cleanup: listen to completed and failed events, then remove the individual job key immediately. No polling interval, no batch operations — just real-time, per-job precision.

A simple event-driven cleanup listener

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

const connection = new Redis({
  host: 'localhost',
  port: 6379,
  maxRetriesPerRequest: null,
});

const queue = new Queue('orders', { connection });
const queueEvents = new QueueEvents('orders', { connection });

queueEvents.on('completed', async ({ jobId }: { jobId: string }) => {
  try {
    const job = await Job.fromId(queue, jobId);
    if (!job) return;

    // Only remove jobs older than 60 seconds (keep recent ones visible)
    const ageSeconds = (Date.now() - job.timestamp) / 1000;
    if (ageSeconds < 60) return;

    // Remove the job from the queue and its log key
    await job.remove();
    await connection.del(`bull:orders:${jobId}:logs`);

    console.log(`Removed completed job ${jobId} (age: ${ageSeconds.toFixed(0)}s)`);
  } catch (error) {
    console.error(`Failed to clean up job ${jobId}:`, error);
  }
});

Conditional cleanup with retention rules

For production systems, you rarely want a blanket delete-all policy. Different job types have different retention requirements — email notifications can be purged after minutes, while financial reconciliation jobs need hours of inspectability.

interface CleanupRule {
  jobName: string;
  keepIf: (job: Job) => boolean;
  onRemove?: (job: Job) => Promise<void>;
}

class EventDrivenCleanup {
  private rules: CleanupRule[] = [];
  private queue: Queue;
  private queueEvents: QueueEvents;
  private connection: Redis;

  constructor(queueName: string, connection: Redis) {
    this.connection = connection;
    this.queue = new Queue(queueName, { connection });
    this.queueEvents = new QueueEvents(queueName, { connection });
  }

  addRule(rule: CleanupRule): void {
    this.rules.push(rule);
  }

  start(): void {
    this.queueEvents.on('completed', async ({ jobId }) => {
      await this.handleJobFinalized(jobId, 'completed');
    });
    this.queueEvents.on('failed', async ({ jobId }) => {
      await this.handleJobFinalized(jobId, 'failed');
    });
  }

  private async handleJobFinalized(
    jobId: string,
    status: 'completed' | 'failed',
  ): Promise<void> {
    const job = await Job.fromId(this.queue, jobId);
    if (!job) return;

    for (const rule of this.rules) {
      if (rule.jobName !== job.name && rule.jobName !== '*') continue;
      if (rule.keepIf(job)) return; // keep-rule matched; skip removal
    }

    // Remove log key and job data hash, then remove from meta-sets
    await this.connection.del(`bull:${this.queue.name}:${jobId}:logs`);
    await this.connection.unlink(`bull:${this.queue.name}:${jobId}`);
    await job.remove();
  }

  async close(): Promise<void> {
    await this.queueEvents.close();
    await this.queue.close();
  }
}

// Usage
const cleanup = new EventDrivenCleanup('orders', connection);

cleanup.addRule({
  jobName: 'send-email',
  keepIf: (job) => {
    const ageSeconds = (Date.now() - job.timestamp) / 1000;
    return ageSeconds < 300; // keep email jobs for 5 minutes
  },
});

cleanup.addRule({
  jobName: 'generate-report',
  keepIf: () => false, // never keep reports
  onRemove: async (job) => {
    await cleanupExternalStorage(job.data.reportId);
  },
});

cleanup.start();

Important caveat: QueueEvents uses Redis streams with at-most-once delivery. If the listener crashes after receiving the event but before calling remove(), the job survives. Design for idempotency — and pair event-driven cleanup with a periodic SCAN sweep (Section 2) as a safety net.

---

Section 2: Redis SCAN-Based Gradual Cleanup

While event-driven cleanup is ideal for real-time precision, it can't catch everything. What about jobs that completed before your cleanup listener started? Or jobs missed due to a listener restart?

Enter SCAN-based cleanup — a gradual, cursor-based approach that iterates over job keys in small batches without blocking Redis.

The key difference from clean(): clean() calls ZREMRANGEBYSCORE on the sorted set, which scans the entire set — O(N) blocking. SCAN-based cleanup uses ZSCAN with a cursor, processing 100-200 keys per round, then UNLINK (non-blocking delete in Redis 4.0+).

The SCAN-based cleanup worker

import { Redis } from 'ioredis';

interface ScanCleanupConfig {
  queueName: string;
  connection: Redis;
  batchSize?: number;
  maxJobAgeSeconds?: number;
  intervalMs?: number;
}

class ScanBasedCleanupWorker {
  private connection: Redis;
  private config: Required<ScanCleanupConfig>;
  private active = false;

  constructor(config: ScanCleanupConfig) {
    this.connection = config.connection;
    this.config = {
      batchSize: config.batchSize ?? 100,
      maxJobAgeSeconds: config.maxJobAgeSeconds ?? 86400,
      intervalMs: config.intervalMs ?? 60_000,
      ...config,
    };
  }

  start(): void {
    this.active = true;
    this.runSweepCycle();
  }

  stop(): void {
    this.active = false;
  }

  private async runSweepCycle(): Promise<void> {
    while (this.active) {
      const removed = await this.sweepCompletedJobs();
      if (removed > 0) {
        console.log(
          `Sweep removed ${removed} jobs for queue "${this.config.queueName}"`,
        );
      }
      await this.sleep(this.config.intervalMs);
    }
  }

  private async sweepCompletedJobs(): Promise<number> {
    const keyPrefix = `bull:${this.config.queueName}:`;
    let removedTotal = 0;

    for (const setKey of [`${keyPrefix}completed`, `${keyPrefix}failed`]) {
      let cursor = '0';
      let removed = 0;
      const now = Date.now();
      const maxAgeMs = this.config.maxJobAgeSeconds * 1000;

      do {
        const [nextCursor, members] = await this.connection.zscan(
          setKey, cursor, 'COUNT', this.config.batchSize,
        );
        cursor = nextCursor;

        const idsToRemove: string[] = [];
        for (let i = 0; i < members.length; i += 2) {
          const jobId = members[i];
          const score = parseInt(members[i + 1], 10);
          if (now - score > maxAgeMs) {
            idsToRemove.push(jobId);
          }
        }

        if (idsToRemove.length > 0) {
          const multi = this.connection.multi();
          multi.zrem(setKey, ...idsToRemove);
          multi.unlink(
            ...idsToRemove.map((id) => `${keyPrefix}${id}`),
          );
          multi.unlink(
            ...idsToRemove.map((id) => `${keyPrefix}${id}:logs`),
          );
          await multi.exec();
          removed += idsToRemove.length;
        }
      } while (cursor !== '0');

      removedTotal += removed;
    }

    return removedTotal;
  }

  private sleep(ms: number): Promise<void> {
    return new Promise((resolve) => setTimeout(resolve, ms));
  }
}

// Run every 2 minutes, remove jobs older than 1 hour
const redis = new Redis({ host: 'localhost', port: 6379, maxRetriesPerRequest: null });

const cleanupWorker = new ScanBasedCleanupWorker({
  queueName: 'orders',
  connection: redis,
  batchSize: 200,
  maxJobAgeSeconds: 3600,
  intervalMs: 120_000,
});

cleanupWorker.start();

Why UNLINK instead of DEL? DEL is synchronous and blocks Redis while reclaiming memory. UNLINK (Redis 4.0+) performs the deletion in a background thread and returns immediately. For workloads that delete thousands of keys per sweep, the difference is substantial.

Trade-off: SCAN-based cleanup is not real-time — it depends on the sweep interval. But it's the safest option for queues with hundreds of thousands of completed jobs where clean() causes measurable latency.

---

Section 3: Redis EXPIRE at the Individual Key Level

The most efficient cleanup mechanism in Redis is the one that requires zero application code: setting a TTL on each key and letting Redis handle deletion internally.

BullMQ doesn't set TTL on job data keys by default — this is by design, as some applications need long-lived job data for observability. But you can layer TTLs on top using either QueueEvents (as in Section 1) or directly inside your worker.

EXPIRE-based cleanup via QueueEvents

interface ExpirePolicy {
  jobDataTTL: number;   // seconds until job hash expires
  jobLogTTL: number;    // seconds until job logs expire
  keepIf?: (job: Job) => boolean;
}

class ExpireBasedCleanup {
  private connection: Redis;
  private queue: Queue;
  private queueEvents: QueueEvents;
  private policies: Map<string, ExpirePolicy> = new Map();

  constructor(queueName: string, connection: Redis) {
    this.connection = connection;
    this.queue = new Queue(queueName, { connection });
    this.queueEvents = new QueueEvents(queueName, { connection });
  }

  setPolicy(jobName: string, policy: ExpirePolicy): void {
    this.policies.set(jobName, policy);
  }

  start(): void {
    this.queueEvents.on('completed', async ({ jobId }) => {
      await this.applyExpire(jobId);
    });
    this.queueEvents.on('failed', async ({ jobId }) => {
      await this.applyExpire(jobId);
    });
  }

  private async applyExpire(jobId: string): Promise<void> {
    const job = await Job.fromId(this.queue, jobId);
    if (!job) return;

    const policy = this.policies.get(job.name) ?? this.policies.get('*');
    if (!policy) return;
    if (policy.keepIf && policy.keepIf(job)) return;

    const keyPrefix = `bull:${this.queue.name}:`;
    const multi = this.connection.multi();
    multi.expire(`${keyPrefix}${jobId}`, policy.jobDataTTL);
    multi.expire(`${keyPrefix}${jobId}:logs`, policy.jobLogTTL);
    await multi.exec();
  }

  async close(): Promise<void> {
    await this.queueEvents.close();
    await this.queue.close();
  }
}

// Usage
const expireCleanup = new ExpireBasedCleanup('orders', redis);

expireCleanup.setPolicy('*', {
  jobDataTTL: 3600,
  jobLogTTL: 7200,
});

expireCleanup.setPolicy('generate-report', {
  jobDataTTL: 86400,
  jobLogTTL: 86400,
  keepIf: (job) => {
    const result = job.returnvalue as { status?: string };
    return result?.status === 'failed';
  },
});

expireCleanup.start();

Post-processing EXPIRE from inside the worker

If you prefer a simpler approach without QueueEvents, set TTLs directly in the worker after processing:

const worker = new Worker(
  'orders',
  async (job: Job) => {
    const result = await processOrder(job.data);

    // Set TTL on this job's Redis keys right after processing
    const multi = redis.multi();
    multi.expire(`bull:orders:${job.id}`, 1800);       // data: 30 min
    multi.expire(`bull:orders:${job.id}:logs`, 3600);  // logs: 1 hour
    await multi.exec();

    return result;
  },
  { connection: redis },
);

How Redis expiration works under the hood: Redis uses a hybrid lazy + periodic expiration strategy. Expired keys are freed on access (lazy deletion) and during a background cycle that runs every 100ms, sampling a subset of TTL'd keys. This means a key with a 3600-second TTL won't necessarily be freed at exactly 3600.000 seconds — but for cleanup purposes, the ~100ms granularity is more than sufficient.

Limitation: EXPIRE only covers individual job data hashes and log keys. The completed and failed sorted sets (bull:orders:completed, bull:orders:failed) are not TTL'd — those still need removeOnComplete/removeOnFail or a SCAN sweep. EXPIRE is best used as a fallback layer: even if your cleanup process dies, job data won't linger past the TTL.

---

Section 4: Dead Letter Queue Cleanup Lifecycle

BullMQ has no built-in dead letter queue, but the pattern is essential for jobs that exhaust their retries. Failed jobs that exceed maxAttempts are moved to a separate DLQ queue for forensic inspection — but they shouldn't live there forever.

A pragmatic DLQ lifecycle has three phases:

• Warm phase: The job is fully inspectable (can view data, result, stack traces via a dashboard). This phase lasts minutes to hours.
• Cold phase: The full job hash is expired. Only a tiny metadata marker (~200 bytes) remains with an archive timestamp and purge date.
• Purge phase: All traces of the job are permanently deleted, including metadata markers and sorted set entries.

interface DLQCleanupPolicy {
  warmRetentionMinutes: number;
  coldRetentionDays: number;
  autoPurgeAfterDays: number;
}

class DeadLetterCleanup {
  private dlq: Queue;
  private dlqEvents: QueueEvents;
  private connection: Redis;

  constructor(dlqQueueName: string, connection: Redis) {
    this.dlq = new Queue(dlqQueueName, { connection });
    this.dlqEvents = new QueueEvents(dlqQueueName, { connection });
    this.connection = connection;
  }

  start(policy: DLQCleanupPolicy): void {
    this.dlqEvents.on('completed', async ({ jobId }) => {
      await this.applyDLQExpiration(jobId, policy);
    });

    // Hourly purge of expired DLQ metadata
    setInterval(async () => {
      await this.purgeExpiredDLQJobs(policy);
    }, 3_600_000);
  }

  private async applyDLQExpiration(
    jobId: string,
    policy: DLQCleanupPolicy,
  ): Promise<void> {
    const multi = this.connection.multi();
    multi.expire(`bull:${this.dlq.name}:${jobId}`, policy.warmRetentionMinutes * 60);

    // Archive marker persists after the job data hash expires
    multi.setex(
      `bull:${this.dlq.name}:${jobId}:archive`,
      policy.coldRetentionDays * 86400,
      JSON.stringify({ archivedAt: Date.now(), purgeAfter: policy.autoPurgeAfterDays }),
    );

    await multi.exec();
  }

  private async purgeExpiredDLQJobs(policy: DLQCleanupPolicy): Promise<void> {
    const keyPattern = `bull:${this.dlq.name}:*:archive`;
    let cursor = '0';

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

      for (const key of keys) {
        const raw = await this.connection.get(key);
        if (!raw) continue;

        const meta = JSON.parse(raw) as { archivedAt: number; purgeAfter: number };
        const ageDays = (Date.now() - meta.archivedAt) / 86_400_000;

        if (ageDays > meta.purgeAfter) {
          const jobId = key
            .replace(`bull:${this.dlq.name}:`, '')
            .replace(':archive', '');

          const multi = this.connection.multi();
          multi.del(key);
          multi.zrem(`bull:${this.dlq.name}:completed`, jobId);
          multi.zrem(`bull:${this.dlq.name}:failed`, jobId);
          await multi.exec();
        }
      }
    } while (cursor !== '0');
  }

  async close(): Promise<void> {
    await this.dlqEvents.close();
    await this.dlq.close();
  }
}

The archive marker pattern is critical for memory efficiency: a full job hash with data, result, and stack traces may be 2-50 KB, while the archive marker is under 200 bytes. Converting warm-phase jobs to cold-phase markers at scale can reclaim gigabytes of Redis memory.

---

Section 5: Cleanup Strategy Benchmarking

Choosing the right strategy requires data, not intuition. Here's a benchmark harness that measures Redis CPU usage, wall-clock time, and memory impact across all four strategies (plus clean() as a baseline).

interface BenchmarkResult {
  strategy: string;
  totalTimeMs: number;
  redisCpuDelta: number;
  memoryFreedBytes: number;
}

class CleanupBenchmark {
  private connection: Redis;
  private queue: Queue;

  constructor(queueName: string, connection: Redis) {
    this.queue = new Queue(queueName, { connection });
    this.connection = connection;
  }

  async seedJobs(count: number): Promise<void> {
    const batchSize = 500;
    for (let i = 0; i < count; i += batchSize) {
      const batch = [];
      for (let j = 0; j < batchSize && i + j < count; j++) {
        batch.push({
          name: 'benchmark-job',
          data: { id: i + j, payload: `x`.repeat(1024) },
          opts: { removeOnComplete: false, removeOnFail: false },
        });
      }
      await this.queue.addBulk(batch);
    }
  }

  async benchmarkClean(): Promise<BenchmarkResult> {
    const cpuBefore = await this.getRedisCpu();
    const memBefore = await this.getRedisMemory();
    const start = Date.now();

    await this.queue.clean(0, 0, 'completed');
    await this.queue.clean(0, 0, 'failed');

    const totalTimeMs = Date.now() - start;
    const cpuAfter = await this.getRedisCpu();
    const memAfter = await this.getRedisMemory();

    return {
      strategy: 'Queue.clean()',
      totalTimeMs,
      redisCpuDelta: cpuAfter - cpuBefore,
      memoryFreedBytes: memBefore - memAfter,
    };
  }

  async benchmarkScanBased(batchSize = 100): Promise<BenchmarkResult> {
    const cpuBefore = await this.getRedisCpu();
    const memBefore = await this.getRedisMemory();
    const start = Date.now();

    const prefix = `bull:${this.queue.name}:`;
    for (const setKey of [`${prefix}completed`, `${prefix}failed`]) {
      let cursor = '0';
      do {
        const [nextCursor, members] = await this.connection.zscan(
          setKey, cursor, 'COUNT', batchSize,
        );
        cursor = nextCursor;

        const ids: string[] = [];
        for (let i = 0; i < members.length; i += 2) {
          ids.push(members[i]);
        }

        if (ids.length > 0) {
          const multi = this.connection.multi();
          multi.zrem(setKey, ...ids);
          multi.unlink(...ids.map((id) => `${prefix}${id}`));
          multi.unlink(...ids.map((id) => `${prefix}${id}:logs`));
          await multi.exec();
        }
      } while (cursor !== '0');
    }

    const totalTimeMs = Date.now() - start;
    const cpuAfter = await this.getRedisCpu();
    const memAfter = await this.getRedisMemory();

    return {
      strategy: 'SCAN-based',
      totalTimeMs,
      redisCpuDelta: cpuAfter - cpuBefore,
      memoryFreedBytes: memBefore - memAfter,
    };
  }

  private async getRedisCpu(): Promise<number> {
    const info = await this.connection.info('CPU');
    const match = info.match(/used_cpu_sys:([0-9.]+)/);
    return match ? parseFloat(match[1]) : 0;
  }

  private async getRedisMemory(): Promise<number> {
    const info = await this.connection.info('MEMORY');
    const match = info.match(/used_memory:(\d+)/);
    return match ? parseInt(match[1], 10) : 0;
  }
}

Expected benchmark results (100K jobs)

Strategy	Time (100K jobs)	Redis CPU Δ	Blocking?	Real-time?
Queue.clean()	~4-8s	Moderate (+5-10%)	Yes (O(N))	No
Event-driven	~0.1s per job	Low (+1-3%)	No	Yes
SCAN-based (batch=100)	~15-30s	Low (+2-5%)	Per-batch only	No
EXPIRE-based	~0s (deferred)	Very low	No	Deferred

The numbers tell a clear story: clean() is fastest in wall-clock time for bulk deletion but blocks Redis during execution. Event-driven is the hands-down winner for real-time systems but requires an idempotency strategy. SCAN-based is the safest choice for large queues where any blocking is unacceptable.

Production recommendation: Use event-driven as the primary strategy for real-time precision, SCAN-based as a safety net with a 2x interval, and EXPIRE on job data keys as a last-resort guarantee.

---

Section 6: Multi-Queue Cleanup Orchestration

In production systems, queues rarely exist in isolation. A typical pipeline might have ingest → enrich → transform → export, where each queue depends on the previous one. Cleanup must respect these dependencies — you should never delete from enrich before transform has finished processing, or you'll orphan dependent data.

The solution is a cleanup orchestrator that uses topological sorting to process queues in dependency order, with locking guardrails to prevent concurrent cleanup of the same queue.

interface QueueDependency {
  queueName: string;
  dependsOn: string[];
}

class CleanupOrchestrator {
  private connection: Redis;
  private queues: Map<string, Queue> = new Map();
  private dependencyGraph: Map<string, string[]> = new Map();

  constructor(
    private config: { dependencies: QueueDependency[] },
    connection: Redis,
  ) {
    this.connection = connection;

    for (const dep of config.dependencies) {
      this.queues.set(dep.queueName, new Queue(dep.queueName, { connection }));
      for (const parent of dep.dependsOn) {
        if (!this.dependencyGraph.has(parent)) {
          this.dependencyGraph.set(parent, []);
        }
        this.dependencyGraph.get(parent)!.push(dep.queueName);
      }
    }
  }

  async runFullCleanup(): Promise<string> {
    const runId = `cleanup-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`;
    const sorted = this.topologicalSort();
    if (sorted === null) {
      throw new Error('Circular dependency detected in queue cleanup graph');
    }

    for (const queueName of sorted) {
      const lockKey = `cleanup:lock:${queueName}`;

      // Guardrail: prevent double-cleanup with SETNX lock
      const acquired = await this.connection.setnx(lockKey, runId);
      if (acquired === 0) {
        throw new Error(`Queue "${queueName}" is already being cleaned`);
      }

      // Lock TTL prevents persistence if orchestrator crashes
      await this.connection.expire(lockKey, 3600);

      try {
        const queue = this.queues.get(queueName)!;
        await this.sweepQueue(queue);
        console.log(`Queue "${queueName}" cleaned in run ${runId}`);
      } finally {
        // Only release the lock if we still hold it
        const currentLock = await this.connection.get(lockKey);
        if (currentLock === runId) {
          await this.connection.del(lockKey);
        }
      }
    }

    return runId;
  }

  private async sweepQueue(queue: Queue): Promise<void> {
    const prefix = `bull:${queue.name}:`;
    for (const setSuffix of ['completed', 'failed']) {
      let cursor = '0';
      do {
        const [nextCursor, members] = await this.connection.zscan(
          `${prefix}${setSuffix}`, cursor, 'COUNT', 100,
        );
        cursor = nextCursor;

        const ids: string[] = [];
        for (let i = 0; i < members.length; i += 2) {
          ids.push(members[i]);
        }

        if (ids.length > 0) {
          const multi = this.connection.multi();
          multi.zrem(`${prefix}${setSuffix}`, ...ids);
          multi.unlink(...ids.map((id) => `${prefix}${id}`));
          multi.unlink(...ids.map((id) => `${prefix}${id}:logs`));
          await multi.exec();
        }
      } while (cursor !== '0');
    }
  }

  private topologicalSort(): string[] | null {
    const inDegree = new Map<string, number>();
    const adj = new Map<string, string[]>();

    for (const dep of this.config.dependencies) {
      if (!inDegree.has(dep.queueName)) inDegree.set(dep.queueName, 0);
      if (!adj.has(dep.queueName)) adj.set(dep.queueName, []);

      for (const parent of dep.dependsOn) {
        if (!inDegree.has(parent)) inDegree.set(parent, 0);
        if (!adj.has(parent)) adj.set(parent, []);
        adj.get(parent)!.push(dep.queueName);
        inDegree.set(dep.queueName, (inDegree.get(dep.queueName) ?? 0) + 1);
      }
    }

    const queue: string[] = [];
    for (const [node, deg] of inDegree) {
      if (deg === 0) queue.push(node);
    }

    const result: string[] = [];
    while (queue.length > 0) {
      const node = queue.shift()!;
      result.push(node);
      for (const neighbor of adj.get(node) ?? []) {
        const newDeg = (inDegree.get(neighbor) ?? 1) - 1;
        inDegree.set(neighbor, newDeg);
        if (newDeg === 0) queue.push(neighbor);
      }
    }

    return result.length === inDegree.size ? result : null;
  }

  async close(): Promise<void> {
    for (const queue of this.queues.values()) {
      await queue.close();
    }
  }
}

// Usage: ingest → enrich → transform → export pipeline
const orchestrator = new CleanupOrchestrator(
  {
    dependencies: [
      { queueName: 'ingest', dependsOn: [] },
      { queueName: 'enrich', dependsOn: ['ingest'] },
      { queueName: 'transform', dependsOn: ['enrich'] },
      { queueName: 'export', dependsOn: ['transform'] },
    ],
  },
  redis,
);

const runId = await orchestrator.runFullCleanup();

The orchestrator implements three guardrails:

1. SETNX locks per queue — prevents two orchestrators from cleaning the same queue simultaneously
2. Lock TTL — if the orchestrator crashes mid-cleanup, the lock expires after 1 hour
3. Circular dependency detection — the topological sort fails fast if the dependency graph has cycles

---

Conclusion: Building Your Cleanup Stack

No single cleanup strategy fits all queues. Your choice depends on throughput, job size, retention requirements, and Redis cluster sensitivity. Instead of picking one, layer them:

Primary:  Event-driven (QueueEvents)  →  Real-time, per-job precision
Safety:   SCAN-based sweep             →  Catches missed jobs at a 2x interval
Fallback: EXPIRE on job data keys      →  Last-resort, zero-code data expiration
Archival: Tiered pipeline              →  Long-term retention without Redis bloat
Orchestration: Topological ordering    →  Safe cleanup across dependent queues

Your action checklist:

• Measure current clean() latency at peak queue size using redis-cli --latency
• Deploy event-driven cleanup with QueueEvents as your primary strategy
• Implement a SCAN-based safety sweep at 2x the cleanup interval
• Add EXPIRE on job data keys as a last-resort TTL
• Set up cleanup locks for multi-queue environments
• Benchmark your chosen strategy using the harness in Section 5
• Monitor cleanup throughput in your observability dashboard

Cleanup is a systems design problem, not a one-liner API call. The strategies in this post give you the tools to design a cleanup architecture that scales with your queue — from hundreds of jobs to millions.

---

Want to see your cleanup strategy in action? Queue Hub provides real-time visibility into queue size, job retention, and Redis memory usage. Set up cleanup policies and monitor their impact from a single dashboard. Try Queue Hub today.