diff --git a/CHANGELOG.md b/CHANGELOG.md index 4f2c73d..544d057 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,28 @@ # Changelog +## 0.13.1 - 2026-08-16 + +- Back idle actor, effect, reminder, and broadcast polling off exponentially + from the configured fast interval to a new one-second idle ceiling. Any + processed work or wake-up resets the role immediately, and actor polling + remains capped by the lease-renewal interval. +- Expose each role's `current_polling_interval` and emit + `solid_objects.polling.interval_changed` instrumentation for every idle, + work, and wake-up transition. +- Warn once when live processes share the database without a configured + cross-process wake-up adapter. +- Make the in-process wake-up generation-aware so a signal committed between + an empty claim and the wait is not missed. PostgreSQL and Redis adapters now + expose the same watch contract. +- Add a reproducible four-role SQLite idle benchmark and repair the benchmark + schema setup for the current operation columns. +- **Behavior change:** `polling_interval` is now the fast interval after + activity, not a constant idle cadence. Existing explicit values back off to + `idle_polling_interval`, which defaults to one second. Set both options to + the same value to preserve a fixed cadence. Existing custom wake-up adapters + that return `nil` remain at the fast cadence until they return `false` for a + timeout and `true` for a notification. + ## 0.13.0 - 2026-08-15 - **Breaking:** make observables invalidation-only by default. An ordinary diff --git a/Gemfile.lock b/Gemfile.lock index 68e1420..caec902 100644 --- a/Gemfile.lock +++ b/Gemfile.lock @@ -1,7 +1,7 @@ PATH remote: . specs: - solid_objects (0.13.0) + solid_objects (0.13.1) actioncable (>= 8.0) actionpack (>= 8.0) actionview (>= 8.0) @@ -384,7 +384,7 @@ CHECKSUMS rubocop-rails-omakase (1.1.0) sha256=2af73ac8ee5852de2919abbd2618af9c15c19b512c4cfc1f9a5d3b6ef009109d ruby-progressbar (1.13.0) sha256=80fc9c47a9b640d6834e0dc7b3c94c9df37f08cb072b7761e4a71e22cff29b33 securerandom (0.4.1) sha256=cc5193d414a4341b6e225f0cb4446aceca8e50d5e1888743fac16987638ea0b1 - solid_objects (0.13.0) + solid_objects (0.13.1) sqlite3 (2.9.5-aarch64-linux-gnu) sha256=78075b6337d3d182c6d2b4691049ed45cd220826160c9ea18946bf6a1de200dc sqlite3 (2.9.5-aarch64-linux-musl) sha256=18c801185deb4adc01ddb281e8f672a39e3d1729979ca91e39439cd3eac0402d sqlite3 (2.9.5-arm-linux-gnu) sha256=1bdfca0c7d63998c60b0f4a8e3c8df2d33800ccc4abd2d612eddbbbc92a4c48b diff --git a/README.md b/README.md index 964d7b6..f668e36 100644 --- a/README.md +++ b/README.md @@ -1014,6 +1014,7 @@ Important defaults: | Setting | Default | | --- | ---: | | `polling_interval` | 0.1 seconds | +| `idle_polling_interval` | 1 second | | `sync_polling_interval` | 0.05 seconds | | `lease_duration` | 30 seconds | | `lease_renewal_interval` | 10 seconds | @@ -1038,6 +1039,14 @@ Payload, state, and result limits; retry delay; table prefix; logging; wake-up; broadcast; database; and authorization adapters are also configurable. Invalid lease intervals, component counts, and size limits fail fast at boot. +`polling_interval` is the fast interval after work or a wake-up. Consecutive +empty passes double it up to `idle_polling_interval`. Actor workers never wait +longer than `lease_renewal_interval`. Set the fast and idle values equal for a +fixed cadence. The default wake-up reaches only the current Ruby process; +configure PostgreSQL notifications or optional Redis Pub/Sub when separate +processes need low-latency delivery. The runtime warns once when it sees that +topology without an adapter. + ## Workers and operations `solid_objects start` runs actor, effect, reminder, and broadcast roles under @@ -1299,8 +1308,8 @@ Partially implemented: - the supervisor starts and drains roles but does not replace a crashed role or run periodic maintenance automatically; -- cross-process wake-up uses polling; PostgreSQL notifications and optional - Redis acceleration are not implemented; +- PostgreSQL notifications and optional Redis acceleration are implemented, + but adapter selection remains explicit and polling is the durable fallback; - live observable and component replacement work, while Turbo append actions remain future work; - local admission limits exist, but distributed rate limits and global diff --git a/benchmark/idle_polling.rb b/benchmark/idle_polling.rb new file mode 100644 index 0000000..45ebf5d --- /dev/null +++ b/benchmark/idle_polling.rb @@ -0,0 +1,98 @@ +# frozen_string_literal: true + +require "json" +require "rbconfig" +require_relative "support" + +class CountingWakeUp < SolidObjects::WakeUp + def initialize + super + @count_mutex = Mutex.new + @poll_count = 0 + end + + def wait(timeout:, generation: nil) + @count_mutex.synchronize { @poll_count += 1 } + super + end + + def reset_count + @count_mutex.synchronize { @poll_count = 0 } + end + + def poll_count + @count_mutex.synchronize { @poll_count } + end +end + +def measure(interval:, warmup:, duration:, wake_up:) + SolidObjects.configuration.polling_interval = interval + SolidObjects.configuration.idle_polling_interval = 1.0 + components = [ + SolidObjects::Worker.new, + SolidObjects::EffectExecutor.new, + SolidObjects::ReminderScheduler.new, + SolidObjects::BroadcastExecutor.new + ] + threads = components.map { |component| Thread.new { component.run } } + + sleep warmup + wake_up.reset_count + cpu_started_at = Process.times + wall_started_at = Process.clock_gettime(Process::CLOCK_MONOTONIC) + sleep duration + wall_elapsed = Process.clock_gettime(Process::CLOCK_MONOTONIC) - wall_started_at + cpu_finished_at = Process.times + cpu_elapsed = cpu_finished_at.utime + cpu_finished_at.stime - + cpu_started_at.utime - cpu_started_at.stime + poll_count = wake_up.poll_count + + { + polling_interval: interval, + idle_polling_interval: SolidObjects.configuration.idle_polling_interval, + current_intervals: components.map(&:current_polling_interval), + polls: poll_count, + polls_per_second: (poll_count / wall_elapsed).round(3), + idle_cpu_percent: ((cpu_elapsed / wall_elapsed) * 100).round(3) + } +ensure + components&.each(&:request_shutdown) + threads&.each { |thread| thread.join(2) } + components&.each(&:stop) +end + +intervals = ENV.fetch("INTERVALS", "0.02,0.1,0.5").split(",").map do |value| + Float(value).tap { |interval| raise ArgumentError, "intervals must be positive" unless interval.positive? } +end +warmup = Float(ENV.fetch("WARMUP", "3")) +duration = Float(ENV.fetch("DURATION", "10")) +raise ArgumentError, "warmup must be positive" unless warmup.positive? +raise ArgumentError, "duration must be positive" unless duration.positive? + +SolidObjectsBenchmark.setup +wake_up = CountingWakeUp.new +SolidObjects.configuration.wake_up_adapter = wake_up +database_version = ActiveRecord::Base.connection.select_value("SELECT sqlite_version()") +results = intervals.map { |interval| measure(interval:, warmup:, duration:, wake_up:) } +puts JSON.pretty_generate( + measured_at: Time.now.utc.iso8601, + package_version: SolidObjects::VERSION, + runtime: { + ruby: RUBY_DESCRIPTION, + platform: RUBY_PLATFORM, + cpu: RbConfig::CONFIG.fetch("host_cpu") + }, + database: { + adapter: "sqlite", + version: database_version, + path: SolidObjectsBenchmark::DATABASE_PATH + }, + methodology: { + roles: %w[actors effects reminders broadcasts], + warmup_seconds: warmup, + duration_seconds: duration, + cpu_percent: "process user plus system CPU time divided by wall time" + }, + results: +) +SolidObjectsBenchmark.teardown diff --git a/benchmark/support.rb b/benchmark/support.rb index a47182a..c2ae12a 100644 --- a/benchmark/support.rb +++ b/benchmark/support.rb @@ -402,8 +402,10 @@ def establish_connection def migrate require_relative "../db/migrate/20260805000000_create_solid_objects_tables" require_relative "../db/migrate/20260806000000_add_state_revision_to_solid_objects_instances" + require_relative "../db/migrate/20260813000000_rename_message_dispatch_columns" CreateSolidObjectsTables.new.migrate(:up) AddStateRevisionToSolidObjectsInstances.new.migrate(:up) + RenameMessageDispatchColumns.new.migrate(:up) end # @rbs () -> void diff --git a/docs/adr/0011-wake-up-strategy.md b/docs/adr/0011-wake-up-strategy.md index 49cdefa..4a133d7 100644 --- a/docs/adr/0011-wake-up-strategy.md +++ b/docs/adr/0011-wake-up-strategy.md @@ -11,6 +11,10 @@ Polling adds latency and database queries. PostgreSQL notifications are transact Database rows remain the only durable source of work and results. Wake-up adapters only prompt workers and synchronous waiters to re-query those rows. +Actor, effect, reminder, and broadcast roles double consecutive empty waits +from `polling_interval` to `idle_polling_interval`. Work and notifications reset +the wait immediately. Actor workers clamp the ceiling to +`lease_renewal_interval`. The interface supports: @@ -40,4 +44,6 @@ Timeout does not cancel durable work. - A reconnecting PostgreSQL listener must commit `LISTEN`, inspect current state, and then wait. - 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. - Notification payloads never contain actor arguments or results. diff --git a/docs/architecture.md b/docs/architecture.md index 96ba76e..1668251 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -365,8 +365,8 @@ it drains earlier messages and the target through the same activation and executor used by workers. If another process owns the actor, the caller waits for the row to become completed, rejected, dead-lettered, destroyed, or timed out. Every wait re-queries durable rows. The implemented wake-up interface -provides same-process signaling, bounded polling, and dependency injection. -PostgreSQL `LISTEN/NOTIFY` and optional Redis Pub/Sub are planned adapters. +provides generation-aware same-process signaling, adaptive bounded polling, +PostgreSQL `LISTEN/NOTIFY`, and optional Redis Pub/Sub. The normal path does not wait for a worker polling interval because the caller assists execution immediately. End-to-end latency still includes earlier diff --git a/docs/benchmarks.md b/docs/benchmarks.md index b1a1398..59cf623 100644 --- a/docs/benchmarks.md +++ b/docs/benchmarks.md @@ -5,6 +5,44 @@ They include the runtime's Active Record and database query overhead and will vary with hardware, schema size, connection pools, durability settings, and contention. +## Idle SQLite polling + +Run the four-role idle harness with: + +```bash +bundle exec ruby -Ilib benchmark/idle_polling.rb +``` + +It warms each interval for three seconds, measures for ten seconds, and reports +process user plus system CPU time divided by wall time. Measured August 16, +2026 on an Apple M5 with Ruby 4.0.6 and SQLite 3.53.2. The before run used +0.13.0; the after run used the prepared 0.13.1 tree. + +| Fast interval | Before polls/s | Before CPU | After polls/s | After CPU | +| ---: | ---: | ---: | ---: | ---: | +| 20 ms | 165.340 | 8.401% | 3.998 | 0.947% | +| 100 ms | 38.396 | 2.925% | 3.999 | 0.482% | +| 500 ms | 7.998 | 2.061% | 3.996 | 0.283% | + +The after run reached the one-second ceiling for the actor, effect, reminder, +and broadcast roles. These are developer-laptop measurements, not a CPU +guarantee; timer scheduling, YJIT, the SQLite file, and unrelated host activity +affect short samples. + +Five SQLite samples measured durable enqueue through committed completion after +2.5 seconds of idleness. The polling-only multi-process harness submits just +after an empty pass, so it measures approximately the full polling wait rather +than average arrival latency. + +| Topology | 0.13.0 p50 | Prepared 0.13.1 p50 | +| --- | ---: | ---: | +| One process, in-process wake-up | 43.360 ms | 50.339 ms | +| Two processes, polling only | 117.787 ms | 1,028.006 ms | + +The local wake-up keeps the one-process path prompt after backoff. The +polling-only row is the explicit tradeoff: use PostgreSQL notifications or +optional Redis Pub/Sub when separate processes need low-latency delivery. + ## Production-shaped adoption measurement An adoption evaluation measured Solid Objects 0.2.0 from a macOS Rails process diff --git a/docs/development.md b/docs/development.md index ee31eac..41568cd 100644 --- a/docs/development.md +++ b/docs/development.md @@ -140,6 +140,7 @@ COUNT=500 CONCURRENCY=4 bundle exec ruby -Ilib benchmark/concurrent_actors.rb COUNT=100 bundle exec ruby -Ilib benchmark/sync_latency.rb COUNT=500 bundle exec ruby -Ilib benchmark/activation_cache.rb bundle exec ruby -Ilib benchmark/query_count.rb +bundle exec ruby -Ilib benchmark/idle_polling.rb ``` SQLite is the default. Set `SOLID_OBJECTS_DATABASE_URL` to benchmark a dedicated diff --git a/docs/operations.md b/docs/operations.md index e5f58e1..032874e 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -69,6 +69,7 @@ Important controls include: - `lease_duration` - `lease_renewal_interval` - `polling_interval` +- `idle_polling_interval` - `max_mailbox_length` - payload, state, and result byte limits - retry attempts and delay @@ -81,6 +82,29 @@ Keep lease duration comfortably above renewal interval and expected database pause time. A handler can exceed the pass-duration budget because Ruby code is not safely preempted; alert on message duration and isolate untrusted work. +## Polling and wake-up adapters + +`polling_interval` is the fast interval after work or a wake-up. Consecutive +empty actor, effect, reminder, and broadcast passes double that role's wait up +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. + +Each role exposes `current_polling_interval`. +`solid_objects.polling.interval_changed` reports the role, reason, previous +interval, and current interval. The polling-only warning is also emitted as +`solid_objects.polling.only_cross_process_wake_up` instrumentation. Custom +adapters should return `true` for a notification and `false` for a timeout; an +older adapter that returns `nil` remains compatible and keeps the fast cadence. + ## Graceful shutdown The supervisor requests shutdown, stops new claims, lets active loops return, diff --git a/docs/roadmap.md b/docs/roadmap.md index e29138c..d2f2b45 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -83,11 +83,14 @@ 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 - `polling_interval`, 100 ms. An adapter removes that floor, measured 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. + 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 diff --git a/lib/solid_objects.rb b/lib/solid_objects.rb index e5983a6..4729969 100644 --- a/lib/solid_objects.rb +++ b/lib/solid_objects.rb @@ -52,6 +52,7 @@ require "solid_objects/wake_up_adapters/postgresql" require "solid_objects/wake_up_adapters/redis" require "solid_objects/wake_up_adapters" +require "solid_objects/polling_backoff" require "solid_objects/effect_registry" require "solid_objects/commit_action_registry" require "solid_objects/lease" @@ -153,6 +154,7 @@ def mutable_copy(value) # @rbs () -> void def reset! + ProcessRegistry.reset_polling_warning! if defined?(ProcessRegistry) @configuration = Configuration.new @registry = ActorRegistry.new @client = nil diff --git a/lib/solid_objects/broadcast_executor.rb b/lib/solid_objects/broadcast_executor.rb index 1710b1f..dd70af5 100644 --- a/lib/solid_objects/broadcast_executor.rb +++ b/lib/solid_objects/broadcast_executor.rb @@ -1,11 +1,14 @@ # rbs_inline: enabled +require "solid_objects/polling_backoff" + module SolidObjects class BroadcastExecutor # @rbs @process_registry: ProcessRegistry # @rbs @database_adapter: DatabaseAdapter # @rbs @stopped: bool # @rbs @shutdown_requested: bool + # @rbs @polling_backoff: PollingBackoff # @rbs (?process_registry: ProcessRegistry, ?database_adapter: DatabaseAdapter) -> void def initialize( @@ -17,6 +20,17 @@ def initialize( process_registry.register(kind: "broadcast") @stopped = false @shutdown_requested = false + @polling_backoff = PollingBackoff.new( + minimum_interval: SolidObjects.configuration.polling_interval, + maximum_interval: SolidObjects.configuration.idle_polling_interval, + on_change: ->(transition) do + SolidObjects.instrument( + :"polling.interval_changed", + role: "broadcasts", + **transition + ) + end + ) end # @rbs () -> bool @@ -46,11 +60,23 @@ def stop # @rbs () -> void def run + ProcessRegistry.warn_if_polling_is_only_cross_process_wake_up + until shutdown_requested? + wake_up = SolidObjects.wake_up + watch = wake_up.respond_to?(:watch) ? wake_up.watch : wake_up worked = run_once - next if worked - - SolidObjects.wake_up.wait(timeout: SolidObjects.configuration.polling_interval) + if worked + polling_backoff.reset(:work) + next + end + + notified = watch.wait(timeout: current_polling_interval) + if notified == false + polling_backoff.record_idle + else + polling_backoff.reset(:wake_up) + end end ensure stop @@ -72,9 +98,14 @@ def shutdown_requested? @shutdown_requested end + # @rbs () -> Float + def current_polling_interval + polling_backoff.current_interval + end + private - attr_reader :process_registry, :database_adapter + attr_reader :process_registry, :database_adapter, :polling_backoff # @rbs () -> Broadcast? def claim_next diff --git a/lib/solid_objects/configuration.rb b/lib/solid_objects/configuration.rb index 45e8939..e7b6678 100644 --- a/lib/solid_objects/configuration.rb +++ b/lib/solid_objects/configuration.rb @@ -4,6 +4,7 @@ module SolidObjects class Configuration # @rbs @table_name_prefix: String # @rbs @polling_interval: Float + # @rbs @idle_polling_interval: Float # @rbs @sync_polling_interval: Float # @rbs @lease_duration: Float # @rbs @lease_renewal_interval: Float @@ -49,6 +50,7 @@ class Configuration attr_accessor :table_name_prefix, :polling_interval, + :idle_polling_interval, :sync_polling_interval, :lease_duration, :lease_renewal_interval, @@ -99,6 +101,7 @@ class Configuration def initialize @table_name_prefix = "solid_objects_" @polling_interval = 0.1 + @idle_polling_interval = 1.0 @sync_polling_interval = 0.05 @lease_duration = 30.0 @lease_renewal_interval = 10.0 @@ -241,6 +244,7 @@ def validate! def positive_values { polling_interval:, + idle_polling_interval:, sync_polling_interval:, lease_duration:, lease_renewal_interval:, diff --git a/lib/solid_objects/effect_executor.rb b/lib/solid_objects/effect_executor.rb index 8f6dbec..d6f4aa9 100644 --- a/lib/solid_objects/effect_executor.rb +++ b/lib/solid_objects/effect_executor.rb @@ -1,5 +1,7 @@ # rbs_inline: enabled +require "solid_objects/polling_backoff" + module SolidObjects EffectContext = Data.define(:id, :attempt, :source_message_id, :actor_type, :actor_id) @@ -10,6 +12,7 @@ class EffectExecutor # @rbs @database_adapter: DatabaseAdapter # @rbs @stopped: bool # @rbs @shutdown_requested: bool + # @rbs @polling_backoff: PollingBackoff # @rbs (?process_registry: ProcessRegistry, ?database_adapter: DatabaseAdapter) -> void def initialize( @@ -21,6 +24,17 @@ def initialize( process_registry.register(kind: "effect") @stopped = false @shutdown_requested = false + @polling_backoff = PollingBackoff.new( + minimum_interval: SolidObjects.configuration.polling_interval, + maximum_interval: SolidObjects.configuration.idle_polling_interval, + on_change: ->(transition) do + SolidObjects.instrument( + :"polling.interval_changed", + role: "effects", + **transition + ) + end + ) end # @rbs () -> bool @@ -50,11 +64,23 @@ def stop # @rbs () -> void def run + ProcessRegistry.warn_if_polling_is_only_cross_process_wake_up + until shutdown_requested? + wake_up = SolidObjects.wake_up + watch = wake_up.respond_to?(:watch) ? wake_up.watch : wake_up worked = run_once - next if worked + if worked + polling_backoff.reset(:work) + next + end - SolidObjects.wake_up.wait(timeout: SolidObjects.configuration.polling_interval) + notified = watch.wait(timeout: current_polling_interval) + if notified == false + polling_backoff.record_idle + else + polling_backoff.reset(:wake_up) + end end ensure stop @@ -76,9 +102,14 @@ def shutdown_requested? @shutdown_requested end + # @rbs () -> Float + def current_polling_interval + polling_backoff.current_interval + end + private - attr_reader :process_registry, :database_adapter + attr_reader :process_registry, :database_adapter, :polling_backoff # @rbs () -> Effect? def claim_next diff --git a/lib/solid_objects/polling_backoff.rb b/lib/solid_objects/polling_backoff.rb new file mode 100644 index 0000000..45a629f --- /dev/null +++ b/lib/solid_objects/polling_backoff.rb @@ -0,0 +1,45 @@ +# rbs_inline: enabled + +module SolidObjects + class PollingBackoff + # @rbs @minimum_interval: Float + # @rbs @maximum_interval: Float + # @rbs @current_interval: Float + # @rbs @on_change: Proc? + + attr_reader :current_interval + + # @rbs (minimum_interval: Numeric, maximum_interval: Numeric, ?on_change: Proc?) -> void + def initialize(minimum_interval:, maximum_interval:, on_change: nil) + @minimum_interval = minimum_interval.to_f + @maximum_interval = [ @minimum_interval, maximum_interval.to_f ].max + @current_interval = @minimum_interval + @on_change = on_change + end + + # @rbs () -> void + def record_idle + change([ current_interval * 2, @maximum_interval ].min, :idle) + end + + # @rbs (:work | :wake_up) -> void + def reset(reason) + change(@minimum_interval, reason) + end + + private + + # @rbs (Float, :idle | :work | :wake_up) -> void + def change(interval, reason) + return if interval == current_interval + + previous_interval = current_interval + @current_interval = interval + @on_change&.call( + previous_interval:, + current_interval: interval, + reason: + ) + end + end +end diff --git a/lib/solid_objects/process_registry.rb b/lib/solid_objects/process_registry.rb index 27caa6a..4a69f3d 100644 --- a/lib/solid_objects/process_registry.rb +++ b/lib/solid_objects/process_registry.rb @@ -4,6 +4,9 @@ module SolidObjects class ProcessRegistry + @polling_warning_mutex = Mutex.new + @polling_warning_emitted = false + class << self # @rbs (?now: Time) -> Integer def cleanup_dead(now: SolidObjects.database_adapter.database_now) @@ -53,8 +56,56 @@ def deregister(process_record, now: SolidObjects.database_adapter.database_now) true end + # @rbs () -> void + def warn_if_polling_is_only_cross_process_wake_up + return if SolidObjects.configuration.wake_up_adapter + + polling_warning_mutex.synchronize do + return if polling_warning_emitted? + return unless another_live_process? + + payload = { + event: "solid_objects.polling_only_cross_process_wake_up", + polling_interval: SolidObjects.configuration.polling_interval, + idle_polling_interval: SolidObjects.configuration.idle_polling_interval + } + SolidObjects.configuration.logger.warn(payload) + SolidObjects.instrument( + :"polling.only_cross_process_wake_up", + **payload.except(:event) + ) + @polling_warning_emitted = true + end + end + + # @rbs () -> void + def reset_polling_warning! + polling_warning_mutex.synchronize { @polling_warning_emitted = false } + end + private + # @rbs () -> bool + def another_live_process? + alive_after = SolidObjects.database_adapter.database_now - + SolidObjects.configuration.process_alive_threshold + Process + .where.not(shutdown_state: "stopped") + .where(last_heartbeat_at: alive_after..) + .where("hostname <> ? OR pid <> ?", Socket.gethostname, ::Process.pid) + .exists? + end + + # @rbs () -> bool + def polling_warning_emitted? + @polling_warning_emitted + end + + # @rbs () -> Mutex + def polling_warning_mutex + @polling_warning_mutex ||= Mutex.new + end + # @rbs (Process, Time) -> void def cleanup_process(process_record, now) deregister(process_record, now:) diff --git a/lib/solid_objects/reminder_scheduler.rb b/lib/solid_objects/reminder_scheduler.rb index 0a3a132..5236c1c 100644 --- a/lib/solid_objects/reminder_scheduler.rb +++ b/lib/solid_objects/reminder_scheduler.rb @@ -1,11 +1,14 @@ # rbs_inline: enabled +require "solid_objects/polling_backoff" + module SolidObjects class ReminderScheduler # @rbs @process_registry: ProcessRegistry # @rbs @database_adapter: DatabaseAdapter # @rbs @stopped: bool # @rbs @shutdown_requested: bool + # @rbs @polling_backoff: PollingBackoff # @rbs (?process_registry: ProcessRegistry, ?database_adapter: DatabaseAdapter) -> void def initialize( @@ -17,6 +20,17 @@ def initialize( process_registry.register(kind: "reminder") @stopped = false @shutdown_requested = false + @polling_backoff = PollingBackoff.new( + minimum_interval: SolidObjects.configuration.polling_interval, + maximum_interval: SolidObjects.configuration.idle_polling_interval, + on_change: ->(transition) do + SolidObjects.instrument( + :"polling.interval_changed", + role: "reminders", + **transition + ) + end + ) end # @rbs (?now: Time?) -> bool @@ -45,11 +59,23 @@ def stop # @rbs () -> void def run + ProcessRegistry.warn_if_polling_is_only_cross_process_wake_up + until shutdown_requested? + wake_up = SolidObjects.wake_up + watch = wake_up.respond_to?(:watch) ? wake_up.watch : wake_up worked = run_once - next if worked - - SolidObjects.wake_up.wait(timeout: SolidObjects.configuration.polling_interval) + if worked + polling_backoff.reset(:work) + next + end + + notified = watch.wait(timeout: current_polling_interval) + if notified == false + polling_backoff.record_idle + else + polling_backoff.reset(:wake_up) + end end ensure stop @@ -71,9 +97,14 @@ def shutdown_requested? @shutdown_requested end + # @rbs () -> Float + def current_polling_interval + polling_backoff.current_interval + end + private - attr_reader :process_registry, :database_adapter + attr_reader :process_registry, :database_adapter, :polling_backoff # @rbs (now: Time?) -> Reminder? def claim_next(now:) diff --git a/lib/solid_objects/version.rb b/lib/solid_objects/version.rb index ac26020..457e47f 100644 --- a/lib/solid_objects/version.rb +++ b/lib/solid_objects/version.rb @@ -1,5 +1,5 @@ # rbs_inline: enabled module SolidObjects - VERSION = "0.13.0" + VERSION = "0.13.1" end diff --git a/lib/solid_objects/wake_up.rb b/lib/solid_objects/wake_up.rb index 50805b4..2f0f38a 100644 --- a/lib/solid_objects/wake_up.rb +++ b/lib/solid_objects/wake_up.rb @@ -2,23 +2,55 @@ module SolidObjects class WakeUp + class Watch + # @rbs @wake_up: WakeUp + # @rbs @generation: Integer + + # @rbs (WakeUp, Integer) -> void + def initialize(wake_up, generation) + @wake_up = wake_up + @generation = generation + end + + # @rbs (timeout: Numeric) -> bool + def wait(timeout:) + @wake_up.wait(timeout:, generation: @generation) + end + end + # @rbs @mutex: Mutex # @rbs @condition: Thread::ConditionVariable + # @rbs @generation: Integer # @rbs () -> void def initialize @mutex = Mutex.new @condition = Thread::ConditionVariable.new + @generation = 0 end # @rbs () -> void def signal - mutex.synchronize { condition.broadcast } + mutex.synchronize do + @generation += 1 + condition.broadcast + end end - # @rbs (timeout: Numeric) -> void - def wait(timeout:) - mutex.synchronize { condition.wait(mutex, timeout) } + # @rbs () -> Watch + def watch + mutex.synchronize { Watch.new(self, @generation) } + end + + # @rbs (timeout: Numeric, ?generation: Integer?) -> bool + def wait(timeout:, generation: nil) + mutex.synchronize do + return true if generation && @generation != generation + + generation ||= @generation + condition.wait(mutex, timeout) + @generation != generation + end end private diff --git a/lib/solid_objects/wake_up_adapters/postgresql.rb b/lib/solid_objects/wake_up_adapters/postgresql.rb index 47a8bcc..f66469c 100644 --- a/lib/solid_objects/wake_up_adapters/postgresql.rb +++ b/lib/solid_objects/wake_up_adapters/postgresql.rb @@ -45,6 +45,12 @@ def wait(timeout:) false end + # @rbs () -> self + def watch + listen + self + end + # Starts listening before a caller blocks, so a notification sent between # startup and the first wait is not missed. # @rbs () -> bool diff --git a/lib/solid_objects/wake_up_adapters/redis.rb b/lib/solid_objects/wake_up_adapters/redis.rb index 56d6f18..2f3b7e5 100644 --- a/lib/solid_objects/wake_up_adapters/redis.rb +++ b/lib/solid_objects/wake_up_adapters/redis.rb @@ -12,6 +12,22 @@ module WakeUpAdapters # polling interval remains the upper bound, so a missed or failed # notification costs latency rather than correctness. class Redis + class Watch + # @rbs @adapter: Redis + # @rbs @generation: Integer + + # @rbs (Redis, Integer) -> void + def initialize(adapter, generation) + @adapter = adapter + @generation = generation + end + + # @rbs (timeout: Numeric) -> bool + def wait(timeout:) + @adapter.wait(timeout:, generation: @generation) + end + end + CHANNEL = "solid_objects_wake_up" FAILED_WAIT_INTERVAL = 0.05 SUBSCRIBE_TIMEOUT = 5.0 @@ -52,9 +68,9 @@ def signal # The counter is snapshotted before subscribing, and re-checked before # blocking, so a signal delivered while this caller was still getting # ready is observed rather than absorbed into the new baseline. - # @rbs (timeout: Numeric) -> bool - def wait(timeout:) - signalled = mutex.synchronize { @signalled } + # @rbs (timeout: Numeric, ?generation: Integer?) -> bool + def wait(timeout:, generation: nil) + signalled = generation || mutex.synchronize { @signalled } return paced_failure(timeout) unless listen mutex.synchronize do @@ -65,6 +81,12 @@ def wait(timeout:) end end + # @rbs () -> Watch + def watch + listen + mutex.synchronize { Watch.new(self, @signalled) } + end + # Redis delivers to a subscribed connection only, and a subscribed # connection cannot serve other callers, so one background subscription # per process fans out to every waiting role in memory. Subscribing diff --git a/lib/solid_objects/worker.rb b/lib/solid_objects/worker.rb index 3d35dc6..7f0e53d 100644 --- a/lib/solid_objects/worker.rb +++ b/lib/solid_objects/worker.rb @@ -5,6 +5,7 @@ require "solid_objects/activation" require "solid_objects/executor" require "solid_objects/lease_renewer" +require "solid_objects/polling_backoff" module SolidObjects class Worker @@ -13,6 +14,7 @@ class Worker # @rbs @activations: Hash[Integer, Activation] # @rbs @stopped: bool # @rbs @shutdown_requested: bool + # @rbs @polling_backoff: PollingBackoff # @rbs (?process_registry: ProcessRegistry) -> void def initialize(process_registry: ProcessRegistry.new) @@ -22,6 +24,23 @@ def initialize(process_registry: ProcessRegistry.new) @activations = {} @stopped = false @shutdown_requested = false + @polling_backoff = PollingBackoff.new( + minimum_interval: [ + SolidObjects.configuration.polling_interval, + SolidObjects.configuration.lease_renewal_interval + ].min, + maximum_interval: [ + SolidObjects.configuration.idle_polling_interval, + SolidObjects.configuration.lease_renewal_interval + ].min, + on_change: ->(transition) do + SolidObjects.instrument( + :"polling.interval_changed", + role: "actors", + **transition + ) + end + ) end # @rbs () -> Integer @@ -70,11 +89,23 @@ def run_until_idle(max_passes: 1_000) # @rbs () -> void def run + ProcessRegistry.warn_if_polling_is_only_cross_process_wake_up + until shutdown_requested? + wake_up = SolidObjects.wake_up + watch = wake_up.respond_to?(:watch) ? wake_up.watch : wake_up processed = run_once - next if processed.positive? + if processed.positive? + polling_backoff.reset(:work) + next + end - SolidObjects.wake_up.wait(timeout: SolidObjects.configuration.polling_interval) + notified = watch.wait(timeout: current_polling_interval) + if notified == false + polling_backoff.record_idle + else + polling_backoff.reset(:wake_up) + end end ensure stop @@ -107,9 +138,14 @@ def shutdown_requested? @shutdown_requested end + # @rbs () -> Float + def current_polling_interval + polling_backoff.current_interval + end + private - attr_reader :process_registry, :activation_manager, :activations + attr_reader :process_registry, :activation_manager, :activations, :polling_backoff # @rbs () -> Activation? def cached_ready_activation diff --git a/sig/generated/lib/solid_objects/broadcast_executor.rbs b/sig/generated/lib/solid_objects/broadcast_executor.rbs index de0a737..e2776be 100644 --- a/sig/generated/lib/solid_objects/broadcast_executor.rbs +++ b/sig/generated/lib/solid_objects/broadcast_executor.rbs @@ -2,6 +2,8 @@ module SolidObjects class BroadcastExecutor + @polling_backoff: PollingBackoff + @shutdown_requested: bool @stopped: bool @@ -31,12 +33,17 @@ module SolidObjects # @rbs () -> bool def shutdown_requested?: () -> bool + # @rbs () -> Float + def current_polling_interval: () -> Float + private attr_reader process_registry: untyped attr_reader database_adapter: untyped + attr_reader polling_backoff: untyped + # @rbs () -> Broadcast? def claim_next: () -> Broadcast? diff --git a/sig/generated/lib/solid_objects/configuration.rbs b/sig/generated/lib/solid_objects/configuration.rbs index fcf749e..8432232 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 + @table_name_prefix: String + @process_alive_threshold: Float @shutdown_timeout: Float @@ -56,10 +58,10 @@ module SolidObjects @authorize_administration: Proc - @table_name_prefix: String - @polling_interval: Float + @idle_polling_interval: Float + @sync_polling_interval: Float @lease_duration: Float @@ -94,6 +96,8 @@ module SolidObjects attr_accessor polling_interval: untyped + attr_accessor idle_polling_interval: untyped + attr_accessor sync_polling_interval: untyped attr_accessor lease_duration: untyped diff --git a/sig/generated/lib/solid_objects/effect_executor.rbs b/sig/generated/lib/solid_objects/effect_executor.rbs index d258d18..0dc4c4c 100644 --- a/sig/generated/lib/solid_objects/effect_executor.rbs +++ b/sig/generated/lib/solid_objects/effect_executor.rbs @@ -23,6 +23,8 @@ module SolidObjects class EffectExecutor ACTOR_MESSAGE_EFFECT: ::String + @polling_backoff: PollingBackoff + @shutdown_requested: bool @stopped: bool @@ -52,12 +54,17 @@ module SolidObjects # @rbs () -> bool def shutdown_requested?: () -> bool + # @rbs () -> Float + def current_polling_interval: () -> Float + private attr_reader process_registry: untyped attr_reader database_adapter: untyped + attr_reader polling_backoff: untyped + # @rbs () -> Effect? def claim_next: () -> Effect? diff --git a/sig/generated/lib/solid_objects/polling_backoff.rbs b/sig/generated/lib/solid_objects/polling_backoff.rbs new file mode 100644 index 0000000..3d68f85 --- /dev/null +++ b/sig/generated/lib/solid_objects/polling_backoff.rbs @@ -0,0 +1,29 @@ +# Generated from lib/solid_objects/polling_backoff.rb with RBS::Inline + +module SolidObjects + class PollingBackoff + @minimum_interval: Float + + @maximum_interval: Float + + @current_interval: Float + + @on_change: Proc? + + attr_reader current_interval: untyped + + # @rbs (minimum_interval: Numeric, maximum_interval: Numeric, ?on_change: Proc?) -> void + def initialize: (minimum_interval: Numeric, maximum_interval: Numeric, ?on_change: Proc?) -> void + + # @rbs () -> void + def record_idle: () -> void + + # @rbs (:work | :wake_up) -> void + def reset: (:work | :wake_up) -> void + + private + + # @rbs (Float, :idle | :work | :wake_up) -> void + def change: (Float, :idle | :work | :wake_up) -> void + end +end diff --git a/sig/generated/lib/solid_objects/process_registry.rbs b/sig/generated/lib/solid_objects/process_registry.rbs index 02c3657..f166f13 100644 --- a/sig/generated/lib/solid_objects/process_registry.rbs +++ b/sig/generated/lib/solid_objects/process_registry.rbs @@ -8,6 +8,21 @@ module SolidObjects # @rbs (Process, ?now: Time) -> bool def self.deregister: (Process, ?now: Time) -> bool + # @rbs () -> void + def self.warn_if_polling_is_only_cross_process_wake_up: () -> void + + # @rbs () -> void + def self.reset_polling_warning!: () -> void + + # @rbs () -> bool + private def self.another_live_process?: () -> bool + + # @rbs () -> bool + private def self.polling_warning_emitted?: () -> bool + + # @rbs () -> Mutex + private def self.polling_warning_mutex: () -> Mutex + # @rbs (Process, Time) -> void private def self.cleanup_process: (Process, Time) -> void diff --git a/sig/generated/lib/solid_objects/reminder_scheduler.rbs b/sig/generated/lib/solid_objects/reminder_scheduler.rbs index a516f3d..7200313 100644 --- a/sig/generated/lib/solid_objects/reminder_scheduler.rbs +++ b/sig/generated/lib/solid_objects/reminder_scheduler.rbs @@ -2,6 +2,8 @@ module SolidObjects class ReminderScheduler + @polling_backoff: PollingBackoff + @shutdown_requested: bool @stopped: bool @@ -31,12 +33,17 @@ module SolidObjects # @rbs () -> bool def shutdown_requested?: () -> bool + # @rbs () -> Float + def current_polling_interval: () -> Float + private attr_reader process_registry: untyped attr_reader database_adapter: untyped + attr_reader polling_backoff: untyped + # @rbs (now: Time?) -> Reminder? def claim_next: (now: Time?) -> Reminder? diff --git a/sig/generated/lib/solid_objects/wake_up.rbs b/sig/generated/lib/solid_objects/wake_up.rbs index bf144ef..6419976 100644 --- a/sig/generated/lib/solid_objects/wake_up.rbs +++ b/sig/generated/lib/solid_objects/wake_up.rbs @@ -2,18 +2,35 @@ module SolidObjects class WakeUp + class Watch + @wake_up: WakeUp + + @generation: Integer + + # @rbs (WakeUp, Integer) -> void + def initialize: (WakeUp, Integer) -> void + + # @rbs (timeout: Numeric) -> bool + def wait: (timeout: Numeric) -> bool + end + @mutex: Mutex @condition: Thread::ConditionVariable + @generation: Integer + # @rbs () -> void def initialize: () -> void # @rbs () -> void def signal: () -> void - # @rbs (timeout: Numeric) -> void - def wait: (timeout: Numeric) -> void + # @rbs () -> Watch + def watch: () -> Watch + + # @rbs (timeout: Numeric, ?generation: Integer?) -> bool + def wait: (timeout: Numeric, ?generation: Integer?) -> bool private 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 f363278..655bae3 100644 --- a/sig/generated/lib/solid_objects/wake_up_adapters/postgresql.rbs +++ b/sig/generated/lib/solid_objects/wake_up_adapters/postgresql.rbs @@ -31,6 +31,9 @@ module SolidObjects # @rbs (timeout: Numeric) -> bool def wait: (timeout: Numeric) -> bool + # @rbs () -> self + def watch: () -> self + # Starts listening before a caller blocks, so a notification sent between # startup and the first wait is not missed. # @rbs () -> bool 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 7ae2a98..4b2ef54 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,18 @@ module SolidObjects # polling interval remains the upper bound, so a missed or failed # notification costs latency rather than correctness. class Redis + class Watch + @adapter: Redis + + @generation: Integer + + # @rbs (Redis, Integer) -> void + def initialize: (Redis, Integer) -> void + + # @rbs (timeout: Numeric) -> bool + def wait: (timeout: Numeric) -> bool + end + CHANNEL: ::String FAILED_WAIT_INTERVAL: ::Float @@ -43,8 +55,11 @@ module SolidObjects # The counter is snapshotted before subscribing, and re-checked before # blocking, so a signal delivered while this caller was still getting # ready is observed rather than absorbed into the new baseline. - # @rbs (timeout: Numeric) -> bool - def wait: (timeout: Numeric) -> bool + # @rbs (timeout: Numeric, ?generation: Integer?) -> bool + def wait: (timeout: Numeric, ?generation: Integer?) -> bool + + # @rbs () -> Watch + def watch: () -> Watch # Redis delivers to a subscribed connection only, and a subscribed # connection cannot serve other callers, so one background subscription diff --git a/sig/generated/lib/solid_objects/worker.rbs b/sig/generated/lib/solid_objects/worker.rbs index 2ba1e3e..c2e7798 100644 --- a/sig/generated/lib/solid_objects/worker.rbs +++ b/sig/generated/lib/solid_objects/worker.rbs @@ -2,6 +2,8 @@ module SolidObjects class Worker + @polling_backoff: PollingBackoff + @shutdown_requested: bool @stopped: bool @@ -36,6 +38,9 @@ module SolidObjects # @rbs () -> bool def shutdown_requested?: () -> bool + # @rbs () -> Float + def current_polling_interval: () -> Float + private attr_reader process_registry: untyped @@ -44,6 +49,8 @@ module SolidObjects attr_reader activations: untyped + attr_reader polling_backoff: untyped + # @rbs () -> Activation? def cached_ready_activation: () -> Activation? diff --git a/test/integration/polling_test.rb b/test/integration/polling_test.rb new file mode 100644 index 0000000..6ec021d --- /dev/null +++ b/test/integration/polling_test.rb @@ -0,0 +1,272 @@ +# frozen_string_literal: true + +require "database_test_helper" +require "solid_objects/polling_backoff" +require "timeout" + +class PollingTest < ActiveSupport::TestCase + class WakeLatencyActor < SolidObjects::Actor + actor_type "polling-wake-latency" + + attribute :count, default: 0 + + def increment + self.count += 1 + end + end + + class RecordingLogger + attr_reader :warnings + + def initialize + @warnings = [] + end + + def warn(entry) + warnings << entry + end + end + + class ImmediateTimeoutWakeUp + attr_reader :intervals + attr_accessor :on_wait + + def initialize + @intervals = [] + @on_wait = method(:itself) + end + + def wait(timeout:) + intervals << timeout + on_wait.call + false + end + + def signal + end + end + + class LegacyWakeUp + attr_reader :intervals + attr_accessor :on_wait + + def initialize + @intervals = [] + @on_wait = method(:itself) + end + + def wait(timeout:) + intervals << timeout + on_wait.call + nil + end + + def signal + end + end + + class SnapshotWakeUp + class Watch + def initialize(adapter) + @adapter = adapter + end + + def wait(timeout:) + @adapter.watched_wait(timeout:) + end + end + + attr_reader :events + attr_accessor :on_wait + + def initialize + @events = [] + @on_wait = method(:itself) + end + + def watch + events << :watch + Watch.new(self) + end + + def wait(timeout:) + raise "runtime role waited without a wake-up snapshot" + end + + def watched_wait(timeout:) + events << :wait + on_wait.call + false + end + + def signal + end + end + + test "backs an idle worker off to the configured ceiling and reports each transition" do + wake_up = ImmediateTimeoutWakeUp.new + SolidObjects.configuration.polling_interval = 0.025 + SolidObjects.configuration.idle_polling_interval = 1.0 + SolidObjects.configuration.wake_up_adapter = wake_up + events = [] + subscription = ActiveSupport::Notifications.subscribe( + "solid_objects.polling.interval_changed" + ) { |event| events << event.payload } + worker = SolidObjects::Worker.new + wake_up.on_wait = -> { worker.request_shutdown if wake_up.intervals.length == 7 } + + worker.run + + assert_equal [ 0.025, 0.05, 0.1, 0.2, 0.4, 0.8, 1.0 ], wake_up.intervals + assert_equal 1.0, worker.current_polling_interval + assert_equal({ + role: "actors", + reason: :idle, + previous_interval: 0.8, + current_interval: 1.0 + }, events.last) + ensure + ActiveSupport::Notifications.unsubscribe(subscription) if subscription + worker&.stop + end + + test "backs idle effect, reminder, and broadcast roles off to the configured ceiling" do + component = nil + { + SolidObjects::EffectExecutor => "effects", + SolidObjects::ReminderScheduler => "reminders", + SolidObjects::BroadcastExecutor => "broadcasts" + }.each do |component_class, role| + wake_up = ImmediateTimeoutWakeUp.new + SolidObjects.configuration.polling_interval = 0.025 + SolidObjects.configuration.idle_polling_interval = 1.0 + SolidObjects.configuration.wake_up_adapter = wake_up + SolidObjects.instance_variable_set(:@wake_up, nil) + component = component_class.new + wake_up.on_wait = -> { component.request_shutdown if wake_up.intervals.length == 7 } + + component.run + + assert_equal [ 0.025, 0.05, 0.1, 0.2, 0.4, 0.8, 1.0 ], wake_up.intervals, role + assert_equal 1.0, component.current_polling_interval, role + component.stop + end + ensure + component&.stop + end + + test "never backs an actor worker off beyond its lease renewal interval" do + wake_up = ImmediateTimeoutWakeUp.new + SolidObjects.configuration.polling_interval = 0.025 + SolidObjects.configuration.idle_polling_interval = 1.0 + SolidObjects.configuration.lease_duration = 0.3 + SolidObjects.configuration.lease_renewal_interval = 0.1 + SolidObjects.configuration.wake_up_adapter = wake_up + worker = SolidObjects::Worker.new + wake_up.on_wait = -> { worker.request_shutdown if wake_up.intervals.length == 5 } + + worker.run + + assert_equal [ 0.025, 0.05, 0.1, 0.1, 0.1 ], wake_up.intervals + ensure + worker&.stop + end + + test "keeps a legacy wake-up adapter at the fast interval" do + wake_up = LegacyWakeUp.new + SolidObjects.configuration.polling_interval = 0.025 + SolidObjects.configuration.idle_polling_interval = 1.0 + SolidObjects.configuration.wake_up_adapter = wake_up + worker = SolidObjects::Worker.new + wake_up.on_wait = -> { worker.request_shutdown if wake_up.intervals.length == 3 } + + worker.run + + assert_equal [ 0.025, 0.025, 0.025 ], wake_up.intervals + ensure + worker&.stop + end + + test "captures wake-up state before checking for actor work" do + wake_up = SnapshotWakeUp.new + SolidObjects.configuration.wake_up_adapter = wake_up + worker = SolidObjects::Worker.new + wake_up.on_wait = -> { worker.request_shutdown } + + worker.run + + assert_equal [ :watch, :wait ], wake_up.events + ensure + worker&.stop + end + + test "processes local work promptly after reaching the idle ceiling" do + SolidObjects.configuration.polling_interval = 0.025 + SolidObjects.configuration.idle_polling_interval = 1.0 + worker = SolidObjects::Worker.new + worker_thread = Thread.new { worker.run } + Timeout.timeout(3) do + sleep 0.005 until worker.current_polling_interval >= 1.0 + end + + started_at = Process.clock_gettime(Process::CLOCK_MONOTONIC) + message = WakeLatencyActor.ref("local").async.increment + Timeout.timeout(0.4) do + sleep 0.005 until message.status == "completed" + end + elapsed = Process.clock_gettime(Process::CLOCK_MONOTONIC) - started_at + + assert_operator elapsed, :<, 0.4 + ensure + worker&.request_shutdown + worker_thread&.join(2) + worker&.stop + end + + test "warns once when another process shares the database without a wake-up adapter" do + logger = RecordingLogger.new + SolidObjects.configuration.logger = logger + SolidObjects.configuration.polling_interval = 0.025 + SolidObjects.configuration.idle_polling_interval = 1.0 + SolidObjects::Process.create!( + id: SecureRandom.uuid, + kind: "worker", + hostname: "another-host", + pid: ::Process.pid + 1, + started_at: Time.current, + last_heartbeat_at: Time.current, + metadata: {} + ) + workers = [ SolidObjects::Worker.new, SolidObjects::Worker.new ] + workers.each(&:request_shutdown) + + workers.each(&:run) + + assert_equal [ { + event: "solid_objects.polling_only_cross_process_wake_up", + polling_interval: 0.025, + idle_polling_interval: 1.0 + } ], logger.warnings + ensure + workers&.each(&:stop) + end + + test "does not warn when a cross-process wake-up adapter is configured" do + logger = RecordingLogger.new + SolidObjects.configuration.logger = logger + SolidObjects.configuration.wake_up_adapter = ImmediateTimeoutWakeUp.new + SolidObjects::Process.create!( + id: SecureRandom.uuid, + kind: "worker", + hostname: "another-host", + pid: ::Process.pid + 1, + started_at: Time.current, + last_heartbeat_at: Time.current, + metadata: {} + ) + + SolidObjects::ProcessRegistry.warn_if_polling_is_only_cross_process_wake_up + + assert_empty logger.warnings + end +end diff --git a/test/integration/postgresql_wake_up_test.rb b/test/integration/postgresql_wake_up_test.rb index ad0af08..bacb78d 100644 --- a/test/integration/postgresql_wake_up_test.rb +++ b/test/integration/postgresql_wake_up_test.rb @@ -36,13 +36,22 @@ class PostgresqlWakeUpTest < ActiveSupport::TestCase @adapter.listen started = monotonic_now - @adapter.wait(timeout: 0.2) + notified = @adapter.wait(timeout: 0.2) elapsed = monotonic_now - started + assert_equal false, notified assert_operator elapsed, :>=, 0.15 assert_operator elapsed, :<, 2.0 end + test "a watch observes a signal delivered before waiting" do + watch = @adapter.watch + + signal_from_another_connection + + assert_equal true, watch.wait(timeout: 0.2) + 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 @@ -160,6 +169,7 @@ class PostgresqlWakeUpTest < ActiveSupport::TestCase test "the adapter satisfies the wake-up contract" do assert_respond_to @adapter, :signal assert_respond_to @adapter, :wait + assert_respond_to @adapter, :watch SolidObjects.configuration.wake_up_adapter = @adapter assert_same @adapter, SolidObjects.wake_up diff --git a/test/integration/redis_wake_up_test.rb b/test/integration/redis_wake_up_test.rb index 99b13ec..783096f 100644 --- a/test/integration/redis_wake_up_test.rb +++ b/test/integration/redis_wake_up_test.rb @@ -38,13 +38,26 @@ class RedisWakeUpTest < ActiveSupport::TestCase @adapter.listen started = monotonic_now - @adapter.wait(timeout: 0.3) + notified = @adapter.wait(timeout: 0.3) elapsed = monotonic_now - started + assert_equal false, notified assert_operator elapsed, :>=, 0.2 assert_operator elapsed, :<, 3.0 end + test "a watch observes a signal delivered before waiting" do + signalled = @adapter.instance_variable_get(:@signalled) + watch = @adapter.watch + + signal_from_another_client + Timeout.timeout(1) do + sleep 0.001 until @adapter.instance_variable_get(:@signalled) > signalled + end + + assert_equal true, watch.wait(timeout: 0.1) + end + # The supervisor shares one adapter across runtime roles, so concurrent # waiters go through a single instance. test "every waiter on one shared adapter wakes on one signal" do @@ -128,6 +141,7 @@ class RedisWakeUpTest < ActiveSupport::TestCase test "the adapter satisfies the wake-up contract" do assert_respond_to @adapter, :signal assert_respond_to @adapter, :wait + assert_respond_to @adapter, :watch SolidObjects.configuration.wake_up_adapter = @adapter assert_same @adapter, SolidObjects.wake_up diff --git a/test/unit/configuration_test.rb b/test/unit/configuration_test.rb index d94d2d7..623a8a8 100644 --- a/test/unit/configuration_test.rb +++ b/test/unit/configuration_test.rb @@ -48,4 +48,13 @@ class ConfigurationTest < ActiveSupport::TestCase assert_equal "instance retention must be positive", error.message end + + test "rejects a non-positive idle polling interval" do + configuration = SolidObjects::Configuration.new + configuration.idle_polling_interval = 0 + + error = assert_raises(ArgumentError) { configuration.validate! } + + assert_equal "idle_polling_interval must be positive", error.message + end end diff --git a/test/unit/polling_backoff_test.rb b/test/unit/polling_backoff_test.rb new file mode 100644 index 0000000..012225a --- /dev/null +++ b/test/unit/polling_backoff_test.rb @@ -0,0 +1,54 @@ +# frozen_string_literal: true + +require "test_helper" +require "solid_objects/polling_backoff" + +class PollingBackoffTest < ActiveSupport::TestCase + test "doubles empty-poll intervals until the idle ceiling" do + backoff = SolidObjects::PollingBackoff.new( + minimum_interval: 0.025, + maximum_interval: 1.0 + ) + + intervals = [ backoff.current_interval ] + 7.times do + backoff.record_idle + intervals << backoff.current_interval + end + + assert_equal [ 0.025, 0.05, 0.1, 0.2, 0.4, 0.8, 1.0, 1.0 ], intervals + end + + test "resets to the fast interval after processed work" do + transitions = [] + backoff = SolidObjects::PollingBackoff.new( + minimum_interval: 0.025, + maximum_interval: 1.0, + on_change: ->(transition) { transitions << transition } + ) + backoff.record_idle + backoff.record_idle + + backoff.reset(:work) + + assert_equal 0.025, backoff.current_interval + assert_equal({ + previous_interval: 0.1, + current_interval: 0.025, + reason: :work + }, transitions.last) + end + + test "resets to the fast interval after a wake-up" do + backoff = SolidObjects::PollingBackoff.new( + minimum_interval: 0.025, + maximum_interval: 1.0 + ) + backoff.record_idle + backoff.record_idle + + backoff.reset(:wake_up) + + assert_equal 0.025, backoff.current_interval + end +end diff --git a/test/unit/wake_up_adapters_test.rb b/test/unit/wake_up_adapters_test.rb index 86417ed..3e9cd59 100644 --- a/test/unit/wake_up_adapters_test.rb +++ b/test/unit/wake_up_adapters_test.rb @@ -29,4 +29,19 @@ class WakeUpAdaptersTest < ActiveSupport::TestCase assert_respond_to adapter, :signal assert_respond_to adapter, :wait end + + test "the in-process wake-up distinguishes a timeout from a signal" do + adapter = SolidObjects::WakeUp.new + + assert_equal false, adapter.wait(timeout: 0.001) + end + + test "the in-process wake-up does not miss a signal sent before waiting" do + adapter = SolidObjects::WakeUp.new + watch = adapter.watch + + adapter.signal + + assert_equal true, watch.wait(timeout: 1.0) + end end