fix(streams): transaction probe must not lease an AR connection

lib/pgbus/streams.rb

@@ -386,10 +386,26 @@ def ensure_queue!
       # after_commit, so we have to check #open? explicitly — otherwise
       # every call path would hit the "deferred" branch and we'd lose the
       # msg_id return value.
+      #
+      # The probe must only inspect a connection the calling thread/fiber
+      # ALREADY leases — `connection_pool.active_connection?`, never
+      # `ActiveRecord::Base.connection`. On Rails 7.2+ `.connection` takes a
+      # sticky executor-scoped lease: on a non-executor thread (the
+      # Coalescer's flush thread, app worker threads) it is never released —
+      # one pool connection pinned per thread — and inside
+      # `with_connection` the sticky flag defeats the block-exit release, so
+      # the CALLER's connection leaks when its thread dies. Both variants
+      # exhausted an exactly-sized pool deterministically (Zazu fan-out
+      # incident, 2026-08-05). Semantics are unchanged: a transaction is
+      # per-lease, so a thread holding no connection can have no open
+      # transaction to defer on — the old code's fresh checkout always
+      # answered nil anyway, at the price of the leak.
       def current_open_transaction
         return nil unless defined?(::ActiveRecord::Base)
 
-        connection = ::ActiveRecord::Base.connection
+        connection = ::ActiveRecord::Base.connection_pool.active_connection?
+        return nil unless connection
+
         transaction = connection.current_transaction
         transaction if transaction.open?
       rescue StandardError => e

spec/pgbus/streams/transaction_probe_spec.rb

@@ -0,0 +1,71 @@
+# frozen_string_literal: true
+
+require "rails_helper"
+
+# Regression for the durable-broadcast connection leak (Zazu incident:
+# fan-out spec pool exhaustion, 2026-08-05).
+#
+# `Stream#current_open_transaction` used to probe via
+# `ActiveRecord::Base.connection`, which on Rails 7.2+ takes a STICKY,
+# executor-scoped lease:
+#
+#   - On a thread/fiber with no prior lease (the Coalescer's flush thread,
+#     any non-executor background thread), the lease is never released —
+#     one AR pool connection pinned forever per thread.
+#   - Inside `connection_pool.with_connection`, the sticky flag makes
+#     with_connection SKIP its release at block exit, so the caller's
+#     connection leaks when the thread dies.
+#
+# A freshly-leased connection can never carry the caller's open transaction
+# anyway (transactions are per-lease), so the probe must only inspect an
+# EXISTING lease: `connection_pool.active_connection?` — no checkout.
+RSpec.describe Pgbus::Streams::Stream do
+  subject(:stream) { described_class.new("probe-leak", client: client, durable: true) }
+
+  let(:client) do
+    instance_double(
+      Pgbus::Client,
+      ensure_stream_queue: nil,
+      send_stream_message: 1,
+      stream_current_msg_id: 0,
+      read_after: []
+    )
+  end
+
+  let(:pool) { ActiveRecord::Base.connection_pool }
+
+  it "does not lease an AR connection when the probing thread holds none" do
+    leaked = Thread.new do
+      stream.send(:current_open_transaction)
+      pool.active_connection?
+    end.value
+
+    expect(leaked).to be_falsey
+  end
+
+  it "returns nil (broadcast immediately) when the probing thread holds no connection" do
+    expect(Thread.new { stream.send(:current_open_transaction) }.value).to be_nil
+  end
+
+  it "does not defeat with_connection's release when broadcasting from a worker thread" do
+    Thread.new do
+      pool.with_connection { stream.broadcast("<turbo-stream>x</turbo-stream>") }
+    end.join
+
+    dead_leases = pool.connections.select { |c| c.in_use? && !c.owner.alive? }
+    expect(dead_leases).to be_empty
+  end
+
+  it "still finds the open transaction when the probing thread holds a connection inside one" do
+    probe = Thread.new do
+      pool.with_connection do
+        ActiveRecord::Base.transaction do
+          seen = stream.send(:current_open_transaction)
+          { present: !seen.nil?, open: seen&.open? }
+        end
+      end
+    end.value
+
+    expect(probe).to eq(present: true, open: true)
+  end
+end

spec/pgbus/streams_spec.rb

@@ -192,8 +192,10 @@ def run_callbacks! = @callbacks.each(&:call)
         tx = transaction
         conn = Object.new
         conn.define_singleton_method(:current_transaction) { tx }
+        pool = Object.new
+        pool.define_singleton_method(:active_connection?) { conn }
         ar_base = Class.new
-        ar_base.define_singleton_method(:connection) { conn }
+        ar_base.define_singleton_method(:connection_pool) { pool }
         stub_const("ActiveRecord::Base", ar_base)
       end
 
@@ -394,7 +396,10 @@ def rollback!  = @after_rollback.each(&:call).then { @open = false }
         double("ActiveRecord::ConnectionAdapters::AbstractAdapter", current_transaction: transaction)
       end
 
-      let(:ar_base) { double("ActiveRecord::Base", connection: ar_connection) }
+      # The probe reads connection_pool.active_connection? — the existing
+      # lease, never a fresh checkout (see Stream#current_open_transaction).
+      let(:ar_pool) { double("ActiveRecord::ConnectionAdapters::ConnectionPool", active_connection?: ar_connection) }
+      let(:ar_base) { double("ActiveRecord::Base", connection_pool: ar_pool) }
 
       before { stub_const("ActiveRecord::Base", ar_base) }