Settle futures on any exception; validate WorkQueue/Progress input

WorkQueue workers only rescued StandardError and Interrupt. Any other
exception (ScriptError family - including NotImplementedError and
LoadError from a failed require - SystemExit, etc.) escaped the worker
thread without completing its future. The dead worker also kept
counting against max_concurrent, so queued work never started. Callers
blocked in Future#value forever, and SpinGroup#wait spun indefinitely
because the task's future never completed.

WorkQueue changes:

* Workers rescue Exception and fail the future, so waiters always
  observe the error and the worker stays alive for queued work.
  Interrupt handling is unchanged (fail the future, re-raise).
* Every future is settled structurally, from an ensure: Thread#raise
  and Thread#kill can land outside the worker's rescues (e.g.
  WorkQueue#interrupt hitting a worker between Queue#pop and the
  begin). A future abandoned that way is failed with
  WorkQueue::AbandonedTaskError instead of blocking Future#value
  forever.
* interrupt tolerates Interrupt re-raised by Thread#join: the Interrupt
  it raises into a worker can land after that worker has already left
  its rescues (it was terminating), killing it with an unhandled
  Interrupt that join would otherwise re-raise in the interrupting
  thread - replacing whatever that thread was propagating.
* WorkQueue.new rejects max_concurrent < 1, which produced a queue that
  accepted work, never started it, and returned from #wait immediately
  while futures stayed incomplete.

SpinGroup changes:

* Task#check treats any exception other than Interrupt/SystemExit as a
  task failure, reported through the normal debrief. Raising out of the
  render loop skipped work_queue.interrupt, the final render, and the
  debrief, leaving sibling workers running after #wait returned.
* wait interrupts the queue before letting SystemExit propagate, so
  remaining workers aren't orphaned.
* initialize falls back to 1024 for any non-positive max_concurrent
  instead of raising from WorkQueue's new guard.

Progress changes:

* tick validates percent/set_percent explicitly (finite numbers only,
  named in the error message) instead of relying on Array#min raising
  on an incomparable value, and validates before assigning so a raising
  tick leaves the bar renderable from its last valid state.
* The result is clamped to 0.0..1.0; the lower bound was previously
  unguarded, so a negative percent rendered a negative suffix.

Co-authored-by: River <river@shopify.com>
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: 3 people

lib/cli/ui/progress.rb

@@ -77,15 +77,20 @@ def initialize(title = nil, width: Terminal.width, reporter: nil)
       # * +:percent+ - Increment progress by a specific percent amount
       # * +:set_percent+ - Set progress to a specific percent
       #
-      # *Note:* The +:percent+ and +:set_percent must be between 0.00 and 1.0
+      # *Note:* The +:percent+ and +:set_percent+ must be finite numbers. The
+      # resulting progress is clamped to 0.0..1.0.
       #
       #: (?percent: Numeric?, ?set_percent: Numeric?) -> void
       def tick(percent: nil, set_percent: nil)
         raise ArgumentError, 'percent and set_percent cannot both be specified' if percent && set_percent
 
-        @percent_done += percent || 0.01
-        @percent_done = set_percent if set_percent
-        @percent_done = [@percent_done, 1.0].min # Make sure we can't go above 1.0
+        validate_finite!(percent, :percent)
+        validate_finite!(set_percent, :set_percent)
+
+        # Validate and clamp before assigning: a raising tick leaves the bar
+        # renderable from its last valid state.
+        new_percent = set_percent || @percent_done + (percent || 0.01)
+        @percent_done = new_percent.clamp(0.0, 1.0)
 
         # Update terminal progress reporter with current percentage
         @reporter&.set_progress((@percent_done * 100).floor)
@@ -125,6 +130,16 @@ def to_s
 
         [title, bar].compact.join("\n")
       end
+
+      private
+
+      #: (untyped value, Symbol name) -> void
+      def validate_finite!(value, name)
+        return if value.nil?
+        return if value.is_a?(Numeric) && value.real? && value.to_f.finite?
+
+        raise ArgumentError, "#{name} must be a finite number (got #{value.inspect})"
+      end
     end
   end
 end

lib/cli/ui/spinner/spin_group.rb

@@ -69,7 +69,7 @@ def initialize(auto_debrief: true, interrupt_debrief: false, max_concurrent: 0,
           @start = Time.new
           @stopped = false
           @internal_work_queue = work_queue.nil?
-          @work_queue = work_queue || WorkQueue.new(max_concurrent.zero? ? 1024 : max_concurrent) #: WorkQueue
+          @work_queue = work_queue || WorkQueue.new(max_concurrent.positive? ? max_concurrent : 1024) #: WorkQueue
           if block_given?
             yield self
             wait(to: to)
@@ -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,10 @@ def wait(to: $stdout)
           @work_queue.interrupt
           debrief(to: to) if @interrupt_debrief
           stopped? ? false : raise
+        rescue SystemExit
+          # Leaving the render loop mid-flight would orphan the remaining workers.
+          @work_queue.interrupt
+          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
@@ -66,6 +70,8 @@ def start
 
       #: (Integer max_concurrent) -> void
       def initialize(max_concurrent)
+        raise ArgumentError, "max_concurrent must be at least 1 (got #{max_concurrent})" if max_concurrent < 1
+
         @max_concurrent = max_concurrent
         @queue = Queue.new #: Queue
         @mutex = Mutex.new #: Mutex
@@ -105,7 +111,15 @@ def interrupt
           end
           # Interrupt all worker threads
           @workers.each { |worker| worker.raise(Interrupt) if worker.alive? }
-          @workers.each(&:join)
+          @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
           @workers.clear
         end
       end
@@ -116,21 +130,25 @@ def interrupt
       def start_worker
         @workers << Thread.new do
           loop do
-            work = @queue.pop
-            break if work.nil?
-
-            future, block = work
+            future = nil #: Future?
 
             begin
+              work = @queue.pop
+              break if work.nil?
+
+              future, block = work
               future.start
-              result = block.call
-              future.complete(result)
+              future.complete(block.call)
             rescue Interrupt => e
-              future.fail(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
+            rescue Exception => e # rubocop:disable Lint/RescueException
+              future&.fail(e)
+              # Don't re-raise - the future carries the error to callers; allow worker to continue
+            ensure
+              # A worker must never abandon a future: an unsettled future blocks
+              # Future#value forever. No-op once the future is settled.
+              future&.fail(AbandonedTaskError.new('worker exited before completing this task'))
             end
           end
         rescue Interrupt

test/cli/ui/progress_test.rb

@@ -28,6 +28,39 @@ def test_tick_with_set_percent_and_percent_raises
         end
       end
 
+      def test_tick_with_set_percent_below_zero_is_clamped_to_zero
+        assert_bar(set_percent: -0.5, expected_filled: 0, expected_unfilled: 10, suffix: ' 0%  ')
+      end
+
+      def test_tick_with_percent_change_below_zero_is_clamped_to_zero
+        assert_bar(percent: -0.5, expected_filled: 0, expected_unfilled: 10, suffix: ' 0%  ')
+      end
+
+      def test_tick_with_invalid_value_raises_without_corrupting_state
+        capture_io do
+          bar = Progress.new(width: 15)
+          bar.tick(set_percent: 0.4)
+
+          ['0.5', Float::NAN, Float::INFINITY, -Float::INFINITY, nil.to_a].each do |bad|
+            assert_raises(ArgumentError) { bar.tick(set_percent: bad) }
+            assert_raises(ArgumentError) { bar.tick(percent: bad) }
+          end
+
+          # The bar still renders from its last valid state
+          assert_match(/ 40% /, bar.to_s)
+        end
+      end
+
+      def test_tick_error_message_names_the_bad_argument
+        bar = Progress.new(width: 15)
+
+        error = assert_raises(ArgumentError) { bar.tick(set_percent: '0.5') }
+        assert_match(/set_percent must be a finite number/, error.message)
+
+        error = assert_raises(ArgumentError) { bar.tick(percent: Float::NAN) }
+        assert_match(/percent must be a finite number/, error.message)
+      end
+
       def test_with_title
         assert_bar(title: 'Title', set_percent: 0.1, expected_filled: 1, expected_unfilled: 9, suffix: ' 10% ')
       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,64 @@ 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
+
+            late = Queue.new
+            sg = SpinGroup.new(auto_debrief: false)
+            sg.add('exits') { exit(1) }
+            sg.add('slow') do
+              sleep(5)
+              late << :ran
+            end
+
+            error = Timeout.timeout(10) { assert_raises(SystemExit) { sg.wait } }
+
+            assert_equal(1, error.status)
+            assert(late.empty?, 'remaining workers should be interrupted, not orphaned')
+          end
+        end
+
+        def test_spin_group_non_positive_max_concurrent_uses_the_default
+          capture_io do
+            CLI::UI::StdoutRouter.ensure_activated
+
+            [0, -1].each do |max_concurrent|
+              sg = SpinGroup.new(max_concurrent: max_concurrent, auto_debrief: false)
+              sg.add('s') { true }
+
+              assert(Timeout.timeout(10) { sg.wait }, "max_concurrent: #{max_concurrent} should run tasks")
+            end
+          end
+        end
+
         def test_spin_group_success_debrief
           capture_io do
             CLI::UI::StdoutRouter.ensure_activated

test/cli/ui/work_queue_test.rb

@@ -65,6 +65,35 @@ def test_future_error
         assert_raises(StandardError, 'Test error') { future.value }
       end
 
+      def test_future_non_standard_error_completes_future_and_keeps_worker_alive
+        @work_queue = WorkQueue.new(1)
+        failed = @work_queue.enqueue { raise ScriptError, 'boom' }
+        replacement = @work_queue.enqueue { :ran }
+
+        @work_queue.wait
+
+        assert(failed.completed?, 'failing future should be completed')
+        error = assert_raises(ScriptError) { failed.value }
+        assert_equal('boom', error.message)
+        assert(replacement.completed?, 'subsequent task should still run on the same worker')
+        assert_equal(:ran, replacement.value)
+      end
+
+      def test_future_is_settled_when_worker_exits_without_finishing
+        @work_queue = WorkQueue.new(1)
+        abandoned = @work_queue.enqueue { Thread.exit }
+
+        @work_queue.wait
+
+        assert(abandoned.completed?, 'a worker must never leave a future unsettled')
+        assert_raises(WorkQueue::AbandonedTaskError) { abandoned.value }
+      end
+
+      def test_rejects_non_positive_max_concurrent
+        assert_raises(ArgumentError) { WorkQueue.new(0) }
+        assert_raises(ArgumentError) { WorkQueue.new(-1) }
+      end
+
       def test_max_concurrent
         max_concurrent = 2
         @work_queue = WorkQueue.new(max_concurrent)