Add workflow check that local activity exists (#263)

Author: tconley1428

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

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

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

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(

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

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

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

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

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?

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?