Stop worker processes even when finalization rethrows a task failure

src/ParallelTestRunner.jl

@@ -1480,29 +1480,39 @@ function _runtests(mod::Module, args::ParsedArgs;
     # finalization
     #
 
-    # wait for the printer to finish so that all results have been printed
-    close(printer_channel)
-    wait(printer_task)
+    try
+        # wait for the printer to finish so that all results have been printed
+        close(printer_channel)
+        wait(printer_task)
+
+        # wait for worker tasks to catch unhandled exceptions
+        for task in worker_tasks
+            try
+                wait(task)
+            catch err
+                # unwrap TaskFailedException
+                while isa(err, TaskFailedException)
+                    err = current_exceptions(err.task)[1].exception
+                end
 
-    # wait for worker tasks to catch unhandled exceptions
-    for task in worker_tasks
-        try
-            wait(task)
-        catch err
-            # unwrap TaskFailedException
-            while isa(err, TaskFailedException)
-                err = current_exceptions(err.task)[1].exception
+                isa(err, InterruptException) || rethrow()
             end
+        end
+    finally
+        # clean up remaining workers even when a worker or printer task failed and
+        # its exception is propagated, so worker processes don't outlive the run
 
-            isa(err, InterruptException) || rethrow()
+        # a failure during a serial phase can leave the shared worker in its Ref
+        # instead of the pool
+        if serial_worker[] !== nothing && Malt.isrunning(serial_worker[])
+            Malt.stop(serial_worker[])
         end
-    end
 
-    # clean up remaining workers in the pool
-    close(worker_pool)
-    for p in worker_pool
-        if p !== nothing && Malt.isrunning(p)
-            Malt.stop(p)
+        close(worker_pool)
+        for p in worker_pool
+            if p !== nothing && Malt.isrunning(p)
+                Malt.stop(p)
+            end
         end
     end
 

test/runtests.jl

@@ -1257,6 +1257,48 @@ end
         @test ParallelTestRunner.ID_COUNTER[] == old_id_counter + jobs + 1
     end
 
+    @testset "workers stopped when a task fails" begin
+        # a `test_worker` hook throwing during the serial phase strands the shared
+        # worker outside the pool while the task failure is rethrown from
+        # finalization; the worker process must still be stopped
+        before = _count_child_pids()
+        if before < 0
+            # counting child PIDs not supported on this platform
+            @test_skip false
+        else
+            testsuite = Dict(
+                "s1" => :( @test true ),
+                "s2" => :( @test true ),
+            )
+            exception = ErrorException("test_worker exploded")
+            test_worker(name) = name == "s2" ? throw(exception) : nothing
+            try
+                ParallelTestRunner._runtests(
+                    ParallelTestRunner, parse_args(["--jobs=1"]);
+                    testsuite,
+                    tests=["s1", "s2"],
+                    serial=["s1", "s2"],
+                    test_worker,
+                    stdout=devnull,
+                    stderr=devnull,
+                )
+                # the error must propagate out of `_runtests`
+                @test false
+            catch e
+                @test typeof(e) === TaskFailedException
+                @test first(Base.current_exceptions(e.task)).exception == exception
+            end
+            # allow a moment for worker processes to exit
+            after = -1
+            for _ in 1:50
+                sleep(0.1)
+                after = _count_child_pids()
+                after >= 0 && after <= before && break
+            end
+            @test after == before
+        end
+    end
+
     @testset "quickfail in serial phase before parallel" begin
         # The failing serial test runs first: the remaining serial tests and the whole
         # parallel batch should never be started.