redis-stream-client-go

Interactive demo: https://gemini.google.com/share/999b256cffd3

A Redis stream-based client with automatic failure recovery. Built on go-redis and redsync.

Problem

Standard Redis stream recovery via XCLAIM requires:

1. Crashed consumer to come back up and reclaim
2. Stuck consumers (GC pauses) block processing

Other available consumers can't help because Redis is pull-based—they don't know recovery is needed.

Solution

This library provides:

1. Load Balancer Stream (LBS) - Distribute streams across consumers via round-robin
2. Periodic reconciliation scan - Recover work from dead consumers by re-queuing it (XACK + XADD), after verifying the owner's lock has actually expired (so slow-but-alive consumers are never disturbed)
3. Keyspace notifications - A low-latency fast path that surfaces lock expirations as StreamExpired; calling Claim re-queues the orphaned stream for redistribution
4. Redis Cluster (OSS) support - Subscribe to keyspace notifications on every master, with the scan as the authoritative, cluster-safe recovery mechanism

Recovery model

Recovery is handled by a periodic reconciliation scan plus the keyspace-notification fast path:

• A background scan inspects the LBS group's pending entries (XPENDING) every ReconciliationInterval (with jitter). For each entry idle longer than MinIdleTime, it checks whether the owning consumer's lock key still exists:
  • lock present → the consumer is alive (just slow), so the message is left untouched (this prevents duplicate processing of slow consumers);
  • lock absent → the consumer is dead, so the message is recovered by acknowledging the original and re-adding it as a new message (XACK + XADD) for normal round-robin redistribution.
• The XACK happens first, so when several consumers try to recover the same message, exactly one wins and re-queues it.
• Each re-queue increments a _retry_count field on the message. Once it exceeds MaxRetries, the message is routed to the configured DLQStream (or dropped if none is set).

This XACK + XADD approach replaces XCLAIM/XAUTOCLAIM and is safe across multiple Redis Cluster shards, where keyspace notifications alone are not broadcast cluster-wide.

Cluster support

Select the mode with impl.WithClusterMode(...):

• impl.ClusterModeSingleShard (default) — single node, primary/replica, or Sentinel. Keyspace notifications are subscribed to on the connected node.
• impl.ClusterModeOSS — OSS Redis Cluster. Requires the underlying client to be a *redis.ClusterClient. The client enables keyspace notifications and subscribes on every master, and the reconciliation scan is the authoritative recovery mechanism. After a failover, or when a master is added or removed, call ReinitTopology(ctx) to re-initialize against the new topology (it fails with errs.ErrClusterNotSettled if the cluster is not settled yet; see Cluster Topology Changes for when to call it).

client, _ := impl.NewRedisStreamClient(
    clusterClient, "my-service",
    impl.WithClusterMode(impl.ClusterModeOSS),
    impl.WithRecoveryConfig(impl.RecoveryConfig{
        ReconciliationInterval: 60 * time.Second,
        MinIdleTime:            30 * time.Second,
        BatchSize:              50,
        MaxRetries:             3,
        DLQStream:              "my-service-dlq",
        DLQMaxLen:              10000, // approx MAXLEN cap on the DLQ; 0 => unbounded
    }),
)

Note: the reconciliation scan uses XPENDING ... IDLE, which requires Redis 6.2+.

Limitations

Redis consumer groups on a single stream are not supported. Each data stream is exclusively owned by one consumer at a time. If you need multiple consumers reading from the same stream in parallel, use XREADGROUP / XAUTOCLAIM directly via go-redis. See this issue for planned support.

Quick Start

client, _ := impl.NewRedisStreamClient(redisClient, "my-service")
outputChan, _ := client.Init(ctx)

for notification := range outputChan {
    switch notification.Type {
    case notifs.StreamAdded:
        go process(notification.Payload.DataStreamName)
    case notifs.StreamExpired:
        client.Claim(ctx, notification.Payload)

Metrics

The client can emit detailed operational metrics via the metrics.Recorder interface. Supply a recorder when creating the client with impl.WithMetricsRecorder(...). A Prometheus example is included under examples/prometheus/ and full metric semantics are documented in docs/METRICS.md.

rec := prometheusmetric.NewPrometheusRecorder(prometheus.DefaultRegisterer)
client, _ := impl.NewRedisStreamClient(redisClient, "my-service", impl.WithMetricsRecorder(rec))

Architecture

Threading Model

The library spawns multiple goroutines to handle concurrent operations:

┌─────────────────────────────────────────────────────────────────┐
│                         Client Instance                         │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  ┌─────────────────┐                                            │
│  │ LBS Reader      │  1 goroutine - reads from Load Balancer    │
│  │ (blocking read) │  Stream, assigns streams to this consumer  │
│  └─────────────────┘                                            │
│                                                                 │
│  ┌─────────────────┐                                            │
│  │ Keyspace        │  1 goroutine - listens for Redis key       │
│  │ Listener        │  expiration events (pub/sub)               │
│  └─────────────────┘                                            │
│                                                                 │
│  ┌─────────────────┐                                            │
│  │ Key Extender    │  N goroutines - one per active stream      │
│  │ (stream-1)      │  extends distributed lock every hbInterval │
│  ├─────────────────┤                                            │
│  │ Key Extender    │  Goroutines exit when:                     │
│  │ (stream-2)      │  - DoneStream() called                     │
│  ├─────────────────┤  - Lock extension fails                    │
│  │ Key Extender    │  - Context cancelled                       │
│  │ (stream-N)      │                                            │
│  └─────────────────┘                                            │
│                                                                 │
│  ┌─────────────────┐                                            │
│  │ Notification    │  1 goroutine - serializes all              │
│  │ Broker          │  notifications to output channel           │
│  └─────────────────┘                                            │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

Total goroutines per client: 3 + N (where N = active streams)

Key points:

• Each active stream has its own key extender goroutine
• Goroutines are lightweight (~2KB stack) but scale with stream count
• All goroutines are properly cleaned up on Done() or context cancellation

NotificationBroker

The library uses an internal NotificationBroker component to safely manage notifications from multiple concurrent sources. This ensures thread-safe delivery to the output channel and prevents panics during shutdown.

┌─────────────────────┐     ┌─────────────────────┐
│  Key Extenders      │────▶│                     │
│  (one per stream)   │     │                     │
└─────────────────────┘     │                     │
                            │  NotificationBroker │────▶ outputChan
┌─────────────────────┐     │                     │
│  Keyspace Listener  │────▶│  - Thread-safe      │
│  (Redis pub/sub)    │     │  - Graceful shutdown│
└─────────────────────┘     │  - No send panics   │
                            │                     │
┌─────────────────────┐     │                     │
│  LBS Stream Reader  │────▶│                     │
└─────────────────────┘     └─────────────────────┘

Backpressure & Slow Consumers

When your consumer processes messages slower than they arrive:

What happens internally

1. Output channel fills up - The output channel has a buffer (default: 500 notifications)
2. Broker blocks - Once full, the NotificationBroker blocks waiting for space
3. Upstream backs up - LBS reader, keyspace listener, and key extenders block on broker
4. Redis consumer lag increases - Messages accumulate in Redis pending entries list (PEL)

Symptoms of slow consumption

• Growing XPENDING count in Redis
• Increased memory usage in Redis
• Notifications arriving late (stream expirations delayed)
• Potential message recovery by other consumers if idle time exceeded

Mitigation strategies

// 1. Process notifications concurrently
for notification := range outputChan {
    go handleNotification(notification)  // Don't block the reader
}

// 2. Use worker pool for controlled concurrency
workerPool := make(chan struct{}, 10)  // 10 concurrent workers
for notification := range outputChan {
    workerPool <- struct{}{}
    go func(n notifs.RecoverableRedisNotification) {
        defer func() { <-workerPool }()
        handleNotification(n)
    }(notification)
}

// 3. Acknowledge streams promptly
// Call DoneStream() as soon as processing completes
client.DoneStream(ctx, streamName)

Monitoring backpressure

# Check pending messages per consumer
redis-cli XPENDING my-service-input my-service-group

# Check consumer lag
redis-cli XINFO CONSUMERS my-service-input my-service-group

Memory Implications

Buffered Channels

Channel	Default Size	Memory per Item	Purpose
Output channel	500	~200 bytes	Notifications to consumer
Broker input	500	~200 bytes	Internal broker queue
Keyspace channel	500	~100 bytes	Redis pub/sub messages

Approximate memory per client: ~300KB for channels alone

Per-Stream Memory

Each active stream consumes:

• Goroutine stack: ~2KB (grows as needed)
• Mutex state: ~200 bytes (redsync mutex)
• StreamLocksInfo: ~150 bytes (LBSInfo + metadata)

Formula: Total = Base(250KB) + Streams × 2.5KB

Active Streams	Approximate Memory
10	~275 KB
100	~500 KB
1,000	~2.75 MB
10,000	~25 MB

Redis Memory

The library creates Redis keys that consume memory:

• Mutex keys: One per active stream, expires after hbInterval
• LBS stream: Grows until messages are acknowledged
• Consumer group metadata: Small, fixed overhead

Memory Management Best Practices

// 1. Call DoneStream() promptly to release resources
if err := client.DoneStream(ctx, streamName); err != nil {
    slog.Error("Failed to release stream", "error", err)
}

// 2. Monitor active stream count
// High stream counts = high memory + goroutine usage

// 3. Tune channel sizes if needed (advanced)
client, err := impl.NewRedisStreamClient(
    redisClient,
    "my-service",
    impl.WithKspChanSize(50),  // Reduce if memory constrained
)

// 4. Implement circuit breaker for runaway stream counts
const maxStreams = 1000
if activeStreamCount > maxStreams {
    slog.Warn("Too many active streams, rejecting new assignments")
    // Handle gracefully
}

Usage

Just import the library:

go get https://github.com/handcoding-labs/redis-stream-client-go

Create the client:

import rsc "github.com/handcoding-labs/redis-stream-client-go/impl"

client := rsc.NewRedisStreamClient(<go redis client>, <service_name>)

Configuration Options

The client supports optional configuration parameters:

import "github.com/handcoding-labs/redis-stream-client-go/impl"

// Create client with custom configuration
client, err := impl.NewRedisStreamClient(
    redisClient,
    "my-service",
    impl.WithClusterMode(impl.ClusterModeSingleShard), // or ClusterModeOSS for Redis Cluster
    impl.WithRecoveryConfig(impl.RecoveryConfig{
        ReconciliationInterval: 60 * time.Second, // base period of the recovery scan (jitter added)
        MinIdleTime:            30 * time.Second, // min idle before a pending message is recoverable
        BatchSize:              50,               // max pending messages inspected per scan
        MaxRetries:             3,                // re-queues before DLQ routing
        DLQStream:              "my-service-dlq", // empty => defaults to "<service>-input-dlq"
        DLQMaxLen:              10000,            // approx MAXLEN cap on the DLQ; 0 => unbounded
    }),
)

Available Options:

• WithClusterMode(mode): Selects the recovery/keyspace strategy — ClusterModeSingleShard (default) or ClusterModeOSS (requires a *redis.ClusterClient). See Cluster support.
• WithRecoveryConfig(config): Tunes the periodic reconciliation scan (ReconciliationInterval, MinIdleTime, BatchSize, MaxRetries, DLQStream, DLQMaxLen). Defaults: 60s / 30s / 50 / 3 / "<service>-input-dlq" / 10000. A RecoveryConfigBuilder (NewRecoveryConfigBuilder().…Build()) is also available and validates on Build().
• WithRetryConfig(config): Configures retry/backoff for transient LBS read errors.
• WithLogger(logger): Provide a custom *slog.Logger.
• WithMetricsRecorder(recorder): Provide a metrics.Recorder for instrumentation (see docs/METRICS.md).

Deprecated: WithLBSIdleTime and WithLBSRecoveryCount no longer affect recovery (now governed by RecoveryConfig) and will be removed in a future release. See docs/MIGRATION.md.

Environment Variables

The client requires one of the following environment variables to generate unique consumer IDs:

• POD_NAME: Kubernetes pod name (preferred in containerized environments)
• POD_IP: Pod IP address (fallback option)

# Example setup
export POD_NAME=my-consumer-$(hostname)-$(date +%s)
# OR
export POD_IP=$(hostname -I | awk '{print $1}')

The consumer ID will be prefixed with redis-consumer- automatically.

Initialize the client and use the LBC and Key space notification channel for tracking which data streams to read and which have expired respectively:

outputChan, err := client.Init(ctx)

Adding Messages to LBS

To add messages to the Load Balancer Stream (LBS), use the LBSInputMessage structure:

import "github.com/handcoding-labs/redis-stream-client-go/notifs"

// Create an LBS message
lbsMessage := notifs.LBSInputMessage{
    DataStreamName: "user-session-123",
    Info: map[string]interface{}{
        "user_id":    "user-456",
        "session_id": "session-789",
        "priority":   "high",
        "created_at": time.Now().Format(time.RFC3339),
    },
}

// Marshal to JSON
messageData, err := json.Marshal(lbsMessage)
if err != nil {
    return err
}

// Add to LBS stream
result := redisClient.XAdd(ctx, &redis.XAddArgs{
    Stream: "<service_name>-input", // LBS stream name
    Values: map[string]interface{}{
        "lbs-input": string(messageData),
    },
})

The Info field in LBSInputMessage allows you to pass additional metadata that will be available in the notification's AdditionalInfo field when the stream is assigned to a consumer.

Notification Types

There are currently four types of notifications sent on outputChan:

1. StreamAdded - When a new stream gets added to LBS. You should take the stream and start reading your data from it using standard XREAD or XREADGROUP commands as applicable. The notification includes:
   • Payload: Contains DataStreamName and IDInLBS
   • AdditionalInfo: Contains the Info field from the original LBSInputMessage
2. StreamExpired - Sent when a keyspace notification reports that a stream's lock has expired (its owner died or got stuck). Call the Claim API to recover it; this re-queues the stream for redistribution rather than granting ownership directly. Recovery also happens automatically via the periodic reconciliation scan, so handling StreamExpired is an optional low-latency optimization.
3. StreamDisowned - When a client gets stuck (not crashed) and thus automatically relinquishes ownership, another active client will claim it. When the old client comes back, it will fail to extend the lock and thus will be informed that it now doesn't own the stream. The old client should gracefully exit by calling Done API.
4. StreamTerminated - Internal notification indicating the notification channel is closing, typically due to context cancellation or fatal errors. Contains additional info about the termination reason.

Claiming

When you receive a StreamExpired notification, call Claim to recover the expired stream. Claim acknowledges the dead consumer's pending message and re-adds it to the LBS as a new message (XACK + XADD). It does not grant ownership to the caller — the stream is redistributed normally and the consumer that picks it up (possibly this one) receives a StreamAdded. Process the stream then, not immediately after Claim.

case notifs.StreamExpired:
    // Re-queue the expired stream for redistribution. Do NOT process here; wait for the
    // StreamAdded that follows when the re-queued stream is picked up.
    if err := client.Claim(ctx, notification.Payload); err != nil {
        // ErrAlreadyClaimed means another consumer (or the reconciliation scan) already
        // recovered it — this is expected and safe to ignore.
        slog.Debug("stream already recovered by another consumer", "stream", notification.Payload.DataStreamName)
    }

case notifs.StreamAdded:
    // Ownership granted (whether a fresh stream or a recovered/re-queued one).
    go handleStream(ctx, notification.Payload.DataStreamName)

A non-nil error from Claim (ErrAlreadyClaimed) indicates another consumer or the periodic scan already recovered the stream.

Stream lifecycle management

The library provides granular control over stream lifecycle:

Processing individual streams

After processing is done for a specific data stream, call DoneStream to mark the end of processing for that particular stream:

err := client.DoneStream(ctx, <data_stream_name>)

This method:

• Unlocks the distributed lock for the stream
• Acknowledges the message in the LBS stream
• Cleans up internal state for that specific stream

Client shutdown

When the client is shutting down completely, call Done to clean up all streams handled by the client:

err := client.Done(ctx)

This method calls DoneStream for all active streams and then performs additional cleanup like closing channels and canceling contexts. The internal NotificationBroker ensures all pending notifications are drained before the output channel is closed.

Done does not close the Redis client you passed in. Close it after Done returns: the consumer's blocking read on the LBS stream is not interrupted by cancellation, and until its connection is closed it can still be handed one newly added message, which is then recovered by the reconciliation scan after MinIdleTime (and uses one retry). See Shutdown and the Redis connection.

Method ID() can be used to obtain client ID for logging purposes:

client.ID()

Complete Example

Here's a complete working example:

package main

import (
    "context"
    "encoding/json"
    "log/slog"
    "os"
    "os/signal"
    "syscall"
    "time"

    "github.com/redis/go-redis/v9"
    "github.com/handcoding-labs/redis-stream-client-go/impl"
    "github.com/handcoding-labs/redis-stream-client-go/notifs"
    "github.com/handcoding-labs/redis-stream-client-go/types"
)

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    // Set required environment variable
    os.Setenv("POD_NAME", "example-consumer")

    // Create Redis client
    redisClient := redis.NewUniversalClient(&redis.UniversalOptions{
        Addrs: []string{"localhost:6379"},
        DB:    0,
    })
    defer redisClient.Close()

    // Enable keyspace notifications
    redisClient.ConfigSet(ctx, "notify-keyspace-events", "KEx")

    // Create and initialize stream client. WithForceConfigOverride is needed because the keyspace
    // config is already set above; without it Init refuses to start (ErrExistingConfigWithoutOverride).
    client, err := impl.NewRedisStreamClient(redisClient, "example-service", impl.WithForceConfigOverride())
    if err != nil {
        slog.Error("could not initialize", "error", err.Error())
    }
    outputChan, err := client.Init(ctx)
    if err != nil {
        slog.Error("Failed to initialize client", "error", err)
        return
    }

    // Handle graceful shutdown
    sigChan := make(chan os.Signal, 1)
    signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)

    // Process notifications
    go func() {
        for notification := range outputChan {
            switch notification.Type {
            case notifs.StreamAdded:
                slog.Info("New stream assigned", "stream", notification.Payload.DataStreamName)
                // Process the stream and call DoneStream when finished. This fires both for fresh
                // streams and for recovered (re-queued) ones.
                go processStream(ctx, client, notification)

            case notifs.StreamExpired:
                // Re-queue the expired stream for redistribution; processing happens later when the
                // re-queued stream arrives as StreamAdded. Not handling this is also fine — the
                // periodic reconciliation scan recovers it regardless.
                if err := client.Claim(ctx, notification.Payload); err != nil {
                    slog.Debug("stream already recovered elsewhere", "stream", notification.Payload.DataStreamName)
                }

            case notifs.StreamDisowned:
                slog.Warn("Lost stream ownership", "stream", notification.Payload.DataStreamName)
            
            case notifs.StreamTerminated:
                slog.Info("Notification channel closing", "reason", notification.AdditionalInfo["info"])
            }
        }
    }()

    // Add test message to LBS
    go addTestMessage(ctx, redisClient)

    // Wait for shutdown
    <-sigChan
    client.Done(ctx)
}

func processStream(ctx context.Context, client types.RedisStreamClient, notification notifs.RecoverableRedisNotification) {
    streamName := notification.Payload.DataStreamName
    slog.Info("Processing stream", "stream", streamName, "info", notification.AdditionalInfo)
    
    // Simulate processing
    time.Sleep(2 * time.Second)
    
    // Mark stream as done
    if err := client.DoneStream(ctx, streamName); err != nil {
        slog.Error("Failed to mark stream done", "error", err, "stream", streamName)
    } else {
        slog.Info("Stream processing completed", "stream", streamName)
    }
}

func addTestMessage(ctx context.Context, redisClient redis.UniversalClient) {
    time.Sleep(1 * time.Second) // Wait for client to be ready
    
    lbsMessage := notifs.LBSInputMessage{
        DataStreamName: "test-stream-1",
        Info: map[string]interface{}{
            "priority": "high",
            "user_id":  "user-123",
        },
    }
}

Documentation

Document	Description
Architecture	Threading model, NotificationBroker, internal design
Usage Guide	API reference, configuration, notification types
Operations	Memory, limits, backpressure, troubleshooting
Example	Complete working example

Installation

go get github.com/handcoding-labs/redis-stream-client-go

Requirements

• Go 1.25.8+
• Redis 6.0+ with keyspace notifications enabled (notify-keyspace-events KEx)
• Environment variable: POD_NAME or POD_IP

License

Apache-2.0