From e8ad426d0136976f1a5a512168c473c741606015 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Tue, 20 Oct 2026 19:43:59 -0700 Subject: [PATCH 1/3] Added the previous completed and failed tasks to the interceptor info. A plugin that withdraws the output of failed workflow task attempts needs these History facts. Core already applies the events, so it hands them on. --- crates/sdk-core/CHANGELOG.md | 2 + .../core_tests/workflow_task_interceptor.rs | 193 +++++++++++++++++- crates/sdk-core/src/lib.rs | 4 +- crates/sdk-core/src/replay/history_builder.rs | 12 ++ crates/sdk-core/src/worker/extension.rs | 20 ++ crates/sdk-core/src/worker/mod.rs | 4 +- .../workflow/machines/workflow_machines.rs | 76 +++++++ .../src/worker/workflow/managed_run.rs | 5 + crates/sdk-core/src/worker/workflow/mod.rs | 1 + 9 files changed, 311 insertions(+), 6 deletions(-) diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index 3df9ceae48..6d24face63 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/crates/sdk-core/CHANGELOG.md @@ -39,6 +39,8 @@ relevant information. and the completion of output buffered before an eviction. An error from `before_complete`, or a hook that misses the task's deadline, fails the workflow task. The replay worker never calls the hooks. +* The info a `WorkflowTaskInterceptor` gets names the run's last completed workflow task and the + tasks that history shows as failed or timed out since then. ### Security * Ephemeral server downloads now use `zip` 9.0 for ZIP extraction, replacing the 8.x dependency diff --git a/crates/sdk-core/src/core_tests/workflow_task_interceptor.rs b/crates/sdk-core/src/core_tests/workflow_task_interceptor.rs index 3b21d5fa80..9a99ee4efe 100644 --- a/crates/sdk-core/src/core_tests/workflow_task_interceptor.rs +++ b/crates/sdk-core/src/core_tests/workflow_task_interceptor.rs @@ -1,7 +1,7 @@ //! The hooks a [WorkflowTaskInterceptor] gets around each workflow task completion. use crate::{ - WorkerExtension, WorkflowTaskCompletionInfo, WorkflowTaskFailedInfo, + WorkerExtension, WorkflowTaskCompletionInfo, WorkflowTaskEpoch, WorkflowTaskFailedInfo, WorkflowTaskInterceptError, WorkflowTaskInterceptor, init_replay_worker, replay::{HistoryForReplay, ReplayWorkerInput, TestHistoryBuilder, canned_histories}, test_help::{ @@ -29,7 +29,9 @@ use temporalio_common::protos::{ workflow_completion::WorkflowActivationCompletion, }, temporal::api::{ - enums::v1::EventType, workflowservice::v1::RespondWorkflowTaskCompletedResponse, + enums::v1::{EventType, TimeoutType, WorkflowTaskFailedCause}, + history::v1::{WorkflowTaskTimedOutEventAttributes, history_event}, + workflowservice::v1::RespondWorkflowTaskCompletedResponse, }, }; @@ -131,6 +133,51 @@ fn worker_with( mock_worker(mock) } +fn started_ids(epoch: &Option) -> Option { + epoch.as_ref().map(|e| e.started_event_id) +} + +/// Sets the attempt of the last WorkflowTaskScheduled event. +fn set_scheduled_attempt(t: &mut TestHistoryBuilder, attempt: i32) { + let scheduled = t.current_event_id() - 1; + t.modify_event(scheduled, |e| { + if let Some(history_event::Attributes::WorkflowTaskScheduledEventAttributes(a)) = + e.attributes.as_mut() + { + a.attempt = attempt; + } + }); +} + +/// Replays the task that starts the timer, then completes the run in the live task. Returns the +/// info of the live task. +async fn live_task_after_timer( + history: TestHistoryBuilder, + responses: Vec, +) -> WorkflowTaskCompletionInfo { + let recorder = Arc::new(Recorder::default()); + let worker = worker_with( + MockPollCfg::from_resp_batches("fakeid", history, responses, mock_worker_client()), + &recorder, + accept, + ); + let first = worker.poll_workflow_activation().await.unwrap(); + assert!(first.is_replaying); + complete( + &worker, + &first, + vec![start_timer_cmd(1, Duration::from_secs(1))], + ) + .await; + let fired = worker.poll_workflow_activation().await.unwrap(); + assert!(!fired.is_replaying); + complete_and_evict(&worker, &fired.run_id).await; + worker.drain_pollers_and_shutdown().await; + let mut infos = recorder.infos(); + assert_eq!(infos.len(), 1); + infos.remove(0) +} + fn accept(_: usize) -> Result { Ok(Default::default()) } @@ -404,6 +451,10 @@ async fn a_dropped_speculative_task_is_not_accepted() { "evicted", ] ); + // History never keeps the dropped task, so the next task follows the first one. + let infos = recorder.infos(); + assert_eq!(started_ids(&infos[2].previous_completed), Some(3)); + assert!(infos[2].failed_tasks.is_empty()); } #[tokio::test] @@ -466,6 +517,10 @@ async fn the_local_activity_heartbeat_completion_runs_the_hooks() { "evicted", ] ); + // The heartbeat completed the first task, so the next one follows it. + let infos = recorder.infos(); + assert_eq!(infos[0].previous_completed, None); + assert_eq!(started_ids(&infos[1].previous_completed), Some(3)); } #[tokio::test] @@ -579,3 +634,137 @@ async fn the_replay_worker_never_calls_the_hooks() { assert!(recorder.log().is_empty()); } + +#[tokio::test] +async fn a_failed_attempt_is_a_failed_task() { + let mut t = timer_then_complete(); + // Task 3 completed on its third attempt, which the server wrote when it completed. + t.modify_event(2, |e| { + if let Some(history_event::Attributes::WorkflowTaskScheduledEventAttributes(a)) = + e.attributes.as_mut() + { + a.attempt = 3; + } + }); + // Task 8 fails, and the server writes its retry as attempt 2. + t.add_workflow_task_failed_with_failure( + WorkflowTaskFailedCause::Unspecified, + Default::default(), + ); + t.add_workflow_task_scheduled_and_started(); + set_scheduled_attempt(&mut t, 2); + let info = live_task_after_timer(t, vec![ResponseType::AllHistory]).await; + + assert_eq!(info.started_event_id, 11); + let previous = info.previous_completed.unwrap(); + assert_eq!(previous.started_event_id, 3); + assert_eq!(previous.attempt, 3); + assert!(previous.started_time.is_some()); + assert_eq!(info.failed_tasks, [8]); +} + +#[tokio::test] +async fn timed_out_attempts_are_failed_tasks_unless_no_worker_started_them() { + let mut t = timer_then_complete(); + t.add_workflow_task_timed_out(); + // A sticky retry that no worker picks up times out before it starts. + t.add_workflow_task_scheduled(); + let scheduled = t.current_event_id(); + t.add(WorkflowTaskTimedOutEventAttributes { + scheduled_event_id: scheduled, + timeout_type: TimeoutType::ScheduleToStart as i32, + ..Default::default() + }); + t.add_workflow_task_scheduled_and_started(); + let info = live_task_after_timer(t, vec![ResponseType::AllHistory]).await; + + assert_eq!(info.started_event_id, 13); + assert_eq!(info.failed_tasks, [8]); +} + +#[tokio::test] +async fn a_refused_completion_shows_up_as_a_failed_task_of_the_retry() { + let mut t = timer_then_complete(); + let refused = hist_to_poll_resp(&t, "fakeid", ResponseType::AllHistory).resp; + // The server refuses task 8 with UNHANDLED_COMMAND and writes the failure. + t.add_workflow_task_failed_with_failure( + WorkflowTaskFailedCause::UnhandledCommand, + Default::default(), + ); + t.add_workflow_task_scheduled_and_started(); + let recorder = Arc::new(Recorder::default()); + let worker = worker_with( + MockPollCfg::from_resp_batches( + "fakeid", + t.clone(), + [ResponseType::Raw(refused), ResponseType::AllHistory], + mock_worker_client(), + ), + &recorder, + |n| { + if n == 1 { + Err(tonic::Status::invalid_argument("UnhandledCommand")) + } else { + Ok(Default::default()) + } + }, + ); + + let first = worker.poll_workflow_activation().await.unwrap(); + complete( + &worker, + &first, + vec![start_timer_cmd(1, Duration::from_secs(1))], + ) + .await; + let fired = worker.poll_workflow_activation().await.unwrap(); + worker.complete_execution(&fired.run_id).await; + let eviction = worker.poll_workflow_activation().await.unwrap(); + worker + .complete_workflow_activation(WorkflowActivationCompletion::empty(eviction.run_id)) + .await + .unwrap(); + // The retry replays the run from History. + let replay = worker.poll_workflow_activation().await.unwrap(); + assert!(replay.is_replaying); + complete( + &worker, + &replay, + vec![start_timer_cmd(1, Duration::from_secs(1))], + ) + .await; + let fired = worker.poll_workflow_activation().await.unwrap(); + complete_and_evict(&worker, &fired.run_id).await; + worker.drain_pollers_and_shutdown().await; + + assert_eq!( + recorder.log(), + [ + "before 8", + "sent commands=1 force_new=false", + "after 8 false", + "evicted", + "before 11", + "sent commands=1 force_new=false", + "after 11 true", + "evicted", + ] + ); + let infos = recorder.infos(); + assert!(infos[0].failed_tasks.is_empty()); + assert_eq!(started_ids(&infos[1].previous_completed), Some(3)); + assert_eq!(infos[1].failed_tasks, [8]); +} + +#[tokio::test] +async fn a_reset_shows_the_reset_task_as_failed() { + let mut t = timer_then_complete(); + // A reset to task 8 fails it in the new run and schedules a new task. + t.add_workflow_task_failed_new_id(WorkflowTaskFailedCause::ResetWorkflow, "new-run"); + t.add_workflow_task_scheduled_and_started(); + let info = live_task_after_timer(t, vec![ResponseType::AllHistory]).await; + + assert_eq!(info.started_event_id, 11); + assert_eq!(started_ids(&info.previous_completed), Some(3)); + assert_eq!(info.failed_tasks, [8]); +} diff --git a/crates/sdk-core/src/lib.rs b/crates/sdk-core/src/lib.rs index 3e2b6e27e9..db9075f9bb 100644 --- a/crates/sdk-core/src/lib.rs +++ b/crates/sdk-core/src/lib.rs @@ -51,8 +51,8 @@ pub use worker::{ WorkflowErrorType, WorkflowSlotKind, }; pub use worker::{ - RunCloseKind, WorkerExtension, WorkflowTaskCompletionInfo, WorkflowTaskFailedInfo, - WorkflowTaskInterceptError, WorkflowTaskInterceptor, + RunCloseKind, WorkerExtension, WorkflowTaskCompletionInfo, WorkflowTaskEpoch, + WorkflowTaskFailedInfo, WorkflowTaskInterceptError, WorkflowTaskInterceptor, }; use crate::{ diff --git a/crates/sdk-core/src/replay/history_builder.rs b/crates/sdk-core/src/replay/history_builder.rs index 51472482c9..3b01865fc3 100644 --- a/crates/sdk-core/src/replay/history_builder.rs +++ b/crates/sdk-core/src/replay/history_builder.rs @@ -129,10 +129,20 @@ impl TestHistoryBuilder { self.previous_task_completed_id = id; } + /// The started event id of the last scheduled workflow task, or zero if it has not started. + fn started_id_of_current_task(&self) -> i64 { + if self.final_workflow_task_started_event_id > self.workflow_task_scheduled_event_id { + self.final_workflow_task_started_event_id + } else { + 0 + } + } + /// Add a workflow task timed out event. pub fn add_workflow_task_timed_out(&mut self) { let attrs = WorkflowTaskTimedOutEventAttributes { scheduled_event_id: self.workflow_task_scheduled_event_id, + started_event_id: self.started_id_of_current_task(), ..Default::default() }; self.build_and_push_event(EventType::WorkflowTaskTimedOut, attrs.into()); @@ -243,6 +253,7 @@ impl TestHistoryBuilder { ) { let attrs = WorkflowTaskFailedEventAttributes { scheduled_event_id: self.workflow_task_scheduled_event_id, + started_event_id: self.started_id_of_current_task(), cause: cause.into(), failure: Some(failure), ..Default::default() @@ -258,6 +269,7 @@ impl TestHistoryBuilder { ) { let attrs = WorkflowTaskFailedEventAttributes { scheduled_event_id: self.workflow_task_scheduled_event_id, + started_event_id: self.started_id_of_current_task(), cause: cause.into(), new_run_id: new_run_id.into(), ..Default::default() diff --git a/crates/sdk-core/src/worker/extension.rs b/crates/sdk-core/src/worker/extension.rs index 1f1993e02d..45d7521d89 100644 --- a/crates/sdk-core/src/worker/extension.rs +++ b/crates/sdk-core/src/worker/extension.rs @@ -74,6 +74,14 @@ pub struct WorkflowTaskCompletionInfo { /// The time the server started the task, from the poll response. Unset if the server did not /// send it. pub started_time: Option, + /// The last task of the run that completed, from the poll response's + /// `previous_started_event_id` and that task's events. Unset before the first completion. + pub previous_completed: Option, + /// The started event ids of the tasks that History shows as failed or timed out since + /// `previous_completed`, oldest first. Attempts after the first that fail leave no event, so + /// they are not here. + #[builder(default)] + pub failed_tasks: Vec, /// Set when the completion has a command that closes the run. Continue-as-new does not /// count. pub closes_run: Option, @@ -82,6 +90,18 @@ pub struct WorkflowTaskCompletionInfo { pub deadline: Instant, } +/// One attempt of a workflow task, as History records it. +#[derive(Clone, Debug, PartialEq, Eq, bon::Builder)] +#[non_exhaustive] +pub struct WorkflowTaskEpoch { + /// The id of the task's `WorkflowTaskStarted` event. + pub started_event_id: i64, + /// The attempt of the task's `WorkflowTaskScheduled` event. Zero if Core no longer has it. + pub attempt: u32, + /// The time of the task's `WorkflowTaskStarted` event. + pub started_time: Option, +} + /// The command with which a completion closes its run. #[derive(Clone, Copy, Debug, PartialEq, Eq)] #[non_exhaustive] diff --git a/crates/sdk-core/src/worker/mod.rs b/crates/sdk-core/src/worker/mod.rs index 085b54ffaa..84701de3c1 100644 --- a/crates/sdk-core/src/worker/mod.rs +++ b/crates/sdk-core/src/worker/mod.rs @@ -12,8 +12,8 @@ mod workflow; pub(crate) const PAYLOADS_TOO_LARGE_FAILURE_TYPE: &str = "PayloadsTooLarge"; pub use extension::{ - RunCloseKind, WorkerExtension, WorkflowTaskCompletionInfo, WorkflowTaskFailedInfo, - WorkflowTaskInterceptError, WorkflowTaskInterceptor, + RunCloseKind, WorkerExtension, WorkflowTaskCompletionInfo, WorkflowTaskEpoch, + WorkflowTaskFailedInfo, WorkflowTaskInterceptError, WorkflowTaskInterceptor, }; use temporalio_client::{Connection, PayloadErrorLimits, worker::NamespaceDescriptionSource}; use temporalio_common::{ 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 e26aabd7f9..b76ff03cdf 100644 --- a/crates/sdk-core/src/worker/workflow/machines/workflow_machines.rs +++ b/crates/sdk-core/src/worker/workflow/machines/workflow_machines.rs @@ -23,6 +23,7 @@ use crate::{ telemetry::{VecDisplayer, metrics::MetricsContext}, worker::{ ExecutingLAId, LocalActRequest, LocalActivityExecutionResult, LocalActivityResolution, + WorkflowTaskEpoch, workflow::{ CommandAnnotations, CommandID, DrivenWorkflow, HistoryUpdate, InternalFlagsRef, LocalResolution, OutgoingJob, ProtoCommandExt, RunBasics, WFCommand, WFCommandVariant, @@ -77,6 +78,54 @@ use temporalio_common::{ type Result = std::result::Result; +/// The workflow task events of History that a workflow task interceptor needs. +#[derive(Default)] +struct TaskHistory { + /// The attempt of the last WorkflowTaskScheduled event. + scheduled_attempt: u32, + /// The task of the last WorkflowTaskStarted event. + started: Option, + /// The task of the last WorkflowTaskCompleted event. + completed: Option, + /// Started event ids of the failed and timed-out tasks since `completed`. + failed: Vec, +} + +impl TaskHistory { + fn apply(&mut self, event: &HistoryEvent) { + match &event.attributes { + Some(history_event::Attributes::WorkflowTaskScheduledEventAttributes(a)) => { + self.scheduled_attempt = u32::try_from(a.attempt).unwrap_or_default(); + } + Some(history_event::Attributes::WorkflowTaskStartedEventAttributes(_)) => { + self.started = Some( + WorkflowTaskEpoch::builder() + .started_event_id(event.event_id) + .attempt(self.scheduled_attempt) + .maybe_started_time(event.event_time.and_then(|t| t.try_into().ok())) + .build(), + ); + } + Some(history_event::Attributes::WorkflowTaskCompletedEventAttributes(_)) => { + self.completed = self.started.clone(); + self.failed.clear(); + } + Some(history_event::Attributes::WorkflowTaskFailedEventAttributes(a)) + if a.started_event_id > 0 => + { + self.failed.push(a.started_event_id); + } + // A schedule-to-start timeout has no started event. + Some(history_event::Attributes::WorkflowTaskTimedOutEventAttributes(a)) + if a.started_event_id > 0 => + { + self.failed.push(a.started_event_id); + } + _ => {} + } + } +} + slotmap::new_key_type! { struct MachineKey; } /// 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 @@ -90,6 +139,8 @@ pub(crate) struct WorkflowMachines { protocol_msgs: Vec, /// EventId of the last handled WorkflowTaskStarted event current_started_event_id: i64, + /// The workflow task facts that History shows, for a workflow task interceptor. + task_history: TaskHistory, /// The event id of the next workflow task started event that the machines need to process. /// Eventually, this number should reach the started id in the latest history update, but /// we must incrementally apply the history while communicating with lang. @@ -287,6 +338,7 @@ impl WorkflowMachines { metrics: basics.metrics, // In an ideal world one could say ..Default::default() here and it'd still work. current_started_event_id: 0, + task_history: Default::default(), next_started_event_id: 0, last_processed_event: 0, workflow_start_time: None, @@ -413,6 +465,29 @@ impl WorkflowMachines { self.drive_me.get_started_info() } + /// The completed task with the given started event id, and the failed tasks since the last + /// completed task. + pub(crate) fn task_history_facts( + &self, + previous_started_event_id: i64, + ) -> (Option, Vec) { + let completed = self.task_history.completed.as_ref(); + let previous = (previous_started_event_id > 0).then(|| match completed { + Some(epoch) if epoch.started_event_id == previous_started_event_id => epoch.clone(), + _ => { + warn!( + previous_started_event_id, + "History has no completed workflow task with this started event" + ); + WorkflowTaskEpoch::builder() + .started_event_id(previous_started_event_id) + .attempt(0) + .build() + } + }); + (previous, self.task_history.failed.clone()) + } + pub(crate) fn get_last_wft_started_id(&self) -> i64 { self.current_started_event_id } @@ -716,6 +791,7 @@ impl WorkflowMachines { eid )); } + self.task_history.apply(&event); let next_event = history.peek(); // This definition of replaying here is that we are no longer replaying as soon as we diff --git a/crates/sdk-core/src/worker/workflow/managed_run.rs b/crates/sdk-core/src/worker/workflow/managed_run.rs index 05392f81ad..53d9471df8 100644 --- a/crates/sdk-core/src/worker/workflow/managed_run.rs +++ b/crates/sdk-core/src/worker/workflow/managed_run.rs @@ -200,6 +200,7 @@ impl ManagedRun { let wft_info = WorkflowTaskInfo { attempt: work.attempt, started_event_id: work.update.wft_started_id, + previous_started_event_id: work.update.previous_wft_started_id, started_time: work.started_time, task_token: work.task_token, wf_id: work.execution.workflow_id.clone(), @@ -1299,6 +1300,8 @@ impl ManagedRun { let wft = self.wft.as_ref()?; let machines = &self.wfm.machines; let started_info = machines.get_started_info(); + let (previous_completed, failed_tasks) = + machines.task_history_facts(wft.info.previous_started_event_id); let wft_timeout = started_info .and_then(|i| i.workflow_task_timeout) .unwrap_or(DEFAULT_WFT_TIMEOUT); @@ -1316,6 +1319,8 @@ impl ManagedRun { .started_event_id(wft.info.started_event_id) .attempt(wft.info.attempt) .maybe_started_time(wft.info.started_time) + .maybe_previous_completed(previous_completed) + .failed_tasks(failed_tasks) .maybe_closes_run(run_close_kind(commands)) .deadline(wft.start_time + wft_timeout.mul_f32(WFT_INTERCEPTOR_DEADLINE_FRACTION)) .build(), diff --git a/crates/sdk-core/src/worker/workflow/mod.rs b/crates/sdk-core/src/worker/workflow/mod.rs index 5c95f07702..64e009d25f 100644 --- a/crates/sdk-core/src/worker/workflow/mod.rs +++ b/crates/sdk-core/src/worker/workflow/mod.rs @@ -1158,6 +1158,7 @@ struct WorkflowTaskInfo { attempt: u32, /// From the poll response, like the attempt. started_event_id: i64, + previous_started_event_id: i64, started_time: Option, /// 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 From 99e8cb4861448cc3c1a2b4da70aebcbdb83f6e72 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Tue, 20 Oct 2026 20:02:15 -0700 Subject: [PATCH 2/3] Listed every failed workflow task of the run in the interceptor info. A plugin withdraws a failed task's output at its next write to the same stream, which can come many completed tasks later. A superset of the failed tasks is always safe. --- crates/sdk-core/CHANGELOG.md | 2 +- .../core_tests/workflow_task_interceptor.rs | 59 +++++++++++++++++++ crates/sdk-core/src/worker/extension.rs | 7 ++- .../workflow/machines/workflow_machines.rs | 6 +- 4 files changed, 66 insertions(+), 8 deletions(-) diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index 453d98b99a..b53a119a74 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/crates/sdk-core/CHANGELOG.md @@ -40,7 +40,7 @@ relevant information. a hook that misses the task's deadline, fails the workflow task. The replay worker never calls the hooks. * The info a `WorkflowTaskInterceptor` gets names the run's last completed workflow task and the - tasks that history shows as failed or timed out since then. + tasks of the run that history shows as failed or timed out. a hook that misses the task's deadline, fails the workflow task. Core waits at most five seconds on `after_complete`. The replay worker never calls the hooks. diff --git a/crates/sdk-core/src/core_tests/workflow_task_interceptor.rs b/crates/sdk-core/src/core_tests/workflow_task_interceptor.rs index 41a72c91a8..29a8a7a5f4 100644 --- a/crates/sdk-core/src/core_tests/workflow_task_interceptor.rs +++ b/crates/sdk-core/src/core_tests/workflow_task_interceptor.rs @@ -815,3 +815,62 @@ async fn a_hung_after_complete_does_not_stall_the_run() { ] ); } + +#[tokio::test] +async fn a_failed_task_before_an_earlier_completion_stays_listed() { + let mut t = TestHistoryBuilder::default(); + t.add_wfe_started_with_wft_timeout(Duration::from_secs(10)); + // Task 3 fails, and task 6 completes without touching what task 3 wrote. + t.add_workflow_task_scheduled_and_started(); + t.add_workflow_task_failed_with_failure( + WorkflowTaskFailedCause::Unspecified, + Default::default(), + ); + t.add_full_wf_task(); + let timer_started = t.add_by_type(EventType::TimerStarted); + t.add_timer_fired(timer_started, "1".to_string()); + t.add_workflow_task_scheduled_and_started(); + let info = live_task_after_timer(t, vec![ResponseType::AllHistory]).await; + + assert_eq!(info.started_event_id, 11); + assert_eq!(started_ids(&info.previous_completed), Some(6)); + assert_eq!(info.failed_tasks, [3]); +} + +#[tokio::test] +async fn a_sticky_task_after_a_failure_lists_it() { + let mut t = timer_then_complete(); + // Task 8 times out while the run stays cached, so the next task is a partial history. + t.add_workflow_task_timed_out(); + t.add_workflow_task_scheduled_and_started(); + let recorder = Arc::new(Recorder::default()); + let worker = worker_with( + MockPollCfg::from_resp_batches( + "fakeid", + t, + [ResponseType::ToTaskNum(1), ResponseType::OneTask(2)], + mock_worker_client(), + ), + &recorder, + accept, + ); + + let first = worker.poll_workflow_activation().await.unwrap(); + complete( + &worker, + &first, + vec![start_timer_cmd(1, Duration::from_secs(1))], + ) + .await; + let fired = worker.poll_workflow_activation().await.unwrap(); + assert!(!fired.is_replaying); + complete_and_evict(&worker, &fired.run_id).await; + worker.drain_pollers_and_shutdown().await; + + let infos = recorder.infos(); + assert_eq!(infos.len(), 2); + let last = &infos[1]; + assert_eq!(last.started_event_id, 11); + assert_eq!(started_ids(&last.previous_completed), Some(3)); + assert_eq!(last.failed_tasks, [8]); +} diff --git a/crates/sdk-core/src/worker/extension.rs b/crates/sdk-core/src/worker/extension.rs index 45bb862c5c..522a807df5 100644 --- a/crates/sdk-core/src/worker/extension.rs +++ b/crates/sdk-core/src/worker/extension.rs @@ -82,9 +82,10 @@ pub struct WorkflowTaskCompletionInfo { /// The last task of the run that completed, from the poll response's /// `previous_started_event_id` and that task's events. Unset before the first completion. pub previous_completed: Option, - /// The started event ids of the tasks that History shows as failed or timed out since - /// `previous_completed`, oldest first. Attempts after the first that fail leave no event, so - /// they are not here. + /// The started event ids of all the tasks of the run that History shows as failed or timed + /// out, oldest first. It includes the task a reset failed, which belongs to the reset run. + /// Attempts after the first that fail leave no event, so they are not here. The list is the + /// same after a cache miss, since replay applies all of History. #[builder(default)] pub failed_tasks: Vec, /// Set when the completion has a command that closes the run. Continue-as-new does not 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 b76ff03cdf..cd839388b7 100644 --- a/crates/sdk-core/src/worker/workflow/machines/workflow_machines.rs +++ b/crates/sdk-core/src/worker/workflow/machines/workflow_machines.rs @@ -87,7 +87,7 @@ struct TaskHistory { started: Option, /// The task of the last WorkflowTaskCompleted event. completed: Option, - /// Started event ids of the failed and timed-out tasks since `completed`. + /// Started event ids of all the failed and timed-out tasks of the run. failed: Vec, } @@ -108,7 +108,6 @@ impl TaskHistory { } Some(history_event::Attributes::WorkflowTaskCompletedEventAttributes(_)) => { self.completed = self.started.clone(); - self.failed.clear(); } Some(history_event::Attributes::WorkflowTaskFailedEventAttributes(a)) if a.started_event_id > 0 => @@ -465,8 +464,7 @@ impl WorkflowMachines { self.drive_me.get_started_info() } - /// The completed task with the given started event id, and the failed tasks since the last - /// completed task. + /// The completed task with the given started event id, and all the failed tasks of the run. pub(crate) fn task_history_facts( &self, previous_started_event_id: i64, From 128574c58d826f432c07b8c1bb5c2b4205bc588f Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Tue, 20 Oct 2026 20:39:43 -0700 Subject: [PATCH 3/3] Removed stray merge lines from the changelog. --- crates/sdk-core/CHANGELOG.md | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index b53a119a74..220ba994e1 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/crates/sdk-core/CHANGELOG.md @@ -35,14 +35,12 @@ relevant information. ### Added * A worker can take a `WorkerExtension` whose `WorkflowTaskInterceptor` runs before and after - each workflow task completion Core sends, including the local activity heartbeat completion + each Workflow Task completion Core sends, including the Local Activity heartbeat completion and the completion of output buffered before an eviction. An error from `before_complete`, or - a hook that misses the task's deadline, fails the workflow task. The replay worker never calls - the hooks. -* The info a `WorkflowTaskInterceptor` gets names the run's last completed workflow task and the - tasks of the run that history shows as failed or timed out. - a hook that misses the task's deadline, fails the workflow task. Core waits at most five + a hook that misses the task's deadline, fails the Workflow Task. Core waits at most five seconds on `after_complete`. The replay worker never calls the hooks. +* The info a `WorkflowTaskInterceptor` gets has the run's last completed Workflow Task and the + tasks of the run that history shows as failed or timed out. ### Security * Ephemeral server downloads now use `zip` 9.0 for ZIP extraction, replacing the 8.x dependency