Handle async pool cancellations as fatal

Author: crmne

lib/solid_queue/execution_pools/async_pool.rb

@@ -126,6 +126,9 @@ def drain_queue(task, semaphore)
 
         def perform_execution(execution)
           wrap_in_app_executor { execution.perform }
+        rescue Async::Cancel => error
+          handle_thread_error(error)
+          register_fatal_error(error)
         rescue Exception => error
           handle_thread_error(error)
         ensure

test/unit/execution_pools/async_pool_test.rb

@@ -9,6 +9,13 @@ def perform
     end
   end
 
+  CancelledExecution = Struct.new(:started) do
+    def perform
+      started << true if started
+      raise Async::Cancel.new
+    end
+  end
+
   def test_raises_a_clear_error_when_the_async_gem_is_unavailable
     load_error = LoadError.new("cannot load such file -- async")
 
@@ -59,4 +66,27 @@ def test_waits_for_in_flight_executions_during_shutdown
     assert_nil pool.wait_for_termination(0.01)
     assert pool.wait_for_termination(1.second)
   end
+
+  def test_marks_the_pool_as_fatal_when_an_execution_is_cancelled
+    notifications = Thread::Queue.new
+    started = Thread::Queue.new
+    reported_errors = []
+    original_on_thread_error = SolidQueue.on_thread_error
+    SolidQueue.on_thread_error = ->(error) { reported_errors << error.class.name }
+
+    pool = SolidQueue::ExecutionPools::AsyncPool.new(1, on_state_change: -> { notifications << :changed })
+
+    pool.post CancelledExecution.new(started)
+    Timeout.timeout(1.second) { started.pop }
+    Timeout.timeout(1.second) { notifications.pop }
+
+    error = assert_raises(Async::Cancel) { pool.available_capacity }
+    assert_equal "Task was cancelled", error.message
+    assert_equal [ "Async::Cancel" ], reported_errors
+    assert_raises(Async::Cancel) { pool.metadata }
+  ensure
+    SolidQueue.on_thread_error = original_on_thread_error
+    pool&.shutdown
+    pool&.wait_for_termination(1.second)
+  end
 end