temporalio_sdk/
activities.rs

1//! Functionality related to defining and interacting with activities
2//!
3//!
4//! An example of defining an activity:
5//! ```
6//! use std::sync::{
7//!     Arc,
8//!     atomic::{AtomicUsize, Ordering},
9//! };
10//! use temporalio_macros::{activities, activity_definitions};
11//! use temporalio_sdk::activities::{ActivityContext, ActivityError};
12//!
13//! struct MyActivities {
14//!     counter: AtomicUsize,
15//! }
16//!
17//! #[activities]
18//! impl MyActivities {
19//!     #[activity]
20//!     async fn echo(_ctx: ActivityContext, e: String) -> Result<String, ActivityError> {
21//!         Ok(e)
22//!     }
23//!
24//!     #[activity]
25//!     async fn uses_self(self: Arc<Self>, _ctx: ActivityContext) -> Result<(), ActivityError> {
26//!         self.counter.fetch_add(1, Ordering::Relaxed);
27//!         Ok(())
28//!     }
29//! }
30//!
31//! // If you need to refer to an activity that is defined externally, in a different codebase or
32//! // possibly a different language, use `#[activity_definitions]`. Methods must omit the
33//! // `ActivityContext` parameter and have a body of `unimplemented!()`. Workflows can then call
34//! // these definitions just like real activities.
35//!
36//! struct ExternalActivities;
37//! #[activity_definitions]
38//! impl ExternalActivities {
39//!     #[activity(name = "foo")]
40//!     fn foo(_: String) -> Result<String, ActivityError> {
41//!         unimplemented!()
42//!     }
43//! }
44//! ```
45//!
46//! This will allows you to call the activity from workflow code still, but the actual function
47//! will never be invoked, since you won't have registered it with the worker.
48
49#[doc(inline)]
50pub use temporalio_macros::activities;
51
52use crate::{
53    OutgoingActivityError, OutgoingError,
54    interceptors::{
55        ActivityExecutionValue, ActivityInboundInterceptor, ExecuteActivityInput,
56        ExecuteActivityOutput, Next,
57    },
58    panic_formatter,
59};
60use futures_util::{
61    FutureExt,
62    future::{BoxFuture, ready},
63};
64use prost_types::{Duration, Timestamp};
65use std::{
66    collections::HashMap,
67    fmt::Debug,
68    panic::AssertUnwindSafe,
69    sync::Arc,
70    time::{Duration as StdDuration, SystemTime},
71};
72use temporalio_client::{Client, ClientOptions, Priority, WorkflowExecutionInfo, WorkflowHandle};
73pub use temporalio_common::ActivityError;
74use temporalio_common::{
75    ActivityDefinition, HasWorkflowDefinition, RetryPolicy, WorkflowExecution,
76    data_converters::{
77        DataConverter, DecodablePayloads, GenericPayloadConverter, PayloadConversionError,
78        PayloadConverter, RawValue, SerializationContext, SerializationContextData,
79        TemporalDeserializable, TemporalSerializable,
80    },
81    error::ApplicationFailure,
82    protos::{
83        coresdk::{ActivityHeartbeat, activity_result::ActivityExecutionResult, activity_task},
84        temporal::api::common::v1::Payload,
85        utilities::TryIntoOrNone,
86    },
87};
88use temporalio_sdk_core::Worker as CoreWorker;
89use tokio_util::sync::CancellationToken;
90
91/// Used within activities to get info, heartbeat management etc.
92#[derive(Clone)]
93pub struct ActivityContext {
94    worker: Arc<CoreWorker>,
95    client_options: ClientOptions,
96    cancellation_token: CancellationToken,
97    heartbeat_details: ActivityHeartbeatDetails,
98    header_fields: HashMap<String, Payload>,
99    info: ActivityInfo,
100}
101
102impl ActivityContext {
103    pub(crate) fn new(
104        worker: Arc<CoreWorker>,
105        client_options: ClientOptions,
106        cancellation_token: CancellationToken,
107        task_queue: String,
108        task_token: Vec<u8>,
109        task: activity_task::Start,
110    ) -> (Self, Vec<Payload>) {
111        let activity_task::Start {
112            workflow_namespace,
113            workflow_type,
114            workflow_execution,
115            activity_id,
116            activity_type,
117            header_fields,
118            input,
119            heartbeat_details,
120            scheduled_time,
121            current_attempt_scheduled_time,
122            started_time,
123            attempt,
124            schedule_to_close_timeout,
125            start_to_close_timeout,
126            heartbeat_timeout,
127            retry_policy,
128            is_local,
129            priority,
130            run_id,
131        } = task;
132        let deadline = calculate_deadline(
133            scheduled_time.as_ref(),
134            started_time.as_ref(),
135            start_to_close_timeout.as_ref(),
136            schedule_to_close_timeout.as_ref(),
137        );
138        let heartbeat_details = ActivityHeartbeatDetails::new(
139            heartbeat_details,
140            client_options.data_converter.payload_converter().clone(),
141        );
142
143        (
144            ActivityContext {
145                worker,
146                client_options,
147                cancellation_token,
148                heartbeat_details,
149                header_fields,
150                info: ActivityInfo {
151                    task_token,
152                    task_queue,
153                    workflow_type,
154                    workflow_namespace,
155                    workflow_execution: workflow_execution.map(Into::into),
156                    activity_id,
157                    activity_type,
158                    heartbeat_timeout: heartbeat_timeout.try_into_or_none(),
159                    scheduled_time: scheduled_time.try_into_or_none(),
160                    started_time: started_time.try_into_or_none(),
161                    deadline,
162                    attempt,
163                    current_attempt_scheduled_time: current_attempt_scheduled_time
164                        .try_into_or_none(),
165                    retry_policy: retry_policy.map(Into::into),
166                    is_local,
167                    priority: priority.map(Into::into).unwrap_or_default(),
168                    run_id: (!run_id.is_empty()).then_some(run_id),
169                },
170            },
171            input,
172        )
173    }
174
175    /// Returns a future the completes if and when the activity this was called inside has been
176    /// cancelled
177    pub async fn cancelled(&self) {
178        self.cancellation_token.clone().cancelled().await
179    }
180
181    /// Returns true if this activity has already been cancelled
182    pub fn is_cancelled(&self) -> bool {
183        self.cancellation_token.is_cancelled()
184    }
185
186    /// Extract heartbeat details from last failed attempt. This is used in combination with retry
187    /// policy.
188    pub fn heartbeat_details(&self) -> &ActivityHeartbeatDetails {
189        &self.heartbeat_details
190    }
191
192    /// Record a heartbeat with typed progress details for the currently executing activity.
193    pub async fn record_heartbeat<T>(&self, details: T) -> Result<(), PayloadConversionError>
194    where
195        T: TemporalSerializable + 'static,
196    {
197        if !self.info.is_local {
198            let details = self
199                .client_options
200                .data_converter
201                .to_payloads(&SerializationContextData::Activity, &details)
202                .await?;
203            self.worker.record_activity_heartbeat(ActivityHeartbeat {
204                task_token: self.info.task_token.clone(),
205                details,
206            })
207        }
208        Ok(())
209    }
210
211    /// Returns activity info of the executing activity
212    pub fn info(&self) -> &ActivityInfo {
213        &self.info
214    }
215
216    /// Return a client targeting the same Temporal service and namespace as this activity's worker.
217    pub fn client(&self) -> Client {
218        let connection = self.worker.get_client_connection().expect(
219            "activity context client is unavailable because the worker was not created from a \
220             Temporal client",
221        );
222        Client::new(connection, self.client_options.clone())
223            .expect("client construction from a worker connection should be infallible")
224    }
225
226    /// Return a workflow handle for the workflow execution that started this activity, if any.
227    pub fn workflow_handle<W: HasWorkflowDefinition>(&self) -> Option<WorkflowHandle<Client, W>> {
228        let workflow_execution = self.info.workflow_execution.as_ref()?;
229        let run_id = (!workflow_execution.run_id().is_empty())
230            .then(|| workflow_execution.run_id().to_owned());
231        Some(WorkflowHandle::new(
232            self.client(),
233            WorkflowExecutionInfo {
234                namespace: self.client_options.namespace.clone(),
235                workflow_id: workflow_execution.workflow_id().to_owned(),
236                run_id: run_id.clone(),
237                first_execution_run_id: run_id,
238            },
239        ))
240    }
241
242    /// Get headers attached to this activity
243    pub fn headers(&self) -> &HashMap<String, Payload> {
244        &self.header_fields
245    }
246
247    pub(crate) fn headers_mut(&mut self) -> &mut HashMap<String, Payload> {
248        &mut self.header_fields
249    }
250}
251
252/// Heartbeat details supplied by the previous activity attempt.
253#[derive(Clone, Debug)]
254#[non_exhaustive]
255pub struct ActivityHeartbeatDetails {
256    payloads: DecodablePayloads,
257}
258
259impl ActivityHeartbeatDetails {
260    fn new(payloads: Vec<Payload>, payload_converter: PayloadConverter) -> Self {
261        Self {
262            payloads: DecodablePayloads::new(
263                payloads,
264                payload_converter,
265                SerializationContextData::Activity,
266            ),
267        }
268    }
269
270    /// Deserialize the previous heartbeat details, or return `None` when there are none.
271    pub fn deserialize<T: TemporalDeserializable + 'static>(
272        &self,
273    ) -> Result<Option<T>, PayloadConversionError> {
274        if self.payloads.raw().is_empty() {
275            Ok(None)
276        } else {
277            self.payloads.deserialize().map(Some)
278        }
279    }
280
281    /// Returns the codec-decoded raw heartbeat payloads.
282    pub fn raw(&self) -> &[Payload] {
283        self.payloads.raw()
284    }
285
286    /// Consume these details and return their codec-decoded payloads.
287    pub fn into_raw(self) -> RawValue {
288        self.payloads.into_raw()
289    }
290}
291
292/// Various information about a specific activity attempt.
293#[derive(Clone, Debug)]
294#[non_exhaustive]
295pub struct ActivityInfo {
296    /// An opaque token representing a specific Activity task.
297    pub task_token: Vec<u8>,
298    /// The type of the workflow that invoked this activity.
299    pub workflow_type: String,
300    /// The namespace of the workflow that invoked this activity.
301    pub workflow_namespace: String,
302    /// The execution of the workflow that invoked this activity.
303    pub workflow_execution: Option<WorkflowExecution>,
304    /// The ID of this activity.
305    pub activity_id: String,
306    /// The type of this activity.
307    pub activity_type: String,
308    /// The task queue of this activity.
309    pub task_queue: String,
310    /// The interval within which this activity must heartbeat or be timed out.
311    pub heartbeat_timeout: Option<StdDuration>,
312    /// Time activity was scheduled by a workflow.
313    pub scheduled_time: Option<SystemTime>,
314    /// Time of activity start.
315    pub started_time: Option<SystemTime>,
316    /// Time of activity timeout.
317    pub deadline: Option<SystemTime>,
318    /// Attempt starts from 1, and increase by 1 for every retry, if retry policy is specified.
319    pub attempt: u32,
320    /// Time this attempt at the activity was scheduled.
321    pub current_attempt_scheduled_time: Option<SystemTime>,
322    /// The retry policy for this activity.
323    pub retry_policy: Option<RetryPolicy>,
324    /// Whether or not this is a local activity.
325    pub is_local: bool,
326    /// Priority of this activity. If unset uses [Priority::default].
327    pub priority: Priority,
328    /// Run ID of this activity execution. Only set for standalone activities.
329    pub run_id: Option<String>,
330}
331
332/// Deadline calculation.  This is a port of
333/// https://github.com/temporalio/sdk-go/blob/8651550973088f27f678118f997839fb1bb9e62f/internal/activity.go#L225
334fn calculate_deadline(
335    scheduled_time: Option<&Timestamp>,
336    started_time: Option<&Timestamp>,
337    start_to_close_timeout: Option<&Duration>,
338    schedule_to_close_timeout: Option<&Duration>,
339) -> Option<SystemTime> {
340    match (
341        scheduled_time,
342        started_time,
343        start_to_close_timeout,
344        schedule_to_close_timeout,
345    ) {
346        (
347            Some(scheduled),
348            Some(started),
349            Some(start_to_close_timeout),
350            Some(schedule_to_close_timeout),
351        ) => {
352            let scheduled: SystemTime = maybe_convert_timestamp(scheduled)?;
353            let started: SystemTime = maybe_convert_timestamp(started)?;
354            let start_to_close_timeout: StdDuration = (*start_to_close_timeout).try_into().ok()?;
355            let schedule_to_close_timeout: StdDuration =
356                (*schedule_to_close_timeout).try_into().ok()?;
357
358            let start_to_close_deadline: SystemTime =
359                started.checked_add(start_to_close_timeout)?;
360            if schedule_to_close_timeout > StdDuration::ZERO {
361                let schedule_to_close_deadline =
362                    scheduled.checked_add(schedule_to_close_timeout)?;
363                // Minimum of the two deadlines.
364                if schedule_to_close_deadline < start_to_close_deadline {
365                    Some(schedule_to_close_deadline)
366                } else {
367                    Some(start_to_close_deadline)
368                }
369            } else {
370                Some(start_to_close_deadline)
371            }
372        }
373        _ => None,
374    }
375}
376
377/// Helper function lifted from prost_types::Timestamp implementation to prevent double cloning in
378/// error construction
379fn maybe_convert_timestamp(timestamp: &Timestamp) -> Option<SystemTime> {
380    let mut timestamp = *timestamp;
381    timestamp.normalize();
382
383    let system_time = if timestamp.seconds >= 0 {
384        std::time::UNIX_EPOCH.checked_add(StdDuration::from_secs(timestamp.seconds as u64))
385    } else {
386        std::time::UNIX_EPOCH.checked_sub(StdDuration::from_secs((-timestamp.seconds) as u64))
387    };
388
389    system_time.and_then(|system_time| {
390        system_time.checked_add(StdDuration::from_nanos(timestamp.nanos as u64))
391    })
392}
393
394pub(crate) type ActivityInvocation = Arc<
395    dyn Fn(
396            Vec<Payload>,
397            DataConverter,
398            ActivityContext,
399            Vec<Arc<dyn ActivityInboundInterceptor>>,
400        ) -> ExecuteActivityOutput<'static>
401        + Send
402        + Sync,
403>;
404
405fn call_execute_activity<'a>(
406    interceptors: &'a [Arc<dyn ActivityInboundInterceptor>],
407    input: ExecuteActivityInput,
408    next: Next<'a, ExecuteActivityInput, ExecuteActivityOutput<'a>>,
409) -> ExecuteActivityOutput<'a> {
410    if let Some((first, rest)) = interceptors.split_first() {
411        first.execute_activity(
412            input,
413            Next::new(move |input| call_execute_activity(rest, input, next)),
414        )
415    } else {
416        next.run(input)
417    }
418}
419
420#[doc(hidden)]
421pub trait ActivityImplementer {
422    fn register_all(self: Arc<Self>, defs: &mut ActivityDefinitions);
423}
424
425#[doc(hidden)]
426pub trait ExecutableActivity: ActivityDefinition + Sized {
427    type Implementer: ActivityImplementer + Send + Sync + 'static;
428    fn definition() -> Self;
429    fn execute(
430        receiver: Option<Arc<Self::Implementer>>,
431        ctx: ActivityContext,
432        input: Self::Input,
433    ) -> BoxFuture<'static, Result<Self::Output, ActivityError>>;
434}
435
436#[doc(hidden)]
437pub trait HasOnlyStaticMethods {}
438
439/// Contains activity registrations in a form ready for execution by workers.
440#[derive(Default, Clone)]
441pub struct ActivityDefinitions {
442    activities: HashMap<String, ActivityInvocation>,
443}
444
445impl ActivityDefinitions {
446    /// Registers all activities on an activity implementer.
447    pub fn register_activities<AI: ActivityImplementer>(&mut self, instance: AI) -> &mut Self {
448        let arcd = Arc::new(instance);
449        AI::register_all(arcd, self);
450        self
451    }
452    /// Registers a specific activitiy.
453    pub fn register_activity<AD>(&mut self, instance: Arc<AD::Implementer>) -> &mut Self
454    where
455        AD: ActivityDefinition + ExecutableActivity,
456        AD::Input: Send + Sync,
457        AD::Output: Send + Sync,
458    {
459        self.activities.insert(
460            AD::definition().name().to_string(),
461            Arc::new(move |payloads, dc, c, activity_inbound_interceptors| {
462                let instance = instance.clone();
463                async move {
464                    // Codec application happens at the SDK/Core boundary, so activity
465                    // implementations work with the payload converter directly.
466                    let pc = dc.payload_converter();
467                    let ctx = SerializationContext {
468                        data: &SerializationContextData::Activity,
469                        converter: pc,
470                    };
471                    let input: AD::Input = pc.from_payloads(&ctx, payloads)?;
472                    let input = ExecuteActivityInput::new(c, Box::new(input));
473                    let leaf = activity_inbound_base::<AD>(instance);
474                    let activity_execution =
475                        call_execute_activity(&activity_inbound_interceptors, input, leaf);
476                    match AssertUnwindSafe(activity_execution).catch_unwind().await {
477                        Ok(output) => output,
478                        Err(panic) => Err(ApplicationFailure::new(anyhow::anyhow!(
479                            "Activity function panicked: {}",
480                            panic_formatter(panic)
481                        ))
482                        .into()),
483                    }
484                }
485                .boxed()
486            }),
487        );
488        self
489    }
490
491    pub(crate) fn is_empty(&self) -> bool {
492        self.activities.is_empty()
493    }
494
495    pub(crate) fn get(&self, act_type: &str) -> Option<ActivityInvocation> {
496        self.activities.get(act_type).cloned()
497    }
498
499    pub(crate) fn names(&self) -> Vec<String> {
500        let mut names: Vec<_> = self.activities.keys().cloned().collect();
501        names.sort_unstable();
502        names
503    }
504}
505
506fn activity_inbound_base<'a, AD>(
507    instance: Arc<AD::Implementer>,
508) -> Next<'a, ExecuteActivityInput, ExecuteActivityOutput<'a>>
509where
510    AD: ActivityDefinition + ExecutableActivity,
511    AD::Input: Send + Sync,
512    AD::Output: Send + Sync,
513{
514    Next::new(
515        move |input: ExecuteActivityInput| -> ExecuteActivityOutput<'a> {
516            let (activity_context, args) = input.into_parts();
517            let args = match args.downcast::<AD::Input>() {
518                Ok(args) => args,
519                Err(_) => {
520                    return ready(Err(ApplicationFailure::new(anyhow::anyhow!(
521                    "Activity inbound interceptor returned arguments with wrong concrete type for activity {}",
522                    AD::definition().name()
523                ))
524                .into()))
525                .boxed();
526                }
527            };
528
529            async move {
530                match AssertUnwindSafe(AD::execute(Some(instance), activity_context, *args))
531                    .catch_unwind()
532                    .await
533                {
534                    Ok(result) => {
535                        result.map(|output| Box::new(output) as Box<dyn ActivityExecutionValue>)
536                    }
537                    Err(panic) => Err(ApplicationFailure::new(anyhow::anyhow!(
538                        "Activity function panicked: {}",
539                        panic_formatter(panic)
540                    ))
541                    .into()),
542                }
543            }
544            .boxed()
545        },
546    )
547}
548
549pub(crate) fn activity_error_to_core_result(
550    dc: &DataConverter,
551    err: ActivityError,
552) -> ActivityExecutionResult {
553    match err {
554        ActivityError::Application(app) => ActivityExecutionResult::fail(dc.to_failure(
555            &SerializationContextData::Activity,
556            OutgoingError::Activity(OutgoingActivityError::Application(app)),
557        )),
558        ActivityError::Cancelled { details } => ActivityExecutionResult::cancel(dc.to_failure(
559            &SerializationContextData::Activity,
560            OutgoingError::Activity(OutgoingActivityError::Cancelled { details }),
561        )),
562        ActivityError::WillCompleteAsync => ActivityExecutionResult::will_complete_async(),
563    }
564}
565
566impl Debug for ActivityDefinitions {
567    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
568        f.debug_struct("ActivityDefinitions")
569            .field("activities", &self.activities.keys())
570            .finish()
571    }
572}
573
574#[cfg(test)]
575mod test {
576    use super::*;
577    use rstest::rstest;
578    use temporalio_common::error::{ApplicationErrorCategory, ApplicationFailure};
579
580    #[test]
581    fn activity_heartbeat_details_support_typed_decoding() {
582        let payload_converter = PayloadConverter::default();
583        let payload = payload_converter
584            .to_payload(
585                &SerializationContext {
586                    data: &SerializationContextData::Activity,
587                    converter: &payload_converter,
588                },
589                &"progress".to_owned(),
590            )
591            .unwrap();
592        let details = ActivityHeartbeatDetails::new(vec![payload.clone()], payload_converter);
593
594        assert_eq!(details.raw(), &[payload]);
595        assert_eq!(
596            details.deserialize::<String>().unwrap(),
597            Some("progress".to_owned())
598        );
599    }
600
601    #[test]
602    fn empty_activity_heartbeat_details_decode_to_none() {
603        let details = ActivityHeartbeatDetails::new(Vec::new(), PayloadConverter::default());
604
605        assert_eq!(details.deserialize::<String>().unwrap(), None);
606        assert!(details.into_raw().payloads.is_empty());
607    }
608
609    #[rstest]
610    #[case(true)]
611    #[case(false)]
612    fn activity_error_conversion_is_not_lossy(#[case] non_retryable: bool) {
613        let original = ApplicationFailure::builder(anyhow::anyhow!("big boom"))
614            .type_name("BigBoom".to_owned())
615            .non_retryable(non_retryable)
616            .next_retry_delay(StdDuration::from_secs(3))
617            .category(ApplicationErrorCategory::Benign)
618            .details("details")
619            .build();
620        let err = ActivityError::from(original);
621        let ActivityError::Application(actual) = err else {
622            panic!("application failure should become app failure")
623        };
624        assert_eq!(actual.type_name(), Some("BigBoom"));
625        assert_eq!(actual.is_non_retryable(), non_retryable);
626        assert_eq!(actual.next_retry_delay(), Some(StdDuration::from_secs(3)));
627        assert_eq!(actual.category(), ApplicationErrorCategory::Benign);
628        assert_eq!(actual.to_string(), "big boom");
629    }
630
631    #[test]
632    fn activity_error_from_special_err_becomes_application() {
633        #[derive(Debug, PartialEq)]
634        struct MyError;
635
636        impl std::error::Error for MyError {}
637        impl std::fmt::Display for MyError {
638            fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
639                f.write_str("MyError")
640            }
641        }
642
643        let err = ActivityError::from(MyError);
644        let ActivityError::Application(actual) = err else {
645            panic!("expected application failure, got {err:?}")
646        };
647        assert_eq!(actual.to_string(), "MyError");
648    }
649}
temporalio_sdk/activities.rs

temporalio_sdk/
activities.rs
