From b5b703ec8ae4b3c175790219d9cd98ce101860e7 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Thu, 22 Oct 2026 10:55:33 -0700 Subject: [PATCH] Removed the Workflow Task hooks that no extension uses. --- crates/sdk-core/CHANGELOG.md | 8 +- crates/sdk-core/src/core_tests/mod.rs | 1 - .../core_tests/workflow_task_interceptor.rs | 876 ------------------ crates/sdk-core/src/lib.rs | 5 +- crates/sdk-core/src/protosext/mod.rs | 5 +- crates/sdk-core/src/worker/extension.rs | 152 +-- crates/sdk-core/src/worker/mod.rs | 8 +- .../src/worker/workflow/driven_workflow.rs | 1 - .../src/worker/workflow/history_update.rs | 1 - .../workflow/machines/workflow_machines.rs | 74 -- .../src/worker/workflow/managed_run.rs | 68 +- crates/sdk-core/src/worker/workflow/mod.rs | 127 +-- .../src/worker/workflow/workflow_stream.rs | 9 - 13 files changed, 16 insertions(+), 1319 deletions(-) delete mode 100644 crates/sdk-core/src/core_tests/workflow_task_interceptor.rs diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index fd7d744153..8f4a809b02 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/crates/sdk-core/CHANGELOG.md @@ -34,13 +34,7 @@ relevant information. ## Unreleased ### 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 - 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 - 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. +* A worker can take a `WorkerExtension`, a plugin component that the replay worker leaves out. * A `WorkerExtension` has `call` and `call_async`, which take a method name and encoded input. A bridge can route a generic lang call to them, so the extension needs no bridge function of its own. Core never calls them. diff --git a/crates/sdk-core/src/core_tests/mod.rs b/crates/sdk-core/src/core_tests/mod.rs index ad8c89a753..d7e0db63b5 100644 --- a/crates/sdk-core/src/core_tests/mod.rs +++ b/crates/sdk-core/src/core_tests/mod.rs @@ -6,7 +6,6 @@ mod replay_flag; mod updates; mod workers; mod workflow_cancels; -mod workflow_task_interceptor; mod workflow_tasks; use crate::{ diff --git a/crates/sdk-core/src/core_tests/workflow_task_interceptor.rs b/crates/sdk-core/src/core_tests/workflow_task_interceptor.rs deleted file mode 100644 index 29a8a7a5f4..0000000000 --- a/crates/sdk-core/src/core_tests/workflow_task_interceptor.rs +++ /dev/null @@ -1,876 +0,0 @@ -//! The hooks a [WorkflowTaskInterceptor] gets around each workflow task completion. - -use crate::{ - WorkerExtension, WorkflowTaskCompletionInfo, WorkflowTaskEpoch, WorkflowTaskFailedInfo, - WorkflowTaskInterceptError, WorkflowTaskInterceptor, init_replay_worker, - replay::{HistoryForReplay, ReplayWorkerInput, TestHistoryBuilder, canned_histories}, - test_help::{ - MockPollCfg, PollWFTRespExt, ResponseType, WorkerExt, WorkerTestHelpers, - build_mock_pollers, hist_to_poll_resp, mock_worker, schedule_local_activity_cmd, - start_timer_cmd, test_worker_cfg, - }, - worker::client::mocks::mock_worker_client, -}; -use futures_util::stream; -use std::{ - sync::{Arc, Mutex}, - time::{Duration, Instant}, -}; -use temporalio_common::protos::{ - coresdk::{ - ActivityTaskCompletion, - activity_result::ActivityExecutionResult, - activity_task::activity_task, - workflow_activation::{WorkflowActivation, workflow_activation_job}, - workflow_commands::{ - ActivityCancellationType, CompleteWorkflowExecution, UpdateResponse, - update_response::Response, workflow_command, - }, - workflow_completion::WorkflowActivationCompletion, - }, - temporal::api::{ - enums::v1::{EventType, TimeoutType, WorkflowTaskFailedCause}, - history::v1::{WorkflowTaskTimedOutEventAttributes, history_event}, - workflowservice::v1::RespondWorkflowTaskCompletedResponse, - }, -}; - -/// What the hooks and the mock server saw, in order. -#[derive(Default)] -struct Recorder { - log: Mutex>, - infos: Mutex>, - /// `before_complete` fails for the tasks with these started event ids. - fail_before: Vec, - /// `before_complete` waits this long first. - before_delay: Option, - /// `after_complete` never returns for the tasks with these started event ids. - hang_after: Vec, -} - -impl Recorder { - fn push(&self, entry: impl Into) { - self.log.lock().unwrap().push(entry.into()); - } - - fn log(&self) -> Vec { - self.log.lock().unwrap().clone() - } - - fn infos(&self) -> Vec { - self.infos.lock().unwrap().clone() - } -} - -#[async_trait::async_trait] -impl WorkflowTaskInterceptor for Recorder { - async fn before_complete( - &self, - task: &WorkflowTaskCompletionInfo, - ) -> Result<(), WorkflowTaskInterceptError> { - self.push(format!("before {}", task.started_event_id)); - self.infos.lock().unwrap().push(task.clone()); - if let Some(delay) = self.before_delay { - tokio::time::sleep(delay).await; - } - if self.fail_before.contains(&task.started_event_id) { - return Err(WorkflowTaskInterceptError::new("store down")); - } - Ok(()) - } - - async fn after_complete(&self, task: &WorkflowTaskCompletionInfo, accepted: bool) { - self.push(format!("after {} {accepted}", task.started_event_id)); - if self.hang_after.contains(&task.started_event_id) { - futures_util::future::pending::<()>().await; - } - } - - fn on_task_failed(&self, task: &WorkflowTaskFailedInfo) { - self.push(format!("failed {} {}", task.attempt, task.reported)); - } - - fn on_evicted(&self, _run_id: &str) { - self.push("evicted"); - } -} - -struct Extension(Arc); - -impl WorkerExtension for Extension { - fn name(&self) -> &str { - "recorder" - } - - fn workflow_task_interceptor(&self) -> Option> { - Some(self.0.clone()) - } -} - -/// Builds a worker with the recorder as its interceptor. The mock server logs each completion it -/// gets, and answers with `answer`. -fn worker_with( - mut cfg: MockPollCfg, - recorder: &Arc, - mut answer: impl FnMut(usize) -> Result - + Send - + 'static, -) -> crate::Worker { - // An empty poll stream would start shutdown before the run is evicted. - cfg.make_poll_stream_interminable = true; - let log = recorder.clone(); - let mut completions = 0; - cfg.completion_mock_fn = Some(Box::new(move |c| { - completions += 1; - log.push(format!( - "sent commands={} force_new={}", - c.commands.len(), - c.force_create_new_workflow_task - )); - answer(completions) - })); - let mut mock = build_mock_pollers(cfg); - let extension: Arc = Arc::new(Extension(recorder.clone())); - mock.worker_cfg(|wc| { - wc.max_cached_workflows = 1; - wc.extension = Some(extension); - }); - 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()) -} - -async fn complete( - worker: &crate::Worker, - activation: &WorkflowActivation, - commands: Vec, -) { - worker - .complete_workflow_activation(WorkflowActivationCompletion::from_cmds( - activation.run_id.clone(), - commands, - )) - .await - .unwrap(); -} - -/// Completes the run, then answers the eviction that follows. -async fn complete_and_evict(worker: &crate::Worker, run_id: &str) { - worker.complete_execution(run_id).await; - let eviction = worker.poll_workflow_activation().await.unwrap(); - assert_matches!( - eviction.jobs[0].variant, - Some(workflow_activation_job::Variant::RemoveFromCache(_)) - ); - worker - .complete_workflow_activation(WorkflowActivationCompletion::empty(eviction.run_id)) - .await - .unwrap(); -} - -fn timer_then_complete() -> TestHistoryBuilder { - let mut t = TestHistoryBuilder::default(); - t.add_wfe_started_with_wft_timeout(Duration::from_secs(10)); - 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(); - t -} - -#[tokio::test] -async fn hooks_run_before_and_after_each_completion() { - let recorder = Arc::new(Recorder::default()); - let worker = worker_with( - MockPollCfg::from_resp_batches( - "fakeid", - timer_then_complete(), - [1, 2], - mock_worker_client(), - ), - &recorder, - accept, - ); - let first = worker.poll_workflow_activation().await.unwrap(); - let received = Instant::now(); - complete( - &worker, - &first, - vec![start_timer_cmd(1, Duration::from_secs(1))], - ) - .await; - let second = worker.poll_workflow_activation().await.unwrap(); - complete_and_evict(&worker, &second.run_id).await; - worker.drain_pollers_and_shutdown().await; - - assert_eq!( - recorder.log(), - [ - "before 3", - "sent commands=1 force_new=false", - "after 3 true", - "before 8", - "sent commands=1 force_new=false", - "after 8 true", - "evicted", - ] - ); - let infos = recorder.infos(); - assert_eq!(infos[0].run_id, first.run_id); - assert_eq!(infos[0].workflow_id, "fakeid"); - assert_eq!( - infos[0].namespace, - test_worker_cfg().build().unwrap().namespace - ); - assert_eq!(infos[0].attempt, 1); - assert!(infos[0].started_time.is_some()); - assert_eq!(infos[0].closes_run, None); - assert_eq!(infos[1].closes_run, Some(crate::RunCloseKind::Completed)); - // The hook gets nine of the ten seconds of the task timeout, from the task's receipt. - let left = infos[0].deadline - received; - assert!(left <= Duration::from_secs(9), "{left:?}"); - assert!(left > Duration::from_secs(8), "{left:?}"); - assert!(infos[1].deadline > infos[0].deadline); -} - -#[tokio::test] -async fn an_error_fails_the_task_and_evicts_the_run() { - let recorder = Arc::new(Recorder { - fail_before: vec![3], - ..Default::default() - }); - let mut cfg = - MockPollCfg::from_resp_batches("fakeid", timer_then_complete(), [1], mock_worker_client()); - cfg.num_expected_fails = 1; - cfg.num_expected_completions = Some(0.into()); - let worker = worker_with(cfg, &recorder, accept); - - let first = worker.poll_workflow_activation().await.unwrap(); - complete( - &worker, - &first, - vec![start_timer_cmd(1, Duration::from_secs(1))], - ) - .await; - let eviction = worker.poll_workflow_activation().await.unwrap(); - assert_matches!( - eviction.jobs[0].variant, - Some(workflow_activation_job::Variant::RemoveFromCache(_)) - ); - worker - .complete_workflow_activation(WorkflowActivationCompletion::empty(eviction.run_id)) - .await - .unwrap(); - worker.drain_pollers_and_shutdown().await; - - assert_eq!(recorder.log(), ["before 3", "failed 1 true", "evicted"]); -} - -#[tokio::test] -async fn a_hook_that_misses_the_deadline_fails_the_task() { - let recorder = Arc::new(Recorder { - before_delay: Some(Duration::from_secs(2)), - ..Default::default() - }); - let mut t = TestHistoryBuilder::default(); - t.add_wfe_started_with_wft_timeout(Duration::from_secs(1)); - t.add_workflow_task_scheduled_and_started(); - let mut cfg = MockPollCfg::from_resp_batches("fakeid", t, [1], mock_worker_client()); - cfg.num_expected_fails = 1; - cfg.num_expected_completions = Some(0.into()); - let worker = worker_with(cfg, &recorder, accept); - - let first = worker.poll_workflow_activation().await.unwrap(); - let started = Instant::now(); - complete( - &worker, - &first, - vec![start_timer_cmd(1, Duration::from_secs(1))], - ) - .await; - assert!(started.elapsed() < Duration::from_secs(2)); - let eviction = worker.poll_workflow_activation().await.unwrap(); - worker - .complete_workflow_activation(WorkflowActivationCompletion::empty(eviction.run_id)) - .await - .unwrap(); - worker.drain_pollers_and_shutdown().await; - - assert_eq!(recorder.log(), ["before 3", "failed 1 true", "evicted"]); -} - -#[tokio::test] -async fn a_refused_completion_is_not_accepted() { - let recorder = Arc::new(Recorder::default()); - let mut cfg = - MockPollCfg::from_resp_batches("fakeid", timer_then_complete(), [1], mock_worker_client()); - cfg.num_expected_completions = Some(1.into()); - let worker = worker_with(cfg, &recorder, |_| { - Err(tonic::Status::invalid_argument("UnhandledCommand")) - }); - - let first = worker.poll_workflow_activation().await.unwrap(); - complete( - &worker, - &first, - vec![start_timer_cmd(1, Duration::from_secs(1))], - ) - .await; - let eviction = worker.poll_workflow_activation().await.unwrap(); - worker - .complete_workflow_activation(WorkflowActivationCompletion::empty(eviction.run_id)) - .await - .unwrap(); - worker.drain_pollers_and_shutdown().await; - - assert_eq!( - recorder.log(), - [ - "before 3", - "sent commands=1 force_new=false", - "after 3 false", - "evicted" - ] - ); -} - -#[tokio::test] -async fn a_dropped_speculative_task_is_not_accepted() { - let wfid = "fakeid"; - let mut t = TestHistoryBuilder::default(); - t.add_by_type(EventType::WorkflowExecutionStarted); - t.add_full_wf_task(); - let timer_started = t.add_by_type(EventType::TimerStarted); - let mut spec_task_hist = t.clone(); - spec_task_hist.add_workflow_task_scheduled_and_started(); - let mut real_hist = t.clone(); - real_hist.add_timer_fired(timer_started, "1".to_string()); - real_hist.add_workflow_task_scheduled_and_started(); - - let update_id = "upd-1"; - let mut speculative_task = hist_to_poll_resp(&spec_task_hist, wfid, ResponseType::OneTask(2)); - speculative_task.add_update_request(update_id, 1); - let cfg = MockPollCfg::from_resp_batches( - wfid, - real_hist, - [ - ResponseType::ToTaskNum(1), - speculative_task.into(), - ResponseType::OneTask(2), - ], - mock_worker_client(), - ); - let recorder = Arc::new(Recorder::default()); - let worker = worker_with(cfg, &recorder, |n| { - let mut r = RespondWorkflowTaskCompletedResponse::default(); - if n == 2 { - // The rejection drops the speculative task from History. - r.reset_history_event_id = 3; - } - Ok(r) - }); - - let first = worker.poll_workflow_activation().await.unwrap(); - complete( - &worker, - &first, - vec![start_timer_cmd(1, Duration::from_secs(1))], - ) - .await; - let update = worker.poll_workflow_activation().await.unwrap(); - complete( - &worker, - &update, - vec![ - UpdateResponse { - protocol_instance_id: update_id.to_string(), - response: Some(Response::Rejected(Default::default())), - } - .into(), - ], - ) - .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 3", - "sent commands=1 force_new=false", - "after 3 true", - "before 7", - "sent commands=0 force_new=false", - "after 7 false", - "before 8", - "sent commands=1 force_new=false", - "after 8 true", - "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] -async fn the_local_activity_heartbeat_completion_runs_the_hooks() { - let mut t = TestHistoryBuilder::default(); - t.add_wfe_started_with_wft_timeout(Duration::from_secs(1)); - t.add_full_wf_task(); - t.add_workflow_task_scheduled_and_started(); - let cfg = MockPollCfg::from_resp_batches("fakeid", t, [1, 2], mock_worker_client()); - let recorder = Arc::new(Recorder::default()); - let worker = worker_with(cfg, &recorder, accept); - - let first = worker.poll_workflow_activation().await.unwrap(); - complete( - &worker, - &first, - vec![schedule_local_activity_cmd( - 1, - "1", - ActivityCancellationType::TryCancel, - Duration::from_secs(60), - )], - ) - .await; - let la = worker.poll_activity_task().await.unwrap(); - assert_matches!(&la.variant, Some(activity_task::Variant::Start(s)) if s.is_local); - // Core heartbeats the task at 80% of its one-second timeout, with no lang activation. - let resolved = { - let wf_poll = worker.poll_workflow_activation(); - tokio::pin!(wf_poll); - tokio::select! { - act = &mut wf_poll => panic!("Unexpected activation {act:?}"), - _ = tokio::time::sleep(Duration::from_millis(1500)) => {} - } - assert_eq!( - recorder.log(), - ["before 3", "sent commands=0 force_new=true", "after 3 true"] - ); - worker - .complete_activity_task(ActivityTaskCompletion { - task_token: la.task_token, - result: Some(ActivityExecutionResult::ok(vec![1].into())), - }) - .await - .unwrap(); - wf_poll.await.unwrap() - }; - complete_and_evict(&worker, &resolved.run_id).await; - worker.drain_pollers_and_shutdown().await; - - assert_eq!( - recorder.log(), - [ - "before 3", - "sent commands=0 force_new=true", - "after 3 true", - "before 6", - "sent commands=2 force_new=false", - "after 6 true", - "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] -async fn an_eviction_with_output_runs_the_hooks_before_on_evicted() { - let mut t = TestHistoryBuilder::default(); - // Keep the LA heartbeat from firing mid-test. - t.add_wfe_started_with_wft_timeout(Duration::from_secs(60)); - t.add_workflow_task_scheduled_and_started(); - let cfg = MockPollCfg::from_resp_batches("fakeid", t, [1], mock_worker_client()); - let recorder = Arc::new(Recorder::default()); - let worker = worker_with(cfg, &recorder, accept); - - let first = worker.poll_workflow_activation().await.unwrap(); - let run_id = first.run_id.clone(); - let la_cmd = |seq: u32| { - schedule_local_activity_cmd( - seq, - &seq.to_string(), - ActivityCancellationType::WaitCancellationCompleted, - Duration::from_secs(60), - ) - }; - complete(&worker, &first, vec![la_cmd(1), la_cmd(2)]).await; - let mut tokens = vec![]; - for _ in 0..2 { - let task = worker.poll_activity_task().await.unwrap(); - let Some(activity_task::Variant::Start(start)) = &task.variant else { - panic!("Expected LA start, got {task:?}"); - }; - tokens.push((start.activity_id.clone(), task.task_token)); - } - tokens.sort(); - let (_, first_token) = tokens.remove(0); - worker - .complete_activity_task(ActivityTaskCompletion { - task_token: first_token, - result: Some(ActivityExecutionResult::ok(vec![1].into())), - }) - .await - .unwrap(); - // The marker of the first LA is output that waits on the second one. - let resolved = worker.poll_workflow_activation().await.unwrap(); - complete(&worker, &resolved, vec![]).await; - assert!(recorder.log().is_empty()); - - worker.request_workflow_eviction(&run_id); - let cancel = worker.poll_activity_task().await.unwrap(); - assert_matches!(cancel.variant, Some(activity_task::Variant::Cancel(_))); - worker - .complete_activity_task(ActivityTaskCompletion { - task_token: cancel.task_token, - result: Some(ActivityExecutionResult::cancel_from_details(None)), - }) - .await - .unwrap(); - let eviction = worker.poll_workflow_activation().await.unwrap(); - assert_matches!( - eviction.jobs[0].variant, - Some(workflow_activation_job::Variant::RemoveFromCache(_)) - ); - worker - .complete_workflow_activation(WorkflowActivationCompletion::empty(eviction.run_id)) - .await - .unwrap(); - worker.drain_pollers_and_shutdown().await; - - assert_eq!( - recorder.log(), - [ - "before 3", - "sent commands=1 force_new=true", - "after 3 true", - "evicted" - ] - ); -} - -#[tokio::test] -async fn the_replay_worker_never_calls_the_hooks() { - let recorder = Arc::new(Recorder::default()); - let extension: Arc = Arc::new(Extension(recorder.clone())); - let mut cfg = test_worker_cfg().build().unwrap(); - cfg.extension = Some(extension); - let worker = init_replay_worker(ReplayWorkerInput::new( - cfg, - stream::iter([HistoryForReplay::from(canned_histories::single_timer("1"))]), - )) - .unwrap(); - - 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(); - complete( - &worker, - &fired, - vec![CompleteWorkflowExecution { result: None }.into()], - ) - .await; - while let Ok(activation) = worker.poll_workflow_activation().await { - worker - .complete_workflow_activation(WorkflowActivationCompletion::empty(activation.run_id)) - .await - .unwrap(); - } - worker.shutdown().await; - - 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 { - hang_after: vec![3], - ..Default::default() - }); - let worker = worker_with( - MockPollCfg::from_resp_batches( - "fakeid", - timer_then_complete(), - [1, 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 second = worker.poll_workflow_activation().await.unwrap(); - complete_and_evict(&worker, &second.run_id).await; - worker.drain_pollers_and_shutdown().await; - - assert_eq!( - recorder.log(), - [ - "before 3", - "sent commands=1 force_new=false", - "after 3 true", - "before 8", - "sent commands=1 force_new=false", - "after 8 true", - "evicted", - ] - ); -} - -#[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 fd22a00e9e..c3984975b4 100644 --- a/crates/sdk-core/src/lib.rs +++ b/crates/sdk-core/src/lib.rs @@ -50,10 +50,7 @@ pub use worker::{ WorkerConfigBuilder, WorkerTuner, WorkerValidationError, WorkerVersioningStrategy, WorkflowErrorType, WorkflowSlotKind, }; -pub use worker::{ - ExtensionError, RunCloseKind, WorkerExtension, WorkflowTaskCompletionInfo, WorkflowTaskEpoch, - WorkflowTaskFailedInfo, WorkflowTaskInterceptError, WorkflowTaskInterceptor, -}; +pub use worker::{ExtensionError, WorkerExtension}; use crate::{ replay::{HistoryForReplay, ReplayWorkerInput}, diff --git a/crates/sdk-core/src/protosext/mod.rs b/crates/sdk-core/src/protosext/mod.rs index 0544a42eba..87319956c7 100644 --- a/crates/sdk-core/src/protosext/mod.rs +++ b/crates/sdk-core/src/protosext/mod.rs @@ -12,7 +12,7 @@ use std::{ collections::HashMap, convert::TryFrom, fmt::{Debug, Display, Formatter}, - time::{Duration, Instant, SystemTime}, + time::{Duration, SystemTime}, }; use temporalio_common::protos::{ constants::{LOCAL_ACTIVITY_MARKER_NAME, PATCH_MARKER_NAME}, @@ -59,8 +59,6 @@ pub(crate) struct ValidPollWFTQResponse { pub(crate) started_event_id: i64, /// When the server started the task. From the `started_time` field. pub(crate) started_time: Option, - /// When the worker received the task, before it waits on a busy run or a history fetch. - pub(crate) received: Instant, /// If this is present, `history` will be empty. This is not a very "tight" design, but it's /// enforced at construction time. From the `query` field. pub(crate) legacy_query: Option, @@ -136,7 +134,6 @@ impl TryFrom for ValidPollWFTQResponse { previous_started_event_id, started_event_id, started_time: started_time.and_then(|t| t.try_into().ok()), - received: Instant::now(), legacy_query: query, query_requests, messages, diff --git a/crates/sdk-core/src/worker/extension.rs b/crates/sdk-core/src/worker/extension.rs index fb643f3a95..4ff8a91807 100644 --- a/crates/sdk-core/src/worker/extension.rs +++ b/crates/sdk-core/src/worker/extension.rs @@ -1,27 +1,15 @@ -//! Hooks that let a plugin run code at fixed points of the worker's workflow task lifecycle. - -use std::{ - fmt::Debug, - sync::Arc, - time::{Instant, SystemTime}, -}; +//! A plugin component that a worker carries, which lang calls by method name. /// A component that lang passes to the worker through [crate::WorkerConfig::extension]. /// -/// Core never inspects the extension beyond the hooks it asks for here. Core never calls -/// [WorkerExtension::call] or [WorkerExtension::call_async]. A bridge exposes them to lang, so lang -/// code reaches the extension in the same native library without a bridge function of its own. +/// Core never calls [WorkerExtension::call] or [WorkerExtension::call_async]. A bridge exposes +/// them to lang, so lang code reaches the extension in the same native library without a bridge +/// function of its own. #[async_trait::async_trait] pub trait WorkerExtension: Send + Sync + 'static { /// A name that identifies the extension in logs. fn name(&self) -> &str; - /// The interceptor Core calls around each workflow task completion, if any. Core may ask - /// more than once per worker, so return the same interceptor each time. - fn workflow_task_interceptor(&self) -> Option> { - None - } - /// Runs `method` with encoded input and returns the encoded output. It must not block, since /// lang may call it from workflow code. fn call(&self, method: &str, _input: &[u8]) -> Result, ExtensionError> { @@ -73,140 +61,10 @@ impl ExtensionError { } } -/// Hooks around each `RespondWorkflowTaskCompleted` call that Core makes for a run. -/// -/// Core calls the hooks of one run in order, and the next workflow task of that run waits until -/// they return. Core waits at most five seconds on [WorkflowTaskInterceptor::after_complete], so -/// long work such as retries must run in the background. The replay worker never calls the hooks. -/// -/// A completion that lang sends waits only on its own run. Core sends the local activity heartbeat -/// completion from within the poll for the next activation, so while its hooks run, that poll -/// gives no activation to other runs. -#[async_trait::async_trait] -pub trait WorkflowTaskInterceptor: Send + Sync + 'static { - /// Called before every `RespondWorkflowTaskCompleted` for the run. This includes the - /// completions that have no lang activation, which are the local activity heartbeat - /// completion and the completion of output buffered before an eviction. - /// - /// An error, or a hook that has not returned by `task.deadline`, fails the workflow task - /// instead of completing it, and the run is evicted. - async fn before_complete( - &self, - task: &WorkflowTaskCompletionInfo, - ) -> Result<(), WorkflowTaskInterceptError>; - - /// Called after the server answered the completion, or after the call failed. - /// - /// `accepted` is false when the call failed, and also when the answer has a - /// `reset_history_event_id`, which means the server dropped a speculative task. - async fn after_complete(&self, task: &WorkflowTaskCompletionInfo, accepted: bool); - - /// Called when Core fails a workflow task of the run, whether or not it reports the failure. - fn on_task_failed(&self, _task: &WorkflowTaskFailedInfo) {} - - /// Called when Core removes the run from its cache. When an eviction sends buffered output, - /// this comes after [WorkflowTaskInterceptor::after_complete] for that completion. - fn on_evicted(&self, _run_id: &str) {} -} - -/// What Core knows about a workflow task when it is about to complete it. -#[derive(Clone, Debug, bon::Builder)] -#[builder(on(String, into))] -#[non_exhaustive] -pub struct WorkflowTaskCompletionInfo { - /// The namespace of the run. - pub namespace: String, - /// The workflow id of the run. - pub workflow_id: String, - /// The run id. - pub run_id: String, - /// The first run id of the run chain. - pub first_execution_run_id: String, - /// The workflow type. - pub workflow_type: String, - /// The id of the task's `WorkflowTaskStarted` event, from the poll response. The server writes - /// no events for transient attempts, so attempts after the second of one task share an id. - pub started_event_id: i64, - /// The task's attempt, from the poll response. - pub attempt: u32, - /// 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, - /// When the hook must have returned, so that the task does not time out on the server. Core - /// measures it from the local receipt of the task, with a margin. - 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] -pub enum RunCloseKind { - /// `CompleteWorkflowExecution` - Completed, - /// `FailWorkflowExecution`. A retry policy can still start a new run in the chain. - Failed, - /// `CancelWorkflowExecution` - Cancelled, -} - -/// What Core knows about a workflow task that it fails. -#[derive(Clone, Debug, bon::Builder)] -#[builder(on(String, into))] -#[non_exhaustive] -pub struct WorkflowTaskFailedInfo { - /// The run id. - pub run_id: String, - /// The attempt of the failed task. - pub attempt: u32, - /// True when Core sends `RespondWorkflowTaskFailed`. Core reports only first attempts and - /// lets later ones time out. - pub reported: bool, -} - -/// The error a [WorkflowTaskInterceptor] returns to fail a workflow task. -#[derive(Debug, thiserror::Error)] -#[error("{message}")] -pub struct WorkflowTaskInterceptError { - /// The message that goes into the task failure. - pub message: String, -} - -impl WorkflowTaskInterceptError { - /// Creates an error with a message. - pub fn new(message: impl Into) -> Self { - Self { - message: message.into(), - } - } -} - #[cfg(test)] mod tests { use super::*; + use std::sync::Arc; struct Named; diff --git a/crates/sdk-core/src/worker/mod.rs b/crates/sdk-core/src/worker/mod.rs index fcf58bca48..a37defa00d 100644 --- a/crates/sdk-core/src/worker/mod.rs +++ b/crates/sdk-core/src/worker/mod.rs @@ -11,10 +11,7 @@ mod workflow; /// the error limit, shared so every conversion site reports the same identifier. pub(crate) const PAYLOADS_TOO_LARGE_FAILURE_TYPE: &str = "PayloadsTooLarge"; -pub use extension::{ - ExtensionError, RunCloseKind, WorkerExtension, WorkflowTaskCompletionInfo, WorkflowTaskEpoch, - WorkflowTaskFailedInfo, WorkflowTaskInterceptError, WorkflowTaskInterceptor, -}; +pub use extension::{ExtensionError, WorkerExtension}; use temporalio_client::{Connection, PayloadErrorLimits, worker::NamespaceDescriptionSource}; use temporalio_common::{ protos::{ @@ -305,8 +302,7 @@ pub struct WorkerConfig { #[builder(default = false)] pub disable_payload_error_limit: bool, - /// A plugin component that Core calls at fixed points of the workflow task lifecycle. See - /// [WorkerExtension]. + /// A plugin component that lang calls by method name. See [WorkerExtension]. pub extension: Option>, } diff --git a/crates/sdk-core/src/worker/workflow/driven_workflow.rs b/crates/sdk-core/src/worker/workflow/driven_workflow.rs index 470c8d82fd..9a911d1136 100644 --- a/crates/sdk-core/src/worker/workflow/driven_workflow.rs +++ b/crates/sdk-core/src/worker/workflow/driven_workflow.rs @@ -50,7 +50,6 @@ impl DrivenWorkflow { debug!(run_id = %attribs.original_execution_run_id, "Driven WF start"); let started_info = WorkflowStartedInfo { workflow_task_timeout: attribs.workflow_task_timeout.try_into_or_none(), - first_execution_run_id: attribs.first_execution_run_id.clone(), memo: attribs.memo.clone(), search_attrs: attribs.search_attributes.clone(), retry_policy: attribs.retry_policy.clone(), diff --git a/crates/sdk-core/src/worker/workflow/history_update.rs b/crates/sdk-core/src/worker/workflow/history_update.rs index 003324a37c..638fe3f200 100644 --- a/crates/sdk-core/src/worker/workflow/history_update.rs +++ b/crates/sdk-core/src/worker/workflow/history_update.rs @@ -157,7 +157,6 @@ impl HistoryPaginator { task_token: wft.task_token, attempt: wft.attempt, started_time: wft.started_time, - received: wft.received, execution: wft.workflow_execution, workflow_type: wft.workflow_type, legacy_query: wft.legacy_query, 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 cd839388b7..e26aabd7f9 100644 --- a/crates/sdk-core/src/worker/workflow/machines/workflow_machines.rs +++ b/crates/sdk-core/src/worker/workflow/machines/workflow_machines.rs @@ -23,7 +23,6 @@ use crate::{ telemetry::{VecDisplayer, metrics::MetricsContext}, worker::{ ExecutingLAId, LocalActRequest, LocalActivityExecutionResult, LocalActivityResolution, - WorkflowTaskEpoch, workflow::{ CommandAnnotations, CommandID, DrivenWorkflow, HistoryUpdate, InternalFlagsRef, LocalResolution, OutgoingJob, ProtoCommandExt, RunBasics, WFCommand, WFCommandVariant, @@ -78,53 +77,6 @@ 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 @@ -138,8 +90,6 @@ 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. @@ -337,7 +287,6 @@ 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, @@ -464,28 +413,6 @@ 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 } @@ -789,7 +716,6 @@ 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 72bc92d574..61c79467e6 100644 --- a/crates/sdk-core/src/worker/workflow/managed_run.rs +++ b/crates/sdk-core/src/worker/workflow/managed_run.rs @@ -4,7 +4,7 @@ use crate::{ internal_flags::CoreInternalFlags, protosext::{WorkflowActivationExt, protocol_messages::IncomingProtocolMessage}, worker::{ - LEGACY_QUERY_ID, LocalActRequest, WorkflowErrorType, WorkflowTaskCompletionInfo, + LEGACY_QUERY_ID, LocalActRequest, WorkflowErrorType, workflow::{ ActivationAction, ActivationCompleteOutcome, ActivationCompleteResult, ActivationOrAuto, BufferedTasks, DrivenWorkflow, EvictionRequestResult, @@ -12,11 +12,10 @@ use crate::{ LocalActivityRequestSink, LocalResolution, NextPageReq, OutstandingActivation, OutstandingTask, PermittedWFT, RequestEvictMsg, RunBasics, ServerCommandsWithWorkflowInfo, TaskStorageMetrics, WFCommand, WFCommandVariant, - WFMachinesError, WFT_HEARTBEAT_TIMEOUT_FRACTION, WFT_INTERCEPTOR_DEADLINE_FRACTION, - WFTReportStatus, WftFailureKind, WorkflowTaskInfo, + WFMachinesError, WFT_HEARTBEAT_TIMEOUT_FRACTION, WFTReportStatus, WftFailureKind, + WorkflowTaskInfo, history_update::HistoryPaginator, machines::{MachinesWFTResponseContent, WorkflowMachines}, - run_close_kind, }, }, }; @@ -42,7 +41,6 @@ use temporalio_common::protos::{ workflow_completion, }, temporal::api::{ - command::v1::Command as ProtoCommand, enums::v1::{VersioningBehavior, WorkflowTaskFailedCause}, failure::v1::Failure, }, @@ -50,10 +48,6 @@ use temporalio_common::protos::{ use tokio::sync::oneshot; use tracing::Span; -/// The server's default workflow task timeout. The start event always has the timeout, so this -/// only keeps the hook's deadline finite if it is missing. -const DEFAULT_WFT_TIMEOUT: Duration = Duration::from_secs(10); - type Result = std::result::Result; pub(super) type RunUpdateAct = Option; @@ -202,9 +196,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, wf_id: work.execution.workflow_id.clone(), }; @@ -1244,7 +1236,7 @@ impl ManagedRun { force_new_wft = true; } - let mut outcome = if should_respond || has_query_responses { + let outcome = if should_respond || has_query_responses { // Broken evictions cannot pass should_respond or contain query responses. Keep this // fallback to avoid sending buffered output if those invariants regress. let (commands, messages) = if self.am_broken && data.activation_was_eviction { @@ -1268,7 +1260,6 @@ impl ManagedRun { sdk_metadata: machines_wft_response.metadata_for_complete(), versioning_behavior: data.versioning_behavior, attempt, - task_info: None, }, metrics: self.metrics.clone(), }) @@ -1276,18 +1267,6 @@ impl ManagedRun { ActivationCompleteOutcome::DoNothing }; let replaying = machines_wft_response.replaying; - if let ActivationCompleteOutcome::ReportWFTSuccess(ServerCommandsWithWorkflowInfo { - action: - ActivationAction::WftComplete { - commands, - task_info, - .. - }, - .. - }) = &mut outcome - { - *task_info = self.completion_info(commands).map(Box::new); - } if matches!(outcome, ActivationCompleteOutcome::ReportWFTSuccess(_)) { self.wfm.wft_completion_reported(); } @@ -1297,45 +1276,6 @@ impl ManagedRun { } } - /// The facts a workflow task interceptor gets about the task being completed. `None` when the - /// worker has no extension, or the run has no task. - fn completion_info(&self, commands: &[ProtoCommand]) -> Option { - self.config - .extension - .as_ref()? - .workflow_task_interceptor()?; - 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); - Some( - WorkflowTaskCompletionInfo::builder() - .namespace(self.config.namespace.clone()) - .workflow_id(machines.workflow_id.clone()) - .run_id(machines.run_id.clone()) - .first_execution_run_id( - started_info - .map(|i| i.first_execution_run_id.clone()) - .unwrap_or_default(), - ) - .workflow_type(machines.workflow_type.clone()) - .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), - ) - .build(), - ) - } - /// Names the current workflow task on each new local activity, since a local activity cannot /// learn by itself which task dispatched it. fn stamp_dispatching_task(&self, requests: &mut [LocalActRequest]) { diff --git a/crates/sdk-core/src/worker/workflow/mod.rs b/crates/sdk-core/src/worker/workflow/mod.rs index f0e82a460d..eea2edae20 100644 --- a/crates/sdk-core/src/worker/workflow/mod.rs +++ b/crates/sdk-core/src/worker/workflow/mod.rs @@ -27,8 +27,7 @@ use crate::{ worker::{ ActivitySlotKind, CompleteWfError, LocalActRequest, LocalActivityExecutionResult, LocalActivityResolution, NamespaceCapabilities, PollError, PostActivateHookData, - RunCloseKind, WorkflowSlotKind, WorkflowTaskCompletionInfo, WorkflowTaskFailedInfo, - WorkflowTaskInterceptor, + WorkflowSlotKind, activities::{ActivitiesFromWFTsHandle, LocalActivityManager}, client::{ LegacyQueryResult, REQUEST_TOO_LARGE_KEY, WorkerClient, WorkflowTaskCompletion, @@ -89,7 +88,7 @@ use temporalio_common::{ Memo, MeteringMetadata, RetryPolicy, SearchAttributes, WorkflowExecution, }, enums::v1::{VersioningBehavior, WorkflowTaskFailedCause}, - failure::v1::{ApplicationFailureInfo, Failure as ApiFailure, failure::FailureInfo}, + failure::v1::{ApplicationFailureInfo, failure::FailureInfo}, protocol::v1::Message as ProtocolMessage, query::v1::WorkflowQuery, sdk::v1::{EventGroupMarker, UserMetadata, WorkflowTaskCompletedMetadata}, @@ -118,11 +117,6 @@ pub const LEGACY_QUERY_ID: &str = "legacy_query"; /// What percentage of a WFT timeout we are willing to wait before sending a WFT heartbeat when /// necessary. const WFT_HEARTBEAT_TIMEOUT_FRACTION: f32 = 0.8; -/// What percentage of a WFT timeout may pass before `before_complete` must have returned. It is -/// above the heartbeat fraction, so a heartbeat completion still has time for the hook. -const WFT_INTERCEPTOR_DEADLINE_FRACTION: f32 = 0.9; -/// How long Core waits on `after_complete` before it goes on with the run. -const WFT_INTERCEPTOR_AFTER_COMPLETE_TIMEOUT: Duration = Duration::from_secs(5); type Result = result::Result; type BoxedActivationStream = BoxStream<'static, Result>; @@ -153,7 +147,6 @@ pub(crate) struct Workflows { default_versioning_behavior: Option, namespace_capabilities: Arc, shutdown_token: CancellationToken, - wft_interceptor: Option>, } pub(crate) struct WorkflowBasics { @@ -203,11 +196,6 @@ impl Workflows { let default_versioning_behavior = basics.default_versioning_behavior; let namespace_capabilities = basics.namespace_capabilities.clone(); let shutdown_token = basics.shutdown_token.clone(); - let wft_interceptor = basics - .worker_config - .extension - .as_ref() - .and_then(|e| e.workflow_task_interceptor()); let extracted_wft_stream = WFTExtractor::build( client.clone(), basics.worker_config.fetching_concurrency, @@ -305,7 +293,6 @@ impl Workflows { default_versioning_behavior, namespace_capabilities, shutdown_token, - wft_interceptor, } } @@ -395,46 +382,9 @@ impl Workflows { sdk_metadata, mut versioning_behavior, attempt, - task_info, }, metrics: run_metrics, } => { - let interceptor = self.wft_interceptor.as_deref().zip(task_info.as_deref()); - let before = match (interceptor, self.wft_interceptor.is_some()) { - (Some((interceptor, info)), _) => run_before_complete(interceptor, info).await, - (None, false) => Ok(()), - // Completing without the hook would skip it, so fail the task instead. - (None, true) => { - dbg_panic!("Workflow task completion is missing its interceptor info"); - Err("Workflow task completion is missing its interceptor info".to_string()) - } - }; - if let Err(message) = before { - warn!(run_id, error = %message, "Workflow task interceptor failed the task"); - // The machines already count this task as completed, so the run must replay. - self.request_eviction( - run_id, - "Workflow task interceptor failed the task", - EvictionReason::Fatal, - ); - return self - .handle_activation_failed( - run_id, - completion_time, - FailedActivationWFTReport::new( - task_token, - attempt, - WorkflowTaskFailedCause::WorkflowWorkerUnhandledFailure, - Failure { - failure: Some(ApiFailure::application_failure(message, false)), - ..Default::default() - }, - WftFailureKind::Task, - &run_metrics, - ), - ) - .await; - } let reserved_act_permits = self.reserve_activity_slots_for_outgoing_commands(commands.as_mut_slice()); debug!(commands=%commands.display(), query_responses=%query_responses.display(), @@ -481,7 +431,6 @@ impl Workflows { completion.sticky_attributes = sticky_attrs; let mut reset_last_started_to = None; - let mut accepted = false; self.handle_wft_reporting_errs(run_id, || async { match self .client @@ -494,8 +443,6 @@ impl Workflows { } if response.reset_history_event_id > 0 { reset_last_started_to = Some(response.reset_history_event_id); - } else { - accepted = true; } if let Some(wft) = response.workflow_task { *wft_from_complete = Some(validate_wft(wft)?); @@ -551,9 +498,6 @@ impl Workflows { Ok(()) }) .await; - if let Some((interceptor, info)) = interceptor { - run_after_complete(interceptor, info, accepted).await; - } WFTReportStatus::Reported { reset_last_started_to, completion_time, @@ -605,11 +549,9 @@ impl Workflows { WftFailureKind::Task if attempt > 1 => { debug!(run_id=%run_id, attempt, failure=?failure, "Not reporting workflow task failure on non-first attempt"); - self.notify_task_failed(run_id, attempt, false); return WFTReportStatus::DropWft { completion_time }; } WftFailureKind::Task => { - self.notify_task_failed(run_id, attempt, true); warn!(run_id=%run_id, failure=?failure, "Failing workflow task"); self.handle_wft_reporting_errs(run_id, || async { self.client @@ -625,18 +567,6 @@ impl Workflows { } } - fn notify_task_failed(&self, run_id: &str, attempt: u32, reported: bool) { - if let Some(interceptor) = self.wft_interceptor.as_deref() { - interceptor.on_task_failed( - &WorkflowTaskFailedInfo::builder() - .run_id(run_id) - .attempt(attempt) - .reported(reported) - .build(), - ); - } - } - async fn handle_activation_completed_result( &self, run_id: &str, @@ -1090,7 +1020,6 @@ struct PreparedWFT { task_token: TaskToken, attempt: u32, started_time: Option, - received: Instant, execution: WorkflowExecution, workflow_type: String, legacy_query: Option, @@ -1165,9 +1094,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 /// 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 @@ -1255,8 +1182,6 @@ pub(crate) enum ActivationAction { sdk_metadata: WorkflowTaskCompletedMetadata, versioning_behavior: VersioningBehavior, attempt: u32, - /// Set when the worker has a workflow task interceptor. - task_info: Option>, }, /// We should respond to a legacy query request RespondLegacyQuery { result: Box }, @@ -1280,53 +1205,6 @@ impl EvictionRequestResult { } } -/// Runs the `before_complete` hook within the task's deadline, returning the failure message. -async fn run_before_complete( - interceptor: &dyn WorkflowTaskInterceptor, - info: &WorkflowTaskCompletionInfo, -) -> Result<(), String> { - let deadline = tokio::time::Instant::from_std(info.deadline); - match tokio::time::timeout_at(deadline, interceptor.before_complete(info)).await { - Ok(Ok(())) => Ok(()), - Ok(Err(e)) => Err(e.message), - Err(_) => Err("Workflow task interceptor missed the task deadline".to_string()), - } -} - -/// Runs the `after_complete` hook. The run's next task waits on it, so a hung hook must not stall -/// the run. -async fn run_after_complete( - interceptor: &dyn WorkflowTaskInterceptor, - info: &WorkflowTaskCompletionInfo, - accepted: bool, -) { - let after = interceptor.after_complete(info, accepted); - if tokio::time::timeout(WFT_INTERCEPTOR_AFTER_COMPLETE_TIMEOUT, after) - .await - .is_err() - { - warn!(run_id = %info.run_id, "Workflow task interceptor timed out after the completion"); - } -} - -/// The way the commands close the run, if they do. -fn run_close_kind(commands: &[ProtoCommand]) -> Option { - commands - .iter() - .find_map(|cmd| match cmd.attributes.as_ref() { - Some(Attributes::CompleteWorkflowExecutionCommandAttributes(_)) => { - Some(RunCloseKind::Completed) - } - Some(Attributes::FailWorkflowExecutionCommandAttributes(_)) => { - Some(RunCloseKind::Failed) - } - Some(Attributes::CancelWorkflowExecutionCommandAttributes(_)) => { - Some(RunCloseKind::Cancelled) - } - _ => None, - }) -} - #[allow(clippy::type_complexity)] fn detect_terminal_command( commands: &[ProtoCommand], @@ -1822,7 +1700,6 @@ enum CommandID { #[derive(Debug, Clone)] struct WorkflowStartedInfo { workflow_task_timeout: Option, - first_execution_run_id: String, memo: Option, search_attrs: Option, retry_policy: Option, diff --git a/crates/sdk-core/src/worker/workflow/workflow_stream.rs b/crates/sdk-core/src/worker/workflow/workflow_stream.rs index 53e6583993..d00173f64e 100644 --- a/crates/sdk-core/src/worker/workflow/workflow_stream.rs +++ b/crates/sdk-core/src/worker/workflow/workflow_stream.rs @@ -36,7 +36,6 @@ pub(super) struct WFStream { history_fetch_refcounter: Arc, shutdown_token: CancellationToken, ignore_evicts_on_shutdown: bool, - wft_interceptor: Option>, metrics: MetricsContext, } @@ -98,11 +97,6 @@ impl WFStream { ), shutdown_token: basics.shutdown_token, ignore_evicts_on_shutdown: basics.worker_config.ignore_evicts_on_shutdown, - wft_interceptor: basics - .worker_config - .extension - .as_ref() - .and_then(|e| e.workflow_task_interceptor()), metrics: basics.metrics, runs_needing_fetching: Default::default(), history_fetch_refcounter: Arc::new(HistfetchRC {}), @@ -411,9 +405,6 @@ impl WFStream { if should_evict { debug!(run_id=%run_id, "Evicting run"); self.runs.remove(run_id); - if let Some(interceptor) = &self.wft_interceptor { - interceptor.on_evicted(run_id); - } } let maybe_ready_wft = maybe_buffered .get_next_wft()