diff --git a/CHANGELOG.md b/CHANGELOG.md index dd478451..1937ab36 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -47,6 +47,11 @@ to docs, or any other relevant information. - `Temporalio::Client::WorkflowHandle#result` no longer returns `nil` while a workflow is still running when a history long poll expires without returning an event. +- Canceling a fiber-executor activity no longer wedges the worker. Cancellation was delivered with + `Fiber#raise` from the worker's task-dispatch fiber, which under a scheduler that resumes fibers + with `Fiber#transfer` (such as `async`) never returns, permanently stopping the worker from + dispatching any further activity tasks or workflow activations. The exception is now delivered + through the scheduler's `fiber_interrupt` hook when it provides one. ## [v1.6.0] - 2026-07-16 diff --git a/temporalio/lib/temporalio/worker/activity_executor/fiber.rb b/temporalio/lib/temporalio/worker/activity_executor/fiber.rb index ad4f7508..5ac82c8b 100644 --- a/temporalio/lib/temporalio/worker/activity_executor/fiber.rb +++ b/temporalio/lib/temporalio/worker/activity_executor/fiber.rb @@ -39,8 +39,18 @@ def set_activity_context(defn, context) return unless defn.cancel_raise fiber = ::Fiber.current + scheduler = ::Fiber.scheduler + scheduler = nil unless scheduler.respond_to?(:fiber_interrupt) context&.cancellation&.add_cancel_callback do - fiber.raise(Error::CanceledError.new('Activity canceled')) + error = Error::CanceledError.new('Activity canceled') + # Directly raising from another fiber can strand a `Fiber#transfer` + # based scheduler's current fiber, so we defer to the scheduler to interrupt. + # If on the same fiber, we can just raise directly. + if scheduler.nil? || ::Fiber.current.equal?(fiber) + fiber.raise(error) + else + scheduler.fiber_interrupt(fiber, error) + end end end end diff --git a/temporalio/test/worker_activity_test.rb b/temporalio/test/worker_activity_test.rb index 83dd72d1..5ff743ea 100644 --- a/temporalio/test/worker_activity_test.rb +++ b/temporalio/test/worker_activity_test.rb @@ -475,6 +475,35 @@ def test_activity_shielding end end + class FiberShieldingActivity < ShieldingActivity + activity_executor :fiber # steep:ignore + end + + def test_activity_shielding_fiber + skip_if_fibers_not_supported! + + Async do + act = FiberShieldingActivity.new + execute_activity( + act, + cancel_on_signal: 'cancel-activity', + wait_for_cancellation: true, + heartbeat_timeout: 0.8 + ) do |handle| + # Wait for it to be waiting + act.wait_until_waiting + # Send activity cancel + handle.signal('cancel-activity') + # Wait for completion + error = assert_raises(Temporalio::Error::WorkflowFailedError) { handle.result } + assert_kind_of Temporalio::Error::CanceledError, error.cause + # Confirm thrown in activity but the proper levels reached + assert act.canceled + assert_equal 2, act.levels_reached + end + end + end + class NoRaiseCancellationActivity < Temporalio::Activity::Definition activity_cancel_raise false attr_reader :canceled @@ -520,6 +549,57 @@ def test_no_raise_cancellation end end + class FiberCancellationActivity < Temporalio::Activity::Definition + activity_executor :fiber + + attr_reader :canceled + + def initialize + @started = Queue.new + @canceled = false + end + + def execute + @started.push(nil) + # Heartbeat every 50ms waiting for cancel + loop do + sleep(0.05) + Temporalio::Activity::Context.current.heartbeat + end + rescue Temporalio::Error::CanceledError + @canceled = true + raise + end + + def wait_started + @started.pop + end + end + + def test_fiber_cancellation + skip_if_fibers_not_supported! + # Tests are doubly executed in threaded and fiber, so we start a new Async block just in case + Async do + act = FiberCancellationActivity.new + execute_activity( + act, + cancel_on_signal: 'cancel-activity', + wait_for_cancellation: true, + heartbeat_timeout: 0.8 + ) do |handle| + # Wait for it to start + act.wait_started + # Send activity cancel + handle.signal('cancel-activity') + # The workflow can only reach a terminal state if the worker's task-dispatch fiber + # survived delivering the cancellation to the activity fiber + error = assert_raises(Temporalio::Error::WorkflowFailedError) { handle.result } + assert_kind_of Temporalio::Error::CanceledError, error.cause + assert act.canceled + end + end + end + class WorkerShutdownActivity < Temporalio::Activity::Definition attr_reader :canceled