diff --git a/CHANGELOG.md b/CHANGELOG.md index 0624111..d882d8d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,17 @@ # Changelog +## Unreleased + +- Add `SolidObjects::WakeUpAdapters::Postgresql`, an optional cross-process + wake-up using PostgreSQL notifications. In-process signalling cannot reach a + worker process, so reactive delivery waited out `polling_interval`. With the + adapter configured, measured cross-process wake-up latency drops from 103.7 ms + to 2.9 ms at p50. The polling interval remains the upper bound, and neither + signalling nor waiting raises into its caller. `WakeUpAdapters.for` selects + notifications on PostgreSQL and the in-process default elsewhere; it is not + the default, because the adapter opens a connection per waiting thread + outside the pool and `LISTEN` does not survive a transaction-pooling proxy. + ## 0.7.3 - 2026-08-09 - Coordinate batched component refreshes by revision as well as scope and batch diff --git a/docs/benchmarks.md b/docs/benchmarks.md index 73f87af..ebea3d8 100644 --- a/docs/benchmarks.md +++ b/docs/benchmarks.md @@ -76,6 +76,21 @@ Cable delivery, or browser rendering, which dominate wall-clock time in a real deployment and make the request-count difference matter more than it appears here. End-to-end latency against a deployed application has not been measured. +## Cross-process wake-up + +Measured 2026-08-09 against PostgreSQL 17, 30 samples, with `polling_interval` +at its 100 ms default and a signal sent 2 ms after the waiter began. + +| Wake-up strategy | p50 | p95 | +| --- | ---: | ---: | +| In-process `WakeUp` | 103.7 ms | 105.1 ms | +| `WakeUpAdapters::Postgresql` | 2.9 ms | 5.1 ms | + +The in-process wake-up cannot reach another process, so a worker waits out the +full polling interval no matter how quickly the web process committed. The +notification adapter removes that floor rather than shrinking it, and the +polling interval remains the upper bound if a notification is missed. + ## Durable row growth The storage cost is deterministic even when latency is not: diff --git a/docs/realtime.md b/docs/realtime.md index 5be7673..4a5e4ad 100644 --- a/docs/realtime.md +++ b/docs/realtime.md @@ -110,6 +110,36 @@ applications discover the namespaced engine asset. Applications created with explicitly serve the module. Turbo's normal morph rules still apply; use `data-turbo-permanent` for elements that must never be changed. +## 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 +`polling_interval`, 100 ms by default. + +On PostgreSQL, install the notification adapter to remove that delay: + +```ruby +# config/initializers/solid_objects.rb +configuration.wake_up_adapter = SolidObjects::WakeUpAdapters.for +``` + +`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, so MySQL applications keep polling and tune +`polling_interval`. + +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. + ## Batched component refreshes A component refresh costs one browser request. When one actor mutation changes diff --git a/lib/generators/solid_objects/templates/solid_objects.rb b/lib/generators/solid_objects/templates/solid_objects.rb index 085e1cc..9b7dbe4 100644 --- a/lib/generators/solid_objects/templates/solid_objects.rb +++ b/lib/generators/solid_objects/templates/solid_objects.rb @@ -58,4 +58,12 @@ # configuration.authorize_administration = lambda do |authorization_context:, **| # authorization_context.is_a?(Hash) && authorization_context[:source] == "cli" # end + + # Runtime roles poll for work and are woken early by an in-process signal, + # which cannot reach another process. On PostgreSQL, notifications remove that + # delay. This opens a connection per waiting thread outside the pool and does + # not work through a transaction-pooling proxy such as PgBouncer, so it is + # opt-in: + # + # configuration.wake_up_adapter = SolidObjects::WakeUpAdapters.for end diff --git a/lib/solid_objects.rb b/lib/solid_objects.rb index 11bc94b..31f40dd 100644 --- a/lib/solid_objects.rb +++ b/lib/solid_objects.rb @@ -46,6 +46,8 @@ require "solid_objects/actor_channel" require "solid_objects/action_cable_broadcast_adapter" require "solid_objects/wake_up" +require "solid_objects/wake_up_adapters/postgresql" +require "solid_objects/wake_up_adapters" require "solid_objects/effect_registry" require "solid_objects/commit_action_registry" require "solid_objects/lease" diff --git a/lib/solid_objects/supervisor.rb b/lib/solid_objects/supervisor.rb index 1519050..7600102 100644 --- a/lib/solid_objects/supervisor.rb +++ b/lib/solid_objects/supervisor.rb @@ -44,17 +44,35 @@ def start def stop return unless @started - components.each(&:request_shutdown) - join_until_timeout - components.reject(&:stopped?).each(&:stop) - @started = false - SolidObjects.instrument(:"supervisor.stopped", component_count: components.length) + begin + components.each(&:request_shutdown) + join_until_timeout + components.reject(&:stopped?).each(&:stop) + ensure + # Connections held outside the pool must be released even when a + # component fails to stop, or they accumulate across restarts. + release_wake_up + @started = false + SolidObjects.instrument(:"supervisor.stopped", component_count: components.length) + end end private attr_reader :components, :threads + # A wake-up adapter may hold connections outside the pool, which would + # otherwise accumulate across restarts in one process. + # @rbs () -> void + def release_wake_up + wake_up = SolidObjects.wake_up + return unless wake_up.respond_to?(:stop) + + wake_up.stop + rescue + nil + end + # @rbs (worker_count: Integer, effect_worker_count: Integer, broadcast_worker_count: Integer, reminder_scheduler_count: Integer) -> Array[Worker | EffectExecutor | ReminderScheduler | BroadcastExecutor] def build_components( worker_count:, diff --git a/lib/solid_objects/wake_up_adapters.rb b/lib/solid_objects/wake_up_adapters.rb new file mode 100644 index 0000000..d4f3d50 --- /dev/null +++ b/lib/solid_objects/wake_up_adapters.rb @@ -0,0 +1,23 @@ +# rbs_inline: enabled + +module SolidObjects + module WakeUpAdapters + 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. + # + # @rbs (?untyped) -> untyped + def for(connection = Record.connection) + return Postgresql.new if connection.adapter_name.match?(/postgres/i) + + WakeUp.new + end + end +end diff --git a/lib/solid_objects/wake_up_adapters/postgresql.rb b/lib/solid_objects/wake_up_adapters/postgresql.rb new file mode 100644 index 0000000..47a8bcc --- /dev/null +++ b/lib/solid_objects/wake_up_adapters/postgresql.rb @@ -0,0 +1,136 @@ +# rbs_inline: enabled + +module SolidObjects + module WakeUpAdapters + # Wakes runtime roles across processes using PostgreSQL notifications. + # + # The in-process wake-up cannot reach another process, so a commit in a web + # process leaves a worker waiting out its polling interval. This adapter + # keeps that polling interval as the upper bound and delivers a notification + # when one is available, so a missed or failed notification costs latency + # rather than correctness. + class Postgresql + CHANNEL = "solid_objects_wake_up" + FAILED_WAIT_INTERVAL = 0.05 + + # @rbs @channel: String + # @rbs @mutex: Thread::Mutex + # @rbs @connections: Array[untyped] + + attr_reader :channel + + # @rbs (?channel: String) -> void + def initialize(channel: CHANNEL) + @channel = channel + @mutex = Thread::Mutex.new + @connections = [] + end + + # @rbs () -> bool + def signal + notify_channel + true + rescue => error + instrument_failure(:signal, error) + false + end + + # @rbs (timeout: Numeric) -> bool + def wait(timeout:) + connection = listening_connection + !connection.raw_connection.wait_for_notify(timeout.to_f).nil? + rescue => error + instrument_failure(:wait, error) + pace_after_failure(timeout) + false + end + + # Starts listening before a caller blocks, so a notification sent between + # startup and the first wait is not missed. + # @rbs () -> bool + def listen + listening_connection + true + rescue => error + instrument_failure(:listen, error) + false + end + + # @rbs () -> bool + def stop + open = mutex.synchronize do + listening = connections.dup + connections.clear + listening + end + Thread.current[thread_key] = nil + open.each { |connection| disconnect(connection) } + open.any? + end + + private + + attr_reader :mutex, :connections + + # @rbs () -> void + def notify_channel + Record.connection_pool.with_connection do |connection| + connection.execute("NOTIFY #{connection.quote_table_name(channel)}") + end + end + + # 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 + connection = Thread.current[thread_key] + return connection if connection&.active? + + 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 + def pace_after_failure(timeout) + interval = [ timeout.to_f, FAILED_WAIT_INTERVAL ].min + return unless interval.positive? + + sleep interval + end + + # @rbs (Symbol, Exception) -> void + def instrument_failure(operation, error) + SolidObjects.instrument( + :"wake_up.failed", + adapter: "postgresql", + operation: operation.to_s, + error_class: error.class.name, + error_message: error.message + ) + end + end + end +end diff --git a/sig/generated/lib/solid_objects/supervisor.rbs b/sig/generated/lib/solid_objects/supervisor.rbs index 2c3f622..24f53b5 100644 --- a/sig/generated/lib/solid_objects/supervisor.rbs +++ b/sig/generated/lib/solid_objects/supervisor.rbs @@ -26,6 +26,11 @@ module SolidObjects attr_reader threads: untyped + # A wake-up adapter may hold connections outside the pool, which would + # otherwise accumulate across restarts in one process. + # @rbs () -> void + def release_wake_up: () -> void + # @rbs (worker_count: Integer, effect_worker_count: Integer, broadcast_worker_count: Integer, reminder_scheduler_count: Integer) -> Array[Worker | EffectExecutor | ReminderScheduler | BroadcastExecutor] def build_components: (worker_count: Integer, effect_worker_count: Integer, broadcast_worker_count: Integer, reminder_scheduler_count: Integer) -> Array[Worker | EffectExecutor | ReminderScheduler | BroadcastExecutor] diff --git a/sig/generated/lib/solid_objects/wake_up_adapters.rbs b/sig/generated/lib/solid_objects/wake_up_adapters.rbs new file mode 100644 index 0000000..ed1bc94 --- /dev/null +++ b/sig/generated/lib/solid_objects/wake_up_adapters.rbs @@ -0,0 +1,17 @@ +# Generated from lib/solid_objects/wake_up_adapters.rb with RBS::Inline + +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. + # + # @rbs (?untyped) -> untyped + def self?.for: (?untyped) -> untyped + 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 new file mode 100644 index 0000000..f363278 --- /dev/null +++ b/sig/generated/lib/solid_objects/wake_up_adapters/postgresql.rbs @@ -0,0 +1,75 @@ +# Generated from lib/solid_objects/wake_up_adapters/postgresql.rb with RBS::Inline + +module SolidObjects + module WakeUpAdapters + # Wakes runtime roles across processes using PostgreSQL notifications. + # + # The in-process wake-up cannot reach another process, so a commit in a web + # process leaves a worker waiting out its polling interval. This adapter + # keeps that polling interval as the upper bound and delivers a notification + # when one is available, so a missed or failed notification costs latency + # rather than correctness. + class Postgresql + CHANNEL: ::String + + FAILED_WAIT_INTERVAL: ::Float + + @connections: Array[untyped] + + @mutex: Thread::Mutex + + @channel: String + + attr_reader channel: untyped + + # @rbs (?channel: String) -> void + def initialize: (?channel: String) -> void + + # @rbs () -> bool + def signal: () -> bool + + # @rbs (timeout: Numeric) -> bool + def wait: (timeout: Numeric) -> bool + + # Starts listening before a caller blocks, so a notification sent between + # startup and the first wait is not missed. + # @rbs () -> bool + def listen: () -> bool + + # @rbs () -> bool + def stop: () -> bool + + private + + attr_reader mutex: untyped + + attr_reader connections: untyped + + # @rbs () -> void + def notify_channel: () -> void + + # 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 + + # @rbs (Symbol, Exception) -> void + def instrument_failure: (Symbol, Exception) -> void + end + end +end diff --git a/test/integration/postgresql_wake_up_test.rb b/test/integration/postgresql_wake_up_test.rb new file mode 100644 index 0000000..60933b8 --- /dev/null +++ b/test/integration/postgresql_wake_up_test.rb @@ -0,0 +1,185 @@ +# frozen_string_literal: true + +require "database_test_helper" +require "timeout" + +class PostgresqlWakeUpTest < ActiveSupport::TestCase + setup do + skip unless postgresql? + + @adapter = SolidObjects::WakeUpAdapters::Postgresql.new + end + + teardown { @adapter&.stop } + + test "a signal from another connection wakes a waiter" do + waiting = Queue.new + woken = Queue.new + waiter = Thread.new do + @adapter.listen + waiting << true + started = monotonic_now + @adapter.wait(timeout: 5) + woken << monotonic_now - started + end + Timeout.timeout(5) { waiting.pop } + + signal_from_another_connection + + elapsed = Timeout.timeout(5) { woken.pop } + waiter.join(5) + assert_operator elapsed, :<, 1.0, + "a cross-connection signal should wake the waiter well before the timeout" + end + + test "waiting returns after the timeout when nothing signals" do + @adapter.listen + started = monotonic_now + + @adapter.wait(timeout: 0.2) + + elapsed = monotonic_now - started + assert_operator elapsed, :>=, 0.15 + assert_operator elapsed, :<, 2.0 + end + + # 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 + threads = Array.new(waiters) do + Thread.new do + @adapter.listen + waiting << 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 + + 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 2, connections.size.times.map { connections.pop }.uniq.length + end + + test "stopping the supervisor releases listener connections" do + SolidObjects.configuration.wake_up_adapter = @adapter + SolidObjects.instance_variable_set(:@wake_up, nil) + @adapter.listen + supervisor = SolidObjects::Supervisor.new( + worker_count: 0, + effect_worker_count: 0, + broadcast_worker_count: 0, + reminder_scheduler_count: 0 + ) + supervisor.start + + supervisor.stop + + refute @adapter.send(:connections).any?, + "supervisor shutdown must release connections opened outside the pool" + ensure + SolidObjects.configuration.wake_up_adapter = nil + SolidObjects.instance_variable_set(:@wake_up, nil) + end + + test "a failing component shutdown still releases listener connections" do + SolidObjects.configuration.wake_up_adapter = @adapter + SolidObjects.instance_variable_set(:@wake_up, nil) + @adapter.listen + supervisor = SolidObjects::Supervisor.new( + worker_count: 1, + effect_worker_count: 0, + broadcast_worker_count: 0, + reminder_scheduler_count: 0 + ) + supervisor.start + supervisor.send(:components).each do |component| + component.define_singleton_method(:stop) { raise "boom" } + component.define_singleton_method(:stopped?) { false } + end + + assert_raises(RuntimeError) { supervisor.stop } + + refute @adapter.send(:connections).any?, + "a failed shutdown must still release connections opened outside the pool" + ensure + SolidObjects.configuration.wake_up_adapter = nil + SolidObjects.instance_variable_set(:@wake_up, nil) + end + + test "the helper selects notifications on PostgreSQL" do + assert_instance_of( + SolidObjects::WakeUpAdapters::Postgresql, + SolidObjects::WakeUpAdapters.for(SolidObjects::Record.connection) + ) + end + + test "signalling never raises into the caller" do + broken = SolidObjects::WakeUpAdapters::Postgresql.new(channel: "solid_objects_missing") + broken.define_singleton_method(:notify_channel) { raise "boom" } + + assert_nothing_raised { broken.signal } + ensure + broken&.stop + end + + test "waiting never raises into the caller" do + broken = SolidObjects::WakeUpAdapters::Postgresql.new + broken.define_singleton_method(:listening_connection) { raise "boom" } + started = monotonic_now + + assert_nothing_raised { broken.wait(timeout: 0.1) } + + assert_operator monotonic_now - started, :>=, 0.05, + "a failed wait must still pace itself rather than spin" + ensure + broken&.stop + end + + test "the adapter satisfies the wake-up contract" do + assert_respond_to @adapter, :signal + assert_respond_to @adapter, :wait + SolidObjects.configuration.wake_up_adapter = @adapter + + assert_same @adapter, SolidObjects.wake_up + ensure + SolidObjects.configuration.wake_up_adapter = nil + SolidObjects.instance_variable_set(:@wake_up, nil) + end + + private + + def postgresql? + SolidObjects::Record.connection.adapter_name.match?(/postgres/i) + end + + def monotonic_now + ::Process.clock_gettime(::Process::CLOCK_MONOTONIC) + end + + # Models a commit in a different process: a plain NOTIFY on its own connection. + def signal_from_another_connection + Thread.new { SolidObjects::WakeUpAdapters::Postgresql.new.signal }.join(5) + end +end diff --git a/test/unit/wake_up_adapters_test.rb b/test/unit/wake_up_adapters_test.rb new file mode 100644 index 0000000..86417ed --- /dev/null +++ b/test/unit/wake_up_adapters_test.rb @@ -0,0 +1,32 @@ +# frozen_string_literal: true + +require "test_helper" + +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")) + + assert_instance_of SolidObjects::WakeUpAdapters::Postgresql, adapter + end + + test "falls back to the in-process wake-up for MySQL" do + adapter = SolidObjects::WakeUpAdapters.for(Connection.new("Mysql2")) + + assert_instance_of SolidObjects::WakeUp, adapter + end + + test "falls back to the in-process wake-up for SQLite" do + adapter = SolidObjects::WakeUpAdapters.for(Connection.new("SQLite")) + + assert_instance_of SolidObjects::WakeUp, adapter + end + + test "the fallback satisfies the wake-up contract" do + adapter = SolidObjects::WakeUpAdapters.for(Connection.new("SQLite")) + + assert_respond_to adapter, :signal + assert_respond_to adapter, :wait + end +end