Consumer Groups

Coordinate multiple consumers with consumer groups.

Creating Consumer Groups

Automatic Creation

The @StreamConsumer decorator creates groups automatically:

@Injectable()
export class OrderProcessor {
  @StreamConsumer({
    stream: 'orders',
    group: 'processors',  // Created automatically
  })
  async handle(message: IStreamMessage<Order>): Promise<void> {
    await this.processOrder(message.data);
    await message.ack();
  }
}

Manual Creation

Create groups with specific starting positions:

import { Injectable, Inject, OnModuleInit } from '@nestjs/common';
import { STREAM_CONSUMER, IStreamConsumer } from '@nestjs-redisx/streams';

@Injectable()
export class StreamSetup implements OnModuleInit {
  constructor(
    @Inject(STREAM_CONSUMER) private readonly consumer: IStreamConsumer,
  ) {}

  async onModuleInit(): Promise<void> {
    // Start from beginning
    await this.consumer.createGroup('orders', 'processors', '0');

    // Start from end (new messages only)
    await this.consumer.createGroup('orders', 'analytics', '$');

    // Start from specific ID
    await this.consumer.createGroup(
      'orders',
      'audit',
      '1706123456789-0'
    );
  }
}

Load Balancing Pattern

Multiple consumers in the same group share work:

Implementation

// Worker 1
@StreamConsumer({
  stream: 'orders',
  group: 'processors',
  consumer: 'worker-1',
  concurrency: 5,
})
async handleWorker1(message: IStreamMessage<Order>) {
  await this.process(message.data);
  await message.ack();
}

// Worker 2
@StreamConsumer({
  stream: 'orders',
  group: 'processors',
  consumer: 'worker-2',
  concurrency: 5,
})
async handleWorker2(message: IStreamMessage<Order>) {
  await this.process(message.data);
  await message.ack();
}

Result: Messages distributed evenly, 10 concurrent operations total.

Fan-out Pattern

Multiple groups each receive all messages:

Implementation

// Order processing
@StreamConsumer({ stream: 'orders', group: 'processors' })
async processOrder(message: IStreamMessage<Order>) {
  await this.orderService.fulfill(message.data);
  await message.ack();
}

// Analytics
@StreamConsumer({ stream: 'orders', group: 'analytics' })
async trackMetrics(message: IStreamMessage<Order>) {
  await this.analyticsService.track(message.data);
  await message.ack();
}

// Notifications
@StreamConsumer({ stream: 'orders', group: 'notifications' })
async sendNotification(message: IStreamMessage<Order>) {
  await this.emailService.sendConfirmation(message.data);
  await message.ack();
}

Result: Each group processes all messages independently.

Pending Messages

Messages delivered but not yet acknowledged.

Check Pending Count

const pending = await this.consumer.getPending('orders', 'processors');

console.log({
  total: pending.count,
  oldest: pending.minId,
  newest: pending.maxId,
  consumers: pending.consumers,  // Per-consumer counts
});

// Output:
// {
//   total: 15,
//   oldest: '1706123456789-0',
//   newest: '1706123456799-0',
//   consumers: [
//     { name: 'worker-1', pending: 10 },
//     { name: 'worker-2', pending: 5 },
//   ]
// }

Pending Info

The getPending() method returns summary info:

const pending = await this.consumer.getPending('orders', 'processors');

console.log({
  count: pending.count,        // Total pending
  minId: pending.minId,        // Oldest pending ID
  maxId: pending.maxId,        // Newest pending ID
  consumers: pending.consumers, // Per-consumer counts
});

// For detailed per-message inspection, use Redis CLI:
// redis-cli XPENDING orders processors - + 10

Claiming Idle Messages

Claim messages from dead or slow consumers.

Auto-Claim via Module Config

Each consumer runs a background auto-claim loop driven by the claimIdleTimeout option (default 30000 ms). Every claimIdleTimeout milliseconds the consumer scans the group's pending entries (XPENDING) and reclaims (XCLAIM) any message that has been idle for at least claimIdleTimeout — for example messages left pending by a crashed or stuck consumer. Reclaimed messages flow through the normal handler / retry / DLQ path, so orphaned messages are recovered automatically without any cron job.

@StreamConsumer({
  stream: 'orders',
  group: 'processors',
  claimIdleTimeout: 30000,  // Reclaim messages idle >= 30s, every 30s (default)
})
async handle(message: IStreamMessage<Order>) {
  await this.process(message.data);
  await message.ack();
}

Set claimIdleTimeout: 0 to disable the background auto-claim. The manual claimIdle() method (below) is still available for on-demand claiming — for example to reclaim with a different idle threshold or from an admin task.

Manual Claim

// Claim idle messages (idle > 30 seconds)
const claimed = await this.consumer.claimIdle(
  'orders',
  'processors',
  'worker-new',  // New consumer taking over
  30000,         // Min idle time (ms)
);

console.log(`Claimed ${claimed.length} messages`);

Claim and Process

import { Injectable, Inject } from '@nestjs/common';
import { STREAM_CONSUMER, IStreamConsumer } from '@nestjs-redisx/streams';

@Injectable()
export class IdleMessageClaimer {
  constructor(
    @Inject(STREAM_CONSUMER) private readonly consumer: IStreamConsumer,
  ) {}

  // @Cron('*/1 * * * *')  // Every minute
  async claimIdleMessages(): Promise<void> {
    // Claim messages idle > 60 seconds
    const claimed = await this.consumer.claimIdle(
      'orders',
      'processors',
      'claimer-worker',
      60000,
    );

    if (claimed.length === 0) return;

    // Process claimed messages
    for (const message of claimed) {
      try {
        await this.processMessage(message);
        await message.ack();
      } catch (error) {
        await message.reject(error);
      }
    }
  }

  private async processMessage(message: any): Promise<void> {
    // Process the claimed message
  }
}

Group Information

Use Redis CLI or the Redis driver directly for group/consumer introspection:

# List groups
redis-cli XINFO GROUPS orders

# List consumers in a group
redis-cli XINFO CONSUMERS orders processors

Scaling Consumers

Horizontal Scaling

Add more consumer instances:

# Server 1
node dist/main.js  # Consumer: worker-1

# Server 2
node dist/main.js  # Consumer: worker-2

# Server 3
node dist/main.js  # Consumer: worker-3

All consumers in the same group share work automatically.

Vertical Scaling

Increase concurrency per consumer:

@StreamConsumer({
  stream: 'orders',
  group: 'processors',
  concurrency: 20,  // Process 20 messages simultaneously
  batchSize: 50,    // Fetch 50 at a time
})
async handle(message: IStreamMessage<Order>) {
  await this.process(message.data);
  await message.ack();
}

Cleanup

Use Redis CLI for group/consumer management:

# Delete inactive consumer from group
redis-cli XGROUP DELCONSUMER orders processors worker-old

# Delete entire group (removes all pending tracking)
redis-cli XGROUP DESTROY orders processors

Warning

Deleting a group removes all pending message tracking. Messages remain in the stream.

Best Practices

1. Use descriptive group names:

// ✅ Good
'order-processors'
'analytics-trackers'
'email-senders'

// ❌ Bad
'group1'
'consumers'

2. Pick an appropriate min-idle time when claiming manually:

The claimIdleTimeout config option drives the background auto-claim. When claiming on demand, pass the idle threshold directly to claimIdle():

// For fast operations (< 1s)
await consumer.claimIdle('orders', 'processors', 'worker-recovery', 10000);   // 10s idle

// For slow operations (minutes)
await consumer.claimIdle('orders', 'processors', 'worker-recovery', 300000);  // 5min idle

3. Monitor pending messages:

@Cron('*/5 * * * *')  // Every 5 minutes
async checkPendingMessages() {
  const pending = await this.consumer.getPending('orders', 'processors');

  if (pending.count > 1000) {
    this.alertService.send('High pending message count');
  }
}

4. Clean up dead consumers:

# Check for idle consumers
redis-cli XINFO CONSUMERS orders processors

# Delete consumers idle > 1 hour manually
redis-cli XGROUP DELCONSUMER orders processors dead-worker

Next Steps

• Dead Letter Queue — Handle failed messages
• Monitoring — Track consumer groups