feat: replace supervised roles that die

A role that raised left its thread dead while the process kept running and quietly did less work, with no signal beyond reduced throughput. The supervisor now watches its threads and restarts any that stopped before shutdown was requested, and prunes dead process records on the same monitor. Both intervals are configurable, a failing maintenance pass cannot stop the monitor, and no restart happens once shutdown begins.

Author: cardmagic

CHANGELOG.md

@@ -2,6 +2,15 @@
 
 ## Unreleased
 
+- 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`.
+
+## 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

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
@@ -42,8 +44,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
@@ -73,19 +73,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.

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

lib/solid_objects/supervisor.rb

@@ -4,7 +4,9 @@ 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 (?worker_count: Integer, ?effect_worker_count: Integer, ?broadcast_worker_count: Integer, ?reminder_scheduler_count: Integer) -> void
     def initialize(
@@ -20,7 +22,9 @@ def initialize(
         reminder_scheduler_count:
       )
       @threads = []
+      @monitor = nil
       @started = false
+      @cleaned_up_at = nil
     end
 
     # @rbs () -> void
@@ -36,7 +40,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,14 +50,16 @@ def stop
       return unless @started
 
       begin
+        @started = false
+        @monitor&.join(SolidObjects.configuration.supervisor_monitor_interval * 2)
         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
+        @monitor = nil
         SolidObjects.instrument(:"supervisor.stopped", component_count: components.length)
       end
     end
@@ -61,6 +68,59 @@ 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.
+    # @rbs () -> void
+    def monitor_loop
+      while @started
+        replace_dead_roles
+        cleanup_dead_processes
+        sleep SolidObjects.configuration.supervisor_monitor_interval
+      end
+    rescue
+      retry if @started
+    end
+
+    # @rbs () -> void
+    def replace_dead_roles
+      components.each_with_index do |component, index|
+        thread = threads[index]
+        next if thread&.alive?
+        next if component.stopped?
+
+        threads[index] = supervise(component)
+        SolidObjects.instrument(
+          :"supervisor.role_replaced",
+          role: component.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
+
     # A wake-up adapter may hold connections outside the pool, which would
     # otherwise accumulate across restarts in one process.
     # @rbs () -> void

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

sig/generated/lib/solid_objects/supervisor.rbs

@@ -2,11 +2,15 @@
 
 module SolidObjects
   class Supervisor
-    @components: Array[Worker | EffectExecutor | ReminderScheduler | BroadcastExecutor]
+    @cleaned_up_at: Float
+
+    @started: bool
+
+    @monitor: Thread?
 
     @threads: Array[Thread]
 
-    @started: bool
+    @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 +30,24 @@ 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.
+    # @rbs () -> void
+    def monitor_loop: () -> void
+
+    # @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
+
     # A wake-up adapter may hold connections outside the pool, which would
     # otherwise accumulate across restarts in one process.
     # @rbs () -> void