From 1483c6770019656cc8d65de4eed969d65b87c08a Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 16 Oct 2026 12:30:43 -0700 Subject: [PATCH 1/3] Polled with a backoff on stores that can't wait for a record. A waiting read on a store without a long poll answered at once, so Streams::read spun until its deadline. --- crates/streams/src/reader.rs | 58 +++++++++++++++++++++++++++--------- 1 file changed, 44 insertions(+), 14 deletions(-) diff --git a/crates/streams/src/reader.rs b/crates/streams/src/reader.rs index e1c614e4e..b8adbbeb0 100644 --- a/crates/streams/src/reader.rs +++ b/crates/streams/src/reader.rs @@ -9,11 +9,11 @@ use crate::{ BEGINNING, END, StreamError, StreamResult, StreamStore, cursor_position, mint_cursor, proto::{ ChainId, ReadRecord, ReadRequest, ReadResponse, StoreLatestRequest, StoreReadRequest, - StoredRecord, StreamRecord, StreamRecordKind, Supersession, read_record, + StoreReadResponse, StoredRecord, StreamRecord, StreamRecordKind, Supersession, read_record, }, }; use prost::Message; -use std::collections::HashMap; +use std::{collections::HashMap, time::Duration}; /// The low bits of a Workflow producer's attempt, which count the reruns of one task. The high /// bits name the last task that completed before it. @@ -345,18 +345,21 @@ pub(crate) async fn read_with( state.watermark_known = true; } - let page = store - .read(StoreReadRequest { - chain: Some(target.chain.clone()), - topic: target.topic.clone(), - after_position: position.clone(), - wait: request.wait, - max_records: match request.max_records { - 0 => DEFAULT_MAX_RECORDS, - max => max, - }, - }) - .await?; + let page_request = StoreReadRequest { + chain: Some(target.chain.clone()), + topic: target.topic.clone(), + after_position: position.clone(), + wait: request.wait, + max_records: match request.max_records { + 0 => DEFAULT_MAX_RECORDS, + max => max, + }, + }; + let page = if capabilities.long_poll { + store.read(page_request).await? + } else { + poll(store, page_request).await? + }; if !capabilities.kept_last { let mark = if position.is_empty() { &state.watermark @@ -442,6 +445,33 @@ pub(crate) async fn read_with( )) } +/// The first poll interval of a read on a store with no long poll, and the longest. +const POLL_FIRST: Duration = Duration::from_millis(50); +const POLL_MAX: Duration = Duration::from_secs(1); + +/// Reads a page on a store that can't wait for a record, asking again with a backoff until a +/// record arrives, the stream closes or the request's wait passes. +async fn poll( + store: &dyn StreamStore, + request: StoreReadRequest, +) -> StreamResult { + let wait = request + .wait + .and_then(|wait| Duration::try_from(wait).ok()) + .unwrap_or_default(); + let deadline = tokio::time::Instant::now() + wait; + let mut interval = POLL_FIRST; + loop { + let page = store.read(request.clone()).await?; + let remaining = deadline.saturating_duration_since(tokio::time::Instant::now()); + if !page.records.is_empty() || page.closed || remaining.is_zero() { + return Ok(page); + } + tokio::time::sleep(interval.min(remaining)).await; + interval = (interval * 2).min(POLL_MAX); + } +} + /// Fails as expired when retention may have dropped records after `mark` that the read had not /// delivered, on a store that doesn't keep a topic's last position. Retention drops only the /// oldest records, so while the record at `mark` is still there, nothing after it is gone. It From acd7d34000225ebe6a5628847df31637be28a280 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 16 Oct 2026 12:30:43 -0700 Subject: [PATCH 2/3] Split the conformance suite by capability and ran it on masked stores. Every store runs the required cases. Masked memory stores and plain Redis also run the cases of the fallbacks they take. --- crates/streams/tests/conformance.rs | 291 +++++++++++++++++++++++++++- 1 file changed, 289 insertions(+), 2 deletions(-) diff --git a/crates/streams/tests/conformance.rs b/crates/streams/tests/conformance.rs index 681240dec..cf6b8e939 100644 --- a/crates/streams/tests/conformance.rs +++ b/crates/streams/tests/conformance.rs @@ -15,8 +15,9 @@ use std::{ }; use temporalio_common::protos::temporal::api::common::v1::Payload; use temporalio_streams::{ - BEGINNING, END, MemoryStore, ReadTarget, StreamResult, StreamStore, mint_cursor, proto::*, - read_page, stored_append_record, stream_hash, + BEGINNING, CHAIN_LOG, Capabilities, END, MemoryStore, ReadTarget, StreamResult, StreamStore, + append_batch, close_chain, close_topic, mint_cursor, proto::*, read_page, stored_append_record, + stream_hash, }; #[async_trait::async_trait] @@ -73,6 +74,15 @@ async fn memory() -> Option { Some(Memory(Arc::new(MemoryStore::new()))) } +/// The memory store with only the capabilities `mask` leaves, so each fallback runs. +async fn masked(mask: impl FnOnce(&mut Capabilities)) -> Option { + let mut capabilities = MemoryStore::new().capabilities(); + mask(&mut capabilities); + Some(Memory(Arc::new(MemoryStore::with_capabilities( + capabilities, + )))) +} + #[cfg(feature = "redis")] mod on_redis { use super::*; @@ -266,6 +276,15 @@ mod on_redis { } pub(crate) async fn redis(variable: &str, cluster: bool) -> Option { + redis_in(variable, cluster, false).await + } + + /// Redis with no Lua, so the store keeps no producer state and no last position. + pub(crate) async fn plain_redis(variable: &str, cluster: bool) -> Option { + redis_in(variable, cluster, true).await + } + + async fn redis_in(variable: &str, cluster: bool, plain: bool) -> Option { let Ok(url) = std::env::var(variable) else { eprintln!("set {variable} to run the Redis store cases"); return None; @@ -275,6 +294,7 @@ mod on_redis { let mut options = RedisStoreOptions::new(url.clone()); options.cluster = cluster; options.key_prefix = prefix.clone(); + options.plain = plain; Some(Redis { store: Arc::new(RedisStore::connect(options).await.unwrap()), raw: Raw::connect(&url, cluster).await, @@ -1195,6 +1215,140 @@ mod cases { ["1"] ); } + + /// What a store declares agrees with the calls it answers. + pub(crate) async fn the_declared_capabilities_match_the_calls(case: &dyn Case) { + let store = case.store(); + let declared = store.capabilities(); + let chain = case.chain(); + let unsupported = |result: StreamResult<()>| { + result.is_err_and(|error| error.kind == StreamFailureKind::Unsupported) + }; + assert_eq!( + unsupported(store.close_chain(&chain).await), + !declared.producer_state + ); + assert_eq!( + unsupported(store.close_topic(&chain, OUT).await), + !declared.producer_state + ); + assert_eq!( + unsupported(store.retire_producer(&chain, OUT, "p", 1).await), + !declared.producer_state + ); + assert_eq!( + unsupported(store.list_topics(&chain).await.map(drop)), + !declared.topic_listing + ); + let delete = store + .delete_owner(DeleteOwnerRequest { + namespace: chain.namespace.clone(), + owner_kind: StreamOwnerKind::Workflow as i32, + workflow_id: chain.workflow_id.clone(), + }) + .await; + assert_eq!(unsupported(delete.map(drop)), !declared.delete_owner); + assert!(declared.max_record_bytes > 0 && declared.max_batch_bytes > 0); + } + + /// Without producer state a repeat after a lost answer lands again, and readers drop it. + pub(crate) async fn a_repeat_lands_a_copy_that_readers_drop(case: &dyn Case) { + let chain = case.chain(); + let mut producer = Producer::new(case, &chain, OUT, "p", 1); + let records = [data("1"), data("2")]; + let request = producer.request(&records); + let first = producer.send(&records).await.unwrap(); + let again = case.store().append(request).await.unwrap(); + assert_ne!(again.first_position, first.first_position); + producer.append(&["3"]).await; + let read = read(case, &chain, OUT, BEGINNING).await.unwrap(); + assert_eq!(values(&read), ["1", "2", "3"]); + } + + /// Without producer state a divergent batch lands, and readers keep the first one. + pub(crate) async fn a_divergent_batch_lands_and_readers_keep_the_first(case: &dyn Case) { + let chain = case.chain(); + let mut producer = Producer::new(case, &chain, OUT, "p", 1); + let divergent = producer.request(&[data("other")]); + producer.append(&["1"]).await; + case.store().append(divergent).await.unwrap(); + let read = read(case, &chain, OUT, BEGINNING).await.unwrap(); + assert_eq!(values(&read), ["1"]); + } + + /// Without close flags a close is a record, and every read ends at the first one. + pub(crate) async fn a_close_record_ends_every_read(case: &dyn Case) { + let store = case.store(); + let chain = case.chain(); + let mut producer = Producer::new(case, &chain, OUT, "p", 1); + let landed = producer.append(&["1"]).await; + close_topic(&*store, &chain, OUT, "closer", 1) + .await + .unwrap(); + close_topic(&*store, &chain, OUT, "late", 1).await.unwrap(); + // The store can't refuse what comes after the close, and no reader delivers it. + producer.append(&["2"]).await; + let first = read(case, &chain, OUT, BEGINNING).await.unwrap(); + assert_eq!(values(&first), ["1"]); + assert_eq!( + first.cursor, + cursor(case, &chain, OUT, &landed.last_position) + ); + let next = read_with( + case, + &chain, + OUT, + &first.cursor, + Duration::ZERO, + &first.state, + ) + .await + .unwrap(); + assert!(next.records.is_empty()); + let resumed = read(case, &chain, OUT, &first.cursor).await.unwrap(); + assert!(resumed.records.is_empty()); + } + + /// Without close flags a chain close writes a close on each topic it finds. + pub(crate) async fn a_chain_close_reaches_every_topic(case: &dyn Case) { + let store = case.store(); + let chain = case.chain(); + for topic in [OUT, OTHER] { + let mut producer = Producer::new(case, &chain, topic, "p", 1); + let request = producer.request(&[data(topic)]); + append_batch(&*store, request).await.unwrap(); + producer.sequence += 1; + producer.append(&["after"]).await; + } + close_chain(&*store, &chain, "@run", 1).await.unwrap(); + for topic in [OUT, OTHER] { + let mut producer = Producer::new(case, &chain, topic, "q", 1); + producer.append(&["late"]).await; + let read = read(case, &chain, topic, BEGINNING).await.unwrap(); + assert_eq!(values(&read), [topic, "after"], "{topic}"); + } + // A topic reader skips the names, so the control log reads as empty either way. + let control = read(case, &chain, CHAIN_LOG, BEGINNING).await.unwrap(); + assert!(control.records.is_empty()); + } + + /// Without a kept last position a read whose cursor record retention dropped is expired. + pub(crate) async fn a_lost_cursor_record_reads_as_expired(case: &dyn Case) { + let chain = case.chain(); + let mut producer = Producer::new(case, &chain, OUT, "p", 1); + let first = producer.append(&["1"]).await; + producer.append(&["2", "3"]).await; + let at = cursor(case, &chain, OUT, &first.last_position); + assert_eq!( + values(&read(case, &chain, OUT, &at).await.unwrap()), + ["2", "3"] + ); + case.drop_oldest(&chain, OUT, 1).await; + assert_eq!( + kind(read(case, &chain, OUT, &at).await), + StreamFailureKind::Expired + ); + } } macro_rules! conformance { @@ -1245,6 +1399,8 @@ mod memory { a_refused_append_leaves_no_trace_of_its_topic, retiring_a_producer_drops_only_its_dedupe_state, deleting_an_owner_drops_every_chain_of_it, + the_declared_capabilities_match_the_calls, + a_lost_cursor_record_reads_as_expired, ); } @@ -1286,6 +1442,8 @@ macro_rules! redis_cases { a_closed_topic_leaves_the_others_open, a_closed_topic_answers_a_repeat_and_ends_reads, retiring_a_producer_drops_only_its_dedupe_state, + the_declared_capabilities_match_the_calls, + a_lost_cursor_record_reads_as_expired, ); }; } @@ -1336,3 +1494,132 @@ mod documented_acl_cluster_redis { deleting_an_owner_drops_every_chain_of_it, ); } + +/// The cases every store passes, whatever it declares beyond the log. A cursor whose record +/// retention dropped resumes only with a kept last position or dense positions, so that case +/// runs apart. +macro_rules! required { + ($case:expr) => { + conformance!($case; + append_read_roundtrip, + a_batch_lands_in_order_and_names_its_last_record, + an_empty_append_is_refused, + an_append_without_a_digest_is_refused, + latest_is_empty_on_an_empty_topic, + a_cursor_resumes_strictly_after_its_record, + latest_positions_a_reader_at_the_end, + end_reads_only_what_arrives_after_the_read_starts, + a_read_waits_no_longer_than_asked, + dropping_a_waiting_read_releases_it, + a_reader_that_falls_behind_retention_is_told, + an_empty_read_from_the_beginning_with_nothing_trimmed_goes_on, + rule_64_1_a_gap_in_the_sequence_is_accepted, + producers_and_attempts_dedupe_apart, + a_new_attempt_supersedes_the_old_one, + a_read_resumed_before_a_new_attempt_reports_it, + a_record_that_does_not_parse_fails_with_its_cursor, + finish_is_a_record_of_its_own, + topics_and_chains_are_apart, + a_cursor_from_another_stream_or_store_is_refused, + the_declared_capabilities_match_the_calls, + a_lost_cursor_record_reads_as_expired, + ); + }; +} + +/// The cases of a store with no producer state, where closes are records. +macro_rules! without_producer_state { + ($case:expr) => { + conformance!($case; + a_repeat_lands_a_copy_that_readers_drop, + a_divergent_batch_lands_and_readers_keep_the_first, + a_close_record_ends_every_read, + a_chain_close_reaches_every_topic, + ); + }; +} + +/// A store with only the log, so every fallback runs. +mod memory_log_only { + async fn case() -> Option { + super::masked(|c| { + *c = temporalio_streams::Capabilities { + dense_positions: true, + ..temporalio_streams::Capabilities::log_only() + } + }) + .await + } + required!(case()); + // Dense positions tell a page right after a dropped cursor is whole. + conformance!(case(); beginning_starts_at_the_oldest_record_still_held); + without_producer_state!(case()); +} + +mod memory_without_producer_state { + async fn case() -> Option { + super::masked(|c| c.producer_state = false).await + } + required!(case()); + // Dense positions tell a page right after a dropped cursor is whole. + conformance!(case(); beginning_starts_at_the_oldest_record_still_held); + without_producer_state!(case()); + conformance!(case(); + an_empty_read_from_the_beginning_learns_of_a_trim, + rule_11_2_an_empty_read_goes_on_when_a_quiet_log_comes_back, + deleting_an_owner_drops_every_chain_of_it, + ); +} + +/// No producer state and no topic listing, so a chain close finds topics in the control log. +mod memory_without_listing { + async fn case() -> Option { + super::masked(|c| { + c.producer_state = false; + c.topic_listing = false; + }) + .await + } + required!(case()); + // Dense positions tell a page right after a dropped cursor is whole. + conformance!(case(); beginning_starts_at_the_oldest_record_still_held); + without_producer_state!(case()); +} + +mod memory_without_kept_last { + async fn case() -> Option { + super::masked(|c| c.kept_last = false).await + } + required!(case()); + // Dense positions tell a page right after a dropped cursor is whole. + conformance!(case(); beginning_starts_at_the_oldest_record_still_held); + conformance!(case(); + a_retried_append_returns_the_original_positions, + a_retry_with_other_ciphertext_still_deduplicates, + a_divergent_retry_is_refused, + a_sequence_below_the_newest_is_refused, + rule_9_1_a_batch_inside_the_held_batch_is_refused, + a_closed_chain_refuses_new_batches_but_answers_a_repeat, + a_closed_topic_leaves_the_others_open, + a_closed_topic_answers_a_repeat_and_ends_reads, + retiring_a_producer_drops_only_its_dedupe_state, + ); +} + +#[cfg(feature = "redis")] +mod plain_redis { + required!(super::on_redis::plain_redis("STREAMS_REDIS_URL", false)); + without_producer_state!(super::on_redis::plain_redis("STREAMS_REDIS_URL", false)); +} + +#[cfg(feature = "redis")] +mod plain_cluster_redis { + required!(super::on_redis::plain_redis( + "STREAMS_REDIS_CLUSTER_URL", + true + )); + without_producer_state!(super::on_redis::plain_redis( + "STREAMS_REDIS_CLUSTER_URL", + true + )); +} From 725bee46f85a6498f96317ba8a55004088cfab81 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 16 Oct 2026 12:34:50 -0700 Subject: [PATCH 3/3] Tested the fallbacks under faults and where readers end. A fault wrapper loses answers, lands late copies and copies after a close, and crashes between a topic name and its record. It also counts writes, so the tests show that readers write nothing. --- crates/streams/docs/redis.md | 7 +- crates/streams/tests/fallbacks.rs | 629 ++++++++++++++++++++++++++++++ 2 files changed, 633 insertions(+), 3 deletions(-) create mode 100644 crates/streams/tests/fallbacks.rs diff --git a/crates/streams/docs/redis.md b/crates/streams/docs/redis.md index b00c428a9..da511061a 100644 --- a/crates/streams/docs/redis.md +++ b/crates/streams/docs/redis.md @@ -56,6 +56,7 @@ Listing a chain's topics scans every primary, as a delete does, and needs `+scan **Tests.** The Redis tests run when `STREAMS_REDIS_URL` names a standalone server and `STREAMS_REDIS_CLUSTER_URL` a cluster seed. `scripts/redis-test-servers.sh` starts both. The -conformance suite runs on each, and once more as a user with exactly the ACL above. The -cross-implementation test also needs `STREAMS_BROOK_PY`, a Python SDK checkout whose provider -predates the move. +conformance suite runs on each, once more as a user with exactly the ACL above, and once in +plain mode with the cases of the fallbacks. `tests/fallbacks.rs` adds faults on plain Redis +too. The cross-implementation test also needs `STREAMS_BROOK_PY`, a Python SDK checkout whose +provider predates the move. diff --git a/crates/streams/tests/fallbacks.rs b/crates/streams/tests/fallbacks.rs new file mode 100644 index 000000000..4f7a5e9fd --- /dev/null +++ b/crates/streams/tests/fallbacks.rs @@ -0,0 +1,629 @@ +//! The fallbacks under faults, through the stream layer, on stores with and without producer +//! state: lost answers, late copies, copies after a close, a crash between a topic name and its +//! record, readers that never write, where readers end when the owner ends, and reads that give +//! the same records however they page. + +use prost::Message; +use sha2::{Digest, Sha256}; +use std::{ + collections::HashMap, + sync::{ + Arc, Mutex, + atomic::{AtomicBool, AtomicUsize, Ordering}, + }, + time::Duration, +}; +use temporalio_common::protos::temporal::api::{ + common::v1::Payload, enums::v1::WorkflowExecutionStatus, +}; +use temporalio_streams::{ + BEGINNING, CHAIN_LOG, Capabilities, MemoryStore, OwnerClient, OwnerDescription, OwnerError, + ReadTarget, StreamError, StreamResult, StreamStore, Streams, StreamsOptions, close_chain, + proto::*, read_page, stream_hash, +}; + +/// A store that forwards to another and injects the faults a test asks for. It also counts +/// writes, so a test can tell that a reader made none. +struct Faulty { + inner: Arc, + /// How many of the next appends land and then answer as if the answer was lost. + lose: AtomicUsize, + /// The next append of a topic other than the control log fails before it lands. + fail_next_data: AtomicBool, + /// The next append also keeps a copy to land later. + keep_copy: AtomicBool, + copies: Mutex>, + writes: AtomicUsize, +} + +impl Faulty { + fn new(inner: Arc) -> Arc { + Arc::new(Self { + inner, + lose: AtomicUsize::new(0), + fail_next_data: AtomicBool::new(false), + keep_copy: AtomicBool::new(false), + copies: Mutex::default(), + writes: AtomicUsize::new(0), + }) + } + + /// Lands the copies kept so far, as a network does with a delayed request. A store with + /// producer state refuses a copy of a batch older than the newest. + async fn land_copies(&self) { + let copies = std::mem::take(&mut *self.copies.lock().unwrap()); + for copy in copies { + match self.inner.append(copy).await { + Err(error) if self.inner.capabilities().producer_state => { + assert_eq!(error.kind, StreamFailureKind::ProducerStale); + } + landed => drop(landed.unwrap()), + } + } + } + + fn wrote(&self) { + self.writes.fetch_add(1, Ordering::Relaxed); + } +} + +#[async_trait::async_trait] +impl StreamStore for Faulty { + fn name(&self) -> &str { + self.inner.name() + } + + fn capabilities(&self) -> Capabilities { + self.inner.capabilities() + } + + async fn append(&self, request: StoreAppendRequest) -> StreamResult { + self.wrote(); + if request.topic != CHAIN_LOG && self.fail_next_data.swap(false, Ordering::Relaxed) { + return Err(StreamError::storage("the Worker crashed before the append")); + } + if self.keep_copy.swap(false, Ordering::Relaxed) { + self.copies.lock().unwrap().push(request.clone()); + } + let landed = self.inner.append(request).await?; + let lost = self + .lose + .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |n| n.checked_sub(1)); + if lost.is_ok() { + return Err(StreamError::outcome_unknown("the answer was lost")); + } + Ok(landed) + } + + async fn read(&self, request: StoreReadRequest) -> StreamResult { + self.inner.read(request).await + } + + async fn latest(&self, request: StoreLatestRequest) -> StreamResult { + self.inner.latest(request).await + } + + async fn record_at( + &self, + chain: &ChainId, + topic: &str, + position: &str, + ) -> StreamResult>> { + self.inner.record_at(chain, topic, position).await + } + + async fn newest_ever(&self, chain: &ChainId, topic: &str) -> StreamResult { + self.inner.newest_ever(chain, topic).await + } + + async fn dropped_after( + &self, + chain: &ChainId, + topic: &str, + mark: &str, + ) -> StreamResult> { + self.inner.dropped_after(chain, topic, mark).await + } + + async fn close_chain(&self, chain: &ChainId) -> StreamResult<()> { + self.wrote(); + self.inner.close_chain(chain).await + } + + async fn close_topic(&self, chain: &ChainId, topic: &str) -> StreamResult<()> { + self.wrote(); + self.inner.close_topic(chain, topic).await + } + + async fn retire_producer( + &self, + chain: &ChainId, + topic: &str, + producer_id: &str, + attempt: i64, + ) -> StreamResult<()> { + self.wrote(); + self.inner + .retire_producer(chain, topic, producer_id, attempt) + .await + } + + async fn list_topics(&self, chain: &ChainId) -> StreamResult> { + self.inner.list_topics(chain).await + } + + async fn delete_owner(&self, request: DeleteOwnerRequest) -> StreamResult { + self.wrote(); + self.inner.delete_owner(request).await + } +} + +/// Temporal, with one Workflow whose only run is `run-1`. +struct Owner { + status: Mutex>, +} + +impl Owner { + fn running() -> Arc { + Arc::new(Self { + status: Mutex::new(Ok(WorkflowExecutionStatus::Running)), + }) + } + + fn end(&self, status: WorkflowExecutionStatus) { + *self.status.lock().unwrap() = Ok(status); + } + + fn forget(&self) { + *self.status.lock().unwrap() = Err(()); + } +} + +#[async_trait::async_trait] +impl OwnerClient for Owner { + async fn describe( + &self, + _: &str, + workflow_id: &str, + _: &str, + ) -> Result { + match *self.status.lock().unwrap() { + Ok(status) => Ok(OwnerDescription { + run_id: "run-1".to_string(), + first_run_id: "run-1".to_string(), + status, + }), + Err(()) => Err(OwnerError::NotFound(workflow_id.to_string())), + } + } +} + +static WORKFLOWS: AtomicUsize = AtomicUsize::new(0); + +/// One stream of a Workflow no other test uses, so a shared Redis keeps tests apart. +fn new_stream() -> StreamAddress { + StreamAddress { + namespace: "ns".to_string(), + owner_kind: StreamOwnerKind::Workflow as i32, + workflow_id: format!( + "fallbacks-{}-{}", + std::process::id(), + WORKFLOWS.fetch_add(1, Ordering::Relaxed) + ), + run_id: String::new(), + topic: "out".to_string(), + } +} + +fn chain(stream: &StreamAddress) -> ChainId { + ChainId { + namespace: stream.namespace.clone(), + workflow_id: stream.workflow_id.clone(), + first_run_id: "run-1".to_string(), + } +} + +fn options() -> StreamsOptions { + StreamsOptions { + owner_recheck: Duration::from_millis(300), + owner_check_min: Duration::from_millis(20), + owner_check_max: Duration::from_millis(80), + ..StreamsOptions::default() + } +} + +/// The stores each test runs on: memory with every capability, memory with no producer state, +/// memory with only the log, and Redis in plain mode when `STREAMS_REDIS_URL` is set. +async fn stores() -> Vec<(&'static str, Arc)> { + let full = MemoryStore::new().capabilities(); + let mut stores: Vec<(&'static str, Arc)> = vec![ + ("memory", Arc::new(MemoryStore::new())), + ( + "memory without producer state", + Arc::new(MemoryStore::with_capabilities(Capabilities { + producer_state: false, + ..full + })), + ), + ( + "memory with only the log", + Arc::new(MemoryStore::with_capabilities(Capabilities { + dense_positions: true, + ..Capabilities::log_only() + })), + ), + ]; + #[cfg(feature = "redis")] + if let Ok(url) = std::env::var("STREAMS_REDIS_URL") { + let mut options = temporalio_streams::RedisStoreOptions::new(url); + options.key_prefix = format!("fallbacks-{}", std::process::id()); + options.plain = true; + let store = temporalio_streams::RedisStore::connect(options) + .await + .unwrap(); + stores.push(("plain Redis", Arc::new(store))); + } + stores +} + +fn data(value: &str) -> AppendRecord { + AppendRecord { + kind: StreamRecordKind::Data as i32, + body: Some(Payload { + data: value.as_bytes().to_vec(), + ..Default::default() + }), + content_hash: Sha256::digest(value.as_bytes()).to_vec(), + } +} + +fn batch(stream: &StreamAddress, producer: &str, sequence: i64, values: &[&str]) -> AppendRequest { + AppendRequest { + stream: Some(stream.clone()), + producer: Some(append_request::Producer::Named(NamedProducer { + producer_id: producer.to_string(), + attempt: 1, + })), + sequence, + records: values.iter().map(|value| data(value)).collect(), + digest: Sha256::digest(values.concat().as_bytes()).to_vec(), + } +} + +/// Appends as lang does: a failure that may have landed is sent again. +async fn append(streams: &Streams, request: AppendRequest) -> StreamResult { + loop { + match streams.append(request.clone()).await { + Err(error) + if matches!( + error.kind, + StreamFailureKind::OutcomeUnknown | StreamFailureKind::Storage + ) => {} + answer => return answer, + } + } +} + +/// Reads the stream until the read is done, and answers what it delivered. +async fn read_to_end(streams: &Streams, stream: &StreamAddress) -> Vec { + let mut request = ReadRequest { + stream: Some(stream.clone()), + wait: Some(Duration::from_millis(500).try_into().unwrap()), + ..Default::default() + }; + let mut delivered = vec![]; + for _ in 0..100 { + let page = streams.read(request.clone()).await.unwrap(); + delivered.extend(values(&page.records)); + if page.done { + return delivered; + } + request.after = page.cursor; + request.state = page.state; + } + panic!("the read never ended"); +} + +fn values(records: &[ReadRecord]) -> Vec { + records + .iter() + .filter_map(|record| match record.record.as_ref()? { + read_record::Record::Stored(stored) => { + Some(String::from_utf8(stored.body.clone()?.data).unwrap()) + } + read_record::Record::Superseded(_) => None, + }) + .collect() +} + +#[tokio::test] +async fn a_lost_answer_and_a_late_copy_reach_readers_once() { + for (label, store) in stores().await { + let faulty = Faulty::new(store); + let streams = Streams::new(faulty.clone(), Owner::running(), options()); + let stream = new_stream(); + faulty.lose.store(1, Ordering::Relaxed); + append(&streams, batch(&stream, "p", 1, &["1"])) + .await + .unwrap(); + faulty.keep_copy.store(true, Ordering::Relaxed); + append(&streams, batch(&stream, "p", 2, &["2", "3"])) + .await + .unwrap(); + append(&streams, batch(&stream, "p", 4, &["4"])) + .await + .unwrap(); + // A copy of the second batch lands after the third, as a delayed request does. + faulty.land_copies().await; + append(&streams, batch(&stream, "p", 5, &["5"])) + .await + .unwrap(); + let page = streams + .read(ReadRequest { + stream: Some(stream.clone()), + ..Default::default() + }) + .await + .unwrap(); + assert_eq!(values(&page.records), ["1", "2", "3", "4", "5"], "{label}"); + } +} + +#[tokio::test] +async fn a_copy_after_a_close_is_not_delivered() { + for (label, store) in stores().await { + let faulty = Faulty::new(store.clone()); + let streams = Streams::new(faulty.clone(), Owner::running(), options()); + let stream = new_stream(); + faulty.keep_copy.store(true, Ordering::Relaxed); + append(&streams, batch(&stream, "p", 1, &["1"])) + .await + .unwrap(); + streams + .close(CloseRequest { + stream: Some(stream.clone()), + }) + .await + .unwrap(); + faulty.land_copies().await; + let after = append(&streams, batch(&stream, "p", 2, &["2"])).await; + // With close flags the store refuses the append. Without them it lands unread. + assert_eq!( + after.is_err(), + store.capabilities().producer_state, + "{label}" + ); + assert_eq!(read_to_end(&streams, &stream).await, ["1"], "{label}"); + } +} + +#[tokio::test] +async fn a_crash_between_a_topic_name_and_its_record_leaves_a_harmless_name() { + let store = Arc::new(MemoryStore::with_capabilities(Capabilities { + producer_state: false, + topic_listing: false, + ..MemoryStore::new().capabilities() + })); + let faulty = Faulty::new(store.clone()); + let streams = Streams::new(faulty.clone(), Owner::running(), options()); + let stream = new_stream(); + faulty.fail_next_data.store(true, Ordering::Relaxed); + append(&streams, batch(&stream, "p", 1, &["1"])) + .await + .unwrap(); + // A producer that crashed for good after naming its topic. + let quiet = StreamAddress { + topic: "quiet".to_string(), + ..stream.clone() + }; + faulty.fail_next_data.store(true, Ordering::Relaxed); + let crashed = streams.append(batch(&quiet, "q", 1, &["x"])).await; + assert_eq!(crashed.unwrap_err().kind, StreamFailureKind::Storage); + let names = store + .read(StoreReadRequest { + chain: Some(chain(&stream)), + topic: CHAIN_LOG.to_string(), + max_records: 10, + ..Default::default() + }) + .await + .unwrap(); + assert_eq!(names.records.len(), 3); + close_chain(&*store, &chain(&stream), "@run-1", 1) + .await + .unwrap(); + assert_eq!(read_to_end(&streams, &stream).await, ["1"]); + assert!(read_to_end(&streams, &quiet).await.is_empty()); +} + +#[tokio::test] +async fn readers_write_nothing() { + for (label, store) in stores().await { + for end in ["completed", "terminated", "gone"] { + let faulty = Faulty::new(store.clone()); + let owner = Owner::running(); + let streams = Streams::new(faulty.clone(), owner.clone(), options()); + let stream = new_stream(); + append(&streams, batch(&stream, "p", 1, &["1"])) + .await + .unwrap(); + let before = faulty.writes.load(Ordering::Relaxed); + let mut request = ReadRequest { + stream: Some(stream.clone()), + wait: Some(Duration::from_millis(200).try_into().unwrap()), + ..Default::default() + }; + let first = streams.read(request.clone()).await.unwrap(); + let mut delivered = values(&first.records); + request.after = first.cursor; + request.state = first.state; + match end { + "completed" => owner.end(WorkflowExecutionStatus::Completed), + "terminated" => owner.end(WorkflowExecutionStatus::Terminated), + _ => owner.forget(), + } + // A gone owner ends a read only on a second answer a recheck apart. + let mut done = false; + for _ in 0..20 { + let page = streams.read(request.clone()).await.unwrap(); + delivered.extend(values(&page.records)); + if page.done { + done = true; + break; + } + request.after = page.cursor; + request.state = page.state; + tokio::time::sleep(Duration::from_millis(100)).await; + } + assert!(done, "{label} {end}"); + assert_eq!(delivered, ["1"], "{label} {end}"); + assert_eq!( + faulty.writes.load(Ordering::Relaxed), + before, + "{label} {end}" + ); + } + } +} + +#[tokio::test] +async fn readers_agree_on_the_end_only_after_a_close() { + for (label, store) in stores().await { + // The owner's Core closes the chain after the task that ends it, so every reader ends + // at the same record. + let owner = Owner::running(); + let streams = Streams::new(store.clone(), owner.clone(), options()); + let stream = new_stream(); + append(&streams, batch(&stream, "p", 1, &["1"])) + .await + .unwrap(); + close_chain(&*store, &chain(&stream), "@run-1", 1) + .await + .unwrap(); + owner.end(WorkflowExecutionStatus::Completed); + let _ = streams.append(batch(&stream, "p", 2, &["2"])).await; + assert_eq!(read_to_end(&streams, &stream).await, ["1"], "{label}"); + assert_eq!(read_to_end(&streams, &stream).await, ["1"], "{label}"); + + // A terminate, or a Worker crash before the close, runs no Core step. Readers end on + // the owner's status, each where it drains, until a producer's owner check closes. + for status in [ + WorkflowExecutionStatus::Terminated, + WorkflowExecutionStatus::Completed, + ] { + let owner = Owner::running(); + let streams = Streams::new(store.clone(), owner.clone(), options()); + let stream = new_stream(); + append(&streams, batch(&stream, "p", 1, &["1"])) + .await + .unwrap(); + owner.end(status); + let early = read_to_end(&streams, &stream).await; + // The producer's last owner check is fresh, so it still writes. + append(&streams, batch(&stream, "p", 2, &["2"])) + .await + .unwrap(); + let later = read_to_end(&streams, &stream).await; + assert_eq!(early, ["1"], "{label} {status:?}"); + assert_eq!(later, ["1", "2"], "{label} {status:?}"); + tokio::time::sleep(options().owner_recheck).await; + let refused = streams.append(batch(&stream, "p", 3, &["3"])).await; + assert_eq!(refused.unwrap_err().kind, StreamFailureKind::Closed); + let other = streams.append(batch(&stream, "q", 1, &["4"])).await; + assert_eq!(other.unwrap_err().kind, StreamFailureKind::Closed); + assert_eq!(read_to_end(&streams, &stream).await, ["1", "2"], "{label}"); + } + } +} + +/// Reads every page of a topic with `max_records` per call, carrying the state. +async fn paged(store: &dyn StreamStore, target: &ReadTarget, max_records: u32) -> Vec { + let mut request = ReadRequest { + after: BEGINNING.to_string(), + max_records, + ..Default::default() + }; + let mut delivered = vec![]; + for _ in 0..100 { + let page = read_page(store, target, &request).await.unwrap(); + if page.cursor == request.after && page.records.is_empty() { + return delivered; + } + delivered.extend(values(&page.records)); + request.after = page.cursor; + request.state = page.state; + } + panic!("the read never ended"); +} + +#[tokio::test] +async fn a_read_gives_the_same_records_however_it_pages() { + for (label, store) in stores().await { + if store.capabilities().producer_state { + continue; + } + let stream = new_stream(); + let chain = chain(&stream); + let record = |producer: &str, attempt: i64, sequence: i64, kind: StreamRecordKind| { + let value = format!("{producer}{attempt}.{sequence}"); + StreamRecord { + topic: "out".to_string(), + kind: kind as i32, + producer_id: producer.to_string(), + attempt, + sequence, + body: (kind == StreamRecordKind::Data).then(|| Payload { + data: value.into_bytes(), + ..Default::default() + }), + ..Default::default() + } + }; + let data = StreamRecordKind::Data; + let close = StreamRecordKind::Close; + let log = [ + record("p", 1, 1, data), + record("q", 1, 1, data), + record("p", 1, 1, data), + record("p", 2, 1, data), + record("p", 1, 2, data), + record("q", 1, 2, data), + record("q", 1, 2, data), + record("c", 1, 0, close), + record("p", 2, 2, data), + record("d", 1, 0, close), + ]; + let mut producers = HashMap::new(); + for record in log { + let bytes = record.encode_to_vec(); + let attempt = producers.entry(record.producer_id.clone()).or_insert(0); + *attempt += 1; + store + .append(StoreAppendRequest { + chain: Some(chain.clone()), + topic: "out".to_string(), + producer_id: record.producer_id, + attempt: record.attempt, + sequence: *attempt, + digest: Sha256::digest(&bytes).to_vec(), + records: vec![bytes], + }) + .await + .unwrap(); + } + let target = ReadTarget { + chain: chain.clone(), + topic: "out".to_string(), + stream_hash: stream_hash("ns", "workflow", &stream.workflow_id, "out"), + }; + let whole = paged(&*store, &target, 1000).await; + assert_eq!(whole, ["p1.1", "q1.1", "p2.1", "p1.2", "q1.2"], "{label}"); + for max_records in [1, 2, 3] { + assert_eq!( + paged(&*store, &target, max_records).await, + whole, + "{label} {max_records}" + ); + } + } +}