[Brook] Scoped Workflow producer supersession to the replayed prefix.

crates/protos/protos/local/temporal/sdk/streams/v1/message.proto

@@ -31,6 +31,15 @@ message StreamRecord {
     // The producer's position within its attempt, starting at one, or zero
     // when it does not number its records.
     int64 sequence = 7;
+    // Set by a Workflow run's own producer only: the WorkflowTaskStarted
+    // event id of the task attempt that wrote the record.
+    int64 task_started_event_id = 9;
+    // Set by a Workflow run's own producer, on the first record of each
+    // batch: the WorkflowTaskStarted event ids of the run's last completed
+    // tasks, newest first, at most eight. The first is the attempt's prefix.
+    // A trailing zero means the list goes back to the run's start. A reader
+    // learns from it whether an earlier attempt it delivered completed.
+    repeated int64 completed_tasks = 10;
 }
 
 // What a record means to a reader.

crates/protos/src/protos/mod.rs

@@ -3091,6 +3091,8 @@ mod sdk_helpers {
                     producer_id: "p".to_string(),
                     attempt: 2,
                     sequence: 3,
+                    task_started_event_id: 0,
+                    completed_tasks: vec![],
                 }
             }
 

crates/sdk-core/CHANGELOG.md

@@ -53,6 +53,9 @@ relevant information.
   1 per attempt. Replay appends nothing. A store that refuses the append fails the Workflow Task.
   Once the server accepts a task, Core retires the dedupe sessions of the run's earlier attempts.
   The C bridge's worker options take the process's stream store as `stream_store`.
+* Experimental: a read reports `SUPERSEDED` for a Workflow producer (`@<run id>`) only when its
+  attempt keeps the high bits (the same task) and raises the low 16 bits (a rerun). A higher
+  prefix is the next task and reads with no supersession.
 * Experimental: a Workflow Task append that the store refuses as closed drops that topic's
   records, logs a warning, counts `workflow_stream_output_dropped`, and the task goes on. A
   retry would be refused the same way, also when the run closed the topic itself before its task

crates/sdk-core/src/core_tests/external_streams.rs

@@ -668,6 +668,11 @@ async fn the_next_task_moves_the_prefix_and_starts_the_sequence_again() {
         [(TOPIC.to_string(), producer, 1)],
         "the second task's completion retires the first task's session, and keeps its own"
     );
+    assert_eq!(
+        store.read(TOPIC).await,
+        ["encrypted a", "encrypted b", "encrypted c"],
+        "a later task goes on with no supersession, and a heartbeat task is a task like any"
+    );
 }
 
 #[tokio::test]
@@ -902,6 +907,97 @@ async fn an_unhandled_command_rerun_supersedes_what_the_refused_task_appended()
     );
 }
 
+#[tokio::test]
+async fn a_rerun_that_skips_the_topic_still_withdraws_what_the_failed_task_appended() {
+    // The first completion is refused. The rerun takes another branch, writes nothing to the
+    // topic and completes. The task after it writes again under a new prefix.
+    let mut t = TestHistoryBuilder::default();
+    t.add_by_type(EventType::WorkflowExecutionStarted);
+    t.add_workflow_task_scheduled_and_started();
+    t.add_workflow_task_failed_with_failure(
+        WorkflowTaskFailedCause::UnhandledCommand,
+        Default::default(),
+    );
+    t.add_we_signaled("go", vec![]);
+    t.add_full_wf_task();
+    let timer = t.add_timer_started("1".to_string());
+    t.add_timer_fired(timer, "1".to_string());
+    t.add_workflow_task_scheduled_and_started();
+    let events = t.get_full_history_info().unwrap().events().to_vec();
+    let first_task = TestHistoryBuilder::from_history(events[..3].to_vec());
+    let first_task = hist_to_poll_resp(&first_task, "fakeid", 1.into()).resp;
+    let rerun_task = TestHistoryBuilder::from_history(events[..7].to_vec());
+    let rerun_task = hist_to_poll_resp(&rerun_task, "fakeid", ResponseType::AllHistory).resp;
+    let store = Arc::new(RecordingStore::default());
+    let mut answers = 0;
+    let mut mock_cfg = MockPollCfg::from_resp_batches(
+        "fakeid",
+        t,
+        [
+            ResponseType::Raw(first_task),
+            ResponseType::Raw(rerun_task),
+            ResponseType::AllHistory,
+        ],
+        mock_worker_client(),
+    );
+    mock_cfg.completion_mock_fn = Some(Box::new(move |_| {
+        answers += 1;
+        if answers == 1 {
+            Err(tonic::Status::invalid_argument("UnhandledCommand"))
+        } else {
+            Ok(RespondWorkflowTaskCompletedResponse::default())
+        }
+    }));
+    let mut mock = build_mock_pollers(mock_cfg);
+    with_store(&mut mock, Some(store.clone()));
+    let worker = mock_worker(mock);
+
+    let first = worker.poll_workflow_activation().await.unwrap();
+    complete(
+        &worker,
+        &first.run_id,
+        vec![
+            output_commit_command(records(&["draft"])),
+            CompleteWorkflowExecution::default().into(),
+        ],
+    )
+    .await;
+    let eviction = worker.poll_workflow_activation().await.unwrap();
+    assert!(eviction.eviction_reason().is_some());
+    worker
+        .complete_workflow_activation(WorkflowActivationCompletion::empty(eviction.run_id))
+        .await
+        .unwrap();
+    let rerun = worker.poll_workflow_activation().await.unwrap();
+    complete(
+        &worker,
+        &rerun.run_id,
+        vec![start_timer_cmd(1, Duration::from_secs(10))],
+    )
+    .await;
+    let fired = worker.poll_workflow_activation().await.unwrap();
+    complete(
+        &worker,
+        &fired.run_id,
+        vec![
+            output_commit_command(records(&["final"])),
+            CompleteWorkflowExecution::default().into(),
+        ],
+    )
+    .await;
+    worker.drain_pollers_and_shutdown().await;
+
+    let appends = store.appends();
+    assert_eq!(
+        appends.iter().map(|a| a.attempt).collect::<Vec<_>>(),
+        [attempt(0, 1), attempt(7, 1)]
+    );
+    assert_eq!(
+        store.read(TOPIC).await,
+        ["encrypted draft", "SUPERSEDED", "encrypted final"]
+    );
+}
+
 #[tokio::test]
 async fn a_rerun_after_a_crash_supersedes_what_the_stopped_worker_appended() {
     let store = Arc::new(RecordingStore::default());

crates/sdk-core/src/worker/workflow/machines/workflow_machines.rs

@@ -99,6 +99,16 @@ fn workflow_task_reruns_in(event: &HistoryEvent) -> i64 {
 /// The bits of a Workflow producer's attempt that hold the reruns since the last completed task.
 const STREAM_RETRY_BITS: u32 = 16;
 
+/// How many completed tasks a Workflow producer's records name. A reader learns from them whether
+/// an earlier attempt it delivered completed, as long as fewer tasks completed since.
+const STREAM_COMPLETED_TASKS: usize = 8;
+
+/// A Workflow producer's attempt: the prefix in the high bits, then 1 plus the reruns since it.
+fn stream_attempt(prefix: i64, reruns: i64) -> i64 {
+    let retry = (1 + reruns).min((1 << STREAM_RETRY_BITS) - 1);
+    (prefix << STREAM_RETRY_BITS) | retry
+}
+
 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
@@ -121,6 +131,10 @@ pub(crate) struct WorkflowMachines {
     stream_reruns: i64,
     /// The id of the last WorkflowTaskStarted event applied.
     stream_last_started: i64,
+    /// The WorkflowTaskStarted event ids of the last completed tasks, newest first.
+    stream_completed: std::collections::VecDeque<i64>,
+    /// How many tasks completed in all.
+    stream_completed_count: usize,
     /// 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.
@@ -318,6 +332,8 @@ impl WorkflowMachines {
             stream_prefix: 0,
             stream_reruns: 0,
             stream_last_started: 0,
+            stream_completed: Default::default(),
+            stream_completed_count: 0,
             next_started_event_id: 0,
             last_processed_event: 0,
             workflow_start_time: None,
@@ -452,8 +468,22 @@ impl WorkflowMachines {
     /// completed task's WorkflowTaskStarted event id in the high bits, then 1 plus the reruns
     /// since it. History alone gives it, so a Worker restart or a cache miss gets the same one.
     pub(crate) fn stream_producer_attempt(&self) -> i64 {
-        let retry = (1 + self.stream_reruns).min((1 << STREAM_RETRY_BITS) - 1);
-        (self.stream_prefix << STREAM_RETRY_BITS) | retry
+        stream_attempt(self.stream_prefix, self.stream_reruns)
+    }
+
+    /// The WorkflowTaskStarted event id of the current task.
+    pub(crate) fn stream_task_started(&self) -> i64 {
+        self.stream_last_started
+    }
+
+    /// The WorkflowTaskStarted event ids of the last completed tasks, newest first, with a
+    /// trailing zero when they go back to the run's start.
+    pub(crate) fn stream_completed_tasks(&self) -> Vec<i64> {
+        let mut ids: Vec<i64> = self.stream_completed.iter().copied().collect();
+        if self.stream_completed_count < STREAM_COMPLETED_TASKS {
+            ids.push(0);
+        }
+        ids
     }
 
     pub(crate) fn prepare_for_wft_response(&mut self) -> MachinesWFTResponseContent<'_> {
@@ -727,6 +757,9 @@ impl WorkflowMachines {
                 EventType::WorkflowTaskCompleted => {
                     self.stream_prefix = self.stream_last_started;
                     self.stream_reruns = 0;
+                    self.stream_completed.push_front(self.stream_last_started);
+                    self.stream_completed.truncate(STREAM_COMPLETED_TASKS);
+                    self.stream_completed_count += 1;
                 }
                 _ => self.stream_reruns += workflow_task_reruns_in(&event),
             }
@@ -1928,3 +1961,28 @@ enum CommandIdKind {
     /// A command which is fire-and-forget (ex: Upsert search attribs)
     NeverResolves,
 }
+
+#[cfg(test)]
+mod stream_attempt_tests {
+    use super::*;
+
+    #[test]
+    fn the_attempt_bits_match_the_shared_vectors() {
+        let vectors: serde_json::Value = serde_json::from_str(include_str!(
+            "../../../../../streams/testdata/workflow_attempts.json"
+        ))
+        .unwrap();
+        assert_eq!(vectors["retry_bits"], STREAM_RETRY_BITS);
+        for case in vectors["attempts"].as_array().unwrap() {
+            let (prefix, reruns) = (
+                case["prefix"].as_i64().unwrap(),
+                case["reruns"].as_i64().unwrap(),
+            );
+            assert_eq!(
+                stream_attempt(prefix, reruns),
+                case["attempt"].as_i64().unwrap(),
+                "{case}"
+            );
+        }
+    }
+}

crates/sdk-core/src/worker/workflow/managed_run.rs

@@ -963,6 +963,8 @@ impl ManagedRun {
             run_id: machines.run_id.clone(),
             first_run_id: machines.first_execution_run_id.clone(),
             attempt,
+            task_started: machines.stream_task_started(),
+            completed_tasks: machines.stream_completed_tasks(),
             topics,
         }))
     }

crates/sdk-core/src/worker/workflow/stream_output.rs

@@ -20,6 +20,10 @@ pub(crate) struct OutputBatch {
     pub(crate) first_run_id: String,
     /// The run's producer attempt for the current Workflow Task, from History.
     pub(crate) attempt: i64,
+    /// The current task's WorkflowTaskStarted event id.
+    pub(crate) task_started: i64,
+    /// The last completed tasks' WorkflowTaskStarted event ids, newest first, from History.
+    pub(crate) completed_tasks: Vec<i64>,
     /// One append per topic, in order of first publish.
     pub(crate) topics: Vec<TopicOutput>,
 }
@@ -207,7 +211,14 @@ impl OutputStore {
                         sequence,
                         &append,
                     )
-                    .map(|stored| stored.encode_to_vec())
+                    .map(|mut stored| {
+                        // The attempt names both, so a repeat of the batch stores the same.
+                        stored.task_started_event_id = batch.task_started;
+                        if sequence == topic.sequence {
+                            stored.completed_tasks = batch.completed_tasks.clone();
+                        }
+                        stored.encode_to_vec()
+                    })
                 })
                 .collect::<Result<Vec<_>, _>>()?;
             let request = StoreAppendRequest {