diff --git a/crates/protos/protos/local/temporal/sdk/streams/v1/message.proto b/crates/protos/protos/local/temporal/sdk/streams/v1/message.proto index 99a8f0ff42..3bf4af50ff 100644 --- a/crates/protos/protos/local/temporal/sdk/streams/v1/message.proto +++ b/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. diff --git a/crates/protos/src/protos/mod.rs b/crates/protos/src/protos/mod.rs index 26d77ce504..a4c64848a1 100644 --- a/crates/protos/src/protos/mod.rs +++ b/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![], } } diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index ec805e7f2b..b968e32a50 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/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 (`@`) 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 diff --git a/crates/sdk-core/src/core_tests/external_streams.rs b/crates/sdk-core/src/core_tests/external_streams.rs index 9490a0ed96..5348ec36fa 100644 --- a/crates/sdk-core/src/core_tests/external_streams.rs +++ b/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::>(), + [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()); 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 0f2b1ef480..9286d70486 100644 --- a/crates/sdk-core/src/worker/workflow/machines/workflow_machines.rs +++ b/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, + /// 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 { + let mut ids: Vec = 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}" + ); + } + } +} diff --git a/crates/sdk-core/src/worker/workflow/managed_run.rs b/crates/sdk-core/src/worker/workflow/managed_run.rs index 8f59ce7dc9..40b04a8580 100644 --- a/crates/sdk-core/src/worker/workflow/managed_run.rs +++ b/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, })) } diff --git a/crates/sdk-core/src/worker/workflow/stream_output.rs b/crates/sdk-core/src/worker/workflow/stream_output.rs index ac0c759c51..d7332e22c0 100644 --- a/crates/sdk-core/src/worker/workflow/stream_output.rs +++ b/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, /// One append per topic, in order of first publish. pub(crate) topics: Vec, } @@ -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::, _>>()?; let request = StoreAppendRequest { diff --git a/crates/streams/src/reader.rs b/crates/streams/src/reader.rs index c70ea74921..f07a759f6c 100644 --- a/crates/streams/src/reader.rs +++ b/crates/streams/src/reader.rs @@ -15,6 +15,36 @@ use crate::{ use prost::Message; use std::collections::HashMap; +/// The low bits of a Workflow producer's attempt, which count the reruns of one task. The high +/// bits name the last task that completed before it. +const WORKFLOW_RETRY_BITS: u32 = 16; + +/// Whether `record` withdraws what its producer's `previous` attempt wrote. `started` is the +/// WorkflowTaskStarted event id that `previous` carried. +/// +/// Any producer's higher attempt is a retry, except a Workflow producer's (`@`). It moves +/// to a higher attempt with each completed task, so a later task withdraws nothing by itself. A +/// rerun of the same task withdraws. So does a later task whose list of completed tasks shows +/// that another attempt completed `previous`'s task, which happens when the rerun wrote nothing +/// to the topic. When the list doesn't reach back to that task, nothing is withdrawn. +fn supersedes(previous: i64, started: i64, record: &StreamRecord) -> bool { + if !record.producer_id.starts_with('@') { + return true; + } + let prefix = previous >> WORKFLOW_RETRY_BITS; + if prefix == record.attempt >> WORKFLOW_RETRY_BITS { + return true; + } + let completed = &record.completed_tasks; + // The task completed after `prefix` is the one just before it in the newest-first list. + started > 0 + && !completed.contains(&started) + && completed + .iter() + .position(|&id| id == prefix) + .is_some_and(|at| at > 0) +} + /// How many producers a read remembers the newest attempt of. pub const MAX_TRACKED_PRODUCERS: usize = 1_000; @@ -78,6 +108,10 @@ struct Tracked { /// seen one is forgotten first. #[prost(uint64, tag = "2")] seen: u64, + /// The WorkflowTaskStarted event id the newest attempt's records carried, for a Workflow + /// producer. + #[prost(int64, tag = "3")] + started: i64, } /// What one record tells the read about its producer's attempts. @@ -169,9 +203,12 @@ impl ReadState { if record.attempt == seen { return Observed::Known; } + let started = tracked.started; tracked.attempt = record.attempt; + tracked.started = record.task_started_event_id; self.forget_beyond(MAX_TRACKED_PRODUCERS); - Observed::Newer((seen > 0).then(|| Supersession { + let superseded = seen > 0 && supersedes(seen, started, record); + Observed::Newer(superseded.then(|| Supersession { topic: record.topic.clone(), producer_id: record.producer_id.clone(), previous_attempt: seen, @@ -184,13 +221,14 @@ impl ReadState { /// /// The read turns stored records into what lang delivers. A producer's newer attempt gets a /// `SUPERSEDED` record before it, at the cursor of the record delivered before it, so a read that -/// resumes there reports the supersession again. A record from an attempt below the newest the -/// read delivered is still delivered, marked `stale`. The attempts seen ride in `state`, so this -/// holds across calls. A read that starts from a cursor without state looks up the record at the -/// cursor first, since that record was delivered before. Only that record's producer is known -/// then, and another producer's earlier attempts are not. A read remembers at most -/// [MAX_TRACKED_PRODUCERS] producers in at most [MAX_STATE_BYTES], and the next attempt of one it -/// forgot reads without `SUPERSEDED`. +/// resumes there reports the supersession again. A Workflow producer's attempt for a later task +/// gets one only when the earlier attempt's task never completed. A record from an attempt below +/// the newest the read delivered is still delivered, marked `stale`. The attempts seen ride in +/// `state`, so this holds across calls. A read that starts from a cursor without state looks up +/// the record at the cursor first, since that record was delivered before. Only that record's +/// producer is known then, and another producer's earlier attempts are not. A read remembers at +/// most [MAX_TRACKED_PRODUCERS] producers in at most [MAX_STATE_BYTES], and the next attempt of one +/// it forgot reads without `SUPERSEDED`. /// /// A record that does not parse fails the call as `RECORD` with its cursor, so the caller resumes /// past it on purpose. Skipping it would lose data no one hears of. The records before it in the @@ -651,6 +689,82 @@ mod tests { assert!(!kept.producers.contains_key(&local_call(0))); } + /// The vectors every SDK checks its attempt bits and supersession rule against. + fn vectors() -> serde_json::Value { + serde_json::from_str(include_str!("../testdata/workflow_attempts.json")).unwrap() + } + + #[test] + fn a_workflow_producer_withdraws_an_attempt_whose_task_never_completed() { + let vectors = vectors(); + assert_eq!(vectors["retry_bits"], WORKFLOW_RETRY_BITS); + let written = |written: &serde_json::Value| StreamRecord { + task_started_event_id: written["task_started_event_id"].as_i64().unwrap(), + completed_tasks: written["completed_tasks"] + .as_array() + .map(|ids| ids.iter().map(|id| id.as_i64().unwrap()).collect()) + .unwrap_or_default(), + ..data_record("@run", written["attempt"].as_i64().unwrap(), b"") + }; + for case in vectors["supersession"].as_array().unwrap() { + let mut state = ReadState::default(); + state.observe(&written(&case["previous"])); + let observed = state.observe(&written(&case["next"])); + assert_eq!( + matches!(observed, Observed::Newer(Some(_))), + case["superseded"].as_bool().unwrap(), + "{}", + case["label"] + ); + } + for case in vectors["attempts"].as_array().unwrap() { + let attempt = case["attempt"].as_i64().unwrap(); + assert_eq!( + attempt >> WORKFLOW_RETRY_BITS, + case["prefix"].as_i64().unwrap() + ); + } + } + + #[test] + fn a_workflow_producer_supersedes_only_within_one_task() { + let task = |prefix: i64, retry: i64| (prefix << 16) | retry; + let mut state = ReadState::default(); + assert_eq!( + observe(&mut state, "@run", task(0, 1)), + Observed::Newer(None) + ); + // A completed task moves the prefix, so the next task goes on with no supersession. + assert_eq!( + observe(&mut state, "@run", task(3, 1)), + Observed::Newer(None) + ); + assert_eq!( + observe(&mut state, "@run", task(3, 2)), + Observed::Newer(Some(Supersession { + topic: "out".to_string(), + producer_id: "@run".to_string(), + previous_attempt: task(3, 1), + attempt: task(3, 2), + })) + ); + assert_eq!( + observe(&mut state, "@run", task(9, 1)), + Observed::Newer(None) + ); + // An earlier task's record after a later one is still behind. + assert_eq!( + observe(&mut state, "@run", task(3, 2)), + Observed::Behind { newest: task(9, 1) } + ); + // Any other producer's higher attempt is a retry, whatever its bits hold. + observe(&mut state, "model", task(0, 1)); + assert!(matches!( + observe(&mut state, "model", task(3, 1)), + Observed::Newer(Some(_)) + )); + } + #[test] fn a_repeat_of_the_current_attempt_is_not_reported() { let mut state = ReadState::default(); diff --git a/crates/streams/testdata/workflow_attempts.json b/crates/streams/testdata/workflow_attempts.json new file mode 100644 index 0000000000..7901278088 --- /dev/null +++ b/crates/streams/testdata/workflow_attempts.json @@ -0,0 +1,209 @@ +{ + "retry_bits": 16, + "attempts": [ + { + "prefix": 0, + "reruns": 0, + "attempt": 1 + }, + { + "prefix": 3, + "reruns": 0, + "attempt": 196609 + }, + { + "prefix": 3, + "reruns": 1, + "attempt": 196610 + }, + { + "prefix": 5, + "reruns": 65533, + "attempt": 393214 + }, + { + "prefix": 5, + "reruns": 65534, + "attempt": 393215 + }, + { + "prefix": 5, + "reruns": 70000, + "attempt": 393215 + } + ], + "supersession": [ + { + "label": "a rerun of the same task", + "previous": { + "attempt": 196609, + "task_started_event_id": 5 + }, + "next": { + "attempt": 196610, + "task_started_event_id": 9, + "completed_tasks": [ + 3, + 0 + ] + }, + "superseded": true + }, + { + "label": "the next task after the attempt completed", + "previous": { + "attempt": 196609, + "task_started_event_id": 5 + }, + "next": { + "attempt": 327681, + "task_started_event_id": 7, + "completed_tasks": [ + 5, + 3, + 0 + ] + }, + "superseded": false + }, + { + "label": "a rerun that wrote nothing completed, then the next task wrote", + "previous": { + "attempt": 196609, + "task_started_event_id": 5 + }, + "next": { + "attempt": 589825, + "task_started_event_id": 11, + "completed_tasks": [ + 9, + 3, + 0 + ] + }, + "superseded": true + }, + { + "label": "a completed task that wrote nothing came in between", + "previous": { + "attempt": 196609, + "task_started_event_id": 5 + }, + "next": { + "attempt": 458753, + "task_started_event_id": 9, + "completed_tasks": [ + 7, + 5, + 3, + 0 + ] + }, + "superseded": false + }, + { + "label": "a rerun and a later task came in between", + "previous": { + "attempt": 196609, + "task_started_event_id": 5 + }, + "next": { + "attempt": 786433, + "task_started_event_id": 14, + "completed_tasks": [ + 12, + 9, + 3, + 0 + ] + }, + "superseded": true + }, + { + "label": "the earlier attempt's task is older than the list", + "previous": { + "attempt": 196609, + "task_started_event_id": 5 + }, + "next": { + "attempt": 2621441, + "task_started_event_id": 42, + "completed_tasks": [ + 40, + 38, + 36, + 34, + 32, + 30, + 28, + 26 + ] + }, + "superseded": false + }, + { + "label": "the earlier record has no started id", + "previous": { + "attempt": 196609, + "task_started_event_id": 0 + }, + "next": { + "attempt": 589825, + "task_started_event_id": 11, + "completed_tasks": [ + 9, + 3, + 0 + ] + }, + "superseded": false + }, + { + "label": "a rerun of the first task", + "previous": { + "attempt": 1, + "task_started_event_id": 3 + }, + "next": { + "attempt": 2, + "task_started_event_id": 6, + "completed_tasks": [ + 0 + ] + }, + "superseded": true + }, + { + "label": "the first task failed and its rerun wrote nothing", + "previous": { + "attempt": 1, + "task_started_event_id": 3 + }, + "next": { + "attempt": 393217, + "task_started_event_id": 8, + "completed_tasks": [ + 6, + 0 + ] + }, + "superseded": true + }, + { + "label": "the first task completed", + "previous": { + "attempt": 1, + "task_started_event_id": 3 + }, + "next": { + "attempt": 196609, + "task_started_event_id": 5, + "completed_tasks": [ + 3, + 0 + ] + }, + "superseded": false + } + ] +}