fix: prove the wake-up choice before claiming it

Two claims in selection were not supported by what it measured.

`configured` overwrote the capability that an adapter reports about
itself. A configured `SolidObjects::WakeUp` signals in one process only
and says so, but selection relabelled it `crosses_processes: true`, so
the doctor reported PASS and the process registry dropped the warning
that this exact topology needs. Selection now keeps the capability that
an adapter provides and labels only an adapter that provides none.

The PostgreSQL probe set `application_name` and read it back on the next
statement. A transaction pooler can hand the same backend to two
consecutive autocommit statements, so the read-back succeeds while
`LISTEN` still has no session affinity, and selection memoised
notification support that never fires. The probe now listens, sends one
`NOTIFY` from a second connection, and waits up to two seconds for it to
arrive. Nothing but a delivered notification selects notifications, and
a probe that cannot run falls back to polling rather than assume.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: cardmagic

CHANGELOG.md

@@ -12,10 +12,14 @@
   cross-process wake-up, a connection per waiting thread outside the pool, and
   one `NOTIFY` per enqueue after the commit. Set
   `config.wake_up_adapter = :in_process` to keep polling.
-- Probe the PostgreSQL session before selecting notifications, because `LISTEN`
-  does not survive a transaction pooler such as PgBouncer. A session that does
-  not outlive a statement falls back to polling and warns once. A probe that
-  cannot run is not treated as a pooler.
+- Prove the PostgreSQL notification path before selecting it, because `LISTEN`
+  does not survive a transaction pooler such as PgBouncer. Selection listens,
+  sends one `NOTIFY` from a second connection, and waits up to two seconds for
+  it to arrive. A probe that does not deliver falls back to polling and warns
+  once.
+- Keep the capability that a configured adapter reports about itself. A
+  configured `SolidObjects::WakeUp` now reports `:in_process` and warns, rather
+  than claim that it crosses processes.
 - Report the resolved choice. `SolidObjects.wake_up.capability` names the
   adapter, whether it crosses processes, its measured floor, and why it was
   chosen. The doctor reports it, and the polling-only warning now fires on what

docs/roadmap.md

@@ -131,7 +131,9 @@
   adapter opens a connection per waiting thread outside the pool and adds one
   `NOTIFY` per enqueue, and Redis is not a dependency of this gem.
   `LISTEN` does not survive a transaction-pooling proxy such as PgBouncer, so
-  the session is probed and a pooled one falls back to polling and warns once.
+  selection listens, sends one `NOTIFY` from a second connection, and waits for
+  it to arrive. A probe that does not deliver falls back to polling and warns
+  once.
   `SolidObjects.wake_up.capability` reports the adapter, whether it crosses
   processes, its floor, and why, and the doctor shows the same record.
   MySQL still polls. It has no notification channel, and no MySQL notifier has

lib/solid_objects/wake_up_adapters.rb

@@ -5,6 +5,7 @@ module WakeUpAdapters
     POSTGRESQL_FLOOR_MS = 2.9
     REDIS_FLOOR_MS = 5.7
     REDIS_URL_VARIABLE = "SOLID_OBJECTS_REDIS_URL"
+    PROBE_TIMEOUT_SECONDS = 2.0
 
     @pooled_warning_mutex = Thread::Mutex.new
     @pooled_warning_emitted = false
@@ -40,6 +41,7 @@ def named(name)
     # @rbs (untyped) -> untyped
     def configured(adapter)
       return adapter unless adapter.respond_to?(:capability=)
+      return adapter if adapter.respond_to?(:default_capability)
 
       labelled(adapter, :configured, true, nil, "an adapter was configured, so selection did not run")
     end
@@ -61,21 +63,37 @@ def select(connection = Record.connection)
       return redis_selection(url) if url
 
       family = DatabaseAdapter.family(connection)
-      return postgresql_selection(connection) if family == :postgresql
+      return postgresql_selection if family == :postgresql
 
       polling_selection(family)
     end
 
-    # @rbs (untyped) -> bool?
-    def session_survives_transactions?(connection)
-      previous = connection.select_value("SELECT current_setting('application_name')")
-      token = SecureRandom.hex(8)
-      connection.execute("SET application_name = #{connection.quote(token)}")
-      connection.select_value("SELECT current_setting('application_name')") == token
+    # @rbs (untyped) -> bool
+    def notifications_deliver?(adapter)
+      return false unless adapter.listen
+      return false unless notify_probe_channel(adapter.channel)
+
+      adapter.wait(timeout: PROBE_TIMEOUT_SECONDS)
     rescue
-      nil
+      false
+    ensure
+      adapter.stop
+    end
+
+    # @rbs (String) -> bool
+    def notify_probe_channel(channel)
+      connection = Record.connection_pool.send(:new_connection)
+      connection.execute("NOTIFY #{connection.quote_table_name(channel)}")
+      true
     ensure
-      restore_application_name(connection, previous)
+      disconnect_probe(connection)
+    end
+
+    # @rbs (untyped) -> void
+    def disconnect_probe(connection)
+      connection&.disconnect!
+    rescue
+      nil
     end
 
     # @rbs () -> void
@@ -97,24 +115,23 @@ def redis_selection(url)
       )
     end
 
-    # @rbs (untyped) -> untyped
-    def postgresql_selection(connection)
-      survives = session_survives_transactions?(connection)
-      return pooled_selection if survives == false
+    # @rbs () -> untyped
+    def postgresql_selection
+      adapter = Postgresql.new
+      return pooled_selection unless notifications_deliver?(adapter)
 
       labelled(
-        Postgresql.new, :postgresql_notify, true, POSTGRESQL_FLOOR_MS,
-        survives ? "PostgreSQL LISTEN is available and the session outlives a transaction"
-                 : "PostgreSQL LISTEN was selected without a session probe"
+        adapter, :postgresql_notify, true, POSTGRESQL_FLOOR_MS,
+        "a probe notification arrived, so PostgreSQL LISTEN carries the signal between processes"
       )
     end
 
     # @rbs () -> untyped
     def pooled_selection
       warn_pooled_session_once
       polling_adapter(
-        "the PostgreSQL session does not outlive a transaction, which a transaction " \
-        "pooler such as PgBouncer causes, so LISTEN would never fire"
+        "a probe notification did not arrive, so LISTEN cannot carry the signal " \
+        "between processes; a transaction pooler such as PgBouncer is the usual cause"
       )
     end
 
@@ -141,20 +158,11 @@ def warn_pooled_session_once
 
         SolidObjects.configuration.logger.warn(
           event: "solid_objects.wake_up.pooled_session",
-          reason: "PostgreSQL notifications were not selected because the session " \
-            "does not outlive a transaction"
+          reason: "PostgreSQL notifications were not selected because a probe " \
+            "notification did not arrive"
         )
         @pooled_warning_emitted = true
       end
     end
-
-    # @rbs (untyped, untyped) -> void
-    def restore_application_name(connection, previous)
-      return if previous.nil?
-
-      connection.execute("SET application_name = #{connection.quote(previous)}")
-    rescue
-      nil
-    end
   end
 end

sig/generated/lib/solid_objects/wake_up_adapters.rbs

@@ -8,6 +8,8 @@ module SolidObjects
 
     REDIS_URL_VARIABLE: ::String
 
+    PROBE_TIMEOUT_SECONDS: ::Float
+
     NAMES: untyped
 
     # @rbs (?untyped) -> untyped
@@ -28,8 +30,14 @@ module SolidObjects
     # @rbs (?untyped) -> untyped
     def self?.select: (?untyped) -> untyped
 
-    # @rbs (untyped) -> bool?
-    def self?.session_survives_transactions?: (untyped) -> bool?
+    # @rbs (untyped) -> bool
+    def self?.notifications_deliver?: (untyped) -> bool
+
+    # @rbs (String) -> bool
+    def self?.notify_probe_channel: (String) -> bool
+
+    # @rbs (untyped) -> void
+    def self?.disconnect_probe: (untyped) -> void
 
     # @rbs () -> void
     def self?.reset_pooled_warning!: () -> void
@@ -40,8 +48,8 @@ module SolidObjects
     # @rbs (String) -> untyped
     def self?.redis_selection: (String) -> untyped
 
-    # @rbs (untyped) -> untyped
-    def self?.postgresql_selection: (untyped) -> untyped
+    # @rbs () -> untyped
+    def self?.postgresql_selection: () -> untyped
 
     # @rbs () -> untyped
     def self?.pooled_selection: () -> untyped
@@ -54,8 +62,5 @@ module SolidObjects
 
     # @rbs () -> void
     def self?.warn_pooled_session_once: () -> void
-
-    # @rbs (untyped, untyped) -> void
-    def self?.restore_application_name: (untyped, untyped) -> void
   end
 end

test/integration/wake_up_selection_test.rb

@@ -4,14 +4,26 @@
 require "solid_objects/doctor"
 
 class WakeUpSelectionTest < ActiveSupport::TestCase
+  class CustomAdapter
+    attr_accessor :capability
+
+    def signal = true
+
+    def watch = self
+
+    def wait(timeout:) = false
+  end
+
   setup do
     SolidObjects.reset_wake_up!
     @redis_url = ENV.delete("SOLID_OBJECTS_REDIS_URL")
+    @configured_adapter = SolidObjects.configuration.wake_up_adapter
   end
 
   teardown do
     ENV.delete("SOLID_OBJECTS_REDIS_URL")
     ENV["SOLID_OBJECTS_REDIS_URL"] = @redis_url if @redis_url
+    SolidObjects.configuration.wake_up_adapter = @configured_adapter
     SolidObjects.reset_wake_up!
   end
 
@@ -20,7 +32,32 @@ class WakeUpSelectionTest < ActiveSupport::TestCase
     SolidObjects.configuration.wake_up_adapter = explicit
 
     assert_same explicit, SolidObjects.wake_up
-    assert_equal :configured, SolidObjects.wake_up.capability.adapter
+  end
+
+  test "a configured adapter keeps the capability it reports about itself" do
+    SolidObjects.configuration.wake_up_adapter = SolidObjects::WakeUp.new
+
+    capability = SolidObjects.wake_up.capability
+    assert_equal :in_process, capability.adapter
+    assert_not capability.crosses_processes
+    assert_match(/another process/i, capability.reason)
+  end
+
+  test "the doctor warns about a configured in-process adapter" do
+    SolidObjects.configuration.authorize_administration = ->(**) { true }
+    SolidObjects.configuration.wake_up_adapter = SolidObjects::WakeUp.new
+
+    check = SolidObjects::Doctor.new.call.check(:wake_up)
+    assert_equal :warn, check.status
+    assert_match(/cannot wake another/i, check.message)
+  end
+
+  test "a configured adapter that reports no capability is recorded as configured" do
+    SolidObjects.configuration.wake_up_adapter = CustomAdapter.new
+
+    capability = SolidObjects.wake_up.capability
+    assert_equal :configured, capability.adapter
+    assert capability.crosses_processes
   end
 
   test "in_process opts out of selection" do
@@ -56,13 +93,26 @@ class WakeUpSelectionTest < ActiveSupport::TestCase
     assert_match(/redis/i, capability.reason)
   end
 
-  test "postgresql selects notifications when the session survives transactions" do
+  test "postgresql selects notifications when a probe notification arrives" do
     skip unless database_family == :postgresql
 
     capability = SolidObjects.wake_up.capability
     assert_equal :postgresql_notify, capability.adapter
     assert capability.crosses_processes
     assert_operator capability.measured_floor_ms, :<, 100
+    assert_match(/probe notification arrived/i, capability.reason)
+  end
+
+  test "postgresql polls when a probe notification does not arrive" do
+    skip unless database_family == :postgresql
+
+    with_undelivered_notifications do
+      capability = SolidObjects.wake_up.capability
+
+      assert_equal :polling, capability.adapter
+      assert_not capability.crosses_processes
+      assert_match(/pooler/i, capability.reason)
+    end
   end
 
   test "a database without a channel polls and reports its floor" do
@@ -134,7 +184,16 @@ def with_module_method(name, replacement)
   end
 
   def with_pooled_session(&block)
-    with_module_method(:session_survives_transactions?, ->(_connection) { false }, &block)
+    with_undelivered_notifications(&block)
+  end
+
+  def with_undelivered_notifications
+    adapter_class = SolidObjects::WakeUpAdapters::Postgresql
+    original = adapter_class.instance_method(:wait)
+    adapter_class.define_method(:wait) { |timeout:| false }
+    yield
+  ensure
+    adapter_class.define_method(:wait, original)
   end
 
   def with_unreachable_database(&block)

test/unit/wake_up_adapters_test.rb

@@ -5,10 +5,21 @@
 class WakeUpAdaptersTest < ActiveSupport::TestCase
   Connection = Struct.new(:adapter_name)
 
-  test "selects PostgreSQL notifications for a PostgreSQL connection" do
-    adapter = SolidObjects::WakeUpAdapters.for(Connection.new("PostgreSQL"))
+  test "selects PostgreSQL notifications when a probe notification arrives" do
+    with_delivered_notifications do
+      adapter = SolidObjects::WakeUpAdapters.for(Connection.new("PostgreSQL"))
 
-    assert_instance_of SolidObjects::WakeUpAdapters::Postgresql, adapter
+      assert_instance_of SolidObjects::WakeUpAdapters::Postgresql, adapter
+    end
+  end
+
+  test "polls when a PostgreSQL probe notification does not arrive" do
+    with_undelivered_notifications do
+      adapter = SolidObjects::WakeUpAdapters.for(Connection.new("PostgreSQL"))
+
+      assert_instance_of SolidObjects::WakeUp, adapter
+      assert_equal :polling, adapter.capability.adapter
+    end
   end
 
   test "falls back to the in-process wake-up for MySQL" do
@@ -44,4 +55,23 @@ class WakeUpAdaptersTest < ActiveSupport::TestCase
 
     assert_equal true, watch.wait(timeout: 1.0)
   end
+
+  private
+
+  def with_delivered_notifications(&block)
+    with_probe(->(_adapter) { true }, &block)
+  end
+
+  def with_undelivered_notifications(&block)
+    with_probe(->(_adapter) { false }, &block)
+  end
+
+  def with_probe(replacement)
+    adapters = SolidObjects::WakeUpAdapters
+    original = adapters.method(:notifications_deliver?)
+    adapters.define_singleton_method(:notifications_deliver?, replacement)
+    yield
+  ensure
+    adapters.define_singleton_method(:notifications_deliver?, original)
+  end
 end