refactor(message): remove dead to_delete intent flag

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:

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:

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,15 +456,15 @@ 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
 
 
 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,15 +496,15 @@ 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:
     """B5: a handler that raised without acking must honor the retry strategy, not reject-delete."""
     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