From 1ce7c9d0e16627f9f1501d0fc162af26cefec151 Mon Sep 17 00:00:00 2001 From: AnushkaKulkarni Date: Mon, 27 Jul 2026 15:36:25 -0400 Subject: [PATCH] fix: preserve workflow task failure errors through illegal call validator rescues Co-authored-by: Cursor --- .../workflow_instance/illegal_call_tracer.rb | 9 +++ temporalio/test/deadlock_test.rb | 58 +++++++++++++++++++ 2 files changed, 67 insertions(+) diff --git a/temporalio/lib/temporalio/internal/worker/workflow_instance/illegal_call_tracer.rb b/temporalio/lib/temporalio/internal/worker/workflow_instance/illegal_call_tracer.rb index 2852bd89..f87bbf7d 100644 --- a/temporalio/lib/temporalio/internal/worker/workflow_instance/illegal_call_tracer.rb +++ b/temporalio/lib/temporalio/internal/worker/workflow_instance/illegal_call_tracer.rb @@ -1,5 +1,6 @@ # frozen_string_literal: true +require 'temporalio/internal/workflow_task_failure_error' require 'temporalio/worker/illegal_workflow_call_validator' require 'temporalio/workflow' @@ -92,6 +93,11 @@ def initialize(illegal_calls) )) nil rescue Exception => e # rubocop:disable Lint/RescueException + # Errors that invalidate the whole workflow task (e.g. an async DeadlockError landing while the + # validator runs) must propagate with their class intact so they fail the task instead of being + # flattened into a NondeterminismError message that futures may store + raise if e.is_a?(Internal::WorkflowTaskFailureError) + ", reason: #{e}" end else @@ -108,6 +114,9 @@ def initialize(illegal_calls) )) nil rescue Exception => e # rubocop:disable Lint/RescueException + # See workflow-task-failure comment on the class-level validator rescue above + raise if e.is_a?(Internal::WorkflowTaskFailureError) + ", reason: #{e}" end end diff --git a/temporalio/test/deadlock_test.rb b/temporalio/test/deadlock_test.rb index 52c9f1b3..eb51623d 100644 --- a/temporalio/test/deadlock_test.rb +++ b/temporalio/test/deadlock_test.rb @@ -65,6 +65,22 @@ def execute end end + module ValidatorInterruptProbe + def self.probe; end + end + + class DeadlockDuringValidatorWorkflow < Temporalio::Workflow::Definition + def execute + Temporalio::Workflow::Future.new do + ValidatorInterruptProbe.probe + Temporalio::Workflow.execute_activity(BasicActivity, 1, start_to_close_timeout: 10) + end + # Future intentionally not waited (mirrors a fanout blocked on all_of with other + # futures pending) so a stored failure would never surface + Temporalio::Workflow.wait_condition { false } + end + end + class ProtobufObjectCachePartialCommandsWorkflow < Temporalio::Workflow::Definition ACTIVITY_COUNT = 6 BLOCK_AFTER_ACTIVITY_COUNT = 3 @@ -144,6 +160,48 @@ def test_deadlock_in_future_fails_workflow_task_and_replays_on_new_worker handle&.terminate end + def test_deadlock_during_illegal_call_validator_fails_workflow_task + task_queue = "tq-#{SecureRandom.uuid}" + # Simulates the deadlock watchdog's async raise landing while an illegal call + # validator block is executing (e.g. Thread::Mutex#synchronize inside protobuf's + # ObjectCache during ScheduleActivity arg serialization) + illegal_calls = Temporalio::Worker.default_illegal_workflow_calls.merge( + 'DeadlockTest::ValidatorInterruptProbe' => [ + Temporalio::Worker::IllegalWorkflowCallValidator.new(method_name: :probe) do |_info| + raise Temporalio::Worker::WorkflowExecutor::ThreadPool::DeadlockError, + '[TMPRL1101] Potential deadlock detected: simulated async deadlock raise' + end + ] + ) + worker = Temporalio::Worker.new( + client: env.client, + task_queue:, + workflows: [DeadlockDuringValidatorWorkflow], + activities: [BasicActivity], + illegal_workflow_calls: illegal_calls + ) + # @type var handle: untyped + handle = nil + worker.run do + handle = env.client.start_workflow( + DeadlockDuringValidatorWorkflow, + id: "wf-#{SecureRandom.uuid}", + task_queue: + ) + assert_eventually_task_fail(handle:, message_contains: 'Potential deadlock detected') + # The DeadlockError class must survive the tracer. A laundered + # NondeterminismError would read "Cannot access ..." and, when raised inside a + # future, be stored instead of failing the task at all + failure_messages = handle.fetch_history_events.filter_map do |event| + event.workflow_task_failed_event_attributes&.failure&.message + end + refute_empty(failure_messages) + failure_messages.each { |message| refute_includes(message, 'Cannot access') } + end + ensure + handle&.terminate + end + def test_protobuf_object_cache_mutex_does_not_emit_partial_command_batch task_queue = "tq-#{SecureRandom.uuid}" # @type var release_mutex: (^() -> void)?