durable_workflow/
local_activity.rs

1use super::*;
2
3const MAX_ATTEMPTS: u64 = 100;
4const MAX_HEARTBEATS: usize = 1000;
5const LEASE_RENEWAL_INTERVAL: Duration = Duration::from_secs(1);
6
7/// Retry and timeout settings for an activity executed by the workflow worker.
8///
9/// Local activities bypass the activity queue. They do not accept routing,
10/// schedule-to-start timeouts, or remote activity cancellation policies.
11#[derive(Clone, Debug, Default, PartialEq, Eq)]
12pub struct LocalActivityOptions {
13    pub retry_policy: Option<ActivityRetryPolicy>,
14    pub start_to_close_timeout: Option<Duration>,
15    pub schedule_to_close_timeout: Option<Duration>,
16    pub heartbeat_timeout: Option<Duration>,
17}
18
19impl LocalActivityOptions {
20    pub fn new() -> Self {
21        Self::default()
22    }
23    pub fn retry_policy(mut self, policy: ActivityRetryPolicy) -> Self {
24        self.retry_policy = Some(policy);
25        self
26    }
27    pub fn start_to_close_timeout(mut self, timeout: Duration) -> Self {
28        self.start_to_close_timeout = Some(timeout);
29        self
30    }
31    pub fn schedule_to_close_timeout(mut self, timeout: Duration) -> Self {
32        self.schedule_to_close_timeout = Some(timeout);
33        self
34    }
35    pub fn heartbeat_timeout(mut self, timeout: Duration) -> Self {
36        self.heartbeat_timeout = Some(timeout);
37        self
38    }
39    fn into_activity_options(self) -> ActivityOptions {
40        ActivityOptions {
41            retry_policy: self.retry_policy,
42            start_to_close_timeout: self.start_to_close_timeout,
43            schedule_to_close_timeout: self.schedule_to_close_timeout,
44            heartbeat_timeout: self.heartbeat_timeout,
45            ..ActivityOptions::default()
46        }
47    }
48}
49
50pub(super) fn validate(options: &ValidatedActivityOptions) -> Result<()> {
51    let max_attempts = options
52        .retry_policy
53        .as_ref()
54        .and_then(|policy| policy["max_attempts"].as_u64())
55        .unwrap_or(1);
56    if max_attempts > MAX_ATTEMPTS {
57        return Err(Error::InvalidActivityOptions(ActivityOptionsError::new(
58            ActivityOptionsErrorKind::InvalidMaxAttempts,
59            Some("max_attempts"),
60            "local activities support at most 100 attempts",
61        )));
62    }
63    for seconds in [
64        options.start_to_close_timeout,
65        options.schedule_to_close_timeout,
66        options.heartbeat_timeout,
67    ]
68    .into_iter()
69    .flatten()
70    .chain(options.retry_policy.iter().flat_map(|p| {
71        p["backoff_seconds"]
72            .as_array()
73            .into_iter()
74            .flatten()
75            .filter_map(Value::as_u64)
76    })) {
77        if Instant::now()
78            .checked_add(Duration::from_secs(seconds))
79            .is_none()
80        {
81            return Err(Error::InvalidActivityOptions(ActivityOptionsError::new(
82                ActivityOptionsErrorKind::TimeoutOverflow,
83                Some("local_activity"),
84                "local timing exceeds the monotonic clock range",
85            )));
86        }
87    }
88    Ok(())
89}
90
91impl WorkflowContext {
92    /// Execute a registered activity in this workflow worker and record its outcome.
93    ///
94    /// Committed results are replayed without executing the callback. A worker
95    /// lost before acknowledgment can execute the callback again, so side effects
96    /// must be idempotent. Callbacks must yield to Tokio and avoid blocking work.
97    pub fn local_activity<T: Serialize>(
98        &self,
99        activity_type: impl Into<String>,
100        args: T,
101    ) -> ActivityCall {
102        self.local_activity_with_options(activity_type, LocalActivityOptions::new(), args)
103    }
104
105    pub fn local_activity_with_options<T: Serialize>(
106        &self,
107        activity_type: impl Into<String>,
108        options: LocalActivityOptions,
109        args: T,
110    ) -> ActivityCall {
111        let mut call =
112            self.activity_with_options(activity_type, options.into_activity_options(), args);
113        call.local = true;
114        call
115    }
116
117    /// Lossless Avro result, including bytes and large integers.
118    pub async fn local_activity_avro_value<T: Serialize>(
119        &self,
120        activity_type: impl Into<String>,
121        args: T,
122    ) -> Result<AvroValue> {
123        self.local_activity_avro_value_with_options(
124            activity_type,
125            LocalActivityOptions::new(),
126            args,
127        )
128        .await
129    }
130
131    pub async fn local_activity_avro_value_with_options<T: Serialize>(
132        &self,
133        activity_type: impl Into<String>,
134        options: LocalActivityOptions,
135        args: T,
136    ) -> Result<AvroValue> {
137        let mut call = self.local_activity_with_options(activity_type, options, args);
138        std::future::poll_fn(|cx| Pin::new(&mut call).poll_avro_value(cx)).await
139    }
140
141    pub async fn local_activity_typed<I: Serialize, O: DeserializeOwned>(
142        &self,
143        activity_type: impl Into<String>,
144        args: I,
145    ) -> Result<O> {
146        self.local_activity_typed_with_options(activity_type, LocalActivityOptions::new(), args)
147            .await
148    }
149
150    pub async fn local_activity_typed_with_options<I: Serialize, O: DeserializeOwned>(
151        &self,
152        activity_type: impl Into<String>,
153        options: LocalActivityOptions,
154        args: I,
155    ) -> Result<O> {
156        let activity_type = activity_type.into();
157        let result = self
158            .local_activity_avro_value_with_options(activity_type.clone(), options, args)
159            .await?;
160        decode_handler_result(result, HandlerKind::Activity, &activity_type)
161    }
162}
163
164#[derive(Debug)]
165pub(super) struct Request {
166    pub command_index: usize,
167    pub options: ValidatedActivityOptions,
168    pub arguments: AvroValue,
169    pub result: Arc<Mutex<Option<ActivityOutcome>>>,
170}
171
172#[derive(Debug)]
173pub(super) struct Heartbeats {
174    active: bool,
175    started: Instant,
176    last: Instant,
177    reports: Vec<Value>,
178    capacity: usize,
179    overflowed: bool,
180}
181
182impl Heartbeats {
183    pub fn record<T: Serialize>(&mut self, details: T) -> Result<ActivityHeartbeatResponse> {
184        if !self.active {
185            return Ok(heartbeat_response(false));
186        }
187        if self.reports.len() >= self.capacity {
188            self.overflowed = true;
189            return Err(Error::WorkerLoop(
190                "local_activity_heartbeat_limit_exceeded: at most 1000 reports per local command"
191                    .into(),
192            ));
193        }
194        let details = encode_typed_envelope(&AvroValue::from_serialize(&details)?, DEFAULT_CODEC)?;
195        let now = Instant::now();
196        self.reports.push(
197            json!({"elapsed_ms": millis(now.duration_since(self.started)), "details": details}),
198        );
199        self.last = now;
200        Ok(heartbeat_response(true))
201    }
202}
203
204fn heartbeat_response(active: bool) -> ActivityHeartbeatResponse {
205    ActivityHeartbeatResponse {
206        cancel_requested: false,
207        heartbeat_recorded: active,
208        can_continue: Some(active),
209        reason: (!active).then(|| "local_activity_attempt_closed".into()),
210        run_closed_reason: None,
211        run_closed_at: None,
212        lease_expires_at: None,
213        last_heartbeat_at: None,
214    }
215}
216
217struct CloseAttempt(Arc<Mutex<Heartbeats>>);
218impl Drop for CloseAttempt {
219    fn drop(&mut self) {
220        if let Ok(mut heartbeats) = self.0.lock() {
221            heartbeats.active = false;
222        }
223    }
224}
225
226fn millis(duration: Duration) -> u64 {
227    u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
228}
229
230impl Worker {
231    pub(super) async fn execute_workflow_with_local_activities(
232        &self,
233        task: WorkflowTask,
234    ) -> Result<Option<WorkflowTaskDecision>> {
235        let (task, context, mut future) = match self.prepare_workflow_task_execution(task, None)? {
236            PreparedWorkflowTask::Decision(decision) => return Ok(Some(decision)),
237            PreparedWorkflowTask::Execution {
238                task,
239                context,
240                future,
241            } => (task, context, future),
242        };
243        loop {
244            let outcome = Self::poll_workflow_future(&context, &mut future)?;
245            let requests = {
246                let mut state = context
247                    .state
248                    .lock()
249                    .map_err(|_| Error::WorkflowStatePoisoned)?;
250                std::mem::take(&mut state.local_activity_requests)
251            };
252            if requests.is_empty() {
253                return self
254                    .finish_workflow_task_execution(&task, &context, outcome)
255                    .map(Some);
256            }
257            if !self.client.local_activities_enabled {
258                return Err(Error::WorkerLoop("local_activities_not_enabled: opt in with Worker::local_activities(true) before executing inline local work".into()));
259            }
260            // A custom future cannot publish a terminal decision while ignoring
261            // an unresolved local call. Resolve only genuinely suspended work.
262            if !outcome.is_pending() {
263                return Err(Error::WorkflowYieldedWithoutCommand);
264            }
265            for request in requests {
266                let Some((wire, result)) = self
267                    .execute_local_activity(&task, &context, &request)
268                    .await?
269                else {
270                    // Do not fail, complete, or continue the workflow after a
271                    // refused or uncertain lease. Reclaim owns subsequent work.
272                    return Ok(None);
273                };
274                let mut state = context
275                    .state
276                    .lock()
277                    .map_err(|_| Error::WorkflowStatePoisoned)?;
278                state.commands[request.command_index] = wire;
279                *request
280                    .result
281                    .lock()
282                    .map_err(|_| Error::WorkflowStatePoisoned)? = Some(result);
283            }
284        }
285    }
286
287    async fn local_claim_is_active(&self, task: &WorkflowTask) -> bool {
288        tokio::select! {
289            biased;
290            _ = self.local_worker_stopped() => false,
291            receipt = self.client.heartbeat_workflow_task_with_protocol(task, None, WORKER_PROTOCOL_VERSION) => {
292                receipt.is_ok_and(|receipt| receipt.cancellation_request.is_none())
293            }
294        }
295    }
296
297    async fn local_worker_stopped(&self) {
298        match &self.client.worker_storage_admission {
299            Some(admission) => wait_for_worker_stop(&admission.stop).await,
300            None => std::future::pending::<()>().await,
301        }
302    }
303
304    async fn externalize_local_command(
305        &self,
306        task: &WorkflowTask,
307        context: &WorkflowContext,
308        index: usize,
309        wire: &mut Value,
310    ) -> Result<()> {
311        let mut commands = context
312            .state
313            .lock()
314            .map_err(|_| Error::WorkflowStatePoisoned)?
315            .commands
316            .clone();
317        commands[index] = wire.clone();
318        let mut body = json!({"lease_owner": task.lease_owner, "workflow_task_attempt": task.workflow_task_attempt,
319            "commands": commands});
320        let path = format!("/worker/workflow-tasks/{}/complete", task.task_id);
321        self.client
322            .externalize_runtime_payloads(
323                &mut body,
324                &path,
325                RequestProtocol::Worker(WORKER_PROTOCOL_VERSION),
326            )
327            .await?;
328        *wire = body["commands"][index].take();
329        Ok(())
330    }
331
332    async fn execute_local_activity(
333        &self,
334        task: &WorkflowTask,
335        context: &WorkflowContext,
336        request: &Request,
337    ) -> Result<Option<(Value, ActivityOutcome)>> {
338        let mut wire = context
339            .state
340            .lock()
341            .map_err(|_| Error::WorkflowStatePoisoned)?
342            .commands[request.command_index]
343            .clone();
344        // Serialize and admit input before the callback can produce side effects.
345        if self
346            .externalize_local_command(task, context, request.command_index, &mut wire)
347            .await
348            .is_err()
349        {
350            return Ok(None);
351        }
352        let activity_type = wire["activity_type"]
353            .as_str()
354            .unwrap_or_default()
355            .to_owned();
356        let policy = request.options.retry_policy.as_ref();
357        let max_attempts = policy
358            .and_then(|policy| policy["max_attempts"].as_u64())
359            .unwrap_or(1);
360        let started = Instant::now();
361        let total_deadline = request
362            .options
363            .schedule_to_close_timeout
364            .map(|seconds| started + Duration::from_secs(seconds));
365        let mut attempts = Vec::new();
366        let mut heartbeat_count = 0;
367        for number in 1..=max_attempts {
368            if !self.local_claim_is_active(task).await {
369                return Ok(None);
370            }
371            let attempt_started = Instant::now();
372            let reports = Arc::new(Mutex::new(Heartbeats {
373                active: true,
374                started: attempt_started,
375                last: attempt_started,
376                reports: Vec::new(),
377                capacity: MAX_HEARTBEATS - heartbeat_count,
378                overflowed: false,
379            }));
380            let close = CloseAttempt(reports.clone());
381            let attempt_id = format!(
382                "{:x}",
383                Sha256::digest(
384                    format!(
385                        "{}\0{}\0{}\0{}",
386                        task.task_id, task.workflow_task_attempt, request.command_index, number
387                    )
388                    .as_bytes()
389                )
390            );
391            let ctx = ActivityContext {
392                client: self.client.clone(),
393                task_id: task.task_id.clone(),
394                activity_attempt_id: attempt_id.clone(),
395                lease_owner: task.lease_owner.clone().unwrap_or_default(),
396                activity_type: activity_type.clone(),
397                attempt_number: number,
398                task_queue: self.task_queue.clone(),
399                worker_id: self.worker_id.clone(),
400                claim_guard: None,
401                local_heartbeats: Some(reports.clone()),
402                worker_session: None,
403            };
404            let callback = self
405                .activities
406                .get(&activity_type)
407                .filter(|_| !total_deadline.is_some_and(|deadline| Instant::now() >= deadline));
408            let mut future: ActivityFuture = match callback {
409                Some(callback) => callback(ctx, request.arguments.clone()),
410                None if total_deadline.is_some_and(|deadline| Instant::now() >= deadline) => {
411                    Box::pin(std::future::pending())
412                }
413                None => {
414                    let unknown = activity_type.clone();
415                    Box::pin(async move { Err(Error::ActivityNotRegistered(unknown)) })
416                }
417            };
418            let attempt_deadline = request
419                .options
420                .start_to_close_timeout
421                .map(|seconds| attempt_started + Duration::from_secs(seconds));
422            let outcome = loop {
423                let heartbeat_deadline = request
424                    .options
425                    .heartbeat_timeout
426                    .map(|seconds| {
427                        reports
428                            .lock()
429                            .map(|reports| reports.last + Duration::from_secs(seconds))
430                    })
431                    .transpose()
432                    .map_err(|_| Error::WorkflowStatePoisoned)?;
433                let next_timeout = [
434                    (total_deadline, "schedule_to_close"),
435                    (attempt_deadline, "start_to_close"),
436                    (heartbeat_deadline, "heartbeat"),
437                ]
438                .into_iter()
439                .filter_map(|(deadline, kind)| deadline.map(|d| (d, kind)))
440                .min_by_key(|(deadline, _)| *deadline);
441                let timeout_kind = next_timeout
442                    .map(|(_, kind)| kind)
443                    .unwrap_or("start_to_close");
444                let timeout = async {
445                    match next_timeout {
446                        Some((deadline, _)) => tokio::time::sleep_until(deadline.into()).await,
447                        None => std::future::pending::<()>().await,
448                    }
449                };
450                tokio::pin!(timeout);
451                tokio::select! {
452                    biased;
453                    _ = self.local_worker_stopped() => return Ok(None),
454                    _ = &mut timeout => {
455                        if timeout_kind == "heartbeat" && heartbeat_extended(&reports, request.options.heartbeat_timeout)? { continue; }
456                        break Err((format!("local activity exceeded {timeout_kind} timeout"), "LocalActivityTimeout".to_owned(), false, Some(timeout_kind)));
457                    },
458                    result = &mut future => break result.map_err(local_error),
459                    _ = tokio::time::sleep(LEASE_RENEWAL_INTERVAL) => {
460                        tokio::select! {
461                            biased;
462                            _ = &mut timeout => {
463                                if timeout_kind == "heartbeat" && heartbeat_extended(&reports, request.options.heartbeat_timeout)? { continue; }
464                                break Err((format!("local activity exceeded {timeout_kind} timeout"), "LocalActivityTimeout".to_owned(), false, Some(timeout_kind)));
465                            },
466                            active = self.local_claim_is_active(task) => { if !active { return Ok(None); } }
467                        }
468                    }
469                }
470            };
471            drop(future);
472            drop(close);
473            let (heartbeats, overflowed) = {
474                let mut reports = reports.lock().map_err(|_| Error::WorkflowStatePoisoned)?;
475                (std::mem::take(&mut reports.reports), reports.overflowed)
476            };
477            heartbeat_count += heartbeats.len();
478            let outcome = if overflowed {
479                Err((
480                    "local activity exceeded 1000 heartbeat reports".into(),
481                    "LocalActivityHeartbeatLimit".into(),
482                    true,
483                    None,
484                ))
485            } else {
486                outcome
487            };
488            let mut report = json!({"attempt_number": number, "attempt_id": attempt_id,
489                "duration_ms": millis(attempt_started.elapsed()), "heartbeats": heartbeats});
490            let (result, failure) = match outcome {
491                Ok(result) => match encode_typed_envelope(&result, &task.payload_codec) {
492                    Ok(envelope) => {
493                        wire["result"] = envelope;
494                        (Some(result), None)
495                    }
496                    Err(error) => (
497                        None,
498                        Some((
499                            error.to_string(),
500                            "LocalActivityResultCodecError".into(),
501                            true,
502                            None,
503                        )),
504                    ),
505                },
506                Err(failure) => (None, Some(failure)),
507            };
508            if let Some((message, exception_type, mut non_retryable, timeout_kind)) = failure {
509                non_retryable |= policy
510                    .and_then(|policy| policy["non_retryable_error_types"].as_array())
511                    .is_some_and(|types| {
512                        types
513                            .iter()
514                            .any(|t| t.as_str() == Some(exception_type.as_str()))
515                    });
516                let status = if timeout_kind.is_some() {
517                    "timed_out"
518                } else {
519                    "failed"
520                };
521                report["outcome"] = json!(status);
522                report["message"] = json!(message);
523                report["exception_type"] = json!(exception_type);
524                report["non_retryable"] = json!(non_retryable);
525                if let Some(kind) = timeout_kind {
526                    report["timeout_kind"] = json!(kind);
527                }
528                let retry = number < max_attempts
529                    && !non_retryable
530                    && timeout_kind != Some("schedule_to_close");
531                if retry {
532                    let backoff = policy
533                        .and_then(|p| p["backoff_seconds"].get((number - 1) as usize))
534                        .and_then(Value::as_u64)
535                        .unwrap_or(0);
536                    report["retry_reason"] = json!(if timeout_kind.is_some() {
537                        "timeout"
538                    } else {
539                        "failure"
540                    });
541                    report["backoff_seconds"] = json!(backoff);
542                    attempts.push(report);
543                    let wake = Instant::now() + Duration::from_secs(backoff);
544                    while Instant::now() < wake {
545                        // An elapsed total deadline is reported as the next
546                        // terminal attempt, with no further callback invocation.
547                        if total_deadline.is_some_and(|deadline| Instant::now() >= deadline) {
548                            break;
549                        }
550                        if !self.local_claim_is_active(task).await {
551                            return Ok(None);
552                        }
553                        let deadline = total_deadline.map_or(wake, |d| d.min(wake));
554                        tokio::time::sleep(
555                            deadline
556                                .saturating_duration_since(Instant::now())
557                                .min(LEASE_RENEWAL_INTERVAL),
558                        )
559                        .await;
560                    }
561                    continue;
562                }
563                attempts.push(report);
564                for field in [
565                    "outcome",
566                    "message",
567                    "exception_type",
568                    "non_retryable",
569                    "timeout_kind",
570                ] {
571                    if let Some(value) = attempts.last().unwrap().get(field) {
572                        wire[field] = value.clone();
573                    }
574                }
575                wire["attempts"] = json!(attempts);
576                // Match the terminal fields Server persists. Database-generated
577                // execution/failure IDs are only available on committed replay.
578                let mut payload = wire.clone();
579                payload["attempt_number"] = json!(number);
580                payload["failure_category"] = json!(if timeout_kind.is_some() {
581                    "timeout"
582                } else {
583                    "application"
584                });
585                if timeout_kind.is_some() {
586                    payload["exception_class"] = payload["exception_type"].take();
587                    payload.as_object_mut().unwrap().remove("exception_type");
588                    payload.as_object_mut().unwrap().remove("non_retryable");
589                }
590                let event = HistoryEvent {
591                    event_type: if timeout_kind.is_some() {
592                        "ActivityTimedOut"
593                    } else {
594                        "ActivityFailed"
595                    }
596                    .into(),
597                    payload,
598                    raw: HashMap::new(),
599                };
600                let result =
601                    activity_outcome(&event, &task.payload_codec, Some(activity_type.clone()))?;
602                if self
603                    .externalize_local_command(task, context, request.command_index, &mut wire)
604                    .await
605                    .is_err()
606                    || !self.local_claim_is_active(task).await
607                {
608                    return Ok(None);
609                }
610                return Ok(Some((wire, result)));
611            }
612            report["outcome"] = json!("completed");
613            attempts.push(report);
614            wire["outcome"] = json!("completed");
615            wire["attempts"] = json!(attempts);
616            if self
617                .externalize_local_command(task, context, request.command_index, &mut wire)
618                .await
619                .is_err()
620                || !self.local_claim_is_active(task).await
621            {
622                return Ok(None);
623            }
624            return Ok(Some((wire, Ok(result.unwrap()))));
625        }
626        unreachable!("validated local retry budget includes at least one attempt")
627    }
628}
629
630fn local_error(error: Error) -> (String, String, bool, Option<&'static str>) {
631    let (kind, non_retryable) = match &error {
632        Error::ActivityFailed(failure) => (
633            failure
634                .exception_type
635                .clone()
636                .unwrap_or_else(|| "RustActivityError".into()),
637            failure.non_retryable,
638        ),
639        Error::ActivityNotRegistered(_) => ("ActivityNotRegistered".into(), true),
640        Error::Codec(_) | Error::HandlerType { .. } => ("ActivityCodecError".into(), true),
641        _ => ("RustActivityError".into(), false),
642    };
643    if kind.len() > 255 {
644        return (
645            "local activity exception type exceeded 255 bytes".into(),
646            "LocalActivityFailureMetadataError".into(),
647            true,
648            None,
649        );
650    }
651    (error.to_string(), kind, non_retryable, None)
652}
653
654fn heartbeat_extended(reports: &Arc<Mutex<Heartbeats>>, timeout: Option<u64>) -> Result<bool> {
655    let reports = reports.lock().map_err(|_| Error::WorkflowStatePoisoned)?;
656    Ok(timeout.is_some_and(|seconds| reports.last + Duration::from_secs(seconds) > Instant::now()))
657}
durable_workflow/local_activity.rs

durable_workflow/
local_activity.rs
