fix: Recreate group in case redis flush

 - adding a test
 - reset properly the id
 - adding another rare case when group is tempered by another system, and IDs gets desynchronized. (AI recommendation)

Author: pbonneaudiabolocom

faststream/redis/subscriber/usecases/stream_subscriber.py

@@ -69,8 +69,11 @@ async def _consume(self, *args: Any, start_signal: "Event") -> None:
             start_signal.set()
         await super()._consume(*args, start_signal=start_signal)
 
-    async def _create_group(self) -> None:
-        group_create_id = "$" if self.last_id == ">" else self.last_id
+    async def _create_group(self, reset_counter:bool = False) -> None:
+        if reset_counter:
+            group_create_id = "0"
+        else:
+            group_create_id = "$" if self.last_id == ">" else self.last_id
         try:
             await self._client.xgroup_create(
                 name=self.stream_sub.name,
@@ -155,8 +158,22 @@ async def read_from_group() -> tuple[
                             noack=stream.no_ack,
                         )
                     except ResponseError as e:
-                        if "NOGROUP" in str(e):
-                            await self._create_group()
+                        err_msg = str(e)
+                        known_error:bool = False
+                        if "NOGROUP" in err_msg:
+                            # most likely redis was flushed, so we need to reset our group
+                            await self._create_group(reset_counter=True)
+                            # Important: reset our internal position too
+                            stream.last_id = ">"
+                            known_error = True
+                        if (
+                                "smaller than the first available entry" in err_msg
+                                or "greater than the maximum id" in err_msg
+                        ):
+                            # group was modified by third party and we need to reset our position to an existing id
+                            stream.last_id = "$"
+                            known_error = True
+                        if known_error:
                             return await client.xreadgroup(  # type: ignore[no-any-return]
                                 groupname=stream.group,
                                 consumername=stream.consumer,
@@ -165,7 +182,7 @@ async def read_from_group() -> tuple[
                                 block=stream.polling_interval,
                                 noack=stream.no_ack,
                             )
-                        raise
+                        raise e
 
                 return read_from_group()
 

tests/brokers/redis/test_consume.py

@@ -844,6 +844,52 @@ async def handler(msg: RedisStreamMessage) -> None:
             assert queue_len == 0, (
                 f"Redis stream must be empty here, found {queue_len} messages"
             )
+    async def test_consume_from_group(
+        self,
+        queue: str,
+    ) -> None:
+        event = asyncio.Event()
+
+        consume_broker = self.get_broker(apply_types=True)
+
+        @consume_broker.subscriber(
+            stream=StreamSub(queue, group="group", consumer=queue),
+        )
+        async def handler(msg: RedisMessage) -> None:
+            event.set()
+
+        async with self.patch_broker(consume_broker) as br:
+            await br.start()
+            redis_client = br._connection
+            with (
+                patch.object(redis_client, "xreadgroup", spy_decorator(redis_client.xreadgroup)) as m_readgroup,
+                patch.object(redis_client, "xgroup_create", spy_decorator(redis_client.xgroup_create)) as m_group_create
+            ):
+                await asyncio.wait(
+                    (
+                        asyncio.create_task(br.publish("hello", stream=queue)),
+                        asyncio.create_task(event.wait()),
+                    ),
+                    timeout=3,
+                )
+                await asyncio.sleep(0.1)
+                m_readgroup.mock.assert_called_once()
+                assert event.is_set()
+                await redis_client.flushall()
+                event.clear()
+                await asyncio.sleep(0.1)
+                await asyncio.wait(
+                    (
+                        asyncio.create_task(br.publish("hello again", stream=queue)),
+                        asyncio.create_task(event.wait()),
+                    ),
+                    timeout=3,
+                )
+
+                await asyncio.sleep(0.1)
+                m_group_create.mock.assert_called_once()
+
+                assert event.is_set()
 
     async def test_get_one(
         self,