Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -130,6 +130,7 @@ def execute_activity(
def execute_local_activity(
activity,
*args,
summary:,
schedule_to_close_timeout:,
schedule_to_start_timeout:,
start_to_close_timeout:,
Expand Down Expand Up @@ -157,6 +158,7 @@ def execute_local_activity(
Temporalio::Worker::Interceptor::Workflow::ExecuteLocalActivityInput.new(
activity:,
args:,
summary:,
schedule_to_close_timeout:,
schedule_to_start_timeout:,
start_to_close_timeout:,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -114,7 +114,8 @@ def execute_local_activity(input)
local_retry_threshold: ProtoUtils.seconds_to_duration(input.local_retry_threshold),
attempt: do_backoff&.attempt || 0,
original_schedule_time: do_backoff&.original_schedule_time
)
),
user_metadata: ProtoUtils.to_user_metadata(input.summary, nil, @instance.payload_converter)
)
)
seq
Expand Down
1 change: 1 addition & 0 deletions temporalio/lib/temporalio/worker/interceptor.rb
Original file line number Diff line number Diff line change
Expand Up @@ -228,6 +228,7 @@ def handle_update(input)
ExecuteLocalActivityInput = Data.define(
:activity,
:args,
:summary,
:schedule_to_close_timeout,
:schedule_to_start_timeout,
:start_to_close_timeout,
Expand Down
9 changes: 6 additions & 3 deletions temporalio/lib/temporalio/workflow.rb
Original file line number Diff line number Diff line change
Expand Up @@ -208,14 +208,16 @@ def self.execute_child_workflow(
#
# @param activity [Class<Activity::Definition>, Symbol, String] Activity definition class or name.
# @param args [Array<Object>] Arguments to the activity.
# @param summary [String, nil] Single-line summary for this activity that may appear in CLI/UI. This can be in
# single-line Temporal markdown format. This is currently experimental.
# @param schedule_to_close_timeout [Float, nil] Max amount of time the activity can take from first being scheduled
# to being completed before it times out. This is inclusive of all retries.
# @param schedule_to_start_timeout [Float, nil] Max amount of time the activity can take to be started from first
# being scheduled.
# @param start_to_close_timeout [Float, nil] Max amount of time a single activity run can take from when it starts
# to when it completes. This is per retry.
# @param retry_policy [RetryPolicy] How an activity is retried on failure. If unset, a server-defined default is
# used. Set maximum attempts to 1 to disable retries.
# @param retry_policy [RetryPolicy, nil] How an activity is retried on failure. If unset, a default policy is used.
# Set maximum attempts to 1 to disable retries.
# @param local_retry_threshold [Float, nil] If the activity is retrying and backoff would exceed this value, a timer
# is scheduled and the activity is retried after. Otherwise, backoff will happen internally within the task.
# Defaults to 1 minute.
Expand All @@ -238,6 +240,7 @@ def self.execute_child_workflow(
def self.execute_local_activity(
activity,
*args,
summary: nil,
schedule_to_close_timeout: nil,
schedule_to_start_timeout: nil,
start_to_close_timeout: nil,
Expand All @@ -251,7 +254,7 @@ def self.execute_local_activity(
)
_current.execute_local_activity(
activity, *args,
schedule_to_close_timeout:, schedule_to_start_timeout:, start_to_close_timeout:,
summary:, schedule_to_close_timeout:, schedule_to_start_timeout:, start_to_close_timeout:,
retry_policy:, local_retry_threshold:, cancellation:, cancellation_type:,
activity_id:, arg_hints:, result_hint:
)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,7 @@ module Temporalio
def execute_local_activity: (
singleton(Activity::Definition) | Symbol | String activity,
*Object? args,
summary: String?,
schedule_to_close_timeout: duration?,
schedule_to_start_timeout: duration?,
start_to_close_timeout: duration?,
Expand Down
2 changes: 2 additions & 0 deletions temporalio/sig/temporalio/worker/interceptor.rbs
Original file line number Diff line number Diff line change
Expand Up @@ -175,6 +175,7 @@ module Temporalio
class ExecuteLocalActivityInput
attr_reader activity: String
attr_reader args: Array[Object?]
attr_reader summary: String?
attr_reader schedule_to_close_timeout: duration?
attr_reader schedule_to_start_timeout: duration?
attr_reader start_to_close_timeout: duration?
Expand All @@ -190,6 +191,7 @@ module Temporalio
def initialize: (
activity: String,
args: Array[Object?],
summary: String?,
schedule_to_close_timeout: duration?,
schedule_to_start_timeout: duration?,
start_to_close_timeout: duration?,
Expand Down
1 change: 1 addition & 0 deletions temporalio/sig/temporalio/workflow.rbs
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,7 @@ module Temporalio
def self.execute_local_activity: (
singleton(Activity::Definition) | Symbol | String activity,
*Object? args,
?summary: String?,
?schedule_to_close_timeout: duration?,
?schedule_to_start_timeout: duration?,
?start_to_close_timeout: duration?,
Expand Down
34 changes: 34 additions & 0 deletions temporalio/test/worker_workflow_activity_test.rb
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

require 'securerandom'
require 'temporalio/client'
require 'temporalio/converters/data_converter'
require 'temporalio/testing'
require 'temporalio/worker'
require 'temporalio/workflow'
Expand Down Expand Up @@ -29,6 +30,12 @@ def execute(scenario)
Temporalio::Workflow.execute_local_activity(:SimpleActivity, 'local', start_to_close_timeout: 10)
when :local_string_name
Temporalio::Workflow.execute_local_activity('SimpleActivity', 'local', start_to_close_timeout: 10)
when :remote_with_summary
Temporalio::Workflow.execute_activity(SimpleActivity, 'remote',
start_to_close_timeout: 10, summary: 'remote summary')
when :local_with_summary
Temporalio::Workflow.execute_local_activity(SimpleActivity, 'local',
start_to_close_timeout: 10, summary: 'local summary')
else
raise NotImplementedError
end
Expand Down Expand Up @@ -368,4 +375,31 @@ def test_cancellation_reset
assert_equal 'canceled - paused: false, requested: false, reset: true', handle.result
end
end

def test_activity_summary
data_converter = Temporalio::Converters::DataConverter.default
execute_workflow(SimpleWorkflow, :remote_with_summary, activities: [SimpleActivity]) do |handle|
handle.result
activity_events = handle.fetch_history.events
.select { |e| e.event_type == :EVENT_TYPE_ACTIVITY_TASK_SCHEDULED }
assert_equal 1, activity_events.size
assert_equal 'remote summary', data_converter.from_payload(activity_events.first.user_metadata.summary)
assert_nil activity_events.first.user_metadata.details
end
end

def test_local_activity_summary
data_converter = Temporalio::Converters::DataConverter.default
execute_workflow(SimpleWorkflow, :local_with_summary, activities: [SimpleActivity]) do |handle|
handle.result
print handle.fetch_history.events
activity_events = handle.fetch_history.events.select do |e|
e.event_type == :EVENT_TYPE_MARKER_RECORDED &&
e.marker_recorded_event_attributes.marker_name == 'core_local_activity'
end
assert_equal 1, activity_events.size
assert_equal 'local summary', data_converter.from_payload(activity_events.first.user_metadata.summary)
assert_nil activity_events.first.user_metadata.details
end
end
end
Loading