diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index 18eddd03d..220ba994e 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/crates/sdk-core/CHANGELOG.md @@ -35,10 +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. 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 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 1e2b38be0..29a8a7a5f 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, }, }; @@ -136,6 +138,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()) } @@ -409,6 +456,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] @@ -471,6 +522,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] @@ -585,6 +640,140 @@ 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]); +} + #[tokio::test] async fn a_hung_after_complete_does_not_stall_the_run() { let recorder = Arc::new(Recorder { @@ -626,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/lib.rs b/crates/sdk-core/src/lib.rs index 3e2b6e27e..db9075f9b 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 51472482c..3b01865fc 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 8d948449a..522a807df 100644 --- a/crates/sdk-core/src/worker/extension.rs +++ b/crates/sdk-core/src/worker/extension.rs @@ -79,6 +79,15 @@ 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 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 /// count. pub closes_run: Option, @@ -87,6 +96,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 085b54ffa..84701de3c 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 e26aabd7f..cd839388b 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,53 @@ 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 all the failed and timed-out tasks of the run. + 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(); + } + 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 +138,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 +337,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 +464,28 @@ impl WorkflowMachines { self.drive_me.get_started_info() } + /// 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, + ) -> (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 +789,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 b35089cfc..160b493c9 100644 --- a/crates/sdk-core/src/worker/workflow/managed_run.rs +++ b/crates/sdk-core/src/worker/workflow/managed_run.rs @@ -201,6 +201,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, received: work.received, task_token: work.task_token, @@ -1304,6 +1305,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); @@ -1321,6 +1324,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.info.received + wft_timeout.mul_f32(WFT_INTERCEPTOR_DEADLINE_FRACTION), diff --git a/crates/sdk-core/src/worker/workflow/mod.rs b/crates/sdk-core/src/worker/workflow/mod.rs index b833ff9bf..f0e82a460 100644 --- a/crates/sdk-core/src/worker/workflow/mod.rs +++ b/crates/sdk-core/src/worker/workflow/mod.rs @@ -1165,6 +1165,7 @@ struct WorkflowTaskInfo { attempt: u32, /// From the poll response, like the attempt. started_event_id: i64, + previous_started_event_id: i64, started_time: Option, received: Instant, /// Exists to allow easy tagging of spans with workflow ids. Is duplicative of info inside the