diff --git a/CHANGELOG.md b/CHANGELOG.md index 4e1a95a..c2da92f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,7 +1,13 @@ # Changelog -## Unreleased - +## 0.8.0 - 2026-08-10 + +- Replace a supervised role whose thread died. A role that raised left its + thread dead while the process kept running and quietly did less work; the + supervisor now restarts it until shutdown is requested. Prune dead process + records on an interval as part of the same monitor. Both intervals are + configurable through `supervisor_monitor_interval` and + `dead_process_cleanup_interval`. - Run compatibility CI across the span the gemspec advertises: Ruby 3.3 and 3.4 against Rails 8.0 and 8.1. The suite previously ran on one combination, so `>= 8.0` was a claim rather than a tested guarantee. Set `RAILS_VERSION` to diff --git a/Gemfile.lock b/Gemfile.lock index 4c7447d..fe03b7e 100644 --- a/Gemfile.lock +++ b/Gemfile.lock @@ -1,7 +1,7 @@ PATH remote: . specs: - solid_objects (0.7.3) + solid_objects (0.8.0) actioncable (>= 8.0) actionpack (>= 8.0) actionview (>= 8.0) @@ -373,7 +373,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.7.3) + solid_objects (0.8.0) 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/docs/roadmap.md b/docs/roadmap.md index 7b674f2..a7f68c7 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -31,6 +31,8 @@ - Bounded message/process pruning, actor-type opt-in instance expiration, graceful caller shutdown, committed state snapshots, and an opt-in Minitest helper +- Supervisor role replacement: a role whose thread dies is restarted until + shutdown is requested, and dead process records are pruned on an interval - SQLite, PostgreSQL, and MySQL integration suites - Opt-in cross-process wake-up on PostgreSQL through `WakeUpAdapters.for`, with a listening connection per waiting thread and release on supervisor shutdown @@ -45,8 +47,6 @@ ## Partially implemented -- Supervisor: starts and drains thread roles, but does not replace a crashed - role or run periodic maintenance automatically. - Wake-up strategy: in-process signaling, durable polling, injection, and an opt-in PostgreSQL notification adapter are implemented; a Redis adapter is not. In-process signaling cannot cross process boundaries, so without the @@ -76,19 +76,18 @@ ## Next milestones -1. Add automatic supervisor role replacement and periodic dead-process cleanup. -2. Add an optional Redis wake-up adapter, which is the remaining cross-process +1. Add an optional Redis wake-up adapter, which is the remaining cross-process option for MySQL. The PostgreSQL notification adapter, its latency benchmark, and its concurrency tests are implemented. -3. Add result lookup by request ID and broader deadlock retry classification. -4. Add scheduled retention and stale-process maintenance. -5. Add database/server-version checks and MySQL InnoDB verification at boot. -6. Add Turbo append intents and expand reconnect coverage in a full browser. -7. Add distributed rate limits, global admission hooks, and cache-capacity +2. Add result lookup by request ID and broader deadlock retry classification. +3. Add scheduled retention and stale-process maintenance. +4. Add database/server-version checks and MySQL InnoDB verification at boot. +5. Add Turbo append intents and expand reconnect coverage in a full browser. +6. Add distributed rate limits, global admission hooks, and cache-capacity eviction. -8. Expand security scanning and run compatibility CI across supported Rails and +7. Expand security scanning and run compatibility CI across supported Rails and Ruby versions. -9. Benchmark all workloads under documented hardware/database settings and +8. Benchmark all workloads under documented hardware/database settings and publish adapter-specific adoption measurements. Throughput, synchronous latency, query counts, and the three reactive delivery paths are measured on SQLite; adapter-specific and end-to-end browser measurements are not. diff --git a/lib/solid_objects/configuration.rb b/lib/solid_objects/configuration.rb index d6f83d5..670bf7a 100644 --- a/lib/solid_objects/configuration.rb +++ b/lib/solid_objects/configuration.rb @@ -21,6 +21,8 @@ class Configuration # @rbs @process_heartbeat_interval: Float # @rbs @process_alive_threshold: Float # @rbs @shutdown_timeout: Float + # @rbs @supervisor_monitor_interval: Float + # @rbs @dead_process_cleanup_interval: Float # @rbs @message_retention: Numeric # @rbs @message_retention_by_actor_type: Hash[String, Numeric] # @rbs @instance_retention_by_actor_type: Hash[String, Numeric] @@ -62,6 +64,8 @@ class Configuration :process_heartbeat_interval, :process_alive_threshold, :shutdown_timeout, + :supervisor_monitor_interval, + :dead_process_cleanup_interval, :message_retention, :message_retention_by_actor_type, :instance_retention_by_actor_type, @@ -102,6 +106,8 @@ def initialize @max_attempts = 5 @retry_delay = ->(attempt) { [ 2**(attempt - 1), 60 ].min.to_f } @lock_retry_attempts = 10 + @supervisor_monitor_interval = 1.0 + @dead_process_cleanup_interval = 60.0 @process_heartbeat_interval = 15.0 @process_alive_threshold = 60.0 @shutdown_timeout = 15.0 @@ -138,6 +144,10 @@ def validate! raise ArgumentError, "table_name_prefix must contain lowercase letters, digits, and underscores" end + unless supervisor_monitor_interval.positive? + raise ArgumentError, "supervisor_monitor_interval must be positive" + end + unless lease_duration > lease_renewal_interval raise ArgumentError, "lease_duration must be greater than lease_renewal_interval" end diff --git a/lib/solid_objects/supervisor.rb b/lib/solid_objects/supervisor.rb index 7600102..19ba16d 100644 --- a/lib/solid_objects/supervisor.rb +++ b/lib/solid_objects/supervisor.rb @@ -4,7 +4,10 @@ module SolidObjects class Supervisor # @rbs @components: Array[Worker | EffectExecutor | ReminderScheduler | BroadcastExecutor] # @rbs @threads: Array[Thread] + # @rbs @monitor: Thread? # @rbs @started: bool + # @rbs @cleaned_up_at: Float + # @rbs @lifecycle: Thread::Mutex # @rbs (?worker_count: Integer, ?effect_worker_count: Integer, ?broadcast_worker_count: Integer, ?reminder_scheduler_count: Integer) -> void def initialize( @@ -20,7 +23,10 @@ def initialize( reminder_scheduler_count: ) @threads = [] + @monitor = nil @started = false + @cleaned_up_at = nil + @lifecycle = Thread::Mutex.new end # @rbs () -> void @@ -36,7 +42,8 @@ def start return if @started @started = true - @threads = components.map { |component| Thread.new { component.run } } + @threads = components.map { |component| supervise(component) } + @monitor = Thread.new { monitor_loop } SolidObjects.instrument(:"supervisor.started", component_count: components.length) end @@ -45,6 +52,11 @@ def stop return unless @started begin + # Flipping the flag under the same lock replacement takes means a + # replacement either completes before shutdown reads the component + # list, or never starts. + @lifecycle.synchronize { @started = false } + stop_monitor components.each(&:request_shutdown) join_until_timeout components.reject(&:stopped?).each(&:stop) @@ -52,7 +64,7 @@ def stop # 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 + @monitor = nil SolidObjects.instrument(:"supervisor.stopped", component_count: components.length) end end @@ -61,6 +73,95 @@ def stop attr_reader :components, :threads + # A role that raises leaves its thread dead. Without replacement the + # process keeps running while quietly doing less work, so the supervisor + # watches its threads and restarts any that stopped before shutdown. + # A failing pass must not stop supervision, and must not retry without + # pacing either: a persistently failing database would otherwise spin. + # @rbs () -> void + def monitor_loop + while @started + begin + replace_dead_roles + cleanup_dead_processes + rescue => error + SolidObjects.instrument( + :"supervisor.monitor_failed", + error_class: error.class.name, + error_message: error.message + ) + end + sleep SolidObjects.configuration.supervisor_monitor_interval + end + end + + # A role that raises runs its own shutdown cleanup on the way out, so a + # crashed component reports itself stopped exactly like one that was asked + # to stop. While the supervisor is still running, a dead thread can only + # mean a crash, so replacement keys on the supervisor rather than on the + # component. The crashed instance has already released its process record, + # so a fresh one takes its place. + # @rbs () -> void + def replace_dead_roles + components.each_with_index do |component, index| + thread = threads[index] + next if thread&.alive? + + replaced = @lifecycle.synchronize do + next false unless @started + + replacement = component.class.new + components[index] = replacement + threads[index] = supervise(replacement) + replacement + end + break unless replaced + + SolidObjects.instrument( + :"supervisor.role_replaced", + role: replaced.class.name, + error_class: thread_error(thread) + ) + end + end + + # @rbs (Thread?) -> String? + def thread_error(thread) + thread&.join + nil + rescue => error + error.class.name + end + + # @rbs () -> void + def cleanup_dead_processes + interval = SolidObjects.configuration.dead_process_cleanup_interval + return unless interval.positive? + return if @cleaned_up_at && monotonic_now - @cleaned_up_at < interval + + @cleaned_up_at = monotonic_now + ProcessRegistry.cleanup_dead + end + + # @rbs (untyped) -> Thread + def supervise(component) + Thread.new { component.run } + end + + # The monitor only performs maintenance, so shutdown must never return while + # it is still alive: a pass blocked on the database would otherwise outlive + # the supervisor that owns it. + # @rbs () -> void + def stop_monitor + monitor = @monitor + @monitor = nil + return unless monitor + + monitor.join(SolidObjects.configuration.shutdown_timeout) + monitor.kill if monitor.alive? + monitor.join(SolidObjects.configuration.supervisor_monitor_interval) + end + # A wake-up adapter may hold connections outside the pool, which would # otherwise accumulate across restarts in one process. # @rbs () -> void diff --git a/lib/solid_objects/version.rb b/lib/solid_objects/version.rb index ffe386a..56a06a8 100644 --- a/lib/solid_objects/version.rb +++ b/lib/solid_objects/version.rb @@ -1,5 +1,5 @@ # rbs_inline: enabled module SolidObjects - VERSION = "0.7.3" + VERSION = "0.8.0" end diff --git a/sig/generated/lib/solid_objects/configuration.rbs b/sig/generated/lib/solid_objects/configuration.rbs index 3597980..0dbf707 100644 --- a/sig/generated/lib/solid_objects/configuration.rbs +++ b/sig/generated/lib/solid_objects/configuration.rbs @@ -2,10 +2,12 @@ module SolidObjects class Configuration - @process_alive_threshold: Float - @shutdown_timeout: Float + @supervisor_monitor_interval: Float + + @dead_process_cleanup_interval: Float + @message_retention: Numeric @message_retention_by_actor_type: Hash[String, Numeric] @@ -82,6 +84,8 @@ module SolidObjects @process_heartbeat_interval: Float + @process_alive_threshold: Float + attr_accessor table_name_prefix: untyped attr_accessor polling_interval: untyped @@ -120,6 +124,10 @@ module SolidObjects attr_accessor shutdown_timeout: untyped + attr_accessor supervisor_monitor_interval: untyped + + attr_accessor dead_process_cleanup_interval: untyped + attr_accessor message_retention: untyped attr_accessor message_retention_by_actor_type: untyped diff --git a/sig/generated/lib/solid_objects/supervisor.rbs b/sig/generated/lib/solid_objects/supervisor.rbs index 24f53b5..12d7ac4 100644 --- a/sig/generated/lib/solid_objects/supervisor.rbs +++ b/sig/generated/lib/solid_objects/supervisor.rbs @@ -2,12 +2,18 @@ module SolidObjects class Supervisor - @components: Array[Worker | EffectExecutor | ReminderScheduler | BroadcastExecutor] + @lifecycle: Thread::Mutex - @threads: Array[Thread] + @cleaned_up_at: Float @started: bool + @monitor: Thread? + + @threads: Array[Thread] + + @components: Array[Worker | EffectExecutor | ReminderScheduler | BroadcastExecutor] + # @rbs (?worker_count: Integer, ?effect_worker_count: Integer, ?broadcast_worker_count: Integer, ?reminder_scheduler_count: Integer) -> void def initialize: (?worker_count: Integer, ?effect_worker_count: Integer, ?broadcast_worker_count: Integer, ?reminder_scheduler_count: Integer) -> void @@ -26,6 +32,38 @@ module SolidObjects attr_reader threads: untyped + # A role that raises leaves its thread dead. Without replacement the + # process keeps running while quietly doing less work, so the supervisor + # watches its threads and restarts any that stopped before shutdown. + # A failing pass must not stop supervision, and must not retry without + # pacing either: a persistently failing database would otherwise spin. + # @rbs () -> void + def monitor_loop: () -> void + + # A role that raises runs its own shutdown cleanup on the way out, so a + # crashed component reports itself stopped exactly like one that was asked + # to stop. While the supervisor is still running, a dead thread can only + # mean a crash, so replacement keys on the supervisor rather than on the + # component. The crashed instance has already released its process record, + # so a fresh one takes its place. + # @rbs () -> void + def replace_dead_roles: () -> void + + # @rbs (Thread?) -> String? + def thread_error: (Thread?) -> String? + + # @rbs () -> void + def cleanup_dead_processes: () -> void + + # @rbs (untyped) -> Thread + def supervise: (untyped) -> Thread + + # The monitor only performs maintenance, so shutdown must never return while + # it is still alive: a pass blocked on the database would otherwise outlive + # the supervisor that owns it. + # @rbs () -> void + def stop_monitor: () -> void + # A wake-up adapter may hold connections outside the pool, which would # otherwise accumulate across restarts in one process. # @rbs () -> void diff --git a/test/integration/supervisor_replacement_test.rb b/test/integration/supervisor_replacement_test.rb new file mode 100644 index 0000000..3bd3dbd --- /dev/null +++ b/test/integration/supervisor_replacement_test.rb @@ -0,0 +1,214 @@ +# frozen_string_literal: true + +require "database_test_helper" +require "timeout" + +class SupervisorReplacementTest < ActiveSupport::TestCase + # A component that crashes on its first run and records every run after. + class CrashingRole + class << self + attr_accessor :shared_runs, :shared_crashes + end + + attr_reader :runs + + def initialize(crashes: self.class.shared_crashes || 1) + @crashes = crashes + self.class.shared_runs ||= Queue.new + @runs = self.class.shared_runs + @stopped = false + @shutdown = false + end + + # Mirrors Worker and ReminderScheduler, which run their shutdown cleanup in + # an ensure and therefore report themselves stopped after a crash too. + def run + @runs << monotonic_now + raise "role crashed" if @runs.size <= @crashes + + sleep 0.01 until @shutdown + ensure + stop + end + + def request_shutdown = @shutdown = true + + def stop = @stopped = true + + def stopped? = @stopped + + private + + def monotonic_now = ::Process.clock_gettime(::Process::CLOCK_MONOTONIC) + end + + class HealthyRole < CrashingRole + def initialize(**) = super(crashes: 0) + end + + setup do + [ CrashingRole, HealthyRole ].each do |role| + role.shared_runs = nil + role.shared_crashes = nil + end + SolidObjects.configuration.supervisor_monitor_interval = 0.02 + SolidObjects.configuration.dead_process_cleanup_interval = 0.05 + end + + teardown { @supervisor&.stop } + + test "replaces a role whose thread died" do + role = CrashingRole.new + @supervisor = supervisor_for(role) + + @supervisor.start + + Timeout.timeout(10) { sleep 0.01 until role.runs.size >= 2 } + assert_operator role.runs.size, :>=, 2, + "a crashed role should be run again" + end + + test "keeps replacing a role that keeps crashing" do + CrashingRole.shared_crashes = 3 + role = CrashingRole.new + @supervisor = supervisor_for(role) + + @supervisor.start + + Timeout.timeout(10) { sleep 0.01 until role.runs.size >= 4 } + assert_operator role.runs.size, :>=, 4 + end + + test "does not restart a healthy role" do + role = HealthyRole.new + @supervisor = supervisor_for(role) + + @supervisor.start + sleep 0.2 + + assert_equal 1, role.runs.size + end + + test "does not restart roles after shutdown is requested" do + role = CrashingRole.new(crashes: 1) + @supervisor = supervisor_for(role) + @supervisor.start + Timeout.timeout(10) { sleep 0.01 until role.runs.size >= 2 } + + @supervisor.stop + runs_at_stop = role.runs.size + sleep 0.2 + + assert_equal runs_at_stop, role.runs.size + end + + test "no role outlives shutdown when replacement races it" do + 10.times do + CrashingRole.shared_runs = nil + role = CrashingRole.new + supervisor = supervisor_for(role) + supervisor.start + Timeout.timeout(10) { sleep 0.001 until role.runs.size >= 1 } + + supervisor.stop + + live = supervisor.instance_variable_get(:@threads).select(&:alive?) + assert_empty live, "a replacement started during shutdown must not survive it" + end + end + + test "shutdown does not return while the monitor is still alive" do + SolidObjects.configuration.shutdown_timeout = 0.1 + @supervisor = supervisor_for(HealthyRole.new) + blocking = Queue.new + @supervisor.define_singleton_method(:cleanup_dead_processes) { blocking.pop } + @supervisor.start + sleep 0.1 + + @supervisor.stop + + refute @supervisor.instance_variable_get(:@monitor)&.alive?, + "a blocked monitor must not outlive shutdown" + ensure + SolidObjects.configuration.shutdown_timeout = 5.0 + end + + test "instruments a replacement" do + events = [] + subscription = ActiveSupport::Notifications.subscribe("solid_objects.supervisor.role_replaced") do |event| + events << event.payload + end + role = CrashingRole.new + @supervisor = supervisor_for(role) + + @supervisor.start + Timeout.timeout(10) { sleep 0.01 until events.any? } + + assert_equal "SupervisorReplacementTest::CrashingRole", events.first.fetch(:role) + ensure + ActiveSupport::Notifications.unsubscribe(subscription) if subscription + end + + test "prunes processes whose heartbeat has stopped" do + dead = SolidObjects::Process.create!( + id: SecureRandom.uuid, + kind: "worker", + hostname: "dead-host", + pid: 999_999, + started_at: 1.hour.ago, + last_heartbeat_at: 1.hour.ago, + metadata: {} + ) + @supervisor = supervisor_for(HealthyRole.new) + + @supervisor.start + Timeout.timeout(10) { sleep 0.02 until dead.reload.shutdown_state == "stopped" } + + assert_equal "stopped", dead.reload.shutdown_state + end + + test "a persistently failing maintenance pass stays paced" do + failures = [] + subscription = ActiveSupport::Notifications.subscribe("solid_objects.supervisor.monitor_failed") do + failures << ::Process.clock_gettime(::Process::CLOCK_MONOTONIC) + end + @supervisor = supervisor_for(HealthyRole.new) + @supervisor.define_singleton_method(:cleanup_dead_processes) { raise "boom" } + + @supervisor.start + sleep 0.3 + + # 0.3s at a 0.02s interval bounds the passes; spinning would produce orders + # of magnitude more. + assert_operator failures.size, :>, 1, "the monitor should keep running" + assert_operator failures.size, :<, 40, "the monitor should pace its retries" + ensure + ActiveSupport::Notifications.unsubscribe(subscription) if subscription + end + + test "a failing maintenance pass does not stop the supervisor" do + role = HealthyRole.new + @supervisor = supervisor_for(role) + @supervisor.define_singleton_method(:cleanup_dead_processes) { raise "boom" } + + @supervisor.start + sleep 0.2 + + assert_equal 1, role.runs.size, "the role should still be running" + end + + private + + # Every instance of a role shares one run log, so replacement instances are + # observable the way the supervisor creates them. + def supervisor_for(*roles) + supervisor = SolidObjects::Supervisor.new( + worker_count: 0, + effect_worker_count: 0, + broadcast_worker_count: 0, + reminder_scheduler_count: 0 + ) + supervisor.instance_variable_set(:@components, roles) + supervisor + end +end