From 25f6388764439e101067ec20b49c13a554ff9634 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Thu, 16 Jul 2026 10:49:33 +0300 Subject: [PATCH 1/4] docs(planning): design for NOTIFY-per-transaction dedup Dedup the producer's pg_notify to one statement per (transaction, queue) via a WeakKeyDictionary memo keyed on the innermost active transaction (savepoint- safe via GC, no listener). Behavior-preserving (matches Postgres' per-txn delivery coalescing), default-on, no knob. Benchmark producer select_calls 5000 -> 1, exact-gated. --- .../changes/2026-07-16.02-notify-dedup.md | 150 ++++++++++++++++++ 1 file changed, 150 insertions(+) create mode 100644 planning/changes/2026-07-16.02-notify-dedup.md diff --git a/planning/changes/2026-07-16.02-notify-dedup.md b/planning/changes/2026-07-16.02-notify-dedup.md new file mode 100644 index 0000000..5b46cfa --- /dev/null +++ b/planning/changes/2026-07-16.02-notify-dedup.md @@ -0,0 +1,150 @@ +--- +summary: Dedup the producer's `pg_notify` to one statement per (transaction, queue) instead of one per publish, via a transaction-scoped memo keyed on the innermost active transaction. Behavior-preserving (matches Postgres' own per-transaction delivery coalescing), default-on, no knob. Benchmark producer `select_calls` drops from N to 1 (5000 -> 1 for the bulk-single-queue case), exact-gated. +--- + +# Design: NOTIFY-per-transaction dedup + +## Summary + +`broker.publish` / `publish_batch` run `INSERT … RETURNING id` **plus** +`SELECT pg_notify(channel, queue)` per call, so N publishes in one caller +transaction emit N `pg_notify` statements — the benchmark producer measures +`insert_calls=5000, select_calls=5000`. Postgres already coalesces identical +`(channel, payload)` NOTIFYs *within a transaction* at delivery, so those N +statements produce at most one delivered wake anyway; the cost is N redundant +round-trips. + +This change deduplicates the producer's NOTIFY to **one `pg_notify` per +`(transaction, queue)`** via a transaction-scoped memo. It is **behavior-preserving** +(the subscriber sees the same "≥1 wake per queue per commit" it does today), +**default-on**, and has **no config knob**. The benchmark producer `select_calls` +drops from N to 1 (5000 → 1 for bulk single-queue publish), a clean exact-gated +before/after. + +## Motivation + +The performance review flagged the producer's two-statements-per-publish cost +(`INSERT … RETURNING` + `SELECT pg_notify`). The NOTIFY half is pure overhead when +multiple rows for the same queue commit together: Postgres delivers one wake per +`(channel, payload)` per transaction regardless, so the second-through-Nth +`pg_notify` statements buy nothing but round-trips. For a bulk publish (many rows to +one queue in one transaction) that is N−1 wasted statements. + +Unlike the batched terminal flush (which changed at-least-once redelivery semantics +and was therefore opt-in), NOTIFY dedup changes **nothing observable** — see below — +so it ships default-on with no knob. + +## Design + +### The dedup memo + +The producer gains one field: `self._notified: WeakKeyDictionary[transaction, set[str]]`. +`_notify` becomes dedup-aware; `publish` / `publish_batch` call it unchanged: + +```python +async def _notify(self, session, queue): + # Innermost active transaction: get_nested_transaction() during a savepoint, + # else the outer transaction. Each is a distinct, weak-referenceable object. + txn = session.get_nested_transaction() or session.get_transaction() + if txn is not None: + emitted = self._notified.setdefault(txn, set()) + if queue in emitted: + return # already NOTIFYed this queue in this txn + emitted.add(queue) + await session.execute(text("SELECT pg_notify(:channel, :payload)"), + {"channel": self._channel, "payload": queue}) +``` + +The `WeakKeyDictionary` needs **no explicit reset**: each transaction is a distinct +object with its own set; commit and rollback both drop the object, so GC clears the +entry. If there is no active transaction (`txn is None` — not expected under the +outbox contract, but defensive), it emits unconditionally. + +### Why key on the *innermost* transaction + +`get_transaction()` returns the *outer* transaction even inside a savepoint (verified +against SQLAlchemy 2.0.50). Keying on it would leave a rolled-back savepoint's memo +entry in place, so a later republish of that queue in the outer transaction would be +wrongly skipped → a **missed** NOTIFY. Keying on the innermost transaction +(`get_nested_transaction() or get_transaction()`) fixes this without any event +listener: a savepoint's publishes memoize against the nested transaction object, +which GCs when the savepoint ends (commit or rollback), so the outer republish always +re-emits. + +### Correctness — never misses, and transparent + +- **Flat txn, same queue** (common case + benchmark): 1 NOTIFY per `(txn, queue)`. +- **Multiple queues, one txn**: one NOTIFY per distinct queue. +- **Separate transactions, same session, same queue**: txn₂ is a new object → fresh + set → re-emits. Never suppressed by txn₁ (the load-bearing reset case). +- **Savepoint commit**: outer republish re-emits (memo was on the nested txn) → + redundant NOTIFY that Postgres coalesces → harmless. +- **Savepoint rollback**: the nested txn + its memo GC → outer republish re-emits → + no missed NOTIFY. + +**Behavior-preserving:** Postgres already coalesces identical `(channel, payload)` +NOTIFYs per transaction at delivery and never *guarantees* dedup, so the subscriber +already sees "≥1 wake per queue per commit" and already tolerates coalesced/duplicate +wakes (a NOTIFY sets an `asyncio.Event`; setting it twice is a no-op). We emit exactly +one — within the range Postgres already produces — inline at the first publish, so +delivery timing is unchanged. Hence default-on, no knob. + +### Scope + +The change is entirely in the real `OutboxProducer._notify` (`publisher/producer.py`). +The fake/test-broker producer (`publisher/fake.py`) emits no NOTIFY, so that path is +untouched. Non-Postgres behavior is unchanged (`_notify`'s statement is unchanged). +The existing guards — no NOTIFY for future-dated rows or `timer_id` conflicts (no row +landed) — are upstream of `_notify` and unaffected; dedup only sees calls that already +passed them. + +## Non-goals + +- **No knob / no opt-in.** The change is observably transparent, so there is nothing + to gate. +- **No cross-queue single-statement emit.** A deferred `before_commit` hook could + collapse a multi-queue transaction into one `pg_notify` statement; rejected as extra + machinery for a marginal gain (the common single-queue case is already 1). +- **No change to delivery, ordering, or the transactional producer contract.** + `_notify` still runs on the caller's session and commits with the row; no flush, no + commit. + +## Testing + +Deterministic — counting statements, not timing. A `before_cursor_execute` engine +listener counts `pg_notify` statements (no `pg_stat_statements` dependency). +Integration tests (real `AsyncSession` + Postgres, so the real transaction lifecycle +drives the dedup): + +- Flat, one queue: N publishes in one `begin()` → exactly **1** `pg_notify`. +- Multi-queue: N publishes across M queues in one txn → exactly **M**. +- Separate transactions (same session), same queue → **2** (per-transaction reset). +- Savepoint rollback then republish same queue → the committed row still gets a + `pg_notify` (innermost-key correctness). +- `publish_batch` + `publish` to the same queue in one txn dedup together. +- Future-dated / `timer_id`-conflict skips still emit no NOTIFY (unchanged). + +Delivery is unchanged, so existing subscriber/wakeup tests stand. + +## Benchmark + +The producer scenario (5000 publishes, one transaction, one queue) currently reports +`select_calls=5000` (the `pg_notify` count). After dedup, **`select_calls` drops +5000 → 1**, exact, deterministic, machine-independent — and `select_calls` is already +an exact-gated key. `insert_calls` stays 5000. `benchmarks/baseline.json` is +regenerated in the same PR so `just bench-check` gates the ~5000× NOTIFY round-trip +reduction for the bulk-single-queue case. + +## Risk + +- **Memo scoping is the whole correctness story.** If the memo leaked across + transactions, a second transaction's queue could be wrongly skipped (missed + wakeup). The `WeakKeyDictionary`-on-innermost-transaction design makes leakage + impossible (distinct object per transaction; GC on end). The integration tests pin + the separate-transaction and savepoint-rollback cases. +- **Missed NOTIFY is latency, not loss.** Even in a hypothetical scoping bug, a missed + wake degrades to poll latency (`max_fetch_interval`); polling always backstops, so + no message is lost. The design targets zero missed wakes regardless. +- **SQLAlchemy internal reliance.** `get_transaction` / `get_nested_transaction` are + public `AsyncSession` methods (not private), and the transaction objects' identity + + weak-referenceability were verified against the installed SQLAlchemy 2.0.50. From 2a8414f128d9dfbd2bfe585f89ded0e185a1bc3b Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Thu, 16 Jul 2026 11:04:24 +0300 Subject: [PATCH 2/4] feat: dedup pg_notify to one statement per transaction and queue Postgres already coalesces identical NOTIFYs per transaction at delivery, so N publishes of the same queue in one caller transaction only need one pg_notify -- the rest are wasted round-trips. Memoize emitted queues in a WeakKeyDictionary keyed on the innermost active transaction so entries GC when the transaction ends, meaning a rolled-back savepoint can never suppress a later real NOTIFY in the outer transaction. --- architecture/producer.md | 4 ++ faststream_outbox/publisher/producer.py | 27 +++++-- tests/test_integration.py | 94 +++++++++++++++++++++++++ 3 files changed, 121 insertions(+), 4 deletions(-) diff --git a/architecture/producer.md b/architecture/producer.md index 105292f..6709e32 100644 --- a/architecture/producer.md +++ b/architecture/producer.md @@ -16,6 +16,10 @@ Session-type / queue / activate-args-mutex / tz validation lives in one shared ` `from_cmd` raises (relay chaining is unsupported here). +## NOTIFY dedup per transaction + +`_notify` emits at most one `pg_notify` per `(transaction, queue)`, deduped via a `WeakKeyDictionary` memo keyed on the innermost active transaction (`session.get_nested_transaction() or session.get_transaction()`). This is behavior-preserving and default-on — no config knob: Postgres already coalesces identical NOTIFYs per transaction at delivery, so the dedup only removes redundant round-trips, and the subscriber already tolerates coalesced/duplicate wakes. The memo entry for a transaction GCs when that transaction object is collected, so a rolled-back savepoint's entry can never suppress a later real NOTIFY for the same queue once control returns to the outer transaction. + ## Publisher wrapper `broker.publisher(queue, *, headers=None, title=None, description=None, schema=None, include_in_schema=True)` returns an `OutboxPublisher` — a typed wrapper around `broker.publish` with the same transactional contract. Static decorator headers merge with per-call headers (per-call wins). diff --git a/faststream_outbox/publisher/producer.py b/faststream_outbox/publisher/producer.py index 51495e9..c7d239f 100644 --- a/faststream_outbox/publisher/producer.py +++ b/faststream_outbox/publisher/producer.py @@ -9,6 +9,7 @@ import time import typing +from weakref import WeakKeyDictionary from faststream._internal.endpoint.utils import ParserComposition from faststream._internal.parser import DefaultCodec @@ -47,6 +48,10 @@ def __init__( ) -> None: self._table = table self._channel = f"outbox_{table.name}" + # Per-transaction NOTIFY dedup memo (see _notify). Keyed on the innermost active + # transaction object, weak-referenced so entries GC when the transaction ends -- no + # explicit reset, and a rolled-back savepoint cannot suppress a later real NOTIFY. + self._notified: WeakKeyDictionary[typing.Any, set[str]] = WeakKeyDictionary() self.serializer: SerializerProto | None = None # ProducerProto[0.7] requires a `codec` attribute. The outbox owns its # own encoding pipeline (_encode_payload) and never reads this attribute @@ -210,10 +215,24 @@ async def request(self, cmd: OutboxPublishCommand) -> typing.NoReturn: raise NotImplementedError(_REQUEST_UNSUPPORTED_MSG) 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 the caller's session so NOTIFY commits with the row insert; rollback - # silently discards it. Non-Postgres dialects ignore it. + # ``pg_notify(:channel, :payload)`` -- parameterized so channel and payload + # bind cleanly (raw NOTIFY accepts only literals -- injection-prone). Runs on + # the caller's session so NOTIFY commits with the row insert; rollback silently + # discards it. Non-Postgres dialects ignore it. + # + # Dedup: Postgres already coalesces identical (channel, payload) NOTIFYs per + # transaction at delivery, so a second pg_notify for the same queue in the same + # transaction only wastes a round-trip. Memoize emitted queues per innermost + # active transaction (get_nested_transaction() during a savepoint, else the outer + # transaction) -- each a distinct, weak-referenceable object that GCs when the + # transaction ends, so a rolled-back savepoint cannot suppress a later real NOTIFY. + # No active transaction -> emit unconditionally. + txn = session.get_nested_transaction() or session.get_transaction() + if txn is not None: + emitted = self._notified.setdefault(txn, set()) + if queue in emitted: + return + emitted.add(queue) await session.execute( text("SELECT pg_notify(:channel, :payload)"), {"channel": self._channel, "payload": queue}, diff --git a/tests/test_integration.py b/tests/test_integration.py index 8606159..0b65e6a 100644 --- a/tests/test_integration.py +++ b/tests/test_integration.py @@ -1,6 +1,7 @@ """Integration tests against real Postgres. Requires docker-compose postgres up.""" import asyncio +import contextlib import datetime as _dt import logging import uuid @@ -2242,3 +2243,96 @@ async def test_validate_schema_autovacuum_ok_when_applied_in_named_schema(pg_eng async with pg_engine.begin() as conn: await conn.run_sync(metadata.drop_all) await conn.execute(text(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE')) + + +@contextlib.contextmanager +def _count_pg_notify(engine: AsyncEngine): + """Count executed ``pg_notify`` statements on *engine* via a cursor-execute listener. + + Returns a one-element mutable list so a test can also reset it mid-transaction. + Listens on ``engine.sync_engine`` (the AsyncEngine routes DBAPI execution through it). + """ + count = [0] + + def _listener(conn, cursor, statement, parameters, context, executemany) -> None: # noqa: ARG001 + if "pg_notify" in statement.lower(): + count[0] += 1 + + event.listen(engine.sync_engine, "before_cursor_execute", _listener) + try: + yield count + finally: + event.remove(engine.sync_engine, "before_cursor_execute", _listener) + + +async def test_notify_deduped_within_transaction_same_queue(pg_engine, outbox_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(): + for _ in range(5): + await broker.publish({"n": 1}, queue="orders", session=session) + assert count[0] == 1 # 5 publishes, one queue, one txn -> one pg_notify + + +async def test_notify_one_per_distinct_queue_within_transaction(pg_engine, outbox_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="a", session=session) + await broker.publish({"n": 2}, queue="b", session=session) + await broker.publish({"n": 3}, queue="a", session=session) # deduped + assert count[0] == 2 # queues a, b -> two pg_notify + + +async def test_notify_re_emits_in_separate_transaction(pg_engine, outbox_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: + async with session.begin(): + await broker.publish({"n": 1}, queue="orders", session=session) + await broker.publish({"n": 2}, queue="orders", session=session) # deduped + async with session.begin(): # new transaction, same session + await broker.publish({"n": 3}, queue="orders", session=session) # must re-emit + assert count[0] == 2 # one per transaction (the per-transaction reset) + + +async def test_notify_re_emits_after_savepoint_rollback(pg_engine, outbox_table) -> None: + # Innermost-transaction key: a publish inside a rolled-back savepoint must NOT + # suppress a later publish of the same queue in the outer transaction. + broker = OutboxBroker(pg_engine, outbox_table=outbox_table) + session_factory = async_sessionmaker(pg_engine, expire_on_commit=False) + + class _RollbackError(Exception): + pass + + with _count_pg_notify(pg_engine) as count: + async with session_factory() as session, session.begin(): + with contextlib.suppress(_RollbackError): + async with session.begin_nested(): + await broker.publish({"n": 1}, queue="orders", session=session) + raise _RollbackError # roll the savepoint back + count[0] = 0 # reset AFTER the savepoint: isolate the outer publish + await broker.publish({"n": 2}, queue="orders", session=session) # must emit + assert count[0] == 1 # the outer republish emitted despite the rolled-back savepoint's publish + + +async def test_notify_dedup_spans_publish_batch_and_publish(pg_engine, outbox_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_batch({"n": 1}, {"n": 2}, queue="orders", session=session) # 1 pg_notify + await broker.publish({"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) + 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, activate_in=_dt.timedelta(minutes=5)) + assert count[0] == 0 # future-dated -> no NOTIFY (unchanged), dedup did not break the skip From f371806fb8a80877c8f5143d421f6d4b77e3aed8 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Thu, 16 Jul 2026 11:25:45 +0300 Subject: [PATCH 3/4] fix(publisher): key NOTIFY dedup memo on sync transaction, not async proxy session.get_nested_transaction()/get_transaction() return the async AsyncSessionTransaction proxy, which SQLAlchemy regenerates per call and only weak-references. Under autobegin (no explicit session.begin()) the proxy is GC'd between publishes, so the WeakKeyDictionary memo never survives to the next publish and the dedup silently no-ops. Key on session.sync_session's SessionTransaction instead -- the Session holds it strongly for the transaction's lifetime, so the memo works for both session.begin() and autobegin usage. Add a regression test that reproduces the gap (5 same-queue publishes in one autobegin transaction emitted 5 pg_notify before the fix) and fix two test-session mocks that lacked a sync_session attribute. --- architecture/producer.md | 2 +- faststream_outbox/publisher/producer.py | 10 ++++++++-- tests/test_integration.py | 14 ++++++++++++++ tests/test_middleware_prometheus.py | 2 ++ tests/test_unit.py | 7 +++++-- 5 files changed, 30 insertions(+), 5 deletions(-) diff --git a/architecture/producer.md b/architecture/producer.md index 6709e32..f1e69db 100644 --- a/architecture/producer.md +++ b/architecture/producer.md @@ -18,7 +18,7 @@ Session-type / queue / activate-args-mutex / tz validation lives in one shared ` ## NOTIFY dedup per transaction -`_notify` emits at most one `pg_notify` per `(transaction, queue)`, deduped via a `WeakKeyDictionary` memo keyed on the innermost active transaction (`session.get_nested_transaction() or session.get_transaction()`). This is behavior-preserving and default-on — no config knob: Postgres already coalesces identical NOTIFYs per transaction at delivery, so the dedup only removes redundant round-trips, and the subscriber already tolerates coalesced/duplicate wakes. The memo entry for a transaction GCs when that transaction object is collected, so a rolled-back savepoint's entry can never suppress a later real NOTIFY for the same queue once control returns to the outer transaction. +`_notify` emits at most one `pg_notify` per `(transaction, queue)`, deduped via a `WeakKeyDictionary` memo keyed on the innermost active *sync* transaction (`session.sync_session.get_nested_transaction() or session.sync_session.get_transaction()`). This is behavior-preserving and default-on — no config knob: Postgres already coalesces identical NOTIFYs per transaction at delivery, so the dedup only removes redundant round-trips, and the subscriber already tolerates coalesced/duplicate wakes. The memo entry for a transaction GCs when that transaction object is collected, so a rolled-back savepoint's entry can never suppress a later real NOTIFY for the same queue once control returns to the outer transaction. The guarantee holds for the caller's transaction regardless of whether they use an explicit `session.begin()` or rely on autobegin: keying on the sync `SessionTransaction` (which the `Session` holds strongly for the transaction's lifetime) rather than the async `AsyncSessionTransaction` proxy (regenerated per call, only weak-referenced) means the memo entry survives between autobegin publishes instead of GC'ing and silently losing the dedup. ## Publisher wrapper diff --git a/faststream_outbox/publisher/producer.py b/faststream_outbox/publisher/producer.py index c7d239f..26ae42e 100644 --- a/faststream_outbox/publisher/producer.py +++ b/faststream_outbox/publisher/producer.py @@ -226,8 +226,14 @@ async def _notify(self, session: typing.Any, queue: str) -> None: # active transaction (get_nested_transaction() during a savepoint, else the outer # transaction) -- each a distinct, weak-referenceable object that GCs when the # transaction ends, so a rolled-back savepoint cannot suppress a later real NOTIFY. - # No active transaction -> emit unconditionally. - txn = session.get_nested_transaction() or session.get_transaction() + # Key on the *sync* SessionTransaction, not the async AsyncSessionTransaction proxy: + # SQLAlchemy regenerates that proxy on every call and only weak-references it, so + # under autobegin (no explicit `session.begin()`) it would GC between publishes and + # 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()) if queue in emitted: diff --git a/tests/test_integration.py b/tests/test_integration.py index 0b65e6a..0524be4 100644 --- a/tests/test_integration.py +++ b/tests/test_integration.py @@ -2336,3 +2336,17 @@ async def test_notify_still_skipped_for_future_dated(pg_engine, outbox_table) -> async with session_factory() as session, session.begin(): await broker.publish({"n": 1}, queue="orders", session=session, activate_in=_dt.timedelta(minutes=5)) assert count[0] == 0 # future-dated -> no NOTIFY (unchanged), dedup did not break the skip + + +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 + # transaction proxy. + 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: + for _ in range(5): + await broker.publish({"n": 1}, queue="orders", session=session) + await session.commit() + assert count[0] == 1 # 5 publishes, one queue, one autobegin transaction -> one pg_notify diff --git a/tests/test_middleware_prometheus.py b/tests/test_middleware_prometheus.py index 8d99120..37c984d 100644 --- a/tests/test_middleware_prometheus.py +++ b/tests/test_middleware_prometheus.py @@ -41,6 +41,8 @@ def _session_mock() -> AsyncMock: s = AsyncMock(spec=AsyncSession) s.execute.return_value = MagicMock() s.execute.return_value.scalar.return_value = 42 + # _notify keys its dedup memo on session.sync_session -- give it a plain mock. + s.sync_session = MagicMock() return s diff --git a/tests/test_unit.py b/tests/test_unit.py index 9c30d1e..c5253dc 100644 --- a/tests/test_unit.py +++ b/tests/test_unit.py @@ -95,6 +95,9 @@ def _make_session_mock(*, scalar_return: object = 42) -> AsyncMock: session = AsyncMock(spec=AsyncSession) session.execute.return_value = MagicMock() session.execute.return_value.scalar.return_value = scalar_return + # _notify keys its dedup memo on session.sync_session (the sync SessionTransaction), + # not the spec'd AsyncSession -- give it a plain mock so attribute access succeeds. + session.sync_session = MagicMock() return session @@ -1039,7 +1042,7 @@ async def test_broker_publish_batch_with_activate_at_skips_notify() -> None: async def test_broker_publish_batch_emits_notify_when_activate_at_is_past() -> None: broker = _make_broker() - session = AsyncMock(spec=AsyncSession) + session = _make_session_mock() past = _dt.datetime.now(tz=_dt.UTC) - _dt.timedelta(seconds=5) await broker.publish_batch(b"a", b"b", queue="orders", session=session, activate_at=past) # INSERT + NOTIFY: past activate_at is immediately eligible. @@ -1159,7 +1162,7 @@ async def test_broker_cancel_timer_returns_false_when_nothing_deleted() -> None: async def test_broker_publish_batch_executes_single_insert_for_many_rows() -> None: broker = _make_broker() - session = AsyncMock(spec=AsyncSession) + session = _make_session_mock() await broker.publish_batch(b"a", b"b", b"c", queue="orders", session=session) # Two execute calls: the multi-row INSERT, then SELECT pg_notify(...). assert session.execute.await_count == 2 From 59c0c92c0d85d5a41f2feadd81e5a236191b0a0f Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Thu, 16 Jul 2026 11:35:02 +0300 Subject: [PATCH 4/4] bench: gate the producer NOTIFY dedup (select_calls 5000 -> 1) --- benchmarks/baseline.json | 136 +++++++++++++++++++-------------------- 1 file changed, 68 insertions(+), 68 deletions(-) diff --git a/benchmarks/baseline.json b/benchmarks/baseline.json index 710c326..17d42aa 100644 --- a/benchmarks/baseline.json +++ b/benchmarks/baseline.json @@ -1,7 +1,7 @@ { "runs": { "consumer/w1/b10": { - "calls": 14880, + "calls": 14712, "dead_tup": 10000, "delete_calls": 5000, "delete_calls_per_msg": 1.0, @@ -9,22 +9,22 @@ "index_bytes": 344064, "insert_calls": 0, "messages": 5000, - "msgs_per_second": 441.21057661925147, - "round_trips_per_msg": 2.976, + "msgs_per_second": 409.5546765342868, + "round_trips_per_msg": 2.9424, "select_calls": 0, "tup_del": 5000, "tup_hot_upd": 0, "tup_ins": 0, "tup_upd": 5000, - "wal_bytes": 4490440, - "wal_bytes_per_msg": 898.088, + "wal_bytes": 4497344, + "wal_bytes_per_msg": 899.4688, "wal_fpi": 242, - "wal_records": 36649, - "wal_records_per_msg": 7.3298, - "wall_seconds": 11.332457255019108 + "wal_records": 36554, + "wal_records_per_msg": 7.3108, + "wall_seconds": 12.208382144017378 }, "consumer/w1/b100": { - "calls": 14658, + "calls": 14616, "dead_tup": 10000, "delete_calls": 5000, "delete_calls_per_msg": 1.0, @@ -32,19 +32,19 @@ "index_bytes": 344064, "insert_calls": 0, "messages": 5000, - "msgs_per_second": 450.1358749815796, - "round_trips_per_msg": 2.9316, + "msgs_per_second": 466.9523359149118, + "round_trips_per_msg": 2.9232, "select_calls": 0, "tup_del": 5000, "tup_hot_upd": 0, "tup_ins": 0, "tup_upd": 5000, - "wal_bytes": 4643216, - "wal_bytes_per_msg": 928.6432, + "wal_bytes": 4638792, + "wal_bytes_per_msg": 927.7584, "wal_fpi": 245, - "wal_records": 37462, - "wal_records_per_msg": 7.4924, - "wall_seconds": 11.10775718599325 + "wal_records": 37372, + "wal_records_per_msg": 7.4744, + "wall_seconds": 10.707730994006852 }, "consumer/w1/b100/tfbs100": { "calls": 204, @@ -55,114 +55,114 @@ "index_bytes": 352256, "insert_calls": 0, "messages": 5000, - "msgs_per_second": 9673.101687744125, + "msgs_per_second": 10183.536244050012, "round_trips_per_msg": 0.0408, "select_calls": 0, "tup_del": 5000, "tup_hot_upd": 0, "tup_ins": 0, "tup_upd": 5000, - "wal_bytes": 5580984, - "wal_bytes_per_msg": 1116.1968, + "wal_bytes": 5580992, + "wal_bytes_per_msg": 1116.1984, "wal_fpi": 243, "wal_records": 30330, "wal_records_per_msg": 6.066, - "wall_seconds": 0.5168972850078717 + "wall_seconds": 0.49098858001525514 }, "consumer/w2/b10": { - "calls": 8586, + "calls": 9303, "dead_tup": 10000, "delete_calls": 5000, "delete_calls_per_msg": 1.0, - "heap_bytes": 2121728, - "index_bytes": 278528, + "heap_bytes": 2080768, + "index_bytes": 294912, "insert_calls": 0, "messages": 5000, - "msgs_per_second": 1189.9545047874471, - "round_trips_per_msg": 1.7172, + "msgs_per_second": 975.4114938640164, + "round_trips_per_msg": 1.8606, "select_calls": 0, "tup_del": 5000, "tup_hot_upd": 0, "tup_ins": 0, "tup_upd": 5000, - "wal_bytes": 4721320, - "wal_bytes_per_msg": 944.264, + "wal_bytes": 4704512, + "wal_bytes_per_msg": 940.9024, "wal_fpi": 249, - "wal_records": 34016, - "wal_records_per_msg": 6.8032, - "wall_seconds": 4.201841314003104 + "wal_records": 34340, + "wal_records_per_msg": 6.868, + "wall_seconds": 5.126041707990225 }, "consumer/w2/b100": { - "calls": 8943, + "calls": 9126, "dead_tup": 10000, "delete_calls": 5000, "delete_calls_per_msg": 1.0, - "heap_bytes": 2424832, + "heap_bytes": 2400256, "index_bytes": 294912, "insert_calls": 0, "messages": 5000, - "msgs_per_second": 853.0250182265884, - "round_trips_per_msg": 1.7886, + "msgs_per_second": 1022.17433469905, + "round_trips_per_msg": 1.8252, "select_calls": 0, "tup_del": 5000, "tup_hot_upd": 0, "tup_ins": 0, "tup_upd": 5000, - "wal_bytes": 4668904, - "wal_bytes_per_msg": 933.7808, + "wal_bytes": 4653304, + "wal_bytes_per_msg": 930.6608, "wal_fpi": 243, - "wal_records": 34237, - "wal_records_per_msg": 6.8474, - "wall_seconds": 5.8614927970047574 + "wal_records": 34296, + "wal_records_per_msg": 6.8592, + "wall_seconds": 4.891533499001525 }, "consumer/w4/b10": { - "calls": 8007, + "calls": 8010, "dead_tup": 10000, "delete_calls": 5000, "delete_calls_per_msg": 1.0, - "heap_bytes": 2146304, + "heap_bytes": 2088960, "index_bytes": 278528, "insert_calls": 0, "messages": 5000, - "msgs_per_second": 1161.7567625270935, - "round_trips_per_msg": 1.6014, + "msgs_per_second": 1160.5565577478487, + "round_trips_per_msg": 1.602, "select_calls": 0, "tup_del": 5000, "tup_hot_upd": 0, "tup_ins": 0, "tup_upd": 5000, - "wal_bytes": 4675072, - "wal_bytes_per_msg": 935.0144, - "wal_fpi": 242, - "wal_records": 33711, - "wal_records_per_msg": 6.7422, - "wall_seconds": 4.30382689498947 + "wal_bytes": 4924936, + "wal_bytes_per_msg": 984.9872, + "wal_fpi": 267, + "wal_records": 34166, + "wal_records_per_msg": 6.8332, + "wall_seconds": 4.308277753996663 }, "consumer/w4/b100": { - "calls": 6729, + "calls": 7029, "dead_tup": 10000, "delete_calls": 5000, "delete_calls_per_msg": 1.0, - "heap_bytes": 2768896, + "heap_bytes": 2662400, "index_bytes": 270336, "insert_calls": 0, "messages": 5000, - "msgs_per_second": 1589.6276667449354, - "round_trips_per_msg": 1.3458, + "msgs_per_second": 1729.9536695879196, + "round_trips_per_msg": 1.4058, "select_calls": 0, "tup_del": 5000, "tup_hot_upd": 0, "tup_ins": 0, "tup_upd": 5000, - "wal_bytes": 4923224, - "wal_bytes_per_msg": 984.6448, + "wal_bytes": 4836824, + "wal_bytes_per_msg": 967.3648, "wal_fpi": 243, - "wal_records": 33866, - "wal_records_per_msg": 6.7732, - "wall_seconds": 3.1453906500246376 + "wal_records": 33755, + "wal_records_per_msg": 6.751, + "wall_seconds": 2.890250813012244 }, "producer/w1/b100": { - "calls": 10002, + "calls": 5003, "dead_tup": 0, "delete_calls": 0, "delete_calls_per_msg": 0.0, @@ -170,19 +170,19 @@ "index_bytes": 196608, "insert_calls": 5000, "messages": 5000, - "msgs_per_second": 4169.062244595134, - "round_trips_per_msg": 2.0004, - "select_calls": 5000, + "msgs_per_second": 7409.4357487481775, + "round_trips_per_msg": 1.0006, + "select_calls": 1, "tup_del": 0, "tup_hot_upd": 0, "tup_ins": 5000, "tup_upd": 0, - "wal_bytes": 2925984, - "wal_bytes_per_msg": 585.1968, - "wal_fpi": 1, - "wal_records": 15220, - "wal_records_per_msg": 3.044, - "wall_seconds": 1.1993104699940886 + "wal_bytes": 2918536, + "wal_bytes_per_msg": 583.7072, + "wal_fpi": 0, + "wal_records": 15219, + "wal_records_per_msg": 3.0438, + "wall_seconds": 0.674815217993455 } } }