diff --git a/crates/streams/src/lib.rs b/crates/streams/src/lib.rs index f9ea07e..d1c4daf 100644 --- a/crates/streams/src/lib.rs +++ b/crates/streams/src/lib.rs @@ -29,6 +29,7 @@ pub use runtime::{ MEMORY_RUNTIME_NAME, MemoryStreamRuntime, REDIS_RUNTIME_NAME, RedisStreamRuntime, RuntimeOptions, StreamRuntime, }; +pub use service::{MAX_RECORDS, MAX_WAIT}; /// The protos that lang and the crate exchange. pub mod proto { diff --git a/crates/streams/src/metrics.rs b/crates/streams/src/metrics.rs index b894762..42824b4 100644 --- a/crates/streams/src/metrics.rs +++ b/crates/streams/src/metrics.rs @@ -1,7 +1,7 @@ //! The crate's metrics. They use the meter that lang gives, or a no-op meter. use temporalio_common::telemetry::metrics::{ - Counter, MetricAttributes, MetricParameters, TemporalMeter, + Counter, MetricAttributes, MetricParameters, TemporalMeter, UpDownCounter, }; #[derive(Clone)] @@ -9,6 +9,7 @@ pub(crate) struct Metrics { attributes: MetricAttributes, fenced: Counter, superseded: Counter, + blocking_reads_waiting: UpDownCounter, } fn parameters( @@ -37,6 +38,11 @@ impl Metrics { "Superseded flags that the Store set", "", )), + blocking_reads_waiting: meter.up_down_counter(parameters( + "stream_blocking_reads_waiting", + "Reads that wait on the Store", + "", + )), } } @@ -47,4 +53,9 @@ impl Metrics { pub(crate) fn superseded(&self) { self.superseded.add(1, &self.attributes); } + + pub(crate) fn read_waits(&self, waiting: bool) { + self.blocking_reads_waiting + .add(if waiting { 1 } else { -1 }, &self.attributes); + } } diff --git a/crates/streams/src/runtime.rs b/crates/streams/src/runtime.rs index 1b8621d..3defe2e 100644 --- a/crates/streams/src/runtime.rs +++ b/crates/streams/src/runtime.rs @@ -4,7 +4,10 @@ use crate::{ Reason, StreamError, StreamResult, memory::MemoryStore, metrics::Metrics, - proto::{AppendRecordsRequest, CloseStreamRequest, DescribeStreamRequest, RedisStreamConfig}, + proto::{ + AppendRecordsRequest, CloseStreamRequest, DescribeStreamRequest, ReadRecordsRequest, + RedisStreamConfig, + }, redis::{RedisStore, RedisStoreOptions}, service::StreamService, store::Store, @@ -173,6 +176,10 @@ impl StreamRuntime { .close_stream(decode::(input)?) .await? .encode_to_vec(), + "ReadRecords" => service + .read_records(decode::(input)?) + .await? + .encode_to_vec(), "DescribeStream" => service .describe_stream(decode::(input)?) .await? diff --git a/crates/streams/src/service.rs b/crates/streams/src/service.rs index c80c1df..b6a09e0 100644 --- a/crates/streams/src/service.rs +++ b/crates/streams/src/service.rs @@ -1,21 +1,39 @@ -//! The Stream service operations on the Store, for Activities, clients and the Local Activity that -//! runs each Workflow call. +//! The five Stream service operations on the Store, for Activities, clients and the Local +//! Activity that runs each Workflow call. use crate::{ Reason, StreamError, StreamResult, digest::batch_digest, + epoch::withdraws, ids::check_producer, ids::check_stream_id, metrics::Metrics, proto::{ AppendRecordsRequest, AppendRecordsResponse, CloseStreamRequest, CloseStreamResponse, CreateStreamRequest, CreateStreamResponse, DescribeStreamRequest, DescribeStreamResponse, - Producer, StreamProducerKind, StreamRef, StreamStatus, + Producer, ReadRecordsRequest, ReadRecordsResponse, StreamProducerKind, StreamReadStart, + StreamRecord, StreamRecordKind, StreamRef, StreamStatus, Supersession, }, record::{caller_status, close_record, data_records, supersession}, - store::{Batch, Capabilities, CreateOutcome, MAX_SEQUENCE, Owner, Store, Withdrawal}, + store::{ + Batch, Capabilities, CreateOutcome, FORMAT_VERSION, MAX_SEQUENCE, Owner, Store, + StreamState, Withdrawal, + }, }; +use prost::Message; use std::{sync::Arc, time::Duration}; +use tokio::time::Instant; + +/// The longest wait of one read, for new Records and for the create. +pub const MAX_WAIT: Duration = Duration::from_secs(60); +/// The maximum number of Records in one read. +pub const MAX_RECORDS: usize = 1000; +/// The maximum number of log entries that one page scans. +const MAX_SCAN: usize = 10 * MAX_RECORDS; +/// The number of log entries that a filtered page reads at a time, whatever its `max_records`. +const SCAN_CHUNK: usize = MAX_RECORDS; +/// How often a read looks again for a Stream that nobody created yet. +const CREATE_POLL: Duration = Duration::from_millis(100); /// The Stream service on one Store. pub(crate) struct StreamService { @@ -254,6 +272,367 @@ impl StreamService { retention: Some(proto_duration(state.retention)), }) } + + /// A long poll that returns the Records it found, the close Record, or nothing after `wait`. + pub(crate) async fn read_records( + &self, + request: ReadRecordsRequest, + ) -> StreamResult { + let id = stream_id(request.stream.as_ref())?.to_string(); + let namespace = request.namespace.as_str(); + if request.from_offset < 0 { + return Err(StreamError::new(Reason::Offset, "The offset is below zero")); + } + let max_bytes = max_bytes(request.max_bytes)?; + if request.last < 0 { + return Err(StreamError::invalid("last must be 0 or more")); + } + if request.last > 0 { + if request.wake_only || request.from_offset > 0 { + return Err(StreamError::invalid( + "A read of the last Records takes no from_offset and no wake_only", + )); + } + return self.read_last(request, max_bytes).await; + } + if request.wake_only || request.to_offset != 0 { + return Err(StreamError::new( + Reason::Unsupported, + "The Store reads no to_offset or wake_only, which the native backend serves", + )); + } + let max = match request.max_records { + n if n <= 0 => MAX_RECORDS, + n => (n as usize).min(MAX_RECORDS), + }; + let started = Instant::now(); + let wait = duration(request.wait).unwrap_or_default().min(MAX_WAIT); + let create_wait = duration(request.create_wait) + .unwrap_or_default() + .min(MAX_WAIT); + let deadline = started + wait; + + let mut state = self.created(namespace, &id, started + create_wait).await?; + if state.version > FORMAT_VERSION { + return Err(StreamError::new( + Reason::FormatTooNew, + format!("format {} is newer than {FORMAT_VERSION}", state.version), + )); + } + let from = match request.from_offset { + 0 if request.start == StreamReadStart::End as i32 => state.newest + 1, + from => from, + }; + if from > state.newest + 1 { + return Err(StreamError::new( + Reason::Offset, + format!( + "The offset {from} is past the newest offset + 1, {}", + state.newest + 1 + ), + )); + } + let topics = TopicFilter { + topics: request.topics, + prefixes: request.topic_prefixes, + }; + // Where the next scan starts. A scan that finds only other Topics advances it, so that the + // next wait doesn't wake for the same entries. + let mut cursor = from; + loop { + check_expiry(&state, cursor)?; + let (page, scanned) = self + .page(namespace, &id, &state, cursor, (max, max_bytes), &topics) + .await?; + if let Some(page) = page { + return Ok(page); + } + // A scan that stopped at its cap left entries, so the next scan starts with no wait. + let more = scanned > cursor && scanned <= state.newest; + cursor = scanned; + if deadline <= Instant::now() { + return Ok(ReadRecordsResponse { + next_offset: cursor, + ..Default::default() + }); + } + if more { + continue; + } + let after = (cursor - 1).max(state.newest); + self.metrics.read_waits(true); + let woke = self.store.wait_after(namespace, &id, after, deadline).await; + self.metrics.read_waits(false); + woke?; + state = self.store.read(namespace, &id, 0, 0).await?.0; + } + } + + /// The newest `last` Records that pass the filter, oldest first, and the close Record at the + /// end when the Stream is closed. A Record that a newer `Superseded` item withdraws doesn't + /// count. It never waits for new Records. + async fn read_last( + &self, + request: ReadRecordsRequest, + max_bytes: usize, + ) -> StreamResult { + let id = stream_id(request.stream.as_ref())?.to_string(); + let namespace = request.namespace.as_str(); + let count = (request.last as usize).min(MAX_RECORDS); + let create_wait = duration(request.create_wait) + .unwrap_or_default() + .min(MAX_WAIT); + let state = self + .created(namespace, &id, Instant::now() + create_wait) + .await?; + if state.version > FORMAT_VERSION { + return Err(StreamError::new( + Reason::FormatTooNew, + format!("format {} is newer than {FORMAT_VERSION}", state.version), + )); + } + check_expiry(&state, 0)?; + let topics = TopicFilter { + topics: request.topics, + prefixes: request.topic_prefixes, + }; + let retained = state.retained_from(); + let (mut picked, mut close) = (vec![], None); + let mut withdrawals: Vec = vec![]; + let (mut bytes, mut scanned) = (0, 0); + let mut before = state.newest; + 'scan: while before >= retained && picked.len() < count && scanned < MAX_SCAN { + let entries = self + .store + .read_back(namespace, &id, before, count.clamp(16, MAX_RECORDS)) + .await?; + if entries.is_empty() { + break; + } + scanned += entries.len(); + for entry in entries { + if entry.offset < retained { + break 'scan; + } + before = entry.offset - 1; + // An entry's `Superseded` item withdraws earlier Records, not the entry's own. + if let Some(record) = entry.record { + let record = decode(&record, entry.offset)?; + let size = record.encoded_len(); + // The close Record ends the page and doesn't count toward the byte cap. + if record.kind == StreamRecordKind::Close as i32 { + close = Some(record); + } else if topics.passes(&record.topic) + && !withdrawals.iter().any(|s| withdraws(s, &record)) + { + // A Record over the cap is still returned alone, so that the read always + // returns a Record. + if !picked.is_empty() && bytes + size > max_bytes { + break 'scan; + } + bytes += size; + picked.push(record); + if picked.len() == count { + break 'scan; + } + } + } + if let Some(supersession) = entry.supersession { + withdrawals.push( + Supersession::decode(supersession.as_slice()) + .map_err(|error| StreamError::storage(error.to_string()))?, + ); + } + } + } + picked.reverse(); + picked.extend(close); + Ok(ReadRecordsResponse { + records: picked, + next_offset: state.newest + 1, + done: state.closed_at.is_some(), + ..Default::default() + }) + } + + /// The state of a Stream, after waiting up to `until` for its create. + async fn created( + &self, + namespace: &str, + id: &str, + until: Instant, + ) -> StreamResult { + loop { + let (state, _) = self.store.read(namespace, id, 0, 0).await?; + if state.exists { + return Ok(state); + } + let now = Instant::now(); + if now >= until { + return Err(StreamError::new( + Reason::NotFound, + "Nobody created the Stream", + )); + } + tokio::time::sleep(CREATE_POLL.min(until - now)).await; + } + } + + /// The Records from `from`, or `None` when there is none yet, with the offset after the last + /// entry it scanned. A page that ends in the close Record is done. `(max, max_bytes)` caps + /// the items and their bytes. + async fn page( + &self, + namespace: &str, + id: &str, + state: &StreamState, + from: i64, + (max, max_bytes): (usize, usize), + topics: &TopicFilter, + ) -> StreamResult<(Option, i64)> { + // A read past the close returns the close Record again, so that every read ends with it. + if let Some(closed_at) = state.closed_at.filter(|at| from > *at) { + let (_, entries) = self.store.read(namespace, id, closed_at, 1).await?; + let mut records = vec![]; + for entry in entries { + if let Some(bytes) = entry.record { + records.push(decode(&bytes, entry.offset)?); + } + } + let page = ReadRecordsResponse { + records, + next_offset: from, + done: true, + ..Default::default() + }; + return Ok((Some(page), from)); + } + // A read from the start skips the Records that the Retention hides. A read from an offset + // fails on them, since the Reader asked for those Records. + let retained = state.retained_from(); + let mut start = if from > 0 { from } else { retained }; + let mut records = vec![]; + let mut next = from; + let (mut done, mut full) = (false, false); + let mut bytes = 0; + // A Topic filter can skip many entries, so one page scans a bounded number of them, and + // reads more than `max` at a time. + let chunk = if topics.filters() { SCAN_CHUNK } else { max }; + let mut scanned = 0; + while records.len() < max && !done && !full && scanned < MAX_SCAN { + let (chunk_state, entries) = self.store.read(namespace, id, start, chunk).await?; + // A trim after the state read leaves a gap before these entries. + check_expiry(&chunk_state, if from > 0 { start } else { 0 })?; + if entries.is_empty() { + break; + } + scanned += entries.len(); + for entry in entries { + if entry.offset < retained { + if from > 0 { + return Err(expired()); + } + next = entry.offset + 1; + start = next; + continue; + } + let mut items = vec![]; + if let Some(bytes) = entry.supersession { + let supersession = Supersession::decode(bytes.as_slice()) + .map_err(|error| StreamError::storage(error.to_string()))?; + items.push(StreamRecord { + kind: StreamRecordKind::Superseded as i32, + offset: entry.offset, + supersession: Some(supersession), + ..Default::default() + }); + } + if let Some(bytes) = entry.record { + let record = decode(&bytes, entry.offset)?; + done = record.kind == StreamRecordKind::Close as i32; + if done || topics.passes(&record.topic) { + items.push(record); + } + } + // An entry over the cap is still returned alone, so that the read always makes + // progress. The next read starts at the entry that didn't fit. + let size: usize = items.iter().map(Message::encoded_len).sum(); + if !records.is_empty() && bytes + size > max_bytes { + (done, full) = (false, true); + next = entry.offset; + break; + } + bytes += size; + records.extend(items); + next = entry.offset + 1; + start = next; + if done || records.len() >= max { + break; + } + } + } + if records.is_empty() { + return Ok((None, next)); + } + let page = ReadRecordsResponse { + records, + next_offset: next, + done, + ..Default::default() + }; + Ok((Some(page), next)) + } +} + +/// Which Topics a read returns. A `Superseded` item passes every filter, so the page adds it before +/// the filter runs. +pub(crate) struct TopicFilter { + topics: Vec, + prefixes: Vec, +} + +impl TopicFilter { + fn filters(&self) -> bool { + !self.topics.is_empty() || !self.prefixes.is_empty() + } + + fn passes(&self, topic: &str) -> bool { + (self.topics.is_empty() && self.prefixes.is_empty()) + || self.topics.iter().any(|t| t == topic) + || self.prefixes.iter().any(|p| topic.starts_with(p.as_str())) + } +} + +fn decode(bytes: &[u8], offset: i64) -> StreamResult { + let mut record = StreamRecord::decode(bytes) + .map_err(|error| StreamError::storage(format!("A stored Record is bad: {error}")))?; + record.offset = offset; + Ok(record) +} + +/// Fails a read whose Records the Retention or the expiry removed, so that the Reader always sees +/// the gap. A read from the start of the Stream has no gap. +/// +/// A Tombstone fails every read. An open Stream that expired gets no more Records, so the Reader +/// would wait forever. A closed one lost its log and its close Record. +fn check_expiry(state: &StreamState, from: i64) -> StreamResult<()> { + if state.expired() || (from > 0 && state.trimmed >= from) { + return Err(expired()); + } + Ok(()) +} + +/// The byte cap of a page, from a request's `max_bytes`. Zero is no cap. +fn max_bytes(value: i64) -> StreamResult { + match value { + 0 => Ok(usize::MAX), + n if n < 0 => Err(StreamError::invalid("max_bytes must be 0 or more")), + n => Ok(usize::try_from(n).unwrap_or(usize::MAX)), + } +} + +fn expired() -> StreamError { + StreamError::new(Reason::Expired, "The Records at the offset expired") } /// A Store answer that may have withdrawn Records. @@ -266,3 +645,23 @@ impl Flagged for crate::redis::Appended { self.flagged } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn a_topic_passes_by_exact_name_or_by_prefix() { + let filter = |topics: &[&str], prefixes: &[&str]| TopicFilter { + topics: topics.iter().map(|t| t.to_string()).collect(), + prefixes: prefixes.iter().map(|p| p.to_string()).collect(), + }; + assert!(filter(&[], &[]).passes("any")); + let both = filter(&["turns"], &["tool/"]); + assert!(both.passes("turns")); + assert!(both.passes("tool/search")); + assert!(!both.passes("turns/1")); + assert!(!both.passes("tool")); + assert!(filter(&[], &[""]).passes("x")); + } +} diff --git a/crates/streams/tests/scans.rs b/crates/streams/tests/scans.rs new file mode 100644 index 0000000..989d3b4 --- /dev/null +++ b/crates/streams/tests/scans.rs @@ -0,0 +1,238 @@ +//! How a read scans the log, seen through a memory Store that counts the log reads and can +//! trim under a read. + +use prost::Message; +use std::{ + sync::{ + Arc, + atomic::{AtomicBool, AtomicUsize, Ordering}, + }, + time::Duration, +}; +use temporalio_common::protos::temporal::api::common::v1::Payload; +use temporalio_streams::{ + MemoryStreamRuntime, StreamError, StreamResult, StreamRuntime, + ids::client_producer_id, + memory::MemoryStore, + proto::{ + AppendRecord, AppendRecordsRequest, AppendRecordsResponse, CreateStreamRequest, + CreateStreamResponse, Producer, ProducerEpoch, ReadRecordsRequest, ReadRecordsResponse, + StreamErrorDetails, StreamProducerKind, StreamRef, + }, + store::{Appended, Batch, Capabilities, Created, Entry, Owner, Store, StreamState, Withdrawal}, +}; +use tokio::time::Instant; + +const NAMESPACE: &str = "ns"; +const STREAM: &str = "s"; + +/// The memory Store, with a count of the reads that ask for log entries. Once armed, its next +/// such read finds its first entry trimmed, as a trim just before that read leaves it. +struct Counted { + inner: MemoryStore, + log_reads: AtomicUsize, + trim_next: AtomicBool, +} + +#[async_trait::async_trait] +impl Store for Counted { + fn capabilities(&self) -> Capabilities { + self.inner.capabilities() + } + + fn default_retention(&self) -> Duration { + self.inner.default_retention() + } + + async fn create( + &self, + namespace: &str, + stream: &StreamRef, + retention: Option, + owner: Owner, + withdrawal: &Withdrawal<'_>, + ) -> StreamResult { + self.inner + .create(namespace, stream, retention, owner, withdrawal) + .await + } + + async fn append( + &self, + namespace: &str, + stream: &StreamRef, + batch: &Batch<'_>, + ) -> StreamResult { + self.inner.append(namespace, stream, batch).await + } + + async fn read( + &self, + namespace: &str, + stream_id: &str, + from: i64, + count: usize, + ) -> StreamResult<(StreamState, Vec)> { + if count > 0 { + self.log_reads.fetch_add(1, Ordering::Relaxed); + } + let (mut state, mut entries) = self.inner.read(namespace, stream_id, from, count).await?; + if count > 0 && !entries.is_empty() && self.trim_next.swap(false, Ordering::Relaxed) { + state.trimmed = entries.remove(0).offset; + } + Ok((state, entries)) + } + + async fn wait_after( + &self, + namespace: &str, + stream_id: &str, + after: i64, + deadline: Instant, + ) -> StreamResult { + self.inner + .wait_after(namespace, stream_id, after, deadline) + .await + } +} + +fn counted() -> (Arc, Arc) { + let store = Arc::new(Counted { + inner: MemoryStore::new(Duration::from_secs(3600), Duration::from_secs(7200)), + log_reads: AtomicUsize::new(0), + trim_next: AtomicBool::new(false), + }); + let ext = MemoryStreamRuntime::with_store(store.clone(), None); + (store, ext) +} + +async fn call( + ext: &StreamRuntime, + method: &str, + request: Req, +) -> Resp { + try_call(ext, method, request).await.unwrap() +} + +async fn try_call( + ext: &StreamRuntime, + method: &str, + request: Req, +) -> Result { + let bytes = ext.call(method, &request.encode_to_vec()).await?; + Ok(Resp::decode(bytes.as_slice()).unwrap()) +} + +async fn create(ext: &StreamRuntime) { + let _: CreateStreamResponse = call( + ext, + "CreateStream", + CreateStreamRequest { + namespace: NAMESPACE.to_string(), + stream_id: STREAM.to_string(), + ..Default::default() + }, + ) + .await; +} + +fn client() -> Producer { + Producer { + kind: StreamProducerKind::Client as i32, + id: client_producer_id("me", "p"), + epoch: Some(ProducerEpoch { + attempt: 1, + ..Default::default() + }), + run_id: String::new(), + } +} + +fn stream() -> Option { + Some(StreamRef { + id: STREAM.to_string(), + ..Default::default() + }) +} + +async fn append( + ext: &StreamRuntime, + sequence: i64, + topic: &str, + count: usize, +) -> AppendRecordsResponse { + let records = (0..count) + .map(|i| AppendRecord { + topic: topic.to_string(), + body: Some(Payload { + data: i.to_string().into_bytes(), + ..Default::default() + }), + }) + .collect(); + call( + ext, + "AppendRecords", + AppendRecordsRequest { + namespace: NAMESPACE.to_string(), + stream: stream(), + producer: Some(client()), + sequence, + records, + ..Default::default() + }, + ) + .await +} + +/// A filtered page reads the log in chunks of its own, so `max_records: 1` doesn't make one +/// Store call per entry. +#[tokio::test] +async fn a_filtered_page_reads_the_log_in_chunks_whatever_its_max() { + let (store, ext) = counted(); + create(&ext).await; + for sequence in 1..=5 { + append(&ext, sequence, "a", 1000).await; + } + append(&ext, 6, "b", 1).await; + store.log_reads.store(0, Ordering::Relaxed); + let read: ReadRecordsResponse = call( + &ext, + "ReadRecords", + ReadRecordsRequest { + namespace: NAMESPACE.to_string(), + stream: stream(), + topics: vec!["b".to_string()], + max_records: 1, + ..Default::default() + }, + ) + .await; + assert_eq!(read.records.len(), 1); + let reads = store.log_reads.load(Ordering::Relaxed); + assert!(reads <= 10, "{reads} log reads"); +} + +/// A trim between a read's state and its log read fails a read from an offset as expired, so that +/// the Reader always sees the gap. A read from the start skips the trimmed Records. +#[tokio::test] +async fn a_trim_under_a_read_from_an_offset_fails_it_as_expired() { + let (store, ext) = counted(); + create(&ext).await; + let first = append(&ext, 1, "a", 3).await; + let read = |from_offset| ReadRecordsRequest { + namespace: NAMESPACE.to_string(), + stream: stream(), + from_offset, + ..Default::default() + }; + store.trim_next.store(true, Ordering::Relaxed); + let error = try_call::<_, ReadRecordsResponse>(&ext, "ReadRecords", read(first.first_offset)) + .await + .unwrap_err(); + let details = StreamErrorDetails::decode(error.details().as_slice()).unwrap(); + assert_eq!(details.reason, "stream_expired"); + store.trim_next.store(true, Ordering::Relaxed); + let from_start: ReadRecordsResponse = call(&ext, "ReadRecords", read(0)).await; + assert_eq!(from_start.records.len(), 2); +} diff --git a/crates/streams/tests/service.rs b/crates/streams/tests/service.rs index 6c6084c..cb384bd 100644 --- a/crates/streams/tests/service.rs +++ b/crates/streams/tests/service.rs @@ -1,20 +1,26 @@ -//! The Stream service operations through the runtime's byte call, as lang sends them, on each -//! backend. +//! The five operations through the runtime's byte call, as lang sends them, on each backend. mod support; use support::Needs; use prost::Message; -use std::{sync::Arc, time::Duration}; -use temporalio_common::protos::temporal::api::common::v1::Payload; +use prost_types::Timestamp; +use std::{ + sync::Arc, + time::{Duration, Instant}, +}; +use temporalio_common::protos::temporal::api::{common::v1::Payload, failure::v1::Failure}; use temporalio_streams::{ - RedisStreamRuntime, StreamError, - ids::client_producer_id, + MemoryStreamRuntime, RedisStreamRuntime, StreamError, + epoch::withdraws, + ids::{activity_producer_id, client_producer_id, named_client_producer_id}, proto::{ - AppendRecord, AppendRecordsRequest, AppendRecordsResponse, CreateStreamRequest, - CreateStreamResponse, Producer, ProducerEpoch, StreamErrorDetails, StreamProducerKind, - StreamRef, StreamStatus, + AppendRecord, AppendRecordsRequest, AppendRecordsResponse, CloseStreamRequest, + CloseStreamResponse, CreateStreamRequest, CreateStreamResponse, DescribeStreamRequest, + DescribeStreamResponse, Producer, ProducerEpoch, ReadRecordsRequest, ReadRecordsResponse, + RedisStreamConfig, StreamCloseStatus, StreamErrorDetails, StreamProducerKind, + StreamReadStart, StreamRecordKind, StreamRef, StreamStatus, }, }; use tonic::Code; @@ -23,6 +29,8 @@ struct Case { ext: Arc, namespace: String, stream: String, + /// What the backend gives, for a step that needs more than the case asked for. + gives: Needs, } static NEXT: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); @@ -47,6 +55,7 @@ async fn cases_for(needs: Needs) -> Vec { ext: runtime.ext, namespace: runtime.namespace.unwrap_or(format!("ns-{unique}")), stream: format!("s-{unique}"), + gives: runtime.gives, } }) .collect() @@ -79,6 +88,37 @@ fn client(attempt: i32) -> Producer { } } +/// A client Producer that a test names, at `attempt`. Every backend accepts it as sent. +fn named(name: &str, attempt: i32) -> Producer { + Producer { + kind: StreamProducerKind::Client as i32, + id: named_client_producer_id(name), + epoch: Some(ProducerEpoch { + attempt, + ..Default::default() + }), + run_id: String::new(), + } +} + +/// An Activity Producer of a made-up run, at `attempt`. A higher attempt withdraws the lower ones' +/// Records. The server requires the Activity's task token, so only Redis and memory accept it. +fn activity(attempt: i32, scheduled: i64) -> Producer { + Producer { + kind: StreamProducerKind::Activity as i32, + id: activity_producer_id("r1", "a", "s"), + epoch: Some(ProducerEpoch { + attempt_scheduled_time: Some(Timestamp { + seconds: scheduled, + nanos: 0, + }), + attempt, + ..Default::default() + }), + run_id: "r1".to_string(), + } +} + fn record(topic: &str, text: &str) -> AppendRecord { AppendRecord { topic: topic.to_string(), @@ -142,6 +182,66 @@ impl Case { ) .await } + + async fn read(&self, request: ReadRecordsRequest) -> Result { + call( + &self.ext, + "ReadRecords", + ReadRecordsRequest { + namespace: self.namespace.clone(), + stream: self.stream(), + ..request + }, + ) + .await + } + + async fn describe(&self) -> Result { + call( + &self.ext, + "DescribeStream", + DescribeStreamRequest { + namespace: self.namespace.clone(), + stream: self.stream(), + }, + ) + .await + } + + async fn close(&self, producer: Producer) -> Result { + call( + &self.ext, + "CloseStream", + CloseStreamRequest { + namespace: self.namespace.clone(), + stream: self.stream(), + producer: Some(producer), + result: Some(Payload { + data: b"done".to_vec(), + ..Default::default() + }), + ..Default::default() + }, + ) + .await + } +} + +fn texts(response: &ReadRecordsResponse) -> Vec { + response + .records + .iter() + .map( + |record| match StreamRecordKind::try_from(record.kind).unwrap() { + StreamRecordKind::Data => { + String::from_utf8(record.body.clone().unwrap().data).unwrap() + } + StreamRecordKind::Superseded => "".to_string(), + StreamRecordKind::Close => "".to_string(), + StreamRecordKind::Unspecified => "?".to_string(), + }, + ) + .collect() } /// A create is idempotent for one Retention, and another Retention fails with @@ -187,6 +287,265 @@ async fn a_producer_must_match_its_kind_and_not_be_a_workflow() { } } +/// A read is inclusive from its offset, a read from the end gives only new Records, and an +/// offset past the newest + 1 fails with `stream_offset`. +#[tokio::test] +async fn a_read_is_inclusive_from_its_offset_and_get_offset_reads_only_new_records() { + for case in cases_for(Needs::CALLS).await { + case.create(None).await.unwrap(); + let first = case + .append(client(1), 1, vec![record("a", "1"), record("b", "2")]) + .await + .unwrap(); + let all = case.read(ReadRecordsRequest::default()).await.unwrap(); + assert_eq!(texts(&all), ["1", "2"]); + assert_eq!(all.records[0].offset, first.first_offset); + assert_eq!(all.next_offset, first.last_offset + 1); + let from = case + .read(ReadRecordsRequest { + from_offset: first.last_offset, + ..Default::default() + }) + .await + .unwrap(); + assert_eq!(texts(&from), ["2"]); + let next = case.describe().await.unwrap().last_offset + 1; + let none = case + .read(ReadRecordsRequest { + from_offset: next, + ..Default::default() + }) + .await + .unwrap(); + assert!(none.records.is_empty()); + assert_eq!(none.next_offset, next); + let end = case + .read(ReadRecordsRequest { + start: StreamReadStart::End as i32, + ..Default::default() + }) + .await + .unwrap(); + assert_eq!(end.next_offset, next); + let error = case + .read(ReadRecordsRequest { + from_offset: next + 1, + ..Default::default() + }) + .await + .unwrap_err(); + assert_eq!( + (error.code(), reason(&error).as_str()), + (Code::InvalidArgument, "stream_offset") + ); + } +} + +/// A Topic filter keeps the `Superseded` items, which pass every filter. +#[tokio::test] +async fn a_topic_filter_keeps_superseded_items() { + for case in cases_for(Needs::UNCHECKED).await { + case.create(None).await.unwrap(); + case.append(activity(1, 10), 1, vec![record("a", "old")]) + .await + .unwrap(); + case.append( + activity(2, 20), + 1, + vec![record("b", "b1"), record("a", "a1")], + ) + .await + .unwrap(); + let all = case.read(ReadRecordsRequest::default()).await.unwrap(); + assert_eq!(texts(&all), ["old", "", "b1", "a1"]); + // The item is just before the batch's first Record. Redis stores both in one entry with one + // offset, and the server gives the item its own offset just below. + let superseded = &all.records[1]; + assert!(all.records[0].offset < superseded.offset); + assert!(superseded.offset <= all.records[2].offset); + let supersession = superseded.supersession.as_ref().unwrap(); + assert_eq!(supersession.producer_id, activity(2, 20).id); + let only_a = case + .read(ReadRecordsRequest { + topics: vec!["a".to_string()], + ..Default::default() + }) + .await + .unwrap(); + assert_eq!(texts(&only_a), ["old", "", "a1"]); + } +} + +/// A read waits for the Stream's create only as long as `create_wait`. +#[tokio::test] +async fn a_read_waits_for_the_create_only_as_long_as_create_wait() { + for case in cases_for(Needs::CALLS).await { + let error = case.read(ReadRecordsRequest::default()).await.unwrap_err(); + assert_eq!( + (error.code(), reason(&error).as_str()), + (Code::NotFound, "stream_not_found") + ); + let started = Instant::now(); + let error = case + .read(ReadRecordsRequest { + create_wait: millis(200), + ..Default::default() + }) + .await + .unwrap_err(); + assert_eq!(error.code(), Code::NotFound); + assert!(started.elapsed() >= Duration::from_millis(200)); + let ext = case.ext.clone(); + let (namespace, stream) = (case.namespace.clone(), case.stream.clone()); + let creator = tokio::spawn(async move { + tokio::time::sleep(Duration::from_millis(150)).await; + let request = CreateStreamRequest { + namespace, + stream_id: stream, + ..Default::default() + }; + call::<_, CreateStreamResponse>(&ext, "CreateStream", request).await + }); + let read = case + .read(ReadRecordsRequest { + create_wait: millis(3000), + ..Default::default() + }) + .await + .unwrap(); + assert!(read.records.is_empty()); + creator.await.unwrap().unwrap(); + } +} + +/// A quiet read ends empty after its wait, and a long poll wakes on an append. +#[tokio::test] +async fn a_quiet_read_ends_empty_after_its_wait_and_a_long_poll_wakes_on_append() { + for case in cases_for(Needs::CALLS).await { + case.create(None).await.unwrap(); + let first = case + .append(client(1), 1, vec![record("", "1")]) + .await + .unwrap(); + let from = first.last_offset + 1; + let started = Instant::now(); + let quiet = case + .read(ReadRecordsRequest { + from_offset: from, + wait: millis(200), + ..Default::default() + }) + .await + .unwrap(); + assert!(started.elapsed() >= Duration::from_millis(200)); + assert!(started.elapsed() < Duration::from_secs(2)); + assert!(quiet.records.is_empty()); + assert_eq!(quiet.next_offset, from); + assert!(!quiet.done); + let ext = case.ext.clone(); + let (namespace, stream) = (case.namespace.clone(), case.stream.clone()); + let writer = tokio::spawn(async move { + tokio::time::sleep(Duration::from_millis(150)).await; + let request = AppendRecordsRequest { + namespace, + stream: Some(StreamRef { + id: stream, + ..Default::default() + }), + producer: Some(client(1)), + sequence: 2, + records: vec![record("", "2")], + ..Default::default() + }; + call::<_, AppendRecordsResponse>(&ext, "AppendRecords", request).await + }); + let woke = case + .read(ReadRecordsRequest { + from_offset: from, + wait: millis(5000), + ..Default::default() + }) + .await + .unwrap(); + assert_eq!(texts(&woke), ["2"]); + writer.await.unwrap().unwrap(); + } +} + +/// A close ends every read with its close Record and refuses later appends, and a create finds +/// the Stream closed. +#[tokio::test] +async fn a_close_ends_every_read_with_its_record() { + for case in cases_for(Needs::CALLS).await { + case.create(None).await.unwrap(); + case.append(client(1), 1, vec![record("", "1")]) + .await + .unwrap(); + case.close(client(1)).await.unwrap(); + let read = case.read(ReadRecordsRequest::default()).await.unwrap(); + assert_eq!(texts(&read), ["1", ""]); + assert!(read.done); + assert_eq!(read.records[1].close_result.as_ref().unwrap().data, b"done"); + let again = case + .read(ReadRecordsRequest { + from_offset: read.next_offset, + ..Default::default() + }) + .await + .unwrap(); + assert_eq!(texts(&again), [""]); + assert!(again.done); + assert_eq!( + case.describe().await.unwrap().status, + StreamStatus::Closed as i32 + ); + let error = case + .append(client(1), 2, vec![record("", "2")]) + .await + .unwrap_err(); + assert_eq!( + (error.code(), reason(&error).as_str()), + (Code::FailedPrecondition, "stream_closed") + ); + let created = case.create(None).await.unwrap(); + assert_eq!(created.status, StreamStatus::Closed as i32); + } +} + +/// A read from an offset that the Retention trimmed fails with `stream_expired`. +#[tokio::test] +async fn a_read_from_a_trimmed_offset_fails_as_expired() { + for case in cases_for(Needs::SHORT_TIMES).await { + case.create(millis(200)).await.unwrap(); + // Each write extends the Stream's expiry by one Retention, so only the old Record expires. + let old = case + .append(client(1), 1, vec![record("", "old")]) + .await + .unwrap(); + tokio::time::sleep(Duration::from_millis(120)).await; + case.append(client(1), 2, vec![record("", "mid")]) + .await + .unwrap(); + tokio::time::sleep(Duration::from_millis(120)).await; + case.append(client(1), 3, vec![record("", "new")]) + .await + .unwrap(); + let error = case + .read(ReadRecordsRequest { + from_offset: old.first_offset, + ..Default::default() + }) + .await + .unwrap_err(); + assert_eq!( + (error.code(), reason(&error).as_str()), + (Code::NotFound, "stream_expired") + ); + let from_start = case.read(ReadRecordsRequest::default()).await.unwrap(); + assert!(!texts(&from_start).contains(&"old".to_string())); + } +} + /// An unknown method fails with UNIMPLEMENTED. #[tokio::test] async fn an_unknown_method_is_unimplemented() { @@ -195,3 +554,458 @@ async fn an_unknown_method_is_unimplemented() { assert_eq!(error.code(), Code::Unimplemented); } } + +/// A read of an open Stream that expired fails at once instead of waiting. +#[tokio::test] +async fn a_read_of_an_expired_open_stream_fails_instead_of_waiting() { + for case in cases_for(Needs::SHORT_TIMES).await { + case.create(millis(300)).await.unwrap(); + case.append(client(1), 1, vec![record("", "1")]) + .await + .unwrap(); + tokio::time::sleep(Duration::from_millis(400)).await; + for from_offset in [0, 1] { + let started = Instant::now(); + let error = case + .read(ReadRecordsRequest { + from_offset, + wait: millis(5000), + ..Default::default() + }) + .await + .unwrap_err(); + assert_eq!(reason(&error), "stream_expired"); + assert!(started.elapsed() < Duration::from_secs(2)); + } + let error = case.append(client(1), 0, vec![]).await.unwrap_err(); + assert_eq!(reason(&error), "stream_invalid"); + } +} + +/// A read of a closed Stream whose Retention passed fails as expired, from any offset, instead +/// of waiting out its wait. A last read fails the same way. +#[tokio::test] +async fn a_read_of_an_expired_closed_stream_fails_instead_of_waiting() { + for case in cases_for(Needs::SHORT_TIMES).await { + case.create(millis(300)).await.unwrap(); + let first = case + .append(client(1), 1, vec![record("", "1")]) + .await + .unwrap(); + case.close(client(1)).await.unwrap(); + tokio::time::sleep(Duration::from_millis(500)).await; + let describe = case.describe().await.unwrap(); + assert_eq!(describe.status, StreamStatus::Closed as i32); + for request in [ + ReadRecordsRequest::default(), + ReadRecordsRequest { + from_offset: first.first_offset, + ..Default::default() + }, + ReadRecordsRequest { + from_offset: describe.last_offset + 1, + ..Default::default() + }, + ] { + let started = Instant::now(); + let error = case + .read(ReadRecordsRequest { + wait: millis(1500), + ..request + }) + .await + .unwrap_err(); + assert_eq!(reason(&error), "stream_expired"); + assert!(started.elapsed() < Duration::from_secs(1)); + } + let last = case + .read(ReadRecordsRequest { + last: 5, + ..Default::default() + }) + .await + .unwrap_err(); + assert_eq!(reason(&last), "stream_expired"); + } +} + +/// A filtered read continues past the scan cap with no wait while entries are left, so a match +/// after 10,000 entries of other Topics returns at once. A read with no wait returns an empty page +/// with an advanced `next_offset`, and the next read from there finds the match. +#[tokio::test] +async fn a_filtered_read_scans_past_the_cap_without_waiting() { + for case in cases_for(Needs::CALLS).await { + case.create(None).await.unwrap(); + for sequence in 1..=11 { + let batch = (0..1000).map(|i| record("a", &i.to_string())).collect(); + case.append(client(1), sequence, batch).await.unwrap(); + } + case.append(client(1), 12, vec![record("b", "match")]) + .await + .unwrap(); + let filtered = |wait| ReadRecordsRequest { + topics: vec!["b".to_string()], + wait, + ..Default::default() + }; + let started = Instant::now(); + let read = case.read(filtered(millis(10_000))).await.unwrap(); + assert_eq!(texts(&read), ["match"]); + assert!(started.elapsed() < Duration::from_secs(5)); + let first = case.read(filtered(None)).await.unwrap(); + assert!(first.records.is_empty() && !first.done); + assert!(first.next_offset > 0); + let next = case + .read(ReadRecordsRequest { + from_offset: first.next_offset, + ..filtered(None) + }) + .await + .unwrap(); + assert_eq!(texts(&next), ["match"]); + } +} + +/// A prefix filter reads the Topics under it and keeps the `Superseded` items. +#[tokio::test] +async fn a_prefix_filter_reads_the_topics_under_it_and_keeps_superseded_items() { + for case in cases_for(Needs::UNCHECKED).await { + case.create(None).await.unwrap(); + case.append(activity(1, 10), 1, vec![record("tool/a", "old")]) + .await + .unwrap(); + case.append( + activity(2, 20), + 1, + vec![ + record("turns", "t1"), + record("tool/a", "a1"), + record("tools", "s1"), + record("tool/b", "b1"), + ], + ) + .await + .unwrap(); + let read = case + .read(ReadRecordsRequest { + topics: vec!["turns".to_string()], + topic_prefixes: vec!["tool/".to_string()], + ..Default::default() + }) + .await + .unwrap(); + assert_eq!(texts(&read), ["old", "", "t1", "a1", "b1"]); + // A filter that matches nothing scans the log and advances the offset. + let none = case + .read(ReadRecordsRequest { + topic_prefixes: vec!["nothing/".to_string()], + ..Default::default() + }) + .await + .unwrap(); + assert_eq!(texts(&none), [""]); + } +} + +/// A ref with an owner and one without reach the same Stream. +#[tokio::test] +async fn a_ref_with_an_owner_and_one_without_reach_the_same_stream() { + for case in cases_for(Needs::ID_KEYS).await { + case.create(None).await.unwrap(); + let owned = Some(StreamRef { + id: case.stream.clone(), + owner: support::workflow_execution("agent-1", "run-1"), + ..Default::default() + }); + // An owner that doesn't match the id changes nothing either. + let other = Some(StreamRef { + id: case.stream.clone(), + owner: support::workflow_execution("someone-else", ""), + ..Default::default() + }); + for (sequence, stream) in [(1, owned.clone()), (2, case.stream()), (3, other)] { + let request = AppendRecordsRequest { + namespace: case.namespace.clone(), + stream, + producer: Some(client(1)), + sequence, + records: vec![record("", &sequence.to_string())], + ..Default::default() + }; + let _: AppendRecordsResponse = call(&case.ext, "AppendRecords", request).await.unwrap(); + } + let read = |stream: Option| ReadRecordsRequest { + namespace: case.namespace.clone(), + stream, + ..Default::default() + }; + let plain: ReadRecordsResponse = call(&case.ext, "ReadRecords", read(case.stream())) + .await + .unwrap(); + let with_owner: ReadRecordsResponse = + call(&case.ext, "ReadRecords", read(owned)).await.unwrap(); + assert_eq!(texts(&plain), ["1", "2", "3"]); + assert_eq!(plain, with_owner); + } +} + +/// A create answer names the configured backend, and the Store never reads it. +#[tokio::test] +async fn a_create_answer_names_the_backend_and_the_store_never_reads_it() { + let mut configured = + support::stream_runtimes(|config| config.backend = "redis-a".to_string(), false); + let config = RedisStreamConfig { + backend: "redis-a".to_string(), + ..Default::default() + }; + configured.push(MemoryStreamRuntime::from_config_bytes(&config.encode_to_vec(), None).unwrap()); + for ext in configured { + let unique = unique(); + let case = Case { + ext, + namespace: format!("ns-{unique}"), + stream: format!("s-{unique}"), + gives: Needs::ALL, + }; + let created = case.create(None).await.unwrap(); + let stream = created.stream.unwrap(); + assert_eq!(stream.id, case.stream); + assert_eq!(stream.backend, "redis-a"); + // A ref with any backend, or none, reaches the same Stream. + let named = |backend: &str| { + Some(StreamRef { + id: case.stream.clone(), + backend: backend.to_string(), + ..Default::default() + }) + }; + for (sequence, stream) in [(1, named("redis-a")), (2, named("")), (3, named("other"))] { + let request = AppendRecordsRequest { + namespace: case.namespace.clone(), + stream, + producer: Some(client(1)), + sequence, + records: vec![record("", &sequence.to_string())], + ..Default::default() + }; + let _: AppendRecordsResponse = call(&case.ext, "AppendRecords", request).await.unwrap(); + } + let read = ReadRecordsRequest { + namespace: case.namespace.clone(), + stream: named("other"), + ..Default::default() + }; + let read: ReadRecordsResponse = call(&case.ext, "ReadRecords", read).await.unwrap(); + assert_eq!(texts(&read), ["1", "2", "3"]); + } + // With no backend in the config, the ref has none, and lang reads it as the default. + for case in cases_for(Needs::OWN_CONFIG).await { + let created = case.create(None).await.unwrap(); + assert_eq!(created.stream.unwrap().backend, ""); + } +} + +fn workflow_owner(run: &str) -> Producer { + Producer { + kind: StreamProducerKind::Workflow as i32, + id: temporalio_streams::ids::workflow_producer_id(run), + run_id: run.to_string(), + ..Default::default() + } +} + +/// The user Local Activity `call-1` of `run`, dispatched by the Workflow Task `[started]`. A rerun +/// in a later task keeps the Producer id and has a newer dispatching task. +fn user_local_activity(run: &str, started: i64, seconds: i64) -> Producer { + Producer { + kind: StreamProducerKind::LocalActivity as i32, + id: temporalio_streams::ids::local_activity_producer_id(run, "call-1"), + epoch: Some(ProducerEpoch { + task_started_event_id: started, + task_attempt: 1, + task_started_time: Some(Timestamp { seconds, nanos: 0 }), + attempt: 1, + ..Default::default() + }), + run_id: run.to_string(), + } +} + +/// A Local Activity rerun in a later task supersedes its first attempt. +#[tokio::test] +async fn a_rerun_local_activity_supersedes_its_first_attempt() { + for case in cases_for(Needs::UNCHECKED).await { + let run = format!("{}-run", case.stream); + let _: CreateStreamResponse = call( + &case.ext, + "CreateStream", + CreateStreamRequest { + namespace: case.namespace.clone(), + stream_id: case.stream.clone(), + owner: Some(workflow_owner(&run)), + ..Default::default() + }, + ) + .await + .unwrap(); + let write = |producer: Producer, text: &str| AppendRecordsRequest { + namespace: case.namespace.clone(), + stream: case.stream(), + producer: Some(producer), + sequence: 1, + records: vec![record("", text)], + ..Default::default() + }; + // Task [3] runs the Local Activity, which writes, and then the task fails. + let _: AppendRecordsResponse = call( + &case.ext, + "AppendRecords", + write(user_local_activity(&run, 3, 100), "first"), + ) + .await + .unwrap(); + // The next task [7] runs the same call again, and it writes again. + let _: AppendRecordsResponse = call( + &case.ext, + "AppendRecords", + write(user_local_activity(&run, 7, 200), "second"), + ) + .await + .unwrap(); + // A late write of the failed attempt is fenced. + let late = call::<_, AppendRecordsResponse>( + &case.ext, + "AppendRecords", + AppendRecordsRequest { + sequence: 2, + ..write(user_local_activity(&run, 3, 100), "late") + }, + ) + .await + .unwrap_err(); + assert_eq!(reason(&late), "stream_fenced"); + // A Reader sees the first attempt, then the Superseded item that withdraws it, then the + // second attempt. After it applies the item, only the second attempt is visible. + let read = case.read(ReadRecordsRequest::default()).await.unwrap(); + assert_eq!(texts(&read), ["first", "", "second"]); + let superseded = read.records[1].supersession.as_ref().unwrap(); + assert!(withdraws(superseded, &read.records[0])); + assert!(!withdraws(superseded, &read.records[2])); + } +} + +/// A close Record gives its status, and a caller can't close as terminated or timed out. +#[tokio::test] +async fn a_close_record_gives_its_status_and_a_caller_cant_terminate() { + for case in cases_for(Needs::CALLS).await { + case.create(None).await.unwrap(); + let close = |status: StreamCloseStatus, failure: Option| CloseStreamRequest { + namespace: case.namespace.clone(), + stream: case.stream(), + producer: Some(client(1)), + failure, + status: status as i32, + ..Default::default() + }; + for status in [StreamCloseStatus::Terminated, StreamCloseStatus::TimedOut] { + let error = + call::<_, CloseStreamResponse>(&case.ext, "CloseStream", close(status, None)) + .await + .unwrap_err(); + assert_eq!( + (error.code(), reason(&error).as_str()), + (Code::InvalidArgument, "stream_invalid") + ); + } + let failure = Failure { + message: "boom".to_string(), + ..Default::default() + }; + let failed = close(StreamCloseStatus::Unspecified, Some(failure)); + call::<_, CloseStreamResponse>(&case.ext, "CloseStream", failed) + .await + .unwrap(); + let read = case.read(ReadRecordsRequest::default()).await.unwrap(); + let record = read.records.last().unwrap(); + assert_eq!(record.close_status, StreamCloseStatus::Failed as i32); + assert_eq!(record.close_failure.as_ref().unwrap().message, "boom"); + } +} + +/// A last read gives the newest Records oldest first, skips withdrawn ones, and ends with the +/// close Record of a closed Stream. +#[tokio::test] +async fn a_last_read_gives_the_newest_records_oldest_first_without_withdrawn_ones() { + for case in cases_for(Needs::CALLS).await { + case.create(None).await.unwrap(); + let last = |n: i32, topics: &[&str]| ReadRecordsRequest { + last: n, + topics: topics.iter().map(|t| t.to_string()).collect(), + ..Default::default() + }; + assert!(case.read(last(5, &[])).await.unwrap().records.is_empty()); + let records = vec![record("t", "c1"), record("u", "c2"), record("t", "c3")]; + case.append(client(1), 1, records).await.unwrap(); + if case.gives.gives(Needs::UNCHECKED) { + case.append(activity(1, 10), 1, vec![record("t", "a1")]) + .await + .unwrap(); + // The retry's first batch withdraws a1, so a1 doesn't count. + case.append(activity(2, 20), 1, vec![record("t", "a2")]) + .await + .unwrap(); + } else { + case.append(named("w", 1), 1, vec![record("t", "a2")]) + .await + .unwrap(); + } + assert_eq!( + texts(&case.read(last(3, &[])).await.unwrap()), + ["c2", "c3", "a2"] + ); + let read = case.read(last(2, &["t"])).await.unwrap(); + assert_eq!(texts(&read), ["c3", "a2"]); + assert!(!read.done); + let newest = case.describe().await.unwrap().last_offset; + assert_eq!(read.next_offset, newest + 1); + let error = case + .read(ReadRecordsRequest { + from_offset: 1, + ..last(1, &[]) + }) + .await + .unwrap_err(); + assert_eq!(reason(&error), "stream_invalid"); + case.close(client(1)).await.unwrap(); + let read = case.read(last(1, &[])).await.unwrap(); + assert_eq!(texts(&read), ["a2", ""]); + assert!(read.done); + } +} + +/// A page stays under `max_bytes`, and a larger Record is still returned alone. +#[tokio::test] +async fn a_page_keeps_under_max_bytes_and_a_larger_record_comes_alone() { + for case in cases_for(Needs::CALLS).await { + case.create(None).await.unwrap(); + let big = "x".repeat(1000); + let records = vec![ + record("t", "a"), + record("t", "b"), + record("t", &big), + record("t", "c"), + ]; + case.append(client(1), 1, records).await.unwrap(); + let capped = |from_offset| ReadRecordsRequest { + from_offset, + max_bytes: 200, + ..Default::default() + }; + let first = case.read(capped(0)).await.unwrap(); + assert_eq!(texts(&first), ["a", "b"]); + let second = case.read(capped(first.next_offset)).await.unwrap(); + assert_eq!(texts(&second), [big]); + let third = case.read(capped(second.next_offset)).await.unwrap(); + assert_eq!(texts(&third), ["c"]); + } +}