diff --git a/CHANGELOG.md b/CHANGELOG.md index af1c7b51..8d2ebb9f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -21,6 +21,11 @@ to docs, or any other relevant information. ### Added +#### Standalone Activity operator commands + +- `Client::ActivityHandle` now supports operator commands for standalone activities: `#pause`, + `#unpause`, `#reset`, and `#update_options`. + ### Changed ### Deprecated diff --git a/temporalio/lib/temporalio/client/activity_execution.rb b/temporalio/lib/temporalio/client/activity_execution.rb index 36fe45d2..e9765d8f 100644 --- a/temporalio/lib/temporalio/client/activity_execution.rb +++ b/temporalio/lib/temporalio/client/activity_execution.rb @@ -46,6 +46,12 @@ def schedule_time Internal::ProtoUtils.timestamp_to_time(@raw_info.schedule_time) end + # @return [Time, nil] When the first activity task was made available for dispatch. Equals + # schedule_time + start_delay; equal to schedule_time when no start delay is set. + def execution_time + Internal::ProtoUtils.timestamp_to_time(@raw_info.execution_time) + end + # @return [Time, nil] When the activity reached a terminal state. def close_time Internal::ProtoUtils.timestamp_to_time(@raw_info.close_time) @@ -112,7 +118,17 @@ def heartbeat_timeout Internal::ProtoUtils.duration_to_seconds(@raw_info.heartbeat_timeout) end - # @return [Boolean] Whether the activity has recorded any heartbeat details. + # @return [Float, nil] Delay in seconds before the first activity task is made available for + # dispatch. Not applied to retry attempts. + def start_delay + Internal::ProtoUtils.duration_to_seconds(@raw_info.start_delay) + end + + # Whether heartbeat details are present on this description. False when the activity + # recorded none, and also when {ActivityHandle#describe} was called without + # `include_heartbeat_details:`. + # + # @return [Boolean] Whether heartbeat details are present. def has_heartbeat_details? # rubocop:disable Naming/PredicatePrefix !@raw_info.heartbeat_details&.payloads.nil? && !@raw_info.heartbeat_details.payloads.empty? end @@ -125,6 +141,57 @@ def heartbeat_details(hints: nil) @data_converter.from_payloads(@raw_info.heartbeat_details, hints:) end + # Whether the activity's input is present. False unless {ActivityHandle#describe} was + # called with `include_input:`. + # + # @return [Boolean] Whether input is present. + def has_input? # rubocop:disable Naming/PredicatePrefix + !@raw_description.input.nil? + end + + # Deserialized activity input, one element per argument. Empty when no input is present. + # + # @param hints [Array, nil] Hints, if any, to assist conversion. + # @return [Array] Converted arguments. + def input(hints: nil) + @data_converter.from_payloads(@raw_description.input, hints:) + end + + # Whether the activity closed with a successful result. False while the activity is still + # running, when it closed with a failure, and when {ActivityHandle#describe} was called + # without `include_outcome:`. + # + # @return [Boolean] Whether a result is present. + def has_result? # rubocop:disable Naming/PredicatePrefix + @raw_description.outcome&.value == :result + end + + # Deserialized result the activity closed with. Nil when no result is present (still + # running, closed with a failure, or `include_outcome:` was not requested). + # + # @param result_hint [Object, nil] Hint, if any, to assist conversion. + # @return [Object, nil] Converted result. + def result(result_hint: nil) + return nil unless has_result? + + @data_converter.from_payloads( + @raw_description.outcome.result, hints: Array(result_hint) + ).first + end + + # Failure the activity closed with. Nil when the activity did not close with a failure or + # when {ActivityHandle#describe} was called without `include_outcome:`. + # + # This is the terminal outcome; {#last_failure} is the failure of the most recent attempt, + # which may be set while the activity is still retrying. + # + # @return [Error::Failure, nil] Converted failure. + def failure + return nil unless @raw_description.outcome&.value == :failure + + @data_converter.from_failure(@raw_description.outcome.failure) + end + # @return [RetryPolicy] Retry policy in effect for this activity. def retry_policy RetryPolicy._from_proto(@raw_info.retry_policy) @@ -145,6 +212,15 @@ def attempt @raw_info.attempt end + # Whether a last failure is present on this description. False when the activity has no + # failed attempt, and also when {ActivityHandle#describe} was called without + # `include_last_failure:`. + # + # @return [Boolean] Whether a last failure is present. + def has_last_failure? # rubocop:disable Naming/PredicatePrefix + !@raw_info.last_failure.nil? + end + # @return [Error::Failure, nil] Failure of the last failed attempt if any. def last_failure return nil unless @raw_info.last_failure diff --git a/temporalio/lib/temporalio/client/activity_execution_options.rb b/temporalio/lib/temporalio/client/activity_execution_options.rb new file mode 100644 index 00000000..f66e513f --- /dev/null +++ b/temporalio/lib/temporalio/client/activity_execution_options.rb @@ -0,0 +1,39 @@ +# frozen_string_literal: true + +require 'temporalio/internal/proto_utils' +require 'temporalio/priority' +require 'temporalio/retry_policy' + +module Temporalio + class Client + # The resolved options of a standalone activity execution, as returned by + # {ActivityHandle#update_options}. Reflects the activity's options as the server resolved them + # after the update was applied. + # + # WARNING: Standalone Activities are experimental. + ActivityExecutionOptions = Data.define( + :task_queue, + :schedule_to_close_timeout, + :schedule_to_start_timeout, + :start_to_close_timeout, + :heartbeat_timeout, + :retry_policy, + :priority, + :start_delay + ) do + # @!visibility private + def self._from_proto(options) + new( + task_queue: Internal::ProtoUtils.string_or(options.task_queue&.name, nil), + schedule_to_close_timeout: Internal::ProtoUtils.duration_to_seconds(options.schedule_to_close_timeout), + schedule_to_start_timeout: Internal::ProtoUtils.duration_to_seconds(options.schedule_to_start_timeout), + start_to_close_timeout: Internal::ProtoUtils.duration_to_seconds(options.start_to_close_timeout), + heartbeat_timeout: Internal::ProtoUtils.duration_to_seconds(options.heartbeat_timeout), + retry_policy: options.retry_policy ? RetryPolicy._from_proto(options.retry_policy) : nil, + priority: Priority._from_proto(options.priority), + start_delay: Internal::ProtoUtils.duration_to_seconds(options.start_delay) + ) + end + end + end +end diff --git a/temporalio/lib/temporalio/client/activity_execution_status.rb b/temporalio/lib/temporalio/client/activity_execution_status.rb index e72ef436..3ed8deac 100644 --- a/temporalio/lib/temporalio/client/activity_execution_status.rb +++ b/temporalio/lib/temporalio/client/activity_execution_status.rb @@ -15,6 +15,7 @@ module ActivityExecutionStatus CANCELED = Api::Enums::V1::ActivityExecutionStatus::ACTIVITY_EXECUTION_STATUS_CANCELED TERMINATED = Api::Enums::V1::ActivityExecutionStatus::ACTIVITY_EXECUTION_STATUS_TERMINATED TIMED_OUT = Api::Enums::V1::ActivityExecutionStatus::ACTIVITY_EXECUTION_STATUS_TIMED_OUT + PAUSED = Api::Enums::V1::ActivityExecutionStatus::ACTIVITY_EXECUTION_STATUS_PAUSED end end end diff --git a/temporalio/lib/temporalio/client/activity_handle.rb b/temporalio/lib/temporalio/client/activity_handle.rb index 60c58a38..d8c1b302 100644 --- a/temporalio/lib/temporalio/client/activity_handle.rb +++ b/temporalio/lib/temporalio/client/activity_handle.rb @@ -2,8 +2,12 @@ require 'temporalio/api' require 'temporalio/client/activity_execution' +require 'temporalio/client/activity_execution_options' require 'temporalio/client/interceptor' require 'temporalio/error' +require 'temporalio/internal/proto_utils' +require 'temporalio/priority' +require 'temporalio/retry_policy' module Temporalio class Client @@ -12,6 +16,17 @@ class Client # # WARNING: Standalone Activities are experimental. class ActivityHandle + UPDATABLE_OPTION_PATHS = { + task_queue: 'task_queue.name', + schedule_to_close_timeout: 'schedule_to_close_timeout', + schedule_to_start_timeout: 'schedule_to_start_timeout', + start_to_close_timeout: 'start_to_close_timeout', + heartbeat_timeout: 'heartbeat_timeout', + retry_policy: 'retry_policy', + priority: 'priority', + start_delay: 'start_delay' + }.freeze + # @return [String] ID for the activity. attr_reader :id @@ -55,15 +70,35 @@ def result(result_hint: nil, rpc_options: nil) # Describe the activity. # + # The payload-bearing fields are opt-in because they can be arbitrarily large; request them + # only when needed. Each has a corresponding predicate on the returned description that + # reports whether the server supplied it. + # + # @param include_input [Boolean] If true and the activity received input, include the input. + # @param include_outcome [Boolean] If true and the activity is closed, include the outcome. + # @param include_heartbeat_details [Boolean] If true and the activity recorded heartbeat + # details, include them. + # @param include_last_failure [Boolean] If true and the activity has a failed attempt, include + # the last failure. # @param rpc_options [RPCOptions, nil] Advanced RPC options. # # @return [ActivityExecution::Description] Activity description. # @raise [Error::RPCError] RPC error from call. - def describe(rpc_options: nil) + def describe( + include_input: false, + include_outcome: false, + include_heartbeat_details: false, + include_last_failure: false, + rpc_options: nil + ) @client._impl.describe_activity( Interceptor::DescribeActivityInput.new( activity_id: id, activity_run_id: run_id, + include_input:, + include_outcome:, + include_heartbeat_details:, + include_last_failure:, rpc_options: ) ) @@ -103,6 +138,142 @@ def terminate(reason = nil, rpc_options: nil) nil end + # Pause the activity. A paused activity is not scheduled or retried until it is unpaused via + # {#unpause}. + # + # WARNING: Standalone Activities are experimental. + # + # @param reason [String, nil] Optional reason recorded on the server. + # @param rpc_options [RPCOptions, nil] Advanced RPC options. + # @raise [Error::RPCError] RPC error from call. + def pause(reason = nil, rpc_options: nil) + @client._impl.pause_activity( + Interceptor::PauseActivityInput.new( + activity_id: id, + activity_run_id: run_id, + reason:, + rpc_options: + ) + ) + nil + end + + # Unpause the activity, allowing it to be scheduled or retried again. + # + # WARNING: Standalone Activities are experimental. + # + # @param reason [String, nil] Optional reason recorded on the server. + # @param jitter [Float, nil] If set, the activity will start at a random time within this + # duration (in seconds). + # @param rpc_options [RPCOptions, nil] Advanced RPC options. + # @raise [Error::RPCError] RPC error from call. + def unpause(reason: nil, jitter: nil, rpc_options: nil) + @client._impl.unpause_activity( + Interceptor::UnpauseActivityInput.new( + activity_id: id, + activity_run_id: run_id, + reason:, + jitter:, + rpc_options: + ) + ) + nil + end + + # Reset the activity. Resetting sets the attempt count back to the start, resets the activity's + # timeouts, and clears any recorded heartbeat details. + # + # WARNING: Standalone Activities are experimental. + # + # @param keep_paused [Boolean] If true and the activity is paused, it remains paused after reset. + # @param jitter [Float, nil] If set and the activity is in backoff, it will start at a random + # time within this duration (in seconds). + # @param restore_original_options [Boolean] If true, restore the activity options to the + # originals it was created with. + # @param reset_heartbeat [Boolean] If true, additionally discard any persisted heartbeat details. + # @param rpc_options [RPCOptions, nil] Advanced RPC options. + # @raise [Error::RPCError] RPC error from call. + def reset(keep_paused: false, jitter: nil, restore_original_options: false, + reset_heartbeat: false, rpc_options: nil) + @client._impl.reset_activity( + Interceptor::ResetActivityInput.new( + activity_id: id, + activity_run_id: run_id, + keep_paused:, + jitter:, + restore_original_options:, + reset_heartbeat:, + rpc_options: + ) + ) + nil + end + + # Update the activity's options. Only the options you actually pass are changed; anything you + # omit is left as-is. Passing an option explicitly as `nil` clears it. + # + # WARNING: Standalone Activities are experimental. + # + # @param restore_original [Boolean] If true, restore the options to the originals the activity + # was created with. Mutually exclusive with any other option. + # @param rpc_options [RPCOptions, nil] Advanced RPC options. + # @param options [Hash{Symbol => Object, nil}] The options to change. Keys must be drawn from + # {UPDATABLE_OPTION_PATHS}; anything else raises `ArgumentError`. + # @option options [String, Symbol, nil] :task_queue New task queue. + # @option options [Float, nil] :schedule_to_close_timeout New schedule-to-close timeout in seconds. + # @option options [Float, nil] :schedule_to_start_timeout New schedule-to-start timeout in seconds. + # @option options [Float, nil] :start_to_close_timeout New start-to-close timeout in seconds. + # @option options [Float, nil] :heartbeat_timeout New heartbeat timeout in seconds. + # @option options [RetryPolicy, nil] :retry_policy New retry policy. + # @option options [Priority, nil] :priority New priority. + # @option options [Float, nil] :start_delay New start delay in seconds. + # + # @return [ActivityExecutionOptions] The activity options after the update. + # + # @raise [ArgumentError] If an unknown option is given, if `restore_original` is combined with + # any other option, or if no option is provided and `restore_original` is false. + # @raise [Error::RPCError] RPC error from call. + def update_options(restore_original: false, rpc_options: nil, **options) + unknown = options.keys - UPDATABLE_OPTION_PATHS.keys + unless unknown.empty? + raise ArgumentError, + "Unknown option(s): #{unknown.join(', ')}. " \ + "Expected any of: #{UPDATABLE_OPTION_PATHS.keys.join(', ')}" + end + + if restore_original && !options.empty? + raise ArgumentError, 'restore_original cannot be combined with any other option' + elsif !restore_original && options.empty? + raise ArgumentError, 'At least one option must be set, or restore_original must be used' + end + + proto = Api::Activity::V1::ActivityOptions.new + if options.key?(:task_queue) && (task_queue = options[:task_queue]) + proto.task_queue = Api::TaskQueue::V1::TaskQueue.new(name: task_queue.to_s) + end + %i[schedule_to_close_timeout schedule_to_start_timeout start_to_close_timeout + heartbeat_timeout start_delay].each do |name| + next unless options.key?(name) + + proto[name.to_s] = Internal::ProtoUtils.seconds_to_duration(options[name]) + end + proto.retry_policy = options[:retry_policy]&._to_proto if options.key?(:retry_policy) + proto.priority = options[:priority]&._to_proto if options.key?(:priority) + + @client._impl.update_activity_options( + Interceptor::UpdateActivityOptionsInput.new( + activity_id: id, + activity_run_id: run_id, + activity_options: proto, + update_mask: Google::Protobuf::FieldMask.new( + paths: options.keys.map { |k| UPDATABLE_OPTION_PATHS.fetch(k) } + ), + restore_original:, + rpc_options: + ) + ) + end + private def _process_outcome(outcome, hint) diff --git a/temporalio/lib/temporalio/client/interceptor.rb b/temporalio/lib/temporalio/client/interceptor.rb index 33575b02..8b8e53db 100644 --- a/temporalio/lib/temporalio/client/interceptor.rb +++ b/temporalio/lib/temporalio/client/interceptor.rb @@ -291,6 +291,10 @@ def intercept_client(next_interceptor) DescribeActivityInput = Data.define( :activity_id, :activity_run_id, + :include_input, + :include_outcome, + :include_heartbeat_details, + :include_last_failure, :rpc_options ) @@ -314,6 +318,52 @@ def intercept_client(next_interceptor) :rpc_options ) + # Input for {Outbound.pause_activity}. + # + # WARNING: Standalone Activities are experimental. + PauseActivityInput = Data.define( + :activity_id, + :activity_run_id, + :reason, + :rpc_options + ) + + # Input for {Outbound.unpause_activity}. + # + # WARNING: Standalone Activities are experimental. + UnpauseActivityInput = Data.define( + :activity_id, + :activity_run_id, + :reason, + :jitter, + :rpc_options + ) + + # Input for {Outbound.reset_activity}. + # + # WARNING: Standalone Activities are experimental. + ResetActivityInput = Data.define( + :activity_id, + :activity_run_id, + :keep_paused, + :jitter, + :restore_original_options, + :reset_heartbeat, + :rpc_options + ) + + # Input for {Outbound.update_activity_options}. + # + # WARNING: Standalone Activities are experimental. + UpdateActivityOptionsInput = Data.define( + :activity_id, + :activity_run_id, + :activity_options, + :update_mask, + :restore_original, + :rpc_options + ) + # Input for {Outbound.list_activities}. # # WARNING: Standalone Activities are experimental. @@ -587,6 +637,43 @@ def terminate_activity(input) next_interceptor.terminate_activity(input) end + # Called for every {ActivityHandle.pause} call. + # + # WARNING: Standalone Activities are experimental. + # + # @param input [PauseActivityInput] Input. + def pause_activity(input) + next_interceptor.pause_activity(input) + end + + # Called for every {ActivityHandle.unpause} call. + # + # WARNING: Standalone Activities are experimental. + # + # @param input [UnpauseActivityInput] Input. + def unpause_activity(input) + next_interceptor.unpause_activity(input) + end + + # Called for every {ActivityHandle.reset} call. + # + # WARNING: Standalone Activities are experimental. + # + # @param input [ResetActivityInput] Input. + def reset_activity(input) + next_interceptor.reset_activity(input) + end + + # Called for every {ActivityHandle.update_options} call. + # + # WARNING: Standalone Activities are experimental. + # + # @param input [UpdateActivityOptionsInput] Input. + # @return [Api::Activity::V1::ActivityOptions] Activity options after the update. + def update_activity_options(input) + next_interceptor.update_activity_options(input) + end + # Called for every {Client.list_activities} call. # # WARNING: Standalone Activities are experimental. diff --git a/temporalio/lib/temporalio/internal/client/implementation.rb b/temporalio/lib/temporalio/internal/client/implementation.rb index 1cc7bb67..4ef16e14 100644 --- a/temporalio/lib/temporalio/internal/client/implementation.rb +++ b/temporalio/lib/temporalio/internal/client/implementation.rb @@ -29,7 +29,7 @@ module Temporalio module Internal module Client - class Implementation < Temporalio::Client::Interceptor::Outbound + class Implementation < Temporalio::Client::Interceptor::Outbound # rubocop:disable Metrics/ClassLength # Proto routing convention for standalone activity completion: `*_by_id` requests carry # `resource_id = "activity:"`. See the resource_id field comment on # `RecordActivityTaskHeartbeatByIdRequest` (and the analogous Completed/Failed/Canceled @@ -1025,10 +1025,20 @@ def describe_activity(input) Api::WorkflowService::V1::DescribeActivityExecutionRequest.new( namespace: @client.namespace, activity_id: input.activity_id, - run_id: input.activity_run_id || '' + run_id: input.activity_run_id || '', + include_input: input.include_input, + include_outcome: input.include_outcome, + include_heartbeat_details: input.include_heartbeat_details, + include_last_failure: input.include_last_failure ), rpc_options: Implementation.with_default_rpc_options(input.rpc_options) ) + # Clear payload-bearing fields the caller did not ask for, in case an older + # or buggy server sent them anyway. + resp.input = nil unless input.include_input + resp.outcome = nil unless input.include_outcome + resp.info.heartbeat_details = nil unless input.include_heartbeat_details + resp.info.last_failure = nil unless input.include_last_failure Temporalio::Client::ActivityExecution::Description.new(resp, @client.data_converter) end @@ -1062,6 +1072,72 @@ def terminate_activity(input) nil end + def pause_activity(input) + @client.workflow_service.pause_activity_execution( + Api::WorkflowService::V1::PauseActivityExecutionRequest.new( + namespace: @client.namespace, + activity_id: input.activity_id, + run_id: input.activity_run_id || '', + identity: @client.connection.identity, + request_id: SecureRandom.uuid, + reason: input.reason || '' + ), + rpc_options: Implementation.with_default_rpc_options(input.rpc_options) + ) + nil + end + + def unpause_activity(input) + @client.workflow_service.unpause_activity_execution( + Api::WorkflowService::V1::UnpauseActivityExecutionRequest.new( + namespace: @client.namespace, + activity_id: input.activity_id, + run_id: input.activity_run_id || '', + identity: @client.connection.identity, + request_id: SecureRandom.uuid, + reason: input.reason || '', + jitter: ProtoUtils.seconds_to_duration(input.jitter) + ), + rpc_options: Implementation.with_default_rpc_options(input.rpc_options) + ) + nil + end + + def reset_activity(input) + @client.workflow_service.reset_activity_execution( + Api::WorkflowService::V1::ResetActivityExecutionRequest.new( + namespace: @client.namespace, + activity_id: input.activity_id, + run_id: input.activity_run_id || '', + identity: @client.connection.identity, + request_id: SecureRandom.uuid, + keep_paused: input.keep_paused, + jitter: ProtoUtils.seconds_to_duration(input.jitter), + restore_original_options: input.restore_original_options, + reset_heartbeat: input.reset_heartbeat + ), + rpc_options: Implementation.with_default_rpc_options(input.rpc_options) + ) + nil + end + + def update_activity_options(input) + resp = @client.workflow_service.update_activity_execution_options( + Api::WorkflowService::V1::UpdateActivityExecutionOptionsRequest.new( + namespace: @client.namespace, + activity_id: input.activity_id, + run_id: input.activity_run_id || '', + identity: @client.connection.identity, + request_id: SecureRandom.uuid, + activity_options: input.activity_options, + update_mask: input.update_mask, + restore_original: input.restore_original + ), + rpc_options: Implementation.with_default_rpc_options(input.rpc_options) + ) + Temporalio::Client::ActivityExecutionOptions._from_proto(resp.activity_options) + end + def list_activities(input) Enumerator.new do |yielder| req = Api::WorkflowService::V1::ListActivityExecutionsRequest.new( diff --git a/temporalio/rbi/temporalio/client/activity_execution.rbi b/temporalio/rbi/temporalio/client/activity_execution.rbi index beb51f88..a53dfbca 100644 --- a/temporalio/rbi/temporalio/client/activity_execution.rbi +++ b/temporalio/rbi/temporalio/client/activity_execution.rbi @@ -33,6 +33,9 @@ class Temporalio::Client::ActivityExecution sig { returns(T.nilable(Time)) } def schedule_time; end + sig { returns(T.nilable(Time)) } + def execution_time; end + sig { returns(T.nilable(Time)) } def close_time; end @@ -76,9 +79,30 @@ class Temporalio::Client::ActivityExecution::Description < ::Temporalio::Client: sig { returns(T.nilable(Float)) } def heartbeat_timeout; end + sig { returns(T.nilable(Float)) } + def start_delay; end + sig { returns(T::Boolean) } def has_heartbeat_details?; end + sig { returns(T::Boolean) } + def has_last_failure?; end + + sig { returns(T::Boolean) } + def has_input?; end + + sig { params(hints: T.nilable(T::Array[Object])).returns(T::Array[T.nilable(Object)]) } + def input(hints: T.unsafe(nil)); end + + sig { returns(T::Boolean) } + def has_result?; end + + sig { params(result_hint: T.nilable(Object)).returns(T.nilable(Object)) } + def result(result_hint: T.unsafe(nil)); end + + sig { returns(T.nilable(Temporalio::Error::Failure)) } + def failure; end + sig { params(hints: T.nilable(T::Array[Object])).returns(T::Array[T.nilable(Object)]) } def heartbeat_details(hints: T.unsafe(nil)); end diff --git a/temporalio/rbi/temporalio/client/activity_execution_options.rbi b/temporalio/rbi/temporalio/client/activity_execution_options.rbi new file mode 100644 index 00000000..ca1abbaa --- /dev/null +++ b/temporalio/rbi/temporalio/client/activity_execution_options.rbi @@ -0,0 +1,58 @@ +# typed: true + +class Temporalio::Client::ActivityExecutionOptions + class << self + sig do + params(options: Temporalio::Api::Activity::V1::ActivityOptions) + .returns(Temporalio::Client::ActivityExecutionOptions) + end + def _from_proto(options); end + end + + sig do + params( + task_queue: T.nilable(String), + schedule_to_close_timeout: T.nilable(Float), + schedule_to_start_timeout: T.nilable(Float), + start_to_close_timeout: T.nilable(Float), + heartbeat_timeout: T.nilable(Float), + retry_policy: T.nilable(Temporalio::RetryPolicy), + priority: Temporalio::Priority, + start_delay: T.nilable(Float) + ).void + end + def initialize( + task_queue:, + schedule_to_close_timeout:, + schedule_to_start_timeout:, + start_to_close_timeout:, + heartbeat_timeout:, + retry_policy:, + priority:, + start_delay: + ); end + + sig { returns(T.nilable(String)) } + attr_reader :task_queue + + sig { returns(T.nilable(Float)) } + attr_reader :schedule_to_close_timeout + + sig { returns(T.nilable(Float)) } + attr_reader :schedule_to_start_timeout + + sig { returns(T.nilable(Float)) } + attr_reader :start_to_close_timeout + + sig { returns(T.nilable(Float)) } + attr_reader :heartbeat_timeout + + sig { returns(T.nilable(Temporalio::RetryPolicy)) } + attr_reader :retry_policy + + sig { returns(Temporalio::Priority) } + attr_reader :priority + + sig { returns(T.nilable(Float)) } + attr_reader :start_delay +end diff --git a/temporalio/rbi/temporalio/client/activity_execution_status.rbi b/temporalio/rbi/temporalio/client/activity_execution_status.rbi index ab4828e6..3694f6a9 100644 --- a/temporalio/rbi/temporalio/client/activity_execution_status.rbi +++ b/temporalio/rbi/temporalio/client/activity_execution_status.rbi @@ -8,4 +8,5 @@ module Temporalio::Client::ActivityExecutionStatus CANCELED = T.let(T.unsafe(nil), Integer) TERMINATED = T.let(T.unsafe(nil), Integer) TIMED_OUT = T.let(T.unsafe(nil), Integer) + PAUSED = T.let(T.unsafe(nil), Integer) end diff --git a/temporalio/rbi/temporalio/client/activity_handle.rbi b/temporalio/rbi/temporalio/client/activity_handle.rbi index 79ae52f1..f738d0d6 100644 --- a/temporalio/rbi/temporalio/client/activity_handle.rbi +++ b/temporalio/rbi/temporalio/client/activity_handle.rbi @@ -1,6 +1,8 @@ # typed: true class Temporalio::Client::ActivityHandle + UPDATABLE_OPTION_PATHS = T.let(T.unsafe(nil), T::Hash[Symbol, String]) + sig do params( client: Temporalio::Client, @@ -28,8 +30,22 @@ class Temporalio::Client::ActivityHandle end def result(result_hint: T.unsafe(nil), rpc_options: T.unsafe(nil)); end - sig { params(rpc_options: T.nilable(Temporalio::Client::RPCOptions)).returns(Temporalio::Client::ActivityExecution::Description) } - def describe(rpc_options: T.unsafe(nil)); end + sig do + params( + include_input: T::Boolean, + include_outcome: T::Boolean, + include_heartbeat_details: T::Boolean, + include_last_failure: T::Boolean, + rpc_options: T.nilable(Temporalio::Client::RPCOptions) + ).returns(Temporalio::Client::ActivityExecution::Description) + end + def describe( + include_input: T.unsafe(nil), + include_outcome: T.unsafe(nil), + include_heartbeat_details: T.unsafe(nil), + include_last_failure: T.unsafe(nil), + rpc_options: T.unsafe(nil) + ); end sig { params(reason: T.nilable(String), rpc_options: T.nilable(Temporalio::Client::RPCOptions)).void } def cancel(reason = T.unsafe(nil), rpc_options: T.unsafe(nil)); end @@ -37,6 +53,50 @@ class Temporalio::Client::ActivityHandle sig { params(reason: T.nilable(String), rpc_options: T.nilable(Temporalio::Client::RPCOptions)).void } def terminate(reason = T.unsafe(nil), rpc_options: T.unsafe(nil)); end + sig { params(reason: T.nilable(String), rpc_options: T.nilable(Temporalio::Client::RPCOptions)).void } + def pause(reason = T.unsafe(nil), rpc_options: T.unsafe(nil)); end + + sig do + params( + reason: T.nilable(String), + jitter: T.nilable(Float), + rpc_options: T.nilable(Temporalio::Client::RPCOptions) + ).void + end + def unpause(reason: T.unsafe(nil), jitter: T.unsafe(nil), rpc_options: T.unsafe(nil)); end + + sig do + params( + keep_paused: T::Boolean, + jitter: T.nilable(Float), + restore_original_options: T::Boolean, + reset_heartbeat: T::Boolean, + rpc_options: T.nilable(Temporalio::Client::RPCOptions) + ).void + end + def reset( + keep_paused: T.unsafe(nil), + jitter: T.unsafe(nil), + restore_original_options: T.unsafe(nil), + reset_heartbeat: T.unsafe(nil), + rpc_options: T.unsafe(nil) + ); end + + sig do + params( + restore_original: T::Boolean, + rpc_options: T.nilable(Temporalio::Client::RPCOptions), + options: T.nilable( + T.any(String, Symbol, Integer, Float, Temporalio::RetryPolicy, Temporalio::Priority) + ) + ).returns(Temporalio::Client::ActivityExecutionOptions) + end + def update_options( + restore_original: T.unsafe(nil), + rpc_options: T.unsafe(nil), + **options + ); end + private sig do diff --git a/temporalio/rbi/temporalio/client/interceptor.rbi b/temporalio/rbi/temporalio/client/interceptor.rbi index dcea1d74..593ecb16 100644 --- a/temporalio/rbi/temporalio/client/interceptor.rbi +++ b/temporalio/rbi/temporalio/client/interceptor.rbi @@ -834,6 +834,18 @@ class Temporalio::Client::Interceptor::DescribeActivityInput < ::Data sig { returns(T.nilable(String)) } def activity_run_id; end + sig { returns(T::Boolean) } + def include_input; end + + sig { returns(T::Boolean) } + def include_outcome; end + + sig { returns(T::Boolean) } + def include_heartbeat_details; end + + sig { returns(T::Boolean) } + def include_last_failure; end + sig { returns(T.nilable(Temporalio::Client::RPCOptions)) } def rpc_options; end @@ -849,6 +861,121 @@ class Temporalio::Client::Interceptor::DescribeActivityInput < ::Data end end +class Temporalio::Client::Interceptor::PauseActivityInput < ::Data + sig { returns(String) } + def activity_id; end + + sig { returns(T.nilable(String)) } + def activity_run_id; end + + sig { returns(T.nilable(String)) } + def reason; end + + sig { returns(T.nilable(Temporalio::Client::RPCOptions)) } + def rpc_options; end + + class << self + sig { params(args: T.untyped).returns(Temporalio::Client::Interceptor::PauseActivityInput) } + def new(*args); end + + sig { params(args: T.untyped).returns(Temporalio::Client::Interceptor::PauseActivityInput) } + def [](*args); end + + sig { returns(T::Array[Symbol]) } + def members; end + end +end +class Temporalio::Client::Interceptor::UnpauseActivityInput < ::Data + sig { returns(String) } + def activity_id; end + + sig { returns(T.nilable(String)) } + def activity_run_id; end + + sig { returns(T.nilable(String)) } + def reason; end + + sig { returns(T.nilable(Float)) } + def jitter; end + + sig { returns(T.nilable(Temporalio::Client::RPCOptions)) } + def rpc_options; end + + class << self + sig { params(args: T.untyped).returns(Temporalio::Client::Interceptor::UnpauseActivityInput) } + def new(*args); end + + sig { params(args: T.untyped).returns(Temporalio::Client::Interceptor::UnpauseActivityInput) } + def [](*args); end + + sig { returns(T::Array[Symbol]) } + def members; end + end +end +class Temporalio::Client::Interceptor::ResetActivityInput < ::Data + sig { returns(String) } + def activity_id; end + + sig { returns(T.nilable(String)) } + def activity_run_id; end + + sig { returns(T::Boolean) } + def keep_paused; end + + sig { returns(T.nilable(Float)) } + def jitter; end + + sig { returns(T::Boolean) } + def restore_original_options; end + + sig { returns(T::Boolean) } + def reset_heartbeat; end + + sig { returns(T.nilable(Temporalio::Client::RPCOptions)) } + def rpc_options; end + + class << self + sig { params(args: T.untyped).returns(Temporalio::Client::Interceptor::ResetActivityInput) } + def new(*args); end + + sig { params(args: T.untyped).returns(Temporalio::Client::Interceptor::ResetActivityInput) } + def [](*args); end + + sig { returns(T::Array[Symbol]) } + def members; end + end +end +class Temporalio::Client::Interceptor::UpdateActivityOptionsInput < ::Data + sig { returns(String) } + def activity_id; end + + sig { returns(T.nilable(String)) } + def activity_run_id; end + + sig { returns(Temporalio::Api::Activity::V1::ActivityOptions) } + def activity_options; end + + sig { returns(Google::Protobuf::FieldMask) } + def update_mask; end + + sig { returns(T::Boolean) } + def restore_original; end + + sig { returns(T.nilable(Temporalio::Client::RPCOptions)) } + def rpc_options; end + + class << self + sig { params(args: T.untyped).returns(Temporalio::Client::Interceptor::UpdateActivityOptionsInput) } + def new(*args); end + + sig { params(args: T.untyped).returns(Temporalio::Client::Interceptor::UpdateActivityOptionsInput) } + def [](*args); end + + sig { returns(T::Array[Symbol]) } + def members; end + end +end + class Temporalio::Client::Interceptor::CancelActivityInput < ::Data sig { returns(String) } def activity_id; end @@ -1056,6 +1183,21 @@ class Temporalio::Client::Interceptor::Outbound sig { params(input: Temporalio::Client::Interceptor::TerminateActivityInput).void } def terminate_activity(input); end + sig { params(input: Temporalio::Client::Interceptor::PauseActivityInput).void } + def pause_activity(input); end + + sig { params(input: Temporalio::Client::Interceptor::UnpauseActivityInput).void } + def unpause_activity(input); end + + sig { params(input: Temporalio::Client::Interceptor::ResetActivityInput).void } + def reset_activity(input); end + + sig do + params(input: Temporalio::Client::Interceptor::UpdateActivityOptionsInput) + .returns(Temporalio::Client::ActivityExecutionOptions) + end + def update_activity_options(input); end + sig { params(input: Temporalio::Client::Interceptor::ListActivitiesInput).returns(T::Enumerator[Temporalio::Client::ActivityExecution]) } def list_activities(input); end diff --git a/temporalio/sig/temporalio/client/activity_execution.rbs b/temporalio/sig/temporalio/client/activity_execution.rbs index dfa9a0dd..2b657e0e 100644 --- a/temporalio/sig/temporalio/client/activity_execution.rbs +++ b/temporalio/sig/temporalio/client/activity_execution.rbs @@ -9,6 +9,7 @@ module Temporalio def activity_run_id: -> String? def activity_type: -> String def schedule_time: -> Time? + def execution_time: -> Time? def close_time: -> Time? def status: -> ActivityExecutionStatus::enum def search_attributes: -> SearchAttributes? @@ -25,7 +26,14 @@ module Temporalio def schedule_to_start_timeout: -> Float? def start_to_close_timeout: -> Float? def heartbeat_timeout: -> Float? + def start_delay: -> Float? def has_heartbeat_details?: -> bool + def has_last_failure?: -> bool + def has_input?: -> bool + def input: (?hints: Array[Object]?) -> Array[Object?] + def has_result?: -> bool + def result: (?result_hint: Object?) -> Object? + def failure: -> Error::Failure? def heartbeat_details: (?hints: Array[Object]?) -> Array[Object?] def retry_policy: -> RetryPolicy def last_heartbeat_time: -> Time? diff --git a/temporalio/sig/temporalio/client/activity_execution_options.rbs b/temporalio/sig/temporalio/client/activity_execution_options.rbs new file mode 100644 index 00000000..61ce9c46 --- /dev/null +++ b/temporalio/sig/temporalio/client/activity_execution_options.rbs @@ -0,0 +1,27 @@ +module Temporalio + class Client + class ActivityExecutionOptions + attr_reader task_queue: String? + attr_reader schedule_to_close_timeout: Float? + attr_reader schedule_to_start_timeout: Float? + attr_reader start_to_close_timeout: Float? + attr_reader heartbeat_timeout: Float? + attr_reader retry_policy: RetryPolicy? + attr_reader priority: Priority + attr_reader start_delay: Float? + + def self._from_proto: (untyped options) -> ActivityExecutionOptions + + def initialize: ( + task_queue: String?, + schedule_to_close_timeout: Float?, + schedule_to_start_timeout: Float?, + start_to_close_timeout: Float?, + heartbeat_timeout: Float?, + retry_policy: RetryPolicy?, + priority: Priority, + start_delay: Float? + ) -> void + end + end +end diff --git a/temporalio/sig/temporalio/client/activity_execution_status.rbs b/temporalio/sig/temporalio/client/activity_execution_status.rbs index 39618ec0..37450146 100644 --- a/temporalio/sig/temporalio/client/activity_execution_status.rbs +++ b/temporalio/sig/temporalio/client/activity_execution_status.rbs @@ -10,6 +10,7 @@ module Temporalio CANCELED: enum TERMINATED: enum TIMED_OUT: enum + PAUSED: enum end end end diff --git a/temporalio/sig/temporalio/client/activity_handle.rbs b/temporalio/sig/temporalio/client/activity_handle.rbs index e83a3514..5e91c9cd 100644 --- a/temporalio/sig/temporalio/client/activity_handle.rbs +++ b/temporalio/sig/temporalio/client/activity_handle.rbs @@ -1,6 +1,8 @@ module Temporalio class Client class ActivityHandle + UPDATABLE_OPTION_PATHS: Hash[Symbol, String] + attr_reader id: String attr_reader run_id: String? attr_reader result_hint: Object? @@ -14,12 +16,40 @@ module Temporalio def result: (?result_hint: Object?, ?rpc_options: RPCOptions?) -> Object? - def describe: (?rpc_options: RPCOptions?) -> ActivityExecution::Description + def describe: ( + ?include_input: bool, + ?include_outcome: bool, + ?include_heartbeat_details: bool, + ?include_last_failure: bool, + ?rpc_options: RPCOptions? + ) -> ActivityExecution::Description def cancel: (?String? reason, ?rpc_options: RPCOptions?) -> void def terminate: (?String? reason, ?rpc_options: RPCOptions?) -> void + def pause: (?String? reason, ?rpc_options: RPCOptions?) -> void + + def unpause: ( + ?reason: String?, + ?jitter: Float?, + ?rpc_options: RPCOptions? + ) -> void + + def reset: ( + ?keep_paused: bool, + ?jitter: Float?, + ?restore_original_options: bool, + ?reset_heartbeat: bool, + ?rpc_options: RPCOptions? + ) -> void + + def update_options: ( + ?restore_original: bool, + ?rpc_options: RPCOptions?, + **untyped options + ) -> ActivityExecutionOptions + private def _process_outcome: (Api::Activity::V1::ActivityExecutionOutcome? outcome, Object? hint) -> Object? end end diff --git a/temporalio/sig/temporalio/client/interceptor.rbs b/temporalio/sig/temporalio/client/interceptor.rbs index a411f81f..928e2a1e 100644 --- a/temporalio/sig/temporalio/client/interceptor.rbs +++ b/temporalio/sig/temporalio/client/interceptor.rbs @@ -481,11 +481,19 @@ module Temporalio class DescribeActivityInput attr_reader activity_id: String attr_reader activity_run_id: String? + attr_reader include_input: bool + attr_reader include_outcome: bool + attr_reader include_heartbeat_details: bool + attr_reader include_last_failure: bool attr_reader rpc_options: RPCOptions? def initialize: ( activity_id: String, activity_run_id: String?, + include_input: bool, + include_outcome: bool, + include_heartbeat_details: bool, + include_last_failure: bool, rpc_options: RPCOptions? ) -> void end @@ -518,6 +526,74 @@ module Temporalio ) -> void end + class PauseActivityInput + attr_reader activity_id: String + attr_reader activity_run_id: String? + attr_reader reason: String? + attr_reader rpc_options: RPCOptions? + + def initialize: ( + activity_id: String, + activity_run_id: String?, + reason: String?, + rpc_options: RPCOptions? + ) -> void + end + + class UnpauseActivityInput + attr_reader activity_id: String + attr_reader activity_run_id: String? + attr_reader reason: String? + attr_reader jitter: Float? + attr_reader rpc_options: RPCOptions? + + def initialize: ( + activity_id: String, + activity_run_id: String?, + reason: String?, + jitter: Float?, + rpc_options: RPCOptions? + ) -> void + end + + class ResetActivityInput + attr_reader activity_id: String + attr_reader activity_run_id: String? + attr_reader keep_paused: bool + attr_reader jitter: Float? + attr_reader restore_original_options: bool + attr_reader reset_heartbeat: bool + attr_reader rpc_options: RPCOptions? + + def initialize: ( + activity_id: String, + activity_run_id: String?, + keep_paused: bool, + jitter: Float?, + restore_original_options: bool, + reset_heartbeat: bool, + rpc_options: RPCOptions? + ) -> void + end + + class UpdateActivityOptionsInput + attr_reader activity_id: String + attr_reader activity_run_id: String? + attr_reader activity_options: untyped + attr_reader update_mask: untyped + attr_reader restore_original: bool + attr_reader rpc_options: RPCOptions? + + def initialize: ( + activity_id: String, + activity_run_id: String?, + activity_options: untyped, + update_mask: untyped, + restore_original: bool, + rpc_options: RPCOptions? + ) -> void + end + class ListActivitiesInput attr_reader query: String attr_reader rpc_options: RPCOptions? @@ -611,6 +687,14 @@ module Temporalio def terminate_activity: (TerminateActivityInput input) -> void + def pause_activity: (PauseActivityInput input) -> void + + def unpause_activity: (UnpauseActivityInput input) -> void + + def reset_activity: (ResetActivityInput input) -> void + + def update_activity_options: (UpdateActivityOptionsInput input) -> untyped + def list_activities: (ListActivitiesInput input) -> Enumerator[ActivityExecution, ActivityExecution] def count_activities: (CountActivitiesInput input) -> ActivityExecutionCount diff --git a/temporalio/sig/temporalio/internal/client/implementation.rbs b/temporalio/sig/temporalio/internal/client/implementation.rbs index e6cc9993..9d5a394e 100644 --- a/temporalio/sig/temporalio/internal/client/implementation.rbs +++ b/temporalio/sig/temporalio/internal/client/implementation.rbs @@ -22,6 +22,14 @@ module Temporalio def terminate_activity: (Temporalio::Client::Interceptor::TerminateActivityInput input) -> void + def pause_activity: (Temporalio::Client::Interceptor::PauseActivityInput input) -> void + + def unpause_activity: (Temporalio::Client::Interceptor::UnpauseActivityInput input) -> void + + def reset_activity: (Temporalio::Client::Interceptor::ResetActivityInput input) -> void + + def update_activity_options: (Temporalio::Client::Interceptor::UpdateActivityOptionsInput input) -> untyped + def list_activities: (Temporalio::Client::Interceptor::ListActivitiesInput input) -> Enumerator[Temporalio::Client::ActivityExecution, Temporalio::Client::ActivityExecution] def count_activities: (Temporalio::Client::Interceptor::CountActivitiesInput input) -> Temporalio::Client::ActivityExecutionCount diff --git a/temporalio/test/client_activity_async_completion_test.rb b/temporalio/test/client_activity_async_completion_test.rb index c6c877c5..799355f3 100644 --- a/temporalio/test/client_activity_async_completion_test.rb +++ b/temporalio/test/client_activity_async_completion_test.rb @@ -94,8 +94,8 @@ def test_async_completion_heartbeat_standalone activity_run_id: handle.run_id ) env.client.async_activity_handle(ref).heartbeat('hb-1', 'hb-2') - assert_equal %w[hb-1 hb-2], - env.client.data_converter.from_payloads(handle.describe.raw_info.heartbeat_details) + desc = handle.describe(include_heartbeat_details: true) + assert_equal %w[hb-1 hb-2], env.client.data_converter.from_payloads(desc.raw_info.heartbeat_details) env.client.async_activity_handle(ref).complete('done-after-heartbeat') assert_equal 'done-after-heartbeat', handle.result end @@ -144,8 +144,8 @@ def test_async_completion_heartbeat_and_fail_standalone activity_run_id: handle.run_id ) env.client.async_activity_handle(ref).heartbeat('hb-1', 'hb-2') - assert_equal %w[hb-1 hb-2], - env.client.data_converter.from_payloads(handle.describe.raw_info.heartbeat_details) + desc = handle.describe(include_heartbeat_details: true) + assert_equal %w[hb-1 hb-2], env.client.data_converter.from_payloads(desc.raw_info.heartbeat_details) env.client.async_activity_handle(ref).fail( Temporalio::Error::ApplicationError.new('hb-then-fail', non_retryable: true) ) diff --git a/temporalio/test/client_activity_operator_commands_build_test.rb b/temporalio/test/client_activity_operator_commands_build_test.rb new file mode 100644 index 00000000..ca763603 --- /dev/null +++ b/temporalio/test/client_activity_operator_commands_build_test.rb @@ -0,0 +1,66 @@ +# frozen_string_literal: true + +require 'temporalio/api' +require 'temporalio/client' +require 'test' + +# Unit test for the operator-command request fields that the server does not surface back. +class ClientActivityOperatorCommandsBuildTest < Test + def test_unobservable_request_fields + # Lazy connect so no real connection is opened; the RPCs below are stubbed. + client = Temporalio::Client.connect('localhost:7233', 'test-namespace', lazy_connect: true) + handle = client.activity_handle('act-1', activity_run_id: 'run-1') + + ws = client.workflow_service + captured = {} + + ws.define_singleton_method(:pause_activity_execution) do |req, **_kwargs| + captured[:pause] = req + Temporalio::Api::WorkflowService::V1::PauseActivityExecutionResponse.new + end + ws.define_singleton_method(:unpause_activity_execution) do |req, **_kwargs| + captured[:unpause] = req + Temporalio::Api::WorkflowService::V1::UnpauseActivityExecutionResponse.new + end + ws.define_singleton_method(:reset_activity_execution) do |req, **_kwargs| + captured[:reset] = req + Temporalio::Api::WorkflowService::V1::ResetActivityExecutionResponse.new + end + ws.define_singleton_method(:update_activity_execution_options) do |req, **_kwargs| + captured[:update] = req + Temporalio::Api::WorkflowService::V1::UpdateActivityExecutionOptionsResponse.new( + activity_options: Temporalio::Api::Activity::V1::ActivityOptions.new + ) + end + + begin + handle.pause('because') + handle.unpause(reason: 'go', jitter: 5.0) + handle.reset(jitter: 2.0) + handle.update_options(restore_original: true) + ensure + ws.singleton_class.send(:remove_method, :pause_activity_execution) + ws.singleton_class.send(:remove_method, :unpause_activity_execution) + ws.singleton_class.send(:remove_method, :reset_activity_execution) + ws.singleton_class.send(:remove_method, :update_activity_execution_options) + end + + pause_req = captured.fetch(:pause) + assert_equal 'because', pause_req.reason + refute_empty pause_req.request_id + + unpause_req = captured.fetch(:unpause) + assert_equal 'go', unpause_req.reason + assert_equal 5, unpause_req.jitter.seconds + assert_equal 0, unpause_req.jitter.nanos + refute_empty unpause_req.request_id + + reset_req = captured.fetch(:reset) + assert_equal 2, reset_req.jitter.seconds + assert_equal 0, reset_req.jitter.nanos + refute_empty reset_req.request_id + + update_req = captured.fetch(:update) + refute_empty update_req.request_id + end +end diff --git a/temporalio/test/client_activity_operator_commands_interceptor_test.rb b/temporalio/test/client_activity_operator_commands_interceptor_test.rb new file mode 100644 index 00000000..82b5bbdb --- /dev/null +++ b/temporalio/test/client_activity_operator_commands_interceptor_test.rb @@ -0,0 +1,99 @@ +# frozen_string_literal: true + +require 'securerandom' +require 'temporalio/client' +require 'temporalio/testing' +require 'temporalio/worker' +require 'test' + +# Verifies each operator command (pause/unpause/reset/update_options) flows through the outbound +# client interceptor chain. +class ClientActivityOperatorCommandsInterceptorTest < Test + class SlowActivity < Temporalio::Activity::Definition + def execute + Temporalio::Activity::Context.current.heartbeat + sleep 0.1 until Temporalio::Activity::Context.current.cancellation.canceled? + raise Temporalio::Error::CanceledError, 'canceled' + end + end + + class RecordingInterceptor + include Temporalio::Client::Interceptor + + def initialize(events_array) + @events = events_array + end + + def intercept_client(next_interceptor) + Outbound.new(next_interceptor, @events) + end + + class Outbound < Temporalio::Client::Interceptor::Outbound + def initialize(next_interceptor, events) + super(next_interceptor) + @events = events + end + + def pause_activity(input) + @events << 'pause_activity' + super + end + + def unpause_activity(input) + @events << 'unpause_activity' + super + end + + def reset_activity(input) + @events << 'reset_activity' + super + end + + def update_activity_options(input) + @events << 'update_activity_options' + super + end + end + end + + def client_with_interceptor(events) + interceptor = RecordingInterceptor.new(events) + Temporalio::Client.new(**env.client.options.with(interceptors: [interceptor]).to_h) + end + + def test_interceptor_invokes_each_operator_command + events = [] + client = client_with_interceptor(events) + task_queue = "saa-tq-#{SecureRandom.uuid}" + worker = Temporalio::Worker.new(client: client, task_queue: task_queue, activities: [SlowActivity]) + worker.run do + activity_id = "act-#{SecureRandom.uuid}" + handle = client.start_activity( + SlowActivity, + id: activity_id, task_queue: task_queue, start_to_close_timeout: 60, heartbeat_timeout: 30 + ) + assert_eventually do + assert_equal Temporalio::Client::PendingActivityState::STARTED, handle.describe.run_state + end + + handle.pause('reason') + paused_states = [ + Temporalio::Client::PendingActivityState::PAUSED, + Temporalio::Client::PendingActivityState::PAUSE_REQUESTED + ] + assert_eventually do + assert_includes paused_states, handle.describe.run_state + end + handle.unpause + handle.update_options(start_to_close_timeout: 90.0) + handle.reset + + handle.terminate('cleanup') + end + + assert_includes events, 'pause_activity' + assert_includes events, 'unpause_activity' + assert_includes events, 'reset_activity' + assert_includes events, 'update_activity_options' + end +end diff --git a/temporalio/test/client_activity_operator_commands_test.rb b/temporalio/test/client_activity_operator_commands_test.rb new file mode 100644 index 00000000..a12461c4 --- /dev/null +++ b/temporalio/test/client_activity_operator_commands_test.rb @@ -0,0 +1,528 @@ +# frozen_string_literal: true + +require 'securerandom' +require 'temporalio/client' +require 'temporalio/testing' +require 'temporalio/worker' +require 'test' + +# Tests for the standalone-activity operator commands on ActivityHandle: +# pause / unpause / reset / update_options. Each asserts an observable server state change. +class ClientActivityOperatorCommandsTest < Test + # Long-running activity that heartbeats and runs until cancellation. + class SlowActivity < Temporalio::Activity::Definition + def execute + Temporalio::Activity::Context.current.heartbeat + sleep 0.1 until Temporalio::Activity::Context.current.cancellation.canceled? + raise Temporalio::Error::CanceledError, 'canceled' + end + end + + # Returns immediately. Used together with a start delay so it can be paused while scheduled + # (before it ever runs) and then resumed to a successful completion. + class QuickActivity < Temporalio::Activity::Definition + def execute + 'resumed' + end + end + + # Fails the first two attempts so retries are forced, then succeeds on the third. Used to exercise + # reset against an activity that has recorded more than one attempt. + class FailThenSucceedActivity < Temporalio::Activity::Definition + def execute + if Temporalio::Activity::Context.current.info.attempt < 3 + raise Temporalio::Error::ApplicationError, 'retryable failure' + end + + 'done' + end + end + + # Takes an argument and returns a value derived from it, so a completed execution has both an + # input and a successful outcome to read back off describe. + class EchoActivity < Temporalio::Activity::Definition + def execute(word) + "#{word}-echoed" + end + end + + # Always fails. Paired with a single-attempt retry policy so the activity reaches a terminal + # failure outcome rather than retrying. + class AlwaysFailActivity < Temporalio::Activity::Definition + def execute + raise Temporalio::Error::ApplicationError, 'deliberate failure' + end + end + + # Records heartbeat details on attempt 1, then blocks waiting for cancellation. The heartbeat + # runs on its own — not adjacent to any completion RPC — so the details reliably persist and are + # observable via describe. Later attempts (after a reset or an unpause that spawns a new attempt) + # do not heartbeat, so any operator-driven clearing of the details stays observable. + class HeartbeatOnceActivity < Temporalio::Activity::Definition + def execute + ctx = Temporalio::Activity::Context.current + ctx.heartbeat('hb-details') if ctx.info.attempt == 1 + sleep 0.1 until ctx.cancellation.canceled? + raise Temporalio::Error::CanceledError, 'canceled' + end + end + + # A running activity does not transition straight to PAUSED on pause: the server records + # PAUSE_REQUESTED and only moves to PAUSED once the worker acknowledges (drops the attempt). A + # long-running heartbeating activity that has not yet noticed the pause stays in PAUSE_REQUESTED, + # so both states count as "paused" for an observability assertion. + PAUSED_STATES = [ + Temporalio::Client::PendingActivityState::PAUSED, + Temporalio::Client::PendingActivityState::PAUSE_REQUESTED + ].freeze + + def assert_eventually_paused(handle) + assert_eventually do + assert_includes PAUSED_STATES, handle.describe.run_state + end + end + + def with_activity_worker(activities, &) + task_queue = "saa-tq-#{SecureRandom.uuid}" + worker = Temporalio::Worker.new( + client: env.client, + task_queue: task_queue, + activities: activities + ) + worker.run { yield task_queue } + end + + # Start a SlowActivity and wait until it has actually started running on the worker. + def start_running_slow_activity(task_queue, **kwargs) + activity_id = "act-#{SecureRandom.uuid}" + handle = env.client.start_activity( + SlowActivity, + id: activity_id, task_queue: task_queue, start_to_close_timeout: 60, + heartbeat_timeout: 30, **kwargs + ) + assert_eventually do + desc = handle.describe + assert_equal Temporalio::Client::PendingActivityState::STARTED, desc.run_state + end + handle + end + + def test_unpause_resumes + with_activity_worker([QuickActivity]) do |task_queue| + activity_id = "act-#{SecureRandom.uuid}" + # Start with a long delay so the activity sits in SCHEDULED and can be paused before it runs. + handle = env.client.start_activity( + QuickActivity, + id: activity_id, task_queue: task_queue, start_to_close_timeout: 60, + start_delay: 30.0 + ) + handle.pause('pause-before-unpause') + # A not-yet-started (scheduled) activity transitions fully to PAUSED. + assert_eventually do + assert_equal Temporalio::Client::PendingActivityState::PAUSED, handle.describe.run_state + end + + handle.unpause + # After unpause the activity proceeds and completes successfully (proving it resumed). + assert_equal 'resumed', handle.result + end + end + + def test_reset + with_activity_worker([FailThenSucceedActivity]) do |task_queue| + activity_id = "act-#{SecureRandom.uuid}" + handle = env.client.start_activity( + FailThenSucceedActivity, + id: activity_id, task_queue: task_queue, start_to_close_timeout: 60, + retry_policy: Temporalio::RetryPolicy.new( + initial_interval: 0.2, backoff_coefficient: 1.0, max_interval: 0.2, max_attempts: 50 + ) + ) + # Wait until the activity has recorded more than one attempt (i.e. it has retried). + assert_eventually do + assert_operator handle.describe.attempt, :>, 1 + end + + handle.reset + # After reset the attempt counter goes back to the start. + assert_eventually do + assert_equal 1, handle.describe.attempt + end + handle.terminate('cleanup') + end + end + + def test_update_options_respects_mask + with_activity_worker([SlowActivity]) do |task_queue| + handle = start_running_slow_activity( + task_queue, + start_to_close_timeout: 45, + schedule_to_close_timeout: 120 + ) + + updated = handle.update_options(start_to_close_timeout: 90.0) + + # Returned options: only start_to_close changed; schedule_to_close kept its original value. + assert_equal 90.0, updated.start_to_close_timeout + assert_equal 120.0, updated.schedule_to_close_timeout + + # Confirm via describe that the partial update was applied server-side. + assert_eventually do + desc = handle.describe + assert_equal 90.0, desc.start_to_close_timeout + assert_equal 120.0, desc.schedule_to_close_timeout + end + handle.terminate('cleanup') + end + end + + def test_update_options_all_fields + with_activity_worker([QuickActivity]) do |task_queue| + activity_id = "act-#{SecureRandom.uuid}" + # Start delayed so the activity stays SCHEDULED (never runs) while we update every option and + # observe each one applied. + handle = env.client.start_activity( + QuickActivity, + id: activity_id, task_queue: task_queue, + schedule_to_close_timeout: 100, start_to_close_timeout: 30, start_delay: 300.0 + ) + + updated = handle.update_options( + task_queue: 'updated-tq', + schedule_to_close_timeout: 200.0, + schedule_to_start_timeout: 15.0, + start_to_close_timeout: 90.0, + heartbeat_timeout: 25.0, + retry_policy: Temporalio::RetryPolicy.new(initial_interval: 1.0, backoff_coefficient: 2.0, max_attempts: 7), + priority: Temporalio::Priority.new(priority_key: 3), + start_delay: 500.0 + ) + + # Every field is settable and lands: the returned options reflect each new value. + assert_equal 'updated-tq', updated.task_queue + assert_equal 200.0, updated.schedule_to_close_timeout + assert_equal 15.0, updated.schedule_to_start_timeout + assert_equal 90.0, updated.start_to_close_timeout + assert_equal 25.0, updated.heartbeat_timeout + assert_equal 7, updated.retry_policy&.max_attempts + assert_equal 3, updated.priority.priority_key + assert_equal 500.0, updated.start_delay + + # And describe reflects them server-side. + desc = handle.describe + assert_equal 'updated-tq', desc.task_queue + assert_equal 200.0, desc.schedule_to_close_timeout + assert_equal 15.0, desc.schedule_to_start_timeout + assert_equal 90.0, desc.start_to_close_timeout + assert_equal 25.0, desc.heartbeat_timeout + assert_equal 7, desc.retry_policy&.max_attempts + assert_equal 3, desc.priority.priority_key + assert_equal 500.0, desc.start_delay + + handle.terminate('cleanup') + end + end + + def test_update_options_restore_original_exclusive + with_activity_worker([SlowActivity]) do |task_queue| + handle = start_running_slow_activity(task_queue) + # Wrap the RPC so we can prove it is never reached when the validation fails. + ws = env.client.workflow_service + reached = false + original = ws.method(:update_activity_execution_options) + ws.define_singleton_method(:update_activity_execution_options) do |req, **kwargs| + reached = true + original.call(req, **kwargs) + end + begin + err = assert_raises(ArgumentError) do + handle.update_options(restore_original: true, start_to_close_timeout: 5.0) + end + assert_match(/restore_original cannot be combined/i, err.message) + refute reached, 'update_activity_execution_options RPC should not be reached when validation fails' + ensure + ws.singleton_class.send(:remove_method, :update_activity_execution_options) + end + handle.terminate('cleanup') + end + end + + def test_update_options_requires_at_least_one_option + with_activity_worker([SlowActivity]) do |task_queue| + handle = start_running_slow_activity(task_queue) + # Wrap the RPC so we can prove it is never reached when the validation fails. + ws = env.client.workflow_service + reached = false + original = ws.method(:update_activity_execution_options) + ws.define_singleton_method(:update_activity_execution_options) do |req, **kwargs| + reached = true + original.call(req, **kwargs) + end + begin + err = assert_raises(ArgumentError) { handle.update_options } + assert_match(/at least one option/i, err.message) + refute reached, 'update_activity_execution_options RPC should not be reached when validation fails' + ensure + ws.singleton_class.send(:remove_method, :update_activity_execution_options) + end + handle.terminate('cleanup') + end + end + + def test_update_options_restore_original + with_activity_worker([SlowActivity]) do |task_queue| + handle = start_running_slow_activity(task_queue, start_to_close_timeout: 45) + + # Change an option away from the original. + changed = handle.update_options(start_to_close_timeout: 90.0) + assert_equal 90.0, changed.start_to_close_timeout + + # restore_original alone reverts to the value the activity was created with. + restored = handle.update_options(restore_original: true) + assert_equal 45.0, restored.start_to_close_timeout + handle.terminate('cleanup') + end + end + + def test_update_options_on_paused_activity + with_activity_worker([QuickActivity]) do |task_queue| + activity_id = "act-#{SecureRandom.uuid}" + # Start delayed so the activity sits SCHEDULED and pauses to a true PAUSED state rather than + # the PAUSE_REQUESTED a running activity lands in. + handle = env.client.start_activity( + QuickActivity, + id: activity_id, task_queue: task_queue, + start_to_close_timeout: 45, schedule_to_close_timeout: 120, start_delay: 60.0 + ) + handle.pause('hold') + assert_eventually do + assert_equal Temporalio::Client::PendingActivityState::PAUSED, handle.describe.run_state + end + + # Updating options is legal while paused, and the new value lands. Whole-second timeouts + # round-trip exactly through the protobuf Duration conversion, so assert on equality. + updated = handle.update_options(start_to_close_timeout: 90.0) + assert_equal 90.0, updated.start_to_close_timeout + + desc = handle.describe + assert_equal 90.0, desc.start_to_close_timeout + # The mask is still honored while paused — an option we didn't touch keeps its original value. + assert_equal 120.0, desc.schedule_to_close_timeout + # And the update leaves the activity paused; it is not an implicit unpause. + assert_equal Temporalio::Client::PendingActivityState::PAUSED, desc.run_state + assert_equal Temporalio::Client::ActivityExecutionStatus::PAUSED, desc.status + + handle.terminate('cleanup') + end + end + + def test_describe_paused_activity_reports_paused_status + with_activity_worker([QuickActivity]) do |task_queue| + activity_id = "act-#{SecureRandom.uuid}" + # Start delayed so the activity sits SCHEDULED; pausing from there reaches a true PAUSED + # state rather than the PAUSE_REQUESTED of a running activity. `status` is the overall + # ActivityExecutionStatus (api#834 added PAUSED to it); `run_state` is the finer-grained + # PendingActivityState. Both should read PAUSED here. + handle = env.client.start_activity( + QuickActivity, + id: activity_id, task_queue: task_queue, start_to_close_timeout: 60, start_delay: 30.0 + ) + # Before the pause the activity is simply RUNNING (scheduled, not yet started). + assert_equal Temporalio::Client::ActivityExecutionStatus::RUNNING, handle.describe.status + + handle.pause('hold') + assert_eventually do + desc = handle.describe + assert_equal Temporalio::Client::ActivityExecutionStatus::PAUSED, desc.status + assert_equal Temporalio::Client::PendingActivityState::PAUSED, desc.run_state + end + handle.terminate('cleanup') + end + end + + def test_reset_keeps_paused + with_activity_worker([QuickActivity]) do |task_queue| + activity_id = "act-#{SecureRandom.uuid}" + # Start delayed so the activity sits SCHEDULED and pauses to a true PAUSED state (not the + # PAUSE_REQUESTED of a running activity), which is what keep_paused must preserve across reset. + handle = env.client.start_activity( + QuickActivity, + id: activity_id, task_queue: task_queue, start_to_close_timeout: 60, start_delay: 30.0 + ) + handle.pause('hold') + assert_eventually do + assert_equal Temporalio::Client::PendingActivityState::PAUSED, handle.describe.run_state + end + + handle.reset(keep_paused: true) + # keep_paused means the activity remains paused across the reset. + # DIAGNOSTIC (2026-07-30): bumped from default 10s to 60s. + assert_eventually(timeout: 60.0) do + assert_equal Temporalio::Client::PendingActivityState::PAUSED, handle.describe.run_state + end + handle.terminate('cleanup') + end + end + + def test_reset_restores_original_options + with_activity_worker([SlowActivity]) do |task_queue| + handle = start_running_slow_activity(task_queue, start_to_close_timeout: 45) + + updated = handle.update_options(start_to_close_timeout: 90.0) + assert_equal 90.0, updated.start_to_close_timeout + + handle.reset(restore_original_options: true) + # restore_original_options reverts the changed option to the value the activity was created with. + # DIAGNOSTIC (2026-07-30): bumped from 30s to 60s. + assert_eventually(timeout: 60.0) do + assert_equal 45.0, handle.describe.start_to_close_timeout + end + handle.terminate('cleanup') + end + end + + # Start a HeartbeatOnceActivity and wait until its first attempt has recorded heartbeat details. + # The activity keeps running (sleeping until cancellation) once heartbeat has fired, so pause + # transitions the activity through PAUSE_REQUESTED to PAUSED — assert_eventually_paused tolerates + # both. + def start_heartbeat_ready_activity(task_queue) + activity_id = "act-#{SecureRandom.uuid}" + handle = env.client.start_activity( + HeartbeatOnceActivity, + id: activity_id, task_queue: task_queue, start_to_close_timeout: 60, heartbeat_timeout: 30 + ) + assert_eventually do + assert handle.describe(include_heartbeat_details: true).has_heartbeat_details? + end + handle + end + + # The payload-bearing describe fields are opt-in (api#792). Assert the default really is "off" + # rather than the SDK quietly requesting everything: same activity, same moment, two describes. + def test_describe_payload_fields_are_opt_in + with_activity_worker([HeartbeatOnceActivity]) do |task_queue| + handle = start_heartbeat_ready_activity(task_queue) + refute handle.describe.has_heartbeat_details? + assert_empty handle.describe.heartbeat_details + assert handle.describe(include_heartbeat_details: true).has_heartbeat_details? + assert_equal ['hb-details'], handle.describe(include_heartbeat_details: true).heartbeat_details + handle.terminate('cleanup') + end + end + + # Input and outcome are opt-in like the other payload fields, and the outcome is a + # result-or-failure oneof. A successful activity populates the result arm only. + def test_describe_input_and_result_are_opt_in + with_activity_worker([EchoActivity]) do |task_queue| + handle = env.client.start_activity( + EchoActivity, 'ping', + id: "act-#{SecureRandom.uuid}", task_queue: task_queue, start_to_close_timeout: 60 + ) + assert_equal 'ping-echoed', handle.result + + desc = handle.describe + refute desc.has_input? + assert_empty desc.input + refute desc.has_result? + assert_nil desc.result + + desc = handle.describe(include_input: true, include_outcome: true) + assert desc.has_input? + assert_equal ['ping'], desc.input + assert desc.has_result? + assert_equal 'ping-echoed', desc.result + # A successful outcome has no failure arm. + assert_nil desc.failure + end + end + + # The other arm of the oneof: a terminally failed activity has a failure and no result. + def test_describe_outcome_failure + with_activity_worker([AlwaysFailActivity]) do |task_queue| + handle = env.client.start_activity( + AlwaysFailActivity, + id: "act-#{SecureRandom.uuid}", task_queue: task_queue, start_to_close_timeout: 60, + retry_policy: Temporalio::RetryPolicy.new(max_attempts: 1) + ) + assert_raises(Temporalio::Error::ActivityFailedError) { handle.result } + + desc = handle.describe(include_outcome: true) + refute desc.has_result? + assert_nil desc.result + failure = desc.failure + assert_instance_of Temporalio::Error::ApplicationError, failure + assert_equal 'deliberate failure', failure&.message + end + end + + def test_pause_preserves_heartbeat + with_activity_worker([HeartbeatOnceActivity]) do |task_queue| + handle = start_heartbeat_ready_activity(task_queue) + handle.pause('hold') + assert_eventually_paused(handle) + # Pause never touches heartbeat details — they persist across the transition. + assert handle.describe(include_heartbeat_details: true).has_heartbeat_details? + handle.terminate('cleanup') + end + end + + def test_unpause_preserves_heartbeat + with_activity_worker([HeartbeatOnceActivity]) do |task_queue| + handle = start_heartbeat_ready_activity(task_queue) + handle.pause('hold') + assert_eventually_paused(handle) + + # Unpause preserves heartbeat details. The re-dispatched attempt doesn't heartbeat (only + # attempt 1 does), so the persisted details are stable and observable. + handle.unpause + assert_eventually do + assert handle.describe(include_heartbeat_details: true).has_heartbeat_details? + end + handle.terminate('cleanup') + end + end + + def test_reset_preserves_heartbeat_by_default + with_activity_worker([HeartbeatOnceActivity]) do |task_queue| + handle = start_heartbeat_ready_activity(task_queue) + handle.pause('hold') + assert_eventually_paused(handle) + + # As of api#848 / temporal#11417, reset does NOT clear heartbeat details by default — + # you must pass reset_heartbeat: true. keep_paused so no new attempt reshapes state. + handle.reset(keep_paused: true) + # Give the server time to persist any state change, then confirm details survive. + sleep 2 + assert handle.describe(include_heartbeat_details: true).has_heartbeat_details? + handle.terminate('cleanup') + end + end + + def test_reset_clears_heartbeat_when_flag_set + with_activity_worker([HeartbeatOnceActivity]) do |task_queue| + handle = start_heartbeat_ready_activity(task_queue) + handle.pause('hold') + assert_eventually_paused(handle) + + # Opt-in flag clears details. + handle.reset(keep_paused: true, reset_heartbeat: true) + assert_eventually(timeout: 30.0) do + refute handle.describe(include_heartbeat_details: true).has_heartbeat_details? + end + handle.terminate('cleanup') + end + end + + def test_update_options_preserves_heartbeat + with_activity_worker([HeartbeatOnceActivity]) do |task_queue| + handle = start_heartbeat_ready_activity(task_queue) + handle.pause('hold') + assert_eventually_paused(handle) + + # UpdateOptions changes activity options only; it never touches heartbeat details. + handle.update_options(start_to_close_timeout: 90.0) + assert handle.describe(include_heartbeat_details: true).has_heartbeat_details? + handle.terminate('cleanup') + end + end +end diff --git a/temporalio/test/client_activity_test.rb b/temporalio/test/client_activity_test.rb index af9bb673..25ac7ba3 100644 --- a/temporalio/test/client_activity_test.rb +++ b/temporalio/test/client_activity_test.rb @@ -305,6 +305,7 @@ def test_describe_running_and_terminated_is_accurate assert_equal 'SlowActivity', desc.activity_type # Status should be RUNNING (1). assert_equal Temporalio::Client::ActivityExecutionStatus::RUNNING, desc.status + refute_nil desc.execution_time handle.terminate('test-termination') # After terminate, status should reach TERMINATED eventually. diff --git a/temporalio/test/sig/client_activity_operator_commands_build_test.rbs b/temporalio/test/sig/client_activity_operator_commands_build_test.rbs new file mode 100644 index 00000000..21dec7d5 --- /dev/null +++ b/temporalio/test/sig/client_activity_operator_commands_build_test.rbs @@ -0,0 +1,2 @@ +class ClientActivityOperatorCommandsBuildTest < Test +end diff --git a/temporalio/test/sig/client_activity_operator_commands_interceptor_test.rbs b/temporalio/test/sig/client_activity_operator_commands_interceptor_test.rbs new file mode 100644 index 00000000..c1398b72 --- /dev/null +++ b/temporalio/test/sig/client_activity_operator_commands_interceptor_test.rbs @@ -0,0 +1,3 @@ +class ClientActivityOperatorCommandsInterceptorTest < Test + def client_with_interceptor: (Array[untyped] events) -> Temporalio::Client +end diff --git a/temporalio/test/sig/client_activity_operator_commands_test.rbs b/temporalio/test/sig/client_activity_operator_commands_test.rbs new file mode 100644 index 00000000..673b2513 --- /dev/null +++ b/temporalio/test/sig/client_activity_operator_commands_test.rbs @@ -0,0 +1,13 @@ +class ClientActivityOperatorCommandsTest < Test + PAUSED_STATES: Array[Integer] + + def assert_eventually_paused: (Temporalio::Client::ActivityHandle handle) -> void + + def with_activity_worker: (Array[singleton(Temporalio::Activity::Definition)] activities) { (String) -> untyped } -> untyped + + def start_running_slow_activity: (String task_queue, **untyped kwargs) -> Temporalio::Client::ActivityHandle + + def start_backed_off_heartbeat_activity: (String task_queue) -> Temporalio::Client::ActivityHandle + + def start_heartbeat_ready_activity: (String task_queue) -> Temporalio::Client::ActivityHandle +end