[ISS-2462] fix(redis): list subscriber consume/stop race

docker-compose.yaml

@@ -42,7 +42,7 @@ services:
       - no-new-privileges:true
 
   faststream:
-    build: .
+    image: faststream:local
     volumes:
       - ./:/src
       - /src/.venv/

faststream/_internal/endpoint/subscriber/usecase.py

@@ -147,9 +147,9 @@ async def stop(self) -> None:
 
         Blocks event loop up to graceful_timeout seconds.
         """
-        self.running = False
         if isinstance(self.lock, MultiLock):
             await self.lock.wait_release(self._outer_config.graceful_timeout)
+        self.running = False
 
     def add_call(
         self,

faststream/redis/subscriber/usecases/basic.py

@@ -91,7 +91,8 @@ async def _consume(self, *args: Any, start_signal: anyio.Event) -> None:
 
         while self.running:
             try:
-                await self._get_msgs(*args)
+                with self.lock:
+                    await self._get_msgs(*args)
 
             except Exception as e:  # noqa: PERF203
                 self._log(

justfile

@@ -14,8 +14,8 @@ default:
 [doc("Init infra")]
 [group("infra")]
 init python="3.10":
-  docker build . --build-arg PYTHON_VERSION={{python}}
-  uv sync --group dev
+  docker build . --build-arg PYTHON_VERSION={{python}} --no-cache --tag faststream:local
+  uv sync --group dev -p {{python}}
 
 [doc("Run all containers")]
 [group("infra")]

tests/brokers/redis/test_consume.py

@@ -197,6 +197,38 @@ async def handler(msg) -> None:
 
         mock.assert_called_once_with(b"hello")
 
+    @pytest.mark.slow()
+    @pytest.mark.flaky(reruns=3, reruns_delay=1)
+    async def test_consume_list_with_big_interval(
+        self,
+        queue: str,
+        mock: MagicMock,
+    ):
+        consume_broker = self.get_broker()
+
+        sub = consume_broker.subscriber(list=ListSub(list_name="input-list"))
+        pub = consume_broker.publisher(list=ListSub(list_name="input-list"))
+
+        @sub
+        async def handle(msg: str) -> None:
+            mock(msg)
+
+        original_blpop = Redis.blpop
+
+        async def extended_blpop(self, *args, **kwargs):
+            result = await original_blpop(self, *args, **kwargs)
+            await asyncio.sleep(2)
+            return result
+
+        patcher = patch.object(Redis, "blpop", new=extended_blpop)
+        patcher.start()
+
+        await consume_broker.start()
+        await pub.publish("hello")
+        await asyncio.sleep(1)
+        await consume_broker.stop()
+        mock.assert_called_once_with("hello")
+
     @pytest.mark.slow()
     async def test_consume_list_batch_with_one(
         self,