From 654c2e1d0581766a1bdc80bb332d68fae4968883 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Sat, 17 Oct 2026 13:37:14 +0300 Subject: [PATCH] refactor: DLQ CTE in Core, prebuilt publish INSERTs, drop dead code --- faststream_outbox/client.py | 90 ++++++++----------------- faststream_outbox/publisher/producer.py | 60 +++++++---------- faststream_outbox/schema.py | 2 +- faststream_outbox/schema_validation.py | 2 +- faststream_outbox/testing.py | 9 +-- tests/test_unit.py | 75 ++++++--------------- 6 files changed, 75 insertions(+), 163 deletions(-) diff --git a/faststream_outbox/client.py b/faststream_outbox/client.py index af47532..bf69e7b 100644 --- a/faststream_outbox/client.py +++ b/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", diff --git a/faststream_outbox/publisher/producer.py b/faststream_outbox/publisher/producer.py index 26ae42e..63995e0 100644 --- a/faststream_outbox/publisher/producer.py +++ b/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,6 +80,7 @@ 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 @@ -68,17 +88,6 @@ def _emit_metric(self, event: str, tags: "Mapping[str, typing.Any]") -> None: # 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). diff --git a/faststream_outbox/schema.py b/faststream_outbox/schema.py index fc0992c..5796852 100644 --- a/faststream_outbox/schema.py +++ b/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], ...] = ( diff --git a/faststream_outbox/schema_validation.py b/faststream_outbox/schema_validation.py index 434648c..93a2588 100644 --- a/faststream_outbox/schema_validation.py +++ b/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) diff --git a/faststream_outbox/testing.py b/faststream_outbox/testing.py index 9104125..8870798 100644 --- a/faststream_outbox/testing.py +++ b/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, diff --git a/tests/test_unit.py b/tests/test_unit.py index 8b99df2..f593441 100644 --- a/tests/test_unit.py +++ b/tests/test_unit.py @@ -617,9 +617,8 @@ async def test_broker_publish_executes_insert_then_pg_notify_on_session() -> Non await broker.publish({"order_id": 1}, queue="orders", session=session) # Two execute calls: the INSERT, then SELECT pg_notify(...). assert session.execute.await_count == 2 - insert_stmt = session.execute.await_args_list[0].args[0] + insert_stmt, params = session.execute.await_args_list[0].args assert "INSERT INTO" in str(insert_stmt) - params = insert_stmt.compile().params assert params["queue"] == "orders" assert json.loads(params["payload"]) == {"order_id": 1} assert params["headers"]["content-type"] == "application/json" @@ -634,8 +633,7 @@ async def test_broker_publish_encodes_pydantic_model() -> None: session = _make_session_mock() body = _PydanticBody(order_id=7, name="alpha") await broker.publish(body, queue="orders", session=session) - insert_stmt = session.execute.await_args_list[0].args[0] - params = insert_stmt.compile().params + _, params = session.execute.await_args_list[0].args assert json.loads(params["payload"]) == body.model_dump() assert params["headers"]["content-type"] == "application/json" @@ -998,7 +996,7 @@ async def test_broker_publish_with_activate_at_skips_notify() -> None: fire = dt.datetime.now(tz=dt.UTC) + dt.timedelta(minutes=5) await broker.publish(b"x", queue="orders", session=session, activate_at=fire) assert session.execute.await_count == 1 - params = session.execute.await_args_list[0].args[0].compile().params + _, params = session.execute.await_args_list[0].args assert params["next_attempt_at"] == fire @@ -1036,13 +1034,12 @@ async def test_broker_publish_with_timer_id_uses_on_conflict() -> None: broker = _make_broker() session = _make_session_mock() await broker.publish(b"x", queue="orders", session=session, timer_id="email-123") - insert_stmt = session.execute.await_args_list[0].args[0] - compiled = insert_stmt.compile(dialect=postgresql.dialect()) - sql = str(compiled) + insert_stmt, params = session.execute.await_args_list[0].args + sql = str(insert_stmt.compile(dialect=postgresql.dialect(), column_keys=list(params))) assert "INSERT INTO" in sql assert "ON CONFLICT" in sql assert "DO NOTHING" in sql - assert compiled.params["timer_id"] == "email-123" + assert params["timer_id"] == "email-123" async def test_broker_publish_returns_none_on_timer_id_conflict() -> None: @@ -1255,14 +1252,6 @@ async def test_outbox_producer_request_raises_not_implemented() -> None: await producer.request(cmd) -def test_outbox_producer_connect_disconnect_noop() -> None: - metadata = MetaData() - t = make_outbox_table(metadata) - producer = OutboxProducer(table=t, parser=None, decoder=None) - producer.connect() # must not raise - producer.disconnect() # must not raise - - def test_broker_exposes_outbox_producer() -> None: broker = _make_broker() producer = broker.config.broker_config.producer @@ -1393,20 +1382,6 @@ async def test_fake_outbox_producer_request_raises() -> None: await producer.request(cmd) -def test_fake_outbox_producer_connect_sets_serializer() -> None: - broker = _make_broker() - producer = FakeOutboxProducer(FakeOutboxClient(), broker, serializer=None, run_loops=False) - sentinel = object() - producer.connect(serializer=sentinel) - assert producer._serializer is sentinel # noqa: SLF001 - - -def test_fake_outbox_producer_disconnect_is_noop() -> None: - broker = _make_broker() - producer = FakeOutboxProducer(FakeOutboxClient(), broker, serializer=None, run_loops=False) - producer.disconnect() # must not raise - - def test_broker_client_property_raises_without_engine() -> None: metadata = MetaData() t = make_outbox_table(metadata) @@ -3691,6 +3666,11 @@ def raising_recorder(event: str, tags: typing.Any) -> None: # noqa: ARG001 # --- make_dlq_table ----------------------------------------------------------- +def _dlq_cte_sql(client: OutboxClient, engine: typing.Any) -> str: + assert client._dlq_cte is not None # noqa: SLF001 + return str(client._dlq_cte.compile(engine)) # noqa: SLF001 + + async def test_dlq_cte_uses_schema_qualified_table_names() -> None: """B10: the DLQ CTE must render schema-qualified names when the tables carry a non-default schema.""" engine = create_async_engine("postgresql+asyncpg://u:p@localhost/db") # constructed, never connected @@ -3699,14 +3679,8 @@ async def test_dlq_cte_uses_schema_qualified_table_names() -> None: outbox = make_outbox_table(metadata) dlq = make_dlq_table(metadata) client = OutboxClient(engine, outbox, dlq_table=dlq) - stmt, _params = client._build_dlq_cte_stmt( # noqa: SLF001 - 1, - uuid.uuid4(), - {"failure_reason": "rejected", "last_exception": None}, - ) - sql = str(stmt) - # The buggy quote(table.name) rendered bare "outbox" / "outbox_dlq"; format_table - # carries the schema → "app.outbox" / "app.outbox_dlq" (B10). + sql = _dlq_cte_sql(client, engine) + # A bare table name would resolve through search_path instead of the "app" schema (B10). assert "DELETE FROM app.outbox " in sql assert "INSERT INTO app.outbox_dlq " in sql finally: @@ -3714,22 +3688,16 @@ async def test_dlq_cte_uses_schema_qualified_table_names() -> None: async def test_dlq_cte_quotes_adversarial_table_names() -> None: - """F3-05: the DLQ CTE is the one raw-SQL identifier site — assert adversarial names stay quoted/escaped.""" + """F3-05: adversarial table names in the DLQ CTE stay quoted/escaped.""" engine = create_async_engine("postgresql+asyncpg://u:p@localhost/db") # constructed, never connected try: metadata = MetaData() outbox = make_outbox_table(metadata, table_name='ob"x') dlq = make_dlq_table(metadata, table_name="dlq;drop") client = OutboxClient(engine, outbox, dlq_table=dlq) - stmt, _params = client._build_dlq_cte_stmt( # noqa: SLF001 - 1, - uuid.uuid4(), - {"failure_reason": "rejected", "last_exception": None}, - ) - sql = str(stmt) + sql = _dlq_cte_sql(client, engine) # The embedded double-quote is doubled (escaped) and the ';' name stays wrapped in - # quotes — the identifier cannot break out of its quoting into injectable SQL. A - # refactor dropping ``format_table`` for bare ``table.name`` would fail here. + # quotes — the identifier cannot break out of its quoting into injectable SQL. assert '"ob""x"' in sql assert '"dlq;drop"' in sql finally: @@ -4168,8 +4136,8 @@ def test_default_retry_strategy_pins_documented_parameters() -> None: async def test_dlq_cte_insert_columns_match_make_dlq_table() -> None: """INVARIANT: the DLQ CTE's INSERT column list matches ``make_dlq_table`` minus ``id`` and ``failed_at``. - ``_build_dlq_cte_stmt`` renders the column list as an f-string with nothing linking it to the - table definition. A NOT-NULL-without-default column added to ``make_dlq_table`` would make every + ``_build_dlq_cte`` takes the column list from ``_DLQ_PROJECTION``, not from the table + definition. A NOT-NULL-without-default column added to ``make_dlq_table`` would make every terminal DLQ write fail -- poison rows that retry forever -- and a nullable one would silently drop audit data. The two omitted columns are deliberate: ``id`` autoincrements and ``failed_at`` rides its server default. @@ -4180,12 +4148,7 @@ async def test_dlq_cte_insert_columns_match_make_dlq_table() -> None: engine = create_async_engine("postgresql+asyncpg://u:p@localhost/db") # never connected try: client = OutboxClient(engine, outbox, dlq_table=dlq) - stmt, _ = client._build_dlq_cte_stmt( # noqa: SLF001 - 1, - uuid.uuid4(), - {"failure_reason": "x", "last_exception": "y"}, - ) - sql = str(stmt) + sql = _dlq_cte_sql(client, engine) finally: await engine.dispose()