diff --git a/CHANGELOG.md b/CHANGELOG.md index d185a85..46cf05e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,39 @@ ## 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. +- Prove the PostgreSQL notification path before selecting it, because `LISTEN` + does not survive a transaction pooler such as PgBouncer. Selection listens on + a probe channel, 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. +- Select the wake-up adapter once per process. `SolidObjects.wake_up` memoised + without a lock, so threads that raced for the first use each ran a full + selection. +- 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. +- Poll rather than pretend when a requested adapter cannot be built. + `wake_up_adapter = :postgresql` on a database with no notification channel, + `:redis` without `SOLID_OBJECTS_REDIS_URL`, and a Redis URL without the redis + gem each log `solid_objects.wake_up.unavailable` once and record the reason in + the capability, so the doctor warns rather than claim a cross-process wake-up + that cannot happen. +- Validate `wake_up_adapter` in `configure`. An unknown name raised at the first + wake-up, which is after a commit, rather than at boot. +- 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 diff --git a/docs/adr/0011-wake-up-strategy.md b/docs/adr/0011-wake-up-strategy.md index 4a133d7..0e85f29 100644 --- a/docs/adr/0011-wake-up-strategy.md +++ b/docs/adr/0011-wake-up-strategy.md @@ -25,6 +25,14 @@ The interface supports: MySQL uses polling or optional Redis. SQLite uses polling plus the in-process signal; multi-host SQLite is outside its supported operating model. +Selection is automatic. `wake_up_adapter` takes a name or an adapter and +defaults to `:automatic`, which prefers a configured Redis URL, then PostgreSQL +notifications, then polling. PostgreSQL is chosen only after a probe +notification arrives, because `LISTEN` does not survive a transaction pooler. A +requested adapter that the environment cannot provide polls and records why, +rather than claim a wake-up it cannot deliver. `SolidObjects.wake_up.capability` +reports the choice, and the doctor reports the same record. + The synchronous caller first attempts to claim and execute the actor locally, so the normal path has no worker polling leg. When another process owns the activation, coordination overhead from completion commit until the caller's @@ -45,5 +53,7 @@ Timeout does not cancel durable work. - Redis loss only increases latency and never loses durable work. - Every adapter retains periodic polling to close startup, reconnect, and missed-message races. - A process that returns `false` from a timed wait participates in backoff; an older custom adapter that returns `nil` keeps the fast cadence. -- A multi-process deployment without an adapter trades idle database load for up to the current idle polling interval of notification latency and logs that topology once. +- A multi-process deployment whose installed adapter cannot cross processes trades idle database load for up to the current idle polling interval of notification latency and logs that topology once. +- A PostgreSQL deployment that configures nothing now pays one `NOTIFY` per enqueue after the commit and one listening connection per waiting thread, outside the pool. +- Selection runs once per process, under a lock, because the probe opens connections and waits. - Notification payloads never contain actor arguments or results. diff --git a/docs/operations.md b/docs/operations.md index 50f6d30..38b219f 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -245,13 +245,27 @@ to `idle_polling_interval`, which defaults to one second. Actor workers clamp the ceiling to `lease_renewal_interval` while they may hold cached activations. Set the fast and idle values equal for a fixed cadence. -The default wake-up interrupts waits only in the current Ruby process. When a -live process record shows that the database is shared across processes and no -adapter is configured, the runtime logs -`solid_objects.polling_only_cross_process_wake_up` once. Configure -`WakeUpAdapters::Postgresql` or `WakeUpAdapters::Redis` when separate processes -need prompt delivery. Without one, newly committed work can wait up to the -current idle polling interval. +Solid Objects selects a wake-up adapter on first use. `wake_up_adapter` defaults +to `:automatic`, which prefers `SOLID_OBJECTS_REDIS_URL`, then PostgreSQL +notifications, then polling. `SolidObjects.wake_up.capability` and the +`wake_up` doctor check report what was installed, whether it crosses processes, +its measured floor, and why. + +The in-process signal interrupts waits only in the current Ruby process. When a +live process record shows that the database is shared across processes and the +installed adapter does not cross them, the runtime logs +`solid_objects.polling_only_cross_process_wake_up` once. Newly committed work +can then wait up to the current idle polling interval. + +On PostgreSQL, selection proves the path first: it listens on a probe channel, +notifies it from a second connection, and waits for the notification. A probe +that does not arrive logs `solid_objects.wake_up.pooled_session` once and falls +back to polling, because `LISTEN` does not survive a transaction pooler such as +PgBouncer. A requested adapter that the environment cannot provide, such as +`:postgresql` on MySQL or `:redis` without `SOLID_OBJECTS_REDIS_URL`, logs +`solid_objects.wake_up.unavailable` once and polls rather than claim a +cross-process wake-up that cannot happen. Only an unknown name is refused, and +`configure` refuses it at boot. The warning excludes process rows with the current hostname and PID. It can therefore appear during a rolling deployment or restart overlap when an older diff --git a/docs/realtime.md b/docs/realtime.md index 5b14ee2..04eb011 100644 --- a/docs/realtime.md +++ b/docs/realtime.md @@ -140,46 +140,52 @@ explicitly serve the module. Turbo's normal morph rules still apply; use ## Cross-process wake-up -Runtime roles poll for work and are woken early by an in-process signal. That -signal cannot cross process boundaries, so a commit in a Puma process does not -wake a broadcast executor in a worker process, and delivery waits out +Runtime roles poll for work and are woken early by a signal. An in-process +signal cannot cross process boundaries, so a commit in a Puma process would not +wake a broadcast executor in a worker process, and delivery would wait out `polling_interval`, 100 ms by default. -On PostgreSQL, install the notification adapter to remove that delay: +Solid Objects selects the adapter for you. `wake_up_adapter` defaults to +`:automatic`, which prefers a configured Redis URL, then PostgreSQL +notifications, then polling: ```ruby # config/initializers/solid_objects.rb -configuration.wake_up_adapter = SolidObjects::WakeUpAdapters.for +configuration.wake_up_adapter = :automatic # the default +configuration.wake_up_adapter = :in_process # opt out +configuration.wake_up_adapter = :postgresql # force one +configuration.wake_up_adapter = MyAdapter.new # your own ``` -`WakeUpAdapters.for` returns notifications on PostgreSQL and the in-process -default on SQLite and MySQL, so the same line is safe across adapters. Name -`SolidObjects::WakeUpAdapters::Postgresql.new` directly to require it. - -MySQL has no notification primitive. MySQL applications either keep polling and -tune `polling_interval`, or configure the Redis adapter: - -```ruby -configuration.wake_up_adapter = SolidObjects::WakeUpAdapters::Redis.new( - url: ENV["REDIS_URL"] -) -``` - -Measured latency for a cross-process wake-up drops from 103.8 ms to 5.7 ms at -p50. The `redis` gem is not a dependency of this gem, so applications add it -themselves. One background subscription per process fans out to every waiting -role in memory, rather than one connection per thread, and `WakeUpAdapters.for` -does not select it: Redis is infrastructure this gem otherwise does not require, -so choosing it is explicit. +`SolidObjects.wake_up.capability` reports what was installed, whether it crosses +processes, its measured floor, and why. `bin/rails solid_objects:doctor` reports +the same record. +On PostgreSQL, selection proves the path before it chooses it. It listens on a +probe channel, sends one `NOTIFY` from a second connection, and waits for it to +arrive, because `LISTEN` does not survive a transaction pooler such as +PgBouncer. A probe that does not deliver falls back to polling and warns once. Measured latency for a cross-process wake-up drops from 103.7 ms to 2.9 ms at p50. The adapter keeps `polling_interval` as the upper bound: a missed or failed notification costs latency, never correctness, and signalling never raises into the caller that committed. `LISTEN` needs its own connection, so the adapter opens one outside the pool and releases it on `stop`. -Applications on SQLite or MySQL, or that do not configure the adapter, keep the -existing polling behaviour. +MySQL has no notification primitive, so MySQL applications either keep polling +and tune `polling_interval`, or set `SOLID_OBJECTS_REDIS_URL`, which selects +Redis on any database: + +```bash +SOLID_OBJECTS_REDIS_URL=redis://localhost:6379/0 +``` + +Measured latency for a cross-process wake-up drops from 103.8 ms to 5.7 ms at +p50. The `redis` gem is not a dependency of this gem, so applications add it +themselves, and selection polls and says so when the gem is missing. One +background subscription per process fans out to every waiting role in memory, +rather than one connection per thread. Name +`SolidObjects::WakeUpAdapters::Redis.new(url:)` directly for a URL that does not +come from the environment. ## Batched component refreshes diff --git a/docs/roadmap.md b/docs/roadmap.md index 9bbdaaa..b53f245 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -122,21 +122,23 @@ - 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 + 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, as does a requested adapter the environment cannot provide, so a + downgrade is recorded rather than hidden. + `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 diff --git a/lib/solid_objects.rb b/lib/solid_objects.rb index fcdb0a4..4cbbf03 100644 --- a/lib/solid_objects.rb +++ b/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" @@ -78,6 +79,8 @@ module SolidObjects extend Instrumentation + @wake_up_mutex = Thread::Mutex.new + class << self # @rbs () -> Configuration def configuration @@ -180,11 +183,11 @@ def mutable_copy(value) # @rbs () -> void def reset! ProcessRegistry.reset_polling_warning! if defined?(ProcessRegistry) + reset_wake_up! @configuration = Configuration.new @registry = ActorRegistry.new @client = nil @database_adapter = nil - @wake_up = nil @caller_process = nil @effect_registry = EffectRegistry.new @commit_action_registry = CommitActionRegistry.new @@ -198,9 +201,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 || @wake_up_mutex.synchronize { @wake_up ||= resolve_wake_up } + end + + # @rbs () -> void + def reset_wake_up! + @wake_up_mutex.synchronize { @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 diff --git a/lib/solid_objects/configuration.rb b/lib/solid_objects/configuration.rb index ab5bec4..87b2523 100644 --- a/lib/solid_objects/configuration.rb +++ b/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 } @@ -237,6 +237,7 @@ def validate! raise ArgumentError, "actor type cannot be empty" if actor_type.to_s.empty? raise ArgumentError, "instance retention must be positive" unless retention.positive? end + validate_wake_up_adapter! unless component_path_resolver.nil? || component_path_resolver.respond_to?(:call) raise ArgumentError, "component_path_resolver must respond to call" end @@ -252,6 +253,26 @@ def validate! private + # @rbs () -> void + def validate_wake_up_adapter! + return if wake_up_adapter.nil? + return validate_wake_up_object! unless wake_up_adapter.is_a?(Symbol) + return if WakeUpAdapters::NAMES.include?(wake_up_adapter) + + raise ArgumentError, + "unknown wake_up_adapter #{wake_up_adapter.inspect}, " \ + "expected one of #{WakeUpAdapters::NAMES.join(", ")} or an adapter" + end + + # @rbs () -> void + def validate_wake_up_object! + %i[signal wait].each do |method_name| + next if wake_up_adapter.respond_to?(method_name) + + raise ArgumentError, "wake_up_adapter must respond to #{method_name}" + end + end + # @rbs () -> Hash[Symbol, Numeric] def positive_values { diff --git a/lib/solid_objects/doctor.rb b/lib/solid_objects/doctor.rb index 2b54f1e..4ed4a92 100644 --- a/lib/solid_objects/doctor.rb +++ b/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 - diff --git a/lib/solid_objects/process_registry.rb b/lib/solid_objects/process_registry.rb index 6f76f26..4067656 100644 --- a/lib/solid_objects/process_registry.rb +++ b/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? diff --git a/lib/solid_objects/wake_up.rb b/lib/solid_objects/wake_up.rb index 2f0f38a..ef1fb51 100644 --- a/lib/solid_objects/wake_up.rb +++ b/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 diff --git a/lib/solid_objects/wake_up_adapters.rb b/lib/solid_objects/wake_up_adapters.rb index a0b79e1..7b80c33 100644 --- a/lib/solid_objects/wake_up_adapters.rb +++ b/lib/solid_objects/wake_up_adapters.rb @@ -2,22 +2,217 @@ module SolidObjects module WakeUpAdapters + POSTGRESQL_FLOOR_MS = 2.9 + REDIS_FLOOR_MS = 5.7 + REDIS_URL_VARIABLE = "SOLID_OBJECTS_REDIS_URL" + PROBE_TIMEOUT_SECONDS = 2.0 + PROBE_CHANNEL = "solid_objects_wake_up_probe" + + @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 requested_postgresql + when :redis then requested_redis + else + raise ArgumentError, "unknown wake_up_adapter #{name.inspect}, expected one of #{NAMES.join(", ")} or an adapter" + end + end + + # @rbs () -> untyped + def requested_postgresql + family = DatabaseAdapter.family(Record.connection) + unless family == :postgresql + return unavailable_selection( + "wake_up_adapter :postgresql needs a database with a notification " \ + "channel, and #{family || "this database"} provides none" + ) + end + + labelled(Postgresql.new, :postgresql_notify, true, POSTGRESQL_FLOOR_MS, "PostgreSQL LISTEN was requested") + end + + # @rbs () -> untyped + def requested_redis + url = redis_url + unless url + return unavailable_selection( + "wake_up_adapter :redis needs #{REDIS_URL_VARIABLE}, which is not set" + ) + end + + redis_adapter(url, "Redis was requested") + end + + # @rbs (String, String) -> untyped + def redis_adapter(url, reason) + return unavailable_selection("#{REDIS_URL_VARIABLE} is set, and the redis gem is not installed") unless redis_installed? + + labelled(Redis.new(url:), :redis, true, REDIS_FLOOR_MS, reason) + end + + # @rbs () -> bool + def redis_installed? + require "redis" + true + rescue LoadError + false + end + + # @rbs (String) -> untyped + def unavailable_selection(reason) + SolidObjects.configuration.logger.warn( + event: "solid_objects.wake_up.unavailable", + reason: + ) + polling_adapter(reason) + end + + # @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 + + # @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 if family == :postgresql + + polling_selection(family) + end + + # @rbs () -> bool + def notifications_deliver? + probe = Postgresql.new(channel: PROBE_CHANNEL) + return false unless probe.listen + return false unless notify_probe_channel + + probe.wait(timeout: PROBE_TIMEOUT_SECONDS) + rescue + false + ensure + probe&.stop + end + + # @rbs () -> bool + def notify_probe_channel + connection = Record.connection_pool.send(:new_connection) + connection.execute("NOTIFY #{connection.quote_table_name(PROBE_CHANNEL)}") + true + ensure + disconnect_probe(connection) + end + + # @rbs (untyped) -> void + def disconnect_probe(connection) + connection&.disconnect! + rescue + nil + 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) + redis_adapter( + url, + "#{REDIS_URL_VARIABLE} is set, so Redis carries the signal between processes" + ) + end + + # @rbs () -> untyped + def postgresql_selection + return pooled_selection unless notifications_deliver? + + labelled( + Postgresql.new, :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( + "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 + + # @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 - WakeUp.new + SolidObjects.configuration.logger.warn( + event: "solid_objects.wake_up.pooled_session", + reason: "PostgreSQL notifications were not selected because a probe " \ + "notification did not arrive" + ) + @pooled_warning_emitted = true + end end end end diff --git a/lib/solid_objects/wake_up_adapters/postgresql.rb b/lib/solid_objects/wake_up_adapters/postgresql.rb index f66469c..8c45271 100644 --- a/lib/solid_objects/wake_up_adapters/postgresql.rb +++ b/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 diff --git a/lib/solid_objects/wake_up_adapters/redis.rb b/lib/solid_objects/wake_up_adapters/redis.rb index 2f3b7e5..bc44dc5 100644 --- a/lib/solid_objects/wake_up_adapters/redis.rb +++ b/lib/solid_objects/wake_up_adapters/redis.rb @@ -12,6 +12,8 @@ module WakeUpAdapters # polling interval remains the upper bound, so a missed or failed # notification costs latency rather than correctness. class Redis + include ReportsWakeUpCapability + class Watch # @rbs @adapter: Redis # @rbs @generation: Integer @@ -43,6 +45,16 @@ def wait(timeout:) attr_reader :channel + # @rbs () -> WakeUpCapability + def default_capability + WakeUpCapability.new( + adapter: :redis, + crosses_processes: true, + measured_floor_ms: 5.7, + reason: "Redis carries the signal between processes" + ) + end + # @rbs (?channel: String, ?url: String?, ?client: untyped) -> void def initialize(channel: CHANNEL, url: nil, client: nil) @channel = channel diff --git a/lib/solid_objects/wake_up_capability.rb b/lib/solid_objects/wake_up_capability.rb new file mode 100644 index 0000000..d0f5406 --- /dev/null +++ b/lib/solid_objects/wake_up_capability.rb @@ -0,0 +1,15 @@ +# rbs_inline: enabled + +module SolidObjects + WakeUpCapability = Data.define(:adapter, :crosses_processes, :measured_floor_ms, :reason) + + module ReportsWakeUpCapability + # @rbs (WakeUpCapability) -> void + attr_writer :capability + + # @rbs () -> WakeUpCapability + def capability + @capability ||= default_capability + end + end +end diff --git a/sig/generated/lib/solid_objects.rbs b/sig/generated/lib/solid_objects.rbs index 1eaf5bb..7c56789 100644 --- a/sig/generated/lib/solid_objects.rbs +++ b/sig/generated/lib/solid_objects.rbs @@ -57,6 +57,15 @@ module SolidObjects # @rbs () -> DatabaseAdapter def self.database_adapter: () -> DatabaseAdapter - # @rbs () -> WakeUp - def self.wake_up: () -> WakeUp + # @rbs () -> untyped + def self.wake_up: () -> untyped + + # @rbs () -> void + def self.reset_wake_up!: () -> void + + # @rbs () -> untyped + def self.resolve_wake_up: () -> untyped + + # @rbs (Exception) -> untyped + def self.unreachable_wake_up: (Exception) -> untyped end diff --git a/sig/generated/lib/solid_objects/configuration.rbs b/sig/generated/lib/solid_objects/configuration.rbs index c565f0f..0cecf1f 100644 --- a/sig/generated/lib/solid_objects/configuration.rbs +++ b/sig/generated/lib/solid_objects/configuration.rbs @@ -2,6 +2,8 @@ module SolidObjects class Configuration + @process_heartbeat_interval: Float + @process_alive_threshold: Float @shutdown_timeout: Float @@ -96,8 +98,6 @@ module SolidObjects @lock_retry_attempts: Integer - @process_heartbeat_interval: Float - attr_accessor table_name_prefix: untyped attr_accessor polling_interval: untyped @@ -230,6 +230,12 @@ module SolidObjects private + # @rbs () -> void + def validate_wake_up_adapter!: () -> void + + # @rbs () -> void + def validate_wake_up_object!: () -> void + # @rbs () -> Hash[Symbol, Numeric] def positive_values: () -> Hash[Symbol, Numeric] diff --git a/sig/generated/lib/solid_objects/doctor.rbs b/sig/generated/lib/solid_objects/doctor.rbs index a6ff3c9..dc6c636 100644 --- a/sig/generated/lib/solid_objects/doctor.rbs +++ b/sig/generated/lib/solid_objects/doctor.rbs @@ -75,6 +75,9 @@ module SolidObjects # @rbs () -> Check def check_database_server: () -> Check + # @rbs () -> Check + def check_wake_up: () -> Check + # @rbs () -> Check def check_runtime: () -> Check diff --git a/sig/generated/lib/solid_objects/wake_up.rbs b/sig/generated/lib/solid_objects/wake_up.rbs index 6419976..790a465 100644 --- a/sig/generated/lib/solid_objects/wake_up.rbs +++ b/sig/generated/lib/solid_objects/wake_up.rbs @@ -2,6 +2,8 @@ module SolidObjects class WakeUp + include ReportsWakeUpCapability + class Watch @wake_up: WakeUp @@ -23,6 +25,9 @@ module SolidObjects # @rbs () -> void def initialize: () -> void + # @rbs () -> WakeUpCapability + def default_capability: () -> WakeUpCapability + # @rbs () -> void def signal: () -> void diff --git a/sig/generated/lib/solid_objects/wake_up_adapters.rbs b/sig/generated/lib/solid_objects/wake_up_adapters.rbs index ed1bc94..dbd6107 100644 --- a/sig/generated/lib/solid_objects/wake_up_adapters.rbs +++ b/sig/generated/lib/solid_objects/wake_up_adapters.rbs @@ -2,16 +2,82 @@ module SolidObjects module WakeUpAdapters - # 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. - # + POSTGRESQL_FLOOR_MS: ::Float + + REDIS_FLOOR_MS: ::Float + + REDIS_URL_VARIABLE: ::String + + PROBE_TIMEOUT_SECONDS: ::Float + + PROBE_CHANNEL: ::String + + NAMES: untyped + # @rbs (?untyped) -> untyped def self?.for: (?untyped) -> untyped + + # @rbs (untyped) -> untyped + def self?.build: (untyped) -> untyped + + # @rbs (Symbol) -> untyped + def self?.named: (Symbol) -> untyped + + # @rbs () -> untyped + def self?.requested_postgresql: () -> untyped + + # @rbs () -> untyped + def self?.requested_redis: () -> untyped + + # @rbs (String, String) -> untyped + def self?.redis_adapter: (String, String) -> untyped + + # @rbs () -> bool + def self?.redis_installed?: () -> bool + + # @rbs (String) -> untyped + def self?.unavailable_selection: (String) -> untyped + + # @rbs (untyped) -> untyped + def self?.configured: (untyped) -> untyped + + # @rbs (untyped, Symbol, bool, Numeric?, String) -> untyped + def self?.labelled: (untyped, Symbol, bool, Numeric?, String) -> untyped + + # @rbs (?untyped) -> untyped + def self?.select: (?untyped) -> untyped + + # @rbs () -> bool + def self?.notifications_deliver?: () -> bool + + # @rbs () -> bool + def self?.notify_probe_channel: () -> bool + + # @rbs (untyped) -> void + def self?.disconnect_probe: (untyped) -> void + + # @rbs () -> void + def self?.reset_pooled_warning!: () -> void + + # @rbs () -> String? + def self?.redis_url: () -> String? + + # @rbs (String) -> untyped + def self?.redis_selection: (String) -> untyped + + # @rbs () -> untyped + def self?.postgresql_selection: () -> untyped + + # @rbs () -> untyped + def self?.pooled_selection: () -> untyped + + # @rbs (Symbol?) -> untyped + def self?.polling_selection: (Symbol?) -> untyped + + # @rbs (String) -> untyped + def self?.polling_adapter: (String) -> untyped + + # @rbs () -> void + def self?.warn_pooled_session_once: () -> void end end diff --git a/sig/generated/lib/solid_objects/wake_up_adapters/postgresql.rbs b/sig/generated/lib/solid_objects/wake_up_adapters/postgresql.rbs index 655bae3..e2c2f57 100644 --- a/sig/generated/lib/solid_objects/wake_up_adapters/postgresql.rbs +++ b/sig/generated/lib/solid_objects/wake_up_adapters/postgresql.rbs @@ -10,6 +10,8 @@ module SolidObjects # when one is available, so a missed or failed notification costs latency # rather than correctness. class Postgresql + include ReportsWakeUpCapability + CHANNEL: ::String FAILED_WAIT_INTERVAL: ::Float @@ -22,6 +24,9 @@ module SolidObjects attr_reader channel: untyped + # @rbs () -> WakeUpCapability + def default_capability: () -> WakeUpCapability + # @rbs (?channel: String) -> void def initialize: (?channel: String) -> void diff --git a/sig/generated/lib/solid_objects/wake_up_adapters/redis.rbs b/sig/generated/lib/solid_objects/wake_up_adapters/redis.rbs index 4b2ef54..d5c57d4 100644 --- a/sig/generated/lib/solid_objects/wake_up_adapters/redis.rbs +++ b/sig/generated/lib/solid_objects/wake_up_adapters/redis.rbs @@ -10,6 +10,8 @@ module SolidObjects # polling interval remains the upper bound, so a missed or failed # notification costs latency rather than correctness. class Redis + include ReportsWakeUpCapability + class Watch @adapter: Redis @@ -46,6 +48,9 @@ module SolidObjects attr_reader channel: untyped + # @rbs () -> WakeUpCapability + def default_capability: () -> WakeUpCapability + # @rbs (?channel: String, ?url: String?, ?client: untyped) -> void def initialize: (?channel: String, ?url: String?, ?client: untyped) -> void diff --git a/sig/generated/lib/solid_objects/wake_up_capability.rbs b/sig/generated/lib/solid_objects/wake_up_capability.rbs new file mode 100644 index 0000000..6de316f --- /dev/null +++ b/sig/generated/lib/solid_objects/wake_up_capability.rbs @@ -0,0 +1,28 @@ +# Generated from lib/solid_objects/wake_up_capability.rb with RBS::Inline + +module SolidObjects + class WakeUpCapability < Data + attr_reader adapter(): untyped + + attr_reader crosses_processes(): untyped + + attr_reader measured_floor_ms(): untyped + + attr_reader reason(): untyped + + def self.new: (untyped adapter, untyped crosses_processes, untyped measured_floor_ms, untyped reason) -> instance + | (adapter: untyped, crosses_processes: untyped, measured_floor_ms: untyped, reason: untyped) -> instance + + def self.members: () -> [ :adapter, :crosses_processes, :measured_floor_ms, :reason ] + + def members: () -> [ :adapter, :crosses_processes, :measured_floor_ms, :reason ] + end + + module ReportsWakeUpCapability + # @rbs (WakeUpCapability) -> void + attr_writer capability: untyped + + # @rbs () -> WakeUpCapability + def capability: () -> WakeUpCapability + end +end diff --git a/test/integration/enqueue_statement_count_test.rb b/test/integration/enqueue_statement_count_test.rb index 2059031..5910b5b 100644 --- a/test/integration/enqueue_statement_count_test.rb +++ b/test/integration/enqueue_statement_count_test.rb @@ -15,7 +15,11 @@ def add(product_id:) STEADY_STATE_STATEMENT_COUNT = 10 - setup { CartActor.ensure_registered! } + setup do + CartActor.ensure_registered! + SolidObjects.configuration.wake_up_adapter = :in_process + SolidObjects.reset_wake_up! + end test "a steady-state enqueue never inserts the instance row" do reference = CartActor.ref("alice") diff --git a/test/integration/polling_test.rb b/test/integration/polling_test.rb index 3e9bece..6da9777 100644 --- a/test/integration/polling_test.rb +++ b/test/integration/polling_test.rb @@ -230,6 +230,8 @@ def signal test "warns once when another process shares the database without a wake-up adapter" do logger = RecordingLogger.new + SolidObjects.configuration.wake_up_adapter = :in_process + SolidObjects.reset_wake_up! SolidObjects.configuration.logger = logger SolidObjects.configuration.polling_interval = 0.025 SolidObjects.configuration.idle_polling_interval = 1.0 diff --git a/test/integration/wake_up_selection_test.rb b/test/integration/wake_up_selection_test.rb new file mode 100644 index 0000000..a3af384 --- /dev/null +++ b/test/integration/wake_up_selection_test.rb @@ -0,0 +1,299 @@ +# frozen_string_literal: true + +require "database_test_helper" +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 + + test "an explicitly configured adapter wins" do + explicit = SolidObjects::WakeUp.new + SolidObjects.configuration.wake_up_adapter = explicit + + assert_same explicit, SolidObjects.wake_up + 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 "threads that race for the adapter select it once" do + selections = Queue.new + start = Queue.new + resolved = Queue.new + + with_module_method(:build, ->(_setting) { + selections << true + sleep 0.05 + SolidObjects::WakeUp.new + }) do + threads = 8.times.map do + Thread.new do + start.pop + resolved << SolidObjects.wake_up + end + end + threads.length.times { start << true } + threads.each(&:join) + end + + assert_equal 1, selections.size + assert_equal 1, resolved.size.times.map { resolved.pop.object_id }.uniq.size + end + + test "in_process opts out of selection" do + SolidObjects.configuration.wake_up_adapter = :in_process + + capability = SolidObjects.wake_up.capability + assert_equal :in_process, capability.adapter + assert_not capability.crosses_processes + end + + test "a name selects that adapter without probing" do + ENV["SOLID_OBJECTS_REDIS_URL"] = "redis://127.0.0.1:6379/15" + SolidObjects.configuration.wake_up_adapter = :redis + + capability = SolidObjects.wake_up.capability + assert_equal :redis, capability.adapter + assert_match(/requested/i, capability.reason) + end + + test "a requested postgresql adapter polls when the database has no channel" do + skip if database_family == :postgresql + warnings = [] + SolidObjects.configuration.logger = Logger.new(IO::NULL).tap do |logger| + logger.define_singleton_method(:warn) { |payload| warnings << payload } + end + SolidObjects.configuration.wake_up_adapter = :postgresql + + capability = SolidObjects.wake_up.capability + assert_equal :polling, capability.adapter + assert_not capability.crosses_processes + assert_match(/notification channel/i, capability.reason) + assert_equal [ "solid_objects.wake_up.unavailable" ], + warnings.map { |payload| payload[:event].to_s } + end + + test "a requested redis adapter polls when no url is set" do + warnings = [] + SolidObjects.configuration.logger = Logger.new(IO::NULL).tap do |logger| + logger.define_singleton_method(:warn) { |payload| warnings << payload } + end + SolidObjects.configuration.wake_up_adapter = :redis + + capability = SolidObjects.wake_up.capability + assert_equal :polling, capability.adapter + assert_match(/SOLID_OBJECTS_REDIS_URL/, capability.reason) + assert_equal [ "solid_objects.wake_up.unavailable" ], + warnings.map { |payload| payload[:event].to_s } + end + + test "an adapter without watch is still accepted" do + legacy = Class.new do + def signal = true + + def wait(timeout:) = false + end.new + SolidObjects.configuration.wake_up_adapter = legacy + + assert_same SolidObjects.configuration, SolidObjects.configuration.validate! + assert_same legacy, SolidObjects.wake_up + end + + test "an adapter that cannot signal is refused" do + SolidObjects.configuration.wake_up_adapter = Object.new + + error = assert_raises(ArgumentError) { SolidObjects.configuration.validate! } + assert_match(/signal/, error.message) + end + + test "an unknown name is refused when the configuration is validated" do + SolidObjects.configuration.wake_up_adapter = :carrier_pigeon + + error = assert_raises(ArgumentError) { SolidObjects.configuration.validate! } + assert_match(/carrier_pigeon/, error.message) + assert_match(/automatic/, error.message) + end + + test "an unknown name is refused rather than silently polling" do + SolidObjects.configuration.wake_up_adapter = :carrier_pigeon + + error = assert_raises(ArgumentError) { SolidObjects.wake_up } + assert_match(/carrier_pigeon/, error.message) + assert_match(/automatic/, error.message) + end + + test "a redis url selects redis on any database" do + ENV["SOLID_OBJECTS_REDIS_URL"] = "redis://127.0.0.1:6379/15" + + capability = SolidObjects.wake_up.capability + assert_equal :redis, capability.adapter + assert capability.crosses_processes + assert_match(/redis/i, capability.reason) + end + + 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 listener wakes from a notification sent on another connection" do + skip unless database_family == :postgresql + channel = "solid_objects_probe_parity" + adapter = SolidObjects::WakeUpAdapters::Postgresql.new(channel:) + + assert adapter.listen + SolidObjects::Record.connection_pool.with_connection do |connection| + connection.execute("NOTIFY #{connection.quote_table_name(channel)}") + end + + assert adapter.wait(timeout: 2.0) + ensure + adapter&.stop + end + + test "a database without a channel polls and reports its floor" do + skip if database_family == :postgresql + + capability = SolidObjects.wake_up.capability + assert_equal :polling, capability.adapter + assert_not capability.crosses_processes + assert_equal SolidObjects.configuration.idle_polling_interval * 1_000, + capability.measured_floor_ms + assert_match(/no notification channel/i, capability.reason) + end + + test "a pooled postgresql session falls back to polling and warns once" do + skip unless database_family == :postgresql + warnings = [] + SolidObjects.configuration.logger = Logger.new(IO::NULL).tap do |logger| + logger.define_singleton_method(:warn) { |payload| warnings << payload } + end + with_pooled_session do + capability = SolidObjects.wake_up.capability + + assert_equal :polling, capability.adapter + assert_match(/pool/i, capability.reason) + end + + assert_equal 1, warnings.count { |payload| payload[:event].to_s.include?("wake_up") } + end + + test "the capability names the adapter that is actually installed" do + expected = { + "SolidObjects::WakeUpAdapters::Postgresql" => :postgresql_notify, + "SolidObjects::WakeUpAdapters::Redis" => :redis, + "SolidObjects::WakeUp" => :polling + } + + assert_equal expected.fetch(SolidObjects.wake_up.class.name), + SolidObjects.wake_up.capability.adapter + end + + test "selection survives a database that cannot be reached" do + with_unreachable_database do + capability = SolidObjects.wake_up.capability + + assert_equal :in_process, capability.adapter + assert_not capability.crosses_processes + assert_match(/could not be reached/i, capability.reason) + end + end + + test "the doctor reports the selected adapter" do + SolidObjects.configuration.authorize_administration = ->(**) { true } + check = SolidObjects::Doctor.new.call.check(:wake_up) + + assert check, "the doctor should report a wake_up check" + assert_equal SolidObjects.wake_up.capability.crosses_processes, check.status == :pass + assert_match(/#{SolidObjects.wake_up.capability.adapter}/, check.message) + end + + private + + def with_module_method(name, replacement) + adapters = SolidObjects::WakeUpAdapters + original = adapters.method(name) + adapters.define_singleton_method(name, replacement) + yield + ensure + adapters.define_singleton_method(name, original) + end + + def with_pooled_session(&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) + with_module_method(:select, ->(*) { raise ActiveRecord::ConnectionNotEstablished }, &block) + end +end diff --git a/test/unit/wake_up_adapters_test.rb b/test/unit/wake_up_adapters_test.rb index 3e9cd59..123a060 100644 --- a/test/unit/wake_up_adapters_test.rb +++ b/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(-> { true }, &block) + end + + def with_undelivered_notifications(&block) + with_probe(-> { 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