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

faststream/_internal/endpoint/subscriber/usecase.py

@@ -147,7 +147,10 @@ async def stop(self) -> None:
 
         Blocks event loop up to graceful_timeout seconds.
         """
+        # set running false before releasing to stop new message reading
         self.running = False
+
+        # Wait for already consumed messages to be processed
         if isinstance(self.lock, MultiLock):
             await self.lock.wait_release(self._outer_config.graceful_timeout)
 

faststream/redis/subscriber/usecases/list_subscriber.py

@@ -42,6 +42,7 @@ def __init__(
         calls: "CallsCollection[Any]",
     ) -> None:
         super().__init__(config, specification, calls)
+        self._read_lock = anyio.Lock()
         assert config.list_sub
         self._list_sub = config.list_sub
 
@@ -76,6 +77,12 @@ async def start(self) -> None:
 
         await super().start(self._client)
 
+    @override
+    async def stop(self) -> None:
+        with anyio.move_on_after(self._outer_config.graceful_timeout):
+            async with self._read_lock:
+                await super().stop()
+
     @override
     async def get_one(
         self,
@@ -169,21 +176,22 @@ def __init__(
         super().__init__(config, specification, calls)
 
     async def _get_msgs(self, client: "Redis[bytes]") -> None:
-        raw_msg = await client.blpop(
-            self.list_sub.name,
-            timeout=self.list_sub.polling_interval,
-        )
+        async with self._read_lock:
+            raw_msg = await client.blpop(
+                self.list_sub.name,
+                timeout=self.list_sub.polling_interval,
+            )
 
-        if raw_msg:
-            _, msg_data = raw_msg
+            if raw_msg:
+                _, msg_data = raw_msg
 
-            msg = DefaultListMessage(
-                type="list",
-                data=msg_data,
-                channel=self.list_sub.name,
-            )
+                msg = DefaultListMessage(
+                    type="list",
+                    data=msg_data,
+                    channel=self.list_sub.name,
+                )
 
-            await self.consume_one(msg)
+                await self.consume_one(msg)
 
 
 class ListBatchSubscriber(_ListHandlerMixin):
@@ -199,21 +207,22 @@ def __init__(
         super().__init__(config, specification, calls)
 
     async def _get_msgs(self, client: "Redis[bytes]") -> None:
-        raw_msgs = await client.lpop(
-            name=self.list_sub.name,
-            count=self.list_sub.max_records,
-        )
-
-        if raw_msgs:
-            msg = BatchListMessage(
-                type="blist",
-                channel=self.list_sub.name,
-                data=raw_msgs,
+        async with self._read_lock:
+            raw_msgs = await client.lpop(
+                name=self.list_sub.name,
+                count=self.list_sub.max_records,
             )
 
-            await self.consume_one(msg)
+            if raw_msgs:
+                msg = BatchListMessage(
+                    type="blist",
+                    channel=self.list_sub.name,
+                    data=raw_msgs,
+                )
+
+                await self.consume_one(msg)
 
-        else:
+        if not raw_msgs:
             await anyio.sleep(self.list_sub.polling_interval)
 
 

justfile

@@ -15,7 +15,7 @@ default:
 [group("infra")]
 init python="3.10":
   docker build . --build-arg PYTHON_VERSION={{python}}
-  uv sync --group dev
+  uv sync --group dev -p {{python}}
 
 [doc("Run all containers")]
 [group("infra")]

pyproject.toml

@@ -52,7 +52,7 @@ classifiers = [
     "Framework :: Pydantic :: 2",
 ]
 
-version = "0.6.0rc3"
+version = "0.6.0rc4"
 
 dependencies = [
     "fast-depends[pydantic]>=3.0.0rc1,<4.0.0",