diff --git a/CLAUDE.md b/CLAUDE.md index 8e00f80..25745f7 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -54,9 +54,9 @@ Deep dive: `architecture/relay.md`. User-facing: `docs/usage/relay.md`. ### Timers (delayed delivery) -`activate_in: timedelta` / `activate_at: datetime` (mutually exclusive) set `next_attempt_at`; the fetch CTE's `next_attempt_at <= now()` gates eligibility. `publish` computes server-side via `make_interval` (clock-skew-safe); `publish_batch` is client-side. +`activate_in: timedelta` / `activate_at: datetime` (mutually exclusive) set `next_attempt_at`; the fetch CTE's `next_attempt_at <= now()` gates eligibility. For `publish`, `activate_in` is computed server-side via `make_interval` (clock-skew-safe) while `activate_at` is bound as the caller's absolute literal; `publish_batch` is fully client-side. -`timer_id` (single `publish` only) → partial unique index `(queue, timer_id) WHERE timer_id IS NOT NULL`. Producer uses `pg_insert(...).on_conflict_do_nothing(...)` — re-publishing the same id is a no-op (returns `None`). NOTIFY is skipped when future-dated OR the conflict suppressed the insert. +`timer_id` (single `publish` only) → partial unique index `(queue, timer_id) WHERE timer_id IS NOT NULL`. Producer uses `pg_insert(...).on_conflict_do_nothing(...)` — re-publishing the same id is a no-op (returns `None`). NOTIFY is skipped when future-dated OR the conflict suppressed the insert. The dedup window is **one *live* row per `(queue, timer_id)`** — it resets once the row is delivered (DELETEd) or terminally fails, so `timer_id` is "at most one in flight", not a global once-ever idempotency key (the DLQ keeps `timer_id` non-unique). `broker.cancel_timer(*, queue, timer_id, session)` issues `DELETE WHERE queue=? AND timer_id=? AND acquired_token IS NULL` — **the `acquired_token IS NULL` guard is load-bearing** (preserves the lease-token invariant; returns `False` if a handler is in flight). diff --git a/faststream_outbox/broker.py b/faststream_outbox/broker.py index b2822f1..27b40b7 100644 --- a/faststream_outbox/broker.py +++ b/faststream_outbox/broker.py @@ -292,12 +292,9 @@ def _warn_on_unstarted_foreign_publishers(self) -> None: for call in sub.calls: for pub in call.handler._publishers: # noqa: SLF001 outer = pub._outer_config # noqa: SLF001 # ty: ignore[unresolved-attribute] - if outer is self.config: # pragma: no cover - # Internal outbox publisher in a decorator chain — not intended. - # Handlers should use broker.publish() directly, not decorate with - # an internal publisher. This branch exists for completeness but is - # unreachable in normal usage. - continue # pragma: no cover + # An internal outbox publisher's ``outer`` is this broker's own config, + # which carries a (truthy) ``producer`` — so the wired-producer check + # below already skips it; no separate ``outer is self.config`` branch needed. producer = getattr(outer, "producer", None) if producer: continue # already wired / started @@ -502,6 +499,10 @@ async def cancel_timer( Treat ``False`` as "no cancellation needed from your transaction", not as "cancellation failed". + + Like :meth:`publish`, this runs on your session **without committing** — a returned + ``True`` only becomes durable when your transaction commits. If you branch on it and + then roll back, the cancellation rolls back with you. """ if not isinstance(session, AsyncSession): msg = "broker.cancel_timer requires an sqlalchemy.ext.asyncio.AsyncSession" @@ -525,8 +526,9 @@ async def fetch_unprocessed( """ Return outbox rows currently in the table — pending, in-flight, or future-dated. - Intended for test assertions: a successful delivery deletes the row, so anything - still in the table is "unprocessed". Pass *queue* to filter to a single queue; + Intended for test assertions and lease-free operator inspection (the lease-free + read path `get_one()`/`__aiter__()` point you here): a successful delivery deletes + the row, so anything still in the table is "unprocessed". Pass *queue* to filter to a single queue; omit it to return rows across all queues. *limit* caps the result set (default 1000) so an accidental call against a backlogged production table does not OOM the process. Runs on the caller's session (same transactional @@ -536,6 +538,12 @@ async def fetch_unprocessed( if not isinstance(session, AsyncSession): msg = "broker.fetch_unprocessed requires an sqlalchemy.ext.asyncio.AsyncSession" raise TypeError(msg) + if limit < 1: + # F4-04: a non-positive limit otherwise hits SQL (LIMIT -1 → DB error) or + # silently returns nothing (LIMIT 0); reject it up front, consistently with + # the fake. + msg = f"limit must be >= 1, got {limit}" + raise ValueError(msg) t = self._outbox_table stmt = select(*t.c).order_by(t.c.id).limit(limit) if queue is not None: diff --git a/faststream_outbox/client.py b/faststream_outbox/client.py index a460fb6..2164bc6 100644 --- a/faststream_outbox/client.py +++ b/faststream_outbox/client.py @@ -16,6 +16,7 @@ """ import abc +import asyncio import datetime as _dt import uuid from typing import TYPE_CHECKING @@ -58,6 +59,10 @@ from alembic.migration import MigrationContext as _AlembicMigrationContext +# Upper bound on the ``ping()`` liveness probe so a half-dead socket can't hang it. +_PING_TIMEOUT_SECONDS = 5.0 + + class AbstractOutboxClient(abc.ABC): """ Outbox client interface. @@ -408,10 +413,13 @@ async def validate_schema(self) -> None: raise RuntimeError(msg) async def ping(self) -> bool: + # Bound the probe: an unwrapped connect+SELECT 1 against a half-dead TCP socket + # can hang on the kernel keepalive default (~hours), defeating ping()'s purpose + # as a liveness check (F2-12; mirrors the LISTEN health probe's wait_for). try: - async with self._engine.connect() as conn: + async with asyncio.timeout(_PING_TIMEOUT_SECONDS), self._engine.connect() as conn: await conn.execute(text("SELECT 1")) - except Exception: # noqa: BLE001 + except Exception: # noqa: BLE001 # includes TimeoutError on a hung probe return False return True @@ -465,8 +473,9 @@ def _validate_index_predicates_sync(connection: "Connection", table: "Table") -> """ Compare the live partial-index WHERE predicates against what the package expects (S2). - Alembic's index diff ignores ``postgresql_where``, so two drifts pass :func:`_run_validate` - yet break the producer's ``ON CONFLICT`` arbiter inference at publish time, both flagged here: + Alembic's index diff ignores ``postgresql_where``, so the alembic autogenerate pass + (:func:`_run_validate`) does not catch two drifts that break the producer's ``ON CONFLICT`` + arbiter inference at publish time — both flagged by this separate probe: * a **wrong** predicate (e.g. ``{table}_timer_id_uq`` built ``WHERE timer_id IS NULL`` instead of ``IS NOT NULL``), and diff --git a/faststream_outbox/metrics/prometheus.py b/faststream_outbox/metrics/prometheus.py index feb4204..add2495 100644 --- a/faststream_outbox/metrics/prometheus.py +++ b/faststream_outbox/metrics/prometheus.py @@ -291,7 +291,7 @@ def __call__(self, event: str, tags: Mapping[str, typing.Any]) -> None: # noqa: # below is the canonical error counter; ``_published_duration`` records # every attempt (with the status label) so failed-publish latency stays # observable. - count = tags.get("count", 1) + count = tags.get("count", 0) # default 0 matches the OTel adapter (F6-02); producer always sets it if status == "error": # P28: an error lands 0 messages but is one failed publish. Increment the # status="error" series (the exact label dashboards alert on) by 1, rather diff --git a/faststream_outbox/subscriber/usecase.py b/faststream_outbox/subscriber/usecase.py index 5c29971..9139b94 100644 --- a/faststream_outbox/subscriber/usecase.py +++ b/faststream_outbox/subscriber/usecase.py @@ -856,8 +856,8 @@ async def process_message(self, msg: OutboxInnerMessage) -> "Response": # noqa: Outbox-specific process_message — header propagation (G3) hook. Optionally fills empty Response headers with the inbound message's - headers when ``propagate_inbound_headers=True``. Task 5 adds the - OutboxResponse + foreign-publisher dual-fire guard (G1) here too. + headers when ``propagate_inbound_headers=True``, and runs the + OutboxResponse + foreign-publisher dual-fire guard here too. # Upstream equivalent (replaced): # SubscriberUsecase.process_message diff --git a/faststream_outbox/testing.py b/faststream_outbox/testing.py index 5cebdf1..10a52f6 100644 --- a/faststream_outbox/testing.py +++ b/faststream_outbox/testing.py @@ -549,6 +549,11 @@ async def fake_fetch_unprocessed( limit: int = 1000, ) -> list[OutboxInnerMessage]: del session + if limit < 1: + # Mirror the real broker's validation so a non-positive limit can't + # silently mis-slice (rows[:-1]) under the test broker (F4-04). + msg = f"limit must be >= 1, got {limit}" + raise ValueError(msg) rows = sorted(fake_client.rows, key=lambda r: r.id) if queue is not None: rows = [r for r in rows if r.queue == queue] 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 dc30b3e..bae4a77 100644 --- a/planning/audits/2026-06-14-deep-audit-pass3-findings.md +++ b/planning/audits/2026-06-14-deep-audit-pass3-findings.md @@ -83,6 +83,8 @@ None. ### Low +> **Safe-Lows batch RESOLVED (2026-06-14, PR #89)** — code: **F4-04** (`fetch_unprocessed` rejects `limit < 1`, real + fake), **F2-12** (`ping()` bounded by `asyncio.timeout(_PING_TIMEOUT_SECONDS)`), **F6-02** (Prometheus `published` count default unified to `0`, matching OTel), **F6-17** (removed the dead `outer is self.config` branch). Docs/comments: **F6-01** (stale "Task 5"), **F6-04** (`_run_validate` phrasing), **F6-13** (CLAUDE.md `make_interval`/`activate_at`), **F6-15** (`fetch_unprocessed` docstring), **F1-05** (`cancel_timer` commit-contingent return), **F2-06** (`timer_id` dedup window). Still open after this batch: **F2-10** (validate `indisunique` — needs a drifted-schema integration test, own change), **F6-05** (`_utcnow` dedup — needs a shared-module decision), **F5-04** (parser correlation_id fallback — subtle), and the larger refactor / test-hardening Lows below. + **Validation symmetry (gaps left by PR #83's eager-validation fix):** > **F4-01/F4-02/F4-06/F4-10 RESOLVED (2026-06-14, PR #86)** — collapsed into one shared `_validate_publish_args(context, *, queue, session, activate_in, activate_at)` in `response.py` (order: activate-args → session → queue), called by the `OutboxPublishCommand` constructor, `OutboxResponse.__init__`, and `broker.publish_batch`'s empty branch. Tests: `tests/test_unit.py::test_outbox_response_rejects_{empty_queue,non_str_queue,non_async_session}_eagerly` + `::test_broker_publish_batch_empty_rejects_empty_queue`. Three dual-fire tests that passed `session=None` to `OutboxResponse` as a shortcut were corrected to a valid session — exactly the latent misuse F4-02 surfaces. **F4-04 (negative-limit `fetch_unprocessed`) remains open.** - **[F4-01]** `OutboxResponse.__init__` validated `activate_*`/tz eagerly but **not `queue`** (non-str/empty/>255) — a bad `queue` deferred to `as_publish_command()` at dispatch, masquerading as a handler failure and burning the inbound row's retry budget. The exact footgun #83 closed for `activate_*`, left open for `queue`. CONFIRMED. (`response.py:139-162`) diff --git a/tests/test_fake.py b/tests/test_fake.py index d5185de..3689e45 100644 --- a/tests/test_fake.py +++ b/tests/test_fake.py @@ -440,6 +440,16 @@ async def test_fake_broker_fetch_unprocessed_respects_limit() -> None: assert len(all_rows) == 5 +async def test_fake_broker_fetch_unprocessed_rejects_non_positive_limit() -> None: + """F4-04: the test broker mirrors the real limit validation (no silent rows[:-1] divergence on limit=-1).""" + broker = _make_broker() + test_broker = TestOutboxBroker(broker) + async with test_broker: + for bad in (0, -1): + with pytest.raises(ValueError, match="limit"): + await broker.fetch_unprocessed(limit=bad) # ty: ignore[missing-argument] + + async def test_fake_broker_router_subscriber_receives_publish() -> None: received: list[str] = [] router = OutboxRouter() diff --git a/tests/test_metrics_prometheus.py b/tests/test_metrics_prometheus.py index 0fa6b6a..959930f 100644 --- a/tests/test_metrics_prometheus.py +++ b/tests/test_metrics_prometheus.py @@ -132,6 +132,20 @@ def test_prometheus_published_event_uses_destination_label() -> None: ) +def test_prometheus_published_without_count_defaults_to_zero() -> None: + """F6-02: a published success event missing ``count`` records no message (default 0, matching the OTel adapter).""" + reg, rec = _make_recorder() + rec("published", {"queue": "q", "status": "success", "duration_seconds": 0.001}) + assert ( + _sample( + reg, + "faststream_published_messages_total", + {"app_name": "", "broker": "outbox", "destination": "q", "status": "success"}, + ) + is None + ) + + def test_prometheus_published_error_status_total_fires_at_count_zero() -> None: """ P28: a status="error" published event (count=0) must increment the error-status total. diff --git a/tests/test_unit.py b/tests/test_unit.py index 624f76e..b691393 100644 --- a/tests/test_unit.py +++ b/tests/test_unit.py @@ -657,6 +657,27 @@ async def test_broker_ping_when_engine_query_fails() -> None: assert await broker.ping() is False +async def test_broker_ping_times_out_on_hung_query(monkeypatch: pytest.MonkeyPatch) -> None: + """F2-12: ping() bounds SELECT 1 with a timeout so a half-dead socket can't hang the liveness probe.""" + monkeypatch.setattr("faststream_outbox.client._PING_TIMEOUT_SECONDS", 0.01) + + class _HangConn: + async def execute(self, *_args: object, **_kwargs: object) -> None: + await asyncio.sleep(1.0) # longer than the patched timeout → asyncio.timeout cancels it + + class _ConnCtx: + async def __aenter__(self) -> "_HangConn": + return _HangConn() + + async def __aexit__(self, *_exc: object) -> bool: + return False + + engine = AsyncMock() + engine.connect = _ConnCtx # AsyncEngine.connect() returns an async CM (sync call) + broker = _make_broker(engine) + assert await broker.ping() is False + + async def test_broker_stop_logs_subscriber_failure_and_completes() -> None: """A raising sub.stop must be logged via the helper, swallowed, and not re-raised.""" broker = _make_broker() @@ -3595,6 +3616,14 @@ async def test_broker_publish_batch_empty_rejects_empty_queue() -> None: await broker.publish_batch(queue="", session=_make_session_mock()) # no bodies +async def test_fetch_unprocessed_rejects_non_positive_limit() -> None: + """F4-04: a non-positive limit raises rather than hitting SQL (limit=-1) or silently returning none (limit=0).""" + broker = _make_broker() + for bad in (0, -1): + with pytest.raises(ValueError, match="limit"): + await broker.fetch_unprocessed(session=_make_session_mock(), limit=bad) + + def test_asyncapi_document_populates_channels_and_operations() -> None: """ Regression: ``BrokerSpec(url=[])`` produced a structurally empty AsyncAPI document.