From a759d7213921c9adf5efec5a608cbcf6d4cc03cf Mon Sep 17 00:00:00 2001 From: Moe Dashti Date: Mon, 26 Oct 2026 22:11:46 -0700 Subject: [PATCH 01/10] Added the Store trait and its capabilities. A Store declares the rules it applies atomically beyond the log, and the service refuses a call that needs a capability the Store lacks. The trait is small so that the Redis Store and a native Store can implement it too. --- crates/streams/Cargo.toml | 2 + crates/streams/src/lib.rs | 1 + crates/streams/src/store.rs | 349 ++++++++++++++++++++++++++++++++++++ 3 files changed, 352 insertions(+) create mode 100644 crates/streams/src/store.rs diff --git a/crates/streams/Cargo.toml b/crates/streams/Cargo.toml index 277de8918..9afcc64ff 100644 --- a/crates/streams/Cargo.toml +++ b/crates/streams/Cargo.toml @@ -12,10 +12,12 @@ keywords = ["temporal", "workflow", "streams"] categories = ["development-tools"] [dependencies] +async-trait = "0.1" prost = { workspace = true } prost-types = { workspace = true } sha2 = "0.10" thiserror = { workspace = true } +tokio = { version = "1.47", default-features = false, features = ["rt", "sync", "time"] } tonic = { workspace = true } [dependencies.temporalio-common] diff --git a/crates/streams/src/lib.rs b/crates/streams/src/lib.rs index d4879e593..bacac0a12 100644 --- a/crates/streams/src/lib.rs +++ b/crates/streams/src/lib.rs @@ -12,6 +12,7 @@ mod error; pub mod ids; pub mod offset; pub mod protos; +pub mod store; pub use error::{Reason, StreamError, StreamResult}; diff --git a/crates/streams/src/store.rs b/crates/streams/src/store.rs new file mode 100644 index 000000000..5d1f156a2 --- /dev/null +++ b/crates/streams/src/store.rs @@ -0,0 +1,349 @@ +//! The calls that every Store gives the Stream service. The Redis Store of +//! `temporalio-streams-redis` is for production, and the memory Store is for tests. Both apply the +//! same rules. + +use crate::{ + Reason, StreamError, StreamResult, + ids::encode_part, + proto::{Producer, StreamRef, WorkflowTaskEpoch}, +}; +use std::time::Duration; +use temporalio_common::protos::temporal::api::enums::v1::ExecutionType; +use tokio::time::Instant; + +/// The format version this crate writes. A Stream with a higher one refuses writes. +pub const FORMAT_VERSION: u32 = 1; +/// The maximum size of a stored Record, in bytes. +pub const MAX_RECORD_BYTES: usize = 1536 * 1024; +/// The maximum size of a stored append, in bytes. +pub const MAX_BATCH_BYTES: usize = 4 * 1024 * 1024; +/// The highest batch `sequence`. A Lua number is exact for integers only up to this value. +pub const MAX_SEQUENCE: i64 = (1 << 53) - 1; +/// How long the meta key stays after the log key expires, so that a read can tell an expired Stream +/// from a missing one. +pub const TOMBSTONE: Duration = Duration::from_secs(30 * 24 * 60 * 60); +/// The longest Retention or safety TTL, which is the largest proto Duration. Its ms plus the time +/// of a write stay exact in a Lua number. +pub const MAX_TTL: Duration = Duration::from_secs(315_576_000_000); + +/// The owner run chain of a ref as `:`, each part encoded, or empty for +/// another owner. A Store keeps it on a Workflow-owned Stream and checks creates and closes on it. +pub fn owner_chain(stream: &StreamRef) -> String { + stream + .owner + .as_ref() + .filter(|owner| owner.r#type == ExecutionType::Workflow as i32) + .map_or_else(String::new, |owner| { + format!( + "{}:{}", + encode_part(&owner.business_id), + encode_part(&owner.run_id) + ) + }) +} + +/// Who created a Stream, which decides its TTL while it is open. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Owner { + /// Until it closes, the Stream expires `safety_ttl` after its last write. + Workflow, + /// The Stream expires one Retention after its last write. + Client, +} + +/// What a create found. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum CreateOutcome { + /// The create wrote the key. + Created, + /// The Stream exists and is open. The create doesn't change its Retention. + Exists, +} + +/// The result of a create. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct Created { + /// What the create found. + pub outcome: CreateOutcome, + /// The Retention of the Stream, the stored one if it existed. + pub retention: Duration, + /// Whether the create withdrew a retry run's previous run. + pub flagged: bool, +} + +/// The withdrawal facts of a batch, and the `Superseded` item that they make. +#[derive(Debug, Clone, Default)] +pub struct Withdrawal<'a> { + /// A retry run's previous run. + pub withdraw_run: &'a str, + /// The encoded Supersession, which the log keeps when the batch withdraws something. + pub supersession: &'a [u8], + /// The run's task facts of a Local Activity batch, which withdraw the run's orphaned Local + /// Activity Records. + pub tasks: Option>, +} + +/// The run's Workflow Task facts that a Local Activity batch gives. A task of the run after +/// `completed` and before the batch's own task, or one in `failed_tasks`, never completed, so +/// its Local Activity Records are orphans. +#[derive(Debug, Clone, Copy, Default)] +pub struct TaskFacts<'a> { + /// The run's last completed Workflow Task, `None` before the first one. + pub completed: Option<&'a WorkflowTaskEpoch>, + /// The started event ids of the run's failed Workflow Tasks. + pub failed_tasks: &'a [i64], + /// The encoded Workflow-kind Supersession that withdraws the orphans. + pub supersession: &'a [u8], +} + +/// One batch of one Producer. +#[derive(Debug, Clone)] +pub struct Batch<'a> { + /// The Producer, with its epoch. + pub producer: &'a Producer, + /// The batch number in this epoch. + pub sequence: i64, + /// The encoded Records. + pub records: &'a [Vec], + /// The [crate::digest::batch_digest] of the batch, which the dedupe entry keeps. + pub digest: &'a str, + /// What the batch withdraws. + pub withdrawal: Withdrawal<'a>, +} + +/// The result of an append. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct Appended { + /// The offsets of the first and the last entry, `None` for an empty batch with no + /// `Superseded` item. + pub offsets: Option<(i64, i64)>, + /// The Store already had the batch, so the offsets are those of the first result. + pub resend: bool, + /// The batch withdrew earlier Records. + pub flagged: bool, +} + +/// The result of a close. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct Closed { + /// The offset of the close Record. + pub offset: i64, + /// The Stream was already closed. + pub already: bool, +} + +/// The state of a Stream as one read saw it. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct StreamState { + /// The meta key exists, live or as a Tombstone. + pub exists: bool, + /// The format version. + pub version: u32, + /// The offset of the close Record, if the Stream is closed. + pub closed_at: Option, + /// The offset of the newest entry written, zero if none. + pub newest: i64, + /// The offset of the newest entry that Retention dropped, zero if none. + pub trimmed: i64, + /// The Retention. + pub retention: Duration, + /// When the Stream stops taking writes, in ms since 1970. + pub expires_ms: i64, + /// The server time of the read, in ms since 1970. + pub now_ms: i64, +} + +impl StreamState { + /// The Stream is a Tombstone. + pub fn expired(&self) -> bool { + self.exists && self.expires_ms <= self.now_ms + } + + /// The offset below which Retention dropped or hides the Records, as of the read. + pub fn retained_from(&self) -> i64 { + let cut = self.now_ms - self.retention.as_millis() as i64; + let hidden = if cut > 0 { + cut << crate::offset::SEQ_BITS + } else { + 0 + }; + hidden.max(self.trimmed + 1) + } +} + +/// One log entry. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Entry { + /// Its offset. + pub offset: i64, + /// The encoded Record, absent on an entry that only withdraws. + pub record: Option>, + /// The encoded Supersession of a flagged batch. + pub supersession: Option>, +} + +/// The rules that a Store applies atomically beyond the log. The service checks a capability before +/// a call that needs it. So a Store that lacks the capability refuses the call, and doesn't give +/// weaker guarantees without notice. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct Capabilities(u8); + +impl Capabilities { + /// Each append checks the Producer's epoch and its last `sequence` before it writes. + pub const PRODUCER_FENCE: Self = Self(1); + /// Each append flags the first batch of a Producer's newer epoch, which withdraws the + /// Records of its earlier ones. + pub const WITHDRAWAL: Self = Self(1 << 1); + /// A retry run's first write withdraws the previous run and fences its Producers. + pub const RUN_FENCE: Self = Self(1 << 2); + /// A close writes the close Record and every later append is refused. + pub const CLOSE: Self = Self(1 << 3); + /// An open Workflow-owned Stream keeps the safety TTL, not the Retention. + pub const SAFETY_TTL: Self = Self(1 << 4); + /// Every capability. + pub const ALL: Self = Self((1 << 5) - 1); + + const NAMES: [(Self, &'static str); 5] = [ + (Self::PRODUCER_FENCE, "producer_fence"), + (Self::WITHDRAWAL, "withdrawal"), + (Self::RUN_FENCE, "run_fence"), + (Self::CLOSE, "close"), + (Self::SAFETY_TTL, "safety_ttl"), + ]; + + /// Both sets. + pub const fn union(self, other: Self) -> Self { + Self(self.0 | other.0) + } + + /// This set without the other one. + pub const fn without(self, other: Self) -> Self { + Self(self.0 & !other.0) + } + + /// Whether every capability of `other` is in this set. + pub const fn contains(self, other: Self) -> bool { + self.0 & other.0 == other.0 + } + + /// The names of the capabilities in this set. + pub fn names(self) -> Vec<&'static str> { + Self::NAMES + .iter() + .filter(|(capability, _)| self.contains(*capability)) + .map(|(_, name)| *name) + .collect() + } + + /// Refuses a call that needs a capability this set lacks, and names the missing ones. + pub fn require(self, needed: Self) -> StreamResult<()> { + let missing = needed.without(self); + if missing == Self::default() { + return Ok(()); + } + Err(unsupported(&missing.names().join(", "))) + } +} + +fn unsupported(capability: &str) -> StreamError { + StreamError::new( + Reason::Unsupported, + format!("The Store lacks the capability {capability}"), + ) +} + +/// The calls of a Store. Each one is atomic, as one Lua script is on Redis. +/// +/// The required calls make the log contract. They are a create key, an atomic batch append, an +/// ordered read with the Stream's state, stable offsets, prefix Retention and a wait for the next +/// entry. An append applies the rules of the capabilities that the Store declares. The optional +/// calls fail by default. +#[async_trait::async_trait] +pub trait Store: Send + Sync + 'static { + /// What the Store does atomically beyond the log. It has no default, so a Store that wraps + /// another one must say what it forwards. + fn capabilities(&self) -> Capabilities; + + /// The Retention of a Stream whose create sets none. + fn default_retention(&self) -> Duration; + + /// Creates a Stream, or finds it when it is open and has the create's owner. The calls that + /// write take the Stream's whole ref, since the server finds a Workflow's Stream by its owner. + /// With [Capabilities::RUN_FENCE], a retry run's create withdraws its previous run. + async fn create( + &self, + namespace: &str, + stream: &StreamRef, + retention: Option, + owner: Owner, + withdrawal: &Withdrawal<'_>, + ) -> StreamResult; + + /// Appends the whole batch, or nothing of it. The batch's Records get consecutive offsets that + /// end at the last offset of [Appended::offsets], so a caller knows the offset of each one. + async fn append( + &self, + namespace: &str, + stream: &StreamRef, + batch: &Batch<'_>, + ) -> StreamResult; + + /// Reads the Stream's state and up to `count` entries from `from`, inclusive. Zero reads + /// from the start of the log. + async fn read( + &self, + namespace: &str, + stream_id: &str, + from: i64, + count: usize, + ) -> StreamResult<(StreamState, Vec)>; + + /// Waits until the log has an entry after `after`, or `deadline` passes. Returns whether an + /// entry came. It may return `false` before the deadline, and then the caller reads again. + async fn wait_after( + &self, + namespace: &str, + stream_id: &str, + after: i64, + deadline: Instant, + ) -> StreamResult; + + /// Closes a Stream with an encoded close Record. Needs [Capabilities::CLOSE]. + async fn close( + &self, + _namespace: &str, + _stream: &StreamRef, + _producer: &Producer, + _record: &[u8], + ) -> StreamResult { + Err(unsupported("close")) + } + + /// Reads up to `count` entries at or before `before`, newest first, for a read of the newest + /// Records. + async fn read_back( + &self, + _namespace: &str, + _stream_id: &str, + _before: i64, + _count: usize, + ) -> StreamResult> { + Err(unsupported("last")) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn a_missing_capability_is_refused_by_name() { + let some = Capabilities::PRODUCER_FENCE.union(Capabilities::CLOSE); + some.require(Capabilities::CLOSE).unwrap(); + let error = some + .require(Capabilities::CLOSE.union(Capabilities::SAFETY_TTL)) + .unwrap_err(); + assert_eq!(error.reason, Reason::Unsupported); + assert!(error.message.ends_with("safety_ttl"), "{}", error.message); + } +} From 73feb8c071899ad7d99c1dad8f2b18d2bcfb2f1d Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 30 Oct 2026 00:23:35 -0700 Subject: [PATCH 02/10] Removed Store::default_retention, which nothing calls. Each Store applies its own default Retention when a create sets none. --- crates/streams/src/store.rs | 3 --- 1 file changed, 3 deletions(-) diff --git a/crates/streams/src/store.rs b/crates/streams/src/store.rs index 5d1f156a2..da26c461e 100644 --- a/crates/streams/src/store.rs +++ b/crates/streams/src/store.rs @@ -264,9 +264,6 @@ pub trait Store: Send + Sync + 'static { /// another one must say what it forwards. fn capabilities(&self) -> Capabilities; - /// The Retention of a Stream whose create sets none. - fn default_retention(&self) -> Duration; - /// Creates a Stream, or finds it when it is open and has the create's owner. The calls that /// write take the Stream's whole ref, since the server finds a Workflow's Stream by its owner. /// With [Capabilities::RUN_FENCE], a retry run's create withdraws its previous run. From 939e7c603cdafd6ac96ed2cf10421cfc9592b940 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 30 Oct 2026 10:29:57 -0700 Subject: [PATCH 03/10] Added the next_sequence capability for an append with no sequence. A call with no writer state sends no sequence. Only the Store can number such a batch in the same step as its dedupe check, so a Store declares that it does. --- crates/streams/src/store.rs | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/crates/streams/src/store.rs b/crates/streams/src/store.rs index da26c461e..639c8ea62 100644 --- a/crates/streams/src/store.rs +++ b/crates/streams/src/store.rs @@ -101,7 +101,8 @@ pub struct TaskFacts<'a> { pub struct Batch<'a> { /// The Producer, with its epoch. pub producer: &'a Producer, - /// The batch number in this epoch. + /// The batch number in this epoch. Zero takes the next one of the epoch, which needs + /// [Capabilities::NEXT_SEQUENCE]. Such a batch has no dedupe, so its retry writes it again. pub sequence: i64, /// The encoded Records. pub records: &'a [Vec], @@ -200,15 +201,19 @@ impl Capabilities { pub const CLOSE: Self = Self(1 << 3); /// An open Workflow-owned Stream keeps the safety TTL, not the Retention. pub const SAFETY_TTL: Self = Self(1 << 4); + /// An append with sequence zero takes the next one of the Producer's epoch, in the same atomic + /// step as the dedupe check. So two such batches never get the same number. + pub const NEXT_SEQUENCE: Self = Self(1 << 5); /// Every capability. - pub const ALL: Self = Self((1 << 5) - 1); + pub const ALL: Self = Self((1 << 6) - 1); - const NAMES: [(Self, &'static str); 5] = [ + const NAMES: [(Self, &'static str); 6] = [ (Self::PRODUCER_FENCE, "producer_fence"), (Self::WITHDRAWAL, "withdrawal"), (Self::RUN_FENCE, "run_fence"), (Self::CLOSE, "close"), (Self::SAFETY_TTL, "safety_ttl"), + (Self::NEXT_SEQUENCE, "next_sequence"), ]; /// Both sets. From ebbadcd6964e347b3f7924099dc661e60483f4ea Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sat, 31 Oct 2026 00:58:55 -0700 Subject: [PATCH 04/10] Renamed Supersession to Superseded, and StreamRef to StreamHandle. --- crates/streams/src/store.rs | 33 +++++++++++++++++---------------- 1 file changed, 17 insertions(+), 16 deletions(-) diff --git a/crates/streams/src/store.rs b/crates/streams/src/store.rs index 639c8ea62..c54534455 100644 --- a/crates/streams/src/store.rs +++ b/crates/streams/src/store.rs @@ -5,7 +5,7 @@ use crate::{ Reason, StreamError, StreamResult, ids::encode_part, - proto::{Producer, StreamRef, WorkflowTaskEpoch}, + proto::{Producer, StreamHandle, WorkflowTaskEpoch}, }; use std::time::Duration; use temporalio_common::protos::temporal::api::enums::v1::ExecutionType; @@ -26,10 +26,11 @@ pub const TOMBSTONE: Duration = Duration::from_secs(30 * 24 * 60 * 60); /// of a write stay exact in a Lua number. pub const MAX_TTL: Duration = Duration::from_secs(315_576_000_000); -/// The owner run chain of a ref as `:`, each part encoded, or empty for -/// another owner. A Store keeps it on a Workflow-owned Stream and checks creates and closes on it. -pub fn owner_chain(stream: &StreamRef) -> String { - stream +/// The owner run chain of a handle as `:`, each part encoded, or empty +/// for another owner. A Store keeps it on a Workflow-owned Stream and checks creates and closes on +/// it. +pub fn owner_chain(handle: &StreamHandle) -> String { + handle .owner .as_ref() .filter(|owner| owner.r#type == ExecutionType::Workflow as i32) @@ -76,8 +77,8 @@ pub struct Created { pub struct Withdrawal<'a> { /// A retry run's previous run. pub withdraw_run: &'a str, - /// The encoded Supersession, which the log keeps when the batch withdraws something. - pub supersession: &'a [u8], + /// The encoded `Superseded` facts, which the log keeps when the batch withdraws something. + pub superseded: &'a [u8], /// The run's task facts of a Local Activity batch, which withdraw the run's orphaned Local /// Activity Records. pub tasks: Option>, @@ -92,8 +93,8 @@ pub struct TaskFacts<'a> { pub completed: Option<&'a WorkflowTaskEpoch>, /// The started event ids of the run's failed Workflow Tasks. pub failed_tasks: &'a [i64], - /// The encoded Workflow-kind Supersession that withdraws the orphans. - pub supersession: &'a [u8], + /// The encoded Workflow-kind `Superseded` facts that withdraw the orphans. + pub superseded: &'a [u8], } /// One batch of one Producer. @@ -179,8 +180,8 @@ pub struct Entry { pub offset: i64, /// The encoded Record, absent on an entry that only withdraws. pub record: Option>, - /// The encoded Supersession of a flagged batch. - pub supersession: Option>, + /// The encoded `Superseded` facts of a flagged batch. + pub superseded: Option>, } /// The rules that a Store applies atomically beyond the log. The service checks a capability before @@ -270,12 +271,12 @@ pub trait Store: Send + Sync + 'static { fn capabilities(&self) -> Capabilities; /// Creates a Stream, or finds it when it is open and has the create's owner. The calls that - /// write take the Stream's whole ref, since the server finds a Workflow's Stream by its owner. - /// With [Capabilities::RUN_FENCE], a retry run's create withdraws its previous run. + /// write take the Stream's whole handle, since the server finds a Workflow's Stream by its + /// owner. With [Capabilities::RUN_FENCE], a retry run's create withdraws its previous run. async fn create( &self, namespace: &str, - stream: &StreamRef, + handle: &StreamHandle, retention: Option, owner: Owner, withdrawal: &Withdrawal<'_>, @@ -286,7 +287,7 @@ pub trait Store: Send + Sync + 'static { async fn append( &self, namespace: &str, - stream: &StreamRef, + handle: &StreamHandle, batch: &Batch<'_>, ) -> StreamResult; @@ -314,7 +315,7 @@ pub trait Store: Send + Sync + 'static { async fn close( &self, _namespace: &str, - _stream: &StreamRef, + _handle: &StreamHandle, _producer: &Producer, _record: &[u8], ) -> StreamResult { From b26552bfadc015d293512370391549ad9251c873 Mon Sep 17 00:00:00 2001 From: Moe Dashti Date: Wed, 4 Nov 2026 10:49:59 -0700 Subject: [PATCH 05/10] Took a batch's digest as its 16 bytes. The batch digest is binary, so a Store gets the bytes that it keeps in the dedupe entry. --- crates/streams/src/store.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/streams/src/store.rs b/crates/streams/src/store.rs index c54534455..049951ddd 100644 --- a/crates/streams/src/store.rs +++ b/crates/streams/src/store.rs @@ -108,7 +108,7 @@ pub struct Batch<'a> { /// The encoded Records. pub records: &'a [Vec], /// The [crate::digest::batch_digest] of the batch, which the dedupe entry keeps. - pub digest: &'a str, + pub digest: [u8; crate::digest::DIGEST_LEN], /// What the batch withdraws. pub withdrawal: Withdrawal<'a>, } From fd25ab3e666ce6af678ed1aecdd4be0d474412da Mon Sep 17 00:00:00 2001 From: Moe Dashti Date: Wed, 4 Nov 2026 17:51:09 -0700 Subject: [PATCH 06/10] Gave a Store's create a conflict policy and a request id. A create of an existing Stream fails unless it uses that Stream or retries the create that made it. The create's inputs are one struct, so the call stays short. --- crates/streams/src/store.rs | 52 ++++++++++++++++++++++++++++++------- 1 file changed, 42 insertions(+), 10 deletions(-) diff --git a/crates/streams/src/store.rs b/crates/streams/src/store.rs index 049951ddd..6f714199f 100644 --- a/crates/streams/src/store.rs +++ b/crates/streams/src/store.rs @@ -44,20 +44,49 @@ pub fn owner_chain(handle: &StreamHandle) -> String { } /// Who created a Stream, which decides its TTL while it is open. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] pub enum Owner { /// Until it closes, the Stream expires `safety_ttl` after its last write. Workflow, - /// The Stream expires one Retention after its last write. + /// The Stream has no owner. It expires one Retention after its last write. + #[default] Client, } +/// What a create does when its Stream exists and is open, and the create isn't a retry of the +/// create that made the Stream. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub enum ConflictPolicy { + /// The create fails with [Reason::AlreadyExists]. + #[default] + Fail, + /// The create gets the open Stream, which keeps its own Retention. + UseExisting, +} + +/// A create of a Stream. +#[derive(Debug, Clone, Default)] +pub struct Create<'a> { + /// The Retention, `None` for the Store default. + pub retention: Option, + /// Who creates the Stream. + pub owner: Owner, + /// What the create does when the Stream exists. + pub conflict: ConflictPolicy, + /// Unique for each create call, and the same on its retries. Empty for none. The Store keeps + /// the request id of the create that made the Stream, so a retry after a lost answer succeeds. + pub request_id: &'a str, + /// What a retry run's create withdraws. + pub withdrawal: Withdrawal<'a>, +} + /// What a create found. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum CreateOutcome { - /// The create wrote the key. + /// The create wrote the key, or it is a retry of the create that wrote it. Created, - /// The Stream exists and is open. The create doesn't change its Retention. + /// The Stream exists and is open, and the create uses it. The create doesn't change its + /// Retention. Exists, } @@ -270,16 +299,19 @@ pub trait Store: Send + Sync + 'static { /// another one must say what it forwards. fn capabilities(&self) -> Capabilities; - /// Creates a Stream, or finds it when it is open and has the create's owner. The calls that - /// write take the Stream's whole handle, since the server finds a Workflow's Stream by its - /// owner. With [Capabilities::RUN_FENCE], a retry run's create withdraws its previous run. + /// Creates a Stream. The calls that write take the Stream's whole handle, since the server + /// finds a Workflow's Stream by its owner. + /// + /// The create of an existing Stream fails with [Reason::Expired], [Reason::NotOwner] or + /// [Reason::Closed], in this order. Then a retry of the create that made the Stream, with the + /// same request id, gets [CreateOutcome::Created] and changes nothing. Any other create + /// applies its [ConflictPolicy]. With [Capabilities::RUN_FENCE], a retry run's create + /// withdraws its previous run. async fn create( &self, namespace: &str, handle: &StreamHandle, - retention: Option, - owner: Owner, - withdrawal: &Withdrawal<'_>, + create: &Create<'_>, ) -> StreamResult; /// Appends the whole batch, or nothing of it. The batch's Records get consecutive offsets that From e68c219702188ef91dbdb9f1de1c976e6436df73 Mon Sep 17 00:00:00 2001 From: Moe Dashti Date: Wed, 4 Nov 2026 22:34:57 -0700 Subject: [PATCH 07/10] Made a Store's close mark the Stream closed with no Record. A close writes no log entry, so the state says the Stream is closed and its newest offset is final. A wait for the next entry also ends at a close. --- crates/streams/src/store.rs | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-) diff --git a/crates/streams/src/store.rs b/crates/streams/src/store.rs index 6f714199f..7df2f2ffe 100644 --- a/crates/streams/src/store.rs +++ b/crates/streams/src/store.rs @@ -157,8 +157,6 @@ pub struct Appended { /// The result of a close. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct Closed { - /// The offset of the close Record. - pub offset: i64, /// The Stream was already closed. pub already: bool, } @@ -170,8 +168,8 @@ pub struct StreamState { pub exists: bool, /// The format version. pub version: u32, - /// The offset of the close Record, if the Stream is closed. - pub closed_at: Option, + /// The Stream is closed. Every later append fails, so `newest` is final. + pub closed: bool, /// The offset of the newest entry written, zero if none. pub newest: i64, /// The offset of the newest entry that Retention dropped, zero if none. @@ -227,7 +225,8 @@ impl Capabilities { pub const WITHDRAWAL: Self = Self(1 << 1); /// A retry run's first write withdraws the previous run and fences its Producers. pub const RUN_FENCE: Self = Self(1 << 2); - /// A close writes the close Record and every later append is refused. + /// A close marks the Stream closed and wakes the waiting reads. Every later append is + /// refused. pub const CLOSE: Self = Self(1 << 3); /// An open Workflow-owned Stream keeps the safety TTL, not the Retention. pub const SAFETY_TTL: Self = Self(1 << 4); @@ -333,8 +332,9 @@ pub trait Store: Send + Sync + 'static { count: usize, ) -> StreamResult<(StreamState, Vec)>; - /// Waits until the log has an entry after `after`, or `deadline` passes. Returns whether an - /// entry came. It may return `false` before the deadline, and then the caller reads again. + /// Waits until the log has an entry after `after`, the Stream is closed, or `deadline` passes. + /// Returns whether an entry came or the Stream is closed. It may return `false` before the + /// deadline, and then the caller reads again. async fn wait_after( &self, namespace: &str, @@ -343,13 +343,13 @@ pub trait Store: Send + Sync + 'static { deadline: Instant, ) -> StreamResult; - /// Closes a Stream with an encoded close Record. Needs [Capabilities::CLOSE]. + /// Closes a Stream. A close writes no log entry, and a close of a closed Stream changes + /// nothing. Needs [Capabilities::CLOSE]. async fn close( &self, _namespace: &str, _handle: &StreamHandle, _producer: &Producer, - _record: &[u8], ) -> StreamResult { Err(unsupported("close")) } From fb05230ec1b1888fbbdf89a59830ca20f6b54d1b Mon Sep 17 00:00:00 2001 From: Moe Dashti Date: Wed, 4 Nov 2026 22:46:51 -0700 Subject: [PATCH 08/10] Took the Store's Stream as a StreamRef in a stream_ref parameter. The proto type is now StreamRef. A bare ref is a Rust keyword, so the parameter is stream_ref. --- crates/streams/src/store.rs | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/crates/streams/src/store.rs b/crates/streams/src/store.rs index 7df2f2ffe..97bac1055 100644 --- a/crates/streams/src/store.rs +++ b/crates/streams/src/store.rs @@ -5,7 +5,7 @@ use crate::{ Reason, StreamError, StreamResult, ids::encode_part, - proto::{Producer, StreamHandle, WorkflowTaskEpoch}, + proto::{Producer, StreamRef, WorkflowTaskEpoch}, }; use std::time::Duration; use temporalio_common::protos::temporal::api::enums::v1::ExecutionType; @@ -26,11 +26,11 @@ pub const TOMBSTONE: Duration = Duration::from_secs(30 * 24 * 60 * 60); /// of a write stay exact in a Lua number. pub const MAX_TTL: Duration = Duration::from_secs(315_576_000_000); -/// The owner run chain of a handle as `:`, each part encoded, or empty +/// The owner run chain of a ref as `:`, each part encoded, or empty /// for another owner. A Store keeps it on a Workflow-owned Stream and checks creates and closes on /// it. -pub fn owner_chain(handle: &StreamHandle) -> String { - handle +pub fn owner_chain(stream_ref: &StreamRef) -> String { + stream_ref .owner .as_ref() .filter(|owner| owner.r#type == ExecutionType::Workflow as i32) @@ -298,7 +298,7 @@ pub trait Store: Send + Sync + 'static { /// another one must say what it forwards. fn capabilities(&self) -> Capabilities; - /// Creates a Stream. The calls that write take the Stream's whole handle, since the server + /// Creates a Stream. The calls that write take the Stream's whole ref, since the server /// finds a Workflow's Stream by its owner. /// /// The create of an existing Stream fails with [Reason::Expired], [Reason::NotOwner] or @@ -309,7 +309,7 @@ pub trait Store: Send + Sync + 'static { async fn create( &self, namespace: &str, - handle: &StreamHandle, + stream_ref: &StreamRef, create: &Create<'_>, ) -> StreamResult; @@ -318,7 +318,7 @@ pub trait Store: Send + Sync + 'static { async fn append( &self, namespace: &str, - handle: &StreamHandle, + stream_ref: &StreamRef, batch: &Batch<'_>, ) -> StreamResult; @@ -348,7 +348,7 @@ pub trait Store: Send + Sync + 'static { async fn close( &self, _namespace: &str, - _handle: &StreamHandle, + _stream_ref: &StreamRef, _producer: &Producer, ) -> StreamResult { Err(unsupported("close")) From 2a07d1d2ab9467139a230a011da3e358b4f91e14 Mon Sep 17 00:00:00 2001 From: Moe Dashti Date: Fri, 6 Nov 2026 17:13:44 -0700 Subject: [PATCH 09/10] Put the Producer kind and the writer's withdrawal into variants. The review asked to stop representing invalid states. The Producer kind is now a oneof with run_id only in the run-based variants, so a client can't carry a run. A write's task facts and withdraw_run move into RunWithdrawal, which only a run writer sets. --- .../temporal/api/stream/v1/message.proto | 102 +++++++++++------- .../streamservice/v1/request_response.proto | 2 +- .../temporal/sdk/core/streams/streams.proto | 7 +- 3 files changed, 68 insertions(+), 43 deletions(-) diff --git a/crates/protos/protos/api_upstream/temporal/api/stream/v1/message.proto b/crates/protos/protos/api_upstream/temporal/api/stream/v1/message.proto index 63bca5f6c..7385028a1 100644 --- a/crates/protos/protos/api_upstream/temporal/api/stream/v1/message.proto +++ b/crates/protos/protos/api_upstream/temporal/api/stream/v1/message.proto @@ -29,27 +29,6 @@ message StreamRef { string backend = 3; } -// Experimental. Might change in the future. -// The ordered attempt of a Producer. A higher epoch supersedes the Records of a lower epoch of the -// same Producer. The case is the kind of the Producer. It decides how the Store orders the epochs, -// and which earlier Records a higher epoch withdraws. -message ProducerEpoch { - oneof kind { - // A Workflow run, which writes in its Workflow Task. A higher epoch withdraws the Records - // of the run's Workflow Tasks that never completed, also those of their Local Activities. - WorkflowTaskEpoch workflow = 1; - // A Local Activity. A higher epoch withdraws the Producer's lower epochs. - LocalActivityEpoch local_activity = 2; - // An Activity of a Workflow. A higher epoch withdraws the Producer's lower epochs. - ActivityEpoch activity = 3; - // A Standalone Activity. A higher epoch withdraws the Producer's lower epochs. - ActivityEpoch standalone_activity = 4; - // A client. A higher epoch only fences the lower ones, since a restarted client's earlier - // Records are real messages, not the Records of a failed attempt. - ClientEpoch client = 5; - } -} - // Experimental. Might change in the future. // One Workflow Task attempt of a run. It orders by started_event_id, then attempt, then // started_time. @@ -86,32 +65,78 @@ message ClientEpoch { } // Experimental. Might change in the future. -// The code that wrote a batch. Its id stays the same across attempts, and its epoch orders them. +// A Workflow run Producer. It writes in its Workflow Task. A higher epoch withdraws the Records of +// the run's Workflow Tasks that never completed, with those of their Local Activities. +message WorkflowProducer { + // The first execution run id of the run chain. + string run_id = 1; + WorkflowTaskEpoch epoch = 2; +} + +// Experimental. Might change in the future. +// A Local Activity Producer. A higher epoch withdraws the Producer's lower epochs. +message LocalActivityProducer { + string run_id = 1; + LocalActivityEpoch epoch = 2; +} + +// Experimental. Might change in the future. +// An Activity Producer, for an Activity of a Workflow or a Standalone Activity. A higher epoch +// withdraws the Producer's lower epochs. +message ActivityProducer { + string run_id = 1; + ActivityEpoch epoch = 2; +} + +// Experimental. Might change in the future. +// A client Producer. A higher epoch only fences the lower ones, since a restarted client's earlier +// Records are real messages, not the Records of a failed attempt. A client has no run. +message ClientProducer { + ClientEpoch epoch = 1; +} + +// Experimental. Might change in the future. +// The code that wrote a batch. Its id stays the same across attempts, and the kind's case orders +// its epochs and picks which earlier Records a higher epoch withdraws. A run-based kind carries its +// run, so a client can't. message Producer { // Stable across attempts, such as `wf:` for the Workflow Producer of a run. string id = 1; - // Required. Its case gives the kind of the Producer. - ProducerEpoch epoch = 2; - // The run of the Producer, so nobody parses the id. Empty for a client. - string run_id = 3; + // Required. The case is the kind of the Producer. + oneof kind { + WorkflowProducer workflow = 2; + LocalActivityProducer local_activity = 3; + ActivityProducer activity = 4; + ActivityProducer standalone_activity = 5; + ClientProducer client = 6; + } } // Experimental. Might change in the future. // The Producer of a write, and which earlier Records the write withdraws. An append, a close and // a writer carry the same fields, so the Store orders and withdraws each write the same way. message StreamWriter { - // Required on a write. Its epoch's case picks the withdrawal rule. + // Required on a write. Its Producer's kind picks the withdrawal rule. Producer producer = 1; - // Workflow and Local Activity only. The run's last completed Workflow Task, from a Local - // Activity's start. Its started_event_id is 0 before the first. The write withdraws the run's - // Local Activity Records whose Workflow Task never completed. - WorkflowTaskEpoch previous_completed_task = 2; - // Workflow and Local Activity only. The started event ids of the run's failed or timed out - // Workflow Tasks. The write withdraws the run's Local Activity Records of these tasks. - repeated int64 failed_tasks = 3; - // Retry run only. The previous run, on the retry run's first write. Every Record whose - // Producer carries it is withdrawn. - string withdraw_run = 4; + // Set only by a Workflow or Local Activity write, since only a run withdraws orphaned Local + // Activity Records or a previous run. Unset for an Activity or client write. + RunWithdrawal withdrawal = 2; +} + +// Experimental. Might change in the future. +// What a Workflow or Local Activity write withdraws from its run. One write can carry both the task +// facts and a run to withdraw, so these are fields of one message and not a oneof. +message RunWithdrawal { + // The run's last completed Workflow Task, from a Local Activity's start. Its started_event_id + // is 0 before the first. The write withdraws the run's Local Activity Records whose Workflow + // Task never completed. + WorkflowTaskEpoch previous_completed_task = 1; + // The started event ids of the run's failed or timed out Workflow Tasks. The write withdraws + // the run's Local Activity Records of these tasks. + repeated int64 failed_tasks = 2; + // Retry run only. The previous run, on the retry run's first write. Every Record whose Producer + // carries it is withdrawn. + string withdraw_run = 3; } // Experimental. Might change in the future. @@ -125,8 +150,7 @@ message AppendRecord { // batch of a higher epoch, or for a retry run's first write, and a read builds a Superseded item // from them. A reader drops the earlier Records that the facts match. message StreamSuperseded { - // The writer of the batch that withdraws. Its Producer's epoch is empty when only - // withdraw_run applies. + // The writer of the batch that withdraws, with its withdrawal facts. StreamWriter writer = 1; } diff --git a/crates/protos/protos/api_upstream/temporal/api/streamservice/v1/request_response.proto b/crates/protos/protos/api_upstream/temporal/api/streamservice/v1/request_response.proto index cdd61456e..463f91194 100644 --- a/crates/protos/protos/api_upstream/temporal/api/streamservice/v1/request_response.proto +++ b/crates/protos/protos/api_upstream/temporal/api/streamservice/v1/request_response.proto @@ -75,7 +75,7 @@ message AppendRecordsRequest { int64 sequence = 4; repeated temporal.api.stream.v1.AppendRecord records = 5; // Only read by the server, on a Workflow Producer's batch. The batch withdraws no Workflow - // Task and keeps no task facts, but the writer's withdraw_run still applies. + // Task and keeps no task facts, but the writer's withdrawal withdraw_run still applies. bool defer_withdrawal = 6; // Only read by the server. The server checks the Producer against the Activity's current // attempt, so nobody else writes as that Activity. diff --git a/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto b/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto index 0c9b198b7..050ac393a 100644 --- a/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto +++ b/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto @@ -20,9 +20,10 @@ import "temporal/api/stream/v1/message.proto"; message OpenWriterRequest { string namespace = 1; temporal.api.stream.v1.StreamRef ref = 2; - // Each batch sends it as AppendRecords does. For a Local Activity it has the run's task facts, - // so the writes withdraw the run's orphaned Local Activity Records. Its withdraw_run must be - // empty, since only a retry run's first append withdraws a run, and a writer sends many. + // Each batch sends it as AppendRecords does. For a Local Activity its withdrawal has the run's + // task facts, so the writes withdraw the run's orphaned Local Activity Records. The withdrawal's + // withdraw_run must be empty, since only a retry run's first append withdraws a run, and a + // writer sends many. temporal.api.stream.v1.StreamWriter writer = 3; // A write sends at most this many Records in one batch. Zero is 1000. int32 max_batch_size = 4; From 21f92479b048a39e47571c7dd2142932d4b8723c Mon Sep 17 00:00:00 2001 From: Moe Dashti Date: Fri, 6 Nov 2026 18:28:46 -0700 Subject: [PATCH 10/10] Read the Producer kind through the oneof. --- crates/streams/src/epoch.rs | 127 ++++++++++++++++++++------------ crates/streams/src/ids.rs | 50 ++++++++----- crates/streams/tests/vectors.rs | 97 ++++++++++++++---------- 3 files changed, 173 insertions(+), 101 deletions(-) diff --git a/crates/streams/src/epoch.rs b/crates/streams/src/epoch.rs index 697d66ef6..15b1ca6a2 100644 --- a/crates/streams/src/epoch.rs +++ b/crates/streams/src/epoch.rs @@ -7,8 +7,8 @@ use crate::{ StreamError, StreamResult, proto::{ - Producer, ProducerEpoch, StreamRecord, StreamSuperseded, WorkflowTaskEpoch, - producer_epoch::Kind, stream_record::Item, + Producer, StreamRecord, StreamSuperseded, StreamWriter, WorkflowTaskEpoch, producer::Kind, + stream_record::Item, }, }; use prost_types::Timestamp; @@ -42,22 +42,52 @@ impl From<&Kind> for ProducerKind { } } -/// Refuses a Producer with no epoch, or with an epoch that has no case. +/// Refuses a Producer with no kind case. impl TryFrom<&Producer> for ProducerKind { type Error = StreamError; fn try_from(producer: &Producer) -> StreamResult { - let epoch = producer - .epoch - .as_ref() - .ok_or_else(|| StreamError::invalid("The Producer has no epoch"))?; - kind(epoch).ok_or_else(|| StreamError::invalid("The Producer's epoch has no kind")) + kind_of(producer).ok_or_else(|| StreamError::invalid("The Producer has no kind")) } } -/// The kind of an epoch's case, `None` when the epoch has no case. -fn kind(epoch: &ProducerEpoch) -> Option { - epoch.kind.as_ref().map(ProducerKind::from) +/// The kind case of a Producer, `None` when it has none. +pub fn kind_of(producer: &Producer) -> Option { + producer.kind.as_ref().map(ProducerKind::from) +} + +/// The run of a Producer. Empty for a client, which has no run. +pub fn producer_run_id(producer: &Producer) -> &str { + match &producer.kind { + Some(Kind::Workflow(w)) => &w.run_id, + Some(Kind::LocalActivity(l)) => &l.run_id, + Some(Kind::Activity(a) | Kind::StandaloneActivity(a)) => &a.run_id, + Some(Kind::Client(_)) | None => "", + } +} + +/// The run that a writer withdraws, empty for none. +pub fn writer_withdraw_run(writer: &StreamWriter) -> &str { + writer + .withdrawal + .as_ref() + .map_or("", |w| w.withdraw_run.as_str()) +} + +/// The started event ids of the run's failed Workflow Tasks that a writer reports. +pub fn writer_failed_tasks(writer: &StreamWriter) -> &[i64] { + writer + .withdrawal + .as_ref() + .map_or(&[], |w| w.failed_tasks.as_slice()) +} + +/// The run's last completed Workflow Task that a writer reports, `None` for none. +pub fn writer_previous_completed_task(writer: &StreamWriter) -> Option<&WorkflowTaskEpoch> { + writer + .withdrawal + .as_ref() + .and_then(|w| w.previous_completed_task.as_ref()) } /// The length of a [task_key]. @@ -92,20 +122,24 @@ pub fn task_key(task: &WorkflowTaskEpoch) -> [u8; TASK_KEY_LEN] { } /// The key of an epoch. It is the [task_of] key, then the attempt's scheduled seconds and nanos -/// and the attempt, in 8, 4 and 4 bytes. A field that the case doesn't have is zero, and an epoch +/// and the attempt, in 8, 4 and 4 bytes. A field that the case doesn't have is zero, and a Producer /// with no case is all zeros. -pub fn epoch_key(epoch: &ProducerEpoch) -> [u8; EPOCH_KEY_LEN] { - let (scheduled, attempt) = match &epoch.kind { - Some(Kind::LocalActivity(local)) => (None, local.attempt), +pub fn epoch_key(kind: Option<&Kind>) -> [u8; EPOCH_KEY_LEN] { + let (scheduled, attempt) = match kind { + Some(Kind::LocalActivity(local)) => (None, local.epoch.map_or(0, |e| e.attempt)), Some(Kind::Activity(activity) | Kind::StandaloneActivity(activity)) => { - (activity.attempt_scheduled_time.as_ref(), activity.attempt) + let epoch = activity.epoch; + ( + epoch.and_then(|e| e.attempt_scheduled_time), + epoch.map_or(0, |e| e.attempt), + ) } - Some(Kind::Client(client)) => (None, client.attempt), + Some(Kind::Client(client)) => (None, client.epoch.map_or(0, |e| e.attempt)), Some(Kind::Workflow(_)) | None => (None, 0), }; - let (seconds, nanos) = time_fields(scheduled); + let (seconds, nanos) = time_fields(scheduled.as_ref()); let mut key = [0; EPOCH_KEY_LEN]; - key[..TASK_KEY_LEN].copy_from_slice(&task_key(&task_of(epoch))); + key[..TASK_KEY_LEN].copy_from_slice(&task_key(&task_of(kind))); key[24..32].copy_from_slice(&be64(seconds)); key[32..36].copy_from_slice(&be32(nanos)); key[36..].copy_from_slice(&be32(attempt)); @@ -114,16 +148,16 @@ pub fn epoch_key(epoch: &ProducerEpoch) -> [u8; EPOCH_KEY_LEN] { /// The Workflow Task of an epoch. It is a Workflow's own task, or the task that dispatched a Local /// Activity, and zero for the other kinds. -pub fn task_of(epoch: &ProducerEpoch) -> WorkflowTaskEpoch { - match &epoch.kind { - Some(Kind::Workflow(task)) => *task, - Some(Kind::LocalActivity(local)) => local.task.unwrap_or_default(), +pub fn task_of(kind: Option<&Kind>) -> WorkflowTaskEpoch { + match kind { + Some(Kind::Workflow(w)) => w.epoch.unwrap_or_default(), + Some(Kind::LocalActivity(local)) => local.epoch.and_then(|e| e.task).unwrap_or_default(), _ => WorkflowTaskEpoch::default(), } } /// Orders two epochs field by field. -pub fn compare_epochs(a: &ProducerEpoch, b: &ProducerEpoch) -> Ordering { +pub fn compare_epochs(a: Option<&Kind>, b: Option<&Kind>) -> Ordering { epoch_key(a).cmp(&epoch_key(b)) } @@ -150,48 +184,52 @@ pub fn withdraws(superseded: &StreamSuperseded, record: &StreamRecord) -> bool { let Some(producer) = data.producer.as_ref() else { return false; }; - if !writer.withdraw_run.is_empty() && producer.run_id == writer.withdraw_run { + let withdraw_run = writer_withdraw_run(writer); + if !withdraw_run.is_empty() && producer_run_id(producer) == withdraw_run { return true; } let Some(withdrawing) = writer.producer.as_ref() else { return false; }; - let (Some(epoch), Some(old)) = (withdrawing.epoch.as_ref(), producer.epoch.as_ref()) else { + let (new, old) = (withdrawing.kind.as_ref(), producer.kind.as_ref()); + let (Some(new_kind), Some(old_kind)) = (new, old) else { return false; }; - match kind(epoch) { - None | Some(ProducerKind::Client) => return false, - Some(ProducerKind::Workflow) => {} + match ProducerKind::from(new_kind) { + ProducerKind::Client => return false, + ProducerKind::Workflow => {} same => { - return kind(old) == same + return ProducerKind::from(old_kind) == same && producer.id == withdrawing.id - && compare_epochs(old, epoch) == Ordering::Less; + && compare_epochs(old, new) == Ordering::Less; } } if !of_run(producer, withdrawing) { return false; } - let (task, old) = (task_of(epoch), task_of(old)); - let started = old.started_event_id; - if writer.failed_tasks.contains(&started) { + let (task, old_task) = (task_of(new), task_of(old)); + let started = old_task.started_event_id; + if writer_failed_tasks(writer).contains(&started) { return true; } - let completed = writer.previous_completed_task.as_ref(); + let completed = writer_previous_completed_task(writer); let after = completed.map_or(0, |c| c.started_event_id); if after < started && started < task.started_event_id { return true; } let lower_at = |at: &WorkflowTaskEpoch| { - started == at.started_event_id && compare_tasks(&old, at) == Ordering::Less + started == at.started_event_id && compare_tasks(&old_task, at) == Ordering::Less }; lower_at(&task) || completed.is_some_and(lower_at) } /// Whether a Producer is the Workflow run that withdraws, or one of its Local Activities. fn of_run(producer: &Producer, withdrawing: &Producer) -> bool { - match producer.epoch.as_ref().and_then(kind) { + match kind_of(producer) { Some(ProducerKind::Workflow) => producer.id == withdrawing.id, - Some(ProducerKind::LocalActivity) => producer.run_id == withdrawing.run_id, + Some(ProducerKind::LocalActivity) => { + producer_run_id(producer) == producer_run_id(withdrawing) + } _ => false, } } @@ -199,17 +237,14 @@ fn of_run(producer: &Producer, withdrawing: &Producer) -> bool { #[cfg(test)] mod tests { use super::*; - use crate::proto::ClientEpoch; + use crate::proto::{ClientEpoch, ClientProducer, producer::Kind}; // Without the clamp, the bytes of a negative field would outrank every real epoch. #[test] fn a_negative_field_compares_as_zero() { - let negative = ProducerEpoch { - kind: Some(Kind::Client(ClientEpoch { attempt: -1 })), - }; - assert_eq!( - compare_epochs(&negative, &ProducerEpoch::default()), - Ordering::Equal - ); + let negative = Kind::Client(ClientProducer { + epoch: Some(ClientEpoch { attempt: -1 }), + }); + assert_eq!(compare_epochs(Some(&negative), None), Ordering::Equal); } } diff --git a/crates/streams/src/ids.rs b/crates/streams/src/ids.rs index d50cf47f3..45eb1e77b 100644 --- a/crates/streams/src/ids.rs +++ b/crates/streams/src/ids.rs @@ -11,7 +11,11 @@ //! script. Each part is percent-encoded, so a `:` or a brace in an id can't make two Streams share //! a key. -use crate::{StreamError, StreamResult, epoch::ProducerKind, proto::Producer}; +use crate::{ + StreamError, StreamResult, + epoch::{ProducerKind, producer_run_id}, + proto::Producer, +}; /// The longest Stream id, in bytes. pub const MAX_STREAM_ID_BYTES: usize = 4096; @@ -151,13 +155,14 @@ pub fn check_producer(producer: &Producer) -> StreamResult<()> { kind, ProducerKind::Workflow | ProducerKind::LocalActivity | ProducerKind::Activity ); - if runs && decode_part(parts[0]).as_deref() != Some(producer.run_id.as_str()) { + if runs && decode_part(parts[0]).as_deref() != Some(producer_run_id(producer)) { return Err(StreamError::invalid(format!( "The Producer id {:?} names another run than its run id {:?}", - producer.id, producer.run_id + producer.id, + producer_run_id(producer) ))); } - if kind == ProducerKind::Client && !producer.run_id.is_empty() { + if kind == ProducerKind::Client && !producer_run_id(producer).is_empty() { return Err(StreamError::invalid("A client Producer has no run id")); } Ok(()) @@ -207,20 +212,34 @@ impl StreamKeys { #[cfg(test)] mod tests { use super::*; - use crate::proto::{ProducerEpoch, producer_epoch::Kind}; + use crate::proto::{ + ActivityProducer, ClientProducer, LocalActivityProducer, WorkflowProducer, producer::Kind, + }; fn producer(kind: ProducerKind, id: &str, run_id: &str) -> Producer { + let run_id = run_id.to_string(); let kind = match kind { - ProducerKind::Workflow => Kind::Workflow(Default::default()), - ProducerKind::LocalActivity => Kind::LocalActivity(Default::default()), - ProducerKind::Activity => Kind::Activity(Default::default()), - ProducerKind::StandaloneActivity => Kind::StandaloneActivity(Default::default()), - ProducerKind::Client => Kind::Client(Default::default()), + ProducerKind::Workflow => Kind::Workflow(WorkflowProducer { + run_id, + ..Default::default() + }), + ProducerKind::LocalActivity => Kind::LocalActivity(LocalActivityProducer { + run_id, + ..Default::default() + }), + ProducerKind::Activity => Kind::Activity(ActivityProducer { + run_id, + ..Default::default() + }), + ProducerKind::StandaloneActivity => Kind::StandaloneActivity(ActivityProducer { + run_id, + ..Default::default() + }), + ProducerKind::Client => Kind::Client(ClientProducer::default()), }; Producer { id: id.to_string(), - epoch: Some(ProducerEpoch { kind: Some(kind) }), - run_id: run_id.to_string(), + kind: Some(kind), } } @@ -267,14 +286,9 @@ mod tests { producer(ProducerKind::Workflow, &workflow_producer_id(run), "other"), producer(ProducerKind::Activity, "act:run%3A1:a", run), Producer { - epoch: None, - ..producer(ProducerKind::Workflow, "wf:x", "x") - }, - Producer { - epoch: Some(ProducerEpoch::default()), + kind: None, ..producer(ProducerKind::Workflow, "wf:x", "x") }, - producer(ProducerKind::Client, "client:a:b", "run"), producer(ProducerKind::Client, "client:a|b", ""), producer(ProducerKind::Client, "client:run%3a1", ""), ]; diff --git a/crates/streams/tests/vectors.rs b/crates/streams/tests/vectors.rs index 853a4e96c..d824561a4 100644 --- a/crates/streams/tests/vectors.rs +++ b/crates/streams/tests/vectors.rs @@ -12,9 +12,10 @@ use temporalio_streams::{ ids::{self, StreamKeys}, offset::{entry_id_of_offset, offset_of_entry_id}, proto::{ - ActivityEpoch, AppendRecord, ClientEpoch, LocalActivityEpoch, Producer, ProducerEpoch, - StreamData, StreamRecord, StreamSuperseded, StreamWriter, WorkflowTaskEpoch, - producer_epoch::Kind, stream_record::Item, + ActivityEpoch, ActivityProducer, AppendRecord, ClientEpoch, ClientProducer, + LocalActivityEpoch, LocalActivityProducer, Producer, RunWithdrawal, StreamData, + StreamRecord, StreamSuperseded, StreamWriter, WorkflowProducer, WorkflowTaskEpoch, + producer::Kind, stream_record::Item, }, }; @@ -39,31 +40,43 @@ fn number(value: &Value, field: &str) -> i64 { value[field].as_i64().unwrap_or(0) } -/// The epoch of a kind, from the flat fields of the vectors. The case takes the fields of its kind. -fn epoch(kind: &str, value: &Value) -> ProducerEpoch { +/// The Producer kind of a vector, from its flat fields and its run. The case takes the fields of +/// its kind, and a run-based kind carries the run. `empty` has no case. +fn kind_of(kind: &str, value: &Value, run_id: &str) -> Option { let task = WorkflowTaskEpoch { started_event_id: number(value, "task_started_event_id"), attempt: number(value, "task_attempt") as i32, started_time: time(&value["task_started_time"]), }; let attempt = number(value, "attempt") as i32; - let activity = ActivityEpoch { - attempt_scheduled_time: time(&value["attempt_scheduled_time"]), - attempt, - }; - let kind = match kind { - "workflow" => Some(Kind::Workflow(task)), - "local_activity" => Some(Kind::LocalActivity(LocalActivityEpoch { - task: Some(task), + let run_id = run_id.to_string(); + let activity = ActivityProducer { + run_id: run_id.clone(), + epoch: Some(ActivityEpoch { + attempt_scheduled_time: time(&value["attempt_scheduled_time"]), attempt, + }), + }; + match kind { + "workflow" => Some(Kind::Workflow(WorkflowProducer { + run_id, + epoch: Some(task), + })), + "local_activity" => Some(Kind::LocalActivity(LocalActivityProducer { + run_id, + epoch: Some(LocalActivityEpoch { + task: Some(task), + attempt, + }), })), "activity" => Some(Kind::Activity(activity)), "standalone_activity" => Some(Kind::StandaloneActivity(activity)), - "client" => Some(Kind::Client(ClientEpoch { attempt })), + "client" => Some(Kind::Client(ClientProducer { + epoch: Some(ClientEpoch { attempt }), + })), "empty" => None, other => panic!("unknown kind {other}"), - }; - ProducerEpoch { kind } + } } #[test] @@ -137,10 +150,10 @@ fn offsets_match() { #[test] fn epoch_keys_match() { for case in load("epochs.json").as_array().unwrap() { - let epoch = epoch(text(case, "kind"), &case["epoch"]); - assert_eq!(epoch_key(&epoch).to_vec(), hex(text(case, "key"))); + let kind = kind_of(text(case, "kind"), &case["epoch"], ""); + assert_eq!(epoch_key(kind.as_ref()).to_vec(), hex(text(case, "key"))); assert_eq!( - task_key(&task_of(&epoch)).to_vec(), + task_key(&task_of(kind.as_ref())).to_vec(), hex(text(case, "task_key")) ); } @@ -152,29 +165,34 @@ fn the_appendix_e_withdrawals_match() { assert!(cases.as_array().unwrap().len() >= 35); for case in cases.as_array().unwrap() { let s = &case["superseded"]; + // A create's item has no epoch, and a Reader must not read it as an empty one. let withdrawing = Producer { id: text(s, "producer_id").to_string(), - // A create's item has no epoch, and a Reader must not read it as an empty one. - epoch: s - .get("epoch") - .map(|value| epoch(text(s, "producer_kind"), value)), - run_id: s["run_id"].as_str().unwrap_or_default().to_string(), + kind: s.get("epoch").and_then(|value| { + kind_of( + text(s, "producer_kind"), + value, + s["run_id"].as_str().unwrap_or_default(), + ) + }), }; let superseded = StreamSuperseded { writer: Some(StreamWriter { producer: Some(withdrawing), - previous_completed_task: s.get("previous_completed_task").map(|c| { - WorkflowTaskEpoch { - started_event_id: number(c, "started_event_id"), - attempt: number(c, "attempt") as i32, - started_time: time(&c["started_time"]), - } + withdrawal: Some(RunWithdrawal { + previous_completed_task: s.get("previous_completed_task").map(|c| { + WorkflowTaskEpoch { + started_event_id: number(c, "started_event_id"), + attempt: number(c, "attempt") as i32, + started_time: time(&c["started_time"]), + } + }), + failed_tasks: s["failed_tasks"] + .as_array() + .map(|list| list.iter().map(|v| v.as_i64().unwrap()).collect()) + .unwrap_or_default(), + withdraw_run: s["withdraw_run"].as_str().unwrap_or_default().to_string(), }), - failed_tasks: s["failed_tasks"] - .as_array() - .map(|list| list.iter().map(|v| v.as_i64().unwrap()).collect()) - .unwrap_or_default(), - withdraw_run: s["withdraw_run"].as_str().unwrap_or_default().to_string(), }), }; let p = &case["record"]["producer"]; @@ -182,8 +200,13 @@ fn the_appendix_e_withdrawals_match() { item: Some(Item::Data(StreamData { producer: Some(Producer { id: text(p, "id").to_string(), - epoch: p.get("epoch").map(|value| epoch(text(p, "kind"), value)), - run_id: p["run_id"].as_str().unwrap_or_default().to_string(), + kind: p.get("epoch").and_then(|value| { + kind_of( + text(p, "kind"), + value, + p["run_id"].as_str().unwrap_or_default(), + ) + }), }), ..Default::default() })),