From bbcf171f29c1ed4e9d6f2789edd4c9ea06f6a982 Mon Sep 17 00:00:00 2001 From: Maciej Dudkowski Date: Mon, 15 Jun 2026 15:06:09 -0400 Subject: [PATCH] Support Standalone Activities --- crates/client/src/activity.rs | 141 +++++ .../src/activity/activity_execution_info.rs | 570 ++++++++++++++++++ crates/client/src/activity/activity_handle.rs | 191 ++++++ crates/client/src/async_activity_handle.rs | 349 ++++++----- crates/client/src/errors.rs | 179 +++++- crates/client/src/lib.rs | 365 ++++++++++- crates/client/src/options_structs.rs | 194 +++++- crates/client/src/utils.rs | 15 + crates/common-wasm/src/data_converters.rs | 3 +- .../src/data_converters/failure_converter.rs | 12 + crates/common-wasm/src/lib.rs | 54 ++ crates/common/src/lib.rs | 8 +- crates/sdk-core/tests/heavy_tests.rs | 10 +- .../async_activity_client_tests.rs | 19 +- .../integ_tests/standalone_activity_tests.rs | 300 +++++++++ .../integ_tests/workflow_tests/activities.rs | 22 +- crates/sdk-core/tests/main.rs | 1 + crates/sdk/src/activities.rs | 59 +- crates/sdk/src/lib.rs | 14 +- crates/workflow/src/lib.rs | 6 +- crates/workflow/src/workflow_context.rs | 9 +- .../workflow/src/workflow_context/options.rs | 67 +- 22 files changed, 2295 insertions(+), 293 deletions(-) create mode 100644 crates/client/src/activity.rs create mode 100644 crates/client/src/activity/activity_execution_info.rs create mode 100644 crates/client/src/activity/activity_handle.rs create mode 100644 crates/client/src/utils.rs create mode 100644 crates/sdk-core/tests/integ_tests/standalone_activity_tests.rs diff --git a/crates/client/src/activity.rs b/crates/client/src/activity.rs new file mode 100644 index 000000000..7d8a12648 --- /dev/null +++ b/crates/client/src/activity.rs @@ -0,0 +1,141 @@ +mod activity_execution_info; +mod activity_handle; + +use crate::errors::ClientError; +pub use activity_execution_info::{ + ActivityExecutionDescription, ActivityExecutionInfo, ActivityExecutionInfoLike, + ActivityExecutionStatus, PendingActivityState, +}; +pub use activity_handle::ActivityHandle; +use futures_util::{Stream, StreamExt}; +use std::{ + pin::Pin, + task::{Context, Poll}, +}; +use temporalio_common::{ + protos::temporal::api::{ + activity::v1::ActivityExecutionListInfo, + workflowservice::v1::{ + CountActivityExecutionsResponse, count_activity_executions_response, + }, + }, + search_attributes::{SearchAttributeError, SearchAttributeValue}, +}; + +/// A stream of activity executions from a list query. +/// Internally paginates through results from the server. +pub struct ListActivitiesStream { + inner: Pin, ClientError>> + Send>>, + buffer: as IntoIterator>::IntoIter, +} + +impl ListActivitiesStream { + pub(crate) fn new( + stream: impl Stream, ClientError>> + Send + 'static, + ) -> Self { + Self { + inner: Box::pin(stream), + buffer: vec![].into_iter(), + } + } +} + +impl Stream for ListActivitiesStream { + type Item = Result; + + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + loop { + if let Some(info) = self.buffer.next() { + return Poll::Ready(Some(Ok(info.into()))); + } + match self.inner.poll_next_unpin(cx) { + Poll::Ready(Some(Ok(items))) => { + self.buffer = items.into_iter(); + } + Poll::Ready(Some(Err(e))) => { + return Poll::Ready(Some(Err(e))); + } + Poll::Ready(None) => { + return Poll::Ready(None); + } + Poll::Pending => { + return Poll::Pending; + } + } + } + } +} + +/// Result of an activity count operation. +/// +/// If the query includes a group-by clause, `groups` will contain the aggregated +/// counts and `count` will be the sum of all group counts. +#[derive(Debug, Clone)] +pub struct ActivityExecutionCount { + count: usize, + groups: Vec, +} + +impl ActivityExecutionCount { + pub(crate) fn from_response(resp: CountActivityExecutionsResponse) -> Self { + Self { + count: resp.count as usize, + groups: resp + .groups + .into_iter() + .map(ActivityExecutionCountAggregationGroup::from_proto) + .collect(), + } + } + + /// The approximate number of activities matching the query. + /// If grouping was applied, this is the sum of all group counts. + pub fn count(&self) -> usize { + self.count + } + + /// The groups if the query had a group-by clause, or empty if not. + pub fn groups(&self) -> &[ActivityExecutionCountAggregationGroup] { + &self.groups + } +} + +/// Aggregation group from an activity count query with a group-by clause. +#[derive(Debug, Clone)] +pub struct ActivityExecutionCountAggregationGroup { + raw: count_activity_executions_response::AggregationGroup, +} + +impl ActivityExecutionCountAggregationGroup { + fn from_proto(proto: count_activity_executions_response::AggregationGroup) -> Self { + Self { raw: proto } + } + + /// Retrieve a typed group value at `index`. + /// + /// Returns `None` if the index is out of bounds or deserialization fails. + /// Use [`Self::try_get`] for explicit error handling. + pub fn get(&self, index: usize) -> Option { + self.try_get(index).ok().flatten() + } + + /// Retrieve a typed group value at `index`, preserving deserialization + /// errors. + /// + /// Returns `Ok(None)` if the index is out of bounds and `Err` if the + /// payload cannot be deserialized. + pub fn try_get( + &self, + index: usize, + ) -> Result, SearchAttributeError> { + match self.raw.group_values.get(index) { + Some(payload) => T::from_search_attribute_payload(payload).map(Some), + None => Ok(None), + } + } + + /// The approximate number of workflows matching for this group. + pub fn count(&self) -> usize { + self.raw.count as usize + } +} diff --git a/crates/client/src/activity/activity_execution_info.rs b/crates/client/src/activity/activity_execution_info.rs new file mode 100644 index 000000000..bebe85d1a --- /dev/null +++ b/crates/client/src/activity/activity_execution_info.rs @@ -0,0 +1,570 @@ +use crate::Priority; +use std::{ + error::Error, + marker::PhantomData, + time::{Duration, SystemTime}, +}; +use temporalio_common::{ + ActivityDefinition, RetryPolicy, UntypedActivity, WorkerDeploymentVersion, + data_converters::{ + DataConverter, NoopDecodeHint, PayloadConversionError, SerializationContextData, + TemporalDeserializable, + }, + error::IncomingError, + protos::{ + proto_ts_to_system_time, + temporal::api::{ + activity::v1::{ + ActivityExecutionInfo as RawInfo, ActivityExecutionListInfo as RawListInfo, + activity_execution_outcome::Value as ActivityExecutionOutcomeValue, + }, + common::v1::{Payload, Payloads}, + enums::v1::{ + ActivityExecutionStatus as ProtoActivityExecutionStatus, + PendingActivityState as ProtoPendingActivityState, + }, + failure::v1::Failure, + workflowservice::v1::DescribeActivityExecutionResponse, + }, + utilities::TryIntoOrNone, + }, + search_attributes::SearchAttributes, +}; + +/// Common methods of [`ActivityExecutionInfo`] and [`ActivityExecutionDescription`]. +pub trait ActivityExecutionInfoLike { + /// ID of the activity. + fn activity_id(&self) -> &str; + /// Run ID of a particular execution of the activity. + fn activity_run_id(&self) -> &str; + /// Type of the activity. + fn activity_type(&self) -> &str; + /// Time the activity was originally scheduled. + fn schedule_time(&self) -> Option; + /// Time when the activity transitioned to a closed state. + fn close_time(&self) -> Option; + /// A general status for this activity, indicates whether it is currently running or in one of + /// the terminal statuses. + fn status(&self) -> ActivityExecutionStatus; + /// The task queue this activity was scheduled on. + fn task_queue(&self) -> &str; + /// The difference between close time and scheduled time. This field is only populated if + /// the activity is closed. + fn execution_duration(&self) -> Option; +} + +/// Contains basic information about an activity. +/// Obtained from [`Client::list_activities`](crate::Client::list_activities). +pub struct ActivityExecutionInfo { + raw: RawListInfo, +} + +impl From for ActivityExecutionInfo { + fn from(raw: RawListInfo) -> Self { + Self { raw } + } +} + +impl ActivityExecutionInfoLike for ActivityExecutionInfo { + fn activity_id(&self) -> &str { + &self.raw.activity_id + } + + fn activity_run_id(&self) -> &str { + &self.raw.run_id + } + + fn activity_type(&self) -> &str { + self.raw + .activity_type + .as_ref() + .map(|t| t.name.as_str()) + .unwrap_or("") + } + + fn schedule_time(&self) -> Option { + self.raw + .schedule_time + .as_ref() + .and_then(proto_ts_to_system_time) + } + + fn close_time(&self) -> Option { + self.raw + .close_time + .as_ref() + .and_then(proto_ts_to_system_time) + } + + fn status(&self) -> ActivityExecutionStatus { + ProtoActivityExecutionStatus::try_from(self.raw.status) + .map(Into::into) + .unwrap_or(ActivityExecutionStatus::Unknown) + } + + fn task_queue(&self) -> &str { + &self.raw.task_queue + } + + fn execution_duration(&self) -> Option { + self.raw.execution_duration.try_into_or_none() + } +} + +impl ActivityExecutionInfo { + /// Raw Protobuf object from server response. + pub fn raw_info(&self) -> &RawListInfo { + &self.raw + } +} + +/// Contains the current state of the activity execution. +/// Obtained from [`ActivityHandle::describe`](crate::ActivityHandle::describe). +/// +/// To support deserialization of transmitted payloads, the object internally stores a reference +/// to the client used to make the request. For this reason, this type is parametrized with client +/// type. The client reference can be dropped by calling [`simple`](Self::untyped). +pub struct ActivityExecutionDescription +where + ActivityT: ActivityDefinition, +{ + raw_info: RawInfo, + raw_input: Option, + raw_outcome: Option, + data_converter: DataConverter, + serialization_context: SerializationContextData, + _phantom: PhantomData, +} + +impl ActivityExecutionInfoLike for ActivityExecutionDescription +where + ActivityT: ActivityDefinition, +{ + fn activity_id(&self) -> &str { + &self.raw_info.activity_id + } + + fn activity_run_id(&self) -> &str { + &self.raw_info.run_id + } + + fn activity_type(&self) -> &str { + self.raw_info + .activity_type + .as_ref() + .map(|t| t.name.as_str()) + .unwrap_or("") + } + + fn schedule_time(&self) -> Option { + self.raw_info + .schedule_time + .as_ref() + .and_then(proto_ts_to_system_time) + } + + fn close_time(&self) -> Option { + self.raw_info + .close_time + .as_ref() + .and_then(proto_ts_to_system_time) + } + + fn status(&self) -> ActivityExecutionStatus { + ProtoActivityExecutionStatus::try_from(self.raw_info.status) + .map(Into::into) + .unwrap_or(ActivityExecutionStatus::Unknown) + } + + fn task_queue(&self) -> &str { + &self.raw_info.task_queue + } + + fn execution_duration(&self) -> Option { + self.raw_info.execution_duration.try_into_or_none() + } +} + +impl ActivityExecutionDescription +where + ActivityT: ActivityDefinition, +{ + pub(crate) fn new( + data_converter: DataConverter, + serialization_context: SerializationContextData, + response: DescribeActivityExecutionResponse, + ) -> Result> { + let Some(raw_info) = response.info else { + return Err("info missing in describe response".into()); + }; + Ok(Self { + raw_info, + raw_input: response.input, + raw_outcome: response.outcome.and_then(|o| o.value), + data_converter, + serialization_context, + _phantom: PhantomData, + }) + } + + /// Convert to an untyped description object. + pub fn untyped(self) -> ActivityExecutionDescription { + ActivityExecutionDescription { + raw_info: self.raw_info, + raw_input: self.raw_input, + raw_outcome: self.raw_outcome, + data_converter: self.data_converter, + serialization_context: self.serialization_context, + _phantom: PhantomData, + } + } + + /// Raw Protobuf object from server response. + pub fn raw_info(&self) -> &RawInfo { + &self.raw_info + } + + /// True if activity input is present. + /// See [`ActivityDescribeOptions::include_input`](crate::ActivityDescribeOptions::include_input). + /// Use [`input`](Self::input) or [`raw_input`](Self::raw_input) to retrieve it. + pub fn has_input(&self) -> bool { + self.raw_input.is_some() + } + + /// Raw payload of activity input, if it was requested. + pub fn raw_input(&self) -> Option<&Payloads> { + self.raw_input.as_ref() + } + + /// Deserialize activity input. Returns `Ok(None)` if not present. + /// See [`ActivityDescribeOptions::include_input`](crate::ActivityDescribeOptions::include_input). + pub async fn input(&self) -> Result, PayloadConversionError> { + let Some(input) = &self.raw_input else { + return Ok(None); + }; + Ok(Some(self.convert_payloads(input).await?)) + } + + /// True if activity outcome is present. + /// See [`ActivityDescribeOptions::include_outcome`](crate::ActivityDescribeOptions::include_outcome). + /// Use [`outcome`](Self::outcome) or [`raw_outcome`](Self::outcome) to retrieve it. + pub fn has_outcome(&self) -> bool { + self.raw_outcome.is_some() + } + + /// Raw payload of activity output, if it was requested and available. + pub fn raw_outcome(&self) -> Option<&ActivityExecutionOutcomeValue> { + self.raw_outcome.as_ref() + } + + /// Deserialize activity outcome. Returns `Ok(None)` if not present. + /// See [`ActivityDescribeOptions::include_outcome`](crate::ActivityDescribeOptions::include_outcome). + pub async fn outcome( + &self, + ) -> Result>, PayloadConversionError> { + match &self.raw_outcome { + None => Ok(None), + Some(ActivityExecutionOutcomeValue::Result(payloads)) => { + Ok(Some(Ok(self.convert_payloads(payloads).await?))) + } + Some(ActivityExecutionOutcomeValue::Failure(failure)) => { + Ok(Some(Err(self.convert_failure(failure)?))) + } + } + } + + /// More detailed breakdown of [`ActivityExecutionStatus::Running`]. + pub fn run_state(&self) -> PendingActivityState { + ProtoPendingActivityState::try_from(self.raw_info.run_state) + .map(Into::into) + .unwrap_or(PendingActivityState::Unknown) + } + + /// Indicates how long the caller is willing to wait for an activity completion. Limits how long + /// retries will be attempted. + pub fn schedule_to_close_timeout(&self) -> Option { + self.raw_info.schedule_to_close_timeout.try_into_or_none() + } + + /// Limits time an activity task can stay in a task queue before a worker picks it up. This + /// timeout is always non-retryable. + pub fn schedule_to_start_timeout(&self) -> Option { + self.raw_info.schedule_to_start_timeout.try_into_or_none() + } + + /// Maximum time a single activity attempt is allowed to execute after being picked up by + /// a worker. This timeout is always retryable. + pub fn start_to_close_timeout(&self) -> Option { + self.raw_info.start_to_close_timeout.try_into_or_none() + } + + /// Maximum permitted time between successful worker heartbeats. + pub fn heartbeat_timeout(&self) -> Option { + self.raw_info.heartbeat_timeout.try_into_or_none() + } + + /// The retry policy for the activity. + pub fn retry_policy(&self) -> Option { + self.raw_info.retry_policy.clone().map(Into::into) + } + + /// True if heartbeat details are present. + /// See [`ActivityDescribeOptions::include_heartbeat_details`](crate::ActivityDescribeOptions::include_heartbeat_details). + /// Use [`heartbeat_details`](Self::heartbeat_details) or + /// [`raw_info()`](Self::raw_info)`.`[`heartbeat_details`](RawInfo::heartbeat_details) + /// to retrieve them. + pub fn has_heartbeat_details(&self) -> bool { + self.raw_info.heartbeat_details.is_some() + } + + /// Deserialize heartbeat details. Returns `Ok(None)` if not present. + /// See [`ActivityDescribeOptions::include_heartbeat_details`](crate::ActivityDescribeOptions::include_heartbeat_details). + pub async fn heartbeat_details( + &self, + ) -> Result, PayloadConversionError> { + let Some(details) = &self.raw_info.heartbeat_details else { + return Ok(None); + }; + Ok(Some(self.convert_payloads(details).await?)) + } + + /// Time the last heartbeat was recorded. + pub fn last_heartbeat_time(&self) -> Option { + self.raw_info + .last_heartbeat_time + .as_ref() + .and_then(proto_ts_to_system_time) + } + + /// Time the last attempt was started. + pub fn last_started_time(&self) -> Option { + self.raw_info + .last_started_time + .as_ref() + .and_then(proto_ts_to_system_time) + } + + /// The attempt this activity is currently on. Incremented each time a new attempt is scheduled. + pub fn attempt(&self) -> u32 { + self.raw_info.attempt.try_into().unwrap_or_default() + } + + /// How long this activity has been running for, including all attempts and backoff between + /// attempts. + pub fn execution_duration(&self) -> Option { + self.raw_info.execution_duration.try_into_or_none() + } + + /// Scheduled time + schedule to close timeout. + pub fn expiration_time(&self) -> Option { + self.raw_info + .expiration_time + .as_ref() + .and_then(proto_ts_to_system_time) + } + + /// True if last failure is present. + /// See [`ActivityDescribeOptions::include_last_failure`](crate::ActivityDescribeOptions::include_last_failure). + /// Use [`last_failure()`](Self::last_failure) or + /// [`raw_info()`](Self::raw_info)`.`[`last_failure`](RawInfo::last_failure) + /// to retrieve it. + pub fn has_last_failure(&self) -> bool { + self.raw_info.last_failure.is_some() + } + + /// Deserialize last failure. Returns `Ok(None)` if not present. + /// See [`ActivityDescribeOptions::include_last_failure`](crate::ActivityDescribeOptions::include_last_failure). + pub fn last_failure(&self) -> Result, PayloadConversionError> { + let Some(failure) = &self.raw_info.last_failure else { + return Ok(None); + }; + Ok(Some(self.convert_failure(failure)?)) + } + + /// Identity of the last worker that attempted this activity. + pub fn last_worker_identity(&self) -> Option<&str> { + self.raw_info + .last_worker_identity + .is_empty() + .then_some(self.raw_info.last_worker_identity.as_str()) + } + + /// Time from the last attempt failure to the next activity retry. + pub fn current_retry_interval(&self) -> Option { + self.raw_info.current_retry_interval.try_into_or_none() + } + + /// The time when the last activity attempt completed. + pub fn last_attempt_complete_time(&self) -> Option { + self.raw_info + .last_attempt_complete_time + .as_ref() + .and_then(proto_ts_to_system_time) + } + + /// The time when the next activity attempt will be scheduled. + pub fn next_attempt_schedule_time(&self) -> Option { + self.raw_info + .next_attempt_schedule_time + .as_ref() + .and_then(proto_ts_to_system_time) + } + + /// The Worker Deployment Version this activity was dispatched to most recently. + pub fn last_deployment_version(&self) -> Option { + self.raw_info + .last_deployment_version + .clone() + .map(Into::into) + } + + /// Priority metadata. + pub fn priority(&self) -> Priority { + self.raw_info.priority.clone().unwrap_or_default().into() + } + + /// Search attributes of the activity. + pub fn search_attributes(&self) -> Option { + self.raw_info + .search_attributes + .as_ref() + .map(SearchAttributes::from_proto) + } + + /// Deserialize static summary that was set when activity was scheduled. + /// Returns `Ok(None)` if not present. + pub async fn static_summary(&self) -> Result, PayloadConversionError> { + let Some(summary) = self + .raw_info + .user_metadata + .as_ref() + .and_then(|m| m.summary.clone()) + else { + return Ok(None); + }; + Ok(Some(self.convert_payload(summary).await?)) + } + + /// Deserialize static details that were set when activity was scheduled. + /// Returns `Ok(None)` if not present. + pub async fn static_details(&self) -> Result, PayloadConversionError> { + let Some(details) = self + .raw_info + .user_metadata + .as_ref() + .and_then(|m| m.details.clone()) + else { + return Ok(None); + }; + Ok(Some(self.convert_payload(details).await?)) + } + + /// Reason for activity cancellation if activity was canceled and reason was provided. + pub fn canceled_reason(&self) -> Option<&str> { + let reason = self.raw_info.canceled_reason.as_str(); + (!reason.is_empty()).then_some(reason) + } + + /// Time to wait before dispatching the first activity task. + /// This delay is not applied to retry attempts. + pub fn start_delay(&self) -> Option { + self.raw_info.start_delay.try_into_or_none() + } + + async fn convert_payload( + &self, + payload: Payload, + ) -> Result { + self.data_converter + .from_payload(&self.serialization_context, payload) + .await + } + + async fn convert_payloads( + &self, + payloads: &Payloads, + ) -> Result { + self.data_converter + .from_payloads(&self.serialization_context, payloads.payloads.clone()) + .await + } + + fn convert_failure(&self, failure: &Failure) -> Result { + self.data_converter + .to_error(&self.serialization_context, failure.clone(), NoopDecodeHint) + } +} + +/// Execution status of an activity. See [`ActivityExecutionInfoLike::status`]. +#[non_exhaustive] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)] +pub enum ActivityExecutionStatus { + #[default] + /// This variant indicates the server did not specify a value. + Unspecified, + /// The activity has not reached a terminal status. + /// See [`ActivityExecutionDescription::run_state`] for the run state. + Running, + /// The activity completed successfully. + Completed, + /// The activity failed with an error. + Failed, + /// The activity was canceled. Note that cancellation is cooperative and a cancel request does + /// not always result in canceled status. + Canceled, + /// The activity was terminated. + Terminated, + /// The activity timed out. + TimedOut, + /// This variant indicates the server used a value not known by this version of the SDK. + Unknown, +} + +impl From for ActivityExecutionStatus { + fn from(value: ProtoActivityExecutionStatus) -> Self { + match value { + ProtoActivityExecutionStatus::Unspecified => Self::Unspecified, + ProtoActivityExecutionStatus::Running => Self::Running, + ProtoActivityExecutionStatus::Completed => Self::Completed, + ProtoActivityExecutionStatus::Failed => Self::Failed, + ProtoActivityExecutionStatus::Canceled => Self::Canceled, + ProtoActivityExecutionStatus::Terminated => Self::Terminated, + ProtoActivityExecutionStatus::TimedOut => Self::TimedOut, + } + } +} + +/// Detailed state of an activity with [`ActivityExecutionStatus::Running`]. +/// See [`ActivityExecutionDescription::run_state`]. +#[non_exhaustive] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)] +pub enum PendingActivityState { + #[default] + /// This variant indicates the server did not specify a state. + Unspecified, + /// Activity is scheduled for execution but not yet running on a worker. + Scheduled, + /// Activity is running on a worker. + Started, + /// Activity has been requested to cancel. + CancelRequested, + /// Activity is paused on the server, and is not running on a worker. + Paused, + /// Activity is currently running on a worker, but paused on the server. + PauseRequested, + /// This variant indicates the server used a value not known by this version of the SDK. + Unknown, +} + +impl From for PendingActivityState { + fn from(value: ProtoPendingActivityState) -> Self { + match value { + ProtoPendingActivityState::Unspecified => Self::Unspecified, + ProtoPendingActivityState::Scheduled => Self::Scheduled, + ProtoPendingActivityState::Started => Self::Started, + ProtoPendingActivityState::CancelRequested => Self::CancelRequested, + ProtoPendingActivityState::Paused => Self::Paused, + ProtoPendingActivityState::PauseRequested => Self::PauseRequested, + } + } +} diff --git a/crates/client/src/activity/activity_handle.rs b/crates/client/src/activity/activity_handle.rs new file mode 100644 index 000000000..25538e5d0 --- /dev/null +++ b/crates/client/src/activity/activity_handle.rs @@ -0,0 +1,191 @@ +use crate::{ + ActivityCancelOptions, ActivityDescribeOptions, ActivityExecutionDescription, + ActivityTerminateOptions, NamespacedClient, + errors::{ActivityInteractionError, ActivityResultError}, + grpc::WorkflowService, +}; +use std::marker::PhantomData; +use temporalio_common::{ + ActivityDefinition, + data_converters::{NoopDecodeHint, SerializationContextData}, + protos::temporal::api::{ + activity::v1::{ActivityExecutionOutcome, activity_execution_outcome}, + workflowservice::v1::{ + DescribeActivityExecutionRequest, PollActivityExecutionRequest, + RequestCancelActivityExecutionRequest, TerminateActivityExecutionRequest, + }, + }, +}; +use tonic::IntoRequest; +use uuid::Uuid; + +/// Handle associated with a standalone activity execution that can be used to wait for the result +/// or to manage execution of the activity. Obtained from +/// [`Client::start_activity`](crate::Client::start_activity) or +/// [`Client::get_activity_handle`](crate::Client::get_activity_handle). +/// +/// If [`run_id`](Self::run_id) is set, the handle always targets that specific execution. +/// If [`run_id`](Self::run_id) is `None`, each method call targets the latest run of the specified +/// [`activity_id`](Self::activity_id) at the time the method is called - this means consecutive +/// method calls may target different executions if an activity was started again with the same ID. +pub struct ActivityHandle +where + ActivityT: ActivityDefinition, +{ + client: ClientT, + activity_id: String, + run_id: Option, + _phantom: PhantomData, +} + +impl ActivityHandle +where + ActivityT: ActivityDefinition, +{ + pub(crate) fn new(client: ClientT, activity_id: String, run_id: Option) -> Self { + Self { + client, + activity_id, + run_id, + _phantom: PhantomData, + } + } + + /// Activity ID this handle is associated with. + pub fn activity_id(&self) -> &str { + &self.activity_id + } + + /// Run ID of the activity execution this handle is associated with. If `None`, each method call + /// targets the latest run of the specified [`activity_id`](Self::activity_id) at the time the + /// method is called - this means consecutive method calls may target different executions if + /// an activity was started again with the same ID. + pub fn run_id(&self) -> Option<&str> { + self.run_id.as_deref() + } +} + +impl ActivityHandle +where + ClientT: WorkflowService + NamespacedClient + Clone, + ActivityT: ActivityDefinition, +{ + /// Wait for the activity to complete and fetch its result. If the activity was not successful + /// (e.g. failed, canceled, timed out), this method returns [`ActivityResultError::ActivityFailed`]. + pub async fn result(&self) -> Result { + let mut client = self.client.clone(); + loop { + let resp = client + .poll_activity_execution( + PollActivityExecutionRequest { + namespace: client.namespace(), + activity_id: self.activity_id.clone(), + run_id: self.run_id.clone().unwrap_or_default(), + } + .into_request(), + ) + .await? + .into_inner(); + + // If resp.outcome.value is None, poll again + let Some(ActivityExecutionOutcome { + value: Some(outcome), + }) = resp.outcome + else { + continue; + }; + + let dc = client.data_converter(); + let ctx = &SerializationContextData::Activity; + + return match outcome { + activity_execution_outcome::Value::Result(payloads) => { + Ok(dc.from_payloads(ctx, payloads.payloads).await?) + } + activity_execution_outcome::Value::Failure(failure) => { + Err(ActivityResultError::ActivityFailed(dc.to_error( + ctx, + failure, + NoopDecodeHint, + )?)) + } + }; + } + } + + /// Describes the current state of the activity execution. + pub async fn describe( + &self, + options: ActivityDescribeOptions, + ) -> Result, ActivityInteractionError> { + let mut client = self.client.clone(); + let resp = client + .describe_activity_execution( + DescribeActivityExecutionRequest { + namespace: client.namespace(), + activity_id: self.activity_id.clone(), + run_id: self.run_id.clone().unwrap_or_default(), + include_input: options.include_input, + include_outcome: options.include_outcome, + include_heartbeat_details: options.include_heartbeat_details, + include_last_failure: options.include_last_failure, + ..Default::default() + } + .into_request(), + ) + .await? + .into_inner(); + + Ok(ActivityExecutionDescription::new( + client.data_converter().clone(), + SerializationContextData::Activity, + resp, + )?) + } + + /// Requests cancellation of the activity. Does not wait for the cancellation to complete. + pub async fn cancel( + &self, + options: ActivityCancelOptions, + ) -> Result<(), ActivityInteractionError> { + let mut client = self.client.clone(); + client + .request_cancel_activity_execution( + RequestCancelActivityExecutionRequest { + namespace: client.namespace(), + activity_id: self.activity_id.clone(), + run_id: self.run_id.clone().unwrap_or_default(), + identity: client.identity(), + request_id: Uuid::new_v4().to_string(), + reason: options.reason, + } + .into_request(), + ) + .await?; + + Ok(()) + } + + /// Terminates activity execution. + pub async fn terminate( + &self, + options: ActivityTerminateOptions, + ) -> Result<(), ActivityInteractionError> { + let mut client = self.client.clone(); + client + .terminate_activity_execution( + TerminateActivityExecutionRequest { + namespace: client.namespace(), + activity_id: self.activity_id.clone(), + run_id: self.run_id.clone().unwrap_or_default(), + identity: client.identity(), + request_id: Uuid::new_v4().to_string(), + reason: options.reason, + } + .into_request(), + ) + .await?; + + Ok(()) + } +} diff --git a/crates/client/src/async_activity_handle.rs b/crates/client/src/async_activity_handle.rs index a4cbb23f6..d7b8b78bf 100644 --- a/crates/client/src/async_activity_handle.rs +++ b/crates/client/src/async_activity_handle.rs @@ -23,18 +23,26 @@ use tonic::IntoRequest; /// Identifies an async activity for completion outside a worker. #[derive(Debug, Clone)] +#[non_exhaustive] pub enum ActivityIdentifier { /// Identify activity by its task token TaskToken(TaskToken), - /// Identify activity by workflow and activity IDs. - ById { + /// Identify workflow activity by workflow and activity IDs. + ByIdWorkflow { /// ID of the workflow that scheduled this activity. workflow_id: String, - /// Run ID of the workflow (optional - if not provided, targets the latest run). - run_id: String, + /// Run ID of the workflow (if not provided, targets the latest run). + run_id: Option, /// ID of the activity to complete. activity_id: String, }, + /// Identify standalone activity by activity ID. + ByIdStandalone { + /// ID of the activity to complete. + activity_id: String, + /// Run ID of the activity (if not provided, targets the latest run). + run_id: Option, + }, } impl ActivityIdentifier { @@ -43,19 +51,56 @@ impl ActivityIdentifier { Self::TaskToken(token) } - /// Create an identifier from workflow and activity IDs. Use an empty run id to target the - /// latest workflow execution. - pub fn by_id( + /// Create an identifier of a workflow activity from workflow and activity IDs. Set run_id to + /// None to target the latest workflow execution. + pub fn by_id_workflow( workflow_id: impl Into, - run_id: impl Into, + run_id: Option>, activity_id: impl Into, ) -> Self { - Self::ById { + Self::ByIdWorkflow { workflow_id: workflow_id.into(), - run_id: run_id.into(), + run_id: run_id.map(Into::into), activity_id: activity_id.into(), } } + + /// Create an identifier of a standalone activity from activity IDs. Set run_id to None to + /// target the latest workflow execution. + pub fn by_id_standalone( + activity_id: impl Into, + run_id: Option>, + ) -> Self { + Self::ByIdStandalone { + activity_id: activity_id.into(), + run_id: run_id.map(Into::into), + } + } + + /// Internal use only. Returns workflow ID, run ID and activity ID as separate strings. + fn id_tuple(&self) -> (String, String, String) { + debug_assert!(!matches!(self, ActivityIdentifier::TaskToken(..))); + match self { + ActivityIdentifier::ByIdWorkflow { + workflow_id, + run_id, + activity_id, + } => ( + workflow_id.clone(), + run_id.clone().unwrap_or_default(), + activity_id.clone(), + ), + ActivityIdentifier::ByIdStandalone { + activity_id, + run_id, + } => ( + "".into(), + run_id.clone().unwrap_or_default(), + activity_id.clone(), + ), + _ => unreachable!("Unknown activity identifier variant"), + } + } } /// Handle for completing activities asynchronously (outside the worker). @@ -98,43 +143,37 @@ impl AsyncActivityHandle { }), None => None, }; - match &self.identifier { - ActivityIdentifier::TaskToken(token) => { - WorkflowService::respond_activity_task_completed( - &mut self.client.clone(), - RespondActivityTaskCompletedRequest { - task_token: token.0.clone(), - result, - identity: self.client.identity(), - namespace: self.client.namespace(), - ..Default::default() - } - .into_request(), - ) - .await - .map_err(AsyncActivityError::from_status)?; - } - ActivityIdentifier::ById { - workflow_id, - run_id, - activity_id, - } => { - WorkflowService::respond_activity_task_completed_by_id( - &mut self.client.clone(), - RespondActivityTaskCompletedByIdRequest { - namespace: self.client.namespace(), - workflow_id: workflow_id.clone(), - run_id: run_id.clone(), - activity_id: activity_id.clone(), - result, - identity: self.client.identity(), - resource_id: Default::default(), - } - .into_request(), - ) - .await - .map_err(AsyncActivityError::from_status)?; - } + if let ActivityIdentifier::TaskToken(token) = &self.identifier { + WorkflowService::respond_activity_task_completed( + &mut self.client.clone(), + RespondActivityTaskCompletedRequest { + task_token: token.0.clone(), + result, + identity: self.client.identity(), + namespace: self.client.namespace(), + ..Default::default() + } + .into_request(), + ) + .await + .map_err(AsyncActivityError::from_status)?; + } else { + let (workflow_id, run_id, activity_id) = self.identifier.id_tuple(); + WorkflowService::respond_activity_task_completed_by_id( + &mut self.client.clone(), + RespondActivityTaskCompletedByIdRequest { + namespace: self.client.namespace(), + workflow_id, + run_id, + activity_id, + result, + identity: self.client.identity(), + resource_id: Default::default(), + } + .into_request(), + ) + .await + .map_err(AsyncActivityError::from_status)?; } Ok(()) } @@ -170,45 +209,39 @@ impl AsyncActivityHandle { }), None => None, }; - match &self.identifier { - ActivityIdentifier::TaskToken(token) => { - WorkflowService::respond_activity_task_failed( - &mut self.client.clone(), - RespondActivityTaskFailedRequest { - task_token: token.0.clone(), - failure: Some(failure), - identity: self.client.identity(), - namespace: self.client.namespace(), - last_heartbeat_details, - ..Default::default() - } - .into_request(), - ) - .await - .map_err(AsyncActivityError::from_status)?; - } - ActivityIdentifier::ById { - workflow_id, - run_id, - activity_id, - } => { - WorkflowService::respond_activity_task_failed_by_id( - &mut self.client.clone(), - RespondActivityTaskFailedByIdRequest { - namespace: self.client.namespace(), - workflow_id: workflow_id.clone(), - run_id: run_id.clone(), - activity_id: activity_id.clone(), - failure: Some(failure), - identity: self.client.identity(), - last_heartbeat_details, - resource_id: Default::default(), - } - .into_request(), - ) - .await - .map_err(AsyncActivityError::from_status)?; - } + if let ActivityIdentifier::TaskToken(token) = &self.identifier { + WorkflowService::respond_activity_task_failed( + &mut self.client.clone(), + RespondActivityTaskFailedRequest { + task_token: token.0.clone(), + failure: Some(failure), + identity: self.client.identity(), + namespace: self.client.namespace(), + last_heartbeat_details, + ..Default::default() + } + .into_request(), + ) + .await + .map_err(AsyncActivityError::from_status)?; + } else { + let (workflow_id, run_id, activity_id) = self.identifier.id_tuple(); + WorkflowService::respond_activity_task_failed_by_id( + &mut self.client.clone(), + RespondActivityTaskFailedByIdRequest { + namespace: self.client.namespace(), + workflow_id: workflow_id.clone(), + run_id: run_id.clone(), + activity_id: activity_id.clone(), + failure: Some(failure), + identity: self.client.identity(), + last_heartbeat_details, + resource_id: Default::default(), + } + .into_request(), + ) + .await + .map_err(AsyncActivityError::from_status)?; } Ok(()) } @@ -228,43 +261,37 @@ impl AsyncActivityHandle { }), None => None, }; - match &self.identifier { - ActivityIdentifier::TaskToken(token) => { - WorkflowService::respond_activity_task_canceled( - &mut self.client.clone(), - RespondActivityTaskCanceledRequest { - task_token: token.0.clone(), - details, - identity: self.client.identity(), - namespace: self.client.namespace(), - ..Default::default() - } - .into_request(), - ) - .await - .map_err(AsyncActivityError::from_status)?; - } - ActivityIdentifier::ById { - workflow_id, - run_id, - activity_id, - } => { - WorkflowService::respond_activity_task_canceled_by_id( - &mut self.client.clone(), - RespondActivityTaskCanceledByIdRequest { - namespace: self.client.namespace(), - workflow_id: workflow_id.clone(), - run_id: run_id.clone(), - activity_id: activity_id.clone(), - details, - identity: self.client.identity(), - ..Default::default() - } - .into_request(), - ) - .await - .map_err(AsyncActivityError::from_status)?; - } + if let ActivityIdentifier::TaskToken(token) = &self.identifier { + WorkflowService::respond_activity_task_canceled( + &mut self.client.clone(), + RespondActivityTaskCanceledRequest { + task_token: token.0.clone(), + details, + identity: self.client.identity(), + namespace: self.client.namespace(), + ..Default::default() + } + .into_request(), + ) + .await + .map_err(AsyncActivityError::from_status)?; + } else { + let (workflow_id, run_id, activity_id) = self.identifier.id_tuple(); + WorkflowService::respond_activity_task_canceled_by_id( + &mut self.client.clone(), + RespondActivityTaskCanceledByIdRequest { + namespace: self.client.namespace(), + workflow_id: workflow_id.clone(), + run_id: run_id.clone(), + activity_id: activity_id.clone(), + details, + identity: self.client.identity(), + ..Default::default() + } + .into_request(), + ) + .await + .map_err(AsyncActivityError::from_status)?; } Ok(()) } @@ -290,47 +317,41 @@ impl AsyncActivityHandle { }), None => None, }; - match &self.identifier { - ActivityIdentifier::TaskToken(token) => { - let resp = WorkflowService::record_activity_task_heartbeat( - &mut self.client.clone(), - RecordActivityTaskHeartbeatRequest { - task_token: token.0.clone(), - details, - identity: self.client.identity(), - namespace: self.client.namespace(), - resource_id: Default::default(), - } - .into_request(), - ) - .await - .map_err(AsyncActivityError::from_status)? - .into_inner(); - Ok(ActivityHeartbeatResponse::from(resp)) - } - ActivityIdentifier::ById { - workflow_id, - run_id, - activity_id, - } => { - let resp = WorkflowService::record_activity_task_heartbeat_by_id( - &mut self.client.clone(), - RecordActivityTaskHeartbeatByIdRequest { - namespace: self.client.namespace(), - workflow_id: workflow_id.clone(), - run_id: run_id.clone(), - activity_id: activity_id.clone(), - details, - identity: self.client.identity(), - resource_id: Default::default(), - } - .into_request(), - ) - .await - .map_err(AsyncActivityError::from_status)? - .into_inner(); - Ok(ActivityHeartbeatResponse::from(resp)) - } + if let ActivityIdentifier::TaskToken(token) = &self.identifier { + let resp = WorkflowService::record_activity_task_heartbeat( + &mut self.client.clone(), + RecordActivityTaskHeartbeatRequest { + task_token: token.0.clone(), + details, + identity: self.client.identity(), + namespace: self.client.namespace(), + resource_id: Default::default(), + } + .into_request(), + ) + .await + .map_err(AsyncActivityError::from_status)? + .into_inner(); + Ok(ActivityHeartbeatResponse::from(resp)) + } else { + let (workflow_id, run_id, activity_id) = self.identifier.id_tuple(); + let resp = WorkflowService::record_activity_task_heartbeat_by_id( + &mut self.client.clone(), + RecordActivityTaskHeartbeatByIdRequest { + namespace: self.client.namespace(), + workflow_id: workflow_id.clone(), + run_id: run_id.clone(), + activity_id: activity_id.clone(), + details, + identity: self.client.identity(), + resource_id: Default::default(), + } + .into_request(), + ) + .await + .map_err(AsyncActivityError::from_status)? + .into_inner(); + Ok(ActivityHeartbeatResponse::from(resp)) } } } diff --git a/crates/client/src/errors.rs b/crates/client/src/errors.rs index d1498bee1..bdcd2b936 100644 --- a/crates/client/src/errors.rs +++ b/crates/client/src/errors.rs @@ -3,8 +3,14 @@ use crate::{WorkflowExecutionStatus, workflow_handle::WorkflowResultDetails}; use http::uri::InvalidUri; use temporalio_common::{ - data_converters::PayloadConversionError, error::IncomingError, - protos::temporal::api::failure::v1::Failure, + data_converters::PayloadConversionError, + error::IncomingError, + protos::{ + temporal::api::{ + errordetails::v1::ActivityExecutionAlreadyStartedFailure, failure::v1::Failure, + }, + utilities::decode_status_detail, + }, }; use tonic::Code; @@ -311,3 +317,172 @@ impl AsyncActivityError { #[derive(Debug, thiserror::Error)] #[non_exhaustive] pub enum ClientNewError {} + +/// Errors returned by methods on [crate::ActivityHandle] that don't need more specific error types. +#[derive(Debug, thiserror::Error)] +#[non_exhaustive] +pub enum ActivityInteractionError { + /// The activity was not found. + #[error("Activity not found")] + NotFound(#[source] tonic::Status), + + /// Error deserializing output. + #[error("Payload conversion error: {0}")] + PayloadConversion(#[from] PayloadConversionError), + + /// An uncategorized RPC error from the server. + #[error("Server error: {0}")] + Rpc(#[source] tonic::Status), + + /// Other errors. + #[error(transparent)] + Other(#[from] Box), +} + +impl From for ActivityInteractionError { + fn from(status: tonic::Status) -> Self { + if status.code() == Code::NotFound { + Self::NotFound(status) + } else { + Self::Rpc(status) + } + } +} + +/// Errors that can occur when starting a standalone activity. +#[allow(clippy::large_enum_variant)] +#[derive(Debug, thiserror::Error)] +#[non_exhaustive] +pub enum StartActivityError { + /// There's a conflicting activity execution with the same ID according to chosen ID reuse + /// policy and ID conflict policy. + #[error("Activity already started with run_id={run_id}")] + AlreadyStarted { + /// Run ID of the existing execution with the same activity ID. + run_id: String, + /// Raw error from the server. + #[source] + source: tonic::Status, + }, + + /// Error serializing input. + #[error("Payload conversion error: {0}")] + PayloadConversion(#[from] PayloadConversionError), + + /// An uncategorized RPC error from the server. + #[error("Server error: {0}")] + Rpc(#[source] tonic::Status), + + /// Other errors. + #[error(transparent)] + Other(#[from] Box), +} + +impl From for StartActivityError { + fn from(status: tonic::Status) -> Self { + if status.code() == tonic::Code::AlreadyExists + && let Some(details) = + decode_status_detail::(status.details()) + { + StartActivityError::AlreadyStarted { + run_id: details.run_id, + source: status, + } + } else { + StartActivityError::Rpc(status) + } + } +} + +/// Errors returned by [`crate::ActivityHandle::result`]. +#[allow(clippy::large_enum_variant)] +#[derive(Debug, thiserror::Error)] +#[non_exhaustive] +pub enum ActivityResultError { + /// Activity execution did not complete successfully. + #[error("Activity failed: {0}")] + ActivityFailed(#[source] IncomingError), + + /// The activity was not found. + #[error("Activity not found")] + NotFound(#[source] tonic::Status), + + /// Error deserializing output. + #[error("Payload conversion error: {0}")] + PayloadConversion(#[from] PayloadConversionError), + + /// An uncategorized RPC error from the server. + #[error("Server error: {0}")] + Rpc(#[source] tonic::Status), + + /// Other errors. + #[error(transparent)] + Other(#[from] Box), +} + +impl From for ActivityResultError { + fn from(status: tonic::Status) -> Self { + if status.code() == Code::NotFound { + Self::NotFound(status) + } else { + Self::Rpc(status) + } + } +} + +/// Errors that can occur when executing a standalone activity. +#[allow(clippy::large_enum_variant)] +#[derive(Debug, thiserror::Error)] +#[non_exhaustive] +pub enum ExecuteActivityError { + /// Activity did not complete successfully. + #[error("Activity failed: {0}")] + ActivityFailed(#[source] IncomingError), + + /// There's a conflicting activity execution with the same ID according to chosen ID reuse + /// policy and ID conflict policy. + #[error("Activity already started with run_id={run_id}")] + AlreadyStarted { + /// Run ID of the existing execution with the same activity ID. + run_id: String, + /// Raw error from the server. + #[source] + source: tonic::Status, + }, + + /// Error serializing input or output. + #[error("Payload conversion error: {0}")] + PayloadConversion(#[from] PayloadConversionError), + + /// An uncategorized RPC error from the server. + #[error("Server error: {0}")] + Rpc(#[source] tonic::Status), + + /// Other errors. + #[error(transparent)] + Other(#[from] Box), +} + +impl From for ExecuteActivityError { + fn from(value: StartActivityError) -> Self { + match value { + StartActivityError::AlreadyStarted { run_id, source } => { + Self::AlreadyStarted { run_id, source } + } + StartActivityError::PayloadConversion(e) => Self::PayloadConversion(e), + StartActivityError::Rpc(e) => Self::Rpc(e), + StartActivityError::Other(e) => Self::Other(e), + } + } +} + +impl From for ExecuteActivityError { + fn from(value: ActivityResultError) -> Self { + match value { + ActivityResultError::ActivityFailed(e) => Self::ActivityFailed(e), + ActivityResultError::PayloadConversion(e) => Self::PayloadConversion(e), + ActivityResultError::NotFound(e) | ActivityResultError::Rpc(e) => Self::Rpc(e), + ActivityResultError::Other(e) => Self::Other(e), + } + } +} diff --git a/crates/client/src/lib.rs b/crates/client/src/lib.rs index abc2ea62f..56997d2d2 100644 --- a/crates/client/src/lib.rs +++ b/crates/client/src/lib.rs @@ -7,6 +7,7 @@ #[macro_use] extern crate tracing; +mod activity; mod async_activity_handle; pub mod callback_based; mod dns; @@ -27,6 +28,7 @@ mod retry; pub mod schedules; #[cfg(test)] mod test_helpers; +pub(crate) mod utils; pub mod worker; mod workflow_handle; mod workflow_status; @@ -36,16 +38,16 @@ pub use crate::{ request_extensions::PayloadErrorLimits, retry::{CallType, RETRYABLE_ERROR_CODES}, }; +pub use activity::*; pub use async_activity_handle::{ ActivityHeartbeatResponse, ActivityIdentifier, AsyncActivityHandle, }; -#[doc(hidden)] -pub use retry::jittered; - pub use metrics::{LONG_REQUEST_LATENCY_HISTOGRAM_NAME, REQUEST_LATENCY_HISTOGRAM_NAME}; pub use options_structs::*; pub use replaceable::SharedReplaceableClient; pub use retry::RetryOptions; +#[doc(hidden)] +pub use retry::jittered; pub use temporalio_common::{Memo, RetryPolicy}; /// Potentially dangerous TLS related functionality. pub mod danger { @@ -69,6 +71,7 @@ use crate::{ }, metrics::{ChannelOrGrpcOverride, GrpcMetricSvc, MetricsContext}, request_extensions::RequestExt, + utils::try_into_or_box_err, worker::ClientWorkerSet, }; use errors::*; @@ -85,7 +88,7 @@ use std::{ time::{Duration, SystemTime}, }; use temporalio_common::{ - HasWorkflowDefinition, + ActivityDefinition, HasWorkflowDefinition, UntypedActivity, data_converters::{ DataConverter, GenericPayloadConverter, PayloadConverter, SerializationContext, SerializationContextData, @@ -97,8 +100,11 @@ use temporalio_common::{ proto_ts_to_system_time, temporal::api::{ cloud::cloudservice::v1::cloud_service_client::CloudServiceClient, - common::v1::WorkflowType, - enums::v1::TaskQueueKind, + common::v1::{ActivityType, WorkflowType}, + enums::v1::{ + ActivityIdConflictPolicy as ProtoActivityIdConflictPolicy, + ActivityIdReusePolicy as ProtoActivityIdReusePolicy, TaskQueueKind, + }, errordetails::v1::WorkflowExecutionAlreadyStartedFailure, operatorservice::v1::operator_service_client::OperatorServiceClient, sdk::v1::UserMetadata, @@ -830,12 +836,108 @@ impl Client { /// Get a handle to complete an activity asynchronously. /// /// An activity returning `ActivityError::WillCompleteAsync` can be completed with this handle. + /// + /// To get a handle to a standalone activity that can be used to wait for result and manage + /// the execution, see [`get_activity_handle`](Self::get_activity_handle). pub fn get_async_activity_handle( &self, identifier: ActivityIdentifier, ) -> AsyncActivityHandle { WorkflowClientTrait::get_async_activity_handle(self, identifier) } + + /// Start a standalone activity. + /// + /// Returns [`ActivityHandle`] that can be used to wait for result or to perform other + /// operations on the activity. + pub async fn start_activity( + &self, + activity: A, + input: A::Input, + options: ActivityStartOptions, + ) -> Result, StartActivityError> + where + A: ActivityDefinition, + { + WorkflowClientTrait::start_activity(self, activity, input, options).await + } + + /// Start a standalone activity and wait for its result. + /// + /// Equivalent to `start_activity(...).await?.result()`. + pub async fn execute_activity( + &self, + activity: A, + input: A::Input, + options: ActivityStartOptions, + ) -> Result + where + A: ActivityDefinition, + { + WorkflowClientTrait::execute_activity(self, activity, input, options).await + } + + /// Get a handle to an existing standalone activity execution. If `run_id` is not specified, + /// the handle always targets the latest execution with matching ID. + /// + /// Note that the validity of the handle is not checked until a method is called on it. + /// If invalid ID or run ID is used, the method will return `NotFound` error. + /// + /// To get an untyped handle, use [`get_untyped_activity_handle`](Self::get_untyped_activity_handle). + /// + /// To get a handle that can be used to complete an activity asynchronously, + /// see [`get_async_activity_handle`](Self::get_async_activity_handle). + pub fn get_activity_handle( + &self, + activity: A, + id: impl Into, + run_id: Option, + ) -> ActivityHandle + where + Self: Sized, + A: ActivityDefinition, + { + WorkflowClientTrait::get_activity_handle(self, activity, id, run_id) + } + + /// Get an untyped handle to an existing standalone activity execution. If `run_id` is not + /// specified, the handle always targets the latest execution with matching ID. + /// + /// Note that the validity of the handle is not checked until a method is called on it. + /// If invalid ID or run ID is used, the method will return `NotFound` error. + /// + /// To get a typed handle, use [`get_activity_handle`](Self::get_activity_handle). + /// + /// To get a handle that can be used to complete an activity asynchronously, + /// see [`get_async_activity_handle`](Self::get_async_activity_handle). + pub fn get_untyped_activity_handle( + &self, + id: impl Into, + run_id: Option, + ) -> ActivityHandle + where + Self: Sized, + { + WorkflowClientTrait::get_untyped_activity_handle(self, id, run_id) + } + + /// List activities matching a query. Returns a stream that lazily paginates through results. + pub fn list_activities( + &self, + query: impl Into, + options: ActivityListOptions, + ) -> ListActivitiesStream { + WorkflowClientTrait::list_activities(self, query, options) + } + + /// Count activities matching a query. + pub async fn count_activities( + &self, + query: impl Into, + options: ActivityCountOptions, + ) -> Result { + WorkflowClientTrait::count_activities(self, query, options).await + } } impl NamespacedClient for Client { @@ -914,6 +1016,62 @@ pub(crate) trait WorkflowClientTrait: NamespacedClient { ) -> AsyncActivityHandle where Self: Sized; + + /// Start a standalone activity. + fn start_activity( + &self, + activity: A, + input: A::Input, + options: ActivityStartOptions, + ) -> impl Future, StartActivityError>> + where + Self: Sized, + A: ActivityDefinition; + + /// Start a standalone activity and wait for its result. + fn execute_activity( + &self, + activity: A, + input: A::Input, + options: ActivityStartOptions, + ) -> impl Future> + where + Self: Sized, + A: ActivityDefinition; + + /// Get a handle to a previously started standalone activity. + fn get_activity_handle( + &self, + activity: A, + id: impl Into, + run_id: Option, + ) -> ActivityHandle + where + Self: Sized, + A: ActivityDefinition; + + /// Get an untyped handle to a previously started standalone activity. + fn get_untyped_activity_handle( + &self, + id: impl Into, + run_id: Option, + ) -> ActivityHandle + where + Self: Sized; + + /// List activities matching a query. Returns a stream that lazily paginates through results. + fn list_activities( + &self, + query: impl Into, + _options: ActivityListOptions, + ) -> ListActivitiesStream; + + /// Count activities matching a query. + fn count_activities( + &self, + query: impl Into, + _options: ActivityCountOptions, + ) -> impl Future>; } /// A client that is bound to a namespace @@ -1444,6 +1602,201 @@ where { AsyncActivityHandle::new(self.clone(), identifier) } + + async fn start_activity( + &self, + activity: A, + input: A::Input, + options: ActivityStartOptions, + ) -> Result, StartActivityError> + where + Self: Sized, + A: ActivityDefinition, + { + let mut client = self.clone(); + let dc = client.data_converter(); + let sc = &SerializationContextData::Activity; + + let close_timeouts = options.close_timeouts.into_values(); + let user_metadata = { + let summary = match &options.static_summary { + Some(summary) => Some(dc.to_payload(sc, summary).await?), + None => None, + }; + let details = match &options.static_details { + Some(details) => Some(dc.to_payload(sc, details).await?), + None => None, + }; + (summary.is_some() || details.is_some()).then_some(UserMetadata { summary, details }) + }; + + let resp = client + .start_activity_execution( + StartActivityExecutionRequest { + namespace: client.namespace(), + identity: client.identity(), + request_id: Uuid::new_v4().to_string(), + activity_id: options.id.clone(), + activity_type: Some(ActivityType { + name: activity.name().to_string(), + }), + task_queue: Some(TaskQueue { + name: options.task_queue, + kind: TaskQueueKind::Normal.into(), + normal_name: "".to_string(), + }), + schedule_to_close_timeout: try_into_or_box_err( + close_timeouts.schedule_to_close, + StartActivityError::Other, + )?, + schedule_to_start_timeout: try_into_or_box_err( + options.schedule_to_start_timeout, + StartActivityError::Other, + )?, + start_to_close_timeout: try_into_or_box_err( + close_timeouts.start_to_close, + StartActivityError::Other, + )?, + heartbeat_timeout: try_into_or_box_err( + options.heartbeat_timeout, + StartActivityError::Other, + )?, + retry_policy: options.retry_policy.map(Into::into), + input: dc.to_payloads(sc, &input).await?.into_payloads(), + id_reuse_policy: ProtoActivityIdReusePolicy::from(options.id_reuse_policy) + .into(), + id_conflict_policy: ProtoActivityIdConflictPolicy::from( + options.id_conflict_policy, + ) + .into(), + search_attributes: options.search_attributes.map(SearchAttributes::into_proto), + header: options.header, + user_metadata, + priority: Some(options.priority.into()), + start_delay: try_into_or_box_err( + options.start_delay, + StartActivityError::Other, + )?, + ..Default::default() + } + .into_request(), + ) + .await? + .into_inner(); + + Ok(ActivityHandle::new( + client, + options.id, + (!resp.run_id.is_empty()).then_some(resp.run_id), + )) + } + + async fn execute_activity( + &self, + activity: A, + input: A::Input, + options: ActivityStartOptions, + ) -> Result + where + Self: Sized, + A: ActivityDefinition, + { + Ok(self + .start_activity(activity, input, options) + .await? + .result() + .await?) + } + + fn get_activity_handle( + &self, + _activity: A, + id: impl Into, + run_id: Option, + ) -> ActivityHandle + where + Self: Sized, + A: ActivityDefinition, + { + ActivityHandle::new(self.clone(), id.into(), run_id) + } + + fn get_untyped_activity_handle( + &self, + id: impl Into, + run_id: Option, + ) -> ActivityHandle + where + Self: Sized, + { + ActivityHandle::new(self.clone(), id.into(), run_id) + } + + fn list_activities( + &self, + query: impl Into, + _options: ActivityListOptions, + ) -> ListActivitiesStream { + let client = self.clone(); + let namespace = client.namespace(); + let query = query.into(); + + ListActivitiesStream::new(stream::unfold( + Some(vec![]), // empty token for initial query, None if done + move |next_page_token| { + let mut client = client.clone(); + let namespace = namespace.clone(); + let query = query.clone(); + + async move { + // making it more visible that we're terminating stream here + #[allow(clippy::question_mark)] + let Some(token): Option> = next_page_token else { + return None; + }; + + match WorkflowService::list_activity_executions( + &mut client, + ListActivityExecutionsRequest { + namespace, + page_size: 0, // Use server default + next_page_token: token.clone(), + query, + } + .into_request(), + ) + .await + .map(|r| r.into_inner()) + { + Ok(resp) => Some(( + Ok(resp.executions), + (!resp.next_page_token.is_empty()).then_some(resp.next_page_token), + )), + Err(e) => Some((Err(e.into()), Some(token))), + } + } + }, + )) + } + + async fn count_activities( + &self, + query: impl Into, + _options: ActivityCountOptions, + ) -> Result { + let mut client = self.clone(); + let resp = client + .count_activity_executions( + CountActivityExecutionsRequest { + namespace: client.namespace(), + query: query.into(), + } + .into_request(), + ) + .await? + .into_inner(); + Ok(ActivityExecutionCount::from_response(resp)) + } } macro_rules! dbg_panic { diff --git a/crates/client/src/options_structs.rs b/crates/client/src/options_structs.rs index aaf7743ed..20b120504 100644 --- a/crates/client/src/options_structs.rs +++ b/crates/client/src/options_structs.rs @@ -2,7 +2,7 @@ use crate::{HttpConnectProxyOptions, RetryOptions, VERSION, callback_based}; use http::Uri; use std::{collections::HashMap, sync::Arc, time::Duration}; use temporalio_common::{ - RetryPolicy, + ActivityCloseTimeoutOptions, RetryPolicy, data_converters::DataConverter, protos::temporal::api::{ common::{ @@ -10,7 +10,9 @@ use temporalio_common::{ v1::{Header, Payloads}, }, enums::v1::{ - ArchivalState, HistoryEventFilterType, QueryRejectCondition, WorkflowIdConflictPolicy, + ActivityIdConflictPolicy as ProtoActivityIdConflictPolicy, + ActivityIdReusePolicy as ProtoActivityIdReusePolicy, ArchivalState, + HistoryEventFilterType, QueryRejectCondition, WorkflowIdConflictPolicy, WorkflowIdReusePolicy, }, replication::v1::ClusterReplicationConfig, @@ -568,3 +570,191 @@ pub struct WorkflowListOptions { #[derive(Debug, Clone, Default, bon::Builder)] #[non_exhaustive] pub struct WorkflowCountOptions {} + +/// Options for starting a standalone activity. +#[derive(Clone, Debug, bon::Builder)] +#[builder(start_fn = new, on(String, into))] +pub struct ActivityStartOptions { + /// Task queue to run this activity on. + #[builder(start_fn)] + pub task_queue: String, + /// Activity ID of the started activity. It's recommended to use a meaningful business ID. + #[builder(start_fn)] + pub id: String, + /// Combined schedule-to-close and start-to-close timeout options. At least one of them is + /// required. + #[builder(setters(name = combined_close_timeouts, vis = ""))] + pub close_timeouts: ActivityCloseTimeoutOptions, + /// If set, specifies maximum time the activity can wait in the task queue before being picked + /// up by a worker. This timeout is non-retryable. + pub schedule_to_start_timeout: Option, + /// If set, specifies maximum time between successful heartbeats. + pub heartbeat_timeout: Option, + /// Controls how Activity is retried. If not set, the server will assign default retry policy. + #[builder(into)] + pub retry_policy: Option, + /// Priority to use when starting this activity. + #[builder(default)] + pub priority: Priority, + /// Specifies behavior if there's a *closed* activity with the same ID. + #[builder(default)] + pub id_reuse_policy: ActivityIdReusePolicy, + /// Specifies behavior if there's a *running* activity with the same ID. Note that there can + /// only be one running activity for each Activity ID. + #[builder(default)] + pub id_conflict_policy: ActivityIdConflictPolicy, + /// Search attributes for the activity. + pub search_attributes: Option, + /// Headers to include with the start request. + pub header: Option
, + /// Single-line static summary for the activity, shown in the Temporal UI. + pub static_summary: Option, + /// Multi-line static details for the activity, shown in the Temporal UI. + pub static_details: Option, + /// Time to wait before dispatching the first activity task. + /// This delay is not applied to retry attempts. + pub start_delay: Option, +} + +impl ActivityStartOptionsBuilder +where + S::CloseTimeouts: activity_start_options_builder::IsUnset, +{ + /// _**Required if**_ [`start_to_close_timeout`](Self::start_to_close_timeout) is not set. + /// + /// If set, specifies total time the activity is allowed to run, including retries. + /// + /// Calling this method prevents calling + /// [`schedule_to_close_timeout`](Self::schedule_to_close_timeout). To set both timeouts, use + /// [`close_timeouts`](Self::close_timeouts). + pub fn schedule_to_close_timeout( + self, + value: Duration, + ) -> ActivityStartOptionsBuilder> { + self.combined_close_timeouts(ActivityCloseTimeoutOptions::ScheduleToClose(value)) + } + + /// _**Required if**_ [`schedule_to_close_timeout`](Self::schedule_to_close_timeout) is not set. + /// + /// If set, specifies maximum time of a single Activity execution attempt. + /// + /// Note that the Temporal Server doesn't detect Worker process failures directly. It relies on + /// this timeout to detect that an Activity that didn't complete on time. So this timeout should + /// be as short as the longest possible execution of the Activity body. Potentially long running + /// Activities must specify [`heartbeat_timeout`](Self::heartbeat_timeout) and heartbeat from + /// the activity periodically for timely failure detection. + /// + /// Calling this method prevents calling + /// [`schedule_to_close_timeout`](Self::schedule_to_close_timeout). To set both timeouts, use + /// [`close_timeouts`](Self::close_timeouts). + pub fn start_to_close_timeout( + self, + value: Duration, + ) -> ActivityStartOptionsBuilder> { + self.combined_close_timeouts(ActivityCloseTimeoutOptions::StartToClose(value)) + } + + /// Sets both [`schedule_to_close_timeout`](Self::schedule_to_close_timeout) and + /// [`start_to_close_timeout`](Self::start_to_close_timeout). See the respective methods for + /// meaning of the options. + pub fn close_timeouts( + self, + schedule_to_close: Duration, + start_to_close: Duration, + ) -> ActivityStartOptionsBuilder> { + self.combined_close_timeouts(ActivityCloseTimeoutOptions::Both { + schedule_to_close, + start_to_close, + }) + } +} + +/// Specifies behavior when starting a standalone activity if there's a *closed* activity with +/// the same ID. See [`ActivityStartOptions::id_reuse_policy`]. +#[non_exhaustive] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)] +pub enum ActivityIdReusePolicy { + #[default] + /// Always allow starting an activity using the same activity ID. This is the default. + AllowDuplicate, + /// Allow starting an activity using the same ID only when the last execution did not complete + /// successfully. + AllowDuplicateFailedOnly, + /// Do not permit re-use of the ID for this activity. + RejectDuplicate, +} + +impl From for ProtoActivityIdReusePolicy { + fn from(value: ActivityIdReusePolicy) -> Self { + match value { + ActivityIdReusePolicy::AllowDuplicate => Self::AllowDuplicate, + ActivityIdReusePolicy::AllowDuplicateFailedOnly => Self::AllowDuplicateFailedOnly, + ActivityIdReusePolicy::RejectDuplicate => Self::RejectDuplicate, + } + } +} + +/// Specifies behavior when starting a standalone activity if there's a *running* activity with +/// the same ID. See [`ActivityStartOptions::id_conflict_policy`]. +#[non_exhaustive] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)] +pub enum ActivityIdConflictPolicy { + #[default] + /// Don't start a new activity; instead return + /// [`StartActivityError::AlreadyStarted`](crate::errors::StartActivityError::AlreadyStarted). + Fail, + /// Don't start a new activity; instead return a handle for the running activity. + UseExisting, +} + +impl From for ProtoActivityIdConflictPolicy { + fn from(value: ActivityIdConflictPolicy) -> Self { + match value { + ActivityIdConflictPolicy::Fail => Self::Fail, + ActivityIdConflictPolicy::UseExisting => Self::UseExisting, + } + } +} + +/// Options for listing activities. +#[derive(Debug, Clone, Default, bon::Builder)] +#[non_exhaustive] +pub struct ActivityListOptions {} + +/// Options for counting activities. +#[derive(Debug, Clone, Default, bon::Builder)] +#[non_exhaustive] +pub struct ActivityCountOptions {} + +/// Controls which optional fields will be requested in +/// [`ActivityHandle::describe`](crate::ActivityHandle::describe) operation. The fields will be +/// present in returned [`ActivityExecutionDescription`](crate::ActivityExecutionDescription), +/// subject to data availability and server support. +/// +/// Note that these fields contain payloads that can be arbitrarily large. It's recommended not to +/// include them unless they're needed. +#[derive(Debug, Clone, Default)] +pub struct ActivityDescribeOptions { + /// If `true` and the activity received input, the input will be included. + pub include_input: bool, + /// If `true` and the activity is closed, the activity outcome will be included. + pub include_outcome: bool, + /// If `true` and the activity sent heartbeat details, the heartbeat details will be included. + pub include_heartbeat_details: bool, + /// If `true` and the activity has a failed attempt, the last failure will be included. + pub include_last_failure: bool, +} + +/// Options for [`ActivityHandle::cancel`](crate::ActivityHandle::cancel). +#[derive(Debug, Clone, Default)] +pub struct ActivityCancelOptions { + /// Reason for cancellation. Can be empty. + pub reason: String, +} + +/// Options for [`ActivityHandle::terminate`](crate::ActivityHandle::terminate). +#[derive(Debug, Clone, Default)] +pub struct ActivityTerminateOptions { + /// Reason for termination. Can be empty. + pub reason: String, +} diff --git a/crates/client/src/utils.rs b/crates/client/src/utils.rs new file mode 100644 index 000000000..9a6dc909d --- /dev/null +++ b/crates/client/src/utils.rs @@ -0,0 +1,15 @@ +use std::error::Error; + +pub(crate) fn try_into_or_box_err( + val: Option, + map_err: MapErr, +) -> Result, E> +where + A: TryInto, + >::Error: Error + Send + Sync + 'static, + MapErr: FnOnce(Box) -> E, +{ + val.map(TryInto::try_into) + .transpose() + .map_err(|e| map_err(Box::from(e))) +} diff --git a/crates/common-wasm/src/data_converters.rs b/crates/common-wasm/src/data_converters.rs index 023a16a3c..672fd77ae 100644 --- a/crates/common-wasm/src/data_converters.rs +++ b/crates/common-wasm/src/data_converters.rs @@ -5,7 +5,8 @@ mod failure_converter; pub use failure_converter::{ ActivityExecutionDecodeHint, ChildWorkflowExecutionDecodeHint, ChildWorkflowStartDecodeHint, - DefaultFailureConverter, FailureConverter, FailureDecodeHint, WorkflowSignalDecodeHint, + DefaultFailureConverter, FailureConverter, FailureDecodeHint, NoopDecodeHint, + WorkflowSignalDecodeHint, }; use crate::protos::temporal::api::common::v1::Payload; diff --git a/crates/common-wasm/src/data_converters/failure_converter.rs b/crates/common-wasm/src/data_converters/failure_converter.rs index 973d5aa9f..344b66905 100644 --- a/crates/common-wasm/src/data_converters/failure_converter.rs +++ b/crates/common-wasm/src/data_converters/failure_converter.rs @@ -59,6 +59,18 @@ pub trait FailureDecodeHint { fn adapt(self, normalized: IncomingError) -> Self::Output; } +/// No-op decode hint; returns the error unchanged. +#[derive(Debug, Clone, Copy)] +pub struct NoopDecodeHint; + +impl FailureDecodeHint for NoopDecodeHint { + type Output = IncomingError; + + fn adapt(self, normalized: IncomingError) -> Self::Output { + normalized + } +} + /// Decode hint for activity execution results. #[derive(Debug, Clone, Copy)] pub struct ActivityExecutionDecodeHint { diff --git a/crates/common-wasm/src/lib.rs b/crates/common-wasm/src/lib.rs index fb3d7b5b7..5ffc1238c 100644 --- a/crates/common-wasm/src/lib.rs +++ b/crates/common-wasm/src/lib.rs @@ -7,6 +7,8 @@ #[macro_use] extern crate tracing; +use std::time::Duration; + mod activity_definition; pub mod data_converters; pub mod error; @@ -48,3 +50,55 @@ macro_rules! dbg_panic { } #[allow(unused_imports)] pub(crate) use dbg_panic; + +/// Represents Activity schedule-to-close and start-to-close timeouts for the purposes of specifying +/// Activity options. Specifying at least one of them is required, but specifying both is also +/// allowed. Note that this type does not cover all available timeout options for an Activity. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ActivityCloseTimeoutOptions { + /// Total time the Activity is allowed to run, including retries. + ScheduleToClose(Duration), + /// Maximum time of a single Activity execution attempt. Note that the Temporal Server doesn't + /// detect Worker process failures directly. It relies on this timeout to detect that an + /// Activity that didn't complete on time. So this timeout should be as short as the longest + /// possible execution of the Activity body. Potentially long running Activities must specify + /// `heartbeat_timeout` in options and heartbeat from the activity periodically for timely + /// failure detection. + StartToClose(Duration), + /// Applies both execution-attempt and overall-completion bounds. + Both { + /// Total time the Activity is allowed to run, including retries. + schedule_to_close: Duration, + /// Maximum time of a single Activity execution attempt. + start_to_close: Duration, + }, +} + +impl ActivityCloseTimeoutOptions { + /// For internal use. Converts options to [`ActivityCloseTimeoutValues`]. + #[doc(hidden)] + pub fn into_values(self) -> ActivityCloseTimeoutValues { + let (schedule_to_close, start_to_close) = match self { + ActivityCloseTimeoutOptions::ScheduleToClose(t) => (Some(t), None), + ActivityCloseTimeoutOptions::StartToClose(t) => (None, Some(t)), + ActivityCloseTimeoutOptions::Both { + schedule_to_close, + start_to_close, + } => (Some(schedule_to_close), Some(start_to_close)), + }; + + ActivityCloseTimeoutValues { + schedule_to_close, + start_to_close, + } + } +} + +/// For internal use. Obtained from [`ActivityCloseTimeoutOptions`]. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[non_exhaustive] // disallow direct construction +#[doc(hidden)] +pub struct ActivityCloseTimeoutValues { + pub schedule_to_close: Option, + pub start_to_close: Option, +} diff --git a/crates/common/src/lib.rs b/crates/common/src/lib.rs index 36a384a28..00a78a1b1 100644 --- a/crates/common/src/lib.rs +++ b/crates/common/src/lib.rs @@ -18,10 +18,10 @@ pub mod protos; pub mod telemetry; pub mod worker; pub use temporalio_common_wasm::{ - ActivityDefinition, ActivityError, HasWorkflowDefinition, Memo, Priority, QueryDefinition, - RetryPolicy, SignalDefinition, UntypedActivity, UntypedWorkflow, UpdateDefinition, - WorkerDeploymentVersion, WorkflowDefinition, WorkflowExecution, data_converters, error, - search_attributes, + ActivityCloseTimeoutOptions, ActivityDefinition, ActivityError, HasWorkflowDefinition, Memo, + Priority, QueryDefinition, RetryPolicy, SignalDefinition, UntypedActivity, UntypedWorkflow, + UpdateDefinition, WorkerDeploymentVersion, WorkflowDefinition, WorkflowExecution, + data_converters, error, search_attributes, }; macro_rules! dbg_panic { diff --git a/crates/sdk-core/tests/heavy_tests.rs b/crates/sdk-core/tests/heavy_tests.rs index 50a80a86e..4a610925d 100644 --- a/crates/sdk-core/tests/heavy_tests.rs +++ b/crates/sdk-core/tests/heavy_tests.rs @@ -5,10 +5,9 @@ pub(crate) mod common; #[path = "heavy_tests/fuzzy_workflow.rs"] mod fuzzy_workflow; -use crate::common::get_integ_runtime_options; use common::{ - CoreWfStarter, activity_functions::StdActivities, init_integ_telem, prom_metrics, rand_6_chars, - workflows::LaProblemWorkflow, + CoreWfStarter, activity_functions::StdActivities, get_integ_runtime_options, init_integ_telem, + prom_metrics, rand_6_chars, workflows::LaProblemWorkflow, }; use futures_util::{ StreamExt, @@ -32,6 +31,7 @@ use temporalio_common::{ use temporalio_macros::{activities, workflow, workflow_methods}; use temporalio_common::{ + ActivityCloseTimeoutOptions, protos::{ coresdk::workflow_commands::ActivityCancellationType, temporal::api::enums::v1::WorkflowIdReusePolicy, @@ -39,7 +39,7 @@ use temporalio_common::{ worker::WorkerTaskTypes, }; use temporalio_sdk::{ - ActivityCloseTimeouts, ActivityOptions, SyncWorkflowContext, WorkflowContext, WorkflowResult, + ActivityOptions, SyncWorkflowContext, WorkflowContext, WorkflowResult, activities::{ActivityContext, ActivityError}, workflows, }; @@ -60,7 +60,7 @@ impl ActivityLoadWf { .execute_activity( StdActivities::echo, input_str.clone(), - ActivityOptions::with_close_timeouts(ActivityCloseTimeouts::Both { + ActivityOptions::with_close_timeouts(ActivityCloseTimeoutOptions::Both { start_to_close: Duration::from_secs(8), schedule_to_close: Duration::from_secs(8), }) diff --git a/crates/sdk-core/tests/integ_tests/async_activity_client_tests.rs b/crates/sdk-core/tests/integ_tests/async_activity_client_tests.rs index 6dfe1057f..0cf1e52d9 100644 --- a/crates/sdk-core/tests/integ_tests/async_activity_client_tests.rs +++ b/crates/sdk-core/tests/integ_tests/async_activity_client_tests.rs @@ -44,8 +44,8 @@ async fn async_activity_completions( #[derive(Clone)] struct SharedActivityInfo { task_token: Vec, - workflow_id: String, - run_id: String, + workflow_id: Option, + workflow_run_id: Option, activity_id: String, } @@ -76,11 +76,10 @@ async fn async_activity_completions( } let activity_info = ctx.info(); - let wf_exec = activity_info.workflow_execution.as_ref().unwrap(); let info = SharedActivityInfo { task_token: activity_info.task_token.clone(), - workflow_id: wf_exec.workflow_id().to_owned(), - run_id: wf_exec.run_id().to_owned(), + workflow_id: activity_info.workflow_id.clone(), + workflow_run_id: activity_info.workflow_run_id.clone(), activity_id: activity_info.activity_id.clone(), }; let _ = self.info_tx.send(info).await; @@ -161,10 +160,10 @@ async fn async_activity_completions( let info = info_rx.recv().await.expect("should receive activity info"); eprintln!( - "DEBUG: Received activity info - task_token_len={}, workflow_id={}, run_id={}, activity_id={}", + "DEBUG: Received activity info - task_token_len={}, workflow_id={:?}, run_id={:?}, activity_id={}", info.task_token.len(), info.workflow_id, - info.run_id, + info.workflow_run_id, info.activity_id ); @@ -175,7 +174,11 @@ async fn async_activity_completions( } IdentifierType::ById => { eprintln!("DEBUG: Using ById identifier"); - ActivityIdentifier::by_id(info.workflow_id, info.run_id, info.activity_id) + ActivityIdentifier::by_id_workflow( + info.workflow_id.unwrap(), + Some(info.workflow_run_id.unwrap()), + info.activity_id, + ) } }; diff --git a/crates/sdk-core/tests/integ_tests/standalone_activity_tests.rs b/crates/sdk-core/tests/integ_tests/standalone_activity_tests.rs new file mode 100644 index 000000000..4429891fd --- /dev/null +++ b/crates/sdk-core/tests/integ_tests/standalone_activity_tests.rs @@ -0,0 +1,300 @@ +use crate::common::{get_integ_client, init_integ_telem, integ_namespace, rand_6_chars}; +use anyhow::anyhow; +use futures_util::{FutureExt, StreamExt, pin_mut, stream}; +use std::{ + collections::HashSet, + panic, + panic::{AssertUnwindSafe, resume_unwind}, + sync::Arc, + time::Duration, +}; +use temporalio_client::{ + ActivityCancelOptions, ActivityDescribeOptions, ActivityExecutionInfoLike, + ActivityExecutionStatus, ActivityStartOptions, ActivityTerminateOptions, Client, + NamespacedClient, errors::ActivityResultError, +}; +use temporalio_common::{ActivityError, error::IncomingError}; +use temporalio_macros::activities; +use temporalio_sdk::{Worker, WorkerOptions, activities::ActivityContext}; +use uuid::Uuid; + +const TASK_QUEUE_PREFIX: &str = "standalone_activity_tests"; + +struct Activities; + +#[activities] +impl Activities { + #[activity] + async fn echo(_ctx: ActivityContext, e: String) -> Result { + Ok(e) + } + + #[activity] + async fn fail_when_canceled( + self: Arc, + ctx: ActivityContext, + ) -> Result<(), ActivityError> { + let mut ticker = tokio::time::interval(Duration::from_millis(100)); + loop { + tokio::select! { biased; + _ = ctx.cancelled() => return Err(anyhow!("canceled").into()), + _ = ticker.tick() => { let _ = ctx.record_heartbeat(()).await; }, + } + } + } +} + +async fn run_test(test: impl AsyncFnOnce(Client, String)) { + let rt = init_integ_telem().unwrap(); + let client = get_integ_client( + integ_namespace(), + rt.telemetry().get_temporal_metric_meter(), + ) + .await; + let task_queue = format!("{TASK_QUEUE_PREFIX}_{}", rand_6_chars()); + let mut worker = Worker::new( + rt, + client.clone(), + WorkerOptions::new(task_queue.clone()).build(), + ) + .unwrap(); + let shutdown_handle = worker.shutdown_handle(); + worker.register_activities(Activities); + + let worker_fut = worker.run(); + let test_fut = async { + let result = AssertUnwindSafe(test(client, task_queue)) + .catch_unwind() + .await; + shutdown_handle(); + result + }; + pin_mut!(worker_fut); + pin_mut!(test_fut); + + tokio::select! { + test_result = &mut test_fut => { + let worker_result = worker_fut.await; + if let Err(panic) = test_result { + resume_unwind(panic); + } + worker_result.unwrap(); + }, + worker_result = &mut worker_fut => { + worker_result.unwrap(); + if let Err(panic) = test_fut.await { + resume_unwind(panic); + } + } + } +} + +macro_rules! test_options { + ($task_queue:expr) => { + ActivityStartOptions::new($task_queue, Uuid::new_v4()) + .schedule_to_close_timeout(Duration::from_secs(60)) + }; +} + +#[tokio::test] +async fn get_result() { + run_test(async |client, tq| { + let options = test_options!(tq).build(); + let arg = "Hello"; + + let handle = client + .start_activity(Activities::echo, arg.into(), options.clone()) + .await + .unwrap(); + assert_eq!(handle.activity_id(), options.id); + assert!(handle.run_id().is_some()); + assert_eq!(handle.result().await.unwrap(), arg); + + let new_handle = client.get_activity_handle( + Activities::echo, + handle.activity_id(), + handle.run_id().map(Into::into), + ); + assert_eq!(new_handle.result().await.unwrap(), arg); + + let untyped_handle = client + .get_untyped_activity_handle(handle.activity_id(), handle.run_id().map(Into::into)); + assert_eq!( + untyped_handle + .result() + .await + .unwrap() + .to_value::(client.data_converter().payload_converter()), + arg + ); + + let wrong_run_id = loop { + let uuid = Some(Uuid::new_v4().to_string()); + if uuid.as_deref() != handle.run_id() { + break uuid; + } + }; + + let handle_wrong_run_id = + client.get_activity_handle(Activities::echo, handle.activity_id(), wrong_run_id); + assert_matches!( + handle_wrong_run_id.result().await, + Err(ActivityResultError::NotFound(_)) + ); + + let handle_no_run_id = + client.get_activity_handle(Activities::echo, handle.activity_id(), None); + assert_eq!(handle_no_run_id.result().await.unwrap(), arg); + }) + .await; +} + +#[tokio::test] +async fn execute() { + run_test(async |client, tq| { + let arg = "Hello"; + let result = client + .execute_activity(Activities::echo, arg.into(), test_options!(tq).build()) + .await + .unwrap(); + + assert_eq!(result, arg); + }) + .await; +} + +#[tokio::test] +async fn describe() { + run_test(async |client, tq| { + let options = test_options!(tq).build(); + let arg = "Hello"; + + let handle = client + .start_activity(Activities::echo, arg.into(), options.clone()) + .await + .unwrap(); + let result = handle.result().await.unwrap(); + + let desc = handle + .describe(ActivityDescribeOptions { + include_input: true, + include_outcome: true, + ..Default::default() + }) + .await + .unwrap(); + + assert_eq!(desc.activity_id(), options.id); + assert_eq!(Some(desc.activity_run_id()), handle.run_id()); + assert_eq!(desc.status(), ActivityExecutionStatus::Completed); + assert_eq!(desc.input().await.unwrap(), Some(arg.to_string())); + assert_eq!(desc.outcome().await.unwrap().unwrap().unwrap(), result); + }) + .await; +} + +#[tokio::test] +async fn cancel() { + run_test(async |client, tq| { + let reason = "test cancel"; + let handle = client + .start_activity( + Activities::fail_when_canceled, + (), + test_options!(tq).build(), + ) + .await + .unwrap(); + handle + .cancel(ActivityCancelOptions { + reason: reason.into(), + }) + .await + .unwrap(); + + assert_matches!( + handle.result().await, + Err(ActivityResultError::ActivityFailed( + IncomingError::Cancelled(_) + )) + ); + let desc = handle.describe(Default::default()).await.unwrap(); + assert_eq!(desc.status(), ActivityExecutionStatus::Canceled); + assert_eq!(desc.canceled_reason(), Some(reason)); + }) + .await; +} + +#[tokio::test] +async fn terminate() { + run_test(async |client, tq| { + let reason = "test terminate"; + let handle = client + .start_activity( + Activities::fail_when_canceled, + (), + test_options!(tq).build(), + ) + .await + .unwrap(); + handle + .terminate(ActivityTerminateOptions { + reason: reason.into(), + }) + .await + .unwrap(); + + assert_matches!( + handle.result().await, + Err(ActivityResultError::ActivityFailed( + IncomingError::Terminated(_) + )) + ); + let desc = handle.describe(Default::default()).await.unwrap(); + assert_eq!(desc.status(), ActivityExecutionStatus::Terminated); + }) + .await; +} + +#[tokio::test] +async fn list_and_count() { + run_test(async |client, tq| { + let query = format!("TaskQueue='{tq}'"); + + let started_activity_ids: HashSet<_> = stream::iter(0..3) + .then(async |_| { + client + .start_activity( + Activities::echo, + "Hello".into(), + test_options!(tq.clone()).build(), + ) + .await + .unwrap() + .activity_id() + .to_string() + }) + .collect() + .await; + + // in loop because of eventual consistency + loop { + let count = client + .count_activities(query.clone(), Default::default()) + .await + .unwrap(); + if count.count() == started_activity_ids.len() { + break; + } + tokio::time::sleep(Duration::from_millis(100)).await; + } + + let list_activity_ids: HashSet<_> = client + .list_activities(query.clone(), Default::default()) + .map(|a| a.unwrap().activity_id().to_string()) + .collect() + .await; + assert_eq!(list_activity_ids, started_activity_ids); + }) + .await; +} diff --git a/crates/sdk-core/tests/integ_tests/workflow_tests/activities.rs b/crates/sdk-core/tests/integ_tests/workflow_tests/activities.rs index d4a7dde8c..11721a20f 100644 --- a/crates/sdk-core/tests/integ_tests/workflow_tests/activities.rs +++ b/crates/sdk-core/tests/integ_tests/workflow_tests/activities.rs @@ -134,7 +134,7 @@ struct ActivityInterceptorRecord { interceptor: &'static str, phase: ActivityInterceptorPhase, activity_type: String, - workflow_type: String, + workflow_type: Option, is_local: bool, input: Option, output: Option, @@ -353,7 +353,7 @@ async fn activity_interceptor_wraps_activity_execution() { interceptor: "outer", phase: ActivityInterceptorPhase::Before, activity_type: "StdActivities::echo".to_owned(), - workflow_type: wf_name.to_owned(), + workflow_type: Some(wf_name.to_owned()), is_local: false, input: Some(input.clone()), output: None, @@ -363,7 +363,7 @@ async fn activity_interceptor_wraps_activity_execution() { interceptor: "inner", phase: ActivityInterceptorPhase::Before, activity_type: "StdActivities::echo".to_owned(), - workflow_type: wf_name.to_owned(), + workflow_type: Some(wf_name.to_owned()), is_local: false, input: Some(input.clone()), output: None, @@ -373,7 +373,7 @@ async fn activity_interceptor_wraps_activity_execution() { interceptor: "inner", phase: ActivityInterceptorPhase::After, activity_type: "StdActivities::echo".to_owned(), - workflow_type: wf_name.to_owned(), + workflow_type: Some(wf_name.to_owned()), is_local: false, input: None, output: Some(input.clone()), @@ -383,7 +383,7 @@ async fn activity_interceptor_wraps_activity_execution() { interceptor: "outer", phase: ActivityInterceptorPhase::After, activity_type: "StdActivities::echo".to_owned(), - workflow_type: wf_name.to_owned(), + workflow_type: Some(wf_name.to_owned()), is_local: false, input: None, output: Some(input.clone()), @@ -433,7 +433,7 @@ async fn activity_interceptor_wraps_local_activity_execution() { interceptor: "local", phase: ActivityInterceptorPhase::Before, activity_type: "StdActivities::echo".to_owned(), - workflow_type: wf_name.to_owned(), + workflow_type: Some(wf_name.to_owned()), is_local: true, input: Some(input.clone()), output: None, @@ -443,7 +443,7 @@ async fn activity_interceptor_wraps_local_activity_execution() { interceptor: "local", phase: ActivityInterceptorPhase::After, activity_type: "StdActivities::echo".to_owned(), - workflow_type: wf_name.to_owned(), + workflow_type: Some(wf_name.to_owned()), is_local: true, input: None, output: Some(input.clone()), @@ -571,7 +571,7 @@ async fn activity_interceptor_observes_activity_error() { interceptor: "failure", phase: ActivityInterceptorPhase::Before, activity_type: "FailingActivities::fail".to_owned(), - workflow_type: wf_name.to_owned(), + workflow_type: Some(wf_name.to_owned()), is_local: false, input: Some(input), output: None, @@ -581,7 +581,7 @@ async fn activity_interceptor_observes_activity_error() { interceptor: "failure", phase: ActivityInterceptorPhase::After, activity_type: "FailingActivities::fail".to_owned(), - workflow_type: wf_name.to_owned(), + workflow_type: Some(wf_name.to_owned()), is_local: false, input: None, output: None, @@ -668,7 +668,7 @@ async fn activity_interceptor_observes_activity_panic() { interceptor: "panic", phase: ActivityInterceptorPhase::Before, activity_type: "PanickingActivities::panic_activity".to_owned(), - workflow_type: wf_name.to_owned(), + workflow_type: Some(wf_name.to_owned()), is_local: false, input: Some(input), output: None, @@ -678,7 +678,7 @@ async fn activity_interceptor_observes_activity_panic() { interceptor: "panic", phase: ActivityInterceptorPhase::After, activity_type: "PanickingActivities::panic_activity".to_owned(), - workflow_type: wf_name.to_owned(), + workflow_type: Some(wf_name.to_owned()), is_local: false, input: None, output: None, diff --git a/crates/sdk-core/tests/main.rs b/crates/sdk-core/tests/main.rs index c9e885844..8e59b8bc6 100644 --- a/crates/sdk-core/tests/main.rs +++ b/crates/sdk-core/tests/main.rs @@ -23,6 +23,7 @@ mod integ_tests { mod polling_tests; mod queries_tests; mod schedule_tests; + mod standalone_activity_tests; mod update_tests; mod visibility_tests; mod worker_heartbeat_tests; diff --git a/crates/sdk/src/activities.rs b/crates/sdk/src/activities.rs index 57d61d4d4..854ecae3b 100644 --- a/crates/sdk/src/activities.rs +++ b/crates/sdk/src/activities.rs @@ -72,7 +72,7 @@ use std::{ use temporalio_client::{Client, ClientOptions, Priority, WorkflowExecutionInfo, WorkflowHandle}; pub use temporalio_common::ActivityError; use temporalio_common::{ - ActivityDefinition, HasWorkflowDefinition, RetryPolicy, WorkflowExecution, + ActivityDefinition, HasWorkflowDefinition, RetryPolicy, data_converters::{ DataConverter, DecodablePayloads, GenericPayloadConverter, PayloadConversionError, PayloadConverter, RawValue, SerializationContext, SerializationContextData, @@ -139,6 +139,10 @@ impl ActivityContext { heartbeat_details, client_options.data_converter.payload_converter().clone(), ); + let (workflow_id, workflow_run_id) = workflow_execution + .map(|we| (we.workflow_id, we.run_id)) + .unzip(); + let activity_run_id = (workflow_id.is_none() && !run_id.is_empty()).then_some(run_id); ( ActivityContext { @@ -150,9 +154,10 @@ impl ActivityContext { info: ActivityInfo { task_token, task_queue, - workflow_type, - workflow_namespace, - workflow_execution: workflow_execution.map(Into::into), + workflow_type: (!workflow_type.is_empty()).then_some(workflow_type), + namespace: workflow_namespace, + workflow_id, + workflow_run_id, activity_id, activity_type, heartbeat_timeout: heartbeat_timeout.try_into_or_none(), @@ -165,7 +170,7 @@ impl ActivityContext { retry_policy: retry_policy.map(Into::into), is_local, priority: priority.map(Into::into).unwrap_or_default(), - run_id: (!run_id.is_empty()).then_some(run_id), + activity_run_id, }, }, input, @@ -225,18 +230,22 @@ impl ActivityContext { /// Return a workflow handle for the workflow execution that started this activity, if any. pub fn workflow_handle(&self) -> Option> { - let workflow_execution = self.info.workflow_execution.as_ref()?; - let run_id = (!workflow_execution.run_id().is_empty()) - .then(|| workflow_execution.run_id().to_owned()); - Some(WorkflowHandle::new( - self.client(), - WorkflowExecutionInfo { - namespace: self.client_options.namespace.clone(), - workflow_id: workflow_execution.workflow_id().to_owned(), - run_id: run_id.clone(), - first_execution_run_id: run_id, - }, - )) + self.info().workflow_id.clone().map(|workflow_id| { + debug_assert!( + self.info().workflow_run_id.is_some(), + "workflow_run_id should be set when workflow_id is set" + ); + let run_id = self.info.workflow_run_id.clone(); + WorkflowHandle::new( + self.client(), + WorkflowExecutionInfo { + namespace: self.client_options.namespace.clone(), + workflow_id, + run_id: run_id.clone(), + first_execution_run_id: run_id, + }, + ) + }) } /// Get headers attached to this activity @@ -295,12 +304,14 @@ impl ActivityHeartbeatDetails { pub struct ActivityInfo { /// An opaque token representing a specific Activity task. pub task_token: Vec, - /// The type of the workflow that invoked this activity. - pub workflow_type: String, - /// The namespace of the workflow that invoked this activity. - pub workflow_namespace: String, - /// The execution of the workflow that invoked this activity. - pub workflow_execution: Option, + /// The type of the workflow that invoked this activity. None for standalone activities. + pub workflow_type: Option, + /// The namespace of this activity. + pub namespace: String, + /// ID of the workflow that invoked this activity. None for standalone activities. + pub workflow_id: Option, + /// Run ID of the workflow that invoked this activity. None for standalone activities. + pub workflow_run_id: Option, /// The ID of this activity. pub activity_id: String, /// The type of this activity. @@ -326,7 +337,7 @@ pub struct ActivityInfo { /// Priority of this activity. If unset uses [Priority::default]. pub priority: Priority, /// Run ID of this activity execution. Only set for standalone activities. - pub run_id: Option, + pub activity_run_id: Option, } /// Deadline calculation. This is a port of diff --git a/crates/sdk/src/lib.rs b/crates/sdk/src/lib.rs index 7d7c039d2..b0ceba8b1 100644 --- a/crates/sdk/src/lib.rs +++ b/crates/sdk/src/lib.rs @@ -89,8 +89,8 @@ pub use crate::error::{ }; pub use temporalio_client::Namespace; pub use temporalio_workflow::{ - ActivityCancellationType, ActivityCloseTimeouts, ActivityOptions, BaseWorkflowContext, - CancellableFuture, ChildWorkflowCancellationType, ChildWorkflowOptions, ContinueAsNewOptions, + ActivityCancellationType, ActivityOptions, BaseWorkflowContext, CancellableFuture, + ChildWorkflowCancellationType, ChildWorkflowOptions, ContinueAsNewOptions, ContinueAsNewVersioningBehavior, ExternalWorkflowHandle, LocalActivityOptions, Memo, MemoValue, MemoValues, NamespacedWorkflowInfo, NexusOperationCancellationType, NexusOperationOptions, ParentClosePolicy, PatchActivationCallback, PatchActivationInput, RetryPolicy, Signal, @@ -1088,10 +1088,12 @@ impl ActivityHalf { tokio::spawn(async move { let act_fut = async move { - if let Some(info) = &ctx.info().workflow_execution { - Span::current() - .record("temporalWorkflowID", info.workflow_id()) - .record("temporalRunID", info.run_id()); + let span = Span::current(); + if let Some(workflow_id) = &ctx.info().workflow_id { + span.record("temporalWorkflowID", workflow_id); + } + if let Some(workflow_run_id) = &ctx.info().workflow_run_id { + span.record("temporalRunID", workflow_run_id); } (act_fn)(args, data_converter, ctx, activity_inbound_interceptors).await } diff --git a/crates/workflow/src/lib.rs b/crates/workflow/src/lib.rs index 518ac4832..d515fe3e8 100644 --- a/crates/workflow/src/lib.rs +++ b/crates/workflow/src/lib.rs @@ -29,7 +29,7 @@ pub use runtime::model::{TimerResult, WorkflowResult, WorkflowTermination}; #[doc(hidden)] pub use runtime::{SdkWakeGuard, is_sdk_wake}; pub use temporalio_common_wasm::{ - Memo, RetryPolicy, + ActivityCloseTimeoutOptions, Memo, RetryPolicy, error::{ ActivityExecutionError, ChildWorkflowExecutionError, ChildWorkflowStartError, RetryState, TimeoutType, WorkflowSignalError, @@ -38,8 +38,8 @@ pub use temporalio_common_wasm::{ #[doc(hidden)] pub use workflow_context::PatchActivationCaller; pub use workflow_context::{ - ActivityCancellationType, ActivityCloseTimeouts, ActivityOptions, BaseWorkflowContext, - CancellableFuture, ChildWorkflowCancellationType, ChildWorkflowOptions, ContinueAsNewOptions, + ActivityCancellationType, ActivityOptions, BaseWorkflowContext, CancellableFuture, + ChildWorkflowCancellationType, ChildWorkflowOptions, ContinueAsNewOptions, ContinueAsNewVersioningBehavior, ExternalWorkflowHandle, LocalActivityOptions, NamespacedWorkflowInfo, NexusOperationCancellationType, NexusOperationOptions, ParentClosePolicy, PatchActivationCallback, PatchActivationInput, Signal, SignalData, diff --git a/crates/workflow/src/workflow_context.rs b/crates/workflow/src/workflow_context.rs index 681bd28fb..5a4e56258 100644 --- a/crates/workflow/src/workflow_context.rs +++ b/crates/workflow/src/workflow_context.rs @@ -2,11 +2,10 @@ mod options; mod view; pub use options::{ - ActivityCancellationType, ActivityCloseTimeouts, ActivityOptions, - ChildWorkflowCancellationType, ChildWorkflowOptions, ContinueAsNewOptions, - ContinueAsNewVersioningBehavior, LocalActivityOptions, NexusOperationCancellationType, - NexusOperationOptions, ParentClosePolicy, Signal, SignalData, TimerOptions, VersioningIntent, - WorkflowIdReusePolicy, + ActivityCancellationType, ActivityOptions, ChildWorkflowCancellationType, ChildWorkflowOptions, + ContinueAsNewOptions, ContinueAsNewVersioningBehavior, LocalActivityOptions, + NexusOperationCancellationType, NexusOperationOptions, ParentClosePolicy, Signal, SignalData, + TimerOptions, VersioningIntent, WorkflowIdReusePolicy, }; pub use temporalio_common_wasm::protos::coresdk::child_workflow::StartChildWorkflowExecutionFailedCause; pub use view::{NamespacedWorkflowInfo, WorkflowContextView}; diff --git a/crates/workflow/src/workflow_context/options.rs b/crates/workflow/src/workflow_context/options.rs index 39b5de48b..f0e7cefc2 100644 --- a/crates/workflow/src/workflow_context/options.rs +++ b/crates/workflow/src/workflow_context/options.rs @@ -2,7 +2,7 @@ use std::{collections::HashMap, time::Duration}; use crate::{MemoValues, runtime::types::ContinueAsNewRequest}; use temporalio_common_wasm::{ - Priority, RetryPolicy, + ActivityCloseTimeoutOptions, Priority, RetryPolicy, data_converters::{ GenericPayloadConverter, PayloadConversionError, PayloadConverter, SerializationContext, SerializationContextData, @@ -285,9 +285,9 @@ impl From for NexusOperationCancellationTyp pub struct ActivityOptions { /// Timeouts for activity completion. /// - /// See [`ActivityCloseTimeouts`] for the meaning of each timeout variant. + /// See [`ActivityCloseTimeoutOptions`] for the meaning of each timeout variant. #[builder(start_fn)] - pub close_timeouts: ActivityCloseTimeouts, + pub close_timeouts: ActivityCloseTimeoutOptions, /// Identifier to use for tracking the activity in Workflow history. /// The `activityId` can be accessed by the activity function. /// Does not need to be unique. @@ -324,14 +324,14 @@ pub struct ActivityOptions { } impl ActivityOptions { - /// Returns a builder with `close_timeout` set to [`ActivityCloseTimeouts::StartToClose`]. + /// Returns a builder with `close_timeout` set to [`ActivityCloseTimeoutOptions::StartToClose`]. pub fn with_start_to_close_timeout(duration: Duration) -> ActivityOptionsBuilder { - Self::with_close_timeouts(ActivityCloseTimeouts::StartToClose(duration)) + Self::with_close_timeouts(ActivityCloseTimeoutOptions::StartToClose(duration)) } - /// Returns a builder with `close_timeout` set to [`ActivityCloseTimeouts::ScheduleToClose`]. + /// Returns a builder with `close_timeout` set to [`ActivityCloseTimeoutOptions::ScheduleToClose`]. pub fn with_schedule_to_close_timeout(duration: Duration) -> ActivityOptionsBuilder { - Self::with_close_timeouts(ActivityCloseTimeouts::ScheduleToClose(duration)) + Self::with_close_timeouts(ActivityCloseTimeoutOptions::ScheduleToClose(duration)) } /// Creates activity options with only `start_to_close_timeout` set. @@ -349,43 +349,6 @@ impl ActivityOptions { } } -/// The timeouts applied to an activity's completion. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum ActivityCloseTimeouts { - /// Total time that a workflow is willing to wait for Activity to complete. - /// `ActivityCloseTimeouts::ScheduleToClose` limits the total time of an Activity's execution - /// including retries (use `ActivityCloseTimeouts::StartToClose` to limit the time of a single - /// attempt). - ScheduleToClose(Duration), - /// Maximum time of a single Activity execution attempt. Note that the Temporal Server doesn't - /// detect Worker process failures directly. It relies on this timeout to detect that an - /// Activity that didn't complete on time. So this timeout should be as short as the longest - /// possible execution of the Activity body. Potentially long running Activities must specify - /// `ActivityOptions::heartbeat_timeout` and heartbeat from the activity periodically for timely - /// failure detection. - StartToClose(Duration), - /// Applies both execution-attempt and overall-completion bounds. - Both { - /// Maximum time of a single Activity execution attempt. - start_to_close: Duration, - /// Total time that a workflow is willing to wait for Activity to complete. - schedule_to_close: Duration, - }, -} - -impl ActivityCloseTimeouts { - fn into_durations(self) -> (Option, Option) { - match self { - Self::ScheduleToClose(schedule_to_close) => (None, Some(schedule_to_close)), - Self::StartToClose(start_to_close) => (Some(start_to_close), None), - Self::Both { - start_to_close, - schedule_to_close, - } => (Some(start_to_close), Some(schedule_to_close)), - } - } -} - impl ActivityOptions { pub(crate) fn into_command( self, @@ -393,8 +356,7 @@ impl ActivityOptions { activity_type: String, args: Vec, ) -> WorkflowCommand { - let (start_to_close_timeout, schedule_to_close_timeout) = - self.close_timeouts.into_durations(); + let close_timeouts = self.close_timeouts.into_values(); command_with_metadata( workflow_command::Variant::ScheduleActivity(ScheduleActivity { seq, @@ -402,12 +364,14 @@ impl ActivityOptions { activity_id: self.activity_id.unwrap_or_else(|| seq.to_string()), task_queue: self.task_queue.unwrap_or_default(), arguments: args, - schedule_to_close_timeout: schedule_to_close_timeout + schedule_to_close_timeout: close_timeouts + .schedule_to_close .and_then(|duration| duration.try_into().ok()), schedule_to_start_timeout: self .schedule_to_start_timeout .and_then(|duration| duration.try_into().ok()), - start_to_close_timeout: start_to_close_timeout + start_to_close_timeout: close_timeouts + .start_to_close .and_then(|duration| duration.try_into().ok()), heartbeat_timeout: self .heartbeat_timeout @@ -829,7 +793,6 @@ pub struct ContinueAsNewOptions { /// search attributes. pub search_attributes: Option, /// If set, the new workflow will have this retry policy. If `None`, reuses the current policy. - #[builder(into)] pub retry_policy: Option, /// Whether the new workflow should run on a worker with a compatible build id. pub versioning_intent: Option, @@ -999,7 +962,7 @@ mod tests { assert_eq!( opts.close_timeouts, - ActivityCloseTimeouts::StartToClose(Duration::from_secs(5)) + ActivityCloseTimeoutOptions::StartToClose(Duration::from_secs(5)) ); assert_eq!(opts.heartbeat_timeout, Some(Duration::from_secs(2))); } @@ -1012,14 +975,14 @@ mod tests { assert_eq!( opts.close_timeouts, - ActivityCloseTimeouts::ScheduleToClose(Duration::from_secs(5)) + ActivityCloseTimeoutOptions::ScheduleToClose(Duration::from_secs(5)) ); assert_eq!(opts.heartbeat_timeout, Some(Duration::from_secs(2))); } #[test] fn activity_options_both_close_timeouts_map_to_command() { - let req = ActivityOptions::with_close_timeouts(ActivityCloseTimeouts::Both { + let req = ActivityOptions::with_close_timeouts(ActivityCloseTimeoutOptions::Both { start_to_close: Duration::from_secs(3), schedule_to_close: Duration::from_secs(8), })