From 6e9f09d5309d48eb35d96a60f2f97e9c6d7adcdf Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Fri, 17 Jul 2026 09:49:42 +0300 Subject: [PATCH] refactor(message): remove dead to_delete intent flag `OutboxInnerMessage.to_delete` was set on every ack/terminal path but read by zero production code: `dispatch_one` branches on `terminal_failure_reason` then `pending_delay_seconds`, and `_flush_terminal`/`_terminal_has_dlq` read `terminal_failure_reason`. The flag was a misleading third field a maintainer could edit while changing nothing. Remove the field and its 5 assignments; re-express the 16 test assertions on the live decision inputs the dispatcher actually reads (`terminal_failure_reason` / `pending_delay_seconds`), so the tests now pin the real outcome. Reword the one stale comment/docstring that named the flag. Behavior-preserving. Full suite green (599 passed, coverage 100%), just lint + ty clean. Co-Authored-By: Claude Opus 4.8 (1M context) --- faststream_outbox/message.py | 8 +------ faststream_outbox/subscriber/usecase.py | 2 +- tests/test_unit.py | 29 +++++++++++-------------- 3 files changed, 15 insertions(+), 24 deletions(-) diff --git a/faststream_outbox/message.py b/faststream_outbox/message.py index 74cdd88..223da85 100644 --- a/faststream_outbox/message.py +++ b/faststream_outbox/message.py @@ -40,7 +40,7 @@ class OutboxInnerMessage: """In-memory copy of a claimed outbox row, plus ack/nack/reject intent helpers. - The ack/nack/reject methods set in-memory intent flags (``to_delete``, + The ack/nack/reject methods set in-memory intent flags (``terminal_failure_reason``, ``pending_delay_seconds``). The worker loop reads those flags and issues the actual DB write, scoped by ``acquired_token``. """ @@ -66,7 +66,6 @@ class OutboxInnerMessage: last_exception: BaseException | None = None state_set: bool = field(default=False, init=False) - to_delete: bool = field(default=False, init=False) # Set by ``_nack`` when the strategy schedules a retry; consumed by the # subscriber's ``_flush_retry`` to drive ``mark_pending_with_lease``. pending_delay_seconds: float | None = field(default=None, init=False) @@ -101,7 +100,6 @@ async def _update_state_if_not_set(self, fn: Callable[[], Awaitable[None]]) -> N async def _ack(self) -> None: self._record_attempt() - self.to_delete = True async def _nack(self) -> None: self._record_attempt() @@ -127,18 +125,15 @@ async def _nack(self) -> None: self.retry_strategy, self, ) - self.to_delete = True self.terminal_failure_reason = "retry_terminal" return if delay is None: - self.to_delete = True self.terminal_failure_reason = "retry_terminal" else: self.pending_delay_seconds = delay async def _reject(self) -> None: self._record_attempt() - self.to_delete = True self.terminal_failure_reason = "rejected" def _record_attempt(self) -> None: @@ -151,7 +146,6 @@ def _record_attempt(self) -> None: def allow_delivery(self, *, max_deliveries: int | None, logger: "LoggerProto | None") -> bool: """If ``max_deliveries`` is set and exceeded, mark for deletion without invoking the handler.""" if max_deliveries is not None and self.deliveries_count > max_deliveries: - self.to_delete = True self.state_set = True self.terminal_failure_reason = "max_deliveries" if logger is not None: diff --git a/faststream_outbox/subscriber/usecase.py b/faststream_outbox/subscriber/usecase.py index 068520f..ee95b96 100644 --- a/faststream_outbox/subscriber/usecase.py +++ b/faststream_outbox/subscriber/usecase.py @@ -857,7 +857,7 @@ async def _flush_terminal( return False # Build the DLQ payload only when this row is terminal-by-failure AND the # broker is configured with a DLQ table. Success-by-ack rows reach this - # method too (terminal=True via to_delete) but carry + # method too (routed here as a plain terminal delete) but carry # ``terminal_failure_reason is None`` and must not land in the DLQ. dlq_payload: dict[str, typing.Any] | None = None if row.terminal_failure_reason is not None and self._outer_config.dlq_table is not None: diff --git a/tests/test_unit.py b/tests/test_unit.py index c0ed811..dc60da0 100644 --- a/tests/test_unit.py +++ b/tests/test_unit.py @@ -371,14 +371,15 @@ def _make_msg(**overrides: object) -> OutboxInnerMessage: async def test_inner_message_ack_marks_for_delete() -> None: msg = _make_msg() await msg.ack() - assert msg.to_delete + assert msg.terminal_failure_reason is None + assert msg.pending_delay_seconds is None assert msg.state_set async def test_inner_message_nack_with_no_strategy_is_terminal() -> None: msg = _make_msg() await msg.nack() - assert msg.to_delete + assert msg.terminal_failure_reason == "retry_terminal" def test_constant_retry_rejects_oversize_jitter() -> None: @@ -455,7 +456,7 @@ async def h2(body: dict) -> None: ... async def test_inner_message_nack_with_strategy_schedules_retry() -> None: msg = _make_msg(retry_strategy=ConstantRetry(delay_seconds=60)) await msg.nack() - assert not msg.to_delete + assert msg.terminal_failure_reason is None # retry scheduled, not terminal assert msg.last_attempt_at is not None assert msg.pending_delay_seconds == 60.0 @@ -463,7 +464,7 @@ async def test_inner_message_nack_with_strategy_schedules_retry() -> None: async def test_inner_message_reject_is_terminal() -> None: msg = _make_msg() await msg.reject() - assert msg.to_delete + assert msg.terminal_failure_reason == "rejected" async def test_inner_message_double_ack_is_noop() -> None: @@ -477,13 +478,13 @@ async def test_inner_message_double_ack_is_noop() -> None: def test_allow_delivery_under_cap() -> None: msg = _make_msg(deliveries_count=3) assert msg.allow_delivery(max_deliveries=5, logger=None) is True - assert not msg.to_delete + assert not msg.state_set def test_allow_delivery_exceeds_cap_marks_for_delete() -> None: msg = _make_msg(deliveries_count=10) assert msg.allow_delivery(max_deliveries=5, logger=None) is False - assert msg.to_delete + assert msg.terminal_failure_reason == "max_deliveries" def test_allow_delivery_no_cap_is_always_true() -> None: @@ -495,7 +496,7 @@ async def test_assert_state_set_rejects_when_not_set() -> None: msg = _make_msg() await msg.assert_state_set(logger=None) assert msg.state_set - assert msg.to_delete # reject path → terminal + assert msg.terminal_failure_reason == "rejected" # reject path → terminal async def test_assert_state_set_with_exception_nacks_for_retry() -> None: @@ -503,7 +504,7 @@ async def test_assert_state_set_with_exception_nacks_for_retry() -> None: msg = _make_msg(retry_strategy=ConstantRetry(delay_seconds=60), last_exception=RuntimeError("boom")) await msg.assert_state_set(logger=None) assert msg.state_set - assert not msg.to_delete # nack scheduled a retry, did not delete + assert msg.pending_delay_seconds is not None # nack scheduled a retry, did not delete assert msg.pending_delay_seconds == 60.0 assert msg.terminal_failure_reason is None @@ -513,7 +514,6 @@ async def test_assert_state_set_with_exception_no_strategy_is_terminal_retry() - msg = _make_msg(last_exception=RuntimeError("boom")) await msg.assert_state_set(logger=None) assert msg.state_set - assert msg.to_delete assert msg.terminal_failure_reason == "retry_terminal" @@ -552,7 +552,6 @@ async def test_nack_with_raising_strategy_degrades_to_retry_terminal() -> None: msg = _make_msg(retry_strategy=_RaisingRetryStrategy()) await msg.nack() assert msg.state_set - assert msg.to_delete assert msg.terminal_failure_reason == "retry_terminal" @@ -1631,7 +1630,7 @@ async def test_outbox_message_reject_calls_raw_then_super() -> None: correlation_id="1", ) await msg.reject() - assert inner.to_delete # raw_message.reject ran + assert inner.terminal_failure_reason == "rejected" # raw_message.reject ran assert msg.committed is not None # super().reject ran @@ -2171,7 +2170,7 @@ async def handle(body: dict) -> None: ... delete_spy.assert_not_awaited() assert not msg.state_set - assert not msg.to_delete + assert msg.terminal_failure_reason is None # untouched row, no terminal intent assert not any(e == "acked" for e, _ in events) assert not any(e.startswith("nacked") for e, _ in events) @@ -3664,7 +3663,6 @@ async def test_terminal_failure_reason_set_on_max_deliveries() -> None: async def test_terminal_failure_reason_set_on_retry_terminal_without_strategy() -> None: msg = _make_msg() await msg.nack() - assert msg.to_delete assert msg.terminal_failure_reason == "retry_terminal" @@ -3684,14 +3682,13 @@ async def test_terminal_failure_reason_unset_when_retry_scheduled() -> None: async def test_terminal_failure_reason_set_on_reject() -> None: msg = _make_msg() await msg.reject() - assert msg.to_delete assert msg.terminal_failure_reason == "rejected" async def test_terminal_failure_reason_unset_on_ack() -> None: msg = _make_msg() await msg.ack() - assert msg.to_delete + assert msg.pending_delay_seconds is None # ack path: plain delete, no retry assert msg.terminal_failure_reason is None @@ -3736,7 +3733,7 @@ async def test_flush_terminal_builds_dlq_payload_when_failure_reason_set() -> No async def test_flush_terminal_no_dlq_payload_on_ack_path() -> None: - """Success-by-ack reaches _flush_terminal too (via to_delete) but reason stays None → no DLQ.""" + """Success-by-ack reaches _flush_terminal too (as a plain terminal delete) but reason stays None → no DLQ.""" broker, test_broker = _make_broker_with_dlq() fake = FakeOutboxClient() test_broker.fake_client = fake