fix(pubsub): use remaining timeout in listen() to prevent cumulative timeout drift

- Track remaining timeout across loop iterations instead of passing
the full timeout each time. This ensures listen(timeout=N) returns
within approximately N seconds total, not N seconds per iteration.
- Addresses cursor[bot] review comments about full timeout being
passed each loop iteration.

Author: Developer

redis/asyncio/client.py

@@ -1440,16 +1440,20 @@ async def listen(self, timeout: Optional[float] = None) -> AsyncIterator:
         """
         if timeout is not None:
             start = time.monotonic()
+            remaining = timeout
+        else:
+            remaining = None
         while self.subscribed:
             response = await self.handle_message(
-                await self.parse_response(block=(timeout is None), timeout=timeout)
+                await self.parse_response(block=(timeout is None), timeout=remaining)
             )
             if response is not None:
                 yield response
             if timeout is not None:
                 elapsed = time.monotonic() - start
                 if elapsed >= timeout:
                     break
+                remaining = timeout - elapsed
 
     async def get_message(
         self, ignore_subscribe_messages: bool = False, timeout: Optional[float] = 0.0

redis/client.py

@@ -1449,16 +1449,20 @@ def listen(self, timeout: Optional[float] = None):
         """
         if timeout is not None:
             start = time.monotonic()
+            remaining = timeout
+        else:
+            remaining = None
         while self.subscribed:
             response = self.handle_message(
-                self.parse_response(block=(timeout is None), timeout=timeout)
+                self.parse_response(block=(timeout is None), timeout=remaining)
             )
             if response is not None:
                 yield response
             if timeout is not None:
                 elapsed = time.monotonic() - start
                 if elapsed >= timeout:
                     break
+                remaining = timeout - elapsed
 
     def get_message(
         self, ignore_subscribe_messages: bool = False, timeout: float = 0.0