diff --git a/crates/protos/protos/local/temporal/sdk/core/activity_task/activity_task.proto b/crates/protos/protos/local/temporal/sdk/core/activity_task/activity_task.proto index ca2540424..d0cd45fab 100644 --- a/crates/protos/protos/local/temporal/sdk/core/activity_task/activity_task.proto +++ b/crates/protos/protos/local/temporal/sdk/core/activity_task/activity_task.proto @@ -77,6 +77,12 @@ message WorkflowTaskAttempt { uint32 attempt = 2; // When the server started the task. google.protobuf.Timestamp started_time = 3; + // The started event id of the run's last completed workflow task, from the poll response. + // 0 when no task of the run has completed. + int64 previous_started_event_id = 4; + // The started event ids of the run's failed and timed out workflow tasks in history, in event + // order. It holds the last 100 only, to keep each start small. + repeated int64 failed_tasks = 5; } // Attempt to cancel a running activity diff --git a/crates/sdk-core/src/core_tests/local_activity_workflow_task.rs b/crates/sdk-core/src/core_tests/local_activity_workflow_task.rs index 8ad9dc705..da3732cb1 100644 --- a/crates/sdk-core/src/core_tests/local_activity_workflow_task.rs +++ b/crates/sdk-core/src/core_tests/local_activity_workflow_task.rs @@ -111,6 +111,8 @@ async fn a_local_activity_names_the_live_task_that_dispatched_it() { started_event_id: 8, attempt: 1, started_time: Some(started_time.into()), + previous_started_event_id: 3, + failed_tasks: vec![], }) ); finish(worker, la).await; @@ -135,6 +137,10 @@ async fn a_rerun_after_a_timed_out_task_names_the_later_task() { // History has no marker for it, so it runs again, dispatched by the live task. let la = worker.poll_activity_task().await.unwrap(); assert_eq!(dispatching(start_of(&la)), (9, 1)); + // The start also names the last completed task and the task that timed out. + let task = start_of(&la).workflow_task.as_ref().unwrap(); + assert_eq!(task.previous_started_event_id, 3); + assert_eq!(task.failed_tasks, [6]); finish(worker, la).await; } diff --git a/crates/sdk-core/src/worker/workflow/machines/workflow_machines.rs b/crates/sdk-core/src/worker/workflow/machines/workflow_machines.rs index e26aabd7f..cd31ec838 100644 --- a/crates/sdk-core/src/worker/workflow/machines/workflow_machines.rs +++ b/crates/sdk-core/src/worker/workflow/machines/workflow_machines.rs @@ -78,6 +78,10 @@ use temporalio_common::{ type Result = std::result::Result; slotmap::new_key_type! { struct MachineKey; } + +/// The most failed workflow tasks that a local activity start names. +const MAX_FAILED_WFTS: usize = 100; + /// Handles all the logic for driving a workflow. It orchestrates many state machines that together /// comprise the logic of an executing workflow. One instance will exist per currently executing /// (or cached) workflow on the worker. @@ -86,6 +90,9 @@ pub(crate) struct WorkflowMachines { /// kept because the lang side polls & completes for every workflow task, but we do not need /// to poll the server that often during replay. last_history_from_server: HistoryUpdate, + /// The started event ids of the run's failed and timed out workflow tasks, for the local + /// activities that need to know which tasks never completed. + failed_wft_started_ids: VecDeque, /// Protocol messages that have yet to be processed for the current WFT. protocol_msgs: Vec, /// EventId of the last handled WorkflowTaskStarted event @@ -276,6 +283,7 @@ impl WorkflowMachines { }; Self { last_history_from_server: basics.history, + failed_wft_started_ids: VecDeque::new(), protocol_msgs: vec![], workflow_id: basics.workflow_id, workflow_type: basics.workflow_type, @@ -345,6 +353,7 @@ impl WorkflowMachines { } pub(crate) fn new_history_from_server(&mut self, update: HistoryUpdate) -> Result<()> { + self.note_failed_wfts(update.get_events()); self.last_history_from_server = update; self.replaying = self.preparing_legacy_query || self.last_history_from_server.previous_wft_started_id > 0; @@ -417,6 +426,38 @@ impl WorkflowMachines { self.current_started_event_id } + pub(crate) fn failed_wft_started_ids(&self) -> Vec { + self.failed_wft_started_ids.iter().copied().collect() + } + + /// Only the last ones are kept, which bounds the list on a long run. A local activity's task + /// facts only need the failed tasks since its run's last write to a Stream. + fn note_failed_wfts(&mut self, events: &[HistoryEvent]) { + for event in events { + let started = match &event.attributes { + Some(history_event::Attributes::WorkflowTaskFailedEventAttributes(a)) => { + a.started_event_id + } + Some(history_event::Attributes::WorkflowTaskTimedOutEventAttributes(a)) => { + a.started_event_id + } + _ => continue, + }; + // A page of history may come again after a cache miss. + if self + .failed_wft_started_ids + .back() + .is_some_and(|&last| last >= started) + { + continue; + } + if self.failed_wft_started_ids.len() == MAX_FAILED_WFTS { + self.failed_wft_started_ids.pop_front(); + } + self.failed_wft_started_ids.push_back(started); + } + } + pub(crate) fn history_size_bytes(&self) -> u64 { self.history_size_bytes } diff --git a/crates/sdk-core/src/worker/workflow/managed_run.rs b/crates/sdk-core/src/worker/workflow/managed_run.rs index 61c79467e..0c8d2aa24 100644 --- a/crates/sdk-core/src/worker/workflow/managed_run.rs +++ b/crates/sdk-core/src/worker/workflow/managed_run.rs @@ -197,6 +197,7 @@ impl ManagedRun { attempt: work.attempt, started_event_id: work.update.wft_started_id, started_time: work.started_time, + previous_started_event_id: work.update.previous_wft_started_id, task_token: work.task_token, wf_id: work.execution.workflow_id.clone(), }; @@ -1292,10 +1293,12 @@ impl ManagedRun { started_event_id: wft.info.started_event_id, attempt: wft.info.attempt, started_time: wft.info.started_time.map(Into::into), + previous_started_event_id: wft.info.previous_started_event_id, + failed_tasks: self.wfm.machines.failed_wft_started_ids(), }; for request in requests { if let LocalActRequest::New(new_la) = request { - new_la.workflow_task = Some(task); + new_la.workflow_task = Some(task.clone()); } } } diff --git a/crates/sdk-core/src/worker/workflow/mod.rs b/crates/sdk-core/src/worker/workflow/mod.rs index eea2edae2..c19d693ce 100644 --- a/crates/sdk-core/src/worker/workflow/mod.rs +++ b/crates/sdk-core/src/worker/workflow/mod.rs @@ -1095,6 +1095,8 @@ struct WorkflowTaskInfo { /// From the poll response, like the attempt. started_event_id: i64, started_time: Option, + /// The started event id of the run's last completed task, from the poll response. + previous_started_event_id: i64, /// Exists to allow easy tagging of spans with workflow ids. Is duplicative of info inside the /// run machines themselves, but that can't be accessed easily. Would be nice to somehow have a /// shared repository, or refcounts, or whatever, for strings like these that get duped all