fix(client): serialize pgmq schema install across clients and processes

CHANGELOG.md

@@ -2,6 +2,8 @@
 
 ### Fixed
 
+- **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.
 
 - **Ephemeral broadcasts over the PG NOTIFY payload cap no longer fail — loudly on the sync path, silently in the coalescer — they auto-degrade to a durable publish (issue #391).** Ephemeral frames ride the NOTIFY payload itself, which PostgreSQL caps below 8000 bytes. Any rendered-component broadcast (a progress card with Tailwind classes easily exceeds it) previously raised `PGMQ::Errors::ConnectionError: … payload string too long` — an error class that sent diagnosis toward the connection, not the payload — and on the `coalesce:` path that raise happened inside the coalescer's flush thread, reaching no caller, no ErrorReporter, no log: small frames delivered, big frames vanished, and the operator saw "SSE works but updates don't arrive". Three changes: **(1)** `Stream#broadcast` now measures the wrapped JSON before the NOTIFY and publishes an over-budget frame durably instead (payload stored in PGMQ, the queue's insert trigger fires the NOTIFY as a bare wake on the same channel the subscriber already LISTENs on) — delivery semantics preserved on both the sync and coalesced paths, warn-logged and instrumented (`pgbus.stream.broadcast` with `ephemeral_fallback: true`). **(2)** Direct `Client#notify_stream` callers get publish-time validation: a typed `Pgbus::Streams::PayloadTooLarge` raised at the call site for payloads exceeding `Pgbus::Client::NotifyStream::NOTIFY_PAYLOAD_LIMIT_BYTES` (7999 bytes, the largest accepted payload), naming the stream, the byte count, and the durable-mode escape hatch. **(3)** The coalescer's flush thread routes every flush error through `ErrorReporter` (same report-don't-log reasoning as #352) — a background thread swallowing delivery failures is invisible to APM by construction. ⚠️ **Upgrade note for 0.13 installs:** `streams_default_broadcast_mode` defaults to `:ephemeral`, and that default is a **behavior change** for apps broadcasting rendered components (what `broadcast_render`-style usage produces) — before this fix, any frame over ~8KB was silently lost or misdiagnosed. Durable is the right mode for turbo-stream UI regardless (since-id replay needs the archive): pin `config.streams_default_broadcast_mode = :durable`, or use `streams_durable_patterns` for the streams that need it; the auto-fallback now covers whatever stays ephemeral. Refs #391.

lib/pgbus/client.rb

@@ -20,6 +20,39 @@ class Client
     PGMQ_REQUIRE_MUTEX = Mutex.new
     private_constant :PGMQ_REQUIRE_MUTEX
 
+    # Fixed advisory-lock key serializing pgmq schema installation across
+    # processes (issue #397). "pgmqinst" in ASCII hex — arbitrary but stable;
+    # it only has to be identical in every process that can install.
+    PGMQ_INSTALL_LOCK_KEY = 0x70676D71_696E7374
+
+    PGMQ_META_CHECK_SQL = "SELECT 1 FROM pg_tables WHERE schemaname = 'pgmq' AND tablename = 'meta' LIMIT 1"
+    private_constant :PGMQ_META_CHECK_SQL
+
+    # 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::…)
+    # guards used elsewhere in this file).
+    DUPLICATE_INSTALL_ERROR_CLASSES = %w[
+      PG::UniqueViolation
+      PG::DuplicateSchema
+      PG::DuplicateTable
+      PG::DuplicateObject
+      PG::DuplicateFunction
+    ].freeze
+    private_constant :DUPLICATE_INSTALL_ERROR_CLASSES
+
+    # Process-wide, not per-instance: on the shared-AR Proc path two Client
+    # instances share one underlying libpq connection while each holding their
+    # own @pgmq_mutex, so a per-instance guard cannot serialize bootstrap DDL —
+    # concurrent install traffic desyncs the protocol ("message type 0x…
+    # arrived from server while idle") and wedges a thread on a socket read
+    # (issue #397, forensics in getzazu/app#3413).
+    @pgmq_install_mutex = Mutex.new
+
+    class << self
+      attr_reader :pgmq_install_mutex
+    end
+
     # Throttle window for PGMQ's enable_notify_insert trigger. Postgres
     # NOTIFYs are coalesced into one wake-up per window, so a value of 250ms
     # means: at most 4 broadcasts/sec per queue, regardless of insert rate.
@@ -245,10 +278,7 @@ def physical_queue_names(logical_name)
     # present, no tracking row) apart from "PGMQ not installed at all".
     def pgmq_installed?
       with_raw_connection do |conn|
-        result = conn.exec(
-          "SELECT 1 FROM pg_tables WHERE schemaname = 'pgmq' AND tablename = 'meta' LIMIT 1"
-        )
-        result.ntuples.positive?
+        conn.exec(PGMQ_META_CHECK_SQL).ntuples.positive?
       end
     end
 
@@ -943,13 +973,10 @@ def collect_configured_queues
     def ensure_pgmq_schema
       return if @schema_ensured
 
-      synchronized do
+      self.class.pgmq_install_mutex.synchronize do
         return if @schema_ensured
 
-        with_raw_connection do |raw_conn|
-          exists = raw_conn.exec("SELECT 1 FROM pg_tables WHERE schemaname = 'pgmq' AND tablename = 'meta' LIMIT 1")
-          install_pgmq_schema(raw_conn) if exists.ntuples.zero?
-        end
+        with_raw_connection { |raw_conn| install_pgmq_schema_serialized(raw_conn) }
         @schema_ensured = true
       end
     rescue StandardError => e
@@ -958,6 +985,36 @@ 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).
+    def install_pgmq_schema_serialized(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
+      begin
+        conn.exec("ROLLBACK")
+      rescue StandardError
+        # A connection broken enough to refuse ROLLBACK also fails the
+        # re-check below, which surfaces the state honestly; re-raising the
+        # ROLLBACK error here would mask the original install failure.
+      end
+      raise e unless duplicate_install_error?(e)
+
+      # 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?
+    end
+
+    def duplicate_install_error?(error)
+      DUPLICATE_INSTALL_ERROR_CLASSES.include?(error.class.name)
+    end
+
     def install_pgmq_schema(conn)
       mode = config.pgmq_schema_mode
 

spec/pgbus/client_spec.rb

@@ -51,6 +51,13 @@ def initialize(*args, **kwargs); end
 
     before do
       allow(client).to receive(:with_raw_connection).and_yield(raw_conn)
+      # Transaction + advisory-lock framing around check+install (issue #397).
+      # Allowed here so every example in this describe tolerates the framing;
+      # the framing-specific examples assert on these explicitly.
+      allow(raw_conn).to receive(:exec).with("BEGIN")
+      allow(raw_conn).to receive(:exec).with(/pg_advisory_xact_lock/)
+      allow(raw_conn).to receive(:exec).with("COMMIT")
+      allow(raw_conn).to receive(:exec).with("ROLLBACK")
     end
 
     it "installs via embedded SQL when pgmq.meta missing and no extension (auto mode)" do
@@ -112,6 +119,107 @@ def initialize(*args, **kwargs); end
 
       expect(raw_conn).to have_received(:exec).with(/pg_tables.*pgmq.*meta/).once
     end
+
+    it "wraps check+install in a transaction holding the install advisory lock (#397)" 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("BEGIN").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(/pg_tables.*pgmq.*meta/).ordered
+      expect(raw_conn).to have_received(:exec).with(Pgbus::PgmqSchema.install_sql).ordered
+      expect(raw_conn).to have_received(:exec).with("COMMIT").ordered
+    end
+
+    context "when another process wins the install race (#397)" do
+      before { require "pg" }
+
+      it "treats a duplicate-object install failure as installed by the winner" 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 violates unique constraint "pg_namespace_nspname_index"')
+        )
+
+        expect { client.ensure_queue("jobs") }.not_to raise_error
+
+        expect(raw_conn).to have_received(:exec).with("ROLLBACK")
+        expect(raw_conn).to have_received(:exec).with(/pg_tables.*pgmq.*meta/).twice
+      end
+
+      it "still wraps as SchemaNotReady when the re-check finds no schema" do
+        check = double("check_result")
+        allow(check).to receive(:ntuples).and_return(0, 0)
+        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") }.to raise_error(
+          Pgbus::SchemaNotReady, /PGMQ schema installation failed/
+        )
+      end
+    end
+
+    # Helpers for the process-wide serialization example: connection doubles
+    # pre-stubbed with the transaction/advisory-lock framing.
+    def framed_conn(name)
+      conn = double(name)
+      allow(conn).to receive(:exec).with("BEGIN")
+      allow(conn).to receive(:exec).with(/pg_advisory_xact_lock/)
+      allow(conn).to receive(:exec).with("COMMIT")
+      conn
+    end
+
+    # Blocks inside the install until `release` is signalled, reporting entry
+    # on `started` — lets the example hold client A mid-install deterministically.
+    def install_blocking_conn(started, release)
+      conn = framed_conn("conn_a")
+      allow(conn).to receive(:exec).with(/pg_tables.*pgmq.*meta/).and_return(double(ntuples: 0))
+      allow(conn).to receive(:exec).with(/pg_available_extensions/).and_return(double(ntuples: 0))
+      allow(conn).to receive(:exec).with(Pgbus::PgmqSchema.install_sql) do
+        started << true
+        release.pop
+      end
+      conn
+    end
+
+    def unensured_client(conn)
+      c = described_class.new(config, schema_ensured: false)
+      allow(c).to receive(:tune_autovacuum)
+      allow(c).to receive(:notify_trigger_current?).and_return(false)
+      allow(c).to receive(:with_raw_connection).and_yield(conn)
+      c
+    end
+
+    it "serializes installs process-wide across client instances (#397)" do
+      install_started = Queue.new
+      release_install = Queue.new
+      conn_a = install_blocking_conn(install_started, release_install)
+      conn_b = framed_conn("conn_b")
+      allow(conn_b).to receive(:exec).with(/pg_tables.*pgmq.*meta/).and_return(double(ntuples: 1))
+      client_a = client
+      allow(client_a).to receive(:with_raw_connection).and_yield(conn_a)
+      client_b = unensured_client(conn_b)
+
+      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
+      expect(conn_b).not_to have_received(:exec)
+
+      release_install << true
+      [thread_a, thread_b].each { |t| t.join(5) }
+      expect(conn_b).to have_received(:exec).with(/pg_tables.*pgmq.*meta/).once
+      expect(conn_b).not_to have_received(:exec).with(Pgbus::PgmqSchema.install_sql)
+    end
   end
 
   describe "#ensure_queue" do