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/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 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 + )); +} 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}" + ); + } + } +}