fix: serialize role replacement with shutdown

The monitor could pass its started check immediately before stop cleared the flag, then swap in a replacement while shutdown walked the component list, leaving a role that never received request_shutdown alive after stop returned. Flip the flag and perform the swap under one lock, so a replacement either completes before shutdown reads the list or never starts.

Author: cardmagic

lib/solid_objects/supervisor.rb

@@ -7,6 +7,7 @@ class Supervisor
     # @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(
@@ -25,6 +26,7 @@ def initialize(
       @monitor = nil
       @started = false
       @cleaned_up_at = nil
+      @lifecycle = Thread::Mutex.new
     end
 
     # @rbs () -> void
@@ -50,7 +52,10 @@ def stop
       return unless @started
 
       begin
-        @started = false
+        # 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 }
         @monitor&.join(SolidObjects.configuration.supervisor_monitor_interval * 2)
         components.each(&:request_shutdown)
         join_until_timeout
@@ -101,14 +106,20 @@ def replace_dead_roles
       components.each_with_index do |component, index|
         thread = threads[index]
         next if thread&.alive?
-        break unless @started
 
-        replacement = component.class.new
-        components[index] = replacement
-        threads[index] = supervise(replacement)
+        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: replacement.class.name,
+          role: replaced.class.name,
           error_class: thread_error(thread)
         )
       end

sig/generated/lib/solid_objects/supervisor.rbs

@@ -2,6 +2,8 @@
 
 module SolidObjects
   class Supervisor
+    @lifecycle: Thread::Mutex
+
     @cleaned_up_at: Float
 
     @started: bool

test/integration/supervisor_replacement_test.rb

@@ -102,6 +102,21 @@ def initialize(**) = super(crashes: 0)
     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 "instruments a replacement" do
     events = []
     subscription = ActiveSupport::Notifications.subscribe("solid_objects.supervisor.role_replaced") do |event|