Settle futures on any exception; validate WorkQueue/Progress input

lib/cli/ui/spinner/spin_group.rb

@@ -143,7 +143,12 @@ def check
               result = @future.value
               @success = true
               @success = false if result == TASK_FAILED
-            rescue => exc
+            rescue Interrupt, SystemExit
+              raise
+            rescue Exception => exc # rubocop:disable Lint/RescueException
+              # Any other exception (including ScriptError and friends) is a task
+              # failure, reported through the normal debrief rather than raised
+              # out of the middle of SpinGroup#wait's render loop.
               @exception = exc
               @success = false
             end
@@ -439,6 +444,11 @@ def wait(to: $stdout)
           @work_queue.interrupt
           debrief(to: to) if @interrupt_debrief
           stopped? ? false : raise
+        rescue SystemExit
+          # An internal queue has no owner left to stop its remaining workers.
+          # A shared queue remains the caller's responsibility.
+          @work_queue.interrupt if @internal_work_queue
+          raise
         end
 
         #: (String message) -> void

lib/cli/ui/work_queue.rb

@@ -4,6 +4,10 @@
 module CLI
   module UI
     class WorkQueue
+      # Raised into a future whose worker exited without settling it: Thread#raise
+      # and Thread#kill can land outside the worker's rescues.
+      class AbandonedTaskError < StandardError; end
+
       class Future
         #: -> void
         def initialize
@@ -17,21 +21,27 @@ def initialize
 
         #: (untyped result) -> void
         def complete(result)
-          @mutex.synchronize do
-            @completed = true
-            @result = result
-            @condition.broadcast
+          Thread.handle_interrupt(Exception => :never) do
+            @mutex.synchronize do
+              return if @completed
+
+              @result = result
+              @completed = true
+              @condition.broadcast
+            end
           end
         end
 
         #: (Exception error) -> void
         def fail(error)
-          @mutex.synchronize do
-            return if @completed
+          Thread.handle_interrupt(Exception => :never) do
+            @mutex.synchronize do
+              return if @completed
 
-            @completed = true
-            @error = error
-            @condition.broadcast
+              @error = error
+              @completed = true
+              @condition.broadcast
+            end
           end
         end
 
@@ -66,76 +76,132 @@ def start
 
       #: (Integer max_concurrent) -> void
       def initialize(max_concurrent)
-        @max_concurrent = max_concurrent
+        @max_concurrent = [max_concurrent, 1].max #: Integer
         @queue = Queue.new #: Queue
         @mutex = Mutex.new #: Mutex
-        @condition = ConditionVariable.new #: ConditionVariable
         @workers = [] #: Array[Thread]
+        @active_workers = 0 #: Integer
       end
 
       #: { -> untyped } -> Future
       def enqueue(&block)
         future = Future.new
         @mutex.synchronize do
-          start_worker if @workers.size < @max_concurrent
+          @queue.push([future, block])
+          start_worker if @active_workers < @max_concurrent
         end
-        @queue.push([future, block])
         future
       end
 
       #: -> void
       def close
-        @queue.close
+        @mutex.synchronize { @queue.close }
       end
 
       #: -> void
       def wait
-        @queue.close
-        @workers.each(&:join)
+        close
+
+        joined = 0
+        loop do
+          workers = @mutex.synchronize { @workers.drop(joined) }
+          break if workers.empty?
+
+          workers.each(&:join)
+          joined += workers.size
+        end
       end
 
       #: -> void
       def interrupt
-        @mutex.synchronize do
+        workers = @mutex.synchronize do
           @queue.close
           # Fail any remaining tasks in the queue
           until @queue.empty?
             future, _block = @queue.pop(true)
             future&.fail(Interrupt.new)
           end
-          # Interrupt all worker threads
-          @workers.each { |worker| worker.raise(Interrupt) if worker.alive? }
-          @workers.each(&:join)
+          @workers.dup
+        end
+
+        current_worker = workers.include?(Thread.current)
+        other_workers = workers.reject { |worker| worker == Thread.current }
+
+        # Interrupt worker threads without holding @mutex: workers retire under
+        # that mutex, so joining them while holding it would deadlock.
+        other_workers.each do |worker|
+          worker.raise(Interrupt) if worker.alive?
+        rescue ThreadError
+          # The worker exited between alive? and raise.
+        end
+        other_workers.each do |worker|
+          worker.join
+        rescue Interrupt
+          # The Interrupt raised above can land after a worker has left its
+          # rescues (it was already terminating); the worker then dies with
+          # it and join re-raises it here. That must not replace whatever
+          # this thread is propagating (e.g. the SystemExit that triggered
+          # this interrupt).
+        end
+
+        @mutex.synchronize do
           @workers.clear
         end
+
+        raise Interrupt if current_worker
       end
 
       private
 
       #: -> void
       def start_worker
-        @workers << Thread.new do
-          loop do
-            work = @queue.pop
-            break if work.nil?
-
-            future, block = work
-
-            begin
-              future.start
-              result = block.call
-              future.complete(result)
-            rescue Interrupt => e
-              future.fail(e)
-              raise # Always re-raise interrupts to terminate the worker
-            rescue StandardError => e
-              future.fail(e)
-              # Don't re-raise standard errors - allow worker to continue
+        worker = Thread.new do
+          Thread.handle_interrupt(Exception => :never) do
+            loop do
+              future = nil #: Future?
+
+              begin
+                work = nil #: untyped
+                Thread.handle_interrupt(Exception => :on_blocking) do
+                  work = @queue.pop
+                end
+
+                break if work.nil?
+
+                future, block = work
+                future.start
+                result = Thread.handle_interrupt(Exception => :immediate) { block.call }
+                future.complete(result)
+              rescue Interrupt => e
+                future&.fail(e)
+                raise
+              rescue StandardError => e
+                future&.fail(e)
+                # Ordinary task failures do not poison the worker.
+              rescue Exception => e # rubocop:disable Lint/RescueException
+                future&.fail(e)
+                raise
+              ensure
+                # A worker must never abandon a future: an unsettled future
+                # blocks Future#value forever. No-op once it is settled.
+                future&.fail(AbandonedTaskError.new('worker exited before completing this task'))
+              end
             end
           end
         rescue Interrupt
           # Clean exit on interrupt
+        rescue Exception # rubocop:disable Lint/RescueException
+          # The future carries the exception to its caller. Fatal exceptions
+          # terminate this worker; the ensure below replaces it if needed.
+        ensure
+          @mutex.synchronize do
+            @active_workers -= 1
+            start_worker if @active_workers < @max_concurrent && !@queue.empty?
+          end
         end
+
+        @workers << worker
+        @active_workers += 1
       end
     end
   end

test/cli/ui/spinner/spin_group_test.rb

@@ -1,6 +1,7 @@
 # frozen_string_literal: true
 
 require 'test_helper'
+require 'timeout'
 
 module CLI
   module UI
@@ -36,6 +37,111 @@ def test_spin_group_auto_debrief_false
           assert_equal('', err)
         end
 
+        def test_spin_group_non_standard_error_is_reported_as_a_task_failure
+          capture_io do
+            CLI::UI::StdoutRouter.ensure_activated
+
+            failures = []
+            sibling_ran = false
+
+            sg = SpinGroup.new
+            sg.failure_debrief { |title, exception, _out, _err| failures << [title, exception] }
+            sg.add('raises') { raise NotImplementedError, 'not done yet' }
+            sg.add('sibling') { sibling_ran = true }
+
+            # Before WorkQueue settled futures for non-StandardError exceptions
+            # this spun forever; the timeout keeps a regression from wedging the
+            # suite instead of failing it.
+            refute(Timeout.timeout(10) { sg.wait })
+
+            assert(sibling_ran, 'sibling task should still run to completion')
+            assert_equal(1, failures.size)
+            title, error = failures.first
+            assert_equal('raises', title)
+            assert_instance_of(NotImplementedError, error)
+            assert_equal('not done yet', error.message)
+          end
+        end
+
+        def test_spin_group_system_exit_propagates_and_interrupts_remaining_work
+          capture_io do
+            CLI::UI::StdoutRouter.ensure_activated
+
+            slow_started = Queue.new
+            interrupted = Queue.new
+            sg = SpinGroup.new(auto_debrief: false)
+            sg.add('exits') do
+              slow_started.pop
+              exit(1)
+            end
+            sg.add('slow') do
+              slow_started << true
+              sleep(5)
+            rescue Interrupt
+              interrupted << true
+              raise
+            end
+
+            error = Timeout.timeout(10) { assert_raises(SystemExit) { sg.wait } }
+
+            assert_equal(1, error.status)
+            assert(Timeout.timeout(1) { interrupted.pop }, 'remaining workers should observe Interrupt')
+            workers = sg.instance_variable_get(:@work_queue).instance_variable_get(:@workers)
+            refute(workers.any?(&:alive?), 'remaining workers should be terminated before wait raises')
+          end
+        end
+
+        def test_spin_group_zero_max_concurrent_uses_the_default
+          capture_io do
+            CLI::UI::StdoutRouter.ensure_activated
+
+            sg = SpinGroup.new(max_concurrent: 0, auto_debrief: false)
+            sg.add('s') { true }
+
+            assert(Timeout.timeout(10) { sg.wait })
+          end
+        end
+
+        def test_spin_group_negative_max_concurrent_runs_serially
+          capture_io do
+            CLI::UI::StdoutRouter.ensure_activated
+
+            mutex = Mutex.new
+            running = 0
+            max_running = 0
+            sg = SpinGroup.new(max_concurrent: -1, auto_debrief: false)
+            3.times do |i|
+              sg.add("task #{i}") do
+                mutex.synchronize do
+                  running += 1
+                  max_running = [max_running, running].max
+                end
+                sleep(0.01)
+                mutex.synchronize { running -= 1 }
+              end
+            end
+
+            assert(Timeout.timeout(10) { sg.wait })
+            assert_equal(1, max_running)
+          end
+        end
+
+        def test_spin_group_system_exit_does_not_interrupt_a_shared_work_queue
+          capture_io do
+            CLI::UI::StdoutRouter.ensure_activated
+
+            work_queue = WorkQueue.new(1)
+            sg = SpinGroup.new(auto_debrief: false, work_queue: work_queue)
+            sg.add('exits') { exit(1) }
+
+            assert_raises(SystemExit) { sg.wait }
+
+            followup = work_queue.enqueue { :ran }
+            work_queue.wait
+            assert_equal(:ran, followup.value)
+          end
+        end
+
         def test_spin_group_success_debrief
           capture_io do
             CLI::UI::StdoutRouter.ensure_activated