Merge pull request #55794 from rails/fix-55513

[Fix #55513] parallel tests hanging when worker processes die abruptly

Author: byroot

activesupport/CHANGELOG.md

@@ -1,3 +1,10 @@
+*   Fix parallel tests hanging when worker processes die abruptly.
+
+    Previously, if a worker process was killed (e.g., OOM killed, `kill -9`) during parallel
+    test execution, the test suite would hang forever waiting for the dead worker.
+
+    *Joshua Young*
+
 *   `ActiveSupport::FileUpdateChecker` does not depend on `Time.now` to prevent unnecessary reloads with time travel test helpers
 
     *Jan Grodowski*

activesupport/lib/active_support/testing/parallelization.rb

@@ -47,8 +47,19 @@ def size
       end
 
       def shutdown
+        dead_worker_pids = @worker_pool.filter_map do |pid|
+          Process.waitpid(pid, Process::WNOHANG)
+        rescue Errno::ECHILD
+          pid
+        end
+        @queue_server.remove_dead_workers(dead_worker_pids)
+
         @queue_server.shutdown
-        @worker_pool.each { |pid| Process.waitpid pid }
+        @worker_pool.each do |pid|
+          Process.waitpid(pid)
+        rescue Errno::ECHILD
+          nil
+        end
       end
     end
   end

activesupport/lib/active_support/testing/parallelization/server.rb

@@ -14,6 +14,7 @@ class Server
         def initialize
           @queue = Queue.new
           @active_workers = Concurrent::Map.new
+          @worker_pids = Concurrent::Map.new
           @in_flight = Concurrent::Map.new
         end
 
@@ -40,12 +41,24 @@ def pop
           end
         end
 
-        def start_worker(worker_id)
+        def start_worker(worker_id, worker_pid)
           @active_workers[worker_id] = true
+          @worker_pids[worker_id] = worker_pid
         end
 
-        def stop_worker(worker_id)
+        def stop_worker(worker_id, worker_pid)
           @active_workers.delete(worker_id)
+          @worker_pids.delete(worker_id)
+        end
+
+        def remove_dead_workers(dead_pids)
+          dead_pids.each do |dead_pid|
+            worker_id = @worker_pids.key(dead_pid)
+            if worker_id
+              @active_workers.delete(worker_id)
+              @worker_pids.delete(worker_id)
+            end
+          end
         end
 
         def active_workers?

activesupport/lib/active_support/testing/parallelization/worker.rb

@@ -18,7 +18,7 @@ def start
             DRb.stop_service
 
             @queue = DRbObject.new_with_uri(@url)
-            @queue.start_worker(@id)
+            @queue.start_worker(@id, Process.pid)
 
             begin
               after_fork
@@ -29,7 +29,7 @@ def start
             set_process_title("(stopping)")
 
             run_cleanup
-            @queue.stop_worker(@id)
+            @queue.stop_worker(@id, Process.pid)
           end
         end
 

activesupport/test/parallelization_test.rb

@@ -0,0 +1,22 @@
+# frozen_string_literal: true
+
+require_relative "abstract_unit"
+
+class ParallelizationTest < ActiveSupport::TestCase
+  test "shutdown handles dead workers gracefully" do
+    parallelization = ActiveSupport::Testing::Parallelization.new(1)
+    parallelization.start
+
+    sleep 0.25
+
+    server = parallelization.instance_variable_get(:@queue_server)
+    assert server.active_workers?
+
+    worker_pids = parallelization.instance_variable_get(:@worker_pool)
+    Process.kill("KILL", worker_pids.first)
+    sleep 0.25
+
+    Timeout.timeout(2.5, Minitest::Assertion, "Expected shutdown to not hang") { parallelization.shutdown }
+    assert_not server.active_workers?
+  end
+end