fix: resolve suspected audit findings S1, S2, S4, S5 (S3 already fixed)

faststream_outbox/__init__.py

@@ -1,7 +1,6 @@
 import functools
 import typing
 
-import faststream.asgi.factories.asyncapi.try_it_out
 from faststream._internal.broker import BrokerUsecase
 from faststream._internal.testing.broker import TestBroker
 
@@ -40,6 +39,11 @@
 ]
 
 try:
+    # S4: import inside the guard too — if upstream moves/removes the module, this
+    # raises ImportError here and is tolerated, instead of breaking ``import
+    # faststream_outbox`` from an unguarded top-level import.
+    import faststream.asgi.factories.asyncapi.try_it_out
+
     original_get_broker_registry = faststream.asgi.factories.asyncapi.try_it_out._get_broker_registry  # noqa: SLF001
 
     @functools.lru_cache(maxsize=1)

faststream_outbox/client.py

@@ -392,6 +392,11 @@ async def validate_schema(self) -> None:
         """
         async with self._engine.connect() as conn:
             errors = await conn.run_sync(_validate_schema_sync, self._table)
+            # S2: alembic's autogenerate diff compares index columns + uniqueness but NOT
+            # the partial-index WHERE predicate, so a wrong postgresql_where slips through
+            # and later breaks the producer's ON CONFLICT arbiter. Probe the predicates
+            # directly against the live catalog.
+            errors.extend(await conn.run_sync(_validate_index_predicates_sync, self._table))
             if self._dlq_table is not None:
                 errors.extend(await conn.run_sync(_validate_dlq_schema_sync, self._dlq_table))
         if errors:
@@ -425,6 +430,69 @@ def _row_to_message(row: dict) -> OutboxInnerMessage:
     )
 
 
+def _normalize_predicate(predicate: str) -> str:
+    """Canonicalize a Postgres partial-index predicate for comparison (lowercase, drop parens/space)."""
+    return " ".join(predicate.lower().replace("(", " ").replace(")", " ").split())
+
+
+# Expected partial-index predicates, keyed by the index-name suffix make_outbox_table uses.
+# These are what the fetch CTE and the producer's ON CONFLICT arbiter rely on (S2).
+_EXPECTED_INDEX_PREDICATES = {
+    "_pending_idx": "acquired_token is null",
+    "_timer_id_uq": "timer_id is not null",
+    "_lease_idx": "acquired_token is not null",
+}
+
+# No ``indpred IS NOT NULL`` filter: we must also catch an expected index that exists but
+# was created NON-partial (predicate NULL) — that breaks ON CONFLICT inference the same way
+# a wrong predicate does, and alembic's diff won't flag it either. ``pg_get_expr`` returns
+# NULL for a non-partial index, which we treat as a distinct error below.
+_INDEX_PREDICATE_QUERY = text(
+    "SELECT c.relname AS index_name, pg_get_expr(i.indpred, i.indrelid) AS predicate "
+    "FROM pg_index i "
+    "JOIN pg_class c ON c.oid = i.indexrelid "
+    "JOIN pg_class t ON t.oid = i.indrelid "
+    "JOIN pg_namespace n ON n.oid = t.relnamespace "
+    "WHERE t.relname = :table AND n.nspname = COALESCE(:schema, current_schema())",
+)
+
+
+def _validate_index_predicates_sync(connection: "Connection", table: "Table") -> list[str]:
+    """
+    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:
+
+    * a **wrong** predicate (e.g. ``{table}_timer_id_uq`` built ``WHERE timer_id IS NULL``
+      instead of ``IS NOT NULL``), and
+    * a present-but-**non-partial** index (a plain ``UNIQUE (queue, timer_id)`` with no
+      ``WHERE``) — ``indpred`` is NULL, which alembic also can't distinguish from the partial form.
+
+    An index that is **absent** entirely is left to the alembic existence diff.
+    """
+    rows = (
+        connection.execute(
+            _INDEX_PREDICATE_QUERY,
+            {"table": table.name, "schema": table.schema},
+        )
+        .mappings()
+        .all()
+    )
+    live = {row["index_name"]: row["predicate"] for row in rows}  # predicate is None for a non-partial index
+    errors: list[str] = []
+    for suffix, want in _EXPECTED_INDEX_PREDICATES.items():
+        name = f"{table.name}{suffix}"
+        if name in live:
+            predicate = live[name]
+            if predicate is None:
+                errors.append(f"index {name!r} is not a partial index (expected predicate '{want}')")
+            elif _normalize_predicate(predicate) != want:
+                got = _normalize_predicate(predicate)
+                errors.append(f"index {name!r} has wrong partial predicate: expected '{want}', got '{got}'")
+    return errors
+
+
 def _validate_schema_sync(connection: "Connection", table: "Table") -> list[str]:
     """Run the outbox-table validation pass; see :func:`_run_validate` for the diff machinery."""
     return _run_validate(connection, table, make_outbox_table)

faststream_outbox/subscriber/usecase.py

@@ -86,6 +86,10 @@ class _OutboxConfigError(RuntimeError): ...
 # TCP socket can hang on the kernel keepalive default (~2.5h on Linux) before failing.
 _LISTEN_HEALTH_CHECK_INTERVAL = 30.0
 _LISTEN_HEALTH_CHECK_TIMEOUT = 5.0
+# Bound the graceful close of the LISTEN connection on teardown (S1): a graceful
+# close on the same half-dead socket the health probe just detected can block on
+# the kernel keepalive; cap it and fall back to an immediate terminate.
+_LISTEN_CLOSE_TIMEOUT = 5.0
 
 
 def _compute_backoff(attempt: int, ceiling: float, *, base: float = 1.0) -> float:
@@ -290,7 +294,7 @@ async def _open_fetch_resources(
                 yield {"fetch_conn": fetch_conn, "listen_conn": listen_conn}
             finally:
                 if listen_conn is not None:
-                    await listen_conn.close()
+                    await self._close_listen_connection(listen_conn)
 
     async def _fetch_inner(
         self,
@@ -401,6 +405,21 @@ async def _open_listen_connection(self, engine: "AsyncEngine") -> "_asyncpg.Conn
                     await conn.close()
         return conn if listening else None
 
+    async def _close_listen_connection(self, listen_conn: "_asyncpg.Connection") -> None:
+        """
+        Close the raw LISTEN connection without letting teardown wedge the fetch loop (S1).
+
+        A graceful ``close()`` on a half-dead socket can block on the kernel keepalive
+        (the same socket the bounded health probe may have just flagged). Cap the graceful
+        close, then fall back to ``terminate()`` (immediate, no network round-trip). Both
+        are best-effort — teardown must never raise.
+        """
+        try:
+            await asyncio.wait_for(listen_conn.close(), timeout=_LISTEN_CLOSE_TIMEOUT)
+        except Exception:  # noqa: BLE001  (includes TimeoutError) — fall back to a hard terminate
+            with suppress(Exception):
+                listen_conn.terminate()
+
     def _on_notify(self, *args: object) -> None:
         """
         Asyncpg notification callback: ``(connection, pid, channel, payload)``.

faststream_outbox/testing.py

@@ -353,7 +353,11 @@ async def publish_batch(self, cmd: OutboxPublishCommand) -> None:
         next_at = _compute_next_at_client_side(cmd.activate_in, cmd.activate_at)
         recorder = self._broker.config.broker_config.metrics_recorder
         total_size = 0
-        landed = 0
+        landed_ids: list[int] = []
+        # S5: insert the WHOLE batch first, then emit ``published``, then dispatch. In
+        # production the batch INSERT commits atomically before the subscriber fetches, so
+        # a handler never sees a half-inserted batch and ``published`` precedes delivery.
+        # Dispatching per-body mid-feed (the old shape) reproduced neither property.
         for body in bodies:
             payload, hdrs = _encode_payload(body, headers=cmd.headers, serializer=self._serializer)
             total_size += len(payload)
@@ -364,20 +368,21 @@ async def publish_batch(self, cmd: OutboxPublishCommand) -> None:
                 next_attempt_at=next_at,
             )
             if row_id is not None:
-                landed += 1
-            if not self._run_loops and row_id is not None:
-                await _sync_dispatch(self._fake_client, self._broker, cmd.queue, row_id)
+                landed_ids.append(row_id)
         _safe_emit(
             recorder,
             "published",
             {
                 "queue": cmd.queue,
                 "status": "success",
-                "count": landed,
+                "count": len(landed_ids),
                 "size_bytes": total_size,
                 "duration_seconds": 0.0,
             },
         )
+        if not self._run_loops:
+            for row_id in landed_ids:
+                await _sync_dispatch(self._fake_client, self._broker, cmd.queue, row_id)
 
     async def request(self, cmd: OutboxPublishCommand) -> typing.NoReturn:
         msg = "OutboxBroker does not support request-reply"
@@ -476,7 +481,10 @@ async def fake_publish_batch(
         next_at = _compute_next_at_client_side(activate_in, activate_at)
         recorder = broker.config.broker_config.metrics_recorder
         total_size = 0
-        landed = 0
+        landed_ids: list[int] = []
+        # S5: insert the whole batch, emit ``published``, then dispatch — mirroring the
+        # production order (atomic batch INSERT → published → subscriber fetch) so handlers
+        # never observe a half-inserted batch and the event order isn't inverted.
         for body in bodies:
             payload, hdrs = _encode_payload(body, headers=headers, serializer=serializer)
             total_size += len(payload)
@@ -487,20 +495,21 @@ async def fake_publish_batch(
                 next_attempt_at=next_at,
             )
             if row_id is not None:
-                landed += 1
-            if not run_loops and row_id is not None:
-                await _sync_dispatch(fake_client, broker, queue, row_id)
+                landed_ids.append(row_id)
         _safe_emit(
             recorder,
             "published",
             {
                 "queue": queue,
                 "status": "success",
-                "count": landed,
+                "count": len(landed_ids),
                 "size_bytes": total_size,
                 "duration_seconds": 0.0,
             },
         )
+        if not run_loops:
+            for row_id in landed_ids:
+                await _sync_dispatch(fake_client, broker, queue, row_id)
 
     return fake_publish_batch
 

planning/active/2026-06-12-code-audit-findings.md

@@ -86,6 +86,30 @@ additional test findings, improvements P1–P35, and the doc / CLAUDE.md drift i
 | S4 | __init__.py:4,42-57 | The try-it-out resilience guard protects the attribute access but the module import itself sits unguarded at top level — an upstream module move breaks `import faststream_outbox` entirely, defeating the guard's stated purpose |
 | S5 | testing.py:341-364,461-484 | Sync-mode batch publish dispatches handlers per-body mid-feed (handlers observe half-inserted batches — impossible in production) and emits `published` after the handlers ran (inverted event order) |
 
+### Suspected — resolution (2026-06-13)
+
+All five investigated against current `main`; each confirmed or dismissed (no longer "unproven"):
+
+- **S1 — confirmed, fixed.** The teardown `listen_conn.close()` was unbounded. Now bounded
+  via `asyncio.wait_for(..., _LISTEN_CLOSE_TIMEOUT)` with a `terminate()` fallback, so a
+  half-dead socket can't wedge the fetch loop on shutdown.
+- **S2 — confirmed, fixed.** A docker probe showed `validate_schema()` passes with a wrong
+  `timer_id_uq` predicate (alembic's diff ignores `postgresql_where`). Added
+  `_validate_index_predicates_sync` — a `pg_get_expr(indpred, …)` probe comparing each
+  partial index's predicate against the expected value.
+- **S3 — dismissed (already resolved by P17).** Emitting `acked`/`nacked_*` only after a
+  successful flush means a lease-lost row emits `lease_lost` *instead*, never a paired
+  ack/nack — so `processed_total` no longer double-counts. Pinned by
+  `test_dispatch_one_lease_lost_emits_only_lease_lost_not_acked`.
+- **S4 — confirmed, fixed.** The `try_it_out` import sat outside the resilience guard.
+  Moved inside the `try/except (AttributeError, ImportError)` so an upstream module
+  move/rename no longer breaks `import faststream_outbox`.
+- **S5 — confirmed, fixed.** Sync-mode batch publish dispatched each handler mid-feed and
+  emitted `published` last. Now inserts the whole batch, emits `published`, then dispatches
+  — mirroring production (atomic batch INSERT → published → subscriber fetch).
+
+Fixes shipped together; see PR for tests.
+
 ## Bug details (what the table can't carry)
 
 **B1 + B2 (drain spin / dead restart).** The two-flag drain design keeps

tests/test_fake.py

@@ -104,6 +104,36 @@ async def handle(body: dict) -> None:
     assert test_broker.fake_client.rows == []  # row deleted after ack
 
 
+async def test_sync_batch_inserts_whole_batch_before_dispatch_and_published_first() -> None:
+    """S5: sync-mode batch inserts the whole batch before any handler runs, and ``published`` precedes delivery."""
+    events: list[str] = []
+    seen_counts: list[int] = []
+    metadata = MetaData()
+    t = make_outbox_table(metadata)
+
+    def _recorder(event: str, tags: Mapping[str, typing.Any]) -> None:
+        del tags
+        events.append(event)
+
+    broker = OutboxBroker(outbox_table=t, metrics_recorder=_recorder)
+
+    @broker.subscriber("orders")
+    async def handle(body: str) -> None:
+        del body
+        seen_counts.append(len(test_broker.fake_client.rows))
+
+    test_broker = TestOutboxBroker(broker)
+    async with test_broker:
+        await broker.publish_batch("a", "b", "c", queue="orders")  # ty: ignore[missing-argument]
+
+    # The whole 3-row batch was present before the first handler ran (the old per-feed
+    # dispatch would have shown only 1), and ``published`` was emitted before delivery.
+    assert seen_counts
+    assert seen_counts[0] == 3
+    assert events
+    assert events[0] == "published"
+
+
 async def test_fake_broker_publish_batch_triggers_handler() -> None:
     broker = _make_broker()
     received: list[str] = []

tests/test_integration.py

@@ -45,6 +45,43 @@ async def _row_count(engine: AsyncEngine, table) -> int:
         return len(result.all())
 
 
+async def test_validate_schema_detects_wrong_partial_index_predicate(
+    pg_engine: AsyncEngine,
+    outbox_table: Table,
+) -> None:
+    """S2: a partial index built with the wrong WHERE predicate is caught (alembic's diff ignores it)."""
+    idx = f"{outbox_table.name}_timer_id_uq"
+    async with pg_engine.begin() as conn:
+        await conn.execute(text(f'DROP INDEX "{idx}"'))
+        # Recreate with the wrong predicate: should be ``timer_id IS NOT NULL``.
+        await conn.execute(
+            text(f'CREATE UNIQUE INDEX "{idx}" ON {outbox_table.name} (queue, timer_id) WHERE timer_id IS NULL'),
+        )
+    client = OutboxClient(pg_engine, outbox_table)
+    with pytest.raises(RuntimeError, match="wrong partial predicate"):
+        await client.validate_schema()
+
+
+async def test_validate_schema_detects_non_partial_index_on_present_index(
+    pg_engine: AsyncEngine,
+    outbox_table: Table,
+) -> None:
+    """
+    S2 (review #1): an expected partial index recreated NON-partial breaks ON CONFLICT too, and is caught.
+
+    Alembic's diff can't distinguish a plain UNIQUE (queue, timer_id) from the partial
+    form (it ignores postgresql_where), and the indpred-NULL row would otherwise be
+    silently skipped by the probe.
+    """
+    idx = f"{outbox_table.name}_timer_id_uq"
+    async with pg_engine.begin() as conn:
+        await conn.execute(text(f'DROP INDEX "{idx}"'))
+        await conn.execute(text(f'CREATE UNIQUE INDEX "{idx}" ON {outbox_table.name} (queue, timer_id)'))
+    client = OutboxClient(pg_engine, outbox_table)
+    with pytest.raises(RuntimeError, match="not a partial index"):
+        await client.validate_schema()
+
+
 async def test_validate_schema_passes_for_correct_table(pg_engine, outbox_table) -> None:
     client = OutboxClient(pg_engine, outbox_table)
     await client.validate_schema()  # should not raise

tests/test_unit.py

@@ -2363,6 +2363,32 @@ async def test_open_listen_connection_closes_conn_when_add_listener_fails() -> N
     fake_conn.close.assert_awaited_once()
 
 
+async def test_close_listen_connection_graceful_close_succeeds() -> None:
+    """S1: a healthy graceful close is awaited; terminate() is not used."""
+    sub = _make_subscriber_for_listener_test()
+    conn = MagicMock()
+    conn.close = AsyncMock()
+    conn.terminate = MagicMock()
+    await sub._close_listen_connection(conn)  # noqa: SLF001
+    conn.close.assert_awaited_once()
+    conn.terminate.assert_not_called()
+
+
+async def test_close_listen_connection_falls_back_to_terminate_on_hang(monkeypatch: pytest.MonkeyPatch) -> None:
+    """S1: a graceful close that hangs on a half-dead socket is bounded and falls back to terminate()."""
+    monkeypatch.setattr("faststream_outbox.subscriber.usecase._LISTEN_CLOSE_TIMEOUT", 0.05)
+    sub = _make_subscriber_for_listener_test()
+    conn = MagicMock()
+
+    async def _hang() -> None:
+        await asyncio.sleep(3600)
+
+    conn.close = _hang
+    conn.terminate = MagicMock()
+    await sub._close_listen_connection(conn)  # noqa: SLF001  # must return promptly, not hang
+    conn.terminate.assert_called_once()
+
+
 async def test_subscriber_start_resets_stopping_flag() -> None:
     """B2: start() clears _stopping so a stop()->start() cycle fetches again instead of hot-spinning."""
     sub = _make_subscriber_for_listener_test()