Automating Redis Queue Health Responses — Self-Healing and Auto-Remediation for BullMQ

Your pager goes off at 3:14 AM on a Tuesday. Redis memory is at 97%. Workers are stalling. Jobs are piling up. You SSH in, run CONFIG SET maxmemory-policy noeviction, drain a few queues, restart the worker pod, and go back to bed. Five hours later, the same thing happens again.

This is the manual remediation death spiral — and it's the norm for most BullMQ deployments. Monitoring tools will happily wake you up for every eviction, every OOM, every stalled worker, but they won't lift a finger to fix anything. The result? Burned-out on-call engineers, avoidable SLA violations, and Redis bills that climb because nobody scales down at 3 AM.

This post changes that. We're building a closed-loop automation pipeline that takes every health signal from your Redis-backed BullMQ queues and turns it into an immediate, safe, reversible remediation — no human in the loop. You'll walk away with production-ready TypeScript code for a health-check daemon, Kubernetes auto-scaling with KEDA, cloud provider auto-remediation (AWS, Azure, Upstash), and a webhook-driven self-healing layer that integrates with PagerDuty and Slack.

---

1. The Metrics-to-Actions Pipeline: Close the Loop

Every self-healing system follows the same control loop. Don't overthink it — four stages, one cycle:

Collect -> Evaluate -> Decide -> Act -> Log -> Verify

+----------+    +--------------+    +------------+    +-----------+
| COLLECT  |--->|  EVALUATE    |--->|  DECIDE    |--->|  ACT      |
|          |    |              |    |            |    |           |
| Redis    |    | Threshold    |    | Select     |    | CLIENT    |
| INFO     |    | comparison   |    | remediation|    | KILL      |
| BullMQ   |    | Rate-of-     |    | strategy   |    | CONFIG    |
| getJobs  |    | change check |    | Cooldown   |    | SET       |
| OS/proc  |    | Anomaly      |    | guard      |    | K8s API   |
+----------+    +--------------+    +------------+    +-----------+
                                                             |
                                                             v
                                                      +-----------+
                                                      | VERIFY    |
                                                      | + LOG     |
                                                      +-----------+

The key difference between a monitoring dashboard and a self-healing system is the decision and action stages. A dashboard collects and evaluates, then pages a human. A self-healing system collects, evaluates, decides, acts, and verifies — all in under 30 seconds.

1.1 Metrics Collection

Poll Redis every N seconds. You need four data categories:

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

interface CollectedMetrics {
  memory: { used: number; max: number; percent: number; rss: number };
  evictions: { total: number; delta: number };
  clients: { connected: number; byFlags: Record<string, number> };
  queues: Map<string, { waiting: number; active: number; delayed: number; failed: number }>;
  latency: number; // P99 in ms
}

async function collectAllMetrics(
  redis: IORedis,
  queueNames: string[],
  lastEvictedKeys: number
): Promise<CollectedMetrics> {
  const info = await redis.info('all');

  const usedMemory = parseInt(info.match(/used_memory:(\d+)/)?.[1] ?? '0', 10);
  const maxmemory = parseInt(info.match(/maxmemory:(\d+)/)?.[1] ?? '0', 10);
  const evictedTotal = parseInt(info.match(/evicted_keys:(\d+)/)?.[1] ?? '0', 10);
  const connected = parseInt(info.match(/connected_clients:(\d+)/)?.[1] ?? '0', 10);

  const queueMetrics = new Map<string, { waiting: number; active: number; delayed: number; failed: number }>();
  for (const name of queueNames) {
    const q = new Queue(name, { connection: redis });
    const counts = await q.getJobCounts();
    queueMetrics.set(name, {
      waiting: counts.waiting ?? 0,
      active: counts.active ?? 0,
      delayed: counts.delayed ?? 0,
      failed: counts.failed ?? 0,
    });
  }

  return {
    memory: {
      used: usedMemory,
      max: maxmemory,
      percent: maxmemory > 0 ? usedMemory / maxmemory : 0,
      rss: parseInt(info.match(/used_memory_rss:(\d+)/)?.[1] ?? '0', 10),
    },
    evictions: { total: evictedTotal, delta: Math.max(0, evictedTotal - lastEvictedKeys) },
    clients: {
      connected,
      byFlags: parseClientFlags(info),
    },
    queues: queueMetrics,
    latency: await measureLatency(redis),
  };
}

1.2 Evaluation Engine

Feed metrics through a rules engine. Each rule produces a signal with severity, suggested action, and cooldown:

interface RemediationSignal {
  severity: 'info' | 'warning' | 'critical';
  ruleName: string;
  currentValue: number;
  threshold: number;
  suggestedAction: string;
  lastFired: number; // timestamp -- for cooldown enforcement
  dryRun: boolean;
}

function evaluateRules(
  metrics: CollectedMetrics,
  state: { lastFired: Record<string, number>; lastEvicted: number; cooldowns: Record<string, number> },
  dryRun: boolean
): RemediationSignal[] {
  const now = Date.now();
  const signals: RemediationSignal[] = [];
  const cf = state.cooldowns;

  // Rule 1: Memory pressure
  if (metrics.memory.percent > 0.80) {
    const rule = 'memory-pressure';
    if (now - (state.lastFired[rule] ?? 0) > (cf.memoryCleanup ?? 300_000)) {
      signals.push({
        severity: metrics.memory.percent > 0.95 ? 'critical' : 'warning',
        ruleName: rule,
        currentValue: metrics.memory.percent,
        threshold: 0.80,
        suggestedAction: 'cleanup-jobs',
        lastFired: state.lastFired[rule] ?? 0,
        dryRun,
      });
    }
  }

  // Rule 2: Eviction detected -- fire immediately (data loss in progress)
  if (metrics.evictions.delta > 0) {
    const rule = 'eviction-detected';
    signals.push({
      severity: 'critical',
      ruleName: rule,
      currentValue: metrics.evictions.delta,
      threshold: 0,
      suggestedAction: 'set-noeviction',
      lastFired: 0,
      dryRun,
    });
  }

  // Rule 3: Queue depth anomaly (sudden spike)
  for (const [name, qm] of metrics.queues) {
    const totalPending = qm.waiting + qm.delayed;
    if (totalPending > 10_000) {
      signals.push({
        severity: totalPending > 50_000 ? 'critical' : 'warning',
        ruleName: `queue-depth-${name}`,
        currentValue: totalPending,
        threshold: 10_000,
        suggestedAction: 'scale-workers',
        lastFired: state.lastFired[`queue-depth-${name}`] ?? 0,
        dryRun,
      });
    }
  }

  return signals.filter(s => dryRun || (now - s.lastFired > (state.cooldowns[s.ruleName] ?? 60_000)));
}

---

2. Auto-Remediation Playbook

A remediation is just a function that takes a signal and returns a result. Here's the playbook for six common failure modes.

2.1 Memory Pressure -- Controlled Evacuation

When memory hits 80%, don't wait for the OOM killer. Proactively drain the oldest completed and failed jobs from your largest queues. At 95%, escalates to cloud API calls that resize the Redis instance.

async function remediateMemoryPressure(
  redis: IORedis,
  queueNames: string[],
  targetFreePercent: number = 0.75
): Promise<{ queue: string; removed: number; freedBytes: number }[]> {
  const results: { queue: string; removed: number; freedBytes: number }[] = [];

  for (const name of queueNames) {
    const queue = new Queue(name, { connection: redis });
    const counts = await queue.getJobCounts();
    const totalRetained = (counts.completed ?? 0) + (counts.failed ?? 0);
    if (totalRetained === 0) continue;

    const toRemove = Math.ceil(totalRetained * targetFreePercent);
    const completed = await queue.getJobs(['completed'], 0, toRemove);
    const failed = await queue.getJobs(['failed'], 0, toRemove);
    const ids = [...completed.map(j => j.id!), ...failed.map(j => j.id!)];

    if (ids.length > 0) {
      const before = await redis.info('memory');
      await queue.remove(ids);
      const after = await redis.info('memory');
      const beforeMem = parseInt(before.match(/used_memory:(\d+)/)?.[1] ?? '0', 10);
      const afterMem = parseInt(after.match(/used_memory:(\d+)/)?.[1] ?? '0', 10);

      results.push({
        queue: name,
        removed: ids.length,
        freedBytes: beforeMem - afterMem,
      });
    }
  }

  return results;
}

2.2 Stalled Workers -- CLIENT KILL and Respawn

BullMQ detects stalled jobs via missed heartbeats, but it doesn't kill the zombie worker connection. That's your job -- literally.

interface StalledWorker {
  id: string;
  queue: string;
  redisClientId: string;
  hostname: string;
  idleMs: number;
}

async function killStalledWorkers(
  redis: IORedis,
  allowlist: Set<string> = new Set(['health-daemon-conn', 'queuehub-agent']),
  idleThresholdMs: number = 60_000
): Promise<StalledWorker[]> {
  const clientList: string = await redis.call('CLIENT', 'LIST');
  const lines = clientList.split('\n').filter(Boolean);
  const killed: StalledWorker[] = [];

  for (const line of lines) {
    const idMatch = line.match(/id=(\d+)/);
    const nameMatch = line.match(/name=([^\s]+)/);
    const idleMatch = line.match(/idle=(\d+)/);
    const flagsMatch = line.match(/flags=([^\s]+)/);

    if (!idMatch || !nameMatch || !idleMatch) continue;

    const clientName = nameMatch[1];
    const clientId = idMatch[1];
    const idleSeconds = parseInt(idleMatch[1], 10);

    if (allowlist.has(clientName)) continue;

    const flags = flagsMatch?.[1] ?? '';
    if (!flags.includes('b') && idleSeconds * 1000 < idleThresholdMs) continue;

    await redis.call('CLIENT', 'KILL', 'ID', clientId);
    killed.push({
      id: clientName,
      queue: extractQueueFromClientName(clientName),
      redisClientId: clientId,
      hostname: line.match(/addr=([^\s]+)/)?.[1] ?? 'unknown',
      idleMs: idleSeconds * 1000,
    });
  }

  return killed;
}

function extractQueueFromClientName(name: string): string {
  const match = name.match(/^bull-(.+?)-worker/);
  return match?.[1] ?? name;
}

After killing a stalled worker, trigger a respawn via your orchestrator:

async function respawnWorker(deploymentName: string, namespace: string = 'default'): Promise<void> {
  // Kubernetes: delete pod so ReplicaSet recreates it
  // kubectl delete pod -l app=bullmq-worker-${deploymentName}

  // systemd: SSH target and systemctl restart bullmq-worker@${deploymentName}

  // Docker Compose: docker compose up -d --scale ${deploymentName}=<current+1>

  console.log(`[AUTO-REMEDIATION] Triggered respawn for ${deploymentName}`);
}

2.3 Eviction Safety -- Instant noeviction Lockdown

Redis evictions in a BullMQ context are data corruption events -- they silently delete job payloads, worker heartbeats, and rate-limit counters. The only safe response is an immediate switch to noeviction policy, followed by a maxmemory increase.

class EvictionSafetyGuard {
  private lastEvictedKeys = 0;

  constructor(
    private redis: IORedis,
    private maxmemoryBufferPercent: number = 0.15
  ) {}

  async checkAndRemediate(dryRun: boolean): Promise<boolean> {
    const stats = await this.redis.info('stats');
    const match = stats.match(/evicted_keys:(\d+)/);
    if (!match) return false;

    const current = parseInt(match[1], 10);
    const delta = current - this.lastEvictedKeys;
    this.lastEvictedKeys = current;

    if (delta <= 0) return false;

    if (!dryRun) {
      await this.redis.config('SET', 'maxmemory-policy', 'noeviction');
      await this.redis.config('REWRITE');

      const info = await this.redis.info('memory');
      const currentMem = parseInt(info.match(/used_memory:(\d+)/)?.[1] ?? '0', 10);
      const newMax = Math.round(currentMem * (1 + this.maxmemoryBufferPercent));
      await this.redis.config('SET', 'maxmemory', String(newMax));
    }

    console.log(
      dryRun
        ? `[DRY-RUN] Would switch to noeviction (evicted delta=${delta})`
        : `[AUTO-REMEDIATION] Switched to noeviction, increased maxmemory`
    );

    return true;
  }
}

2.4 Latency Spike -- Dynamic Queue Splitting

When P99 latency exceeds 500ms for three consecutive cycles, the cause is usually one slow job type blocking the entire queue. Auto-split:

async function splitQueueByLatency(
  redis: IORedis,
  originalName: string,
  classifyFn: (jobData: Record<string, unknown>) => 'fast' | 'slow'
): Promise<{ fastLane: string; slowLane: string }> {
  const fastName = `${originalName}-fast`;
  const slowName = `${originalName}-slow`;

  const original = new Queue(originalName, { connection: redis });
  const fast = new Queue(fastName, { connection: redis });
  const slow = new Queue(slowName, { connection: redis });

  const waiting = await original.getJobs(['waiting']);
  for (const job of waiting) {
    const lane = classifyFn(job.data as Record<string, unknown>);
    const target = lane === 'fast' ? fast : slow;
    await target.add(job.name, job.data, job.opts);
    await job.remove();
  }

  return { fastLane: fastName, slowLane: slowName };
}

---

3. Building the Health-Check Daemon

Wrap the playbook into a standalone HealthCheckDaemon class -- a single process that runs alongside your workers. Four responsibilities: poll, evaluate, act, log.

3.1 Complete Daemon Skeleton

interface DaemonConfig {
  connection: IORedis.RedisOptions;
  pollIntervalMs: number;
  dryRun: boolean;
  queues: string[];
  thresholds: {
    memoryPercent: number;
    evictionDelta: number;
    latencyP99Ms: number;
    queueDepthWarning: number;
    queueDepthCritical: number;
    stalledWorkerIdleMs: number;
  };
  cooldowns: {
    memoryCleanupMs: number;
    workerKillMs: number;
    queueSplitMs: number;
  };
  actions: {
    enabled: {
      jobCleanup: boolean;
      killStalledWorkers: boolean;
      evictionSafety: boolean;
      queueSplit: boolean;
      scaleWorkers: boolean;
    };
  };
}

class HealthCheckDaemon {
  private redis: IORedis;
  private state = {
    lastFired: {} as Record<string, number>,
    lastEvictedKeys: 0,
    consecutiveLatencySpikes: 0,
  };
  private timer: ReturnType<typeof setInterval> | null = null;

  constructor(private config: DaemonConfig) {
    this.redis = new IORedis(config.connection);
  }

  async start(): Promise<void> {
    console.log(`[DAEMON] Starting -- poll every ${this.config.pollIntervalMs}ms, dryRun=${this.config.dryRun}`);
    await this.cycle();
    this.timer = setInterval(() => this.cycle(), this.config.pollIntervalMs);
  }

  stop(): void {
    if (this.timer) clearInterval(this.timer);
    this.redis.disconnect();
  }

  private async cycle(): Promise<void> {
    const start = Date.now();
    try {
      const metrics = await collectAllMetrics(this.redis, this.config.queues, this.state.lastEvictedKeys);
      this.state.lastEvictedKeys = metrics.evictions.total;

      const signals = evaluateRules(metrics, this.state, this.config.dryRun);
      for (const signal of signals) {
        await this.dispatch(signal);
        this.state.lastFired[signal.ruleName] = Date.now();
      }

      console.log(`[DAEMON] Cycle: ${signals.length} signal(s), ${Date.now() - start}ms`);
    } catch (err) {
      console.error('[DAEMON] Cycle failed:', err);
    }
  }

  private async dispatch(signal: RemediationSignal): Promise<void> {
    const entry = { timestamp: new Date().toISOString(), ...signal };
    console.log(signal.dryRun ? `[DRY-RUN] ${JSON.stringify(entry)}` : `[ACTION] ${JSON.stringify(entry)}`);
    if (signal.dryRun) return;

    switch (signal.suggestedAction) {
      case 'cleanup-jobs':
        await remediateMemoryPressure(this.redis, this.config.queues);
        break;
      case 'set-noeviction':
        const guard = new EvictionSafetyGuard(this.redis);
        await guard.checkAndRemediate(false);
        break;
      case 'kill-stalled-workers':
        await killStalledWorkers(this.redis);
        break;
      case 'scale-workers':
        // Forward to KEDA / K8s API
        break;
    }
  }
}

3.2 Health Endpoint (for K8s Probes)

import http from 'http';

function createHealthEndpoint(daemon: HealthCheckDaemon): http.Server {
  return http.createServer((req, res) => {
    if (req.url === '/health') {
      res.writeHead(200, { 'Content-Type': 'application/json' });
      res.end(JSON.stringify({
        status: 'ok',
        uptime: process.uptime(),
        lastCycle: Date.now(),
      }));
    } else {
      res.writeHead(404);
      res.end();
    }
  }).listen(9090);
}

---

4. Kubernetes: HPA + KEDA for Queue-Based Auto-Scaling

Monitoring is reactive. Scaling is proactive. When queue depth spikes, you don't want a human to increase the replica count -- you want the cluster to do it automatically.

4.1 KEDA -- The Recommended Path

KEDA (Kubernetes Event-Driven Autoscaler) natively supports Redis list length as a scaling trigger. BullMQ stores waiting jobs in Redis lists, making KEDA a perfect fit:

apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: bullmq-email-scaler
spec:
  scaleTargetRef:
    name: bullmq-email-worker
  pollingInterval: 15
  cooldownPeriod: 120
  minReplicaCount: 2
  maxReplicaCount: 30
  triggers:
    - type: redis
      metadata:
        address: REDIS_SERVICE:6379
        passwordFromEnv: REDIS_PASSWORD
        listLength: "200"
        listName: "bull:email:wait"
        enableTLS: "false"

BullMQ Redis key patterns for KEDA:

Queue State	Redis Key	KEDA Trigger
Waiting	bull:{name}:wait	listLength
Delayed	bull:{name}:delayed	listLength (if using ZSET adapter)
Failed (backlog)	bull:{name}:failed	listLength

4.2 Custom Metrics HPA (No KEDA)

If KEDA isn't an option, deploy a sidecar metrics adapter that reads BullMQ queue depth and exposes it as a Prometheus gauge:

import IORedis from 'ioredis';
import { Queue } from 'bullmq';
import express from 'express';

const app = express();
const redis = new IORedis({ host: process.env.REDIS_HOST });

app.get('/metrics', async (_req, res) => {
  const queue = new Queue(process.env.QUEUE_NAME ?? 'default', { connection: redis });
  const counts = await queue.getJobCounts();
  const depth = (counts.waiting ?? 0) + (counts.delayed ?? 0);

  const body = [
    '# HELP bullmq_queue_depth Total pending jobs in queue',
    '# TYPE bullmq_queue_depth gauge',
    `bullmq_queue_depth{queue="${process.env.QUEUE_NAME}"} ${depth}`,
  ].join('\n');

  res.set('Content-Type', 'text/plain');
  res.end(body);
});

app.listen(8080);

Then reference the custom metric in your HPA:

apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: bullmq-worker-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: bullmq-worker
  minReplicas: 2
  maxReplicas: 30
  metrics:
    - type: Pods
      pods:
        metric:
          name: bullmq_queue_depth
        target:
          type: AverageValue
          averageValue: "50"

4.3 Direct K8s API Scaling from the Daemon

The health daemon can also scale deployments directly when a queue depth signal fires:

import { KubeConfig, AppsV1Api } from '@kubernetes/client-node';

async function scaleDeployment(
  namespace: string,
  name: string,
  replicas: number
): Promise<void> {
  const kc = new KubeConfig();
  kc.loadFromDefault();
  const api = kc.makeApiClient(AppsV1Api);

  await api.patchNamespacedDeploymentScale(name, namespace, {
    spec: { replicas },
  }, undefined, undefined, undefined, undefined, {
    headers: { 'Content-Type': 'application/merge-patch+json' },
  });

  console.log(`[K8S] Scaled ${name} to ${replicas} replicas`);
}

---

5. Cloud Provider Auto-Remediation

When Redis-level actions aren't enough (e.g., the entire cluster needs a node reboot for activedefrag), escalate to cloud provider APIs.

5.1 AWS ElastiCache -- Reboot or Resize

import { ElastiCacheClient, RebootCacheClusterCommand } from '@aws-sdk/client-elasticache';

async function rebootElastiCache(clusterId: string, region: string): Promise<void> {
  const client = new ElastiCacheClient({ region });
  await client.send(new RebootCacheClusterCommand({
    CacheClusterId: clusterId,
    CacheNodeIdsToReboot: ['0001'],
  }));
  console.log(`[AWS] Rebooted ${clusterId}`);
}

5.2 Upstash Redis -- Scale Memory via REST API

async function upstashScaleDatabase(databaseId: string, newMemoryMb: number, token: string): Promise<void> {
  const res = await fetch(`https://api.upstash.com/v2/redis/database/${databaseId}`, {
    method: 'PATCH',
    headers: {
      Authorization: `Bearer ${token}`,
      'Content-Type': 'application/json',
    },
    body: JSON.stringify({ maxMemory: newMemoryMb }),
  });
  if (!res.ok) throw new Error(`Upstash scale failed: ${res.status}`);
  console.log(`[UPSTASH] Scaled ${databaseId} to ${newMemoryMb}MB`);
}

---

6. Webhook-Driven Self-Healing

The daemon doesn't have to act alone. Route signals through webhooks for PagerDuty runbook automation or Slack incident commands.

6.1 PagerDuty Automation Action Bridge

import express from 'express';

const app = express();
app.use(express.json());

app.post('/webhook/pagerduty', async (req, res) => {
  const event = req.body;
  console.log(`[PD] Received: ${event.event_type} -- ${event.incident?.title}`);

  if (event.event_type === 'incident.triggered') {
    const title = event.incident.title.toLowerCase();
    if (title.includes('redis memory')) {
      await remediateMemoryPressure(/* ... */);
    } else if (title.includes('stalled worker')) {
      await killStalledWorkers(/* ... */);
    } else if (title.includes('eviction')) {
      await new EvictionSafetyGuard(/* ... */).checkAndRemediate(false);
    }
  }

  res.status(200).json({ received: true });
});

app.listen(8080);

6.2 Slack Incident Command with Remediation Buttons

// In your Slack slash command handler:
app.post('/slack/incident', async (req, res) => {
  const { text, channel_id, user_id } = req.body;
  const [incidentType, target] = text.split(' ');

  res.json({
    response_type: 'ephemeral',
    blocks: [
      {
        type: 'section',
        text: { type: 'mrkdwn', text: `*Incident:* ${incidentType}\n*Target:* ${target}\n*Reported by:* <@${user_id}>` },
      },
      {
        type: 'actions',
        elements: [
          {
            type: 'button',
            text: { type: 'plain_text', text: 'Restart Worker' },
            value: JSON.stringify({ action: 'restart-worker', target }),
            action_id: 'remediate_restart',
          },
          {
            type: 'button',
            text: { type: 'plain_text', text: 'Force Clean Queue' },
            value: JSON.stringify({ action: 'clean-queue', target }),
            action_id: 'remediate_clean',
          },
        ],
      },
    ],
  });
});

---

7. Safe Deployment -- Don't Break Production

Auto-remediation is powerful, but it's also dangerous. A bug in your daemon can do more damage than the failure it's trying to fix. Follow these safety rules:

Rule	Implementation
Start in dry-run mode	Set dryRun: true for the first 2 weeks. Log every action the daemon would have taken.
One action at a time	Enable eviction-safety first (lowest blast radius). Enable worker-kill last (highest blast radius).
Circuit breaker	Max 3 actions per hour per rule. If exceeded, disable all auto-remediation and page the on-call.
Allowlist connections	Never CLIENT KILL your own daemon connection, QueueHub agent, or known monitoring tools.
Kill-switch webhook	POST /killswitch disables all auto-actions instantly. Document the curl command in your runbook.
Log everything	Every action, dry-run or real, gets a structured log entry with timestamp, rule, value, threshold, and outcome.

---

8. How QueueHub Powers Your Auto-Remediation Pipeline

QueueHub isn't just a dashboard for passive monitoring -- it's the remediation control plane for your auto-healing pipeline.

• Live Workers View -- See every connected worker, its Redis client ID, heartbeat age, and current job. This is the data your health daemon needs to identify CLIENT KILL targets and verify that respawned workers have joined.
• Agent Tunnel -- When Redis is behind a VPC or firewall, the QueueHub agent provides a WebSocket tunnel. Your health daemon connects through the same tunnel, so it can reach private Redis instances without VPN.
• Bulk Job Actions -- When auto-cleanup removes the wrong jobs, use QueueHub to promote, retry, or bulk-move jobs from the dead-letter queue back into processing.
• Multi-Backend -- A single daemon monitors BullMQ and BeeQueue simultaneously. One control loop, two queue engines.

The full pipeline:

+---------------------------+     +--------------+     +--------------------+
|  Private Redis            |---->| QueueHub     |---->| QueueHub Cloud     |
|  (VPC / ElastiCache)      |     | Agent        |     | (Live Workers)     |
+---------------------------+     +------+-------+     +----------+---------+
                                         |                          |
                                   +-----v-------+            +-----v-----------+
                                   |  Health      |<-----------| QueueHub API    |
                                   |  Daemon      |            | (worker data)   |
                                   +-----+-------+            +-----------------+
                                         |
                                   +-----v-------+
                                   |  Actions     |
                                   |  * CLIENT    |
                                   |    KILL      |
                                   |  * CONFIG    |
                                   |    SET       |
                                   |  * Cleanup   |
                                   |  * K8s       |
                                   |    scale     |
                                   +-------------+

Ready to close the loop on your queue health? Try QueueHub free and start building your auto-remediation pipeline today. Your 3 AM self will thank you.