The Reliability Trap of Redis Pub/Sub

Many engineering teams reach for Redis Pub/Sub when they need real-time message broadcasting between microservices. While Pub/Sub offers blazing speed and minimal memory overhead, it operates on a strict 'fire-and-forget' model. If a subscriber disconnects due to a deployment, GC pause, or network partition, messages published during that window evaporate permanently. There is no persistence, no concept of consumer acknowledgment, and no replay mechanism.

Redis Streams (introduced in Redis 5.0) resolves this fundamental gap by providing an append-only, disk-persisted log structure modeled after Apache Kafka, but with sub-millisecond in-memory access speeds. However, simply switching from PUBLISH to XADD is not enough. To build production-grade event processors, you must master consumer groups, manage the Pending Entries List (PEL), and engineer resilient dead-letter mechanisms.

Stream Anatomy: Entries, Groups, and the PEL

When you append events using XADD, Redis assigns a monotonically increasing entry ID composed of a millisecond timestamp and a sequence number (e.g., 1712839201942-0). When you read these entries via a consumer group using XREADGROUP, Redis tracks message delivery state via two core internal data structures:

  • Last Delivered ID: A high-water mark pointer indicating the latest stream entry delivered to any consumer in the group.
  • Pending Entries List (PEL): A per-consumer tracking table recording every message that has been fetched but not yet acknowledged via XACK. Each record stores the message ID, consumer name, delivery count, and idle time.

If a worker crashes while processing an entry, the message remains locked in that worker's PEL. Unless another worker actively reclaims it, that event will stall indefinitely.

Step 1: Ingesting Events with Controlled Stream Growth

Unbounded streams will eventually exhaust RAM. When appending events, use the approximate trimming modifier (~) with MAXLEN. This avoids expensive reallocations by trimming entire radix tree nodes rather than individual entries:

// Append order event and trim stream to roughly 100,000 entries
XADD orders_stream MAXLEN ~ 100000 * order_id 98421 user_id 4012 amount 149.99 currency USD status CREATED

Next, initialize the consumer group. Setting the start ID to $ instructs the group to only process events arriving after group creation, while 0 replays the entire stream history:

// Create group 'order_processors' reading from the end of 'orders_stream'
// MKSTREAM automatically creates the stream key if it does not exist
XGROUP CREATE orders_stream order_processors $ MKSTREAM

Step 2: Building the Resilient Consumer Loop

A production worker must handle two distinct queues of work: fetching new unread messages and reclaiming abandoned messages from failed workers. The following Node.js implementation using ioredis demonstrates this dual-phase lifecycle:

import Redis from "ioredis";

const redis = new Redis({ host: "127.0.0.1", port: 6379 });
const STREAM = "orders_stream";
const GROUP = "order_processors";
const CONSUMER_NAME = `worker-${process.pid}`;
const MAX_RETRIES = 3;
const DLQ_STREAM = "orders_stream_dlq";

async function processOrder(messageId, payload) {
  // Business logic simulation
  if (Math.random() < 0.05) {
    throw new Error("Simulated database transient error");
  }
  console.log(`[${CONSUMER_NAME}] Processed ${messageId}:`, payload);
}

async function handleDeadLetter(messageId, payload, error) {
  await redis.xadd(
    DLQ_STREAM,
    "*",
    "original_id", messageId,
    "error", error.message,
    "failed_at", Date.now().toString(),
    "payload", JSON.stringify(payload)
  );
  await redis.xack(STREAM, GROUP, messageId);
  console.error(`[DLQ] Quarantined poisoned entry ${messageId}`);
}

async function reclaimStaleMessages() {
  const MIN_IDLE_TIME_MS = 30000; // 30 seconds idle
  let cursor = "0-0";

  while (true) {
    // XAUTOCLAIM safely transfers ownership of idle pending entries
    const [nextCursor, entries] = await redis.xautoclaim(
      STREAM,
      GROUP,
      CONSUMER_NAME,
      MIN_IDLE_TIME_MS,
      cursor,
      "COUNT", 50
    );

    for (const [id, fields] of entries) {
      const payload = Object.fromEntries(
        fields.reduce((acc, cur, i, arr) => {
          if (i % 2 === 0) acc.push([cur, arr[i + 1]]);
          return acc;
        }, [])
      );

      // Check delivery count using XPENDING
      const [[, , , deliveryCount]] = await redis.xpending(
        STREAM, GROUP, id, id, 1
      );

      if (deliveryCount > MAX_RETRIES) {
        await handleDeadLetter(id, payload, new Error("Max retry limit exceeded"));
        continue;
      }

      try {
        await processOrder(id, payload);
        await redis.xack(STREAM, GROUP, id);
      } catch (err) {
        console.warn(`[RECLAIM-FAIL] Delivery ${deliveryCount} failed for ${id}`);
      }
    }

    cursor = nextCursor;
    if (cursor === "0-0") break;
  }
}

async function startConsumer() {
  console.log(`Starting consumer ${CONSUMER_NAME}...`);

  // Periodically check for abandoned messages
  setInterval(reclaimStaleMessages, 10000);

  while (true) {
    try {
      // Read only new messages targeting this group
      const response = await redis.xreadgroup(
        "GROUP", GROUP, CONSUMER_NAME,
        "BLOCK", 2000,
        "COUNT", 10,
        "STREAMS", STREAM,
        ">"
      );

      if (!response) continue;

      const [, messages] = response[0];
      for (const [id, fields] of messages) {
        const payload = Object.fromEntries(
          fields.reduce((acc, cur, i, arr) => {
            if (i % 2 === 0) acc.push([cur, arr[i + 1]]);
            return acc;
          }, [])
        );

        try {
          await processOrder(id, payload);
          await redis.xack(STREAM, GROUP, id);
        } catch (err) {
          console.error(`Processing error on ${id}: ${err.message}`);
          // Do not XACK; message remains in PEL for retry or XAUTOCLAIM
        }
      }
    } catch (loopError) {
      console.error("Consumer loop error:", loopError);
      await new Promise((r) => setTimeout(r, 1000));
    }
  }
}

startConsumer();

Key Architectural Mechanics

1. The Special ID Symbols

When interacting with consumer groups, two ID operators have distinct semantics:

  • >: Requests entries that have never been delivered to any consumer in this consumer group.
  • 0 or valid IDs: Reads pending entries assigned to the calling consumer that have not been acknowledged.

2. XAUTOCLAIM vs. XCLAIM

Prior to Redis 6.2, reclaiming orphaned messages required a two-step cycle: querying XPENDING to discover idle IDs, followed by issuing an XCLAIM command. Redis 6.2 introduced XAUTOCLAIM, which merges inspection, filtration by idle time, and ownership reassignment into an atomic O(1) operation per entry, drastically reducing network round-trips.

3. Dead-Letter Queues (DLQ)

Without an explicit retry ceiling, a malformed payload (poison pill) will crash your consumers repeatedly. By querying the delivery counter in the PEL via XPENDING, workers can detect when an entry has failed beyond the threshold (e.g., 3 attempts), route it to a secondary DLQ stream, and send an XACK to remove it from the active processing pipeline.

Production Failure Modes and Tuning

Warning: Forgetting to execute XACK does not break immediate reads, but it inflates Redis memory over time. Each unacknowledged message adds metadata to the group's internal PEL radix tree.

Keep these operational rules in mind before deploying to production:

  1. Monitor PEL Growth: Track the metric via XINFO GROUPS orders_stream. A continuously climbing pending count indicates crashing workers, slow execution, or missing XACK calls.
  2. Graceful Shutdown: Listen for SIGTERM signals. Allow in-flight handlers to finish and invoke XACK before terminating the process to avoid unnecessary XAUTOCLAIM churn.
  3. Idempotency is Mandatory: Because Redis Streams guarantees at-least-once delivery in consumer group mode, duplicate deliveries will occur during worker crashes or reclaims. Ensure your downstream persistence uses idempotency keys or unique database constraints.

Key Takeaways

  • Redis Pub/Sub is best suited for ephemeral signals; Redis Streams provides durable event logs with consumer tracking.
  • The Pending Entries List (PEL) tracks in-flight messages; messages remain there until explicitly confirmed with XACK.
  • Use XAUTOCLAIM periodically to gracefully recover orphaned entries from ungraceful worker shutdowns.
  • Combine retry counters with a secondary Dead-Letter Queue stream to prevent poison pills from blocking your pipeline indefinitely.