fix: remove corrupt timers on decode so the parser has no side effects (#113)

Author: lesnik512

faststream_redis_timers/message.py

@@ -21,16 +21,26 @@ class TimerStreamMessage(StreamMessage["TimerMessage"]):
     (with its score pushed forward by `lease_ttl`) while the handler runs.
     `ack()` and `reject()` atomically remove it; `nack()` is a no-op so the
     lease expires and another worker re-claims the timer.
+
+    A timer whose payload cannot be parsed is removed and raises on `decode()`,
+    so the handler never runs.
     """
 
     def __init__(
         self,
         *args: typing.Any,
         _remove: "Callable[[], Awaitable[None]] | None" = None,
+        _discard_corrupt: "Callable[[], Awaitable[typing.NoReturn]] | None" = None,
         **kwargs: typing.Any,
     ) -> None:
         super().__init__(*args, **kwargs)
         self._remove = _remove
+        self._discard_corrupt = _discard_corrupt
+
+    async def decode(self) -> typing.Any:
+        if self._discard_corrupt is not None:
+            await self._discard_corrupt()
+        return await super().decode()
 
     async def _commit(self) -> None:
         if self._remove is None:

faststream_redis_timers/parser/parser.py

@@ -19,19 +19,13 @@ def __init__(self, config: "TimersSubscriberConfig") -> None:
 
     async def parse_message(self, msg: "TimerMessage") -> TimerStreamMessage:
         timer_id = msg["timer_id"]
-        outer_config = self._config._outer_config  # noqa: SLF001
-        store = outer_config.store
+        store = self._config._outer_config.store  # noqa: SLF001
+        discard_corrupt = None
         try:
             body, headers = TimerMessageFormat.parse(msg["data"])
         except ValueError as e:
-            await store.remove(self._config.full_topic, timer_id)
-            outer_config.logger.log(
-                f"Timer {timer_id!r} on {self._config.full_topic!r} removed: "
-                f"its {len(msg['data'])}-byte payload cannot be parsed",
-                logging.ERROR,
-                exc_info=e,
-            )
-            raise
+            body, headers = msg["data"], {}
+            discard_corrupt = partial(self._discard_corrupt, msg, e)
         return TimerStreamMessage(
             raw_message=msg,
             body=body,
@@ -41,7 +35,19 @@ async def parse_message(self, msg: "TimerMessage") -> TimerStreamMessage:
             correlation_id=headers.get("correlation_id", timer_id),
             reply_to=headers.get("reply_to", ""),
             _remove=partial(store.remove, self._config.full_topic, timer_id),
+            _discard_corrupt=discard_corrupt,
+        )
+
+    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"])
+        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",
+            logging.ERROR,
+            exc_info=error,
         )
+        raise error
 
     async def decode_message(self, msg: TimerStreamMessage) -> typing.Any:
         return decode_message(msg)

tests/test_envelope.py

@@ -3,10 +3,12 @@
 import logging
 import os
 import time
+import typing
 import uuid
 
 import pytest
 from faststream import Context
+from faststream.message import StreamMessage
 from redis.asyncio import Redis
 
 from faststream_redis_timers import TimersBroker
@@ -199,7 +201,7 @@ async def test_legacy_envelope_with_non_hex_body_is_removed_and_logged_once(
     seen: list[bytes] = []
 
     @broker.subscriber("topic")
-    async def handler(body: bytes) -> None:  # pragma: no cover - never invoked; the parser rejects the timer
+    async def handler(body: bytes) -> None:  # pragma: no cover - never invoked; the timer is corrupt
         seen.append(body)
 
     with caplog.at_level(logging.ERROR, logger=f"corrupt-{suffix}"):
@@ -212,3 +214,34 @@ async def handler(body: bytes) -> None:  # pragma: no cover - never invoked; the
     removals = [record for record in caplog.records if "removed" in record.getMessage()]
     assert len(removals) == 1
     assert "'old-timer'" in removals[0].getMessage()
+    assert not [record for record in caplog.records if record.levelno == logging.CRITICAL]
+
+
+async def test_corrupt_timer_never_reaches_handler_with_custom_decoder(redis_client: Redis) -> None:
+    suffix = uuid.uuid4().hex
+    broker = TimersBroker(
+        redis_client,
+        timeline_key=f"corrupt_dec_tl_{suffix}",
+        payloads_key=f"corrupt_dec_pl_{suffix}",
+    )
+    timeline_key = f"corrupt_dec_tl_{suffix}:topic"
+    payloads_key = f"corrupt_dec_pl_{suffix}:topic"
+
+    await redis_client.zadd(timeline_key, {"old-timer": time.time() - 1})
+    await redis_client.hset(payloads_key, "old-timer", b'{"b": "zz"}')
+
+    async def raw_body_decoder(message: StreamMessage[typing.Any]) -> bytes:  # pragma: no cover - never invoked
+        return message.body
+
+    seen: list[bytes] = []
+
+    @broker.subscriber("topic", decoder=raw_body_decoder)
+    async def handler(body: bytes) -> None:  # pragma: no cover - never invoked; the timer is corrupt
+        seen.append(body)
+
+    async with broker:
+        await asyncio.sleep(0.3)
+
+    assert seen == []
+    assert await redis_client.zscore(timeline_key, "old-timer") is None
+    assert not await redis_client.hexists(payloads_key, "old-timer")