From e57112efe15b7014c818283028c1da15420a11d2 Mon Sep 17 00:00:00 2001
From: mhenrixon <mikael@mhenrixon.com>
Date: Wed, 12 Aug 2026 17:11:02 +0200
Subject: [PATCH 1/4] fix(client): schema install must not commit or destroy a
 caller's transaction

Follow-up to #398 review (P1 + P3):

- The Proc-supplied shared connection can arrive mid-transaction
  (perform_later inside an application `transaction do`); the unconditional
  BEGIN was a warning no-op there, so the matching COMMIT/ROLLBACK operated
  on the CALLER's transaction. The framing is now transaction_status-aware:
  idle -> owned BEGIN..COMMIT, in-transaction -> SAVEPOINT/RELEASE with
  ROLLBACK TO SAVEPOINT on failure.
- The process-wide serialization spec's fixed `sleep 0.05` could false-pass
  on a saturated scheduler; it now waits deterministically for thread B to
  block or terminate before asserting zero connection traffic.

Refs #398
---
 CHANGELOG.md              |  2 ++
 lib/pgbus/client.rb       | 64 +++++++++++++++++++++++++++++++++------
 spec/pgbus/client_spec.rb | 63 +++++++++++++++++++++++++++++++++++++-
 3 files changed, 118 insertions(+), 11 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 198d6f0..1aedc6a 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,8 @@
 
 ### Fixed
 
+- **The schema-install transaction framing no longer commits or destroys a caller's open transaction (#398 review follow-up).** The #397 fix wrapped check+install in `BEGIN`…`COMMIT`/`ROLLBACK` unconditionally. On the Proc-supplied shared-connection path (the Rails lambda), the connection can arrive **mid-transaction** — e.g. `perform_later` inside an application `transaction do` block — where `BEGIN` is a warning-level no-op and the matching `COMMIT`/`ROLLBACK` then commits half of, or destroys, the *caller's* transaction. The framing is now `transaction_status`-aware: an idle connection gets the owned `BEGIN`…`COMMIT` as before; a connection already inside a transaction rides it via `SAVEPOINT pgbus_pgmq_install` / `RELEASE` (`ROLLBACK TO SAVEPOINT` on failure), so the caller's transaction is never touched. On the savepoint path the advisory lock joins the caller's transaction and is held until it ends — over-holding only delays a concurrent installer, never corrupts it. Refs #398.
+
 - **PGMQ schema installation is now race-safe across clients and processes (issue #397).** `ensure_pgmq_schema` guarded check+install with `@schema_ensured` + `synchronized` — both per-instance, and `synchronized` is a no-op on the dedicated-connection path — so two Client instances (or two threads on the dedicated path) could install concurrently. The loser's `PG::UniqueViolation` (`Key (nspname)=(pgmq) already exists`) surfaced as `SchemaNotReady` even though the schema was fine, and on the shared-AR Proc path — where two instances each hold their *own* mutex around one shared libpq connection — the concurrent install traffic desynced the protocol (`message type 0x… arrived from server while idle`) and left a thread blocked on a socket read forever (downstream forensics: a CI shard going silent until the merge queue's timeout evicted the PR, getzazu/app#3413). Three changes: **(1)** schema bootstrap is serialized process-wide through a class-level mutex, not per-instance state; **(2)** check+install runs inside one explicit transaction holding `pg_advisory_xact_lock` on a fixed key (`Pgbus::Client::PGMQ_INSTALL_LOCK_KEY`), serializing installers across processes — xact-scoped so the lock releases itself at COMMIT/ROLLBACK and stays safe through transaction-pooling poolers, where a session lock's unlock could land on a different server connection; **(3)** a duplicate-object install failure (`PG::UniqueViolation`, `PG::DuplicateSchema`, `PG::DuplicateTable`, `PG::DuplicateObject`, `PG::DuplicateFunction` — a process without the advisory lock, e.g. older pgbus or the extension path, won the race) is rescued by re-checking `pgmq.meta`: present means proceed as installed, absent means the original error is re-raised wrapped in `SchemaNotReady`. Refs #397.
 
 - **SSE delivery no longer strips newlines from broadcast payloads — multiline payloads are framed as consecutive `data:` lines per the SSE spec (issue #392).** `Streams::Envelope.message` collapsed `\r`/`\n` in the payload to nothing before writing the single `data:` line, silently corrupting any whitespace-significant broadcast (pre-formatted `<pre>` content, textarea seeds, JSON-in-data frames) on **both** the ephemeral and durable delivery paths — HTML's whitespace tolerance is why it went unnoticed. A multiline payload is now split on `\r\n`/`\r`/`\n` into consecutive `data:` lines, which EventSource clients rejoin with `\n`, making delivery lossless (a trailing newline survives via an empty final `data:` line; `\r` variants normalize to `\n` — SSE line terminators cannot be carried raw). The original injection defense is preserved: every payload line carries the `data:` prefix followed by one space, so a crafted payload still cannot forge `id:`/`event:` fields, and single-line fields (event names, comments) still strip newlines. The `<pgbus-stream-source>` element's fetch-path parser had the matching client-side bug — it joined `data:` lines without `\n` *and* `trim()`ed payload whitespace — and now follows EventSource semantics (join with `\n`, strip only the single leading space). Refs #392.
diff --git a/lib/pgbus/client.rb b/lib/pgbus/client.rb
index 722defe..227a0c9 100644
--- a/lib/pgbus/client.rb
+++ b/lib/pgbus/client.rb
@@ -28,6 +28,9 @@ class Client
     PGMQ_META_CHECK_SQL = "SELECT 1 FROM pg_tables WHERE schemaname = 'pgmq' AND tablename = 'meta' LIMIT 1"
     private_constant :PGMQ_META_CHECK_SQL
 
+    PGMQ_INSTALL_SAVEPOINT = "pgbus_pgmq_install"
+    private_constant :PGMQ_INSTALL_SAVEPOINT
+
     # Install-race losers see the winner's DDL as one of these. Matched by
     # class NAME so the check works whether or not the pg gem's generated
     # error classes are loaded in this process (mirrors the defined?(PG::…)
@@ -985,30 +988,71 @@ def ensure_pgmq_schema
             "Ensure the pgbus database exists and migrations have been run."
     end
 
-    # Check-and-install inside one transaction holding a fixed advisory lock:
-    # pg_advisory_xact_lock serializes installers across processes and releases
-    # itself at COMMIT/ROLLBACK — safe through transaction-pooling poolers,
-    # where a session-level lock could be released on a different server
-    # connection than the one that acquired it (issue #397).
+    # Check-and-install under a fixed advisory lock: pg_advisory_xact_lock
+    # serializes installers across processes and releases itself when its
+    # transaction ends — safe through transaction-pooling poolers, where a
+    # session-level lock could be released on a different server connection
+    # than the one that acquired it (issue #397).
+    #
+    # The transactional framing must respect who owns the transaction. A
+    # Proc-supplied shared connection (the Rails-lambda path) can arrive
+    # mid-transaction — e.g. perform_later inside an application
+    # `transaction do` block. BEGIN there is a warning-level no-op, and the
+    # matching COMMIT/ROLLBACK would then commit or destroy the CALLER's
+    # transaction (#398 review). So: own the transaction only when the
+    # connection is idle; ride the caller's transaction via a savepoint
+    # otherwise.
     def install_pgmq_schema_serialized(conn)
+      if inside_caller_transaction?(conn)
+        install_pgmq_schema_in_savepoint(conn)
+      else
+        install_pgmq_schema_in_own_transaction(conn)
+      end
+    end
+
+    # respond_to? guard: a Proc can hand back any connection-shaped object;
+    # only a real PG::Connection reports transaction_status (and its presence
+    # guarantees the PG constants below are loaded).
+    def inside_caller_transaction?(conn)
+      conn.respond_to?(:transaction_status) && conn.transaction_status != PG::PQTRANS_IDLE
+    end
+
+    def install_pgmq_schema_in_own_transaction(conn)
       conn.exec("BEGIN")
       conn.exec("SELECT pg_advisory_xact_lock(#{PGMQ_INSTALL_LOCK_KEY})")
       install_pgmq_schema(conn) if conn.exec(PGMQ_META_CHECK_SQL).ntuples.zero?
       conn.exec("COMMIT")
     rescue StandardError => e
+      recover_from_install_failure(conn, e, "ROLLBACK")
+    end
+
+    # The advisory lock joins the CALLER's transaction here, so it is held
+    # until that transaction ends — longer than the install needs, but xact
+    # locks cannot be released early by design, and over-holding only delays
+    # a concurrent installer, never corrupts it.
+    def install_pgmq_schema_in_savepoint(conn)
+      conn.exec("SAVEPOINT #{PGMQ_INSTALL_SAVEPOINT}")
+      conn.exec("SELECT pg_advisory_xact_lock(#{PGMQ_INSTALL_LOCK_KEY})")
+      install_pgmq_schema(conn) if conn.exec(PGMQ_META_CHECK_SQL).ntuples.zero?
+      conn.exec("RELEASE SAVEPOINT #{PGMQ_INSTALL_SAVEPOINT}")
+    rescue StandardError => e
+      recover_from_install_failure(conn, e, "ROLLBACK TO SAVEPOINT #{PGMQ_INSTALL_SAVEPOINT}")
+    end
+
+    def recover_from_install_failure(conn, error, rollback_sql)
       begin
-        conn.exec("ROLLBACK")
+        conn.exec(rollback_sql)
       rescue StandardError
-        # A connection broken enough to refuse ROLLBACK also fails the
+        # A connection broken enough to refuse the rollback also fails the
         # re-check below, which surfaces the state honestly; re-raising the
-        # ROLLBACK error here would mask the original install failure.
+        # rollback error here would mask the original install failure.
       end
-      raise e unless duplicate_install_error?(e)
+      raise error unless duplicate_install_error?(error)
 
       # A process without the advisory lock (older pgbus, or the extension
       # path) won the install race — re-check instead of failing on its
       # success.
-      raise e if conn.exec(PGMQ_META_CHECK_SQL).ntuples.zero?
+      raise error if conn.exec(PGMQ_META_CHECK_SQL).ntuples.zero?
     end
 
     def duplicate_install_error?(error)
diff --git a/spec/pgbus/client_spec.rb b/spec/pgbus/client_spec.rb
index 409f603..21b1eae 100644
--- a/spec/pgbus/client_spec.rb
+++ b/spec/pgbus/client_spec.rb
@@ -135,6 +135,57 @@ def initialize(*args, **kwargs); end
       expect(raw_conn).to have_received(:exec).with("COMMIT").ordered
     end
 
+    context "when the connection is already inside a caller's transaction (#398 review P1)" do
+      before do
+        require "pg"
+        allow(raw_conn).to receive(:transaction_status).and_return(PG::PQTRANS_INTRANS)
+        allow(raw_conn).to receive(:exec).with(/SAVEPOINT/)
+      end
+
+      it "frames check+install in a savepoint and never issues BEGIN/COMMIT" do
+        allow(raw_conn).to receive(:exec).with(/pg_tables.*pgmq.*meta/).and_return(double(ntuples: 0))
+        allow(raw_conn).to receive(:exec).with(/pg_available_extensions/).and_return(double(ntuples: 0))
+        allow(raw_conn).to receive(:exec).with(Pgbus::PgmqSchema.install_sql).and_return(nil)
+
+        client.ensure_queue("jobs")
+
+        expect(raw_conn).to have_received(:exec).with("SAVEPOINT pgbus_pgmq_install").ordered
+        expect(raw_conn).to have_received(:exec)
+          .with("SELECT pg_advisory_xact_lock(#{Pgbus::Client::PGMQ_INSTALL_LOCK_KEY})").ordered
+        expect(raw_conn).to have_received(:exec).with(Pgbus::PgmqSchema.install_sql).ordered
+        expect(raw_conn).to have_received(:exec).with("RELEASE SAVEPOINT pgbus_pgmq_install").ordered
+        expect(raw_conn).not_to have_received(:exec).with("BEGIN")
+        expect(raw_conn).not_to have_received(:exec).with("COMMIT")
+      end
+
+      it "rolls back to the savepoint — never the whole transaction — on a duplicate install" do
+        check = double("check_result")
+        allow(check).to receive(:ntuples).and_return(0, 1)
+        allow(raw_conn).to receive(:exec).with(/pg_tables.*pgmq.*meta/).and_return(check)
+        allow(raw_conn).to receive(:exec).with(/pg_available_extensions/).and_return(double(ntuples: 0))
+        allow(raw_conn).to receive(:exec).with(Pgbus::PgmqSchema.install_sql).and_raise(
+          PG::UniqueViolation.new("ERROR: duplicate key value")
+        )
+
+        expect { client.ensure_queue("jobs") }.not_to raise_error
+
+        expect(raw_conn).to have_received(:exec).with("ROLLBACK TO SAVEPOINT pgbus_pgmq_install")
+        expect(raw_conn).not_to have_received(:exec).with("ROLLBACK")
+        expect(raw_conn).not_to have_received(:exec).with("COMMIT")
+      end
+    end
+
+    it "owns the transaction when the connection reports an idle status" do
+      require "pg"
+      allow(raw_conn).to receive(:transaction_status).and_return(PG::PQTRANS_IDLE)
+      allow(raw_conn).to receive(:exec).with(/pg_tables.*pgmq.*meta/).and_return(double(ntuples: 1))
+
+      client.ensure_queue("jobs")
+
+      expect(raw_conn).to have_received(:exec).with("BEGIN")
+      expect(raw_conn).to have_received(:exec).with("COMMIT")
+    end
+
     context "when another process wins the install race (#397)" do
       before { require "pg" }
 
@@ -212,7 +263,17 @@ def unensured_client(conn)
       thread_a = Thread.new { client_a.ensure_queue("jobs") }
       install_started.pop
       thread_b = Thread.new { client_b.ensure_queue("jobs") }
-      sleep 0.05 # window in which an unserialized B would (wrongly) hit its connection
+      # Deterministic (no fixed sleep, #398 review P3): with A parked inside its
+      # install, B is the only runnable thread — wait until it either blocks
+      # (serialized: parked on the install mutex) or terminates (unserialized:
+      # it ran its whole path). At that settled point, zero execs on B's
+      # connection is exactly the serialization property; an unserialized B has
+      # terminated WITH execs recorded and fails the assertion every time.
+      5_000.times do
+        break if [false, nil, "sleep"].include?(thread_b.status)
+
+        sleep 0.001
+      end
       expect(conn_b).not_to have_received(:exec)
 
       release_install << true

From fa9db0f1eb239093164ca00a984263aeb22aea10 Mon Sep 17 00:00:00 2001
From: mhenrixon <mikael@mhenrixon.com>
Date: Wed, 12 Aug 2026 17:26:13 +0200
Subject: [PATCH 2/4] fix(client): don't cache schema_ensured from a
 savepoint-path install
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

A savepoint-path ensure rides the caller's transaction, so the install is
only durable once THAT commits — caching @schema_ensured there means an
outer rollback leaves the schema missing while every future check is
skipped. Only the owned-COMMIT path caches now; the savepoint path
re-checks on the next ensure (one SELECT).

Also: the serialization spec's settle-wait now fails explicitly if thread B
never settles, instead of silently degrading back into a fixed-duration
window.

Refs #399 review
---
 CHANGELOG.md              |  2 +-
 lib/pgbus/client.rb       | 26 ++++++++++++++++----------
 spec/pgbus/client_spec.rb | 29 ++++++++++++++++++++++++-----
 3 files changed, 41 insertions(+), 16 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 1aedc6a..b330ff0 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,7 +2,7 @@
 
 ### Fixed
 
-- **The schema-install transaction framing no longer commits or destroys a caller's open transaction (#398 review follow-up).** The #397 fix wrapped check+install in `BEGIN`…`COMMIT`/`ROLLBACK` unconditionally. On the Proc-supplied shared-connection path (the Rails lambda), the connection can arrive **mid-transaction** — e.g. `perform_later` inside an application `transaction do` block — where `BEGIN` is a warning-level no-op and the matching `COMMIT`/`ROLLBACK` then commits half of, or destroys, the *caller's* transaction. The framing is now `transaction_status`-aware: an idle connection gets the owned `BEGIN`…`COMMIT` as before; a connection already inside a transaction rides it via `SAVEPOINT pgbus_pgmq_install` / `RELEASE` (`ROLLBACK TO SAVEPOINT` on failure), so the caller's transaction is never touched. On the savepoint path the advisory lock joins the caller's transaction and is held until it ends — over-holding only delays a concurrent installer, never corrupts it. Refs #398.
+- **The schema-install transaction framing no longer commits or destroys a caller's open transaction (#398 review follow-up).** The #397 fix wrapped check+install in `BEGIN`…`COMMIT`/`ROLLBACK` unconditionally. On the Proc-supplied shared-connection path (the Rails lambda), the connection can arrive **mid-transaction** — e.g. `perform_later` inside an application `transaction do` block — where `BEGIN` is a warning-level no-op and the matching `COMMIT`/`ROLLBACK` then commits half of, or destroys, the *caller's* transaction. The framing is now `transaction_status`-aware: an idle connection gets the owned `BEGIN`…`COMMIT` as before; a connection already inside a transaction rides it via `SAVEPOINT pgbus_pgmq_install` / `RELEASE` (`ROLLBACK TO SAVEPOINT` on failure), so the caller's transaction is never touched. On the savepoint path the advisory lock joins the caller's transaction and is held until it ends — over-holding only delays a concurrent installer, never corrupts it — and `@schema_ensured` is NOT cached there: the install is only durable once the caller commits, so a cached true after an outer rollback would skip every future check against a missing schema. Refs #398.
 
 - **PGMQ schema installation is now race-safe across clients and processes (issue #397).** `ensure_pgmq_schema` guarded check+install with `@schema_ensured` + `synchronized` — both per-instance, and `synchronized` is a no-op on the dedicated-connection path — so two Client instances (or two threads on the dedicated path) could install concurrently. The loser's `PG::UniqueViolation` (`Key (nspname)=(pgmq) already exists`) surfaced as `SchemaNotReady` even though the schema was fine, and on the shared-AR Proc path — where two instances each hold their *own* mutex around one shared libpq connection — the concurrent install traffic desynced the protocol (`message type 0x… arrived from server while idle`) and left a thread blocked on a socket read forever (downstream forensics: a CI shard going silent until the merge queue's timeout evicted the PR, getzazu/app#3413). Three changes: **(1)** schema bootstrap is serialized process-wide through a class-level mutex, not per-instance state; **(2)** check+install runs inside one explicit transaction holding `pg_advisory_xact_lock` on a fixed key (`Pgbus::Client::PGMQ_INSTALL_LOCK_KEY`), serializing installers across processes — xact-scoped so the lock releases itself at COMMIT/ROLLBACK and stays safe through transaction-pooling poolers, where a session lock's unlock could land on a different server connection; **(3)** a duplicate-object install failure (`PG::UniqueViolation`, `PG::DuplicateSchema`, `PG::DuplicateTable`, `PG::DuplicateObject`, `PG::DuplicateFunction` — a process without the advisory lock, e.g. older pgbus or the extension path, won the race) is rescued by re-checking `pgmq.meta`: present means proceed as installed, absent means the original error is re-raised wrapped in `SchemaNotReady`. Refs #397.
 
diff --git a/lib/pgbus/client.rb b/lib/pgbus/client.rb
index 227a0c9..f942c05 100644
--- a/lib/pgbus/client.rb
+++ b/lib/pgbus/client.rb
@@ -979,8 +979,21 @@ def ensure_pgmq_schema
       self.class.pgmq_install_mutex.synchronize do
         return if @schema_ensured
 
-        with_raw_connection { |raw_conn| install_pgmq_schema_serialized(raw_conn) }
-        @schema_ensured = true
+        # Cache only a durable result: true only when this call owned the
+        # COMMIT. A savepoint-path ensure rides the CALLER's transaction — if
+        # that later rolls back the schema is gone (and even a schema found
+        # already-present there may be the caller's own uncommitted work), so
+        # a cached true would skip every future check (#399 review).
+        durable = with_raw_connection do |raw_conn|
+          if inside_caller_transaction?(raw_conn)
+            install_pgmq_schema_in_savepoint(raw_conn)
+            false
+          else
+            install_pgmq_schema_in_own_transaction(raw_conn)
+            true
+          end
+        end
+        @schema_ensured = true if durable
       end
     rescue StandardError => e
       raise Pgbus::SchemaNotReady,
@@ -1002,14 +1015,7 @@ def ensure_pgmq_schema
     # transaction (#398 review). So: own the transaction only when the
     # connection is idle; ride the caller's transaction via a savepoint
     # otherwise.
-    def install_pgmq_schema_serialized(conn)
-      if inside_caller_transaction?(conn)
-        install_pgmq_schema_in_savepoint(conn)
-      else
-        install_pgmq_schema_in_own_transaction(conn)
-      end
-    end
-
+    #
     # respond_to? guard: a Proc can hand back any connection-shaped object;
     # only a real PG::Connection reports transaction_status (and its presence
     # guarantees the PG constants below are loaded).
diff --git a/spec/pgbus/client_spec.rb b/spec/pgbus/client_spec.rb
index 21b1eae..ad21390 100644
--- a/spec/pgbus/client_spec.rb
+++ b/spec/pgbus/client_spec.rb
@@ -158,6 +158,17 @@ def initialize(*args, **kwargs); end
         expect(raw_conn).not_to have_received(:exec).with("COMMIT")
       end
 
+      it "does not cache schema_ensured — the install is only durable once the caller commits" do
+        allow(raw_conn).to receive(:exec).with(/pg_tables.*pgmq.*meta/).and_return(double(ntuples: 0))
+        allow(raw_conn).to receive(:exec).with(/pg_available_extensions/).and_return(double(ntuples: 0))
+        allow(raw_conn).to receive(:exec).with(Pgbus::PgmqSchema.install_sql).and_return(nil)
+
+        client.ensure_queue("jobs")
+        client.ensure_queue("events")
+
+        expect(raw_conn).to have_received(:exec).with(/pg_tables.*pgmq.*meta/).twice
+      end
+
       it "rolls back to the savepoint — never the whole transaction — on a duplicate install" do
         check = double("check_result")
         allow(check).to receive(:ntuples).and_return(0, 1)
@@ -250,6 +261,18 @@ def unensured_client(conn)
       c
     end
 
+    # Bounded deterministic wait: true once the thread is blocked or
+    # terminated, false if it never settles (#398/#399 review — a fixed sleep
+    # can false-pass, and silent fall-through re-creates a fixed sleep).
+    def settled?(thread)
+      5_000.times do
+        return true if [false, nil, "sleep"].include?(thread.status)
+
+        sleep 0.001
+      end
+      false
+    end
+
     it "serializes installs process-wide across client instances (#397)" do
       install_started = Queue.new
       release_install = Queue.new
@@ -269,11 +292,7 @@ def unensured_client(conn)
       # it ran its whole path). At that settled point, zero execs on B's
       # connection is exactly the serialization property; an unserialized B has
       # terminated WITH execs recorded and fails the assertion every time.
-      5_000.times do
-        break if [false, nil, "sleep"].include?(thread_b.status)
-
-        sleep 0.001
-      end
+      expect(settled?(thread_b)).to be(true), "thread B never settled (blocked or terminated) within the wait budget"
       expect(conn_b).not_to have_received(:exec)
 
       release_install << true

From ae7f6b619eb6f9a233c2f921079d3e790ec1b18f Mon Sep 17 00:00:00 2001
From: mhenrixon <mikael@mhenrixon.com>
Date: Wed, 12 Aug 2026 17:41:49 +0200
Subject: [PATCH 3/4] fix(client): queue-creation cache follows the same
 durability rule

Queue DDL on the shared Proc-supplied connection joins the caller's open
transaction; caching @queues_created there outlives a caller rollback, so
later ensures skip recreation and message operations fail against a
missing queue. When the DDL rides a caller's transaction the queue is
created (idempotent CREATE IF NOT EXISTS) but not cached; dedicated-path
DDL runs on pgmq-ruby's own pool connections and caches as before.

Refs #399 review
---
 CHANGELOG.md              |  2 +-
 lib/pgbus/client.rb       | 54 ++++++++++++++++++++++++++++++---------
 spec/pgbus/client_spec.rb | 36 ++++++++++++++++++++++++++
 3 files changed, 79 insertions(+), 13 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index b330ff0..cd95a63 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,7 +2,7 @@
 
 ### Fixed
 
-- **The schema-install transaction framing no longer commits or destroys a caller's open transaction (#398 review follow-up).** The #397 fix wrapped check+install in `BEGIN`…`COMMIT`/`ROLLBACK` unconditionally. On the Proc-supplied shared-connection path (the Rails lambda), the connection can arrive **mid-transaction** — e.g. `perform_later` inside an application `transaction do` block — where `BEGIN` is a warning-level no-op and the matching `COMMIT`/`ROLLBACK` then commits half of, or destroys, the *caller's* transaction. The framing is now `transaction_status`-aware: an idle connection gets the owned `BEGIN`…`COMMIT` as before; a connection already inside a transaction rides it via `SAVEPOINT pgbus_pgmq_install` / `RELEASE` (`ROLLBACK TO SAVEPOINT` on failure), so the caller's transaction is never touched. On the savepoint path the advisory lock joins the caller's transaction and is held until it ends — over-holding only delays a concurrent installer, never corrupts it — and `@schema_ensured` is NOT cached there: the install is only durable once the caller commits, so a cached true after an outer rollback would skip every future check against a missing schema. Refs #398.
+- **The schema-install transaction framing no longer commits or destroys a caller's open transaction (#398 review follow-up).** The #397 fix wrapped check+install in `BEGIN`…`COMMIT`/`ROLLBACK` unconditionally. On the Proc-supplied shared-connection path (the Rails lambda), the connection can arrive **mid-transaction** — e.g. `perform_later` inside an application `transaction do` block — where `BEGIN` is a warning-level no-op and the matching `COMMIT`/`ROLLBACK` then commits half of, or destroys, the *caller's* transaction. The framing is now `transaction_status`-aware: an idle connection gets the owned `BEGIN`…`COMMIT` as before; a connection already inside a transaction rides it via `SAVEPOINT pgbus_pgmq_install` / `RELEASE` (`ROLLBACK TO SAVEPOINT` on failure), so the caller's transaction is never touched. On the savepoint path the advisory lock joins the caller's transaction and is held until it ends — over-holding only delays a concurrent installer, never corrupts it — and `@schema_ensured` is NOT cached there: the install is only durable once the caller commits, so a cached true after an outer rollback would skip every future check against a missing schema. The same durability rule now governs `@queues_created`: queue DDL on the shared Proc-supplied connection joins the caller's open transaction, so queue creation there runs uncached (idempotent `CREATE IF NOT EXISTS`) and the next ensure re-checks — a cache write outliving a caller rollback would make later message operations fail against a missing queue. Refs #398.
 
 - **PGMQ schema installation is now race-safe across clients and processes (issue #397).** `ensure_pgmq_schema` guarded check+install with `@schema_ensured` + `synchronized` — both per-instance, and `synchronized` is a no-op on the dedicated-connection path — so two Client instances (or two threads on the dedicated path) could install concurrently. The loser's `PG::UniqueViolation` (`Key (nspname)=(pgmq) already exists`) surfaced as `SchemaNotReady` even though the schema was fine, and on the shared-AR Proc path — where two instances each hold their *own* mutex around one shared libpq connection — the concurrent install traffic desynced the protocol (`message type 0x… arrived from server while idle`) and left a thread blocked on a socket read forever (downstream forensics: a CI shard going silent until the merge queue's timeout evicted the PR, getzazu/app#3413). Three changes: **(1)** schema bootstrap is serialized process-wide through a class-level mutex, not per-instance state; **(2)** check+install runs inside one explicit transaction holding `pg_advisory_xact_lock` on a fixed key (`Pgbus::Client::PGMQ_INSTALL_LOCK_KEY`), serializing installers across processes — xact-scoped so the lock releases itself at COMMIT/ROLLBACK and stays safe through transaction-pooling poolers, where a session lock's unlock could land on a different server connection; **(3)** a duplicate-object install failure (`PG::UniqueViolation`, `PG::DuplicateSchema`, `PG::DuplicateTable`, `PG::DuplicateObject`, `PG::DuplicateFunction` — a process without the advisory lock, e.g. older pgbus or the extension path, won the race) is rescued by re-checking `pgmq.meta`: present means proceed as installed, absent means the original error is re-raised wrapped in `SchemaNotReady`. Refs #397.
 
diff --git a/lib/pgbus/client.rb b/lib/pgbus/client.rb
index f942c05..0c97368 100644
--- a/lib/pgbus/client.rb
+++ b/lib/pgbus/client.rb
@@ -323,12 +323,13 @@ def ensure_dead_letter_queue(name)
       dlq_name = config.dead_letter_queue_name(name)
       return if @queues_created[dlq_name]
 
-      @queues_created.compute_if_absent(dlq_name) do
-        synchronized do
-          @pgmq.create(dlq_name)
-          tune_autovacuum(dlq_name)
+      if queue_ddl_rides_caller_transaction?
+        create_dead_letter_queue_physically(dlq_name)
+      else
+        @queues_created.compute_if_absent(dlq_name) do
+          create_dead_letter_queue_physically(dlq_name)
+          true
         end
-        true
       end
     end
 
@@ -1145,17 +1146,46 @@ def with_streams_connection(&)
     def ensure_single_queue(full_name)
       return if @queues_created[full_name]
 
-      @queues_created.compute_if_absent(full_name) do
-        synchronized do
-          @pgmq.create(full_name)
-          tune_autovacuum(full_name)
-          enable_notify_if_needed(full_name, NOTIFY_THROTTLE_MS)
-          create_fifo_index_if_needed(full_name)
+      if queue_ddl_rides_caller_transaction?
+        create_queue_physically(full_name)
+      else
+        @queues_created.compute_if_absent(full_name) do
+          create_queue_physically(full_name)
+          true
         end
-        true
       end
     end
 
+    def create_queue_physically(full_name)
+      synchronized do
+        @pgmq.create(full_name)
+        tune_autovacuum(full_name)
+        enable_notify_if_needed(full_name, NOTIFY_THROTTLE_MS)
+        create_fifo_index_if_needed(full_name)
+      end
+    end
+
+    def create_dead_letter_queue_physically(dlq_name)
+      synchronized do
+        @pgmq.create(dlq_name)
+        tune_autovacuum(dlq_name)
+      end
+    end
+
+    # Queue DDL on the shared Proc-supplied connection joins any transaction
+    # the caller has open, so a @queues_created cache write there outlives a
+    # caller rollback — later ensures would skip recreation and message
+    # operations would fail (#399 review; same durability rule as
+    # @schema_ensured). Create the queue (idempotent CREATE IF NOT EXISTS)
+    # but let the next ensure re-check. Dedicated String/Hash paths run DDL
+    # on pgmq-ruby's own pool connections, never inside an application
+    # transaction, so they always cache.
+    def queue_ddl_rides_caller_transaction?
+      return false unless @shared_connection
+
+      with_raw_connection { |conn| inside_caller_transaction?(conn) }
+    end
+
     def enable_notify_if_needed(full_name, throttle_ms)
       return unless config.listen_notify
       return if notify_trigger_current?(full_name, throttle_ms)
diff --git a/spec/pgbus/client_spec.rb b/spec/pgbus/client_spec.rb
index ad21390..dbb3bb5 100644
--- a/spec/pgbus/client_spec.rb
+++ b/spec/pgbus/client_spec.rb
@@ -302,6 +302,42 @@ def settled?(thread)
     end
   end
 
+  describe "#ensure_queue when queue DDL rides a caller's transaction (#399 review)" do
+    subject(:client) do
+      allow(config).to receive(:connection_options).and_return(-> { raw_conn })
+      allow(PGMQ::Client).to receive(:new).and_return(mock_pgmq)
+      c = described_class.new(config, schema_ensured: true)
+      allow(c).to receive(:tune_autovacuum)
+      allow(c).to receive(:notify_trigger_current?).and_return(false)
+      c
+    end
+
+    let(:raw_conn) { double("raw_conn") }
+
+    before { require "pg" }
+
+    it "creates the queue but does not cache it — the DDL is only durable once the caller commits" do
+      allow(raw_conn).to receive(:transaction_status).and_return(PG::PQTRANS_INTRANS)
+
+      client.ensure_queue("jobs")
+      client.ensure_queue("jobs")
+
+      expect(mock_pgmq).to have_received(:create).with("pgbus_test_jobs").twice
+    end
+
+    it "resumes caching once the shared connection is idle again" do
+      allow(raw_conn).to receive(:transaction_status).and_return(
+        PG::PQTRANS_INTRANS, PG::PQTRANS_IDLE, PG::PQTRANS_IDLE
+      )
+
+      client.ensure_queue("jobs") # inside caller txn — created, not cached
+      client.ensure_queue("jobs") # idle — created and cached
+      client.ensure_queue("jobs") # cache hit
+
+      expect(mock_pgmq).to have_received(:create).with("pgbus_test_jobs").twice
+    end
+  end
+
   describe "#ensure_queue" do
     it "tunes autovacuum when creating a queue" do
       client.ensure_queue("jobs")

From dfac1d3d5a82981b6e25f344c1d848c2d577138d Mon Sep 17 00:00:00 2001
From: mhenrixon <mikael@mhenrixon.com>
Date: Wed, 12 Aug 2026 17:55:54 +0200
Subject: [PATCH 4/4] fix(client): hold the connection mutex for the txn probe
 and schema install
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The shared Proc-supplied PG::Connection is single-owner: every access must
hold @pgmq_mutex. The new transaction-status probe ran outside it, and the
#397/#398 rework had swapped ensure_pgmq_schema's synchronized for the
class-level install mutex — silently dropping the connection-ownership
guard the original code had. Both now hold synchronized (nested inside the
class install mutex; safe order, nothing acquires them reversed).

Also dedupes the create+tune DDL pair shared by queue and DLQ creation.

Refs #399 review
---
 CHANGELOG.md              |  2 +-
 lib/pgbus/client.rb       | 45 +++++++++++++++++++++++++++------------
 spec/pgbus/client_spec.rb | 31 +++++++++++++++++++++++++++
 3 files changed, 63 insertions(+), 15 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index cd95a63..e8bf1ec 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,7 +2,7 @@
 
 ### Fixed
 
-- **The schema-install transaction framing no longer commits or destroys a caller's open transaction (#398 review follow-up).** The #397 fix wrapped check+install in `BEGIN`…`COMMIT`/`ROLLBACK` unconditionally. On the Proc-supplied shared-connection path (the Rails lambda), the connection can arrive **mid-transaction** — e.g. `perform_later` inside an application `transaction do` block — where `BEGIN` is a warning-level no-op and the matching `COMMIT`/`ROLLBACK` then commits half of, or destroys, the *caller's* transaction. The framing is now `transaction_status`-aware: an idle connection gets the owned `BEGIN`…`COMMIT` as before; a connection already inside a transaction rides it via `SAVEPOINT pgbus_pgmq_install` / `RELEASE` (`ROLLBACK TO SAVEPOINT` on failure), so the caller's transaction is never touched. On the savepoint path the advisory lock joins the caller's transaction and is held until it ends — over-holding only delays a concurrent installer, never corrupts it — and `@schema_ensured` is NOT cached there: the install is only durable once the caller commits, so a cached true after an outer rollback would skip every future check against a missing schema. The same durability rule now governs `@queues_created`: queue DDL on the shared Proc-supplied connection joins the caller's open transaction, so queue creation there runs uncached (idempotent `CREATE IF NOT EXISTS`) and the next ensure re-checks — a cache write outliving a caller rollback would make later message operations fail against a missing queue. Refs #398.
+- **The schema-install transaction framing no longer commits or destroys a caller's open transaction (#398 review follow-up).** The #397 fix wrapped check+install in `BEGIN`…`COMMIT`/`ROLLBACK` unconditionally. On the Proc-supplied shared-connection path (the Rails lambda), the connection can arrive **mid-transaction** — e.g. `perform_later` inside an application `transaction do` block — where `BEGIN` is a warning-level no-op and the matching `COMMIT`/`ROLLBACK` then commits half of, or destroys, the *caller's* transaction. The framing is now `transaction_status`-aware: an idle connection gets the owned `BEGIN`…`COMMIT` as before; a connection already inside a transaction rides it via `SAVEPOINT pgbus_pgmq_install` / `RELEASE` (`ROLLBACK TO SAVEPOINT` on failure), so the caller's transaction is never touched. On the savepoint path the advisory lock joins the caller's transaction and is held until it ends — over-holding only delays a concurrent installer, never corrupts it — and `@schema_ensured` is NOT cached there: the install is only durable once the caller commits, so a cached true after an outer rollback would skip every future check against a missing schema. The same durability rule now governs `@queues_created`: queue DDL on the shared Proc-supplied connection joins the caller's open transaction, so queue creation there runs uncached (idempotent `CREATE IF NOT EXISTS`) and the next ensure re-checks — a cache write outliving a caller rollback would make later message operations fail against a missing queue. All shared-connection access in these paths — including the transaction-status probe and the schema install itself — holds the per-instance connection mutex, restoring the single-owner invariant the #397 fix had narrowed. Refs #398.
 
 - **PGMQ schema installation is now race-safe across clients and processes (issue #397).** `ensure_pgmq_schema` guarded check+install with `@schema_ensured` + `synchronized` — both per-instance, and `synchronized` is a no-op on the dedicated-connection path — so two Client instances (or two threads on the dedicated path) could install concurrently. The loser's `PG::UniqueViolation` (`Key (nspname)=(pgmq) already exists`) surfaced as `SchemaNotReady` even though the schema was fine, and on the shared-AR Proc path — where two instances each hold their *own* mutex around one shared libpq connection — the concurrent install traffic desynced the protocol (`message type 0x… arrived from server while idle`) and left a thread blocked on a socket read forever (downstream forensics: a CI shard going silent until the merge queue's timeout evicted the PR, getzazu/app#3413). Three changes: **(1)** schema bootstrap is serialized process-wide through a class-level mutex, not per-instance state; **(2)** check+install runs inside one explicit transaction holding `pg_advisory_xact_lock` on a fixed key (`Pgbus::Client::PGMQ_INSTALL_LOCK_KEY`), serializing installers across processes — xact-scoped so the lock releases itself at COMMIT/ROLLBACK and stays safe through transaction-pooling poolers, where a session lock's unlock could land on a different server connection; **(3)** a duplicate-object install failure (`PG::UniqueViolation`, `PG::DuplicateSchema`, `PG::DuplicateTable`, `PG::DuplicateObject`, `PG::DuplicateFunction` — a process without the advisory lock, e.g. older pgbus or the extension path, won the race) is rescued by re-checking `pgmq.meta`: present means proceed as installed, absent means the original error is re-raised wrapped in `SchemaNotReady`. Refs #397.
 
diff --git a/lib/pgbus/client.rb b/lib/pgbus/client.rb
index 0c97368..dc78f63 100644
--- a/lib/pgbus/client.rb
+++ b/lib/pgbus/client.rb
@@ -985,13 +985,21 @@ def ensure_pgmq_schema
         # that later rolls back the schema is gone (and even a schema found
         # already-present there may be the caller's own uncommitted work), so
         # a cached true would skip every future check (#399 review).
-        durable = with_raw_connection do |raw_conn|
-          if inside_caller_transaction?(raw_conn)
-            install_pgmq_schema_in_savepoint(raw_conn)
-            false
-          else
-            install_pgmq_schema_in_own_transaction(raw_conn)
-            true
+        #
+        # synchronized (the per-instance connection mutex) nests INSIDE the
+        # class-level install mutex — that lock order is safe because no path
+        # acquires them the other way round — so the shared Proc connection is
+        # never touched while another thread of this instance is mid-operation
+        # on it (single-owner invariant; #399 review).
+        durable = synchronized do
+          with_raw_connection do |raw_conn|
+            if inside_caller_transaction?(raw_conn)
+              install_pgmq_schema_in_savepoint(raw_conn)
+              false
+            else
+              install_pgmq_schema_in_own_transaction(raw_conn)
+              true
+            end
           end
         end
         @schema_ensured = true if durable
@@ -1158,18 +1166,20 @@ def ensure_single_queue(full_name)
 
     def create_queue_physically(full_name)
       synchronized do
-        @pgmq.create(full_name)
-        tune_autovacuum(full_name)
+        create_queue_table(full_name)
         enable_notify_if_needed(full_name, NOTIFY_THROTTLE_MS)
         create_fifo_index_if_needed(full_name)
       end
     end
 
     def create_dead_letter_queue_physically(dlq_name)
-      synchronized do
-        @pgmq.create(dlq_name)
-        tune_autovacuum(dlq_name)
-      end
+      synchronized { create_queue_table(dlq_name) }
+    end
+
+    # Runs inside synchronized — callers own the connection mutex.
+    def create_queue_table(name)
+      @pgmq.create(name)
+      tune_autovacuum(name)
     end
 
     # Queue DDL on the shared Proc-supplied connection joins any transaction
@@ -1180,10 +1190,17 @@ def create_dead_letter_queue_physically(dlq_name)
     # but let the next ensure re-check. Dedicated String/Hash paths run DDL
     # on pgmq-ruby's own pool connections, never inside an application
     # transaction, so they always cache.
+    #
+    # The probe itself must hold the connection mutex: even the local
+    # transaction_status read honors the single-owner invariant on the
+    # shared PG::Connection (#399 review). Sequential with — never nested
+    # inside — the create's own synchronized block.
     def queue_ddl_rides_caller_transaction?
       return false unless @shared_connection
 
-      with_raw_connection { |conn| inside_caller_transaction?(conn) }
+      synchronized do
+        with_raw_connection { |conn| inside_caller_transaction?(conn) }
+      end
     end
 
     def enable_notify_if_needed(full_name, throttle_ms)
diff --git a/spec/pgbus/client_spec.rb b/spec/pgbus/client_spec.rb
index dbb3bb5..8cde7a3 100644
--- a/spec/pgbus/client_spec.rb
+++ b/spec/pgbus/client_spec.rb
@@ -186,6 +186,25 @@ def initialize(*args, **kwargs); end
       end
     end
 
+    it "holds the connection mutex while installing on a shared Proc connection" do
+      require "pg"
+      allow(PGMQ::Client).to receive(:new).and_return(mock_pgmq)
+      allow(config).to receive(:connection_options).and_return(-> { raw_conn })
+      shared_client = described_class.new(config, schema_ensured: false)
+      allow(shared_client).to receive(:tune_autovacuum)
+      allow(shared_client).to receive(:notify_trigger_current?).and_return(false)
+      allow(shared_client).to receive(:with_raw_connection).and_yield(raw_conn)
+      owned_during_check = nil
+      allow(raw_conn).to receive(:exec).with(/pg_tables.*pgmq.*meta/) do
+        owned_during_check = shared_client.instance_variable_get(:@pgmq_mutex).owned?
+        double(ntuples: 1)
+      end
+
+      shared_client.ensure_queue("jobs")
+
+      expect(owned_during_check).to be(true)
+    end
+
     it "owns the transaction when the connection reports an idle status" do
       require "pg"
       allow(raw_conn).to receive(:transaction_status).and_return(PG::PQTRANS_IDLE)
@@ -316,6 +335,18 @@ def settled?(thread)
 
     before { require "pg" }
 
+    it "probes the shared connection only while holding the connection mutex" do
+      owned_during_probe = nil
+      allow(raw_conn).to receive(:transaction_status) do
+        owned_during_probe = client.instance_variable_get(:@pgmq_mutex).owned?
+        PG::PQTRANS_IDLE
+      end
+
+      client.ensure_queue("jobs")
+
+      expect(owned_during_probe).to be(true)
+    end
+
     it "creates the queue but does not cache it — the DDL is only durable once the caller commits" do
       allow(raw_conn).to receive(:transaction_status).and_return(PG::PQTRANS_INTRANS)