[Brook] Added the previous completed and failed Workflow Tasks to the interceptor info.

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

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<WorkflowTaskEpoch>) -> Option<i64> {
+    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<ResponseType>,
+) -> 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<RespondWorkflowTaskCompletedResponse, tonic::Status> {
     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]);
+}

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::{

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()

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<SystemTime>,
+    /// 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<WorkflowTaskEpoch>,
+    /// 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<i64>,
     /// Set when the completion has a command that closes the run. Continue-as-new does not
     /// count.
     pub closes_run: Option<RunCloseKind>,
@@ -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<SystemTime>,
+}
+
 /// The command with which a completion closes its run.
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 #[non_exhaustive]