From 8303b8eeb9454667621aae56b7cb9cfd5b6c041e Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Wed, 14 Oct 2026 15:27:09 -0700 Subject: [PATCH 1/3] Superseded a Workflow producer's records only within one task. Each completed task moves a Workflow producer to a higher attempt. A reader that took that for a retry would withdraw records of tasks that completed. --- crates/sdk-core/CHANGELOG.md | 3 + .../src/core_tests/external_streams.rs | 5 ++ crates/streams/src/reader.rs | 57 ++++++++++++++++++- 3 files changed, 63 insertions(+), 2 deletions(-) diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index 1549ff3815..1c39133ec8 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. ### Breaking Changes * Replay detects changes to the service or operation of scheduled Nexus operations, reporting diff --git a/crates/sdk-core/src/core_tests/external_streams.rs b/crates/sdk-core/src/core_tests/external_streams.rs index a2b7b1adfe..b4c480fef2 100644 --- a/crates/sdk-core/src/core_tests/external_streams.rs +++ b/crates/sdk-core/src/core_tests/external_streams.rs @@ -658,6 +658,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" + ); } /// Replays the first task of `history`, then commits `["c"]` live in the task after it, and diff --git a/crates/streams/src/reader.rs b/crates/streams/src/reader.rs index c70ea74921..c381ea3415 100644 --- a/crates/streams/src/reader.rs +++ b/crates/streams/src/reader.rs @@ -15,6 +15,18 @@ 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 a producer's `attempt` withdraws what its `previous` attempt wrote. A Workflow +/// producer (`@`) moves to a higher attempt with each completed task, and only a rerun +/// of the same task withdraws. Any other producer's higher attempt is a retry. +fn supersedes(producer_id: &str, previous: i64, attempt: i64) -> bool { + !producer_id.starts_with('@') + || previous >> WORKFLOW_RETRY_BITS == attempt >> WORKFLOW_RETRY_BITS +} + /// How many producers a read remembers the newest attempt of. pub const MAX_TRACKED_PRODUCERS: usize = 1_000; @@ -171,7 +183,8 @@ impl ReadState { } tracked.attempt = record.attempt; self.forget_beyond(MAX_TRACKED_PRODUCERS); - Observed::Newer((seen > 0).then(|| Supersession { + let superseded = seen > 0 && supersedes(&record.producer_id, seen, record.attempt); + Observed::Newer(superseded.then(|| Supersession { topic: record.topic.clone(), producer_id: record.producer_id.clone(), previous_attempt: seen, @@ -184,7 +197,8 @@ 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 +/// resumes there reports the supersession again. A Workflow producer's attempt for a later task +/// gets none, since only a rerun of the same task withdraws records. 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 @@ -651,6 +665,45 @@ mod tests { assert!(!kept.producers.contains_key(&local_call(0))); } + #[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(); From 971c86e40f6905d039163aecd2eb70884937f22a Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 16 Oct 2026 03:44:48 -0700 Subject: [PATCH 2/3] Withdrew a failed task's records when its rerun skipped the topic. A Workflow's records now carry their task's started event id and the run's last completed tasks. A reader then learns that an earlier attempt never completed, also when its rerun wrote nothing to the topic. The shared vectors pin the attempt bits and this rule for every SDK. --- .../temporal/sdk/streams/v1/message.proto | 9 + crates/protos/src/protos/mod.rs | 2 + .../src/core_tests/external_streams.rs | 91 ++++++++ .../workflow/machines/workflow_machines.rs | 62 +++++- .../src/worker/workflow/managed_run.rs | 2 + .../src/worker/workflow/stream_output.rs | 13 +- crates/streams/src/reader.rs | 85 ++++++- .../streams/testdata/workflow_attempts.json | 209 ++++++++++++++++++ 8 files changed, 458 insertions(+), 15 deletions(-) create mode 100644 crates/streams/testdata/workflow_attempts.json 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/src/core_tests/external_streams.rs b/crates/sdk-core/src/core_tests/external_streams.rs index ec81c23af7..5348ec36fa 100644 --- a/crates/sdk-core/src/core_tests/external_streams.rs +++ b/crates/sdk-core/src/core_tests/external_streams.rs @@ -907,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 524f5fefee..fa5bdb6ebf 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, } @@ -181,7 +185,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 c381ea3415..6f66c18bd2 100644 --- a/crates/streams/src/reader.rs +++ b/crates/streams/src/reader.rs @@ -19,12 +19,30 @@ use std::collections::HashMap; /// bits name the last task that completed before it. const WORKFLOW_RETRY_BITS: u32 = 16; -/// Whether a producer's `attempt` withdraws what its `previous` attempt wrote. A Workflow -/// producer (`@`) moves to a higher attempt with each completed task, and only a rerun -/// of the same task withdraws. Any other producer's higher attempt is a retry. -fn supersedes(producer_id: &str, previous: i64, attempt: i64) -> bool { - !producer_id.starts_with('@') - || previous >> WORKFLOW_RETRY_BITS == attempt >> WORKFLOW_RETRY_BITS +/// 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. @@ -90,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. @@ -181,9 +203,11 @@ 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); - let superseded = seen > 0 && supersedes(&record.producer_id, seen, record.attempt); + let superseded = seen > 0 && supersedes(seen, started, record); Observed::Newer(superseded.then(|| Supersession { topic: record.topic.clone(), producer_id: record.producer_id.clone(), @@ -198,11 +222,11 @@ 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 Workflow producer's attempt for a later task -/// gets none, since only a rerun of the same task withdraws records. 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 +/// 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`. /// @@ -665,6 +689,43 @@ 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; 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 + } + ] +} From da72d2f81b97eeaa7ada00d9eb9ba9cf2f06bf2e Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 16 Oct 2026 04:22:31 -0700 Subject: [PATCH 3/3] Wrapped a doc comment at 100 characters. --- crates/streams/src/reader.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/crates/streams/src/reader.rs b/crates/streams/src/reader.rs index 6f66c18bd2..f07a759f6c 100644 --- a/crates/streams/src/reader.rs +++ b/crates/streams/src/reader.rs @@ -226,9 +226,9 @@ impl ReadState { /// 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`. +/// 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