From 99b1691f7a527a7556d1a4479ffc22ae2f5087b9 Mon Sep 17 00:00:00 2001 From: mhenrixon Date: Wed, 5 Aug 2026 01:24:56 +0200 Subject: [PATCH] fix(streams): transaction probe must not lease an AR connection MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Stream#current_open_transaction probed via ActiveRecord::Base.connection, which on Rails 7.2+ takes a sticky, executor-scoped lease. Two leaks: - On a non-executor thread (the Coalescer's flush thread, app worker threads) the lease is never released — one AR pool connection pinned per thread for its lifetime. - Inside connection_pool.with_connection the sticky flag defeats the block-exit release, so the CALLER's connection leaks when its thread dies (pool reports it dead-but-in-use). Both variants only fire on the durable path — ephemeral broadcasts early-return before the probe — which is why flipping streams_default_broadcast_mode to :durable surfaced it: a downstream fan-out spec with an exactly-sized pool exhausted deterministically (5 workers + 1 coalescer flush = whole pool dead in one run). Probe connection_pool.active_connection? instead: the existing lease or nil, never a checkout. Semantics unchanged — a transaction is per-lease, so a thread holding no connection has no open transaction to defer on; the old fresh checkout always answered nil anyway, at the price of the leak. --- lib/pgbus/streams.rb | 18 ++++- spec/pgbus/streams/transaction_probe_spec.rb | 71 ++++++++++++++++++++ spec/pgbus/streams_spec.rb | 9 ++- 3 files changed, 95 insertions(+), 3 deletions(-) create mode 100644 spec/pgbus/streams/transaction_probe_spec.rb diff --git a/lib/pgbus/streams.rb b/lib/pgbus/streams.rb index 1edfe144..1e8db7c9 100644 --- a/lib/pgbus/streams.rb +++ b/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 diff --git a/spec/pgbus/streams/transaction_probe_spec.rb b/spec/pgbus/streams/transaction_probe_spec.rb new file mode 100644 index 00000000..a4d558c2 --- /dev/null +++ b/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("x") } + 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 diff --git a/spec/pgbus/streams_spec.rb b/spec/pgbus/streams_spec.rb index 6a1d1845..46bda15a 100644 --- a/spec/pgbus/streams_spec.rb +++ b/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) }