feat: select a wake-up adapter automatically

A commit in a web process did not wake a worker process unless somebody
configured an adapter, so delivery waited out the polling interval. The
pieces existed and were measured. Nothing chose between them.

config.wake_up_adapter now takes a name or an adapter, the way
config.cache_store and config.active_job.queue_adapter do, rather than
one setting for a mode and another for an instance. It defaults to
:automatic, which prefers a configured Redis URL, then PostgreSQL
notifications, then polling. :in_process opts out. An unknown name
raises rather than quietly polling, because a typo that silently costs a
second of latency is the failure this is meant to remove.

LISTEN does not survive a transaction pooler, so the PostgreSQL session
is probed first. The probe reports three outcomes rather than two. A
definite false means a pooler took the session, and polling is chosen
with a warning. A probe that could not run at all is not evidence of a
pooler, so notifications are still chosen and the reason says the
session was not probed. Conflating those two would downgrade any
deployment whose connection cannot answer the probe.

The choice is now readable. SolidObjects.wake_up.capability names the
adapter, whether it crosses processes, its measured floor, and why. The
doctor reports it. The polling-only warning now fires on what was
installed rather than on whether a setting was set, which is what it
meant to ask: the old guard returned early for any configured adapter,
and :automatic is always configured.

Two tests moved to :in_process rather than changing what they assert.
The polling warning test is about the warning, not about selection, and
the enqueue statement count is about the enqueue transaction, not about
the NOTIFY that a cross-process adapter adds after the commit.

Author: cardmagic

CHANGELOG.md

@@ -2,6 +2,24 @@
 
 ## Unreleased
 
+- Select a wake-up adapter automatically. `config.wake_up_adapter` now takes a
+  name or an adapter, as `config.cache_store` and
+  `config.active_job.queue_adapter` do, and defaults to `:automatic`. Selection
+  prefers a configured Redis URL, then PostgreSQL notifications, then polling.
+  `:in_process` opts out, and an unknown name raises rather than quietly
+  polling.
+- PostgreSQL deployments that configure nothing now use notifications. They gain
+  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.
+- 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
+  was installed rather than on whether a setting was set.
 - Read the durable row rather than the query cache in `MessageReference#status`,
   `MessageReference#result`, and an actor snapshot. A caller that polls holds one
   query cache for the whole poll, and the worker that finishes the message is

docs/roadmap.md

@@ -122,21 +122,20 @@
 
 - Wake-up strategy: in-process signaling, durable polling, injection, and
   cross-process adapters for PostgreSQL and Redis are implemented and tested.
-  What is not done is making any of them automatic. In-process signaling cannot
-  cross process boundaries, so by default a commit in a web process does not
-  wake a broadcast executor in a worker process and that delivery waits up to
-  the current adaptive polling interval, up to the one-second
-  `idle_polling_interval` default. The runtime warns once when it observes this
-  topology without an adapter. An adapter removes that floor, measured before
-  adaptive polling at 103.7 ms to 2.9 ms at p50 on PostgreSQL and 103.8 ms to
-  5.7 ms on Redis, but each stays opt-in for a reason: the PostgreSQL adapter
-  opens a connection per waiting thread outside the pool and `LISTEN` does not
-  survive a transaction-pooling proxy such as PgBouncer, and Redis is not a
-  dependency of this gem.
-  `WakeUpAdapters.for` selects notifications on PostgreSQL and the in-process
-  default elsewhere; it never selects Redis. An application that configures
-  nothing keeps polling, and MySQL applications keep polling unless they
-  configure Redis explicitly.
+  Selection is automatic. `config.wake_up_adapter` defaults to `:automatic` and
+  prefers a configured Redis URL, then PostgreSQL notifications, then polling,
+  so an application that configures nothing no longer polls on PostgreSQL. An
+  adapter removes the one-second floor, measured before adaptive polling at
+  103.7 ms to 2.9 ms at p50 on PostgreSQL and 103.8 ms to 5.7 ms on Redis. Each
+  carries a cost that selection now states rather than hides: the PostgreSQL
+  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.
+  `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
+  been measured against polling on the same hardware, so none is shipped.
 - Realtime: scalar and dependency-driven keyed ERB component replacement or
   morphing, personalized refresh authorization, revision fencing, coalescing,
   reconnect convergence, batched refreshes, and personalized state payloads are

lib/solid_objects.rb

@@ -49,6 +49,7 @@
 require "solid_objects/actor_channel"
 require "solid_objects/action_cable_broadcast_adapter"
 require "solid_objects/database_adapter"
+require "solid_objects/wake_up_capability"
 require "solid_objects/wake_up"
 require "solid_objects/wake_up_adapters/postgresql"
 require "solid_objects/wake_up_adapters/redis"
@@ -198,9 +199,36 @@ def database_adapter
       @database_adapter ||= DatabaseAdapter.for(SolidObjects::Record.connection)
     end
 
-    # @rbs () -> WakeUp
+    # @rbs () -> untyped
     def wake_up
-      @wake_up ||= configuration.wake_up_adapter || WakeUp.new
+      @wake_up ||= resolve_wake_up
+    end
+
+    # @rbs () -> void
+    def reset_wake_up!
+      @wake_up = nil
+      WakeUpAdapters.reset_pooled_warning!
+    end
+
+    # @rbs () -> untyped
+    def resolve_wake_up
+      WakeUpAdapters.build(configuration.wake_up_adapter)
+    rescue ArgumentError
+      raise
+    rescue => error
+      unreachable_wake_up(error)
+    end
+
+    # @rbs (Exception) -> untyped
+    def unreachable_wake_up(error)
+      adapter = WakeUp.new
+      adapter.capability = WakeUpCapability.new(
+        adapter: :in_process,
+        crosses_processes: false,
+        measured_floor_ms: nil,
+        reason: "the database could not be reached to select an adapter: #{error.class}"
+      )
+      adapter
     end
   end
 end

lib/solid_objects/configuration.rb

@@ -141,7 +141,7 @@ def initialize
       @connects_to = nil
       @stream_signing_secret = nil
       @broadcast_adapter = nil
-      @wake_up_adapter = nil
+      @wake_up_adapter = :automatic
       @component_path_resolver = nil
       @component_authorization_context = ->(controller:) { controller }
       @payload_authorization_context = ->(connection:) { connection }

lib/solid_objects/doctor.rb

@@ -113,6 +113,7 @@ def call
         schema_check,
         check_authorization,
         check_database_server,
+        check_wake_up,
         schema_check.failed? ? skipped_runtime : check_runtime,
         ready_for_round_trip?(configuration_check, schema_check) ?
           check_sync_round_trip :
@@ -208,6 +209,19 @@ def check_database_server
       warn_check(:database_server, "#{error.class}: #{error.message}")
     end
 
+    # @rbs () -> Check
+    def check_wake_up
+      capability = SolidObjects.wake_up.capability
+      floor = capability.measured_floor_ms
+      summary = "#{capability.adapter}: #{capability.reason}"
+      summary += ", floor #{floor} ms" if floor
+      return pass(:wake_up, summary) if capability.crosses_processes
+
+      warn_check(:wake_up, "#{summary}; a commit in one process cannot wake another")
+    rescue => error
+      warn_check(:wake_up, "#{error.class}: #{error.message}")
+    end
+
     # @rbs () -> Check
     def check_runtime
       cutoff = SolidObjects.database_adapter.database_now -

lib/solid_objects/process_registry.rb

@@ -61,7 +61,9 @@ def deregister(process_record, now: SolidObjects.database_adapter.database_now)
 
       # @rbs () -> void
       def warn_if_polling_is_only_cross_process_wake_up
-        return if SolidObjects.configuration.wake_up_adapter
+        wake_up = SolidObjects.wake_up
+        return unless wake_up.respond_to?(:capability)
+        return if wake_up.capability.crosses_processes
 
         polling_warning_mutex.synchronize do
           return if polling_warning_emitted?

lib/solid_objects/wake_up.rb

@@ -2,6 +2,8 @@
 
 module SolidObjects
   class WakeUp
+    include ReportsWakeUpCapability
+
     class Watch
       # @rbs @wake_up: WakeUp
       # @rbs @generation: Integer
@@ -29,6 +31,16 @@ def initialize
       @generation = 0
     end
 
+    # @rbs () -> WakeUpCapability
+    def default_capability
+      WakeUpCapability.new(
+        adapter: :in_process,
+        crosses_processes: false,
+        measured_floor_ms: nil,
+        reason: "in-process signalling, which a commit in another process cannot reach"
+      )
+    end
+
     # @rbs () -> void
     def signal
       mutex.synchronize do

lib/solid_objects/wake_up_adapters.rb

@@ -2,22 +2,159 @@
 
 module SolidObjects
   module WakeUpAdapters
+    POSTGRESQL_FLOOR_MS = 2.9
+    REDIS_FLOOR_MS = 5.7
+    REDIS_URL_VARIABLE = "SOLID_OBJECTS_REDIS_URL"
+
+    @pooled_warning_mutex = Thread::Mutex.new
+    @pooled_warning_emitted = false
+
     module_function
 
-    # Returns the best wake-up strategy for a connection: cross-process
-    # notifications where the database provides them, and the in-process
-    # default everywhere else.
-    #
-    # This is deliberately not the default. A notification adapter opens a
-    # connection per waiting thread outside the pool, and `LISTEN` does not
-    # survive a transaction-pooling proxy such as PgBouncer, so adopting it is
-    # a deployment decision rather than an upgrade side effect.
-    #
+    NAMES = %i[automatic in_process postgresql redis].freeze
+
     # @rbs (?untyped) -> untyped
     def for(connection = Record.connection)
-      return Postgresql.new if DatabaseAdapter.family(connection) == :postgresql
+      select(connection)
+    end
+
+    # @rbs (untyped) -> untyped
+    def build(setting)
+      return select if setting.nil? || setting == :automatic
+      return named(setting) if setting.is_a?(Symbol)
+
+      configured(setting)
+    end
+
+    # @rbs (Symbol) -> untyped
+    def named(name)
+      case name
+      when :in_process then labelled(WakeUp.new, :in_process, false, nil, "in-process signalling was requested")
+      when :postgresql then labelled(Postgresql.new, :postgresql_notify, true, POSTGRESQL_FLOOR_MS, "PostgreSQL LISTEN was requested")
+      when :redis then labelled(Redis.new(url: redis_url), :redis, true, REDIS_FLOOR_MS, "Redis was requested")
+      else
+        raise ArgumentError, "unknown wake_up_adapter #{name.inspect}, expected one of #{NAMES.join(", ")} or an adapter"
+      end
+    end
+
+    # @rbs (untyped) -> untyped
+    def configured(adapter)
+      return adapter unless adapter.respond_to?(:capability=)
+
+      labelled(adapter, :configured, true, nil, "an adapter was configured, so selection did not run")
+    end
+
+    # @rbs (untyped, Symbol, bool, Numeric?, String) -> untyped
+    def labelled(adapter, name, crosses_processes, floor, reason)
+      adapter.capability = WakeUpCapability.new(
+        adapter: name,
+        crosses_processes:,
+        measured_floor_ms: floor,
+        reason:
+      )
+      adapter
+    end
+
+    # @rbs (?untyped) -> untyped
+    def select(connection = Record.connection)
+      url = redis_url
+      return redis_selection(url) if url
+
+      family = DatabaseAdapter.family(connection)
+      return postgresql_selection(connection) 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
+    rescue
+      nil
+    ensure
+      restore_application_name(connection, previous)
+    end
+
+    # @rbs () -> void
+    def reset_pooled_warning!
+      @pooled_warning_mutex.synchronize { @pooled_warning_emitted = false }
+    end
+
+    # @rbs () -> String?
+    def redis_url
+      value = ENV[REDIS_URL_VARIABLE].to_s
+      value.empty? ? nil : value
+    end
+
+    # @rbs (String) -> untyped
+    def redis_selection(url)
+      labelled(
+        Redis.new(url:), :redis, true, REDIS_FLOOR_MS,
+        "#{REDIS_URL_VARIABLE} is set, so Redis carries the signal between processes"
+      )
+    end
+
+    # @rbs (untyped) -> untyped
+    def postgresql_selection(connection)
+      survives = session_survives_transactions?(connection)
+      return pooled_selection if survives == false
+
+      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"
+      )
+    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"
+      )
+    end
+
+    # @rbs (Symbol?) -> untyped
+    def polling_selection(family)
+      polling_adapter(
+        "#{family || "this database"} has no notification channel and " \
+        "#{REDIS_URL_VARIABLE} is not set"
+      )
+    end
+
+    # @rbs (String) -> untyped
+    def polling_adapter(reason)
+      labelled(
+        WakeUp.new, :polling, false,
+        SolidObjects.configuration.idle_polling_interval * 1_000, reason
+      )
+    end
+
+    # @rbs () -> void
+    def warn_pooled_session_once
+      @pooled_warning_mutex.synchronize do
+        return if @pooled_warning_emitted
+
+        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"
+        )
+        @pooled_warning_emitted = true
+      end
+    end
+
+    # @rbs (untyped, untyped) -> void
+    def restore_application_name(connection, previous)
+      return if previous.nil?
 
-      WakeUp.new
+      connection.execute("SET application_name = #{connection.quote(previous)}")
+    rescue
+      nil
     end
   end
 end

lib/solid_objects/wake_up_adapters/postgresql.rb

@@ -10,6 +10,8 @@ module WakeUpAdapters
     # when one is available, so a missed or failed notification costs latency
     # rather than correctness.
     class Postgresql
+      include ReportsWakeUpCapability
+
       CHANNEL = "solid_objects_wake_up"
       FAILED_WAIT_INTERVAL = 0.05
 
@@ -19,6 +21,16 @@ class Postgresql
 
       attr_reader :channel
 
+      # @rbs () -> WakeUpCapability
+      def default_capability
+        WakeUpCapability.new(
+          adapter: :postgresql_notify,
+          crosses_processes: true,
+          measured_floor_ms: 2.9,
+          reason: "PostgreSQL LISTEN carries the signal between processes"
+        )
+      end
+
       # @rbs (?channel: String) -> void
       def initialize(channel: CHANNEL)
         @channel = channel