fix: preserve row in dispatch_one when consume() early-exits on shutdown

If stop() flips running=False between a worker pulling a row from
_inflight and dispatch_one entering consume(), FastStream's base
SubscriberUsecase.consume() returns None via its early-exit guard
without invoking the handler. dispatch_one previously fell into
assert_state_set → reject() → _safe_flush, silently DELETEing the
row and emitting an 'acked' metric for work that never ran.

Restructure the consume() block so a None return + running=False
short-circuits before the flush path; the lease lives until
lease_ttl_seconds expires and another replica reclaims the row on
next start. Behavior is preserved for the consume()-raises path
because the outer except already returned without flushing, making
the previous finally-driven state_set mutation a no-op.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

# Conflicts:
#	CLAUDE.md

Author: lesnik512

CLAUDE.md

@@ -76,6 +76,8 @@ Lease-loss is logged at WARNING with `extra={"event": "lease_lost", "phase": "te
 
 **Writer-connection autocommit.** `_open_worker_resources` configures the per-worker writer connection with `isolation_level="AUTOCOMMIT"`. The two terminal-state writes (`delete_with_lease`, `mark_pending_with_lease`) are each a single statement, so an explicit BEGIN/COMMIT would just add two Postgres round-trips per row with no benefit. Autocommit collapses the per-row cost from three round-trips (BEGIN + DELETE/UPDATE + COMMIT) to one while preserving the lease-token guard — the `WHERE acquired_token = …` clause is what enforces the invariant, not the transaction wrapping the statement. The fetch connection is **not** autocommit-configured: it owns LISTEN/NOTIFY and the CTE-update flow is paid once per batch, where the BEGIN/COMMIT amortizes naturally.
 
+**Shutdown race in `dispatch_one`.** If `stop()` flips `running=False` between a worker pulling a row from `_inflight` and `consume()` being entered, FastStream's base `SubscriberUsecase.consume()` early-exits (`if not self.running: return None`) without running the handler. `dispatch_one` detects this state — `not row.state_set and not self.running` after `consume()` returns without raising — and returns before the `assert_state_set → reject() → _safe_flush` path that would otherwise silently DELETE the row. The lease lives until `lease_ttl_seconds` expires and another replica reclaims the row on next start. No metric fires at this site (the row is neither acked nor nacked). Without this guard, busy subscribers leak rows on every rolling deploy.
+
 ### Test broker
 
 `TestOutboxBroker` (in `testing.py`) swaps in a `FakeOutboxClient` (in-memory list of `_FakeRow` dicts). Two dispatch modes:

faststream_outbox/subscriber/usecase.py

@@ -439,17 +439,22 @@ async def dispatch_one(
         )
         start_perf = time.perf_counter()
         try:
-            try:
-                await self.consume(row)
-            finally:
-                await row.assert_state_set(logger)
+            await self.consume(row)
         except Exception as e:  # noqa: BLE001
             # No metric emitted here intentionally: the row was never marked
             # terminal/retry, so its state is undefined — flushing or emitting an
             # ack/nack would lie. The lease will expire and the row will be
             # reclaimed; the ERROR log is the operator signal.
             self._log(log_level=logging.ERROR, message=f"Outbox worker error: {e!r}", exc_info=e)
             return
+        # Shutdown race: SubscriberUsecase.consume() returns None without invoking
+        # process_message when self.running has been flipped to False by stop().
+        # Detecting that here lets us preserve the row instead of falling through
+        # to assert_state_set → reject() → _safe_flush → DELETE. The row's lease
+        # expires after lease_ttl_seconds and is reclaimed on next start.
+        if not row.state_set and not self.running:
+            return
+        await row.assert_state_set(logger)
         duration_seconds = time.perf_counter() - start_perf
         common = {**base, "deliveries_count": row.deliveries_count, "duration_seconds": duration_seconds}
         if row.last_exception is None:

tests/test_unit.py

@@ -1578,6 +1578,42 @@ async def _boom(_row: object) -> None:
             await sub.dispatch_one(msg, writer_conn=None)
 
 
+async def test_dispatch_one_preserves_row_when_consume_early_exits_on_shutdown() -> None:
+    """
+    Shutdown race in dispatch_one.
+
+    ``SubscriberUsecase.consume()`` returns ``None`` without invoking
+    ``process_message`` when ``running`` is False. Previously ``dispatch_one`` fell
+    into ``assert_state_set → reject() → _safe_flush`` and silently DELETEd the row.
+    The early-return guard preserves the row so its lease expires and another
+    replica reclaims it.
+    """
+    events, recorder = _events_recorder()
+    metadata = MetaData()
+    table = make_outbox_table(metadata)
+    broker = OutboxBroker(outbox_table=table, metrics_recorder=recorder)
+
+    @broker.subscriber("orders")
+    async def handle(body: dict) -> None: ...
+
+    fake = FakeOutboxClient()
+    test_broker = TestOutboxBroker(broker)
+    test_broker.fake_client = fake
+    msg = _make_msg(queue="orders")
+
+    async with test_broker:
+        sub = next(iter(broker._subscribers))  # noqa: SLF001
+        # Simulate the race: a worker has pulled a row from _inflight, then stop()
+        # flipped running to False before dispatch_one's consume() call.
+        sub.running = False
+        with patch.object(fake, "delete_with_lease", new=AsyncMock(return_value=True)) as delete_spy:
+            await sub.dispatch_one(msg, writer_conn=None)
+
+    delete_spy.assert_not_awaited()
+    assert not any(e == "acked" for e, _ in events)
+    assert not any(e.startswith("nacked") for e, _ in events)
+
+
 async def test_flush_terminal_logs_lease_lost_at_warning_with_structured_fields() -> None:
     """M7: when delete returns rowcount=0 (lease reclaimed) the broker emits a WARNING with structured fields."""