Skip to content

Testing Stream Consumer Groups#

Consumer Groups and Message Claiming both rely on Redis's Pending Entries List (PEL) to track messages that were delivered but not yet acknowledged. Exercising that lifecycle against a real Redis instance means waiting out real min_idle_time timers, which makes tests slow and non-deterministic.

TestRedisBroker avoids that by emulating the PEL in memory. It applies the same rules a real Redis consumer group would, but synchronously - so a min_idle_time consumer can reclaim a nacked message within the very same await broker.publish(...) call instead of waiting for a real timeout.

How the Emulated PEL Works#

For every message delivered to a consumer group member, the fake broker applies the same three rules as production Redis:

  1. Successful processing removes the entry from the PEL - nothing is left pending.
  2. A nacked message (e.g. a raised NackMessage) leaves the entry in the PEL, where it stays until a min_idle_time consumer reclaims it.
  3. no_ack=True consumers are never tracked in the PEL at all, matching Redis's NOACK flag, which acknowledges a message the moment it's delivered.

Pending Without a Claimer#

If nothing in the group has min_idle_time set, a nacked message simply stays pending. Here, flaky_worker always fails, and there's no one to reclaim its work:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
from faststream import FastStream, Logger
from faststream.exceptions import NackMessage
from faststream.redis import RedisBroker, StreamSub

broker = RedisBroker()
app = FastStream(broker)


@broker.subscriber(
    stream=StreamSub("orders", group="order-processors", consumer="worker-1"),
)
async def flaky_worker(order_id: str, logger: Logger) -> None:
    logger.info(f"Failed to process order: {order_id}")
    raise NackMessage

Pass a PEL instance to TestRedisBroker to inspect it directly after publishing. A pel fixture keeps every test working against its own instance:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
@pytest.mark.asyncio
async def test_pending_message_stays_without_a_claimer() -> None:
    pel = PEL()
    async with TestRedisBroker(pending_broker, pel=pel) as br:
        await br.publish("order-1", stream="orders")

        pending_worker.mock.assert_called_once_with("order-1")
        # nothing exists to reclaim it, so the entry just stays pending
        assert len(pel.entries) == 1


@pytest.mark.asyncio
async def test_each_group_gets_its_own_pel_entry() -> None:

Multiple Groups Mean Multiple PEL Entries#

The PEL is tracked per consumer group, not per message. If the same stream has several groups subscribed - each modeling an independent workload - a single published message that goes unacknowledged in every group leaves one pending entry per group, not one shared entry:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
from faststream import FastStream, Logger
from faststream.exceptions import NackMessage
from faststream.redis import RedisBroker, StreamSub

broker = RedisBroker()
app = FastStream(broker)


@broker.subscriber(
    stream=StreamSub("orders", group="billing", consumer="worker-1"),
)
async def billing_worker(order_id: str, logger: Logger) -> None:
    logger.info(f"Billing failed for order: {order_id}")
    raise NackMessage


@broker.subscriber(
    stream=StreamSub("orders", group="shipping", consumer="worker-1"),
)
async def shipping_worker(order_id: str, logger: Logger) -> None:
    logger.info(f"Shipping failed for order: {order_id}")
    raise NackMessage

1
2
3
4
5
6
7
8
9
        await br.publish("order-4", stream="orders")

        billing_worker.mock.assert_called_once_with("order-4")
        shipping_worker.mock.assert_called_once_with("order-4")
        # one entry per group, tracked independently for the same message
        assert len(pel.entries) == 2


@pytest.mark.asyncio

This mirrors real Redis: XPENDING is scoped to a single consumer group, so groups never see or interfere with each other's pending entries, even when they're reading the same stream.

Reclaiming a Pending Message#

Add a min_idle_time consumer to the same group, and it reclaims the pending entry once flaky_worker nacks it:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
from faststream import FastStream, Logger
from faststream.exceptions import NackMessage
from faststream.redis import RedisBroker, StreamSub

broker = RedisBroker()
app = FastStream(broker)

while (
    len(broker.subscribers) < 2
    or broker.subscribers[0].specification.call_name != "FlakyWorker"
):
    broker = RedisBroker()
    @broker.subscriber(
        stream=StreamSub("orders", group="order-processors", consumer="worker-1"),
    )
    async def flaky_worker(order_id: str, logger: Logger) -> None:
        logger.info(f"Failed to process order: {order_id}")
        raise NackMessage


    @broker.subscriber(
        stream=StreamSub(
            "orders",
            group="order-processors",
            consumer="claimer",
            min_idle_time=10000,  # 10 seconds
        ),
    )
    async def claiming_worker(order_id: str, logger: Logger) -> None:
        logger.info(f"Recovered order: {order_id}")

1
2
3
4
5
6
    pel = PEL()
    async with TestRedisBroker(reprocessing_broker, pel=pel) as br:
        await br.publish("order", stream="orders")
        reprocessing_worker.mock.assert_called_once_with("order")
        claiming_worker.mock.assert_called_once_with("order")
    assert len(pel.entries) == 0

Tip

Unlike a real broker, the fake min_idle_time consumer doesn't wait for the idle timeout to elapse - it checks the PEL immediately, so the reclaim happens within the same publish() call that produced the pending entry.

no_ack Skips the PEL Entirely#

Because no_ack=True disables acknowledgement altogether, the fake broker never records an entry for it, even when the handler raises:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
from faststream import FastStream, Logger
from faststream.redis import RedisBroker, StreamSub

broker = RedisBroker()
app = FastStream(broker)


@broker.subscriber(stream=StreamSub("orders", no_ack=True))
async def fire_and_forget_worker(order_id: str, logger: Logger) -> None:
    logger.info(f"Processing order: {order_id}")
    error_msg = f"Could not process order: {order_id}"
    raise ValueError(error_msg)

1
2
3
4
5
6
7
    async with TestRedisBroker(no_ack_broker, pel=pel) as br:
        with pytest.raises(ValueError, match="Could not process order"):
            await br.publish("order-3", stream="orders")

        fire_and_forget_worker.mock.assert_called_once_with("order-3")
        # no_ack means nothing was ever recorded, failure or not
        assert pel.entries == {}

Inspecting the PEL#

By default, each TestRedisBroker creates its own private PEL. Passing one explicitly (as in the examples above) lets you assert against it directly - either by reading pel.entries, or through the put/remove spies it exposes:

from unittest.mock import patch

from faststream.redis.testing import PEL, TestRedisBroker

pel = PEL()

async with TestRedisBroker(broker, pel=pel) as br:
    with patch.object(pel, "put") as put_mock:
        await br.publish(...)

    put_mock.assert_not_called()