fix: listen on a connection per waiting thread

The supervisor memoizes one adapter and shares it across runtime roles, so every waiter used one cached connection and called wait_for_notify on it concurrently. A notification was consumed by whichever waiter won, leaving the other roles asleep until their polling timeout, and concurrent use of one connection is unsafe besides. Open and track a listening connection per thread, and disconnect all of them on stop. The earlier test hid this by giving each waiter its own adapter, and passed even against the shared connection because it could not distinguish waking on a notification from returning at the timeout; it now asserts on latency.

Author: cardmagic

lib/solid_objects/wake_up_adapters/postgresql.rb

@@ -15,15 +15,15 @@ class Postgresql
 
       # @rbs @channel: String
       # @rbs @mutex: Thread::Mutex
-      # @rbs @connection: untyped
+      # @rbs @connections: Array[untyped]
 
       attr_reader :channel
 
       # @rbs (?channel: String) -> void
       def initialize(channel: CHANNEL)
         @channel = channel
         @mutex = Thread::Mutex.new
-        @connection = nil
+        @connections = []
       end
 
       # @rbs () -> bool
@@ -58,21 +58,19 @@ def listen
 
       # @rbs () -> bool
       def stop
-        mutex.synchronize do
-          connection = @connection
-          @connection = nil
-          return false unless connection
-
-          connection.disconnect!
-          true
+        open = mutex.synchronize do
+          listening = connections.dup
+          connections.clear
+          listening
         end
-      rescue
-        false
+        Thread.current[thread_key] = nil
+        open.each { |connection| disconnect(connection) }
+        open.any?
       end
 
       private
 
-      attr_reader :mutex
+      attr_reader :mutex, :connections
 
       # @rbs () -> void
       def notify_channel
@@ -81,17 +79,38 @@ def notify_channel
         end
       end
 
-      # A listening connection is dedicated: `LISTEN` is per connection, and a
-      # blocking wait must not hold a connection the rest of the runtime needs.
+      # A listening connection is dedicated and per thread. `LISTEN` is per
+      # connection, a blocking wait must not hold a connection the rest of the
+      # runtime needs, and one connection cannot serve concurrent waiters: the
+      # supervisor shares one adapter across roles, and a notification consumed
+      # by one waiter would leave the others asleep until their poll expired.
       # @rbs () -> untyped
       def listening_connection
-        mutex.synchronize do
-          return @connection if @connection&.active?
+        connection = Thread.current[thread_key]
+        return connection if connection&.active?
 
-          @connection = Record.connection_pool.send(:new_connection)
-          @connection.execute("LISTEN #{@connection.quote_table_name(channel)}")
-          @connection
-        end
+        open_listening_connection
+      end
+
+      # @rbs () -> untyped
+      def open_listening_connection
+        connection = Record.connection_pool.send(:new_connection)
+        connection.execute("LISTEN #{connection.quote_table_name(channel)}")
+        Thread.current[thread_key] = connection
+        mutex.synchronize { connections << connection }
+        connection
+      end
+
+      # @rbs () -> Symbol
+      def thread_key
+        :"solid_objects_wake_up_#{object_id}"
+      end
+
+      # @rbs (untyped) -> void
+      def disconnect(connection)
+        connection.disconnect!
+      rescue
+        nil
       end
 
       # @rbs (Numeric) -> void

sig/generated/lib/solid_objects/wake_up_adapters/postgresql.rbs

@@ -14,7 +14,7 @@ module SolidObjects
 
       FAILED_WAIT_INTERVAL: ::Float
 
-      @connection: untyped
+      @connections: Array[untyped]
 
       @mutex: Thread::Mutex
 
@@ -43,14 +43,28 @@ module SolidObjects
 
       attr_reader mutex: untyped
 
+      attr_reader connections: untyped
+
       # @rbs () -> void
       def notify_channel: () -> void
 
-      # A listening connection is dedicated: `LISTEN` is per connection, and a
-      # blocking wait must not hold a connection the rest of the runtime needs.
+      # A listening connection is dedicated and per thread. `LISTEN` is per
+      # connection, a blocking wait must not hold a connection the rest of the
+      # runtime needs, and one connection cannot serve concurrent waiters: the
+      # supervisor shares one adapter across roles, and a notification consumed
+      # by one waiter would leave the others asleep until their poll expired.
       # @rbs () -> untyped
       def listening_connection: () -> untyped
 
+      # @rbs () -> untyped
+      def open_listening_connection: () -> untyped
+
+      # @rbs () -> Symbol
+      def thread_key: () -> Symbol
+
+      # @rbs (untyped) -> void
+      def disconnect: (untyped) -> void
+
       # @rbs (Numeric) -> void
       def pace_after_failure: (Numeric) -> void
 

test/integration/postgresql_wake_up_test.rb

@@ -43,28 +43,43 @@ class PostgresqlWakeUpTest < ActiveSupport::TestCase
     assert_operator elapsed, :<, 2.0
   end
 
-  test "every waiter wakes on one signal" do
+  # The supervisor memoizes one adapter and shares it across runtime roles, so
+  # concurrent waiters go through a single adapter instance.
+  test "every waiter on one shared adapter wakes on one signal" do
     waiters = 3
     waiting = Queue.new
     woken = Queue.new
-    adapters = Array.new(waiters) { SolidObjects::WakeUpAdapters::Postgresql.new }
-    threads = adapters.map do |adapter|
+    threads = Array.new(waiters) do
       Thread.new do
-        adapter.listen
+        @adapter.listen
         waiting << true
-        adapter.wait(timeout: 5)
-        woken << true
+        started = monotonic_now
+        @adapter.wait(timeout: 5)
+        woken << monotonic_now - started
       end
     end
     waiters.times { Timeout.timeout(5) { waiting.pop } }
 
     signal_from_another_connection
 
-    waiters.times { Timeout.timeout(5) { woken.pop } }
+    elapsed = waiters.times.map { Timeout.timeout(10) { woken.pop } }
+    threads.each { |thread| thread.join(6) }
+    # Waking on the notification, not falling through to the 5 second timeout.
+    assert_operator elapsed.max, :<, 1.0,
+      "every waiter should wake on the notification, not time out"
+  end
+
+  test "each waiting thread listens on its own connection" do
+    connections = Queue.new
+    threads = Array.new(2) do
+      Thread.new do
+        @adapter.listen
+        connections << Thread.current[:"solid_objects_wake_up_#{@adapter.object_id}"].object_id
+      end
+    end
     threads.each { |thread| thread.join(5) }
-    assert_equal 0, woken.size
-  ensure
-    adapters&.each(&:stop)
+
+    assert_equal 2, connections.size.times.map { connections.pop }.uniq.length
   end
 
   test "signalling never raises into the caller" do