test: close the deferred test-hardening items (F7-07, F7-09, F1-08)

faststream_outbox/testing.py

@@ -57,6 +57,20 @@ class _FakeRow:
     timer_id: str | None = None
 
 
+def _claim_fake_row(row: _FakeRow, *, now: _dt.datetime, token: uuid.UUID) -> None:
+    """
+    Set the lease and bump ``deliveries_count`` — the shared claim mechanics (F1-08).
+
+    Both ``FakeOutboxClient.fetch`` (loop mode) and ``_sync_dispatch`` (sync mode) route
+    through here, so the ``max_deliveries`` boundary is exercised on one path. Eligibility
+    gating (``next_attempt_at <= now``) stays in ``fetch`` only — sync mode deliberately
+    fires future-dated rows immediately, so it claims its row without that gate.
+    """
+    row.acquired_at = now
+    row.acquired_token = token
+    row.deliveries_count += 1
+
+
 class FakeOutboxClient(AbstractOutboxClient):
     """In-memory ``OutboxClient`` substitute. Same surface, list-of-rows storage."""
 
@@ -145,9 +159,7 @@ async def fetch(
             key=lambda r: (r.next_attempt_at, r.id),
         )
         for row in eligible[:limit]:
-            row.acquired_at = now
-            row.acquired_token = token
-            row.deliveries_count += 1
+            _claim_fake_row(row, now=now, token=token)
             out.append(_to_inner(row))
         return out
 
@@ -279,9 +291,7 @@ async def _sync_dispatch(fake_client: FakeOutboxClient, broker: OutboxBroker, qu
     fake_row = next((r for r in fake_client.rows if r.id == row_id), None)
     if fake_row is None:  # pragma: no cover  # defensive: feed just returned this id
         return
-    fake_row.acquired_token = uuid.uuid4()
-    fake_row.acquired_at = utcnow()
-    fake_row.deliveries_count += 1
+    _claim_fake_row(fake_row, now=utcnow(), token=uuid.uuid4())
     await subscriber.dispatch_one(_to_inner(fake_row))
 
 

planning/audits/2026-06-14-deep-audit-pass3-findings.md

@@ -186,4 +186,6 @@ Every confirmed finding is now resolved, documented, consciously dropped, or def
 - **F5-04 code** (`correlation_id` fallback → `gen_cor_id`) — `str(msg.id)` is *better* (stable across re-fetch); fallback unreachable for canonical rows. Doc-only.
 - **F3-04** (structured logging) — `!r` already neutralizes CRLF/log-injection; converting many log sites is churn for marginal benefit.
 
-**Deferred to a follow-up (PR #95):** F7-07 (NOTIFY-wakeup determinism), F7-09 (telemetry exact bounds, 4 files), F1-08 (sync dispatch via `fake_client.fetch` — higher risk).
+**Resolved (test/test-infra, PR #95):** F7-07 (NOTIFY-wakeup tests now use a 30s poll interval — a 15× margin the poll path can't meet, so a broken NOTIFY fails rather than flakes), F7-09 (OTel exact counts + `status="acked"` attribute; Prometheus middleware sums matching samples instead of `max(...)`, so a duplicate series pushes past 1.0 and fails), F1-08 (shared `_claim_fake_row` claim mechanics between `FakeOutboxClient.fetch` and `_sync_dispatch` — same `deliveries_count`/lease path; eligibility gating stays in `fetch` so sync-mode still fires future-dated rows immediately).
+
+**Pass-3 audit fully closed.** Every confirmed finding is resolved, documented, or consciously dropped across PRs #85–#95.

tests/test_fake.py

@@ -648,13 +648,14 @@ async def test_loop_mode_feed_wakes_fetch_loop_via_notify() -> None:
     """
     P30: feeding a row in loop mode wakes the fetch loop via NOTIFY, not the full poll interval.
 
-    With a deliberately slow 5s poll, the handler should still run within 2s — only possible
-    if feed() set the subscriber's _notify_event (the production NOTIFY analogue).
+    With a deliberately slow 30s poll, the handler must still run within 2s — a 15x margin
+    that the poll path provably cannot meet, so only the _notify_event wakeup (the production
+    NOTIFY analogue) can deliver it in time. F7-07: makes a broken NOTIFY fail, not flake.
     """
     broker = _make_broker()
     handled = asyncio.Event()
 
-    @broker.subscriber("orders", min_fetch_interval=5.0, max_fetch_interval=5.0)
+    @broker.subscriber("orders", min_fetch_interval=30.0, max_fetch_interval=30.0)
     async def handle(body: str) -> None:
         del body
         handled.set()
@@ -664,15 +665,15 @@ async def handle(body: str) -> None:
         await asyncio.sleep(0.05)  # let the loop do its first (empty) fetch and enter the idle wait
         payload, hdrs = encode_payload("x")
         test_broker.feed("orders", payload, headers=hdrs)
-        await asyncio.wait_for(handled.wait(), timeout=2.0)  # ~5s without the notify wakeup
+        await asyncio.wait_for(handled.wait(), timeout=2.0)  # ~30s without the notify wakeup
 
 
 async def test_loop_mode_publish_wakes_fetch_loop_via_notify() -> None:
     """P30: broker.publish in loop mode wakes the fetch loop (single-publish notify path)."""
     broker = _make_broker()
     handled = asyncio.Event()
 
-    @broker.subscriber("orders", min_fetch_interval=5.0, max_fetch_interval=5.0)
+    @broker.subscriber("orders", min_fetch_interval=30.0, max_fetch_interval=30.0)
     async def handle(body: str) -> None:
         del body
         handled.set()
@@ -690,7 +691,7 @@ async def test_loop_mode_publish_batch_wakes_fetch_loop_via_notify() -> None:
     handled: list[str] = []
     done = asyncio.Event()
 
-    @broker.subscriber("orders", min_fetch_interval=5.0, max_fetch_interval=5.0)
+    @broker.subscriber("orders", min_fetch_interval=30.0, max_fetch_interval=30.0)
     async def handle(body: str) -> None:
         handled.append(body)
         if len(handled) == 2:

tests/test_metrics_opentelemetry.py

@@ -58,8 +58,10 @@ def test_otel_acked_records_process_duration_histogram() -> None:
     metrics = _collect_metrics(reader)
     assert "messaging.process.duration" in metrics
     points = metrics["messaging.process.duration"].data_points
-    # Histogram point: at least one bucket count > 0.
     assert sum(p.count for p in points) == 1
+    # F7-09: pin the status attribute too — the source's acked branch hinges on it, and a
+    # regression mislabeling the outcome would otherwise pass the count-only check.
+    assert dict(points[0].attributes)["messaging.outbox.status"] == "acked"
 
 
 def test_otel_messages_counter_is_optional() -> None:

tests/test_middleware_opentelemetry.py

@@ -102,7 +102,7 @@ async def handle(body: dict) -> None:
     instruments = _instruments(reader)
     assert "messaging.process.duration" in instruments
     points = instruments["messaging.process.duration"].data_points
-    assert any(p.count >= 1 for p in points)
+    assert sum(p.count for p in points) == 1  # exactly one consume — would catch a double-record
     # Confirm the messaging.system attribute is the canonical short name.
     attrs = dict(points[0].attributes)
     assert attrs["messaging.system"] == "outbox"
@@ -123,7 +123,7 @@ async def handle(body: dict) -> None:
     instruments = _instruments(reader)
     assert "messaging.process.messages" in instruments
     points = instruments["messaging.process.messages"].data_points
-    assert sum(p.value for p in points) >= 1
+    assert sum(p.value for p in points) == 1  # exactly one message consumed
 
 
 async def test_outbox_telemetry_middleware_messages_counter_absent_when_disabled() -> None:

tests/test_middleware_prometheus.py

@@ -83,7 +83,7 @@ async def handle(body: dict) -> None:
         if metric.name == "faststream_received_messages":
             for sample in metric.samples:
                 if sample.name.endswith("_total") and sample.labels.get("broker") == "outbox":
-                    found_value = max(found_value, sample.value)
+                    found_value += sample.value  # F7-09: sum, not max — a duplicate series would push this past 1.0
     assert found_value == 1.0
 
 
@@ -107,7 +107,7 @@ async def handle(body: dict) -> None:
                     and sample.labels.get("broker") == "outbox"
                     and sample.labels.get("status") == "acked"
                 ):
-                    acked = max(acked, sample.value)
+                    acked += sample.value  # F7-09: sum, not max — a duplicate series would push this past 1.0
     assert acked == 1.0
 
 
@@ -128,7 +128,7 @@ async def handle(body: dict) -> None:
         if metric.name == "faststream_received_messages_size_bytes":
             for sample in metric.samples:
                 if sample.name.endswith("_count") and sample.labels.get("broker") == "outbox":
-                    observed = max(observed, sample.value)
+                    observed += sample.value  # F7-09: sum, not max — a duplicate series would push this past 1.0
     assert observed == 1.0
 
 
@@ -153,7 +153,7 @@ async def test_outbox_prometheus_middleware_publish_scope_fires_via_real_broker_
                     and sample.labels.get("destination") == "orders"
                     and sample.labels.get("status") == "success"
                 ):
-                    published = max(published, sample.value)
+                    published += sample.value  # F7-09: sum, not max — a duplicate series would push this past 1.0
     assert published == 1.0