diff --git a/docs/concepts/performance.md b/docs/concepts/performance.md index fb17faf..aad1856 100644 --- a/docs/concepts/performance.md +++ b/docs/concepts/performance.md @@ -45,7 +45,8 @@ why). | Need more parallel handler capacity | `max_workers` / `fetch_batch_size` | More concurrency, but mind the [connection budget](../usage/subscriber.md#connection-budget). | Single-publish-per-transaction workloads see no change from NOTIFY dedup (one row -still emits one NOTIFY), and the batched-flush win is largest at low `max_workers` +still emits one NOTIFY), but that NOTIFY rides in the INSERT's `RETURNING`, so it +adds no round trip of its own. The batched-flush win is largest at low `max_workers` (where per-row deletes serialise) and narrows as worker parallelism rises. ## The three levers @@ -53,8 +54,9 @@ still emits one NOTIFY), and the batched-flush win is largest at low `max_worker ### Producer: NOTIFY dedup (automatic) `broker.publish` / `publish_batch` emit one `pg_notify` per (transaction, -queue), so N publishes to the same queue in one transaction cost one NOTIFY -round-trip. Postgres already coalesces identical notifications per transaction +queue), so N publishes to the same queue in one transaction cost at most one +NOTIFY. A single `publish` puts it in the INSERT's `RETURNING`, so it costs no +round trip; `publish_batch` sends it as one extra statement. Postgres already coalesces identical notifications per transaction at delivery, so extra NOTIFYs would be pure waste. It is default-on, has no knob, and is behavior-preserving: the subscriber still gets its wake, fired inline at the first publish. See diff --git a/docs/introduction/how-it-works.md b/docs/introduction/how-it-works.md index 47de409..bcaceab 100644 --- a/docs/introduction/how-it-works.md +++ b/docs/introduction/how-it-works.md @@ -47,11 +47,13 @@ async with session_factory() as session, session.begin(): `publish_batch(*bodies, queue, session, ...)` does the same with a single round-trip for many rows. -The producer also emits `SELECT pg_notify('outbox_', queue)` on the -caller's session right after the INSERT, except when the row is -genuinely future-dated (a future `activate_in` / `activate_at`; a *past* -`activate_at`, e.g. a recovered idempotency token, still notifies) or a -`timer_id` conflict made the insert a no-op. NOTIFY is transactional, so listeners only see it +A single `publish` also sends `pg_notify('outbox_
', queue)` in the +INSERT's `RETURNING` clause, so the wakeup costs no extra round trip. +`publish_batch` sends a separate `SELECT pg_notify(...)` after its INSERT. +Neither notifies when the row is genuinely future-dated (a future +`activate_in` / `activate_at`; a *past* `activate_at`, e.g. a recovered +idempotency token, still notifies), and a `timer_id` conflict returns no +row, so it sends no NOTIFY either. NOTIFY is transactional, so listeners only see it after the user's transaction commits, so atomicity with the row insert is automatic. diff --git a/faststream_outbox/publisher/producer.py b/faststream_outbox/publisher/producer.py index 63995e0..212bb0d 100644 --- a/faststream_outbox/publisher/producer.py +++ b/faststream_outbox/publisher/producer.py @@ -13,7 +13,7 @@ from faststream._internal.endpoint.utils import ParserComposition from faststream._internal.parser import DefaultCodec -from sqlalchemy import Float, Table, bindparam, func, insert, text +from sqlalchemy import Float, Table, bindparam, func, insert, literal, text from sqlalchemy.dialects.postgresql import insert as pg_insert from faststream_outbox._scheduling import is_future_dated, resolve_next_attempt_client_side @@ -33,8 +33,12 @@ 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.""" +def _build_insert_stmts(table: Table, channel: str) -> "dict[tuple[bool, bool, bool], Insert]": + """Build the single-row INSERT once per ``(timer_id set, activate_in set, notify)``; values arrive as parameters. + + The notify variants call ``pg_notify`` in ``RETURNING``, so the NOTIFY costs no extra round trip and + fires only when a row actually landed (an ``ON CONFLICT DO NOTHING`` skip returns no row). + """ # 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)) @@ -43,12 +47,13 @@ def _build_insert_stmts(table: Table) -> "dict[tuple[bool, bool], Insert]": 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), - } + notify = func.pg_notify(literal(channel), table.c.queue) + stmts: dict[tuple[bool, bool, bool], Insert] = {} + for has_timer, base in ((False, plain), (True, timer)): + for has_activate_in, stmt in ((False, base), (True, base.values(next_attempt_at=activate_in))): + stmts[has_timer, has_activate_in, False] = stmt.returning(table.c.id) + stmts[has_timer, has_activate_in, True] = stmt.returning(table.c.id, notify) + return stmts class OutboxProducer: @@ -80,7 +85,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) + self._insert_stmts = _build_insert_stmts(table, self._channel) 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 @@ -143,16 +148,20 @@ async def _do_publish( 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) - - stmt = self._insert_stmts[cmd.timer_id is not None, cmd.activate_in is not None] + # Future-dated rows skip NOTIFY: listeners can't act before the gate opens, and + # polling fires them at the next tick. + emitted = self._notified_queues(cmd.session) + notify = not is_future_dated(cmd.activate_in, cmd.activate_at, utcnow()) and ( + emitted is None or cmd.queue not in emitted + ) + stmt = self._insert_stmts[cmd.timer_id is not None, cmd.activate_in is not None, notify] 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). - if row_id is not None and not is_future: - await self._notify(cmd.session, cmd.queue) + if row_id is not None and notify: + # Re-read: under autobegin the transaction only exists once the INSERT has run. + emitted = self._notified_queues(cmd.session) + if emitted is not None: + emitted.add(cmd.queue) return row_id async def publish_batch(self, cmd: OutboxPublishCommand) -> None: @@ -206,6 +215,14 @@ async def publish_batch(self, cmd: OutboxPublishCommand) -> None: async def request(self, cmd: OutboxPublishCommand) -> typing.NoReturn: raise NotImplementedError(_REQUEST_UNSUPPORTED_MSG) + def _notified_queues(self, session: typing.Any) -> set[str] | None: + """Return the queues already notified in *session*'s innermost transaction, or None outside one.""" + sync_session = session.sync_session + txn = sync_session.get_nested_transaction() or sync_session.get_transaction() + if txn is None: + return None + return self._notified.setdefault(txn, set()) + async def _notify(self, session: typing.Any, queue: str) -> None: # ``pg_notify(:channel, :payload)`` -- parameterized so channel and payload # bind cleanly (raw NOTIFY accepts only literals -- injection-prone). Runs on @@ -224,10 +241,8 @@ async def _notify(self, session: typing.Any, queue: str) -> None: # defeat the memo. The Session holds the sync transaction strongly for its lifetime, # so the key survives across publishes whether the caller uses `session.begin()` or # autobegin. No active transaction -> emit unconditionally. - sync_session = session.sync_session - txn = sync_session.get_nested_transaction() or sync_session.get_transaction() - if txn is not None: - emitted = self._notified.setdefault(txn, set()) + emitted = self._notified_queues(session) + if emitted is not None: if queue in emitted: return emitted.add(queue) diff --git a/pyproject.toml b/pyproject.toml index 5290087..9a22af1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -149,6 +149,9 @@ Changelog = "https://github.com/modern-python/faststream-outbox/releases" [tool.coverage.run] omit = ["benchmarks/*"] +# The C tracer intermittently drops executed lines in async tests; 3.11 falls back to it. +core = "sysmon" +disable_warnings = ["no-sysmon"] [tool.coverage.report] fail_under = 100 diff --git a/tests/test_integration.py b/tests/test_integration.py index 2287ffd..6e425ab 100644 --- a/tests/test_integration.py +++ b/tests/test_integration.py @@ -9,6 +9,7 @@ from typing import Any from unittest import mock +import asyncpg import pytest from faststream import BaseMiddleware, Context, Depends from faststream.kafka import KafkaBroker, TestKafkaBroker @@ -2529,6 +2530,16 @@ async def test_notify_dedup_spans_publish_batch_and_publish(pg_engine, outbox_ta assert count[0] == 1 +async def test_notify_dedup_spans_publish_and_publish_batch(pg_engine: AsyncEngine, outbox_table: Table) -> None: + broker = OutboxBroker(pg_engine, outbox_table=outbox_table) + session_factory = async_sessionmaker(pg_engine, expire_on_commit=False) + with _count_pg_notify(pg_engine) as count: + async with session_factory() as session, session.begin(): + await broker.publish({"n": 1}, queue="orders", session=session) # pg_notify in RETURNING + await broker.publish_batch({"n": 2}, {"n": 3}, queue="orders", session=session) # deduped + assert count[0] == 1 + + async def test_notify_still_skipped_for_future_dated(pg_engine, outbox_table) -> None: broker = OutboxBroker(pg_engine, outbox_table=outbox_table) session_factory = async_sessionmaker(pg_engine, expire_on_commit=False) @@ -2538,6 +2549,57 @@ async def test_notify_still_skipped_for_future_dated(pg_engine, outbox_table) -> assert count[0] == 0 # future-dated -> no NOTIFY (unchanged), dedup did not break the skip +async def test_publish_notifies_in_the_insert_round_trip(pg_engine: AsyncEngine, outbox_table: Table) -> None: + broker = OutboxBroker(pg_engine, outbox_table=outbox_table) + session_factory = async_sessionmaker(pg_engine, expire_on_commit=False) + statements: list[str] = [] + + def _listener(conn, cursor, statement, parameters, context, executemany) -> None: # noqa: ARG001 + statements.append(statement) + + async with session_factory() as session, session.begin(): + await session.connection() + event.listen(pg_engine.sync_engine, "before_cursor_execute", _listener) + try: + await broker.publish({"n": 1}, queue="orders", session=session) + await broker.publish({"n": 2}, queue="orders", session=session) + finally: + event.remove(pg_engine.sync_engine, "before_cursor_execute", _listener) + + assert len(statements) == 2 + assert "pg_notify" in statements[0].lower() + assert "pg_notify" not in statements[1].lower() + + +async def test_notify_sent_after_a_timer_conflict_in_the_same_transaction( + pg_engine: AsyncEngine, outbox_table: Table +) -> None: + broker = OutboxBroker(pg_engine, outbox_table=outbox_table) + session_factory = async_sessionmaker(pg_engine, expire_on_commit=False) + async with session_factory() as session, session.begin(): + await broker.publish({"n": 0}, queue="orders", session=session, timer_id="t-1") + + payloads: list[str] = [] + received = asyncio.Event() + dsn = pg_engine.url.set(drivername="postgresql").render_as_string(hide_password=False) + listener = await asyncpg.connect(dsn) + try: + + def _cb(_conn, _pid, _channel, payload) -> None: + payloads.append(payload) + received.set() + + await listener.add_listener(f"outbox_{outbox_table.name}", _cb) + async with session_factory() as session, session.begin(): + assert await broker.publish({"n": 1}, queue="orders", session=session, timer_id="t-1") is None + assert await broker.publish({"n": 2}, queue="orders", session=session) is not None + await asyncio.wait_for(received.wait(), timeout=2.0) + finally: + await listener.close() + + assert payloads == ["orders"] + + async def test_notify_deduped_in_autobegin_transaction(pg_engine, outbox_table) -> None: # Autobegin (no explicit session.begin()): the dedup must still collapse N publishes # of one queue to one pg_notify. Guards against keying the memo on the GC-fragile async diff --git a/tests/test_unit.py b/tests/test_unit.py index 6e372a3..d45b5e2 100644 --- a/tests/test_unit.py +++ b/tests/test_unit.py @@ -612,21 +612,19 @@ async def test_broker_publish_rejects_non_async_session() -> None: await broker.publish(b"x", queue="orders", session=object()) # ty: ignore[invalid-argument-type] -async def test_broker_publish_executes_insert_then_pg_notify_on_session() -> None: +async def test_broker_publish_executes_one_insert_that_notifies() -> None: broker = _make_broker() session = _make_session_mock() 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 + assert session.execute.await_count == 1 insert_stmt, params = session.execute.await_args_list[0].args - assert "INSERT INTO" in str(insert_stmt) + compiled = insert_stmt.compile(dialect=postgresql.dialect()) + assert "INSERT INTO" in str(compiled) + assert re.search(r"RETURNING outbox\.id, pg_notify\(%\(param_1\)s(::VARCHAR)?, outbox\.queue\)", str(compiled)) + assert compiled.params["param_1"] == "outbox_outbox" assert params["queue"] == "orders" assert json.loads(params["payload"]) == {"order_id": 1} assert params["headers"]["content-type"] == "application/json" - notify_stmt, notify_params = session.execute.await_args_list[1].args - assert "pg_notify" in str(notify_stmt) - assert notify_params["channel"] == "outbox_outbox" - assert notify_params["payload"] == "orders" async def test_broker_publish_encodes_pydantic_model() -> None: @@ -989,6 +987,7 @@ async def test_broker_publish_with_activate_in_skips_notify() -> None: insert_stmt = session.execute.await_args_list[0].args[0] assert "INSERT INTO" in str(insert_stmt) assert "next_attempt_at" in str(insert_stmt) + assert "pg_notify" not in str(insert_stmt) async def test_broker_publish_with_activate_at_skips_notify() -> None: @@ -997,8 +996,9 @@ 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 + insert_stmt, params = session.execute.await_args_list[0].args assert params["next_attempt_at"] == fire + assert "pg_notify" not in str(insert_stmt) async def test_broker_publish_emits_notify_when_activate_at_is_past() -> None: @@ -1008,19 +1008,18 @@ async def test_broker_publish_emits_notify_when_activate_at_is_past() -> None: session = _make_session_mock() past = dt.datetime.now(tz=dt.UTC) - dt.timedelta(seconds=5) await broker.publish(b"x", queue="orders", session=session, activate_at=past) - assert session.execute.await_count == 2 - notify_stmt, notify_params = session.execute.await_args_list[1].args - assert "pg_notify" in str(notify_stmt) - assert notify_params["payload"] == "orders" + assert session.execute.await_count == 1 + insert_stmt, _params = session.execute.await_args_list[0].args + assert "pg_notify" in str(insert_stmt) async def test_broker_publish_emits_notify_when_activate_in_is_zero() -> None: broker = _make_broker() session = _make_session_mock() await broker.publish(b"x", queue="orders", session=session, activate_in=dt.timedelta(0)) - assert session.execute.await_count == 2 - notify_stmt, _params = session.execute.await_args_list[1].args - assert "pg_notify" in str(notify_stmt) + assert session.execute.await_count == 1 + insert_stmt, _params = session.execute.await_args_list[0].args + assert "pg_notify" in str(insert_stmt) async def test_broker_publish_rejects_naive_activate_at() -> None: @@ -3669,7 +3668,7 @@ def _make_producer(recorder: typing.Any) -> OutboxProducer: async def test_producer_publish_emits_error_event_and_reraises_on_sql_failure() -> None: events, recorder = _events_recorder() producer = _make_producer(recorder) - session = AsyncMock(spec=AsyncSession) + session = _make_session_mock() session.execute.side_effect = RuntimeError("forced INSERT failure") cmd = OutboxPublishCommand({"x": 1}, queue="orders", session=session)