diff --git a/faststream_outbox/testing.py b/faststream_outbox/testing.py index f88b648..1bacfad 100644 --- a/faststream_outbox/testing.py +++ b/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)) diff --git a/planning/audits/2026-06-14-deep-audit-pass3-findings.md b/planning/audits/2026-06-14-deep-audit-pass3-findings.md index c7d6420..a0cba6c 100644 --- a/planning/audits/2026-06-14-deep-audit-pass3-findings.md +++ b/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. diff --git a/tests/test_fake.py b/tests/test_fake.py index b02d643..17a3451 100644 --- a/tests/test_fake.py +++ b/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,7 +665,7 @@ 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: @@ -672,7 +673,7 @@ async def test_loop_mode_publish_wakes_fetch_loop_via_notify() -> None: 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: diff --git a/tests/test_metrics_opentelemetry.py b/tests/test_metrics_opentelemetry.py index 93dcd76..f6552b5 100644 --- a/tests/test_metrics_opentelemetry.py +++ b/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: diff --git a/tests/test_middleware_opentelemetry.py b/tests/test_middleware_opentelemetry.py index e49fdf2..e7fbc77 100644 --- a/tests/test_middleware_opentelemetry.py +++ b/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: diff --git a/tests/test_middleware_prometheus.py b/tests/test_middleware_prometheus.py index 9675782..07d492f 100644 --- a/tests/test_middleware_prometheus.py +++ b/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