Gate async workers on fiber isolation

Author: crmne

README.md

@@ -278,6 +278,7 @@ Here's an overview of the different options:
   Check the sections below on [how queue order behaves combined with priorities](#queue-order-and-priorities), and [how the way you specify the queues per worker might affect performance](#queues-specification-and-performance).
 
 - `execution_mode`: controls how a worker executes claimed jobs. `thread` is the default and uses the existing thread pool behavior. `async` executes jobs as fibers on a single reactor thread. `fiber` is accepted as an alias for `async`.
+  Async worker execution requires fiber-scoped isolated execution state. In Rails apps, set `config.active_support.isolation_level = :fiber` before using `execution_mode: async`. Solid Queue refuses to boot async workers when isolation remains thread-scoped.
 - `threads`: this is the execution capacity for a worker in `thread` mode. It is the max size of the thread pool. By default, this is `3`. Only workers have this setting.
 It is recommended to set this value less than or equal to the queue database's connection pool size minus 2, as each worker uses connections for polling and heartbeat and thread mode may use additional connections for job execution.
 - `capacity`: an alias for worker execution capacity. This is the clearer name when `execution_mode: async`, because it refers to in-flight execution capacity rather than operating system threads.
@@ -377,6 +378,8 @@ By default, workers in Solid Queue use a thread pool to run work in multiple thr
 
 Async worker execution is best suited for cooperative, mostly I/O-bound jobs. Blocking or CPU-heavy work still blocks the single reactor thread, so it should not be expected to outperform thread mode for every workload.
 
+Because async workers run multiple fibers on a single thread, Rails must also isolate execution state per fiber rather than per thread. If your app keeps the default thread-scoped isolation level, Solid Queue will raise a boot-time error instead of running async workers with shared Active Record state.
+
 The supervisor is in charge of managing these processes, and it responds to the following signals when running in its own process via `bin/jobs` or with [the Puma plugin](#puma-plugin) with the default `fork` mode:
 - `TERM`, `INT`: starts graceful termination. The supervisor will send a `TERM` signal to its supervised processes, and it'll wait up to `SolidQueue.shutdown_timeout` time until they're done. If any supervised processes are still around by then, it'll send a `QUIT` signal to them to indicate they must exit.
 - `QUIT`: starts immediate termination. The supervisor will send a `QUIT` signal to its supervised processes, causing them to exit immediately.

lib/solid_queue/configuration.rb

@@ -8,6 +8,7 @@ class Configuration
     validate :ensure_valid_recurring_tasks
     validate :ensure_correctly_sized_thread_pool
     validate :ensure_valid_worker_execution_modes
+    validate :ensure_async_workers_use_supported_isolation_level
 
     class Process < Struct.new(:kind, :attributes)
       def instantiate
@@ -105,6 +106,14 @@ def ensure_valid_worker_execution_modes
         end
       end
 
+      def ensure_async_workers_use_supported_isolation_level
+        return unless workers_options.any? { |options| async_worker?(options) }
+
+        SolidQueue::ExecutionPools::AsyncPool.ensure_supported_isolation_level!
+      rescue ArgumentError => error
+        errors.add(:base, error.message)
+      end
+
       def default_options
         {
           mode: ENV["SOLID_QUEUE_SUPERVISOR_MODE"] || :fork,

lib/solid_queue/execution_pools/async_pool.rb

@@ -15,6 +15,29 @@ def initialize(error)
         end
       end
 
+      class UnsupportedIsolationLevelError < ArgumentError
+        def initialize(level)
+          super(
+            "Async execution mode requires fiber-scoped isolated execution state. " \
+            "Set `ActiveSupport::IsolatedExecutionState.isolation_level = :fiber` " \
+            "(or `config.active_support.isolation_level = :fiber` in Rails). " \
+            "Current isolation level: #{level.inspect}"
+          )
+        end
+      end
+
+      class << self
+        def ensure_supported_isolation_level!
+          return if supported_isolation_level?
+
+          raise UnsupportedIsolationLevelError.new(ActiveSupport::IsolatedExecutionState.isolation_level)
+        end
+
+        def supported_isolation_level?
+          ActiveSupport::IsolatedExecutionState.isolation_level == :fiber
+        end
+      end
+
       attr_reader :size
 
       def initialize(size, on_state_change: nil)
@@ -28,6 +51,7 @@ def initialize(size, on_state_change: nil)
         @boot_queue = Thread::Queue.new
 
         load_dependency!
+        self.class.ensure_supported_isolation_level!
 
         @queue = Async::Queue.new
         @reactor_thread = start_reactor

lib/solid_queue/worker.rb

@@ -17,11 +17,13 @@ def initialize(**options)
 
       # Ensure that the queues array is deep frozen to prevent accidental modification
       @queues = Array(options[:queues]).map(&:freeze).freeze
+      @metadata_state_mutex = Mutex.new
+      @metadata_dirty = false
 
       @pool_options = {
         mode: options[:execution_mode],
         size: options[:threads],
-        on_state_change: -> { wake_up }
+        on_state_change: -> { mark_metadata_dirty; wake_up }
       }
 
       super(**options)
@@ -38,7 +40,7 @@ def poll
             pool.post(execution)
           end
 
-          reload_metadata if executions.any?
+          reload_metadata_if_needed(executions.any?)
 
           pool.idle? ? polling_interval : 10.minutes
         end
@@ -78,5 +80,23 @@ def set_procline
       def build_pool
         @pool ||= ExecutionPools.build(**@pool_options)
       end
+
+      def mark_metadata_dirty
+        metadata_state_mutex.synchronize { @metadata_dirty = true }
+      end
+
+      def metadata_state_mutex
+        @metadata_state_mutex
+      end
+
+      def reload_metadata_if_needed(executions_claimed)
+        needs_reload = metadata_state_mutex.synchronize do
+          claimed_or_dirty = executions_claimed || @metadata_dirty
+          @metadata_dirty = false
+          claimed_or_dirty
+        end
+
+        reload_metadata if needs_reload
+      end
   end
 end

test/test_helper.rb

@@ -29,6 +29,19 @@ def write(...)
 Logger::LogDevice.prepend(BlockLogDeviceTimeoutExceptions)
 class ExpectedTestError < RuntimeError; end
 
+module ExecutionIsolationTestHelper
+  def with_execution_isolation(level)
+    previous_level = ActiveSupport::IsolatedExecutionState.isolation_level
+    ActiveSupport::IsolatedExecutionState.isolation_level = level
+    yield
+  ensure
+    ActiveSupport::IsolatedExecutionState.isolation_level = previous_level
+  end
+end
+
+class Minitest::Test
+  include ExecutionIsolationTestHelper
+end
 
 class ActiveSupport::TestCase
   include ConfigurationTestHelper, ProcessesTestHelper, JobsTestHelper

test/unit/configuration_test.rb

@@ -62,27 +62,31 @@ class ConfigurationTest < ActiveSupport::TestCase
   end
 
   test "normalize worker execution modes and capacity aliases" do
-    configuration = SolidQueue::Configuration.new(
-      workers: [
-        { queues: "llm*", execution_mode: :async, capacity: 10 },
-        { queues: "*", execution_mode: :fiber, fibers: 3 }
-      ],
-      dispatchers: [],
-      skip_recurring: true
-    )
+    with_execution_isolation(:fiber) do
+      configuration = SolidQueue::Configuration.new(
+        workers: [
+          { queues: "llm*", execution_mode: :async, capacity: 10 },
+          { queues: "*", execution_mode: :fiber, fibers: 3 }
+        ],
+        dispatchers: [],
+        skip_recurring: true
+      )
 
-    assert configuration.valid?
-    assert_processes configuration, :worker, 2, execution_mode: [ :async, :async ], capacity: [ 10, 3 ], threads: [ nil, nil ]
+      assert configuration.valid?
+      assert_processes configuration, :worker, 2, execution_mode: [ :async, :async ], capacity: [ 10, 3 ], threads: [ nil, nil ]
+    end
   end
 
   test "async worker capacity does not inflate required database pool size" do
-    configuration = SolidQueue::Configuration.new(
-      workers: [ { queues: "llm*", execution_mode: :async, capacity: 1000 } ],
-      dispatchers: [],
-      skip_recurring: true
-    )
+    with_execution_isolation(:fiber) do
+      configuration = SolidQueue::Configuration.new(
+        workers: [ { queues: "llm*", execution_mode: :async, capacity: 1000 } ],
+        dispatchers: [],
+        skip_recurring: true
+      )
 
-    assert configuration.valid?
+      assert configuration.valid?
+    end
   end
 
   test "mulitple workers with the same configuration" do
@@ -193,6 +197,10 @@ class ConfigurationTest < ActiveSupport::TestCase
     assert_not configuration.valid?
     assert_match /Unknown execution mode/, configuration.errors.full_messages.first
 
+    configuration = SolidQueue::Configuration.new(skip_recurring: true, dispatchers: [], workers: [ { execution_mode: :async } ])
+    assert_not configuration.valid?
+    assert_match /requires fiber-scoped isolated execution state/, configuration.errors.full_messages.first
+
     # Not enough DB connections
     configuration = SolidQueue::Configuration.new(workers: [ { queues: "background", threads: 50, polling_interval: 10 } ])
     assert_not configuration.valid?

test/unit/execution_pools/async_pool_test.rb

@@ -36,57 +36,74 @@ def test_build_treats_fiber_as_an_alias_for_async
     assert_equal pool, SolidQueue::ExecutionPools.build(mode: :fiber, size: 5)
   end
 
-  def test_executes_jobs_as_fibers_on_a_single_reactor_thread
-    pool = SolidQueue::ExecutionPools::AsyncPool.new(2)
-    results = Thread::Queue.new
-
-    pool.post Execution.new(nil, results, 0.05)
-    pool.post Execution.new(nil, results, 0.05)
+  def test_raises_a_clear_error_when_isolation_level_is_not_fiber
+    error = assert_raises SolidQueue::ExecutionPools::AsyncPool::UnsupportedIsolationLevelError do
+      SolidQueue::ExecutionPools::AsyncPool.new(3)
+    end
 
-    entries = 2.times.map { Timeout.timeout(1.second) { results.pop } }
+    assert_match /isolation_level = :fiber/, error.message
+  end
 
-    assert_equal 1, entries.map(&:first).uniq.count
-    assert_equal 2, entries.map(&:last).uniq.count
-    assert_equal 2, pool.available_capacity
-    assert_equal 0, pool.metadata[:inflight]
-  ensure
-    pool&.shutdown
-    pool&.wait_for_termination(1.second)
+  def test_executes_jobs_as_fibers_on_a_single_reactor_thread
+    with_execution_isolation(:fiber) do
+      pool = SolidQueue::ExecutionPools::AsyncPool.new(2)
+      results = Thread::Queue.new
+
+      pool.post Execution.new(nil, results, 0.05)
+      pool.post Execution.new(nil, results, 0.05)
+
+      entries = 2.times.map { Timeout.timeout(1.second) { results.pop } }
+
+      assert_equal 1, entries.map(&:first).uniq.count
+      assert_equal 2, entries.map(&:last).uniq.count
+      assert_equal 2, pool.available_capacity
+      assert_equal 0, pool.metadata[:inflight]
+    ensure
+      pool&.shutdown
+      pool&.wait_for_termination(1.second)
+    end
   end
 
   def test_waits_for_in_flight_executions_during_shutdown
-    pool = SolidQueue::ExecutionPools::AsyncPool.new(1)
-    started = Thread::Queue.new
+    with_execution_isolation(:fiber) do
+      pool = SolidQueue::ExecutionPools::AsyncPool.new(1)
+      started = Thread::Queue.new
 
-    pool.post Execution.new(started, nil, 0.1)
-    Timeout.timeout(1.second) { started.pop }
+      pool.post Execution.new(started, nil, 0.1)
+      Timeout.timeout(1.second) { started.pop }
 
-    pool.shutdown
+      pool.shutdown
 
-    assert_nil pool.wait_for_termination(0.01)
-    assert pool.wait_for_termination(1.second)
+      assert_nil pool.wait_for_termination(0.01)
+      assert pool.wait_for_termination(1.second)
+    ensure
+      pool&.shutdown
+      pool&.wait_for_termination(1.second)
+    end
   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)
+    with_execution_isolation(:fiber) do
+      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
 end