chore: clear the safe audit-Low cluster (F4-04, F2-12, F6-02, F6-17 + doc Lows)

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).
 

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:

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

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

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

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]

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`)

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()

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.

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.