From 691957a4f197cbdb6f270e4a10eb518e1f620a4f Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Tue, 13 Oct 2026 12:27:54 -0700 Subject: [PATCH 1/4] Ended reads with their owner and refused lost records. --- crates/streams/src/handle.rs | 348 +++++++++++++++++++++++++++- crates/streams/src/memory.rs | 26 +++ crates/streams/src/reader.rs | 52 ++++- crates/streams/src/redis/mod.rs | 193 ++++++++++++--- crates/streams/src/store.rs | 19 ++ crates/streams/tests/conformance.rs | 109 ++++++++- crates/streams/tests/redis_store.rs | 210 +++++++++++++++++ 7 files changed, 906 insertions(+), 51 deletions(-) diff --git a/crates/streams/src/handle.rs b/crates/streams/src/handle.rs index 38e9e13770..a9fa335ece 100644 --- a/crates/streams/src/handle.rs +++ b/crates/streams/src/handle.rs @@ -21,6 +21,7 @@ use std::{ sync::{Arc, Mutex}, time::Duration, }; +use temporalio_common::protos::temporal::api::enums::v1::WorkflowExecutionStatus; use tokio::time::Instant; /// The longest a producer writes without asking whether its owner's chain ended. @@ -34,6 +35,10 @@ pub struct StreamsOptions { pub owner_recheck: Duration, /// How many producer attempts and chains to remember the last owner check of. pub remembered: usize, + /// How long an idle read waits before it first asks whether its owner ended. + pub owner_check_min: Duration, + /// The longest an idle read waits between two such questions. + pub owner_check_max: Duration, } impl Default for StreamsOptions { @@ -41,6 +46,8 @@ impl Default for StreamsOptions { Self { owner_recheck: OWNER_RECHECK, remembered: 10_000, + owner_check_min: Duration::from_millis(500), + owner_check_max: Duration::from_secs(5), } } } @@ -230,7 +237,15 @@ impl Streams { }) } - /// Reads one page of a stream. The first call resolves the chain, and the state carries it. + /// Reads one page of a stream, waiting up to the request's `wait` for a record. + /// + /// The first call resolves the chain, and the state carries it. While nothing arrives, the + /// read asks whether its owner ended, first after [StreamsOptions::owner_check_min] and + /// then twice as long each time, up to [StreamsOptions::owner_check_max]. A record starts + /// the interval over. So many readers parked on a quiet stream don't load the server. Once + /// the owner ended, or the store marked the stream closed, the read delivers what is left + /// and answers `done`. A read pinned to a run ends when that run closes, even if the chain + /// continued as new. pub async fn read(&self, request: ReadRequest) -> StreamResult { let stream = address(request.stream.as_ref())?; let mut state = ReadState::decode_from(&request.state)?; @@ -242,9 +257,90 @@ impl Streams { topic: stream.topic.clone(), stream_hash: self.hash(stream), }; - let mut response = read_with(&*self.store, &target, &request, &mut state).await?; - response.state = state.encode_to_vec(); - Ok(response) + let wait = request + .wait + .and_then(|wait| Duration::try_from(wait).ok()) + .unwrap_or_default(); + let deadline = Instant::now() + wait; + let minimum = self.options.owner_check_min; + let mut page_request = request.clone(); + let response = loop { + let now = unix_ms(); + if state.check_interval_ms == 0 { + state.check_interval_ms = millis(minimum); + state.next_check_ms = now + state.check_interval_ms; + } + let until_check = Duration::from_millis(state.next_check_ms.saturating_sub(now)); + let remaining = deadline.saturating_duration_since(Instant::now()); + let wait = if state.owner_ended { + Duration::ZERO + } else { + remaining.min(until_check) + }; + page_request.wait = wait.try_into().ok(); + let (mut page, closed) = + read_with(&*self.store, &target, &page_request, &mut state).await?; + page_request.after = page.cursor.clone(); + if !page.records.is_empty() { + state.check_interval_ms = millis(minimum); + state.next_check_ms = unix_ms() + state.check_interval_ms; + break page; + } + if closed && !state.owner_ended { + // One more pass after learning of the close, as after learning the owner ended. + state.owner_ended = true; + continue; + } + if state.owner_ended { + page.done = true; + break page; + } + if unix_ms() >= state.next_check_ms { + state.owner_ended = self.owner_ended(stream, &target.chain).await?; + state.check_interval_ms = (state.check_interval_ms * 2) + .clamp(millis(minimum), millis(self.options.owner_check_max)); + state.next_check_ms = unix_ms() + state.check_interval_ms; + // One more pass after learning the owner ended, so a record that landed + // between the read and the describe is delivered. + if state.owner_ended { + continue; + } + } + if Instant::now() >= deadline { + break page; + } + }; + Ok(ReadResponse { + state: state.encode_to_vec(), + ..response + }) + } + + /// Whether the owner a read follows ended. Following the chain, a run that continued as + /// new is not the end, and a new chain on the Workflow id is. + async fn owner_ended(&self, stream: &StreamAddress, chain: &ChainId) -> StreamResult { + let key = owner_key(stream); + match self.describe(&key, &key.run_id).await { + Err(OwnerError::NotFound(_)) => { + // The chain came from this Workflow, so History has since dropped it, and + // nothing more will be written for it. + if key.run_id.is_empty() { + self.store.close_chain(chain).await?; + } + Ok(true) + } + Err(error) => Err(describe_failed(&error)), + Ok(described) if !key.run_id.is_empty() => { + Ok(described.status != WorkflowExecutionStatus::Running) + } + Ok(latest) if !latest.chain_ended() && latest.first_run_id == chain.first_run_id => { + Ok(false) + } + Ok(_) => { + self.store.close_chain(chain).await?; + Ok(true) + } + } } /// The cursor of a stream's newest record, empty when it holds none. @@ -406,6 +502,18 @@ fn check_topic(topic: &str) -> StreamResult<()> { temporalio_common::streams::check_topic(topic).map_err(StreamError::invalid_argument) } +fn unix_ms() -> u64 { + millis( + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default(), + ) +} + +fn millis(duration: Duration) -> u64 { + u64::try_from(duration.as_millis()).unwrap_or(u64::MAX) +} + fn owner_key(stream: &StreamAddress) -> OwnerKey { OwnerKey { namespace: stream.namespace.clone(), @@ -989,6 +1097,238 @@ mod tests { setup.streams.append(other).await.unwrap(); } + fn quick() -> StreamsOptions { + StreamsOptions { + owner_check_min: Duration::from_millis(20), + owner_check_max: Duration::from_millis(80), + ..StreamsOptions::default() + } + } + + fn read_after(after: &str, state: &[u8], wait: Duration) -> ReadRequest { + ReadRequest { + stream: Some(stream("out")), + after: after.to_string(), + state: state.to_vec(), + wait: Some(wait.try_into().unwrap()), + ..Default::default() + } + } + + /// Reads until the read says it is done, or fails after `limit` calls. + async fn read_to_end(streams: &Streams, wait: Duration) -> (usize, ReadResponse) { + let mut delivered = 0; + let mut last = read_after("", &[], wait); + for _ in 0..50 { + let page = streams.read(last.clone()).await.unwrap(); + delivered += page.records.len(); + if page.done { + return (delivered, page); + } + last = read_after(&page.cursor, &page.state, wait); + } + panic!("the read never ended"); + } + + #[tokio::test] + async fn a_read_ends_when_its_owner_chain_ends_and_marks_it_closed() { + let setup = setup(quick()); + setup.streams.append(append("p", 1, "1")).await.unwrap(); + setup.owners.end("wf", WorkflowExecutionStatus::Completed); + let (delivered, end) = read_to_end(&setup.streams, Duration::from_secs(5)).await; + assert_eq!(delivered, 1); + assert!(end.records.is_empty()); + // The read saw the chain end, and says so to later producers. + assert!(chain_closed(&setup.store, "run-1").await); + } + + #[tokio::test] + async fn a_read_follows_a_chain_that_continued_as_new() { + let setup = setup(quick()); + setup.streams.append(append("p", 1, "1")).await.unwrap(); + setup + .owners + .end("wf", WorkflowExecutionStatus::ContinuedAsNew); + setup.owners.run("wf", "run-2", "run-1"); + let page = setup + .streams + .read(read_after("", &[], Duration::from_millis(200))) + .await + .unwrap(); + let idle = setup + .streams + .read(read_after( + &page.cursor, + &page.state, + Duration::from_millis(200), + )) + .await + .unwrap(); + assert!(!idle.done); + assert!(!chain_closed(&setup.store, "run-1").await); + } + + #[tokio::test] + async fn a_read_pinned_to_a_run_ends_when_that_run_continues() { + let setup = setup(quick()); + setup.streams.append(append("p", 1, "1")).await.unwrap(); + setup + .owners + .runs + .lock() + .unwrap() + .get_mut("run-1") + .unwrap() + .status = WorkflowExecutionStatus::ContinuedAsNew; + setup.owners.run("wf", "run-2", "run-1"); + let mut request = read_after("", &[], Duration::from_secs(5)); + request.stream.as_mut().unwrap().run_id = "run-1".to_string(); + let mut done = false; + for _ in 0..10 { + let page = setup.streams.read(request.clone()).await.unwrap(); + if page.done { + done = true; + break; + } + request.after = page.cursor; + request.state = page.state; + } + assert!(done); + // It stopped with the first run, though the chain still runs. + assert!(!chain_closed(&setup.store, "run-1").await); + } + + #[tokio::test] + async fn a_read_ends_when_a_new_chain_reuses_the_workflow_id() { + let setup = setup(quick()); + setup.streams.append(append("p", 1, "1")).await.unwrap(); + let first = setup + .streams + .read(read_after("", &[], Duration::ZERO)) + .await + .unwrap(); + // The id's latest run is running, but it belongs to another chain. + setup.owners.run("wf", "run-9", "run-9"); + let mut request = read_after(&first.cursor, &first.state, Duration::from_secs(5)); + let mut done = false; + for _ in 0..10 { + let page = setup.streams.read(request.clone()).await.unwrap(); + assert!(page.records.is_empty()); + if page.done { + done = true; + break; + } + request = read_after(&page.cursor, &page.state, Duration::from_secs(5)); + } + assert!(done); + assert!(chain_closed(&setup.store, "run-1").await); + } + + #[tokio::test] + async fn rule_20_3_a_read_ends_when_history_no_longer_holds_its_owner() { + let setup = setup(quick()); + setup.streams.append(append("p", 1, "1")).await.unwrap(); + let first = setup + .streams + .read(read_after("", &[], Duration::ZERO)) + .await + .unwrap(); + setup + .owners + .fail("wf", OwnerError::NotFound("workflow not found".to_string())); + let page = setup + .streams + .read(read_after( + &first.cursor, + &first.state, + Duration::from_secs(5), + )) + .await + .unwrap(); + assert!(page.done); + assert!(chain_closed(&setup.store, "run-1").await); + } + + #[tokio::test] + async fn rule_17_2_an_owner_check_during_a_read_that_fails_is_a_storage_failure() { + let setup = setup(quick()); + let first = setup + .streams + .read(read_after("", &[], Duration::ZERO)) + .await + .unwrap(); + setup + .owners + .fail("wf", OwnerError::Failed("unavailable".to_string())); + let error = setup + .streams + .read(read_after( + &first.cursor, + &first.state, + Duration::from_secs(5), + )) + .await + .unwrap_err(); + assert_eq!(error.kind, StreamFailureKind::Storage); + assert!(error.message.contains("unavailable"), "{error}"); + } + + #[tokio::test] + async fn an_idle_read_backs_off_its_owner_checks() { + let setup = setup(quick()); + let first = setup + .streams + .read(read_after("", &[], Duration::ZERO)) + .await + .unwrap(); + let before = setup.owners.describes(); + let idle = setup + .streams + .read(read_after( + &first.cursor, + &first.state, + Duration::from_secs(1), + )) + .await + .unwrap(); + // Fixed checks every 20 ms would be about 50, and doubling to 80 ms is about 15. + let checks = setup.owners.describes() - before; + assert!((8..=20).contains(&checks), "{checks} checks"); + let state = ReadState::decode_from(&idle.state).unwrap(); + assert_eq!(state.check_interval_ms, 80); + // A record starts the interval over. + setup.streams.append(append("p", 1, "1")).await.unwrap(); + let page = setup + .streams + .read(read_after( + &idle.cursor, + &idle.state, + Duration::from_secs(1), + )) + .await + .unwrap(); + assert_eq!(page.records.len(), 1); + let state = ReadState::decode_from(&page.state).unwrap(); + assert_eq!(state.check_interval_ms, 20); + } + + #[tokio::test] + async fn a_closed_stream_ends_its_reads_without_asking_temporal() { + let setup = setup(quick()); + setup.streams.append(append("p", 1, "1")).await.unwrap(); + setup + .streams + .close(CloseRequest { + stream: Some(stream("out")), + }) + .await + .unwrap(); + let before = setup.owners.describes(); + let (delivered, _) = read_to_end(&setup.streams, Duration::from_secs(5)).await; + assert_eq!(delivered, 1); + assert_eq!(setup.owners.describes(), before); + } + /// The digest the Python SDK's Redis provider took over `[{"n": 1}, {"n": 2}]` from producer /// `p`, attempt 1, sequence 1. const PYTHON_DIGEST: &str = "0499768af1856bc6fd19cf6cf90e28e971099f1680e8347d70bdb4b65273f30f"; diff --git a/crates/streams/src/memory.rs b/crates/streams/src/memory.rs index cf39662f2b..a190c3bfe7 100644 --- a/crates/streams/src/memory.rs +++ b/crates/streams/src/memory.rs @@ -310,6 +310,32 @@ impl StreamStore for MemoryStore { .cloned()) } + async fn newest_ever(&self, chain: &ChainId, topic: &str) -> StreamResult { + let chains = self.chains.lock().unwrap(); + Ok(chains + .get(&chain_key(chain)) + .and_then(|chain| chain.topics.get(topic)) + .and_then(|topic| topic.head().checked_sub(1)) + .map(|offset| offset.to_string()) + .unwrap_or_default()) + } + + async fn dropped_after( + &self, + chain: &ChainId, + topic: &str, + mark: &str, + ) -> StreamResult> { + let mark = parse_position(mark)?; + let chains = self.chains.lock().unwrap(); + Ok(chains + .get(&chain_key(chain)) + .and_then(|chain| chain.topics.get(topic)) + .and_then(|topic| topic.base.checked_sub(1)) + .filter(|dropped| mark.is_none_or(|mark| *dropped > mark)) + .map(|offset| offset.to_string())) + } + async fn close_chain(&self, chain: &ChainId) -> StreamResult<()> { self.chains .lock() diff --git a/crates/streams/src/reader.rs b/crates/streams/src/reader.rs index d517eccb66..b9823b1674 100644 --- a/crates/streams/src/reader.rs +++ b/crates/streams/src/reader.rs @@ -44,6 +44,21 @@ pub(crate) struct ReadState { /// The first run of the chain the read follows, so later calls don't resolve it again. #[prost(string, tag = "3")] pub(crate) first_run_id: String, + /// Set once the read learned its owner ended, so it only drains what is left. + #[prost(bool, tag = "4")] + pub(crate) owner_ended: bool, + /// How long the read waits between owner checks while nothing arrives. + #[prost(uint64, tag = "5")] + pub(crate) check_interval_ms: u64, + /// When the next owner check is due, in milliseconds since the Unix epoch. + #[prost(uint64, tag = "6")] + pub(crate) next_check_ms: u64, + /// The topic's newest position ever when a read that has delivered nothing started. Such a + /// read has no position, so a record retention dropped after this mark was lost to it. + #[prost(string, tag = "7")] + watermark: String, + #[prost(bool, tag = "8")] + watermark_known: bool, /// When this read last saw each producer in `attempts`, on the read's own clock, so the /// least recently seen one is forgotten first. #[prost(map = "string, uint64", tag = "9")] @@ -141,19 +156,19 @@ pub async fn read_page( request: &ReadRequest, ) -> StreamResult { let mut state = ReadState::decode_from(&request.state)?; - let mut response = read_with(store, target, request, &mut state).await?; + let (mut response, _) = read_with(store, target, request, &mut state).await?; response.state = state.encode_to_vec(); Ok(response) } /// [read_page] with the state already decoded, for a caller that keeps fields of its own in it. -/// The answer's `state` is left empty. +/// The answer's `state` is left empty. Also answers whether the store marked the stream closed. pub(crate) async fn read_with( store: &dyn StreamStore, target: &ReadTarget, request: &ReadRequest, state: &mut ReadState, -) -> StreamResult { +) -> StreamResult<(ReadResponse, bool)> { let from_end = request.after == END; let position = if from_end { store @@ -183,6 +198,21 @@ pub(crate) async fn read_with( } } state.started = true; + if position.is_empty() { + if !state.watermark_known { + state.watermark = store.newest_ever(&target.chain, &target.topic).await?; + state.watermark_known = true; + } else if let Some(dropped) = store + .dropped_after(&target.chain, &target.topic, &state.watermark) + .await? + { + return Err(StreamError::expired(format!( + "records on topic {:?} were dropped by retention while this read was behind \ + them, before it delivered any; the newest dropped one is {dropped}", + target.topic + ))); + } + } let page = store .read(StoreReadRequest { @@ -196,6 +226,7 @@ pub(crate) async fn read_with( }, }) .await?; + let closed = page.closed; let mut records = Vec::with_capacity(page.records.len()); let mut previous = start; for stored in page.records { @@ -234,12 +265,15 @@ pub(crate) async fn read_with( }); previous = cursor; } - Ok(ReadResponse { - records, - cursor: previous, - state: Vec::new(), - done: false, - }) + Ok(( + ReadResponse { + records, + cursor: previous, + state: Vec::new(), + done: false, + }, + closed, + )) } /// A stored record as a reader delivers it. Fails on bytes that are no record, and on a kind no diff --git a/crates/streams/src/redis/mod.rs b/crates/streams/src/redis/mod.rs index 538b17570c..e294e7a8db 100644 --- a/crates/streams/src/redis/mod.rs +++ b/crates/streams/src/redis/mod.rs @@ -136,11 +136,18 @@ impl RedisStore { Ok(ChainKeys::new(&self.prefix, chain)) } - /// Reads the chain's close flag and the topic's, which share a slot on a cluster. - fn closed(pipeline: &mut redis::Pipeline, keys: &ChainKeys, topic: &str) { - pipeline - .hget(keys.chain(), CLOSED_FIELD) - .hget(keys.meta(topic), CLOSED_FIELD); + /// What a read needs to know besides the records, in one script so a cluster answers it + /// from the slot the chain's keys share. + fn checks(keys: &ChainKeys, topic: &str) -> Cmd { + let mut eval = redis::cmd("EVAL"); + eval.arg(TOPIC_STATE) + .arg(4) + .arg(keys.log(topic)) + .arg(keys.meta(topic)) + .arg(keys.chain()) + .arg(keys.meta(topic)) + .arg(CLOSED_FIELD); + eval } fn xread( @@ -193,6 +200,100 @@ fn records(reply: Value) -> StreamResult> { .collect()) } +/// Reads whether the log is there, what its meta says and both close flags. Lua turns a missing +/// value into `false`, which keeps the reply's places where a `nil` would cut the list short. +const TOPIC_STATE: &str = "\ +return {redis.call('EXISTS', KEYS[1]), redis.call('EXISTS', KEYS[2]), \ +redis.call('HGET', KEYS[2], 'trimmed'), redis.call('HGET', KEYS[2], 'last'), \ +redis.call('HGET', KEYS[3], ARGV[1]), redis.call('HGET', KEYS[4], ARGV[1])}"; + +/// What the store knows about a topic besides its records. +#[derive(Debug, Default)] +struct TopicState { + log_exists: bool, + meta_exists: bool, + /// The newest id retention trimmed from the log. + trimmed: Option, + /// The newest id the log ever took. It outlives the log in the meta. + last: Option, + chain_closed: bool, + topic_closed: bool, +} + +impl redis::FromRedisValue for TopicState { + fn from_redis_value(value: Value) -> Result { + let text = |value: &Value| match value { + Value::BulkString(bytes) => Some(String::from_utf8_lossy(bytes).into_owned()), + _ => None, + }; + let number = |value: &Value| matches!(value, Value::Int(n) if *n > 0); + let Value::Array(values) = value else { + return Err(format!("a topic state is a list, not {value:?}").into()); + }; + let [log, meta, trimmed, last, chain_closed, topic_closed] = values.as_slice() else { + return Err(format!("a topic state has six values, not {}", values.len()).into()); + }; + Ok(Self { + log_exists: number(log), + meta_exists: number(meta), + trimmed: text(trimmed), + last: text(last), + chain_closed: text(chain_closed).is_some(), + topic_closed: text(topic_closed).is_some(), + }) + } +} + +impl TopicState { + /// Why a read after `position` can't go on, if the store lost records after it. + /// + /// With the log there, records after the position are lost when retention trimmed past it + /// and the read did not get them first. With the log gone but its meta left, they are lost + /// when the log took records after the position. With neither, nothing is known. A read + /// from the beginning loses nothing, since anything trimmed was behind its start. + fn lost( + &self, + topic: &str, + position: Option<(&str, (u64, u64))>, + found: &[StoredRecord], + ) -> StreamResult> { + let Some((position, at)) = position else { + return Ok(None); + }; + if !self.log_exists && found.is_empty() { + if !self.meta_exists { + return Ok(Some(StreamError::not_found(format!( + "topic {topic:?} keeps no log and no tombstone, so nothing is known about \ + what followed {position}" + )))); + } + if let Some(last) = &self.last + && entry(last)? > at + { + return Ok(Some(StreamError::expired(format!( + "the log of topic {topic:?} expired with records after {position}" + )))); + } + return Ok(None); + } + let Some(trimmed) = &self.trimmed else { + return Ok(None); + }; + let watermark = entry(trimmed)?; + let first = found + .first() + .map(|record| entry(&record.position)) + .transpose()?; + if watermark > at && first.is_none_or(|first| watermark < first) { + return Ok(Some(StreamError::expired(format!( + "records after {position} on topic {topic:?} were dropped by retention while \ + this read was behind them; the newest dropped one is {trimmed}" + )))); + } + Ok(None) + } +} + fn wait_of(request: &StoreReadRequest) -> Duration { request.wait.as_ref().map_or(Duration::ZERO, |wait| { Duration::new( @@ -245,58 +346,45 @@ impl StreamStore for RedisStore { async fn read(&self, request: StoreReadRequest) -> StreamResult { let keys = self.keys(request.chain.as_ref())?; - let after = match request.after_position.as_str() { - "" => "0-0".to_string(), - position => { - entry(position)?; - position.to_string() - } + let topic = request.topic.as_str(); + let position = match request.after_position.as_str() { + "" => None, + position => Some((position, entry(position)?)), }; + let after = position.map_or("0-0", |(position, _)| position); let wait = wait_of(&request); let deadline = Instant::now() + wait; let mut pipeline = redis::pipe(); - pipeline.add_command(Self::xread( - &keys, - &request.topic, - &after, - request.max_records, - None, - )); - Self::closed(&mut pipeline, &keys, &request.topic); - let (reply, chain_closed, topic_closed): (Value, Option, Option) = pipeline + pipeline.add_command(Self::xread(&keys, topic, after, request.max_records, None)); + let checks = Self::checks(&keys, topic); + pipeline.add_command(checks.clone()); + let (reply, mut state): (Value, TopicState) = pipeline .query_async(&mut self.shared.clone()) .await .mapped(Call::Read)?; let mut found = records(reply)?; - let mut closed = chain_closed.is_some() || topic_closed.is_some(); - if found.is_empty() && !wait.is_zero() { - let xread = Self::xread( - &keys, - &request.topic, - &after, - request.max_records, - Some(wait), - ); + if found.is_empty() && !wait.is_zero() && state.lost(topic, position, &found)?.is_none() { + let xread = Self::xread(&keys, topic, after, request.max_records, Some(wait)); let blocked = self .blocking - .query(&keys.log(&request.topic), &xread, deadline) + .query(&keys.log(topic), &xread, deadline) .await .mapped(Call::Read)?; if let Some(reply) = blocked { found = records(reply)?; - // Asked after the wait, so a close that came during it is seen. - let mut pipeline = redis::pipe(); - Self::closed(&mut pipeline, &keys, &request.topic); - let (chain_closed, topic_closed): (Option, Option) = pipeline + // Read again after the wait, so a trim or a close that raced it is seen. + state = checks .query_async(&mut self.shared.clone()) .await .mapped(Call::Read)?; - closed = chain_closed.is_some() || topic_closed.is_some(); } } + if let Some(error) = state.lost(topic, position, &found)? { + return Err(error); + } Ok(StoreReadResponse { records: found, - closed, + closed: state.chain_closed || state.topic_closed, }) } @@ -344,6 +432,39 @@ impl StreamStore for RedisStore { })) } + async fn newest_ever(&self, chain: &ChainId, topic: &str) -> StreamResult { + let last: Option = redis::cmd("HGET") + .arg(ChainKeys::new(&self.prefix, chain).meta(topic)) + .arg("last") + .query_async(&mut self.shared.clone()) + .await + .mapped(Call::Read)?; + Ok(last.unwrap_or_default()) + } + + async fn dropped_after( + &self, + chain: &ChainId, + topic: &str, + mark: &str, + ) -> StreamResult> { + let trimmed: Option = redis::cmd("HGET") + .arg(ChainKeys::new(&self.prefix, chain).meta(topic)) + .arg("trimmed") + .query_async(&mut self.shared.clone()) + .await + .mapped(Call::Read)?; + let Some(trimmed) = trimmed else { + return Ok(None); + }; + // A log that expired comes back with `trimmed` set to its old `last`, which is the mark + // itself, so a revive counts as no loss. + if !mark.is_empty() && entry(&trimmed)? <= entry(mark)? { + return Ok(None); + } + Ok(Some(trimmed)) + } + async fn close_chain(&self, chain: &ChainId) -> StreamResult<()> { let keys = ChainKeys::new(&self.prefix, chain); redis::pipe() diff --git a/crates/streams/src/store.rs b/crates/streams/src/store.rs index 405d3c9c97..36bfcd1dae 100644 --- a/crates/streams/src/store.rs +++ b/crates/streams/src/store.rs @@ -50,6 +50,25 @@ pub trait StreamStore: Send + Sync { position: &str, ) -> StreamResult>>; + /// The position of the newest record a topic ever took, even one retention dropped since, + /// or empty when it took none. A read that has delivered nothing keeps it as its mark, since + /// it has no position the store could check. + async fn newest_ever(&self, _chain: &ChainId, _topic: &str) -> StreamResult { + Ok(String::new()) + } + + /// The position of the newest record retention dropped from a topic, when that record came + /// after `mark`, the topic's [StreamStore::newest_ever] when a read started. Such a record + /// was lost to the read. A store that never drops records keeps the default. + async fn dropped_after( + &self, + _chain: &ChainId, + _topic: &str, + _mark: &str, + ) -> StreamResult> { + Ok(None) + } + /// Marks a chain closed, so its topics refuse new batches from producers. async fn close_chain(&self, chain: &ChainId) -> StreamResult<()>; diff --git a/crates/streams/tests/conformance.rs b/crates/streams/tests/conformance.rs index ec8c5f5aa9..65bc08585f 100644 --- a/crates/streams/tests/conformance.rs +++ b/crates/streams/tests/conformance.rs @@ -28,6 +28,10 @@ trait Case: Send + Sync { /// Drops all but the newest `keep` records of a topic, as retention does. async fn drop_oldest(&self, chain: &ChainId, topic: &str, keep: usize); + + /// Lets a quiet topic's records expire whole, as a log does after a retention with no + /// write. What the store knows about the topic stays. + async fn expire_log(&self, chain: &ChainId, topic: &str); } static CHAINS: AtomicUsize = AtomicUsize::new(0); @@ -59,6 +63,10 @@ impl Case for Memory { async fn drop_oldest(&self, chain: &ChainId, topic: &str, keep: usize) { self.0.truncate(chain, topic, keep); } + + async fn expire_log(&self, chain: &ChainId, topic: &str) { + self.0.truncate(chain, topic, 0); + } } async fn memory() -> Option { @@ -179,6 +187,14 @@ mod on_redis { raw.hset(&format!("{log}:meta"), "trimmed", &entries[doomed - 1].0) .await; } + + async fn expire_log(&self, chain: &ChainId, topic: &str) { + // The log key expires and its meta stays, as the tombstone. + let mut raw = self.raw.clone(); + let _: i64 = raw + .query(redis::cmd("DEL").arg(log_key(&self.prefix, chain, topic))) + .await; + } } } @@ -559,6 +575,87 @@ mod cases { ); } + pub(crate) async fn an_empty_read_from_the_beginning_learns_of_a_trim(case: &dyn Case) { + // The read has delivered nothing, so it has no position, but records written after it + // started and dropped before its next call are still lost to it. + let chain = case.chain(); + let first = read(case, &chain, OUT, BEGINNING).await.unwrap(); + assert!(first.records.is_empty()); + assert_eq!(first.cursor, BEGINNING); + Producer::new(case, &chain, OUT, "p", 1) + .append(&["1", "2"]) + .await; + case.drop_oldest(&chain, OUT, 1).await; + assert_eq!( + kind( + read_with( + case, + &chain, + OUT, + &first.cursor, + Duration::ZERO, + &first.state + ) + .await + ), + StreamFailureKind::Expired + ); + // A new read starts at the oldest record left, as a read from the beginning does. + assert_eq!( + values(&read(case, &chain, OUT, BEGINNING).await.unwrap()), + ["2"] + ); + } + + pub(crate) async fn rule_11_2_an_empty_read_goes_on_when_a_quiet_log_comes_back( + case: &dyn Case, + ) { + // A log that expired comes back with its next write, and nothing the read could have + // seen was lost. + let chain = case.chain(); + Producer::new(case, &chain, OUT, "p", 1) + .append(&["1"]) + .await; + case.expire_log(&chain, OUT).await; + let first = read(case, &chain, OUT, BEGINNING).await.unwrap(); + assert!(first.records.is_empty()); + let mut producer = Producer::new(case, &chain, OUT, "p", 1); + producer.sequence = 2; + producer.append(&["2"]).await; + let next = read_with( + case, + &chain, + OUT, + &first.cursor, + Duration::ZERO, + &first.state, + ) + .await + .unwrap(); + assert_eq!(values(&next), ["2"]); + } + + pub(crate) async fn an_empty_read_from_the_beginning_with_nothing_trimmed_goes_on( + case: &dyn Case, + ) { + let chain = case.chain(); + let first = read(case, &chain, OUT, BEGINNING).await.unwrap(); + Producer::new(case, &chain, OUT, "p", 1) + .append(&["1"]) + .await; + let next = read_with( + case, + &chain, + OUT, + &first.cursor, + Duration::ZERO, + &first.state, + ) + .await + .unwrap(); + assert_eq!(values(&next), ["1"]); + } + pub(crate) async fn a_retried_append_returns_the_original_positions(case: &dyn Case) { let chain = case.chain(); let first = Producer::new(case, &chain, OUT, "p", 1) @@ -961,6 +1058,9 @@ mod memory { dropping_a_waiting_read_releases_it, beginning_starts_at_the_oldest_record_still_held, a_reader_that_falls_behind_retention_is_told, + an_empty_read_from_the_beginning_learns_of_a_trim, + an_empty_read_from_the_beginning_with_nothing_trimmed_goes_on, + rule_11_2_an_empty_read_goes_on_when_a_quiet_log_comes_back, a_retried_append_returns_the_original_positions, a_retry_with_other_ciphertext_still_deduplicates, a_divergent_retry_is_refused, @@ -982,8 +1082,7 @@ mod memory { ); } -/// Every case but the ones that need reads that check retention or owner deletes, which come -/// with those features. +/// Every case but owner deletes, which come with that feature. #[cfg(feature = "redis")] macro_rules! redis_cases { ($case:expr) => { @@ -998,6 +1097,12 @@ macro_rules! redis_cases { end_reads_only_what_arrives_after_the_read_starts, a_read_waits_no_longer_than_asked, dropping_a_waiting_read_releases_it, + beginning_starts_at_the_oldest_record_still_held, + a_reader_that_falls_behind_retention_is_told, + an_empty_read_from_the_beginning_learns_of_a_trim, + an_empty_read_from_the_beginning_with_nothing_trimmed_goes_on, + rule_11_2_an_empty_read_goes_on_when_a_quiet_log_comes_back, + a_refused_append_leaves_no_trace_of_its_topic, a_retried_append_returns_the_original_positions, a_retry_with_other_ciphertext_still_deduplicates, a_divergent_retry_is_refused, diff --git a/crates/streams/tests/redis_store.rs b/crates/streams/tests/redis_store.rs index 21f9c23ed9..ad61bd3652 100644 --- a/crates/streams/tests/redis_store.rs +++ b/crates/streams/tests/redis_store.rs @@ -841,3 +841,213 @@ async fn rule_10_1_an_append_that_races_a_topic_close_is_refused_once_the_close_ let length: usize = raw.xlen(log_key(&prefix, &chain, "events")).await.unwrap(); assert_eq!(length, 1); } + +fn read_after(chain: &ChainId, position: &str, max_records: u32) -> StoreReadRequest { + StoreReadRequest { + chain: Some(chain.clone()), + topic: "events".to_string(), + after_position: position.to_string(), + max_records, + ..Default::default() + } +} + +#[tokio::test] +async fn an_expired_log_leaves_a_tombstone_that_tells_expired_from_empty() { + let Some(mut setup) = retained_for(Duration::from_millis(300)).await else { + return; + }; + let chain = chain(); + let first = setup + .store + .append(append(&chain, "p", 1, 1, b"\x01")) + .await + .unwrap(); + let last = setup + .store + .append(append(&chain, "p", 2, 1, b"\x02")) + .await + .unwrap(); + tokio::time::sleep(Duration::from_millis(500)).await; + let log = log_key(&setup.prefix, &chain, "events"); + let exists: bool = setup.raw.exists(&log).await.unwrap(); + assert!(!exists); + let error = setup + .store + .read(read_after(&chain, &first.last_position, 10)) + .await + .unwrap_err(); + assert_eq!(error.kind, StreamFailureKind::Expired); + assert!(error.message.contains("expired"), "{error}"); + // Nothing came after the last record, so the stream is empty after it. + let empty = setup + .store + .read(read_after(&chain, &last.last_position, 10)) + .await + .unwrap(); + assert!(empty.records.is_empty()); + let _: () = setup.raw.del(format!("{log}:meta")).await.unwrap(); + let error = setup + .store + .read(read_after(&chain, &last.last_position, 10)) + .await + .unwrap_err(); + assert_eq!(error.kind, StreamFailureKind::NotFound); + assert!(error.message.contains("no tombstone"), "{error}"); + // A read from the beginning of a missing log just finds no records. + let empty = setup.store.read(read_after(&chain, "", 10)).await.unwrap(); + assert!(empty.records.is_empty()); +} + +#[tokio::test] +async fn retention_expires_a_cursor_only_past_the_newest_trimmed_record() { + let Some(setup) = retained_for(Duration::from_millis(300)).await else { + return; + }; + let chain = chain(); + let first = setup + .store + .append(append(&chain, "p", 1, 1, b"\x01")) + .await + .unwrap(); + let second = setup + .store + .append(append(&chain, "p", 2, 1, b"\x02")) + .await + .unwrap(); + tokio::time::sleep(Duration::from_millis(400)).await; + let third = setup + .store + .append(append(&chain, "p", 3, 1, b"\x03")) + .await + .unwrap(); + let error = setup + .store + .read(read_after(&chain, &first.last_position, 10)) + .await + .unwrap_err(); + assert_eq!(error.kind, StreamFailureKind::Expired); + assert!(error.message.contains("dropped by retention"), "{error}"); + // Nothing after the newest trimmed record was lost, so it still resumes. + let resumed = setup + .store + .read(read_after(&chain, &second.last_position, 10)) + .await + .unwrap(); + let positions: Vec<_> = resumed.records.iter().map(|r| &r.position).collect(); + assert_eq!(positions, [&third.last_position]); +} + +#[tokio::test] +async fn a_read_that_falls_behind_retention_fails_instead_of_skipping() { + let Some(setup) = retained_for(Duration::from_millis(300)).await else { + return; + }; + let chain = chain(); + setup + .store + .append(append(&chain, "p", 1, 2, b"\x01")) + .await + .unwrap(); + // One record per read, so the trim lands between two reads of one batch. + let first = setup.store.read(read_after(&chain, "", 1)).await.unwrap(); + assert_eq!(first.records.len(), 1); + tokio::time::sleep(Duration::from_millis(350)).await; + // This append trims both earlier records, and the second was never delivered. + let third = setup + .store + .append(append(&chain, "p", 3, 1, b"\x02")) + .await + .unwrap(); + let error = setup + .store + .read(read_after(&chain, &first.records[0].position, 1)) + .await + .unwrap_err(); + assert_eq!(error.kind, StreamFailureKind::Expired); + assert!( + error.message.contains("while this read was behind"), + "{error}" + ); + // A read that starts after the trim begins at the oldest record left. + let fresh = setup.store.read(read_after(&chain, "", 1)).await.unwrap(); + assert_eq!(fresh.records[0].position, third.last_position); +} + +#[tokio::test] +async fn rule_22_1_a_lost_connection_on_the_read_checks_is_a_storage_failure() { + let Some(url) = url() else { + return; + }; + let faults = Faults::start(&url).await; + let setup = setup_at(&faults.url(), &url).await; + let chain = chain(); + let landed = setup + .store + .append(append(&chain, "p", 1, 1, b"\x01")) + .await + .unwrap(); + faults.next(LOSE_REQUEST); + let error = setup + .store + .read(read_after(&chain, &landed.last_position, 10)) + .await + .unwrap_err(); + assert_eq!(error.kind, StreamFailureKind::Storage, "{error}"); +} + +#[tokio::test] +async fn rule_20_4_a_malformed_entry_fails_with_its_cursor() { + use temporalio_streams::{ReadTarget, mint_cursor, proto::ReadRequest, read_page, stream_hash}; + let Some(mut setup) = setup().await else { + return; + }; + let chain = chain(); + setup + .store + .append(append(&chain, "p", 1, 1, b"\x01")) + .await + .unwrap(); + let log = log_key(&setup.prefix, &chain, "events"); + let entry: String = setup.raw.xadd(&log, "*", &[("not-r", "x")]).await.unwrap(); + setup + .store + .append(append(&chain, "p", 2, 1, b"\x02")) + .await + .unwrap(); + let target = ReadTarget { + chain: chain.clone(), + topic: "events".to_string(), + stream_hash: stream_hash(&chain.namespace, "workflow", &chain.workflow_id, "events"), + }; + let first = read_page(&setup.store, &target, &ReadRequest::default()) + .await + .unwrap(); + assert_eq!(first.records.len(), 1); + let error = read_page( + &setup.store, + &target, + &ReadRequest { + after: first.cursor.clone(), + state: first.state.clone(), + ..Default::default() + }, + ) + .await + .unwrap_err(); + assert_eq!(error.kind, StreamFailureKind::Record); + let bad = mint_cursor("redis", &target.stream_hash, &entry); + assert_eq!(error.cursor.as_deref(), Some(bad.as_str())); + let after = read_page( + &setup.store, + &target, + &ReadRequest { + after: bad, + state: first.state, + ..Default::default() + }, + ) + .await + .unwrap(); + assert_eq!(after.records.len(), 1); +} From 3c6be557cc22aa1aa4aa97ddcb10714586b3d834 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Tue, 13 Oct 2026 12:29:13 -0700 Subject: [PATCH 2/4] Tested the Core store against the Python Redis provider. --- crates/streams/tests/cross_impl.rs | 418 +++++++++++++++++++++++ crates/streams/tests/cross_impl/brook.py | 144 ++++++++ 2 files changed, 562 insertions(+) create mode 100644 crates/streams/tests/cross_impl.rs create mode 100644 crates/streams/tests/cross_impl/brook.py diff --git a/crates/streams/tests/cross_impl.rs b/crates/streams/tests/cross_impl.rs new file mode 100644 index 0000000000..70585ff96f --- /dev/null +++ b/crates/streams/tests/cross_impl.rs @@ -0,0 +1,418 @@ +//! The Core store and the Python SDK's Redis provider on one Redis, each reading and writing +//! what the other wrote. +//! +//! Set `STREAMS_REDIS_URL`, and `STREAMS_BROOK_PY` to a Python SDK checkout whose Redis provider +//! is the one before the store moved into Core. The Python side runs `tests/cross_impl/brook.py` +//! through `uv`. + +#![cfg(feature = "redis")] + +use prost::Message; +use serde_json::{Value, json}; +use std::{ + path::PathBuf, + process::Command, + sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, + }, + time::Duration, +}; +use temporalio_common::protos::temporal::api::{ + common::v1::Payload, enums::v1::WorkflowExecutionStatus, +}; +use temporalio_streams::{ + OwnerClient, OwnerDescription, OwnerError, ReadTarget, RedisStore, RedisStoreOptions, + StreamStore, Streams, StreamsOptions, mint_cursor, + proto::{ + AppendRecord, AppendRequest, ChainId, NamedProducer, ReadRequest, StoreAppendRequest, + StreamAddress, StreamFailureKind, StreamOwnerKind, StreamRecordKind, + append_request::Producer, read_record, + }, + read_page, stored_append_record, stream_hash, +}; + +static IDS: AtomicUsize = AtomicUsize::new(0); + +const NAMESPACE: &str = "ns"; +const FIRST_RUN: &str = "run-1"; + +struct Both { + url: String, + checkout: PathBuf, + prefix: String, + workflow_id: String, + retention_ms: u64, + core: RedisStore, +} + +async fn pair(retention: Duration) -> Option { + let (Ok(url), Ok(checkout)) = ( + std::env::var("STREAMS_REDIS_URL"), + std::env::var("STREAMS_BROOK_PY"), + ) else { + eprintln!( + "set STREAMS_REDIS_URL and STREAMS_BROOK_PY to run the cross-implementation test" + ); + return None; + }; + let id = IDS.fetch_add(1, Ordering::Relaxed); + let prefix = format!("cross-{}-{id}", std::process::id()); + let mut options = RedisStoreOptions::new(url.clone()); + options.key_prefix = prefix.clone(); + options.retention = retention; + Some(Both { + core: RedisStore::connect(options).await.unwrap(), + url, + checkout: checkout.into(), + prefix, + workflow_id: format!("wf-{id}"), + retention_ms: retention.as_millis() as u64, + }) +} + +impl Both { + /// Runs one operation of the Python provider and answers its JSON. + fn brook(&self, op: Value) -> Value { + let helper = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("tests/cross_impl/brook.py"); + let mut op = op; + let fields = op.as_object_mut().unwrap(); + fields.insert("url".into(), json!(self.url)); + fields.insert("prefix".into(), json!(self.prefix)); + fields.insert("namespace".into(), json!(NAMESPACE)); + fields.insert("workflow_id".into(), json!(self.workflow_id)); + fields.insert("first_run".into(), json!(FIRST_RUN)); + fields.insert("retention_ms".into(), json!(self.retention_ms)); + let output = Command::new("uv") + .args(["run", "--no-sync", "python"]) + .arg(helper) + .arg(op.to_string()) + .current_dir(&self.checkout) + .output() + .unwrap(); + assert!( + output.status.success(), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + let stdout = String::from_utf8(output.stdout).unwrap(); + serde_json::from_str(stdout.lines().last().unwrap()).unwrap() + } + + fn chain(&self) -> ChainId { + ChainId { + namespace: NAMESPACE.to_string(), + workflow_id: self.workflow_id.clone(), + first_run_id: FIRST_RUN.to_string(), + } + } + + fn target(&self, topic: &str) -> ReadTarget { + ReadTarget { + chain: self.chain(), + topic: topic.to_string(), + stream_hash: stream_hash(NAMESPACE, "workflow", &self.workflow_id, topic), + } + } + + fn cursor(&self, topic: &str, position: &str) -> String { + mint_cursor("redis", &self.target(topic).stream_hash, position) + } + + /// Appends JSON values as producer `p` the way lang hands them to Core. + async fn core_append( + &self, + sequence: i64, + values: &[Value], + digest: &[u8], + ) -> (String, String) { + let records: Vec<_> = values + .iter() + .enumerate() + .map(|(index, value)| { + stored_append_record( + "events", + "p", + 1, + sequence + index as i64, + &json_record(value), + ) + .unwrap() + .encode_to_vec() + }) + .collect(); + let landed = self + .core + .append(StoreAppendRequest { + chain: Some(self.chain()), + topic: "events".to_string(), + producer_id: "p".to_string(), + attempt: 1, + sequence, + digest: digest.to_vec(), + records, + }) + .await + .unwrap(); + (landed.first_position, landed.last_position) + } + + /// Reads a topic from `after` with the Core reader, as JSON values. + async fn core_read(&self, topic: &str, after: &str) -> Result, StreamFailureKind> { + let page = read_page( + &self.core, + &self.target(topic), + &ReadRequest { + after: after.to_string(), + ..Default::default() + }, + ) + .await + .map_err(|error| error.kind)?; + Ok(page + .records + .iter() + .filter_map(|record| match record.record.as_ref()? { + read_record::Record::Stored(stored) => { + let body = stored.body.as_ref()?; + serde_json::from_slice(&body.data).ok() + } + read_record::Record::Superseded(_) => None, + }) + .collect()) + } +} + +fn json_record(value: &Value) -> AppendRecord { + let payload = Payload { + metadata: [("encoding".to_string(), b"json/plain".to_vec())].into(), + data: value.to_string().into_bytes(), + ..Default::default() + }; + use sha2::{Digest, Sha256}; + AppendRecord { + kind: StreamRecordKind::Data as i32, + content_hash: Sha256::digest(payload.encode_to_vec()).to_vec(), + body: Some(payload), + } +} + +fn values(records: &Value) -> Vec { + records + .as_array() + .unwrap_or_else(|| panic!("not records: {records}")) + .iter() + .map(|record| record["value"].clone()) + .collect() +} + +fn hex(text: &str) -> Vec { + (0..text.len()) + .step_by(2) + .map(|i| u8::from_str_radix(&text[i..i + 2], 16).unwrap()) + .collect() +} + +const WEEK: Duration = Duration::from_secs(7 * 24 * 60 * 60); +/// Long enough that starting the Python side never outlasts it. +const SHORT: Duration = Duration::from_secs(2); + +#[tokio::test] +async fn what_python_wrote_core_reads_and_resumes() { + let Some(both) = pair(WEEK).await else { + return; + }; + let first = + both.brook(json!({"op": "append", "producer": "p", "values": [{"n": 1}, {"n": 2}]})); + both.brook(json!({"op": "append", "producer": "p", "sequence": 3, "values": [{"n": 3}]})); + assert_eq!( + both.core_read("events", "").await.unwrap(), + [json!({"n": 1}), json!({"n": 2}), json!({"n": 3})] + ); + // A cursor the Python provider minted resumes on the Core store. + let cursor = first["cursor"].as_str().unwrap(); + assert_eq!( + both.core_read("events", cursor).await.unwrap(), + [json!({"n": 3})] + ); + let page = read_page(&both.core, &both.target("events"), &ReadRequest::default()) + .await + .unwrap(); + let Some(read_record::Record::Stored(stored)) = &page.records[2].record else { + panic!("no record"); + }; + assert_eq!( + (stored.producer_id.as_str(), stored.attempt, stored.sequence), + ("p", 1, 3) + ); +} + +#[tokio::test] +async fn what_core_wrote_python_reads_and_resumes() { + let Some(both) = pair(WEEK).await else { + return; + }; + let (_, first) = both.core_append(1, &[json!({"n": 1})], b"\x01").await; + both.core_append(2, &[json!({"n": 2})], b"\x02").await; + let read = both.brook(json!({"op": "read", "count": 2})); + assert_eq!(values(&read), [json!({"n": 1}), json!({"n": 2})]); + assert_eq!(read[0]["producer_id"], "p"); + // A cursor the Core store minted resumes on the Python provider. + let after = + both.brook(json!({"op": "read", "count": 1, "after": both.cursor("events", &first)})); + assert_eq!(values(&after), [json!({"n": 2})]); +} + +/// Temporal as the stream layer asks it: one Workflow, running in its first run. +struct Running; + +#[async_trait::async_trait] +impl OwnerClient for Running { + async fn describe( + &self, + _namespace: &str, + _workflow_id: &str, + _run_id: &str, + ) -> Result { + Ok(OwnerDescription { + run_id: FIRST_RUN.to_string(), + first_run_id: FIRST_RUN.to_string(), + status: WorkflowExecutionStatus::Running, + }) + } +} + +impl Both { + /// The stream layer over a store on the same keys, as lang reaches it. + async fn streams(&self) -> Streams { + let mut options = RedisStoreOptions::new(self.url.clone()); + options.key_prefix = self.prefix.clone(); + let store = RedisStore::connect(options).await.unwrap(); + Streams::new( + Arc::new(store), + Arc::new(Running), + StreamsOptions::default(), + ) + } + + fn append_request(&self, values: &[Value], digest: Vec) -> AppendRequest { + AppendRequest { + stream: Some(StreamAddress { + namespace: NAMESPACE.to_string(), + owner_kind: StreamOwnerKind::Workflow as i32, + workflow_id: self.workflow_id.clone(), + run_id: String::new(), + topic: "events".to_string(), + }), + producer: Some(Producer::Named(NamedProducer { + producer_id: "p".to_string(), + attempt: 1, + })), + sequence: 1, + records: values.iter().map(json_record).collect(), + digest, + } + } +} + +#[tokio::test] +async fn a_batch_python_wrote_dedupes_through_core_with_its_digest() { + // Lang takes the digest and the store keeps it as given, so a retry that carries the same + // digest is recognized whichever SDK wrote first. + let Some(both) = pair(WEEK).await else { + return; + }; + let streams = both.streams().await; + let landed = both.brook(json!({"op": "append", "producer": "p", "values": [{"n": 1}]})); + let digest = hex(landed["digest"].as_str().unwrap()); + let retry = streams + .append(both.append_request(&[json!({"n": 1})], digest)) + .await + .unwrap(); + assert_eq!(retry.last_cursor, landed["cursor"].as_str().unwrap()); + let error = streams + .append(both.append_request(&[json!({"n": 1})], vec![7; 32])) + .await + .unwrap_err(); + assert_eq!(error.kind, StreamFailureKind::ProducerDivergent); + assert_eq!(both.core_read("events", "").await.unwrap().len(), 1); +} + +#[tokio::test] +async fn a_batch_core_wrote_dedupes_through_python() { + let Some(both) = pair(WEEK).await else { + return; + }; + let streams = both.streams().await; + let digest = both.brook(json!({"op": "digest", "producer": "p", "values": [{"n": 1}]})); + let landed = streams + .append(both.append_request(&[json!({"n": 1})], hex(digest.as_str().unwrap()))) + .await + .unwrap(); + let retry = both.brook(json!({"op": "append", "producer": "p", "values": [{"n": 1}]})); + assert_eq!(retry["cursor"].as_str().unwrap(), landed.last_cursor); + assert_eq!(both.core_read("events", "").await.unwrap().len(), 1); +} + +#[tokio::test] +async fn either_side_sees_a_chain_the_other_closed() { + let Some(both) = pair(WEEK).await else { + return; + }; + both.core_append(1, &[json!({"n": 1})], b"\x01").await; + both.core.close_chain(&both.chain()).await.unwrap(); + let refused = both.brook(json!({"op": "append", "producer": "q", "values": [{"n": 2}]})); + assert_eq!(refused["error"], "StreamClosedError", "{refused}"); + + let other = pair(WEEK).await.unwrap(); + other.brook(json!({"op": "close"})); + let error = other + .core + .append(StoreAppendRequest { + chain: Some(other.chain()), + topic: "events".to_string(), + producer_id: "p".to_string(), + attempt: 1, + sequence: 1, + digest: b"\x01".to_vec(), + records: vec![b"\x1a\x06events".to_vec()], + }) + .await + .unwrap_err(); + assert_eq!(error.kind, StreamFailureKind::Closed); +} + +#[tokio::test] +async fn a_trim_core_made_expires_a_python_cursor() { + let Some(both) = pair(SHORT).await else { + return; + }; + both.brook(json!({"op": "append", "producer": "p", "values": [{"n": 1}, {"n": 2}]})); + let delivered = both.brook(json!({"op": "read", "count": 1})); + tokio::time::sleep(SHORT + Duration::from_millis(200)).await; + // This append trims both records, and the reader never got the second. + both.core_append(3, &[json!({"n": 3})], b"\x03").await; + let expired = both.brook(json!({ + "op": "read", + "count": 1, + "after": delivered[0]["cursor"], + })); + assert_eq!(expired["error"], "StreamExpiredError", "{expired}"); +} + +#[tokio::test] +async fn a_trim_python_made_expires_a_core_cursor() { + let Some(both) = pair(SHORT).await else { + return; + }; + let (first, _) = both + .core_append(1, &[json!({"n": 1}), json!({"n": 2})], b"\x01") + .await; + tokio::time::sleep(SHORT + Duration::from_millis(200)).await; + both.brook(json!({"op": "append", "producer": "q", "values": [{"n": 3}]})); + assert_eq!( + both.core_read("events", &both.cursor("events", &first)) + .await, + Err(StreamFailureKind::Expired) + ); +} diff --git a/crates/streams/tests/cross_impl/brook.py b/crates/streams/tests/cross_impl/brook.py new file mode 100644 index 0000000000..6b6b3a0a3c --- /dev/null +++ b/crates/streams/tests/cross_impl/brook.py @@ -0,0 +1,144 @@ +"""Drives the Python SDK's Redis provider for the cross-implementation test. + +Run from a Python SDK checkout with ``uv run --no-sync python brook.py ''``. The argument +names one operation on one stream, and the answer is one JSON line. The provider talks to Redis +as in production. A stand-in for the Temporal client describes one running Workflow, so no +server is needed. +""" + +from __future__ import annotations + +import asyncio +import json +import sys +from datetime import timedelta +from types import SimpleNamespace +from typing import Any + +import temporalio.converter +from temporalio.client import WorkflowExecutionStatus +from temporalio.contrib.streams import Cursor, RecordKind, StreamError, StreamRef +from temporalio.contrib.streams._body import content_fingerprint +from temporalio.contrib.streams._wire import to_wire +from temporalio.contrib.streams.redis import RedisStreams + + +class Owner: + """A Temporal client that knows one running Workflow, whose first run is ``first_run``.""" + + def __init__(self, namespace: str, first_run: str) -> None: + self.namespace = namespace + self.data_converter = temporalio.converter.DataConverter.default + self._first_run = first_run + + def get_workflow_handle(self, workflow_id: str, run_id: str | None = None) -> Any: + async def describe() -> Any: + info = SimpleNamespace(first_run_id=self._first_run) + return SimpleNamespace( + status=WorkflowExecutionStatus.RUNNING, + run_id=run_id or self._first_run, + raw_description=SimpleNamespace(workflow_execution_info=info), + ) + + return SimpleNamespace(describe=describe) + + +def record(item: Any) -> dict[str, Any]: + return { + "cursor": item.cursor.token, + "kind": item.kind.name, + "value": item.value, + "producer_id": item.producer_id, + "attempt": item.attempt, + "sequence": item.sequence, + "run_id": item.run_id, + "stale": item.stale, + } + + +async def run(op: dict[str, Any]) -> Any: + provider = RedisStreams( + op["url"], + key_prefix=op["prefix"], + retention=timedelta(milliseconds=op.get("retention_ms", 7 * 24 * 3600 * 1000)), + poll_interval=timedelta(milliseconds=50), + ) + owner = Owner(op["namespace"], op["first_run"]) + ref = StreamRef.for_workflow(op["workflow_id"]) + stream = provider.get_stream_handle(owner, ref) # type: ignore[arg-type] + topic = op.get("topic", "events") + try: + if op["op"] == "append": + producer = stream.producer( + topic=topic, producer_id=op["producer"], attempt=op.get("attempt", 1) + ) + producer._sequence = op.get("sequence", 1) + converter = owner.data_converter.payload_converter + wires = [ + to_wire( + converter, + topic=topic, + kind=RecordKind.DATA, + value=value, + producer_id=op["producer"], + attempt=op.get("attempt", 1), + sequence=producer._sequence + index, + ) + for index, value in enumerate(op["values"]) + ] + digest = content_fingerprint(wires).hex() + cursor = await producer.append(*op["values"]) + return {"cursor": cursor.token, "digest": digest} + if op["op"] == "digest": + converter = owner.data_converter.payload_converter + wires = [ + to_wire( + converter, + topic=topic, + kind=RecordKind.DATA, + value=value, + producer_id=op["producer"], + attempt=op.get("attempt", 1), + sequence=op.get("sequence", 1) + index, + ) + for index, value in enumerate(op["values"]) + ] + return content_fingerprint(wires).hex() + if op["op"] == "read": + records = stream.read(topic=topic, after=Cursor(op.get("after", ""))) + out = [] + try: + while len(out) < op["count"]: + out.append(record(await asyncio.wait_for(records.__anext__(), 5))) + except asyncio.TimeoutError: + return { + "error": "Timeout", + "message": f"got {len(out)} records", + "cursor": None, + } + finally: + await records.aclose() + return out + if op["op"] == "latest": + return (await stream.latest(topic=topic)).token + if op["op"] == "close": + await provider._mark_closed( + provider._chain_keys( + op["namespace"], op["workflow_id"], op["first_run"] + ) + ) + return None + raise ValueError(f"no operation {op['op']!r}") + except StreamError as error: + cursor = getattr(error, "cursor", None) + return { + "error": type(error).__name__, + "message": str(error), + "cursor": cursor.token if cursor is not None else None, + } + finally: + await provider.close() + + +if __name__ == "__main__": + print(json.dumps(asyncio.run(run(json.loads(sys.argv[1]))))) From 5566dc118a05b3f74068278b994655deebbf8be4 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 16 Oct 2026 03:17:44 -0700 Subject: [PATCH 3/4] Refused a Redis read after a position the topic never held. The memory store refused it. On Redis a failover that lost the tail let a reader skip the records written after it. --- crates/streams/src/redis/mod.rs | 9 +++++++++ crates/streams/tests/redis_store.rs | 22 ++++++++++++++++++++++ 2 files changed, 31 insertions(+) diff --git a/crates/streams/src/redis/mod.rs b/crates/streams/src/redis/mod.rs index f572063ec5..1320a1f811 100644 --- a/crates/streams/src/redis/mod.rs +++ b/crates/streams/src/redis/mod.rs @@ -273,6 +273,15 @@ impl TopicState { let Some((position, at)) = position else { return Ok(None); }; + if let Some(last) = &self.last + && at > entry(last)? + { + // The memory store refuses this too. A failover that lost the tail also lands here, + // and a read that went on would skip the records written after it. + return Ok(Some(StreamError::cursor(format!( + "position {position} is one topic {topic:?} never held" + )))); + } if !self.log_exists && found.is_empty() { if !self.meta_exists { return Ok(Some(StreamError::not_found(format!( diff --git a/crates/streams/tests/redis_store.rs b/crates/streams/tests/redis_store.rs index 87a0122e99..218aabc225 100644 --- a/crates/streams/tests/redis_store.rs +++ b/crates/streams/tests/redis_store.rs @@ -1065,3 +1065,25 @@ async fn rule_20_4_a_malformed_entry_fails_with_its_cursor() { .unwrap(); assert_eq!(after.records.len(), 1); } + +#[tokio::test] +async fn a_position_past_the_head_is_one_the_topic_never_held() { + let Some(setup) = setup().await else { + return; + }; + let chain = chain(); + let landed = setup + .store + .append(append(&chain, "p", 1, 1, b"\x01")) + .await + .unwrap(); + let (ms, _) = landed.last_position.split_once('-').unwrap(); + let past = format!("{}-0", ms.parse::().unwrap() + 1000); + let error = setup + .store + .read(read_after(&chain, &past, 10)) + .await + .unwrap_err(); + assert_eq!(error.kind, StreamFailureKind::Cursor, "{error}"); + assert!(error.message.contains("never held"), "{error}"); +} From e03c16236fe06bf10c4e74bd2396974fd547e8c7 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 16 Oct 2026 03:17:44 -0700 Subject: [PATCH 4/4] Sent the read checks script by hash. --- crates/streams/src/redis/mod.rs | 60 +++++++++++++++++++++-------- crates/streams/tests/redis_store.rs | 35 +++++++++++++++++ 2 files changed, 79 insertions(+), 16 deletions(-) diff --git a/crates/streams/src/redis/mod.rs b/crates/streams/src/redis/mod.rs index 1320a1f811..8e016c29c5 100644 --- a/crates/streams/src/redis/mod.rs +++ b/crates/streams/src/redis/mod.rs @@ -32,9 +32,10 @@ use crate::{ use conn::{BlockingPool, Shared}; use errors::{Call, Mapped}; use keys::{ChainKeys, session_field}; -use redis::{Cmd, Value, streams::StreamReadReply}; +use redis::{Cmd, Script, Value, streams::StreamReadReply}; use std::{ collections::HashMap, + sync::LazyLock, time::{Duration, SystemTime, UNIX_EPOCH}, }; use tokio::time::Instant; @@ -150,11 +151,19 @@ impl RedisStore { } /// What a read needs to know besides the records, in one script so a cluster answers it - /// from the slot the chain's keys share. - fn checks(keys: &ChainKeys, topic: &str) -> Cmd { - let mut eval = redis::cmd("EVAL"); - eval.arg(TOPIC_STATE) - .arg(4) + /// from the slot the chain's keys share. By hash, unless the server answered it lost the + /// script, since the text costs every read its bytes and a parse. + fn checks(keys: &ChainKeys, topic: &str, by_hash: bool) -> Cmd { + let mut eval = if by_hash { + let mut eval = redis::cmd("EVALSHA"); + eval.arg(TOPIC_STATE_SCRIPT.get_hash()); + eval + } else { + let mut eval = redis::cmd("EVAL"); + eval.arg(TOPIC_STATE); + eval + }; + eval.arg(4) .arg(keys.log(topic)) .arg(keys.meta(topic)) .arg(keys.chain()) @@ -220,6 +229,12 @@ return {redis.call('EXISTS', KEYS[1]), redis.call('EXISTS', KEYS[2]), \ redis.call('HGET', KEYS[2], 'trimmed'), redis.call('HGET', KEYS[2], 'last'), \ redis.call('HGET', KEYS[3], ARGV[1]), redis.call('HGET', KEYS[4], ARGV[1])}"; +static TOPIC_STATE_SCRIPT: LazyLock