fix: bound subscriber Redis commands so a silent primary can't stall delivery

README.md

@@ -148,6 +148,10 @@ Per-subscriber knobs (passed to `@broker.subscriber("topic", ...)`):
 - `max_concurrent` (default `5`): maximum number of handlers running in
   parallel per subscriber. Also caps the fetch batch size per poll cycle.
   Handlers must be safe under concurrency in addition to being idempotent.
+- `command_timeout` (default `None`, meaning `lease_ttl`): the most time the
+  subscriber waits on a single Redis command (fetch, claim, ack). When a primary
+  stops answering without closing its connections, the command fails after this
+  long and the poll loop reconnects. Must be greater than 0 and at most `lease_ttl`.
 
 ## Failure modes
 

docs/introduction/how-it-works.md

@@ -67,6 +67,7 @@ The default ack policy is `NACK_ON_ERROR`: the timer is acknowledged on success,
 | `max_polling_interval` | `5.0` s | Ceiling for the adaptive idle backoff: `polling_interval` doubles up to this value on consecutive empty polls. Jitter is applied before the cap, so worst-case delivery latency on a previously-idle topic is `max_polling_interval` |
 | `max_concurrent` | `5` | Max handlers running concurrently per subscriber; also bounds fetch batch size |
 | `lease_ttl` | `30` s | How long a worker holds the lease before another worker may re-claim. Set to ~3–5× the P99 handler runtime: lower values speed up recovery from worker death, higher values tolerate handler GC pauses and clock skew |
+| `command_timeout` | `None` (`lease_ttl`) | Most time the subscriber waits on one Redis command (fetch, claim, ack). A primary that goes silent without closing its connections otherwise stalls the subscriber; redis-py before 8.0 sets no socket timeout by default. Must be > 0 and <= `lease_ttl` |
 
 ## Operational requirements
 

docs/usage/subscriber.md

@@ -69,6 +69,7 @@ Configure polling behaviour per subscriber:
 | `max_polling_interval` | `5.0` s | Cap for adaptive idle backoff (doubles per empty cycle, ±50% jitter applied before the cap) |
 | `max_concurrent` | `5` | Max handlers running in parallel; also caps fetch batch size per poll |
 | `lease_ttl` | `30` s | How long a worker holds the lease before another worker may re-claim |
+| `command_timeout` | `None` (`lease_ttl`) | Most time the subscriber waits on one Redis command (fetch, claim, ack) before dropping the connection and reconnecting. Must be > 0 and <= `lease_ttl` |
 
 ```python
 @broker.subscriber(

faststream_redis_timers/parser/parser.py

@@ -2,6 +2,7 @@
 import typing
 from functools import partial
 
+import anyio
 from faststream.message import decode_message
 
 from faststream_redis_timers.envelope import TimerMessageFormat
@@ -19,7 +20,6 @@ def __init__(self, config: "TimersSubscriberConfig") -> None:
 
     async def parse_message(self, msg: "TimerMessage") -> TimerStreamMessage:
         timer_id = msg["timer_id"]
-        store = self._config._outer_config.store  # noqa: SLF001
         discard_corrupt = None
         try:
             body, headers = TimerMessageFormat.parse(msg["data"])
@@ -34,13 +34,18 @@ async def parse_message(self, msg: "TimerMessage") -> TimerStreamMessage:
             message_id=timer_id,
             correlation_id=headers.get("correlation_id", timer_id),
             reply_to=headers.get("reply_to", ""),
-            _remove=partial(store.remove, self._config.full_topic, timer_id),
+            _remove=partial(self._remove, timer_id),
             _discard_corrupt=discard_corrupt,
         )
 
+    async def _remove(self, timer_id: str) -> None:
+        store = self._config._outer_config.store  # noqa: SLF001
+        with anyio.fail_after(self._config.timer_sub.effective_command_timeout):
+            await store.remove(self._config.full_topic, timer_id)
+
     async def _discard_corrupt(self, msg: "TimerMessage", error: ValueError) -> typing.NoReturn:
         outer_config = self._config._outer_config  # noqa: SLF001
-        await outer_config.store.remove(self._config.full_topic, msg["timer_id"])
+        await self._remove(msg["timer_id"])
         outer_config.logger.log(
             f"Timer {msg['timer_id']!r} on {self._config.full_topic!r} removed: "
             f"its {len(msg['data'])}-byte payload cannot be parsed",

faststream_redis_timers/registrator.py

@@ -24,6 +24,7 @@ def subscriber(  # ty: ignore[invalid-method-override]
         max_polling_interval: float = 5.0,
         max_concurrent: int = 5,
         lease_ttl: int = 30,
+        command_timeout: float | None = None,
         dependencies: Sequence[Dependant] = (),
         parser: CustomCallable | None = None,
         decoder: CustomCallable | None = None,
@@ -37,6 +38,7 @@ def subscriber(  # ty: ignore[invalid-method-override]
             max_polling_interval=max_polling_interval,
             max_concurrent=max_concurrent,
             lease_ttl=lease_ttl,
+            command_timeout=command_timeout,
             config=self.config,  # ty: ignore[invalid-argument-type]
             title_=title_,
             description_=description_,

faststream_redis_timers/router.py

@@ -44,6 +44,7 @@ def __init__(  # noqa: PLR0913
         max_polling_interval: float = 5.0,
         max_concurrent: int = 5,
         lease_ttl: int = 30,
+        command_timeout: float | None = None,
         publishers: Iterable[TimersRoutePublisher] = (),
         dependencies: Sequence[Dependant] = (),
         parser: CustomCallable | None = None,
@@ -59,6 +60,7 @@ def __init__(  # noqa: PLR0913
             max_polling_interval=max_polling_interval,
             max_concurrent=max_concurrent,
             lease_ttl=lease_ttl,
+            command_timeout=command_timeout,
             publishers=publishers,
             dependencies=dependencies,
             parser=parser,

faststream_redis_timers/schemas.py

@@ -1,3 +1,4 @@
+import math
 from dataclasses import dataclass
 
 
@@ -12,3 +13,21 @@ class TimerSub:
     # Maximum concurrent handler invocations; also bounds fetch batch size per poll.
     max_concurrent: int = 5
     lease_ttl: int = 30
+    command_timeout: float | None = None
+
+    def __post_init__(self) -> None:
+        if self.command_timeout is None:
+            return
+        if not math.isfinite(self.command_timeout):
+            msg = f"command_timeout must be finite, got {self.command_timeout}"
+            raise ValueError(msg)
+        if self.command_timeout <= 0:
+            msg = f"command_timeout must be > 0, got {self.command_timeout}"
+            raise ValueError(msg)
+        if self.command_timeout > self.lease_ttl:
+            msg = f"command_timeout ({self.command_timeout}) must be <= lease_ttl ({self.lease_ttl})"
+            raise ValueError(msg)
+
+    @property
+    def effective_command_timeout(self) -> float:
+        return self.lease_ttl if self.command_timeout is None else self.command_timeout

faststream_redis_timers/subscriber/factory.py

@@ -18,6 +18,7 @@ def create_subscriber(  # noqa: PLR0913
     max_polling_interval: float = 5.0,
     max_concurrent: int = 5,
     lease_ttl: int = 30,
+    command_timeout: float | None = None,
     config: "TimersBrokerConfig",
     title_: str | None = None,
     description_: str | None = None,
@@ -29,6 +30,7 @@ def create_subscriber(  # noqa: PLR0913
         max_polling_interval=max_polling_interval,
         max_concurrent=max_concurrent,
         lease_ttl=lease_ttl,
+        command_timeout=command_timeout,
     )
     usecase_config = TimersSubscriberConfig(
         _outer_config=config,

faststream_redis_timers/subscriber/usecase.py

@@ -135,7 +135,8 @@ async def _get_msgs(
             return -1
 
         now = time.time()
-        timer_ids: list[str] = await self._outer_config.store.due(self._config.full_topic, now, free)
+        with anyio.fail_after(self._config.timer_sub.effective_command_timeout):
+            timer_ids: list[str] = await self._outer_config.store.due(self._config.full_topic, now, free)
         if not timer_ids:
             return 0
 
@@ -154,9 +155,10 @@ async def _claim_and_consume(
         try:
             async with limiter:
                 now = time.time()
-                raw_payload: bytes | None = await self._outer_config.store.claim(
-                    self._config.full_topic, timer_id, now, lease_ttl
-                )
+                with anyio.fail_after(self._config.timer_sub.effective_command_timeout):
+                    raw_payload: bytes | None = await self._outer_config.store.claim(
+                        self._config.full_topic, timer_id, now, lease_ttl
+                    )
                 if raw_payload is None:
                     self._log(
                         log_level=logging.DEBUG,

tests/test_command_timeout.py

@@ -0,0 +1,163 @@
+import asyncio
+import contextlib
+import typing
+import uuid
+from collections.abc import AsyncIterator
+
+import pytest
+from redis.asyncio import Redis
+
+from faststream_redis_timers import TimersBroker
+from faststream_redis_timers.router import TimersRoute, TimersRouter
+
+
+class _FreezableProxy:
+    """TCP proxy to Redis; ``freeze()`` silences every open connection while new ones still pass."""
+
+    def __init__(self, target_host: str, target_port: int) -> None:
+        self._target = (target_host, target_port)
+        self._generation = 0
+        self._never = asyncio.Event()
+        self._writers: list[asyncio.StreamWriter] = []
+        self._tasks: set[asyncio.Task[None]] = set()
+        self._server: asyncio.Server | None = None
+        self.port = 0
+
+    async def start(self) -> None:
+        self._server = await asyncio.start_server(self._accept, "127.0.0.1", 0)
+        self.port = self._server.sockets[0].getsockname()[1]
+
+    def freeze(self) -> None:
+        self._generation += 1
+
+    async def stop(self) -> None:
+        assert self._server is not None
+        self._server.close()
+        for task in self._tasks:
+            task.cancel()
+        for writer in self._writers:
+            writer.close()
+        await asyncio.gather(*self._tasks, return_exceptions=True)
+
+    async def _accept(self, client_reader: asyncio.StreamReader, client_writer: asyncio.StreamWriter) -> None:
+        upstream_reader, upstream_writer = await asyncio.open_connection(*self._target)
+        self._writers += [client_writer, upstream_writer]
+        generation = self._generation
+        for reader, writer in ((client_reader, upstream_writer), (upstream_reader, client_writer)):
+            task = asyncio.create_task(self._pipe(reader, writer, generation))
+            self._tasks.add(task)
+
+    async def _pipe(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter, generation: int) -> None:
+        while data := await reader.read(65536):
+            if generation != self._generation:
+                await self._never.wait()
+            writer.write(data)
+            await writer.drain()
+
+
+@pytest.fixture
+async def proxy(redis_client: Redis) -> AsyncIterator[_FreezableProxy]:
+    kwargs = redis_client.connection_pool.connection_kwargs
+    proxy = _FreezableProxy(kwargs["host"], kwargs["port"])
+    await proxy.start()
+    yield proxy
+    await proxy.stop()
+
+
+@pytest.fixture
+async def proxied_client(proxy: _FreezableProxy, redis_client: Redis) -> AsyncIterator[Redis]:
+    db = redis_client.connection_pool.connection_kwargs.get("db", 0)
+    client = Redis(host="127.0.0.1", port=proxy.port, db=db, socket_timeout=None, socket_connect_timeout=None)
+    yield client
+    with contextlib.suppress(Exception):
+        async with asyncio.timeout(1):
+            await client.aclose()  # ty: ignore[unresolved-attribute]
+
+
+def _brokers(proxied_client: Redis, redis_client: Redis) -> tuple[TimersBroker, TimersBroker]:
+    """Return a broker on the proxied client and a publisher on a direct one, sharing keys."""
+    suffix = uuid.uuid4().hex
+    timeline_key, payloads_key = f"ct_tl_{suffix}", f"ct_pl_{suffix}"
+    return (
+        TimersBroker(proxied_client, timeline_key=timeline_key, payloads_key=payloads_key),
+        TimersBroker(redis_client, timeline_key=timeline_key, payloads_key=payloads_key),
+    )
+
+
+async def test_a_fetch_hung_on_a_silent_connection_is_abandoned(
+    proxy: _FreezableProxy, proxied_client: Redis, redis_client: Redis
+) -> None:
+    broker, publisher = _brokers(proxied_client, redis_client)
+    seen: list[str] = []
+    delivered: dict[str, asyncio.Event] = {"before": asyncio.Event(), "after": asyncio.Event()}
+
+    @broker.subscriber("topic", lease_ttl=1, max_polling_interval=0.1)
+    async def handler(body: str) -> None:
+        seen.append(body)
+        delivered[body].set()
+
+    async with broker, publisher:
+        await publisher.publish("before", topic="topic")
+        await asyncio.wait_for(delivered["before"].wait(), timeout=5)
+        proxy.freeze()
+        await asyncio.sleep(0.5)
+        await publisher.publish("after", topic="topic")
+        await asyncio.wait_for(delivered["after"].wait(), timeout=10)
+
+
+async def test_an_ack_hung_on_a_silent_connection_frees_its_handler_slot(
+    proxy: _FreezableProxy, proxied_client: Redis, redis_client: Redis
+) -> None:
+    broker, publisher = _brokers(proxied_client, redis_client)
+    entered = asyncio.Event()
+    release = asyncio.Event()
+    after = asyncio.Event()
+
+    @broker.subscriber("topic", lease_ttl=1, max_concurrent=1, max_polling_interval=0.1)
+    async def handler(body: str) -> None:
+        if body == "after":
+            after.set()
+            return
+        entered.set()
+        await release.wait()
+
+    async with broker, publisher:
+        await publisher.publish("before", topic="topic")
+        await asyncio.wait_for(entered.wait(), timeout=5)
+        proxy.freeze()
+        release.set()
+        await asyncio.sleep(0.2)
+        await publisher.publish("after", topic="topic")
+        await asyncio.wait_for(after.wait(), timeout=15)
+
+
+def test_command_timeout_defaults_to_the_lease(broker: TimersBroker) -> None:
+    subscriber = broker.subscriber("topic", lease_ttl=7)
+    assert subscriber._config.timer_sub.effective_command_timeout == 7  # noqa: SLF001
+
+
+def test_command_timeout_overrides_the_lease(broker: TimersBroker) -> None:
+    subscriber = broker.subscriber("topic", lease_ttl=7, command_timeout=0.5)
+    assert subscriber._config.timer_sub.effective_command_timeout == 0.5  # noqa: SLF001
+
+
+def test_a_route_passes_command_timeout_through(broker: TimersBroker) -> None:
+    async def handler(body: str) -> None: ...
+
+    broker.include_router(TimersRouter(handlers=[TimersRoute(handler, "topic", lease_ttl=7, command_timeout=0.5)]))
+    assert broker.subscribers[0]._config.timer_sub.effective_command_timeout == 0.5  # noqa: SLF001
+
+
+@pytest.mark.parametrize(
+    ("command_timeout", "match"),
+    [
+        (0, "must be > 0"),
+        (-1, "must be > 0"),
+        (float("inf"), "must be finite"),
+        (float("nan"), "must be finite"),
+        (8, "must be <= lease_ttl"),
+    ],
+)
+def test_command_timeout_is_validated(broker: TimersBroker, command_timeout: typing.Any, match: str) -> None:
+    with pytest.raises(ValueError, match=match):
+        broker.subscriber("topic", lease_ttl=7, command_timeout=command_timeout)

tests/test_tuning_defaults.py

@@ -12,6 +12,7 @@
     "max_polling_interval": 5.0,
     "max_concurrent": 5,
     "lease_ttl": 30,
+    "command_timeout": None,
 }
 
 DECLARATION_SITES: typing.Final = (
@@ -28,7 +29,7 @@ def _declared_defaults(site: typing.Any) -> dict[str, typing.Any]:
 
 
 def test_every_declaration_of_a_subscriber_tuning_knob_carries_the_published_default() -> None:
-    """INVARIANT: the four poll-loop knobs default the same at every site that declares them.
+    """INVARIANT: the subscriber tuning knobs default the same at every site that declares them.
 
     Broken by adding a knob, or retuning one, at fewer than all four sites. Each of
     ``TimerSub``, ``create_subscriber``, ``TimersRegistrator.subscriber`` and ``TimersRoute``