temporalio_client/activity/
activity_execution_info.rs

1use crate::Priority;
2use std::{
3    error::Error,
4    marker::PhantomData,
5    time::{Duration, SystemTime},
6};
7use temporalio_common::{
8    ActivityDefinition, RetryPolicy, UntypedActivity, WorkerDeploymentVersion,
9    data_converters::{
10        DataConverter, NoopDecodeHint, PayloadConversionError, SerializationContextData,
11        TemporalDeserializable,
12    },
13    error::IncomingError,
14    protos::{
15        proto_ts_to_system_time,
16        temporal::api::{
17            activity::v1::{
18                ActivityExecutionInfo as RawInfo, ActivityExecutionListInfo as RawListInfo,
19                activity_execution_outcome::Value as ActivityExecutionOutcomeValue,
20            },
21            common::v1::{Payload, Payloads},
22            enums::v1::{
23                ActivityExecutionStatus as ProtoActivityExecutionStatus,
24                PendingActivityState as ProtoPendingActivityState,
25            },
26            failure::v1::Failure,
27            workflowservice::v1::DescribeActivityExecutionResponse,
28        },
29        utilities::TryIntoOrNone,
30    },
31    search_attributes::SearchAttributes,
32};
33
34/// Common methods of [`ActivityExecutionInfo`] and [`ActivityExecutionDescription`].
35pub trait ActivityExecutionInfoLike {
36    /// ID of the activity.
37    fn activity_id(&self) -> &str;
38    /// Run ID of a particular execution of the activity.
39    fn activity_run_id(&self) -> &str;
40    /// Type of the activity.
41    fn activity_type(&self) -> &str;
42    /// Time the activity was originally scheduled.
43    fn schedule_time(&self) -> Option<SystemTime>;
44    /// Time when the activity transitioned to a closed state.
45    fn close_time(&self) -> Option<SystemTime>;
46    /// A general status for this activity, indicates whether it is currently running or in one of
47    /// the terminal statuses.
48    fn status(&self) -> ActivityExecutionStatus;
49    /// The task queue this activity was scheduled on.
50    fn task_queue(&self) -> &str;
51    /// The difference between close time and scheduled time. This field is only populated if
52    /// the activity is closed.
53    fn execution_duration(&self) -> Option<Duration>;
54}
55
56/// Contains basic information about an activity.
57/// Obtained from [`Client::list_activities`](crate::Client::list_activities).
58pub struct ActivityExecutionInfo {
59    raw: RawListInfo,
60}
61
62impl From<RawListInfo> for ActivityExecutionInfo {
63    fn from(raw: RawListInfo) -> Self {
64        Self { raw }
65    }
66}
67
68impl ActivityExecutionInfoLike for ActivityExecutionInfo {
69    fn activity_id(&self) -> &str {
70        &self.raw.activity_id
71    }
72
73    fn activity_run_id(&self) -> &str {
74        &self.raw.run_id
75    }
76
77    fn activity_type(&self) -> &str {
78        self.raw
79            .activity_type
80            .as_ref()
81            .map(|t| t.name.as_str())
82            .unwrap_or("")
83    }
84
85    fn schedule_time(&self) -> Option<SystemTime> {
86        self.raw
87            .schedule_time
88            .as_ref()
89            .and_then(proto_ts_to_system_time)
90    }
91
92    fn close_time(&self) -> Option<SystemTime> {
93        self.raw
94            .close_time
95            .as_ref()
96            .and_then(proto_ts_to_system_time)
97    }
98
99    fn status(&self) -> ActivityExecutionStatus {
100        ProtoActivityExecutionStatus::try_from(self.raw.status)
101            .map(Into::into)
102            .unwrap_or(ActivityExecutionStatus::Unknown)
103    }
104
105    fn task_queue(&self) -> &str {
106        &self.raw.task_queue
107    }
108
109    fn execution_duration(&self) -> Option<Duration> {
110        self.raw.execution_duration.try_into_or_none()
111    }
112}
113
114impl ActivityExecutionInfo {
115    /// Raw Protobuf object from server response.
116    pub fn raw_info(&self) -> &RawListInfo {
117        &self.raw
118    }
119}
120
121/// Contains the current state of the activity execution.
122/// Obtained from [`ActivityHandle::describe`](crate::ActivityHandle::describe).
123/// Methods that deserialize payloads (e.g. [`heartbeat_details`](Self::heartbeat_details)) use
124/// [`DataConverter`] of the client associated with the activity handle.
125pub struct ActivityExecutionDescription<ActivityT = UntypedActivity>
126where
127    ActivityT: ActivityDefinition,
128{
129    raw_info: RawInfo,
130    raw_input: Option<Payloads>,
131    raw_outcome: Option<ActivityExecutionOutcomeValue>,
132    data_converter: DataConverter,
133    serialization_context: SerializationContextData,
134    _phantom: PhantomData<ActivityT>,
135}
136
137impl<ActivityT> ActivityExecutionInfoLike for ActivityExecutionDescription<ActivityT>
138where
139    ActivityT: ActivityDefinition,
140{
141    fn activity_id(&self) -> &str {
142        &self.raw_info.activity_id
143    }
144
145    fn activity_run_id(&self) -> &str {
146        &self.raw_info.run_id
147    }
148
149    fn activity_type(&self) -> &str {
150        self.raw_info
151            .activity_type
152            .as_ref()
153            .map(|t| t.name.as_str())
154            .unwrap_or("")
155    }
156
157    fn schedule_time(&self) -> Option<SystemTime> {
158        self.raw_info
159            .schedule_time
160            .as_ref()
161            .and_then(proto_ts_to_system_time)
162    }
163
164    fn close_time(&self) -> Option<SystemTime> {
165        self.raw_info
166            .close_time
167            .as_ref()
168            .and_then(proto_ts_to_system_time)
169    }
170
171    fn status(&self) -> ActivityExecutionStatus {
172        ProtoActivityExecutionStatus::try_from(self.raw_info.status)
173            .map(Into::into)
174            .unwrap_or(ActivityExecutionStatus::Unknown)
175    }
176
177    fn task_queue(&self) -> &str {
178        &self.raw_info.task_queue
179    }
180
181    fn execution_duration(&self) -> Option<Duration> {
182        self.raw_info.execution_duration.try_into_or_none()
183    }
184}
185
186impl<ActivityT> ActivityExecutionDescription<ActivityT>
187where
188    ActivityT: ActivityDefinition,
189{
190    pub(crate) fn new(
191        data_converter: DataConverter,
192        serialization_context: SerializationContextData,
193        response: DescribeActivityExecutionResponse,
194    ) -> Result<Self, Box<dyn Error + Send + Sync + 'static>> {
195        let Some(raw_info) = response.info else {
196            return Err("info missing in describe response".into());
197        };
198        Ok(Self {
199            raw_info,
200            raw_input: response.input,
201            raw_outcome: response.outcome.and_then(|o| o.value),
202            data_converter,
203            serialization_context,
204            _phantom: PhantomData,
205        })
206    }
207
208    /// Convert to an untyped description object.
209    pub fn untyped(self) -> ActivityExecutionDescription {
210        ActivityExecutionDescription {
211            raw_info: self.raw_info,
212            raw_input: self.raw_input,
213            raw_outcome: self.raw_outcome,
214            data_converter: self.data_converter,
215            serialization_context: self.serialization_context,
216            _phantom: PhantomData,
217        }
218    }
219
220    /// Raw Protobuf object from server response.
221    pub fn raw_info(&self) -> &RawInfo {
222        &self.raw_info
223    }
224
225    /// True if activity input is present.
226    /// See [`ActivityDescribeOptions::include_input`](crate::ActivityDescribeOptions::include_input).
227    /// Use [`input`](Self::input) or [`raw_input`](Self::raw_input) to retrieve it.
228    pub fn has_input(&self) -> bool {
229        self.raw_input.is_some()
230    }
231
232    /// Raw payload of activity input, if it was requested.
233    pub fn raw_input(&self) -> Option<&Payloads> {
234        self.raw_input.as_ref()
235    }
236
237    /// Deserialize activity input. Returns `Ok(None)` if not present.
238    /// See [`ActivityDescribeOptions::include_input`](crate::ActivityDescribeOptions::include_input).
239    pub async fn input(&self) -> Result<Option<ActivityT::Input>, PayloadConversionError> {
240        let Some(input) = &self.raw_input else {
241            return Ok(None);
242        };
243        Ok(Some(self.convert_payloads(input).await?))
244    }
245
246    /// True if activity outcome is present.
247    /// See [`ActivityDescribeOptions::include_outcome`](crate::ActivityDescribeOptions::include_outcome).
248    /// Use [`outcome`](Self::outcome) or [`raw_outcome`](Self::outcome) to retrieve it.
249    pub fn has_outcome(&self) -> bool {
250        self.raw_outcome.is_some()
251    }
252
253    /// Raw payload of activity output, if it was requested and available.
254    pub fn raw_outcome(&self) -> Option<&ActivityExecutionOutcomeValue> {
255        self.raw_outcome.as_ref()
256    }
257
258    /// Deserialize activity outcome. Returns `Ok(None)` if not present.
259    /// See [`ActivityDescribeOptions::include_outcome`](crate::ActivityDescribeOptions::include_outcome).
260    pub async fn outcome(
261        &self,
262    ) -> Result<Option<Result<ActivityT::Output, IncomingError>>, PayloadConversionError> {
263        match &self.raw_outcome {
264            None => Ok(None),
265            Some(ActivityExecutionOutcomeValue::Result(payloads)) => {
266                Ok(Some(Ok(self.convert_payloads(payloads).await?)))
267            }
268            Some(ActivityExecutionOutcomeValue::Failure(failure)) => {
269                Ok(Some(Err(self.convert_failure(failure)?)))
270            }
271        }
272    }
273
274    /// More detailed breakdown of [`ActivityExecutionStatus::Running`].
275    pub fn run_state(&self) -> PendingActivityState {
276        ProtoPendingActivityState::try_from(self.raw_info.run_state)
277            .map(Into::into)
278            .unwrap_or(PendingActivityState::Unknown)
279    }
280
281    /// Indicates how long the caller is willing to wait for an activity completion. Limits how long
282    /// retries will be attempted.
283    pub fn schedule_to_close_timeout(&self) -> Option<Duration> {
284        self.raw_info.schedule_to_close_timeout.try_into_or_none()
285    }
286
287    /// Limits time an activity task can stay in a task queue before a worker picks it up. This
288    /// timeout is always non-retryable.
289    pub fn schedule_to_start_timeout(&self) -> Option<Duration> {
290        self.raw_info.schedule_to_start_timeout.try_into_or_none()
291    }
292
293    /// Maximum time a single activity attempt is allowed to execute after being picked up by
294    /// a worker. This timeout is always retryable.
295    pub fn start_to_close_timeout(&self) -> Option<Duration> {
296        self.raw_info.start_to_close_timeout.try_into_or_none()
297    }
298
299    /// Maximum permitted time between successful worker heartbeats.
300    pub fn heartbeat_timeout(&self) -> Option<Duration> {
301        self.raw_info.heartbeat_timeout.try_into_or_none()
302    }
303
304    /// The retry policy for the activity.
305    pub fn retry_policy(&self) -> Option<RetryPolicy> {
306        self.raw_info.retry_policy.clone().map(Into::into)
307    }
308
309    /// True if heartbeat details are present.
310    /// See [`ActivityDescribeOptions::include_heartbeat_details`](crate::ActivityDescribeOptions::include_heartbeat_details).
311    /// Use [`heartbeat_details`](Self::heartbeat_details) or
312    /// [`raw_info()`](Self::raw_info)`.`[`heartbeat_details`](RawInfo::heartbeat_details)
313    /// to retrieve them.
314    pub fn has_heartbeat_details(&self) -> bool {
315        self.raw_info.heartbeat_details.is_some()
316    }
317
318    /// Deserialize heartbeat details. Returns `Ok(None)` if not present.
319    /// See [`ActivityDescribeOptions::include_heartbeat_details`](crate::ActivityDescribeOptions::include_heartbeat_details).
320    pub async fn heartbeat_details<T: TemporalDeserializable + 'static>(
321        &self,
322    ) -> Result<Option<T>, PayloadConversionError> {
323        let Some(details) = &self.raw_info.heartbeat_details else {
324            return Ok(None);
325        };
326        Ok(Some(self.convert_payloads(details).await?))
327    }
328
329    /// Time the last heartbeat was recorded.
330    pub fn last_heartbeat_time(&self) -> Option<SystemTime> {
331        self.raw_info
332            .last_heartbeat_time
333            .as_ref()
334            .and_then(proto_ts_to_system_time)
335    }
336
337    /// Time the last attempt was started.
338    pub fn last_started_time(&self) -> Option<SystemTime> {
339        self.raw_info
340            .last_started_time
341            .as_ref()
342            .and_then(proto_ts_to_system_time)
343    }
344
345    /// The attempt this activity is currently on. Incremented each time a new attempt is scheduled.
346    pub fn attempt(&self) -> u32 {
347        self.raw_info.attempt.try_into().unwrap_or_default()
348    }
349
350    /// How long this activity has been running for, including all attempts and backoff between
351    /// attempts.
352    pub fn execution_duration(&self) -> Option<Duration> {
353        self.raw_info.execution_duration.try_into_or_none()
354    }
355
356    /// Scheduled time + schedule to close timeout.
357    pub fn expiration_time(&self) -> Option<SystemTime> {
358        self.raw_info
359            .expiration_time
360            .as_ref()
361            .and_then(proto_ts_to_system_time)
362    }
363
364    /// True if last failure is present.
365    /// See [`ActivityDescribeOptions::include_last_failure`](crate::ActivityDescribeOptions::include_last_failure).
366    /// Use [`last_failure()`](Self::last_failure) or
367    /// [`raw_info()`](Self::raw_info)`.`[`last_failure`](RawInfo::last_failure)
368    /// to retrieve it.
369    pub fn has_last_failure(&self) -> bool {
370        self.raw_info.last_failure.is_some()
371    }
372
373    /// Deserialize last failure. Returns `Ok(None)` if not present.
374    /// See [`ActivityDescribeOptions::include_last_failure`](crate::ActivityDescribeOptions::include_last_failure).
375    pub fn last_failure(&self) -> Result<Option<IncomingError>, PayloadConversionError> {
376        let Some(failure) = &self.raw_info.last_failure else {
377            return Ok(None);
378        };
379        Ok(Some(self.convert_failure(failure)?))
380    }
381
382    /// Identity of the last worker that attempted this activity.
383    pub fn last_worker_identity(&self) -> Option<&str> {
384        self.raw_info
385            .last_worker_identity
386            .is_empty()
387            .then_some(self.raw_info.last_worker_identity.as_str())
388    }
389
390    /// Time from the last attempt failure to the next activity retry.
391    pub fn current_retry_interval(&self) -> Option<Duration> {
392        self.raw_info.current_retry_interval.try_into_or_none()
393    }
394
395    /// The time when the last activity attempt completed.
396    pub fn last_attempt_complete_time(&self) -> Option<SystemTime> {
397        self.raw_info
398            .last_attempt_complete_time
399            .as_ref()
400            .and_then(proto_ts_to_system_time)
401    }
402
403    /// The time when the next activity attempt will be scheduled.
404    pub fn next_attempt_schedule_time(&self) -> Option<SystemTime> {
405        self.raw_info
406            .next_attempt_schedule_time
407            .as_ref()
408            .and_then(proto_ts_to_system_time)
409    }
410
411    /// The Worker Deployment Version this activity was dispatched to most recently.
412    pub fn last_deployment_version(&self) -> Option<WorkerDeploymentVersion> {
413        self.raw_info
414            .last_deployment_version
415            .clone()
416            .map(Into::into)
417    }
418
419    /// Priority metadata.
420    pub fn priority(&self) -> Priority {
421        self.raw_info.priority.clone().unwrap_or_default().into()
422    }
423
424    /// Search attributes of the activity.
425    pub fn search_attributes(&self) -> Option<SearchAttributes> {
426        self.raw_info
427            .search_attributes
428            .as_ref()
429            .map(SearchAttributes::from_proto)
430    }
431
432    /// Deserialize static summary that was set when activity was scheduled.
433    /// Returns `Ok(None)` if not present.
434    pub async fn static_summary(&self) -> Result<Option<String>, PayloadConversionError> {
435        let Some(summary) = self
436            .raw_info
437            .user_metadata
438            .as_ref()
439            .and_then(|m| m.summary.clone())
440        else {
441            return Ok(None);
442        };
443        Ok(Some(self.convert_payload(summary).await?))
444    }
445
446    /// Deserialize static details that were set when activity was scheduled.
447    /// Returns `Ok(None)` if not present.
448    pub async fn static_details(&self) -> Result<Option<String>, PayloadConversionError> {
449        let Some(details) = self
450            .raw_info
451            .user_metadata
452            .as_ref()
453            .and_then(|m| m.details.clone())
454        else {
455            return Ok(None);
456        };
457        Ok(Some(self.convert_payload(details).await?))
458    }
459
460    /// Reason for activity cancellation if activity was canceled and reason was provided.
461    pub fn canceled_reason(&self) -> Option<&str> {
462        let reason = self.raw_info.canceled_reason.as_str();
463        (!reason.is_empty()).then_some(reason)
464    }
465
466    /// Time to wait before dispatching the first activity task.
467    /// This delay is not applied to retry attempts.
468    pub fn start_delay(&self) -> Option<Duration> {
469        self.raw_info.start_delay.try_into_or_none()
470    }
471
472    async fn convert_payload<T: TemporalDeserializable + 'static>(
473        &self,
474        payload: Payload,
475    ) -> Result<T, PayloadConversionError> {
476        self.data_converter
477            .from_payload(&self.serialization_context, payload)
478            .await
479    }
480
481    async fn convert_payloads<T: TemporalDeserializable + 'static>(
482        &self,
483        payloads: &Payloads,
484    ) -> Result<T, PayloadConversionError> {
485        self.data_converter
486            .from_payloads(&self.serialization_context, payloads.payloads.clone())
487            .await
488    }
489
490    fn convert_failure(&self, failure: &Failure) -> Result<IncomingError, PayloadConversionError> {
491        self.data_converter
492            .to_error(&self.serialization_context, failure.clone(), NoopDecodeHint)
493    }
494}
495
496/// Execution status of an activity. See [`ActivityExecutionInfoLike::status`].
497#[non_exhaustive]
498#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)]
499pub enum ActivityExecutionStatus {
500    #[default]
501    /// This variant indicates the server did not specify a value.
502    Unspecified,
503    /// The activity has not reached a terminal status.
504    /// See [`ActivityExecutionDescription::run_state`] for the run state.
505    Running,
506    /// The activity completed successfully.
507    Completed,
508    /// The activity failed with an error.
509    Failed,
510    /// The activity was canceled. Note that cancellation is cooperative and a cancel request does
511    /// not always result in canceled status.
512    Canceled,
513    /// The activity was terminated.
514    Terminated,
515    /// The activity timed out.
516    TimedOut,
517    /// The activity is paused.
518    Paused,
519    /// This variant indicates the server used a value not known by this version of the SDK.
520    Unknown,
521}
522
523impl From<ProtoActivityExecutionStatus> for ActivityExecutionStatus {
524    fn from(value: ProtoActivityExecutionStatus) -> Self {
525        match value {
526            ProtoActivityExecutionStatus::Unspecified => Self::Unspecified,
527            ProtoActivityExecutionStatus::Running => Self::Running,
528            ProtoActivityExecutionStatus::Completed => Self::Completed,
529            ProtoActivityExecutionStatus::Failed => Self::Failed,
530            ProtoActivityExecutionStatus::Canceled => Self::Canceled,
531            ProtoActivityExecutionStatus::Terminated => Self::Terminated,
532            ProtoActivityExecutionStatus::TimedOut => Self::TimedOut,
533            ProtoActivityExecutionStatus::Paused => Self::Paused,
534        }
535    }
536}
537
538/// Detailed state of an activity with [`ActivityExecutionStatus::Running`].
539/// See [`ActivityExecutionDescription::run_state`].
540#[non_exhaustive]
541#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)]
542pub enum PendingActivityState {
543    #[default]
544    /// This variant indicates the server did not specify a state.
545    Unspecified,
546    /// Activity is scheduled for execution but not yet running on a worker.
547    Scheduled,
548    /// Activity is running on a worker.
549    Started,
550    /// Activity has been requested to cancel.
551    CancelRequested,
552    /// Activity is paused on the server, and is not running on a worker.
553    Paused,
554    /// Activity is currently running on a worker, but paused on the server.
555    PauseRequested,
556    /// This variant indicates the server used a value not known by this version of the SDK.
557    Unknown,
558}
559
560impl From<ProtoPendingActivityState> for PendingActivityState {
561    fn from(value: ProtoPendingActivityState) -> Self {
562        match value {
563            ProtoPendingActivityState::Unspecified => Self::Unspecified,
564            ProtoPendingActivityState::Scheduled => Self::Scheduled,
565            ProtoPendingActivityState::Started => Self::Started,
566            ProtoPendingActivityState::CancelRequested => Self::CancelRequested,
567            ProtoPendingActivityState::Paused => Self::Paused,
568            ProtoPendingActivityState::PauseRequested => Self::PauseRequested,
569        }
570    }
571}
temporalio_client/activity/activity_execution_info.rs

temporalio_client/activity/
activity_execution_info.rs
