Add workflow check that local activity exists

tconley1428 · tconley1428 · commit 3e7249e1420d · 2025-05-12T09:55:27.000-07:00
diff --git a/temporalio/lib/temporalio/internal/worker/activity_worker.rb b/temporalio/lib/temporalio/internal/worker/activity_worker.rb
@@ -297,6 +297,17 @@ def run_activity(defn, activity, input)
           )
         end
 
+        def assert_valid_activity(activity)
+          defn = @activities[activity]
+          defn = @activities[nil] if !defn && !Internal::ProtoUtils.reserved_name?(activity)
+
+          return unless defn.nil?
+
+          raise ArgumentError,
+                "Activity #{activity} " \
+                "is not registered on this worker, available activities: #{@activities.keys.sort.join(', ')}"
+        end
+
         class RunningActivity < Activity::Context
           attr_reader :info, :cancellation, :worker_shutdown_cancellation, :payload_converter, :logger
           attr_accessor :instance, :_outbound_impl, :_server_requested_cancel
diff --git a/temporalio/lib/temporalio/internal/worker/workflow_instance.rb b/temporalio/lib/temporalio/internal/worker/workflow_instance.rb
@@ -56,7 +56,7 @@ def self.new_completion_with_failure(run_id:, error:, failure_converter:, payloa
                     :pending_external_signals, :pending_external_cancels, :in_progress_handlers, :payload_converter,
                     :failure_converter, :cancellation, :continue_as_new_suggested, :current_history_length,
                     :current_history_size, :replaying, :random, :signal_handlers, :query_handlers, :update_handlers,
-                    :context_frozen
+                    :context_frozen, :assert_valid_local_activity
         attr_accessor :io_enabled, :current_details
 
         def initialize(details)
@@ -108,6 +108,8 @@ def initialize(details)
           @query_handlers = HandlerHash.new(details.definition.queries, Workflow::Definition::Query)
           @update_handlers = HandlerHash.new(details.definition.updates, Workflow::Definition::Update)
 
+          @assert_valid_local_activity = details.assert_valid_local_activity
+
           # Create all things needed from initial job
           @init_job = details.initial_activation.jobs.find { |j| !j.initialize_workflow.nil? }&.initialize_workflow
           raise 'Missing init job from first activation' unless @init_job
diff --git a/temporalio/lib/temporalio/internal/worker/workflow_instance/details.rb b/temporalio/lib/temporalio/internal/worker/workflow_instance/details.rb
@@ -8,7 +8,8 @@ class WorkflowInstance
         class Details
           attr_reader :namespace, :task_queue, :definition, :initial_activation, :logger, :metric_meter,
                       :payload_converter, :failure_converter, :interceptors, :disable_eager_activity_execution,
-                      :illegal_calls, :workflow_failure_exception_types, :unsafe_workflow_io_enabled
+                      :illegal_calls, :workflow_failure_exception_types, :unsafe_workflow_io_enabled,
+                      :assert_valid_local_activity
 
           def initialize(
             namespace:,
@@ -23,7 +24,8 @@ def initialize(
             disable_eager_activity_execution:,
             illegal_calls:,
             workflow_failure_exception_types:,
-            unsafe_workflow_io_enabled:
+            unsafe_workflow_io_enabled:,
+            assert_valid_local_activity:
           )
             @namespace = namespace
             @task_queue = task_queue
@@ -38,6 +40,7 @@ def initialize(
             @illegal_calls = illegal_calls
             @workflow_failure_exception_types = workflow_failure_exception_types
             @unsafe_workflow_io_enabled = unsafe_workflow_io_enabled
+            @assert_valid_local_activity = assert_valid_local_activity
           end
         end
       end
diff --git a/temporalio/lib/temporalio/internal/worker/workflow_instance/outbound_implementation.rb b/temporalio/lib/temporalio/internal/worker/workflow_instance/outbound_implementation.rb
@@ -87,6 +87,8 @@ def execute_local_activity(input)
               raise ArgumentError, 'Activity must have schedule_to_close_timeout or start_to_close_timeout'
             end
 
+            @instance.assert_valid_local_activity.call(input.activity)
+
             execute_activity_with_local_backoffs(local: true, cancellation: input.cancellation) do |do_backoff|
               seq = (@activity_counter += 1)
               @instance.add_command(
diff --git a/temporalio/lib/temporalio/internal/worker/workflow_worker.rb b/temporalio/lib/temporalio/internal/worker/workflow_worker.rb
@@ -69,7 +69,7 @@ def initialize(
           workflow_payload_codec_thread_pool:,
           unsafe_workflow_io_enabled:,
           debug_mode:,
-          on_eviction: nil
+          assert_valid_local_activity:, on_eviction: nil
         )
           @executor = workflow_executor
 
@@ -111,7 +111,8 @@ def initialize(
 
               t
             end.freeze,
-            unsafe_workflow_io_enabled:
+            unsafe_workflow_io_enabled:,
+            assert_valid_local_activity:
           )
           @state.on_eviction = on_eviction if on_eviction
 
@@ -186,14 +187,16 @@ def apply_codec_on_payload_visit(payload_or_payloads, &)
         class State
           attr_reader :workflow_definitions, :bridge_worker, :logger, :metric_meter, :data_converter, :deadlock_timeout,
                       :illegal_calls, :namespace, :task_queue, :disable_eager_activity_execution,
-                      :workflow_interceptors, :workflow_failure_exception_types, :unsafe_workflow_io_enabled
+                      :workflow_interceptors, :workflow_failure_exception_types, :unsafe_workflow_io_enabled,
+                      :assert_valid_local_activity
 
           attr_writer :on_eviction
 
           def initialize(
             workflow_definitions:, bridge_worker:, logger:, metric_meter:, data_converter:, deadlock_timeout:,
             illegal_calls:, namespace:, task_queue:, disable_eager_activity_execution:,
-            workflow_interceptors:, workflow_failure_exception_types:, unsafe_workflow_io_enabled:
+            workflow_interceptors:, workflow_failure_exception_types:, unsafe_workflow_io_enabled:,
+            assert_valid_local_activity:
           )
             @workflow_definitions = workflow_definitions
             @bridge_worker = bridge_worker
@@ -208,6 +211,7 @@ def initialize(
             @workflow_interceptors = workflow_interceptors
             @workflow_failure_exception_types = workflow_failure_exception_types
             @unsafe_workflow_io_enabled = unsafe_workflow_io_enabled
+            @assert_valid_local_activity = assert_valid_local_activity
 
             @running_workflows = {}
             @running_workflows_mutex = Mutex.new
diff --git a/temporalio/lib/temporalio/worker.rb b/temporalio/lib/temporalio/worker.rb
@@ -490,7 +490,8 @@ def initialize(
           workflow_failure_exception_types:,
           workflow_payload_codec_thread_pool:,
           unsafe_workflow_io_enabled:,
-          debug_mode:
+          debug_mode:,
+          assert_valid_local_activity: ->(activity) { _assert_valid_local_activity(activity) }
         )
       end
 
@@ -599,5 +600,17 @@ def _on_shutdown_complete
       @workflow_worker&.on_shutdown_complete
       @workflow_worker = nil
     end
+
+    # @!visibility private
+    def _assert_valid_local_activity(activity)
+      unless @activity_worker.nil?
+        @activity_worker.assert_valid_activity(activity)
+        return
+      end
+
+      raise ArgumentError,
+            "Activity #{activity} " \
+            'is not registered on this worker, no available activities.'
+    end
   end
 end
diff --git a/temporalio/lib/temporalio/worker/workflow_executor/thread_pool.rb b/temporalio/lib/temporalio/worker/workflow_executor/thread_pool.rb
@@ -214,7 +214,8 @@ def create_instance(initial_activation, worker_state)
                 disable_eager_activity_execution: worker_state.disable_eager_activity_execution,
                 illegal_calls: worker_state.illegal_calls,
                 workflow_failure_exception_types: worker_state.workflow_failure_exception_types,
-                unsafe_workflow_io_enabled: worker_state.unsafe_workflow_io_enabled
+                unsafe_workflow_io_enabled: worker_state.unsafe_workflow_io_enabled,
+                assert_valid_local_activity: worker_state.assert_valid_local_activity
               )
             )
           end
diff --git a/temporalio/lib/temporalio/worker/workflow_replayer.rb b/temporalio/lib/temporalio/worker/workflow_replayer.rb
@@ -245,7 +245,8 @@ def initialize(
             workflow_payload_codec_thread_pool: options.workflow_payload_codec_thread_pool,
             unsafe_workflow_io_enabled: options.unsafe_workflow_io_enabled,
             debug_mode: options.debug_mode,
-            on_eviction: proc { |_, remove_job| @last_workflow_remove_job = remove_job } # steep:ignore
+            on_eviction: proc { |_, remove_job| @last_workflow_remove_job = remove_job }, # steep:ignore
+            assert_valid_local_activity: ->(_) {}
           )
 
           # Create the runner
diff --git a/temporalio/sig/temporalio/internal/worker/activity_worker.rbs b/temporalio/sig/temporalio/internal/worker/activity_worker.rbs
@@ -26,6 +26,8 @@ module Temporalio
           Temporalio::Worker::Interceptor::Activity::ExecuteInput input
         ) -> void
 
+        def assert_valid_activity: (String activity) -> void
+
         class RunningActivity < Activity::Context
           attr_accessor instance: Activity::Definition?
           attr_accessor _outbound_impl: Temporalio::Worker::Interceptor::Activity::Outbound?
diff --git a/temporalio/sig/temporalio/internal/worker/workflow_instance.rbs b/temporalio/sig/temporalio/internal/worker/workflow_instance.rbs
@@ -33,6 +33,7 @@ module Temporalio
         attr_reader query_handlers: Hash[String?, Workflow::Definition::Query]
         attr_reader update_handlers: Hash[String?, Workflow::Definition::Update]
         attr_reader context_frozen: bool
+        attr_reader assert_valid_local_activity: ^(String) -> void
 
         attr_accessor io_enabled: bool
         attr_accessor current_details: String?
diff --git a/temporalio/sig/temporalio/internal/worker/workflow_instance/details.rbs b/temporalio/sig/temporalio/internal/worker/workflow_instance/details.rbs
@@ -16,6 +16,7 @@ module Temporalio
           attr_reader illegal_calls: Hash[String, :all | Hash[Symbol, bool]]
           attr_reader workflow_failure_exception_types: Array[singleton(Exception)]
           attr_reader unsafe_workflow_io_enabled: bool
+          attr_reader assert_valid_local_activity: ^(String) -> void
 
           def initialize: (
             namespace: String,
@@ -30,7 +31,8 @@ module Temporalio
             disable_eager_activity_execution: bool,
             illegal_calls: Hash[String, :all | Hash[Symbol, bool]],
             workflow_failure_exception_types: Array[singleton(Exception)],
-            unsafe_workflow_io_enabled: bool
+            unsafe_workflow_io_enabled: bool,
+            assert_valid_local_activity: ^(String) -> void
           ) -> void
         end
       end
diff --git a/temporalio/sig/temporalio/internal/worker/workflow_worker.rbs b/temporalio/sig/temporalio/internal/worker/workflow_worker.rbs
@@ -27,7 +27,8 @@ module Temporalio
           workflow_payload_codec_thread_pool: Temporalio::Worker::ThreadPool?,
           unsafe_workflow_io_enabled: bool,
           debug_mode: bool,
-          ?on_eviction: (^(String run_id, untyped cache_remove_job) -> void)?
+          ?on_eviction: (^(String run_id, untyped cache_remove_job) -> void)?,
+          assert_valid_local_activity: ^(String) -> void
         ) -> void
 
         def handle_activation: (
@@ -63,6 +64,7 @@ module Temporalio
           attr_reader workflow_interceptors: Array[Temporalio::Worker::Interceptor::Workflow]
           attr_reader workflow_failure_exception_types: Array[singleton(Exception)]
           attr_reader unsafe_workflow_io_enabled: bool
+          attr_reader assert_valid_local_activity: ^(String) -> void
 
           attr_writer on_eviction: ^(String run_id, untyped cache_remove_job) -> void
 
@@ -79,7 +81,8 @@ module Temporalio
             disable_eager_activity_execution: bool,
             workflow_interceptors: Array[Temporalio::Worker::Interceptor::Workflow],
             workflow_failure_exception_types: Array[singleton(Exception)],
-            unsafe_workflow_io_enabled: bool
+            unsafe_workflow_io_enabled: bool,
+            assert_valid_local_activity: ^(String) -> void
           ) -> void
 
           def get_or_create_running_workflow: [T] (String run_id) { -> T } -> T
diff --git a/temporalio/sig/temporalio/worker.rbs b/temporalio/sig/temporalio/worker.rbs
@@ -132,5 +132,6 @@ module Temporalio
     def _activity_interceptors: -> Array[Interceptor::Activity]
     def _on_poll_bytes: (Internal::Worker::MultiRunner runner, Symbol worker_type, String bytes) -> void
     def _on_shutdown_complete: -> void
+    def _assert_valid_local_activity: (String) -> void
   end
 end
diff --git a/temporalio/test/worker_workflow_test.rb b/temporalio/test/worker_workflow_test.rb
@@ -9,7 +9,7 @@
 require 'test'
 require 'timeout'
 
-class WorkerWorkflowTest < Test
+class WorkerWorkflowTest < Test # rubocop:disable Metrics/ClassLength
   class SimpleWorkflow < Temporalio::Workflow::Definition
     def execute(name)
       "Hello, #{name}!"
@@ -2285,17 +2285,50 @@ def test_unsafe_io
     assert_includes res, '<html>'
   end
 
-  # TODO(cretz): To test
-  # * Common
-  #   * Eager workflow start
-  #   * Unawaited futures that have exceptions, need to log warning like Java does
-  #   * Enhanced stack trace?
-  #   * Separate abstract/interface demonstration
-  #   * Reset update randomness seed
-  #   * Confirm thread pool does not leak, meaning thread/worker goes away after last workflow
-  #   * Test workflow cancel causing other cancels at the same time but in different coroutines
-  #   * 0-sleep timers vs nil timers
-  #   * Interceptors
-  # * Activity
-  #   * Local activity cancel (currently broken)
+  class MissingLocalActivityWorkflow < Temporalio::Workflow::Definition
+    def execute
+      # Start 30 10ms timers and wait on them all
+      Temporalio::Workflow.execute_local_activity('does-not-exist',
+                                                  start_to_close_timeout: 1000)
+    end
+  end
+
+  def test_missing_local_activity
+    activities = [PatchPreActivity]
+    message_contains = 'Activity does-not-exist is not registered on this worker, ' \
+                       'available activities: PatchPreActivity'
+
+    execute_workflow(MissingLocalActivityWorkflow, activities:) do |handle|
+      assert_eventually_task_fail(handle:,
+                                  message_contains:)
+    end
+  end
+
+  def test_dynamic_local_activity
+    activities = [ReservedNameDynamicActivity]
+
+    assert_equal 'done', execute_workflow(MissingLocalActivityWorkflow, activities:)
+  end
+
+  def test_no_local_activity
+    message_contains = 'Activity does-not-exist is not registered on this worker, no available activities.'
+
+    execute_workflow(MissingLocalActivityWorkflow, false) do |handle|
+      assert_eventually_task_fail(handle:, message_contains:)
+    end
+  end
 end
+
+# TODO(cretz): To test
+# * Common
+#   * Eager workflow start
+#   * Unawaited futures that have exceptions, need to log warning like Java does
+#   * Enhanced stack trace?
+#   * Separate abstract/interface demonstration
+#   * Reset update randomness seed
+#   * Confirm thread pool does not leak, meaning thread/worker goes away after last workflow
+#   * Test workflow cancel causing other cancels at the same time but in different coroutines
+#   * 0-sleep timers vs nil timers
+#   * Interceptors
+# * Activity
+#   * Local activity cancel (currently broken)

Original file line number	Diff line number	Diff line change
`@@ -214,7 +214,8 @@ def create_instance(initial_activation, worker_state)`
`214`	`214`	`disable_eager_activity_execution: worker_state.disable_eager_activity_execution,`
`215`	`215`	`illegal_calls: worker_state.illegal_calls,`
`216`	`216`	`workflow_failure_exception_types: worker_state.workflow_failure_exception_types,`
`217`		`- unsafe_workflow_io_enabled: worker_state.unsafe_workflow_io_enabled`
	`217`	`+ unsafe_workflow_io_enabled: worker_state.unsafe_workflow_io_enabled,`
	`218`	`+ assert_valid_local_activity: worker_state.assert_valid_local_activity`
`218`	`219`	`)`
`219`	`220`	`)`
`220`	`221`	`end`
Original file line number	Diff line number	Diff line change
`@@ -245,7 +245,8 @@ def initialize(`
`245`	`245`	`workflow_payload_codec_thread_pool: options.workflow_payload_codec_thread_pool,`
`246`	`246`	`unsafe_workflow_io_enabled: options.unsafe_workflow_io_enabled,`
`247`	`247`	`debug_mode: options.debug_mode,`
`248`		`- on_eviction: proc { \|_, remove_job\| @last_workflow_remove_job = remove_job } # steep:ignore`
	`248`	`+ on_eviction: proc { \|_, remove_job\| @last_workflow_remove_job = remove_job }, # steep:ignore`
	`249`	`+ assert_valid_local_activity: ->(_) {}`
`249`	`250`	`)`
`250`	`251`
`251`	`252`	`# Create the runner`