refactor: DLQ CTE in Core, prebuilt publish INSERTs, drop dead code

faststream_outbox/client.py

@@ -30,6 +30,7 @@
     cast,
     delete,
     func,
+    insert,
     or_,
     select,
     text,
@@ -50,7 +51,7 @@
     import typing
     from collections.abc import Mapping, Sequence
 
-    from sqlalchemy import Table
+    from sqlalchemy import Insert, Table
     from sqlalchemy.ext.asyncio import AsyncConnection, AsyncEngine
 
 
@@ -140,6 +141,21 @@ async def validate_schema(self, *, check_autovacuum: bool = False) -> None: ...
     async def ping(self) -> bool: ...
 
 
+def _build_dlq_cte(outbox: "Table", dlq: "Table") -> "Insert":
+    """Build the DELETE-then-archive CTE once; the row id, lease token and failure context are bind parameters."""
+    deleted = (
+        delete(outbox)
+        .where(outbox.c.id == bindparam("message_id"), outbox.c.acquired_token == bindparam("acquired_token"))
+        .returning(*(outbox.c[col] for col, _ in _DLQ_PROJECTION))
+        .cte("deleted")
+    )
+    archived = select(
+        *(deleted.c[col] for col, _ in _DLQ_PROJECTION),
+        *(bindparam(col, type_=dlq.c[col].type) for col in _DLQ_INJECTED_COLUMNS),
+    )
+    return insert(dlq).from_select([*(col for _, col in _DLQ_PROJECTION), *_DLQ_INJECTED_COLUMNS], archived)
+
+
 class OutboxClient(AbstractOutboxClient):
     def __init__(
         self,
@@ -155,6 +171,7 @@ def __init__(
         self._engine = engine
         self._table = outbox_table
         self._dlq_table = dlq_table
+        self._dlq_cte = None if dlq_table is None else _build_dlq_cte(outbox_table, dlq_table)
 
     @property
     def table(self) -> "Table":
@@ -238,7 +255,7 @@ async def fetch(
             .returning(*t.c)
         )
         async with conn.begin():
-            result = await conn.execute(stmt, {"lease_ttl": max(0.0, lease_ttl_seconds)})
+            result = await conn.execute(stmt, {"lease_ttl": lease_ttl_seconds})
             rows = result.mappings().all()
         return [_row_to_message(dict(row)) for row in rows]
 
@@ -287,7 +304,7 @@ async def delete_with_lease(
             msg = "OutboxClient.delete_with_lease requires a live AsyncConnection (got None)"
             raise TypeError(msg)
         if dlq_payload is not None:
-            if self._dlq_table is None:
+            if self._dlq_cte is None:
                 # P10: a dlq_payload with no dlq_table would silently degrade to a plain
                 # DELETE — losing the audit row. If the broker/client wiring ever desyncs,
                 # fail loudly rather than drop forensics.
@@ -296,72 +313,19 @@ async def delete_with_lease(
                     "refusing to drop the audit row silently"
                 )
                 raise RuntimeError(msg)
-            stmt, params = self._build_dlq_cte_stmt(message_id, acquired_token, dlq_payload)
-            result = await conn.execute(stmt, params)
+            params = {
+                "message_id": message_id,
+                "acquired_token": acquired_token,
+                "failure_reason": dlq_payload["failure_reason"],
+                "last_exception": dlq_payload["last_exception"],
+            }
+            result = await conn.execute(self._dlq_cte, params)
             return (result.rowcount or 0) > 0
         t = self._table
         del_stmt = delete(t).where(t.c.id == message_id, t.c.acquired_token == acquired_token)
         result = await conn.execute(del_stmt)
         return (result.rowcount or 0) > 0
 
-    def _build_dlq_cte_stmt(
-        self,
-        message_id: int,
-        acquired_token: uuid.UUID,
-        dlq_payload: "Mapping[str, typing.Any]",
-    ) -> "tuple[typing.Any, dict[str, typing.Any]]":
-        """Compose the single-statement DLQ CTE plus the parameter dict.
-
-        Identifiers are quoted via the dialect's identifier preparer so reserved words
-        and odd characters survive interpolation. The outbox/DLQ table names are
-        application-controlled (from the user's ``MetaData``), not request-derived
-        input, so the quoting is a robustness/correctness safeguard not a security
-        boundary.
-        """
-        # ``self._dlq_table`` is guaranteed non-None at the only call site (the guard
-        # in ``delete_with_lease``); local alias narrows the type for the formatting.
-        dlq_table = self._dlq_table
-        assert dlq_table is not None  # noqa: S101
-        preparer = self._engine.dialect.identifier_preparer
-        # ``format_table`` renders the schema-qualified, quoted name (``"app"."outbox"``)
-        # when the Table carries a non-default ``schema=``. ``quote(table.name)`` dropped
-        # the schema, so a ``MetaData(schema="app")`` deployment hit ``UndefinedTable`` on
-        # every terminal failure (poison rows retry forever, the outbox grows) or silently
-        # wrote to a same-named search_path table (B10).
-        outbox_name = preparer.format_table(self._table)
-        dlq_name = preparer.format_table(dlq_table)
-        # Column lists derive from _DLQ_PROJECTION (projected pairs first, then the
-        # injected failure-context columns) so the real CTE and the fake stay in lockstep
-        # off one source. INSERT and SELECT share the same order, so the named columns map
-        # positionally.
-        returning_cols = ", ".join(out for out, _ in _DLQ_PROJECTION)
-        insert_cols = ", ".join([dlq for _, dlq in _DLQ_PROJECTION] + list(_DLQ_INJECTED_COLUMNS))
-        select_exprs = ", ".join(
-            [out for out, _ in _DLQ_PROJECTION] + [f":{col}" for col in _DLQ_INJECTED_COLUMNS],
-        )
-        # S608: outbox_name / dlq_name come from application-defined SQLAlchemy Table
-        # objects (not request input) and are quoted via the dialect's identifier preparer;
-        # the column names come from _DLQ_PROJECTION constants. Values flow through
-        # :bindparam placeholders.
-        cte_sql = (
-            f"WITH deleted AS ("  # noqa: S608
-            f"DELETE FROM {outbox_name} "
-            f"WHERE id = :message_id AND acquired_token = :acquired_token "
-            f"RETURNING {returning_cols}"
-            f") "
-            f"INSERT INTO {dlq_name} ({insert_cols}) "
-            f"SELECT {select_exprs} "
-            f"FROM deleted"
-        )
-        sql = text(cte_sql)
-        params = {
-            "message_id": message_id,
-            "acquired_token": acquired_token,
-            "failure_reason": dlq_payload["failure_reason"],
-            "last_exception": dlq_payload["last_exception"],
-        }
-        return sql, params
-
     async def delete_batch_with_lease(
         self,
         conn: "AsyncConnection | None",

faststream_outbox/publisher/producer.py

@@ -30,6 +30,25 @@
     from fast_depends.library.serializer import SerializerProto
     from faststream._internal.parser import CodecProto
     from faststream._internal.types import AsyncCallable, CustomCallable
+    from sqlalchemy import Insert
+
+
+def _build_insert_stmts(table: Table) -> "dict[tuple[bool, bool], Insert]":
+    """Build the single-row INSERT once per ``(timer_id set, activate_in set)``; values arrive as parameters."""
+    # Server-side compute keeps timing immune to worker/DB clock skew (mirrors
+    # client.mark_pending_with_lease).
+    activate_in = func.now() + func.make_interval(0, 0, 0, 0, 0, 0, bindparam("activate_in_seconds", type_=Float))
+    plain = insert(table)
+    timer = pg_insert(table).on_conflict_do_nothing(
+        index_elements=[table.c.queue, table.c.timer_id],
+        index_where=table.c.timer_id.is_not(None),
+    )
+    return {
+        (False, False): plain.returning(table.c.id),
+        (False, True): plain.values(next_attempt_at=activate_in).returning(table.c.id),
+        (True, False): timer.returning(table.c.id),
+        (True, True): timer.values(next_attempt_at=activate_in).returning(table.c.id),
+    }
 
 
 class OutboxProducer:
@@ -61,24 +80,14 @@ def __init__(
         self._parser = ParserComposition(parser, default.parse_message)
         self._decoder = ParserComposition(decoder, default.decode_message)
         self._metrics_recorder = metrics_recorder
+        self._insert_stmts = _build_insert_stmts(table)
 
     def _emit_metric(self, event: str, tags: "Mapping[str, typing.Any]") -> None:
         # Delegate to the shared swallow-and-DEBUG-log helper so the recorder-isolation
         # contract lives in exactly one place (the subscriber's _emit_metric is the one
         # deliberate exception — it routes through self._log for handler-scoped context).
         _safe_emit(self._metrics_recorder, event, tags)
 
-    def connect(
-        self,
-        connection: typing.Any = None,  # noqa: ARG002
-        serializer: typing.Optional["SerializerProto"] = None,
-    ) -> None:
-        self.serializer = serializer
-
-    def disconnect(self) -> None:
-        # Caller owns the engine — never dispose anything here.
-        pass
-
     async def publish(self, cmd: OutboxPublishCommand) -> int | None:
         start_perf = time.perf_counter()
         size_bytes = 0
@@ -125,37 +134,20 @@ async def _do_publish(
         payload: bytes,
         hdrs: dict[str, str] | None,
     ) -> int | None:
-        t = self._table
-        values: dict[str, typing.Any] = {"queue": cmd.queue, "payload": payload, "headers": hdrs}
-        # Server-side compute keeps timing immune to worker/DB clock skew (mirrors
-        # client.mark_pending_with_lease).
+        params: dict[str, typing.Any] = {"queue": cmd.queue, "payload": payload, "headers": hdrs}
         if cmd.activate_in is not None:
-            values["next_attempt_at"] = func.now() + func.make_interval(
-                0, 0, 0, 0, 0, 0, bindparam("activate_in_seconds", cmd.activate_in.total_seconds(), type_=Float)
-            )
+            params["activate_in_seconds"] = cmd.activate_in.total_seconds()
         elif cmd.activate_at is not None:
-            values["next_attempt_at"] = cmd.activate_at
+            params["next_attempt_at"] = cmd.activate_at
         if cmd.timer_id is not None:
-            values["timer_id"] = cmd.timer_id
+            params["timer_id"] = cmd.timer_id
         # Skip NOTIFY only when the row is genuinely future-dated. A past activate_at
         # (e.g. a recovered idempotency token) is immediately eligible — fire NOTIFY.
         now = utcnow()
         is_future = is_future_dated(cmd.activate_in, cmd.activate_at, now)
 
-        if cmd.timer_id is not None:
-            stmt = (
-                pg_insert(t)
-                .values(**values)
-                .on_conflict_do_nothing(
-                    index_elements=[t.c.queue, t.c.timer_id],
-                    index_where=t.c.timer_id.is_not(None),
-                )
-                .returning(t.c.id)
-            )
-        else:
-            stmt = insert(t).values(**values).returning(t.c.id)
-
-        result = await cmd.session.execute(stmt)
+        stmt = self._insert_stmts[cmd.timer_id is not None, cmd.activate_in is not None]
+        result = await cmd.session.execute(stmt, params)
         row_id: int | None = result.scalar()
         # Skip NOTIFY for future-dated rows (listeners can't act before the gate
         # opens — polling fires them at next tick) and on conflict (no row landed).

faststream_outbox/schema.py

@@ -158,7 +158,7 @@ def make_outbox_table(metadata: "MetaData", table_name: str = "outbox") -> Table
 
 # Outbox-row columns copied verbatim into the DLQ archive row on terminal failure, as
 # ``(outbox_column, dlq_column)`` pairs. The single source both the real client's DLQ
-# CTE (``OutboxClient._build_dlq_cte_stmt``) and the fake (``FakeOutboxClient.delete_with_lease``)
+# CTE (``client._build_dlq_cte``) and the fake (``FakeOutboxClient.delete_with_lease``)
 # build from, so a DLQ column change is one edit here instead of hand-kept parity in two
 # substrates. ``failed_at`` is not listed — it rides the DLQ column's ``server_default``.
 _DLQ_PROJECTION: tuple[tuple[str, str], ...] = (

faststream_outbox/schema_validation.py

@@ -282,7 +282,7 @@ def _run_validate(
     # domain tables (in their own MetaData) don't show up in the diff. Carry the
     # user's schema onto the canonical copy so the autogenerate diff compares
     # ``app.outbox`` against ``app.outbox`` rather than the default search_path —
-    # matching the schema-qualified DLQ CTE in ``_build_dlq_cte_stmt`` (B10).
+    # matching the schema-qualified DLQ CTE in ``client._build_dlq_cte`` (B10).
     canonical_metadata = MetaData(schema=table.schema)
     canonical_factory(canonical_metadata, table.name)
 

faststream_outbox/testing.py

@@ -156,7 +156,7 @@ async def fetch(
         if not queues:
             return []
         now = utcnow()
-        lease_cutoff = now - dt.timedelta(seconds=max(0.0, lease_ttl_seconds))
+        lease_cutoff = now - dt.timedelta(seconds=lease_ttl_seconds)
         token = uuid.uuid4()
         out: list[OutboxInnerMessage] = []
         eligible = sorted(
@@ -485,13 +485,6 @@ async def publish_batch(self, cmd: OutboxPublishCommand) -> None:
     async def request(self, cmd: OutboxPublishCommand) -> typing.NoReturn:
         raise NotImplementedError(_REQUEST_UNSUPPORTED_MSG)
 
-    def connect(self, connection: typing.Any = None, serializer: typing.Any = None) -> None:  # noqa: ARG002
-        if serializer is not None:
-            self._serializer = serializer
-
-    def disconnect(self) -> None:
-        pass
-
 
 def _build_fake_publish(
     fake_client: FakeOutboxClient,