diff --git a/CHANGELOG.md b/CHANGELOG.md index af1c7b51..8b8c9aea 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -21,12 +21,23 @@ to docs, or any other relevant information. ### Added -### Changed +#### Standalone Activity operator commands -### Deprecated +- `Client::ActivityHandle` now supports operator commands for standalone activities: `#pause`, + `#unpause` and `#update_options`. +- Added opt-in payload fields to `Client::ActivityHandle#describe`: `include_input:`, + `include_outcome:`, `include_heartbeat_details:` and `include_last_failure:`, all default + `false`. +- Add missing description fields: `execution_time`, `start_delay`, `total_heartbeat_count`. ### :boom: Breaking Changes +- `Description` payload fields are now opt-in: `input`, `outcome`, `heartbeat_details`, `last_failure`. + +### Changed + +### Deprecated + ### Fixed - Canceling a fiber-executor activity no longer wedges the worker if using a `Fiber#transfer` based scheduler (such as `async`). The exception is now delivered through the scheduler's `fiber_interrupt` hook when it diff --git a/temporalio/lib/temporalio/client/activity_execution.rb b/temporalio/lib/temporalio/client/activity_execution.rb index 36fe45d2..ef9b50bb 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,20 @@ def attempt @raw_info.attempt end + # @return [Integer] Total number of heartbeats recorded across all attempts. + def total_heartbeat_count + @raw_info.total_heartbeat_count + 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..17719cc3 100644 --- a/temporalio/lib/temporalio/client/activity_handle.rb +++ b/temporalio/lib/temporalio/client/activity_handle.rb @@ -2,8 +2,13 @@ require 'temporalio/api' require 'temporalio/client/activity_execution' +require 'temporalio/client/activity_execution_options' +require 'temporalio/client/activity_options' require 'temporalio/client/interceptor' require 'temporalio/error' +require 'temporalio/internal/proto_utils' +require 'temporalio/priority' +require 'temporalio/retry_policy' module Temporalio class Client @@ -55,15 +60,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 +128,102 @@ 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 + + # Update the activity's options. Only the options named by `updates` are changed; anything + # not named is left as-is. + # + # Updates are created from the keys on {ActivityOptions}, via {ActivityOptions::Key#value_set} + # to set an option or {ActivityOptions::Key#value_unset} to clear it. + # + # WARNING: Standalone Activities are experimental. + # + # @param updates [Array] The option updates to apply. At least one is + # required unless `restore_original` is true. + # @param restore_original [Boolean] If true, restore the options to the originals the activity + # was created with. Mutually exclusive with any update. + # @param rpc_options [RPCOptions, nil] Advanced RPC options. + # + # @return [ActivityExecutionOptions] The activity options after the update. + # + # @raise [ArgumentError] If a non-update is given, if `restore_original` is combined with any + # update, or if no update is provided and `restore_original` is false. + # @raise [Error::RPCError] RPC error from call. + def update_options(*updates, restore_original: false, rpc_options: nil) + unless updates.all?(ActivityOptions::Update) + raise ArgumentError, + 'Updates must be created via ActivityOptions::Key#value_set or #value_unset' + end + + if restore_original && !updates.empty? + raise ArgumentError, 'restore_original cannot be combined with any option update' + elsif !restore_original && updates.empty? + raise ArgumentError, + 'At least one option update must be given, or restore_original must be used' + end + + # For repeated keys, later values override previous ones. + by_path = updates.to_h { |update| [update.key.name, update] } + + proto = Api::Activity::V1::ActivityOptions.new + by_path.each_value do |update| + # An unset update names its path but leaves the field absent, which is how the server is + # told to clear the option rather than set it to a value. + update.key._apply(proto, update.value) unless update.value.nil? + end + + @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: by_path.keys), + restore_original:, + rpc_options: + ) + ) + end + private def _process_outcome(outcome, hint) diff --git a/temporalio/lib/temporalio/client/activity_options.rb b/temporalio/lib/temporalio/client/activity_options.rb new file mode 100644 index 00000000..73c2f953 --- /dev/null +++ b/temporalio/lib/temporalio/client/activity_options.rb @@ -0,0 +1,113 @@ +# frozen_string_literal: true + +require 'temporalio/api' +require 'temporalio/internal/proto_utils' + +module Temporalio + class Client + # The activity options that {ActivityHandle#update_options} can change. + # + # Updates are created from the keys below, via {Key#value_set} to set an option or + # {Key#value_unset} to clear it. An option with no update is left untouched. + # + # WARNING: Standalone Activities are experimental. + module ActivityOptions + # Typed key for one updatable activity option. Use the keys on {ActivityOptions} rather than + # constructing these directly. + class Key + # @return [String] Field-mask path this key updates. + attr_reader :name + + # @!visibility private + def initialize(name, &to_proto) + @name = name + @to_proto = to_proto + freeze + end + + # Create an update that sets this option to the given value. + # + # @param value [Object] Value to set. Cannot be nil. + # @return [Update] Created update. + def value_set(value) + raise ArgumentError, 'Value cannot be nil, use value_unset' if value.nil? + + Update.new(self, value) + end + + # Create an update that clears this option server-side. + # + # @return [Update] Created update. + def value_unset + Update.new(self, nil) + end + + # @!visibility private + def _apply(proto, value) + @to_proto.call(proto, value) + end + end + + # A single change to an activity's options that can be separately applied. + class Update + # @return [Key] Key this update applies to. + attr_reader :key + + # @return [Object, nil] Value to set, or `nil` to clear the option. + attr_reader :value + + # Create an update. Users may find it easier to use {Key#value_set} and {Key#value_unset}. + # + # @param key [Key] Key to update. + # @param value [Object, nil] Value to set, or nil to clear the option. + def initialize(key, value) + raise ArgumentError, 'Key must be a key' unless key.is_a?(Key) + + @key = key + @value = value + freeze + end + end + + # @return [Key] New task queue. + TASK_QUEUE = Key.new('task_queue.name') do |proto, value| + proto.task_queue = Api::TaskQueue::V1::TaskQueue.new(name: value.to_s) + end + + # @return [Key] New schedule-to-close timeout in seconds. + SCHEDULE_TO_CLOSE_TIMEOUT = Key.new('schedule_to_close_timeout') do |proto, value| + proto.schedule_to_close_timeout = Internal::ProtoUtils.seconds_to_duration(value) + end + + # @return [Key] New schedule-to-start timeout in seconds. + SCHEDULE_TO_START_TIMEOUT = Key.new('schedule_to_start_timeout') do |proto, value| + proto.schedule_to_start_timeout = Internal::ProtoUtils.seconds_to_duration(value) + end + + # @return [Key] New start-to-close timeout in seconds. + START_TO_CLOSE_TIMEOUT = Key.new('start_to_close_timeout') do |proto, value| + proto.start_to_close_timeout = Internal::ProtoUtils.seconds_to_duration(value) + end + + # @return [Key] New heartbeat timeout in seconds. + HEARTBEAT_TIMEOUT = Key.new('heartbeat_timeout') do |proto, value| + proto.heartbeat_timeout = Internal::ProtoUtils.seconds_to_duration(value) + end + + # @return [Key] New start delay in seconds. + START_DELAY = Key.new('start_delay') do |proto, value| + proto.start_delay = Internal::ProtoUtils.seconds_to_duration(value) + end + + # @return [Key] New retry policy. + RETRY_POLICY = Key.new('retry_policy') do |proto, value| + proto.retry_policy = value._to_proto + end + + # @return [Key] New priority. + PRIORITY = Key.new('priority') do |proto, value| + proto.priority = value._to_proto + end + end + end +end diff --git a/temporalio/lib/temporalio/client/interceptor.rb b/temporalio/lib/temporalio/client/interceptor.rb index 33575b02..af2ed899 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,39 @@ 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.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 +624,34 @@ 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.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..33fbddb1 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,7 +1025,11 @@ 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) ) @@ -1062,6 +1066,54 @@ 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 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..5b849ad8 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 @@ -94,6 +118,9 @@ class Temporalio::Client::ActivityExecution::Description < ::Temporalio::Client: sig { returns(Integer) } def attempt; end + sig { returns(Integer) } + def total_heartbeat_count; end + sig { returns(T.nilable(Temporalio::Error::Failure)) } def last_failure; 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..4320ec85 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,31 @@ 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( + updates: Temporalio::Client::ActivityOptions::Update, + restore_original: T::Boolean, + rpc_options: T.nilable(Temporalio::Client::RPCOptions) + ).returns(Temporalio::Client::ActivityExecutionOptions) + end + def update_options( + *updates, + restore_original: T.unsafe(nil), + rpc_options: T.unsafe(nil) + ); end + private sig do diff --git a/temporalio/rbi/temporalio/client/activity_options.rbi b/temporalio/rbi/temporalio/client/activity_options.rbi new file mode 100644 index 00000000..e2f031bc --- /dev/null +++ b/temporalio/rbi/temporalio/client/activity_options.rbi @@ -0,0 +1,49 @@ +# typed: strong + +module Temporalio + class Client + module ActivityOptions + class Key + sig { returns(String) } + def name; end + + sig do + params( + name: String, + to_proto: T.proc.params(proto: Temporalio::Api::Activity::V1::ActivityOptions, value: Object).void + ).void + end + def initialize(name, &to_proto); end + + sig { params(value: Object).returns(Temporalio::Client::ActivityOptions::Update) } + def value_set(value); end + + sig { returns(Temporalio::Client::ActivityOptions::Update) } + def value_unset; end + + sig { params(proto: Temporalio::Api::Activity::V1::ActivityOptions, value: Object).void } + def _apply(proto, value); end + end + + class Update + sig { returns(Temporalio::Client::ActivityOptions::Key) } + def key; end + + sig { returns(T.nilable(Object)) } + def value; end + + sig { params(key: Temporalio::Client::ActivityOptions::Key, value: T.nilable(Object)).void } + def initialize(key, value); end + end + + TASK_QUEUE = T.let(T.unsafe(nil), Temporalio::Client::ActivityOptions::Key) + SCHEDULE_TO_CLOSE_TIMEOUT = T.let(T.unsafe(nil), Temporalio::Client::ActivityOptions::Key) + SCHEDULE_TO_START_TIMEOUT = T.let(T.unsafe(nil), Temporalio::Client::ActivityOptions::Key) + START_TO_CLOSE_TIMEOUT = T.let(T.unsafe(nil), Temporalio::Client::ActivityOptions::Key) + HEARTBEAT_TIMEOUT = T.let(T.unsafe(nil), Temporalio::Client::ActivityOptions::Key) + START_DELAY = T.let(T.unsafe(nil), Temporalio::Client::ActivityOptions::Key) + RETRY_POLICY = T.let(T.unsafe(nil), Temporalio::Client::ActivityOptions::Key) + PRIORITY = T.let(T.unsafe(nil), Temporalio::Client::ActivityOptions::Key) + end + end +end diff --git a/temporalio/rbi/temporalio/client/interceptor.rbi b/temporalio/rbi/temporalio/client/interceptor.rbi index dcea1d74..c95ef3cb 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,88 @@ 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::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 +1150,19 @@ 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 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..9be974bb 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,12 +26,21 @@ 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? def last_started_time: -> Time? def attempt: -> Integer + + def total_heartbeat_count: -> Integer def last_failure: -> Error::Failure? def expiration_time: -> Time? def last_worker_identity: -> String? 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..4df8a032 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,32 @@ 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 update_options: ( + *ActivityOptions::Update updates, + ?restore_original: bool, + ?rpc_options: RPCOptions? + ) -> ActivityExecutionOptions + private def _process_outcome: (Api::Activity::V1::ActivityExecutionOutcome? outcome, Object? hint) -> Object? end end diff --git a/temporalio/sig/temporalio/client/activity_options.rbs b/temporalio/sig/temporalio/client/activity_options.rbs new file mode 100644 index 00000000..41477c0b --- /dev/null +++ b/temporalio/sig/temporalio/client/activity_options.rbs @@ -0,0 +1,30 @@ +module Temporalio + class Client + module ActivityOptions + class Key + attr_reader name: String + + def initialize: (String name) { (untyped proto, untyped value) -> void } -> void + def value_set: (untyped value) -> Update + def value_unset: () -> Update + def _apply: (untyped proto, untyped value) -> void + end + + class Update + attr_reader key: Key + attr_reader value: untyped? + + def initialize: (Key key, untyped? value) -> void + end + + TASK_QUEUE: Key + SCHEDULE_TO_CLOSE_TIMEOUT: Key + SCHEDULE_TO_START_TIMEOUT: Key + START_TO_CLOSE_TIMEOUT: Key + HEARTBEAT_TIMEOUT: Key + START_DELAY: Key + RETRY_POLICY: Key + PRIORITY: Key + end + end +end diff --git a/temporalio/sig/temporalio/client/interceptor.rbs b/temporalio/sig/temporalio/client/interceptor.rbs index a411f81f..cd82b3ad 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,54 @@ 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 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 +667,12 @@ module Temporalio def terminate_activity: (TerminateActivityInput input) -> void + def pause_activity: (PauseActivityInput input) -> void + + def unpause_activity: (UnpauseActivityInput 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..6a53fea3 100644 --- a/temporalio/sig/temporalio/internal/client/implementation.rbs +++ b/temporalio/sig/temporalio/internal/client/implementation.rbs @@ -22,6 +22,13 @@ 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 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_hints_test.rb b/temporalio/test/client_activity_hints_test.rb index 52de72ef..3759058f 100644 --- a/temporalio/test/client_activity_hints_test.rb +++ b/temporalio/test/client_activity_hints_test.rb @@ -101,6 +101,37 @@ def test_activity_hints_from_definition 'Worker-side result encode should use definition result_hint' end + def test_describe_input_and_result_hints + client = build_tracking_client + task_queue = "saa-hints-tq-#{SecureRandom.uuid}" + Temporalio::Worker.new(client:, task_queue:, activities: [HintActivity]).run do + handle = client.start_activity( + HintActivity, 'described', + id: "act-#{SecureRandom.uuid}", task_queue:, start_to_close_timeout: 10 + ) + handle.result + + desc = handle.describe(include_input: true, include_outcome: true) + @hint_converter.inbound_hints = nil + + assert_equal ['described'], desc.input(hints: [:describe_arg]) + assert_equal 'result-of:described', desc.result(result_hint: :describe_result) + + inbound = @hint_converter.inbound_hints || [] + arg_decode = inbound.find { |e| e[:value] == 'described' } + + refute_nil arg_decode + assert_equal :describe_arg, arg_decode[:hint], + 'describe input hints should reach the payload converter' + + result_decode = inbound.find { |e| e[:value] == 'result-of:described' } + + refute_nil result_decode + assert_equal :describe_result, result_decode[:hint], + 'describe result_hint should reach the payload converter' + end + end + def test_activity_hints_call_site_override client = build_tracking_client task_queue = "saa-hints-tq-#{SecureRandom.uuid}" 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..1062789f --- /dev/null +++ b/temporalio/test/client_activity_operator_commands_build_test.rb @@ -0,0 +1,130 @@ +# frozen_string_literal: true + +require 'temporalio/api' +require 'temporalio/client' +require 'test' + +class ClientActivityOperatorCommandsBuildTest < Test + def test_value_set_of_zero_sends_an_explicit_zero + req = capture_update do |handle| + handle.update_options(Temporalio::Client::ActivityOptions::HEARTBEAT_TIMEOUT.value_set(0)) + end + + assert_equal %w[heartbeat_timeout], req.update_mask.paths.sort + # Present and zero, which is distinct from absent: the caller asked for zero. + assert req.activity_options.has_heartbeat_timeout? + assert_equal 0, req.activity_options.heartbeat_timeout.seconds + assert_equal 0, req.activity_options.heartbeat_timeout.nanos + end + + def test_value_unset_names_the_path_but_leaves_the_field_absent + req = capture_update do |handle| + handle.update_options(Temporalio::Client::ActivityOptions::HEARTBEAT_TIMEOUT.value_unset) + end + + assert_equal %w[heartbeat_timeout], req.update_mask.paths.sort + # Absent, which is how the server is told to clear the option. + refute req.activity_options.has_heartbeat_timeout? + end + + def test_mask_names_only_the_changed_options + req = capture_update do |handle| + handle.update_options( + Temporalio::Client::ActivityOptions::TASK_QUEUE.value_set('new-tq'), + Temporalio::Client::ActivityOptions::START_TO_CLOSE_TIMEOUT.value_set(90.0) + ) + end + + assert_equal %w[start_to_close_timeout task_queue.name], req.update_mask.paths.sort + refute req.restore_original + assert_equal 'new-tq', req.activity_options.task_queue.name + assert_equal 90, req.activity_options.start_to_close_timeout.seconds + end + + def test_a_repeated_key_resolves_to_its_last_update + req = capture_update do |handle| + handle.update_options( + Temporalio::Client::ActivityOptions::HEARTBEAT_TIMEOUT.value_set(5.0), + Temporalio::Client::ActivityOptions::HEARTBEAT_TIMEOUT.value_unset + ) + end + + # The later unset wins, and the path is named once. + assert_equal %w[heartbeat_timeout], req.update_mask.paths.sort + refute req.activity_options.has_heartbeat_timeout? + end + + # Runs the block against a handle whose update RPC is stubbed, returning the captured request. + def capture_update + client = Temporalio::Client.connect('localhost:7233', 'test-namespace', lazy_connect: true) + handle = client.activity_handle('act-1') + + ws = client.workflow_service + captured = {} + 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 + + yield handle + captured.fetch(:update) + end + + def test_omitted_jitter_is_left_off_the_wire + # 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') + + ws = client.workflow_service + captured = {} + + ws.define_singleton_method(:unpause_activity_execution) do |req, **_kwargs| + captured[:unpause] = req + Temporalio::Api::WorkflowService::V1::UnpauseActivityExecutionResponse.new + end + + begin + handle.unpause + ensure + ws.singleton_class.send(:remove_method, :unpause_activity_execution) + end + + refute captured.fetch(:unpause).has_jitter? + end + + 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 + + begin + handle.pause('because') + handle.unpause(reason: 'go', jitter: 5.0) + ensure + ws.singleton_class.send(:remove_method, :pause_activity_execution) + ws.singleton_class.send(:remove_method, :unpause_activity_execution) + end + + pause_req = captured.fetch(:pause) + assert_equal 'because', pause_req.reason + + 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 + 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..b886948b --- /dev/null +++ b/temporalio/test/client_activity_operator_commands_interceptor_test.rb @@ -0,0 +1,127 @@ +# frozen_string_literal: true + +require 'securerandom' +require 'temporalio/client' +require 'temporalio/testing' +require 'temporalio/worker' +require 'test' + +# Verifies each operator command (pause/unpause/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 + + attr_reader :inputs + + def initialize(events_array) + @events = events_array + @inputs = {} + end + + def intercept_client(next_interceptor) + Outbound.new(next_interceptor, @events, @inputs) + end + + class Outbound < Temporalio::Client::Interceptor::Outbound + def initialize(next_interceptor, events, inputs) + super(next_interceptor) + @events = events + @inputs = inputs + end + + def pause_activity(input) + @events << 'pause_activity' + @inputs[:pause_activity] = input + super + end + + def unpause_activity(input) + @events << 'unpause_activity' + @inputs[:unpause_activity] = input + super + end + + def update_activity_options(input) + @events << 'update_activity_options' + @inputs[:update_activity_options] = input + super + end + end + end + + def client_with_interceptor(events, recorder: nil) + interceptor = recorder || 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(Temporalio::Client::ActivityOptions::START_TO_CLOSE_TIMEOUT.value_set(90.0)) + + handle.terminate('cleanup') + end + + assert_includes events, 'pause_activity' + assert_includes events, 'unpause_activity' + assert_includes events, 'update_activity_options' + end + + # Asserts the values a caller passes reach the interceptor chain, not merely that the hook + # fired. A dropped argument between the handle and the chain is invisible to a test that only + # checks which events were recorded. + def test_interceptor_receives_command_arguments + events = [] + recorder = RecordingInterceptor.new(events) + client = client_with_interceptor(events, recorder:) + task_queue = "saa-tq-#{SecureRandom.uuid}" + worker = Temporalio::Worker.new(client:, task_queue:, activities: [SlowActivity]) + worker.run do + handle = client.start_activity( + SlowActivity, + id: "act-#{SecureRandom.uuid}", 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('pause-reason') + handle.unpause(reason: 'unpause-reason', jitter: 5.0) + handle.terminate('cleanup') + end + + assert_equal 'pause-reason', recorder.inputs[:pause_activity].reason + assert_equal 'unpause-reason', recorder.inputs[:unpause_activity].reason + assert_equal 5.0, recorder.inputs[:unpause_activity].jitter + 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..eaed3614 --- /dev/null +++ b/temporalio/test/client_activity_operator_commands_test.rb @@ -0,0 +1,465 @@ +# 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 / 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 + + # Heartbeats continuously. SlowActivity beats only once, so it cannot drive a heartbeat count + # past one however long the test waits. + class FastHeartbeatActivity < Temporalio::Activity::Definition + def execute + ctx = Temporalio::Activity::Context.current + until ctx.cancellation.canceled? + ctx.heartbeat + sleep 0.05 + end + 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 + + # 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 + + # Heartbeats, fails the first attempt, then succeeds. One execution of this carries input, a + # result, heartbeat details and a last failure all at once, which is what lets a single + # describe exercise every payload field. + class HeartbeatFailIncrementActivity < Temporalio::Activity::Definition + def execute(value) + ctx = Temporalio::Activity::Context.current + ctx.heartbeat('heartbeat details') + raise Temporalio::Error::ApplicationError, 'deliberate first-attempt failure' if ctx.info.attempt == 1 + + value + 1 + 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 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 + assert_eventually do + refute_includes PAUSED_STATES, handle.describe.run_state + 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(Temporalio::Client::ActivityOptions::START_TO_CLOSE_TIMEOUT.value_set(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( + Temporalio::Client::ActivityOptions::TASK_QUEUE.value_set('updated-tq'), + Temporalio::Client::ActivityOptions::SCHEDULE_TO_CLOSE_TIMEOUT.value_set(200.0), + Temporalio::Client::ActivityOptions::SCHEDULE_TO_START_TIMEOUT.value_set(15.0), + Temporalio::Client::ActivityOptions::START_TO_CLOSE_TIMEOUT.value_set(90.0), + Temporalio::Client::ActivityOptions::HEARTBEAT_TIMEOUT.value_set(25.0), + Temporalio::Client::ActivityOptions::RETRY_POLICY.value_set( + Temporalio::RetryPolicy.new(initial_interval: 1.0, backoff_coefficient: 2.0, max_attempts: 7) + ), + Temporalio::Client::ActivityOptions::PRIORITY.value_set(Temporalio::Priority.new(priority_key: 3)), + Temporalio::Client::ActivityOptions::START_DELAY.value_set(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(Temporalio::Client::ActivityOptions::START_TO_CLOSE_TIMEOUT.value_set(5.0), + restore_original: true) + 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(Temporalio::Client::ActivityOptions::START_TO_CLOSE_TIMEOUT.value_set(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(Temporalio::Client::ActivityOptions::START_TO_CLOSE_TIMEOUT.value_set(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}" + 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 + + # 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 + + # 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. + # The count tracks heartbeats the server recorded. + def test_describe_reports_total_heartbeat_count + with_activity_worker([FastHeartbeatActivity]) do |task_queue| + handle = env.client.start_activity( + FastHeartbeatActivity, + id: "act-#{SecureRandom.uuid}", task_queue:, + start_to_close_timeout: 60, heartbeat_timeout: 3 + ) + assert_eventually(timeout: 20.0) do + assert_operator handle.describe.total_heartbeat_count, :>=, 2 + end + handle.terminate('cleanup') + end + end + + def test_describe_payloads + with_activity_worker([HeartbeatFailIncrementActivity, AlwaysFailActivity]) do |task_queue| + handle = env.client.start_activity( + HeartbeatFailIncrementActivity, 1, + id: "act-#{SecureRandom.uuid}", task_queue:, + start_to_close_timeout: 60, heartbeat_timeout: 5, + retry_policy: Temporalio::RetryPolicy.new(max_attempts: 2, initial_interval: 0.1) + ) + + assert_equal 2, handle.result + + # Nothing requested: every payload field is absent. + bare = handle.describe + + refute bare.has_input? + refute bare.has_result? + refute bare.has_heartbeat_details? + refute bare.has_last_failure? + assert_empty bare.input + assert_nil bare.result + assert_nil bare.failure + assert_nil bare.last_failure + + # All four requested. The activity succeeded on its second attempt, so it has a result + # and a last failure at the same time, and no terminal failure. + full = handle.describe( + include_input: true, include_outcome: true, + include_heartbeat_details: true, include_last_failure: true + ) + + assert full.has_input? + assert_equal [1], full.input + assert full.has_result? + assert_equal 2, full.result + assert_nil full.failure + assert full.has_heartbeat_details? + assert_equal ['heartbeat details'], full.heartbeat_details + assert full.has_last_failure? + refute_nil full.last_failure + + failed = env.client.start_activity( + AlwaysFailActivity, + id: "act-#{SecureRandom.uuid}", task_queue:, + start_to_close_timeout: 60, + retry_policy: Temporalio::RetryPolicy.new(max_attempts: 1) + ) + assert_raises(Temporalio::Error) { failed.result } + + desc = failed.describe(include_outcome: true, include_last_failure: 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_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(Temporalio::Client::ActivityOptions::START_TO_CLOSE_TIMEOUT.value_set(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 d9b99a77..89405104 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..5ca02de0 --- /dev/null +++ b/temporalio/test/sig/client_activity_operator_commands_build_test.rbs @@ -0,0 +1,3 @@ +class ClientActivityOperatorCommandsBuildTest < Test + def capture_update: () { (Temporalio::Client::ActivityHandle handle) -> void } -> untyped +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..3ce490f2 --- /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, ?recorder: untyped?) -> 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 diff --git a/temporalio/test/test.rb b/temporalio/test/test.rb index cf7f34ee..ddf0da42 100644 --- a/temporalio/test/test.rb +++ b/temporalio/test/test.rb @@ -256,7 +256,7 @@ def initialize else @server = Temporalio::Testing::WorkflowEnvironment.start_local( logger: Logger.new($stdout), - dev_server_download_version: 'v1.7.1-standalone-nexus-operations', + dev_server_download_version: 'v1.8.3-server-1.32.0-162.0', dev_server_extra_args: [ # Allow continue as new to be immediate '--dynamic-config-value', 'history.workflowIdReuseMinimalInterval="0s"',