perf(publisher): dedup pg_notify to one statement per (transaction, queue) (#135)

* 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.

* 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.

* 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.

* bench: gate the producer NOTIFY dedup (select_calls 5000 -> 1)

Author: lesnik512

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 *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
 
 `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).

benchmarks/baseline.json

@@ -1,50 +1,50 @@
 {
   "runs": {
     "consumer/w1/b10": {
-      "calls": 14880,
+      "calls": 14712,
       "dead_tup": 10000,
       "delete_calls": 5000,
       "delete_calls_per_msg": 1.0,
       "heap_bytes": 1875968,
       "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,
       "heap_bytes": 2048000,
       "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,134 +55,134 @@
       "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,
       "heap_bytes": 2162688,
       "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
     }
   }
 }

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,30 @@ 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.
+        # 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:
+                return
+            emitted.add(queue)
         await session.execute(
             text("SELECT pg_notify(:channel, :payload)"),
             {"channel": self._channel, "payload": queue},