diff --git a/CLAUDE.md b/CLAUDE.md index 462ed7b..357e358 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -30,7 +30,7 @@ The package wires a FastStream `Broker`/`Registrator`/`Subscriber` trio whose tr - **Relay to foreign broker** — native FastStream publisher-chain (`@kafka_pub @broker_outbox.subscriber("q")`); bad chain composition raises `_OutboxConfigError` and retries via **lease expiry**, not the retry strategy. → [relay.md](architecture/relay.md) - **Timers** — `activate_in`/`activate_at` are mutually exclusive and gate eligibility; `timer_id` is "at most one *live* row per `(queue, timer_id)`", not a global idempotency key; `cancel_timer`'s `acquired_token IS NULL` guard is load-bearing. → [timers.md](architecture/timers.md) - **User-owned schema** — caller owns the table; the three partial indexes + the `_lease_ck` CHECK are load-bearing; **no `state` column**; `validate_schema()` is opt-in. → [schema.md](architecture/schema.md) -- **Opt-in DLQ** — with `dlq_table=None` every path is **bit-for-bit identical**; the `DLQFailureReason` `Literal` is the **public contract** (changes are API-breaking); branch on `terminal_failure_reason` **before** `last_exception`. → [dlq.md](architecture/dlq.md) +- **Opt-in DLQ** — with `dlq_table=None` every path is **bit-for-bit identical**; the `DLQFailureReason` `Literal` is the **public contract** (changes are API-breaking); delivery records one disjoint `Outcome` (`Ack`/`Retry`/`Terminal`) that `dispatch_one` matches on (`terminal_failure_reason`/`pending_delay_seconds`/`state_set` are read-only views). → [dlq.md](architecture/dlq.md) - **Two-loop subscriber + lease-token invariant** — fetch loop + N worker loops; **every terminal write filters on `acquired_token`** (a stale write finds `rowcount==0` and is dropped) — any new fetch/terminal path must preserve this; `AckPolicy.ACK_FIRST` is rejected at registration; `lease_ttl_seconds` must exceed handler P99. → [subscriber.md](architecture/subscriber.md) - **Drain on stop** — custom `stop()` overrides + the `dispatch_one` shutdown-race guard prevent row leaks on rolling deploys; **re-check both overrides when touching shutdown**. → [drain.md](architecture/drain.md) - **Test broker** — `TestOutboxBroker` swaps in `FakeOutboxClient`; sync mode dispatches the handler before `publish` returns; `test_client_contract.py` pins real-vs-fake parity. → [test-broker.md](architecture/test-broker.md) diff --git a/architecture/dlq.md b/architecture/dlq.md index 1810d51..70a53ea 100644 --- a/architecture/dlq.md +++ b/architecture/dlq.md @@ -12,15 +12,15 @@ User-facing: `docs/usage/dlq.md`. Invariant summary: `CLAUDE.md` § Opt-in DLQ. The CTE's `RETURNING` / `INSERT` / `SELECT` column lists are not hand-written — they derive from `_DLQ_PROJECTION` (the `(outbox_col, dlq_col)` pairs copied verbatim) plus `_DLQ_INJECTED_COLUMNS` (`failure_reason`, `last_exception`, supplied by the caller) in `schema.py`. The fake (`FakeOutboxClient.delete_with_lease`) builds its audit dict from the same constants, so the two substrates can't drift on which columns the archive carries — a DLQ column change is one edit in `schema.py`, verified across both adapters by `tests/test_client_contract.py`. `failed_at` is not in the projection; it rides the DLQ column's `server_default`. -## `terminal_failure_reason` routing +## Terminal-outcome routing -`OutboxInnerMessage.terminal_failure_reason` is set on the three failure paths: +`OutboxInnerMessage.outcome` records a single canonical `Outcome` (`message.py`): `Ack` (success), `Retry(delay_seconds)`, or `Terminal(reason)`. The three failure paths record a `Terminal`: -- `allow_delivery` False → `"max_deliveries"` -- `_nack` strategy-exhausted → `"retry_terminal"` -- `_reject` → `"rejected"` +- `allow_delivery` False → `Terminal("max_deliveries")` +- `_nack` strategy-exhausted → `Terminal("retry_terminal")` +- `_reject` → `Terminal("rejected")` -`_flush_terminal` reads it to decide whether to build a DLQ payload; `dispatch_one` also reads it to pick the `nacked_terminal(reason=…)` tag value. **Branch on `terminal_failure_reason` BEFORE `last_exception`**, so manual `await msg.reject()` (no exception raised) routes correctly to `nacked_terminal(reason="rejected")` instead of the previously-incorrect `acked`. Success (`_ack`) leaves the field None; success rows never touch the DLQ. +`terminal_failure_reason` is now a **read-only view** of `outcome` (`Terminal.reason`, else `None`), so existing readers are unchanged: `_flush_terminal` reads it to decide whether to build a DLQ payload; the `nacked_terminal(reason=…)` tag value comes from the same reason. `dispatch_one` **matches on the disjoint `Outcome` variant** — `Terminal → nacked_terminal`, `Retry → reschedule`, `Ack → delete`. Because the variants are mutually exclusive there is no ordering dependence between the arms: a manual `await msg.reject()` (no exception raised) records `Terminal("rejected")` directly, independent of `last_exception`. Success (`_ack` → `Ack`) leaves `terminal_failure_reason` None; success rows never touch the DLQ. **The `DLQFailureReason` `Literal` type (`message.py`) is the public contract** for this string — operator queries and dashboard labels key off these values, so changing them is API-breaking. diff --git a/faststream_outbox/message.py b/faststream_outbox/message.py index 412366c..6cafa07 100644 --- a/faststream_outbox/message.py +++ b/faststream_outbox/message.py @@ -36,6 +36,32 @@ DLQFailureReason = Literal["max_deliveries", "retry_terminal", "rejected"] +# Canonical delivery outcome. ``ack``/``nack``/``reject`` record exactly one of these +# on ``OutboxInnerMessage.outcome``; the worker loop matches on the variant to pick the +# DB write. The legacy ``state_set``/``terminal_failure_reason``/``pending_delay_seconds`` +# attributes are read-only views derived from it. +@dataclass(frozen=True, slots=True) +class Ack: + """Handler succeeded; the row is deleted (no DLQ).""" + + +@dataclass(frozen=True, slots=True) +class Retry: + """Handler failed; the strategy scheduled another attempt after ``delay_seconds``.""" + + delay_seconds: float + + +@dataclass(frozen=True, slots=True) +class Terminal: + """Delivery failed terminally; the row is deleted (and DLQ'd when configured).""" + + reason: DLQFailureReason + + +Outcome = Ack | Retry | Terminal + + # Header keys the outbox envelope owns on a row's ``headers`` dict. ``_encode_payload`` # (``envelope.py``) writes them, ``OutboxParser.parse_message`` (``parser.py``) reads # them back, and the relay header-propagation (``_maybe_propagate_inbound_headers``) @@ -50,9 +76,11 @@ 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 (``terminal_failure_reason``, - ``pending_delay_seconds``). The worker loop reads those flags and issues the - actual DB write, scoped by ``acquired_token``. + The ack/nack/reject methods record a single canonical ``Outcome`` + (``Ack``/``Retry``/``Terminal``) on ``outcome``. The worker loop matches on the + variant and issues the actual DB write, scoped by ``acquired_token``. + ``state_set``/``terminal_failure_reason``/``pending_delay_seconds`` are read-only + views of ``outcome`` kept for existing readers. """ id: int @@ -75,16 +103,32 @@ class OutboxInnerMessage: retry_strategy: "RetryStrategyProto | None" = None last_exception: BaseException | None = None - state_set: 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) - # Set on terminal-failure paths (``allow_delivery`` False, ``_nack`` exhausted, - # ``_reject``). ``_flush_terminal`` reads it to decide whether to build a DLQ - # payload; ``dispatch_one`` reads it to pick the ``nacked_terminal`` reason - # tag. Stays ``None`` on the success (``_ack``) path so handler-success - # never touches the DLQ. - terminal_failure_reason: "DLQFailureReason | None" = field(default=None, init=False) + # Canonical delivery outcome, set exactly once by ``_ack``/``_nack``/``_reject``/ + # ``allow_delivery``. ``None`` until intent is recorded. The three properties below + # derive the legacy views the worker loop and metrics read. + outcome: "Outcome | None" = field(default=None, init=False) + + @property + def state_set(self) -> bool: + return self.outcome is not None + + @property + def terminal_failure_reason(self) -> "DLQFailureReason | None": + """Terminal reason (``allow_delivery`` False, ``_nack`` exhausted, ``_reject``), else ``None``. + + ``_flush_terminal`` reads it to decide whether to build a DLQ payload; + ``dispatch_one`` reads it to pick the ``nacked_terminal`` reason tag. ``None`` + on the ``Ack``/``Retry`` paths so handler-success never touches the DLQ. + """ + return self.outcome.reason if isinstance(self.outcome, Terminal) else None + + @property + def pending_delay_seconds(self) -> float | None: + """Retry delay set by ``_nack`` when the strategy schedules a retry, else ``None``. + + Consumed by ``_flush_retry`` to drive ``mark_pending_with_lease``. + """ + return self.outcome.delay_seconds if isinstance(self.outcome, Retry) else None # ``**_options`` are accepted-and-ignored: FastStream's AcknowledgementMiddleware # forwards ``message.nack(**extra_options)`` for native idioms like @@ -103,13 +147,13 @@ async def reject(self, **_options: Any) -> None: await self._update_state_if_not_set(self._reject) async def _update_state_if_not_set(self, fn: Callable[[], Awaitable[None]]) -> None: - if self.state_set: + if self.outcome is not None: return await fn() - self.state_set = True async def _ack(self) -> None: self._record_attempt() + self.outcome = Ack() async def _nack(self) -> None: self._record_attempt() @@ -135,16 +179,16 @@ async def _nack(self) -> None: self.retry_strategy, self, ) - self.terminal_failure_reason = "retry_terminal" + self.outcome = Terminal("retry_terminal") return if delay is None: - self.terminal_failure_reason = "retry_terminal" + self.outcome = Terminal("retry_terminal") else: - self.pending_delay_seconds = delay + self.outcome = Retry(delay) async def _reject(self) -> None: self._record_attempt() - self.terminal_failure_reason = "rejected" + self.outcome = Terminal("rejected") def _record_attempt(self) -> None: self.attempts_count += 1 @@ -156,8 +200,7 @@ 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.state_set = True - self.terminal_failure_reason = "max_deliveries" + self.outcome = Terminal("max_deliveries") if logger is not None: logger.log( logging.ERROR, diff --git a/faststream_outbox/subscriber/usecase.py b/faststream_outbox/subscriber/usecase.py index e6cabd9..65447cb 100644 --- a/faststream_outbox/subscriber/usecase.py +++ b/faststream_outbox/subscriber/usecase.py @@ -39,7 +39,7 @@ from faststream.specification.schema import Message, Operation, SubscriberSpec from typing_extensions import override -from faststream_outbox.message import ENVELOPE_MANAGED_HEADERS, OutboxInnerMessage +from faststream_outbox.message import ENVELOPE_MANAGED_HEADERS, OutboxInnerMessage, Retry, Terminal from faststream_outbox.parser.parser import OutboxParser from faststream_outbox.publisher.fake import OutboxFakePublisher from faststream_outbox.response import OutboxResponse @@ -701,23 +701,23 @@ async def dispatch_one( # linear pipeline: guard, consume, branch on outcome, f await row.assert_state_set(logger) duration_seconds = time.perf_counter() - start_perf common = {**base, "deliveries_count": row.deliveries_count, "duration_seconds": duration_seconds} - # Branch on ``terminal_failure_reason`` (set by ``_nack``/``_reject``) before - # ``last_exception`` so manual ``msg.reject()`` (no exception raised) still - # surfaces as ``nacked_terminal(reason="rejected")``, and ack-policy-driven - # rejects (REJECT_ON_ERROR) carry ``reason="rejected"`` rather than the - # incorrect ``"retry_terminal"`` they got under the old ``last_exception``-first - # ordering. Successful handlers leave ``terminal_failure_reason=None``. - if row.terminal_failure_reason is not None: - terminal_tags = self._with_exception_type({**common, "reason": row.terminal_failure_reason}, row) - event, tags = "nacked_terminal", terminal_tags - elif row.pending_delay_seconds is not None: - retry_tags = self._with_exception_type({**common, "next_delay_seconds": row.pending_delay_seconds}, row) - # Retry is a per-row UPDATE -- never batched. Flush inline, unchanged. + # Match on the disjoint ``Outcome`` variant recorded by ack/nack/reject: + # Terminal -> nacked_terminal (delete + DLQ), Retry -> reschedule, Ack -> delete. + # The variants are mutually exclusive, so there is no ordering dependence between + # the arms (e.g. a manual ``msg.reject()`` records ``Terminal("rejected")`` + # directly, independent of ``last_exception``). + outcome = row.outcome + if isinstance(outcome, Terminal): + tags = self._with_exception_type({**common, "reason": outcome.reason}, row) + event = "nacked_terminal" + elif isinstance(outcome, Retry): + retry_tags = self._with_exception_type({**common, "next_delay_seconds": outcome.delay_seconds}, row) + # Retry is a per-row UPDATE -- never batched. Flush inline. # P17: emit only after the flush lands; a lease-lost update emits lease_lost instead. if await self._safe_flush(row, terminal=False, writer_conn=writer_conn): self._emit_metric("nacked_retried", retry_tags) return False - else: + else: # Ack (assert_state_set guaranteed outcome is set, so this is Ack, never None) event, tags = "acked", common # Terminal delete: batch it if batchable, else flush inline. P17 holds either way — # the metric fires only when the delete lands (inline here, or deferred in _flush_buffer). diff --git a/planning/changes/2026-07-17.02-outcome-sum-type.md b/planning/changes/2026-07-17.02-outcome-sum-type.md new file mode 100644 index 0000000..add7567 --- /dev/null +++ b/planning/changes/2026-07-17.02-outcome-sum-type.md @@ -0,0 +1,55 @@ +--- +summary: Replaced OutboxInnerMessage's terminal_failure_reason / pending_delay_seconds / state_set fields with one canonical Outcome sum type (Ack | Retry | Terminal); the old attributes become read-only views. Behavior-preserving; dissolves the terminal-before-retry ordering invariant. +--- + +# Change: Canonical Outcome sum type on OutboxInnerMessage + +**Lane:** lightweight. A behavior-preserving refactor of the load-bearing +terminal/dispatch path that reshapes an internal representation and dissolves a +documented ordering invariant — recorded here so the change is legible. + +## Goal + +Deepen `OutboxInnerMessage`: `ack`/`nack`/`reject`/`allow_delivery` recorded delivery +intent across three loosely-coupled fields (`terminal_failure_reason`, +`pending_delay_seconds`, `state_set`), where "success" was the implicit "all None" +state and "reason and delay both set" was representable-but-never-used. Replace them +with one canonical `Outcome` sum type so success is an explicit variant and the +impossible combination is unrepresentable. + +## Approach (Option A — canonical outcome + derived views) + +- `message.py` defines `Ack` / `Retry(delay_seconds)` / `Terminal(reason)` + (frozen, slotted) and `Outcome = Ack | Retry | Terminal`. `OutboxInnerMessage.outcome` + is the single canonical field; the setters assign it (`_ack` → `Ack()`, `_nack` → + `Retry`/`Terminal("retry_terminal")`, `_reject` → `Terminal("rejected")`, + `allow_delivery` → `Terminal("max_deliveries")`). +- `terminal_failure_reason` / `pending_delay_seconds` / `state_set` become **read-only + `@property` views** of `outcome` (zero-drift derivations, no setters). So the five + other subscriber readers (`_flush_terminal`, `_flush_retry`, `_terminal_has_dlq`, the + shutdown-race `state_set` guard, metric tags) and every read-assertion in the tests + are unchanged. +- `dispatch_one` matches on the disjoint variant (`Terminal → nacked_terminal`, + `Retry → reschedule`, `Ack → delete`). Because the variants are mutually exclusive, + the **terminal-before-retry / before-`last_exception` ordering invariant is dissolved** + — recorded in `architecture/dlq.md` and the `CLAUDE.md` DLQ invariant line. + +The `DLQFailureReason` `Literal` (the public wire contract queried by operators) is +unchanged — it is now `Terminal.reason`'s type. `dlq_table=None` stays bit-for-bit +identical. + +## Files + +- `faststream_outbox/message.py` — `Outcome` types; `outcome` field + three property + views; setters wired to `outcome`. +- `faststream_outbox/subscriber/usecase.py` — `dispatch_one` matches on `outcome` + (imports `Retry`, `Terminal`); no other reader changed. +- `tests/test_unit.py` — the 12 sites that *wrote* the old fields now set `outcome` + (`Retry(...)` / `Terminal(...)`); read-assertions unchanged (properties). +- `architecture/dlq.md`, `CLAUDE.md` — truth-home promotion of the routing model. + +## Verification + +- [x] `just test` — full suite green: 603 passed, Postgres 17, coverage 100.00%. +- [x] `just lint` — clean (ruff + ty). +- [x] `just check-planning` — passes. diff --git a/tests/test_unit.py b/tests/test_unit.py index a783c1d..d24c193 100644 --- a/tests/test_unit.py +++ b/tests/test_unit.py @@ -45,7 +45,7 @@ from faststream_outbox.client import OutboxClient from faststream_outbox.configs import OutboxBrokerConfig from faststream_outbox.envelope import _encode_payload -from faststream_outbox.message import OutboxInnerMessage, OutboxMessage +from faststream_outbox.message import OutboxInnerMessage, OutboxMessage, Retry, Terminal from faststream_outbox.parser.parser import OutboxParser from faststream_outbox.publisher.config import OutboxPublisherSpecificationConfig from faststream_outbox.publisher.fake import OutboxFakePublisher @@ -2065,7 +2065,7 @@ async def test_flush_retry_threads_writer_conn_into_mark_pending(writer_conn: ob fake = FakeOutboxClient() broker, test_broker = _make_broker_for_dispatch(fake) msg = _make_msg() - msg.pending_delay_seconds = 1.0 # set post-construction; field is init=False + msg.outcome = Retry(1.0) # set post-construction; field is init=False conn_arg = MagicMock() if writer_conn == "sentinel" else None with patch.object(fake, "mark_pending_with_lease", new=AsyncMock(return_value=True)) as spy: @@ -2301,7 +2301,7 @@ async def mark_pending_with_lease(self, *args: object, **kwargs: object) -> bool broker, test_broker = _make_broker_for_dispatch(LeaseLostFake()) msg = _make_msg(id=99, queue="orders", deliveries_count=2) - msg.pending_delay_seconds = 1.0 + msg.outcome = Retry(1.0) async with test_broker: sub = next(iter(broker._subscribers)) # noqa: SLF001 @@ -2329,7 +2329,7 @@ async def mark_pending_with_lease(self, *args: object, **kwargs: object) -> bool broker, test_broker = _make_broker_for_dispatch(RaisingFake()) msg = _make_msg() - msg.pending_delay_seconds = 1.0 + msg.outcome = Retry(1.0) async with test_broker: sub = next(iter(broker._subscribers)) # noqa: SLF001 @@ -3258,7 +3258,7 @@ async def mark_pending_with_lease(self, *args: object, **kwargs: object) -> bool test_broker = TestOutboxBroker(broker) test_broker.fake_client = LeaseLostFake() msg = _make_msg(id=99, queue="orders", deliveries_count=2) - msg.pending_delay_seconds = 1.0 + msg.outcome = Retry(1.0) async with test_broker: sub = next(iter(broker._subscribers)) # noqa: SLF001 @@ -3717,7 +3717,7 @@ async def test_flush_terminal_builds_dlq_payload_when_failure_reason_set() -> No fake = FakeOutboxClient() test_broker.fake_client = fake msg = _make_msg(id=7, queue="orders", deliveries_count=3) - msg.terminal_failure_reason = "max_deliveries" + msg.outcome = Terminal("max_deliveries") msg.last_exception = RuntimeError("boom") with patch.object(fake, "delete_with_lease", new=AsyncMock(return_value=True)) as spy: @@ -3755,7 +3755,7 @@ async def test_flush_terminal_no_dlq_payload_when_dlq_unconfigured() -> None: fake = FakeOutboxClient() broker, test_broker = _make_broker_for_dispatch(fake) msg = _make_msg(id=9, queue="orders", deliveries_count=3) - msg.terminal_failure_reason = "max_deliveries" + msg.outcome = Terminal("max_deliveries") with patch.object(fake, "delete_with_lease", new=AsyncMock(return_value=True)) as spy: async with test_broker: @@ -3773,7 +3773,7 @@ async def test_flush_terminal_emits_dlq_written_metric_after_successful_delete() fake = FakeOutboxClient() test_broker.fake_client = fake msg = _make_msg(id=10, queue="orders", deliveries_count=3) - msg.terminal_failure_reason = "retry_terminal" + msg.outcome = Terminal("retry_terminal") msg.last_exception = RuntimeError("boom") with patch.object(fake, "delete_with_lease", new=AsyncMock(return_value=True)): @@ -3795,7 +3795,7 @@ async def test_flush_terminal_does_not_emit_dlq_written_on_lease_lost() -> None: fake = FakeOutboxClient() test_broker.fake_client = fake msg = _make_msg(id=11, queue="orders", deliveries_count=3) - msg.terminal_failure_reason = "max_deliveries" + msg.outcome = Terminal("max_deliveries") with patch.object(fake, "delete_with_lease", new=AsyncMock(return_value=False)): async with test_broker: @@ -3812,7 +3812,7 @@ async def test_flush_terminal_dlq_payload_includes_repr_of_exception() -> None: fake = FakeOutboxClient() test_broker.fake_client = fake msg = _make_msg(id=12, queue="orders") - msg.terminal_failure_reason = "rejected" + msg.outcome = Terminal("rejected") msg.last_exception = ValueError("invalid payload") with patch.object(fake, "delete_with_lease", new=AsyncMock(return_value=True)) as spy: @@ -3831,7 +3831,7 @@ async def test_flush_terminal_dlq_payload_last_exception_none_when_no_exc() -> N fake = FakeOutboxClient() test_broker.fake_client = fake msg = _make_msg(id=13, queue="orders") - msg.terminal_failure_reason = "rejected" + msg.outcome = Terminal("rejected") # last_exception is None — operator chose to drop, no exception context. with patch.object(fake, "delete_with_lease", new=AsyncMock(return_value=True)) as spy: @@ -3854,7 +3854,7 @@ async def test_flush_terminal_dlq_payload_truncates_long_exception_repr() -> Non fake = FakeOutboxClient() test_broker.fake_client = fake msg = _make_msg(id=14, queue="orders") - msg.terminal_failure_reason = "retry_terminal" + msg.outcome = Terminal("retry_terminal") # Build an exception whose ``repr`` is much larger than the cap. huge_payload = "x" * (_LAST_EXCEPTION_MAX_CHARS * 3) msg.last_exception = RuntimeError(huge_payload) @@ -3877,7 +3877,7 @@ async def test_flush_terminal_dlq_payload_short_exception_not_truncated() -> Non fake = FakeOutboxClient() test_broker.fake_client = fake msg = _make_msg(id=15, queue="orders") - msg.terminal_failure_reason = "retry_terminal" + msg.outcome = Terminal("retry_terminal") msg.last_exception = ValueError("short message") with patch.object(fake, "delete_with_lease", new=AsyncMock(return_value=True)) as spy: