From b6d2429722ab002f21ca1d50e942147c96535eb3 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Wed, 28 Oct 2026 18:42:12 +0300 Subject: [PATCH 1/2] fix: bound subscriber Redis commands so a silent primary can't stall delivery --- README.md | 4 + docs/introduction/how-it-works.md | 1 + docs/usage/subscriber.md | 1 + faststream_redis_timers/parser/parser.py | 11 +- faststream_redis_timers/registrator.py | 2 + faststream_redis_timers/router.py | 2 + faststream_redis_timers/schemas.py | 19 ++ faststream_redis_timers/subscriber/factory.py | 2 + faststream_redis_timers/subscriber/usecase.py | 10 +- tests/test_command_timeout.py | 167 ++++++++++++++++++ tests/test_tuning_defaults.py | 3 +- 11 files changed, 214 insertions(+), 8 deletions(-) create mode 100644 tests/test_command_timeout.py diff --git a/README.md b/README.md index 4d50361..aaae5e2 100644 --- a/README.md +++ b/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 diff --git a/docs/introduction/how-it-works.md b/docs/introduction/how-it-works.md index 550f81c..42b54ed 100644 --- a/docs/introduction/how-it-works.md +++ b/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 diff --git a/docs/usage/subscriber.md b/docs/usage/subscriber.md index 438aab9..68bc89d 100644 --- a/docs/usage/subscriber.md +++ b/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( diff --git a/faststream_redis_timers/parser/parser.py b/faststream_redis_timers/parser/parser.py index f87f18b..4cfc88b 100644 --- a/faststream_redis_timers/parser/parser.py +++ b/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", diff --git a/faststream_redis_timers/registrator.py b/faststream_redis_timers/registrator.py index c705dc9..d902631 100644 --- a/faststream_redis_timers/registrator.py +++ b/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_, diff --git a/faststream_redis_timers/router.py b/faststream_redis_timers/router.py index f87b889..4f15220 100644 --- a/faststream_redis_timers/router.py +++ b/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, diff --git a/faststream_redis_timers/schemas.py b/faststream_redis_timers/schemas.py index 18ffb3c..f74e5c5 100644 --- a/faststream_redis_timers/schemas.py +++ b/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 diff --git a/faststream_redis_timers/subscriber/factory.py b/faststream_redis_timers/subscriber/factory.py index e27ca9d..11af4d5 100644 --- a/faststream_redis_timers/subscriber/factory.py +++ b/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, diff --git a/faststream_redis_timers/subscriber/usecase.py b/faststream_redis_timers/subscriber/usecase.py index 88d1d35..d89cbc5 100644 --- a/faststream_redis_timers/subscriber/usecase.py +++ b/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, diff --git a/tests/test_command_timeout.py b/tests/test_command_timeout.py new file mode 100644 index 0000000..8118226 --- /dev/null +++ b/tests/test_command_timeout.py @@ -0,0 +1,167 @@ +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] + + +@contextlib.asynccontextmanager +async def _side_publisher(redis_client: Redis, keys: dict[str, str]) -> AsyncIterator[TimersBroker]: + publisher = TimersBroker(redis_client, **keys) + async with publisher: + yield publisher + + +def _keys() -> dict[str, str]: + suffix = uuid.uuid4().hex + return {"timeline_key": f"ct_tl_{suffix}", "payloads_key": f"ct_pl_{suffix}"} + + +async def test_a_fetch_hung_on_a_silent_connection_is_abandoned( + proxy: _FreezableProxy, proxied_client: Redis, redis_client: Redis +) -> None: + keys = _keys() + broker = TimersBroker(proxied_client, **keys) + 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, _side_publisher(redis_client, keys) as 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: + keys = _keys() + broker = TimersBroker(proxied_client, **keys) + 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, _side_publisher(redis_client, keys) as 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) diff --git a/tests/test_tuning_defaults.py b/tests/test_tuning_defaults.py index a0b9148..77ed0c5 100644 --- a/tests/test_tuning_defaults.py +++ b/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`` From 4705de8ab12da59afb64ff430e1a779e663359dd Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Wed, 28 Oct 2026 18:51:52 +0300 Subject: [PATCH 2/2] test: pass broker keys explicitly so ty accepts the call --- tests/test_command_timeout.py | 26 +++++++++++--------------- 1 file changed, 11 insertions(+), 15 deletions(-) diff --git a/tests/test_command_timeout.py b/tests/test_command_timeout.py index 8118226..c66e49e 100644 --- a/tests/test_command_timeout.py +++ b/tests/test_command_timeout.py @@ -74,23 +74,20 @@ async def proxied_client(proxy: _FreezableProxy, redis_client: Redis) -> AsyncIt await client.aclose() # ty: ignore[unresolved-attribute] -@contextlib.asynccontextmanager -async def _side_publisher(redis_client: Redis, keys: dict[str, str]) -> AsyncIterator[TimersBroker]: - publisher = TimersBroker(redis_client, **keys) - async with publisher: - yield publisher - - -def _keys() -> dict[str, str]: +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 - return {"timeline_key": f"ct_tl_{suffix}", "payloads_key": f"ct_pl_{suffix}"} + 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: - keys = _keys() - broker = TimersBroker(proxied_client, **keys) + broker, publisher = _brokers(proxied_client, redis_client) seen: list[str] = [] delivered: dict[str, asyncio.Event] = {"before": asyncio.Event(), "after": asyncio.Event()} @@ -99,7 +96,7 @@ async def handler(body: str) -> None: seen.append(body) delivered[body].set() - async with broker, _side_publisher(redis_client, keys) as publisher: + async with broker, publisher: await publisher.publish("before", topic="topic") await asyncio.wait_for(delivered["before"].wait(), timeout=5) proxy.freeze() @@ -111,8 +108,7 @@ async def handler(body: str) -> None: async def test_an_ack_hung_on_a_silent_connection_frees_its_handler_slot( proxy: _FreezableProxy, proxied_client: Redis, redis_client: Redis ) -> None: - keys = _keys() - broker = TimersBroker(proxied_client, **keys) + broker, publisher = _brokers(proxied_client, redis_client) entered = asyncio.Event() release = asyncio.Event() after = asyncio.Event() @@ -125,7 +121,7 @@ async def handler(body: str) -> None: entered.set() await release.wait() - async with broker, _side_publisher(redis_client, keys) as publisher: + async with broker, publisher: await publisher.publish("before", topic="topic") await asyncio.wait_for(entered.wait(), timeout=5) proxy.freeze()