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

temporalio_client/activity/
activity_execution_info.rs
