diff --git a/crates/streams/protos/temporal/sdk/streams/v1/runtime.proto b/crates/streams/protos/temporal/sdk/streams/v1/runtime.proto index 2a96a2c2a..2f559b76e 100644 --- a/crates/streams/protos/temporal/sdk/streams/v1/runtime.proto +++ b/crates/streams/protos/temporal/sdk/streams/v1/runtime.proto @@ -292,3 +292,71 @@ message WorkflowCloseRequest { // The runtime sets `namespace` and `producer`. Payloads are in their stored form. temporal.api.streamservice.v1.CloseStreamRequest close = 3; } + +// Method `OpenReader`, synchronous. Opens one read of a Stream for `ReadNext`. The reader keeps the +// resume point, the offsets that it gave, the create wait and the end rules. +message OpenReaderRequest { + string namespace = 1; + temporal.api.stream.v1.StreamRef stream = 2; + // Start at this offset. Zero starts at `start`. + int64 from_offset = 3; + // END reads only the Records after the newest one. + temporal.api.enums.v1.StreamReadStart start = 4; + // Start with the newest `last` Records. Not with `from_offset`. + int32 last = 5; + // Wait for new Records at the newest one. Otherwise the read ends there. + bool wait_new_records = 6; + // How long the read waits for a Stream that nobody created yet. Empty fails at once with + // NOT_FOUND. + google.protobuf.Duration create_wait = 7; + // The most Record bytes in one page. Zero is no cap. + int64 max_bytes = 8; +} + +message OpenReaderResponse { + uint64 reader_id = 1; +} + +// Method `ReadNext`, asynchronous. Reads pages until one has items, the read ends or `max_wait` +// passes, and returns that page. A dropped call leaves the reader as it was. +message ReadNextRequest { + uint64 reader_id = 1; + // The longest wait of this call, below the caller's RPC timeout. Empty is 60 s. + google.protobuf.Duration max_wait = 2; +} + +message ReaderItem { + temporal.api.stream.v1.StreamRecord record = 1; + // The resume point after this item. After a Superseded item it is the item's own offset, + // since the first Record of its batch has the same offset and comes next. + int64 next_offset = 2; + // Superseded only. The offsets of the Records that this read gave and the item withdraws, in + // order. + repeated int64 withdrawn_offsets = 3; +} + +message ReaderPage { + repeated ReaderItem items = 1; + // The resume point after the page. Zero before the first page of a read from END or of a read + // with `last`. + int64 next_offset = 2; + // The read ended, at the close or at the newest Record of a read that doesn't wait. + bool done = 3; +} + +// Method `FreeReader`, synchronous, with `FreeResponse`. Frees the reader. A `ReadNext` in flight +// still completes. +message FreeReaderRequest { + uint64 reader_id = 1; +} + +// Method `Withdraws`, static. Whether a Superseded item withdraws the earlier Records of a +// Producer, such as Records from an earlier read. +message WithdrawsRequest { + temporal.api.stream.v1.Supersession supersession = 1; + temporal.api.stream.v1.Producer producer = 2; +} + +message WithdrawsResponse { + bool withdraws = 1; +} diff --git a/crates/streams/src/epoch.rs b/crates/streams/src/epoch.rs index 795edaf0e..75536304b 100644 --- a/crates/streams/src/epoch.rs +++ b/crates/streams/src/epoch.rs @@ -91,9 +91,14 @@ pub fn withdraws(supersession: &Supersession, record: &StreamRecord) -> bool { let Some(Item::Data(data)) = &record.item else { return false; }; - let Some(producer) = data.producer.as_ref() else { - return false; - }; + data.producer + .as_ref() + .is_some_and(|producer| withdraws_producer(supersession, producer)) +} + +/// Whether a `Superseded` item withdraws the earlier Records of `producer`, by the rule of +/// [withdraws]. +pub fn withdraws_producer(supersession: &Supersession, producer: &Producer) -> bool { if !supersession.withdraw_run.is_empty() && producer.run_id == supersession.withdraw_run { return true; } diff --git a/crates/streams/src/lib.rs b/crates/streams/src/lib.rs index edddad3cc..667a622bf 100644 --- a/crates/streams/src/lib.rs +++ b/crates/streams/src/lib.rs @@ -14,6 +14,7 @@ pub mod memory; mod metrics; pub mod offset; pub mod protos; +mod reader; mod record; mod runtime; mod scope; diff --git a/crates/streams/src/reader.rs b/crates/streams/src/reader.rs new file mode 100644 index 000000000..0d4a742da --- /dev/null +++ b/crates/streams/src/reader.rs @@ -0,0 +1,364 @@ +//! Readers. A reader keeps the state of one read of a Stream, so lang only decodes each item. + +use crate::{ + Reason, StreamError, StreamResult, + epoch::{epoch_key, withdraws_producer}, + proto::{ + OpenReaderRequest, Producer, ReadNextRequest, ReadRecordsRequest, ReadRecordsResponse, + ReaderItem, ReaderPage, StreamProducerKind, StreamReadStart, Supersession, + stream_record::Item, + }, + runtime::Shared, + service::MAX_WAIT, +}; +use parking_lot::Mutex; +use std::{ + collections::HashMap, + sync::{ + Arc, + atomic::{AtomicU64, Ordering}, + }, + time::Duration, +}; +use tokio::time::Instant; + +// An empty page that comes back in less than half of its wait is followed by a pause, so a Store +// that ignores the wait isn't asked again at once. The pause doubles up to the last one. +const FIRST_PAUSE: Duration = Duration::from_millis(50); +const LAST_PAUSE: Duration = Duration::from_secs(2); + +/// The Producer fields that the withdrawal rule reads. +type ProducerKey = (i32, String, String, Option); + +struct State { + /// The request of the next page. + request: ReadRecordsRequest, + wait_new_records: bool, + create_wait: Option, + /// Set by the first `ReadNext`, so that the create wait starts when lang starts to read. + create_until: Option, + /// The resume point, zero before the first page of a read from END or with `last`. + position: i64, + /// The offsets of the Records that the read gave, as runs, by Producer. + shown: HashMap)>, + done: bool, +} + +pub(crate) struct Reader { + /// Allows one `ReadNext` at a time. + state: tokio::sync::Mutex, +} + +#[derive(Default)] +pub(crate) struct Readers { + next: AtomicU64, + table: Mutex>>, +} + +impl Readers { + fn get(&self, id: u64) -> StreamResult> { + self.table + .lock() + .get(&id) + .cloned() + .ok_or_else(unknown_reader) + } + + pub(crate) fn free(&self, id: u64) -> StreamResult<()> { + self.table + .lock() + .remove(&id) + .map(|_| ()) + .ok_or_else(unknown_reader) + } + + pub(crate) fn clear(&self) { + self.table.lock().clear(); + } +} + +fn unknown_reader() -> StreamError { + StreamError::new(Reason::UnknownWriter, "No reader has the id") +} + +fn duration(value: Option) -> Option { + let value = value?; + let duration = Duration::new(value.seconds.max(0) as u64, value.nanos.max(0) as u32); + (!duration.is_zero()).then_some(duration) +} + +/// The time `wait` from now. A wait of more than a year counts as a year, so it can't overflow. +fn after(wait: Duration) -> Instant { + Instant::now() + wait.min(Duration::from_secs(365 * 24 * 60 * 60)) +} + +fn proto_duration(value: Duration) -> prost_types::Duration { + prost_types::Duration { + seconds: value.as_secs() as i64, + nanos: value.subsec_nanos() as i32, + } +} + +fn producer_key(producer: &Producer) -> ProducerKey { + ( + producer.kind, + producer.id.clone(), + producer.run_id.clone(), + producer.epoch.as_ref().map(epoch_key), + ) +} + +/// The page of a read that ended with no new item. +fn ended() -> ReadRecordsResponse { + ReadRecordsResponse { + done: true, + ..Default::default() + } +} + +impl State { + /// Moves the resume point over a page and returns its items. A Superseded item takes the + /// offsets of the Records that it withdraws from this read. + fn take(&mut self, page: ReadRecordsResponse) -> Vec { + let mut items = vec![]; + for record in page.records { + let offset = record.offset; + let mut withdrawn_offsets = vec![]; + match &record.item { + // The first Record of its batch has the same offset and comes next. + Some(Item::Superseded(supersession)) => { + self.position = offset; + withdrawn_offsets = self.withdraw(supersession); + } + Some(Item::Data(data)) => { + self.position = self.position.max(offset + 1); + // Nothing withdraws a client's Records, so a long read of them keeps nothing. + let producer = data.producer.as_ref(); + if let Some(producer) = + producer.filter(|p| p.kind != StreamProducerKind::Client as i32) + { + let (_, runs) = self + .shown + .entry(producer_key(producer)) + .or_insert_with(|| (producer.clone(), vec![])); + match runs.last_mut() { + Some((_, last)) if *last + 1 == offset => *last = offset, + _ => runs.push((offset, offset)), + } + } + } + Some(Item::Close(_)) => { + self.position = self.position.max(offset + 1); + self.done = true; + } + None => self.position = self.position.max(offset + 1), + } + items.push(ReaderItem { + record: Some(record), + next_offset: self.position, + withdrawn_offsets, + }); + if self.done { + return items; + } + } + self.position = self.position.max(page.next_offset); + self.done |= page.done; + items + } + + fn withdraw(&mut self, supersession: &Supersession) -> Vec { + let mut offsets = vec![]; + // A withdrawn Record stays withdrawn, so the read forgets it. + self.shown.retain(|_, (producer, runs)| { + if !withdraws_producer(supersession, producer) { + return true; + } + for (first, last) in runs.iter() { + offsets.extend(*first..=*last); + } + false + }); + offsets.sort_unstable(); + offsets + } +} + +impl Shared { + pub(crate) fn open_reader(&self, request: OpenReaderRequest) -> StreamResult { + if request.from_offset < 0 || request.last < 0 || request.max_bytes < 0 { + return Err(StreamError::invalid( + "from_offset, last and max_bytes must be 0 or more", + )); + } + let from_end = request.from_offset == 0 && request.start == StreamReadStart::End as i32; + if request.last > 0 && (request.from_offset > 0 || from_end) { + return Err(StreamError::invalid( + "A read with last takes no from_offset and no END", + )); + } + let state = State { + request: ReadRecordsRequest { + namespace: request.namespace, + stream: request.stream, + from_offset: request.from_offset, + start: if request.last > 0 { 0 } else { request.start }, + last: request.last, + max_bytes: request.max_bytes, + ..Default::default() + }, + wait_new_records: request.wait_new_records, + create_wait: duration(request.create_wait), + create_until: None, + position: if request.last > 0 || from_end { + 0 + } else { + request.from_offset + }, + shown: HashMap::new(), + done: false, + }; + let id = self.readers.next.fetch_add(1, Ordering::Relaxed) + 1; + let reader = Arc::new(Reader { + state: tokio::sync::Mutex::new(state), + }); + self.readers.table.lock().insert(id, reader); + // A shutdown that already cleared the table doesn't see this reader. + if self.is_shut_down() { + let _ = self.readers.free(id); + return Err(crate::runtime::shut_down()); + } + Ok(id) + } + + pub(crate) async fn read_next(&self, request: ReadNextRequest) -> StreamResult { + let reader = self.readers.get(request.reader_id)?; + let mut state = reader.state.lock().await; + if !state.done { + let create_until = state.create_until.or_else(|| state.create_wait.map(after)); + let deadline = after(duration(request.max_wait).unwrap_or(MAX_WAIT)); + // The state changes only after the last await, so a dropped call changes nothing. + let paged = self.next_page(&state, create_until, deadline).await; + state.create_until = create_until; + let (page, next) = paged?; + state.request = next; + let items = state.take(page); + return Ok(ReaderPage { + items, + next_offset: state.position, + done: state.done, + }); + } + Ok(ReaderPage { + next_offset: state.position, + done: true, + ..Default::default() + }) + } + + /// The next page with items, or the one at the deadline or at the end of the read, and the + /// request of the page after it. + async fn next_page( + &self, + state: &State, + create_until: Option, + deadline: Instant, + ) -> StreamResult<(ReadRecordsResponse, ReadRecordsRequest)> { + let service = self.service().await?; + let mut request = state.request.clone(); + let mut pause = FIRST_PAUSE; + loop { + let now = Instant::now(); + let left = deadline.saturating_duration_since(now); + request.create_wait = create_until + .map(|until| until.saturating_duration_since(now).min(left)) + .filter(|wait| !wait.is_zero()) + .map(proto_duration); + // The newest Records come in one page that never waits. + let last = request.last > 0; + let wait = (state.wait_new_records && !last) + .then(|| left.clamp(Duration::from_millis(1), MAX_WAIT)); + request.wait = wait.map(proto_duration); + let page = match service.read_records(request.clone()).await { + Ok(page) => page, + Err(error) + if error.reason == Reason::NotFound + && create_until.is_some_and(|until| until > Instant::now()) => + { + if Instant::now() >= deadline { + return Ok((ReadRecordsResponse::default(), request)); + } + continue; + } + Err(error) => return Err(error), + }; + let from = request.from_offset; + if page.next_offset > 0 { + (request.from_offset, request.last, request.start) = (page.next_offset, 0, 0); + } + if last { + // A read that waits goes on after the newest Records. + let end = page.done || !state.wait_new_records || page.next_offset == 0; + if !page.records.is_empty() || end { + let done = page.done || end; + return Ok((ReadRecordsResponse { done, ..page }, request)); + } + continue; + } + if !page.records.is_empty() || page.done { + return Ok((page, request)); + } + if !state.wait_new_records { + // An empty page that didn't move is at the newest Record. + if page.next_offset == 0 || page.next_offset <= from { + return Ok(( + ReadRecordsResponse { + next_offset: page.next_offset, + ..ended() + }, + request, + )); + } + continue; + } + let left = deadline.saturating_duration_since(Instant::now()); + if left.is_zero() { + return Ok((page, request)); + } + if now.elapsed() < wait.unwrap_or_default() / 2 { + tokio::time::sleep(pause.min(left)).await; + pause = (pause * 2).min(LAST_PAUSE); + } else { + pause = FIRST_PAUSE; + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::proto::StreamRecord; + + #[test] + fn an_item_of_an_unknown_kind_moves_the_resume_point() { + let mut state = State { + request: ReadRecordsRequest::default(), + wait_new_records: false, + create_wait: None, + create_until: None, + position: 5, + shown: HashMap::new(), + done: false, + }; + let page = ReadRecordsResponse { + records: vec![StreamRecord { + offset: 9, + item: None, + }], + ..Default::default() + }; + let items = state.take(page); + assert_eq!((items[0].next_offset, state.position), (10, 10)); + } +} diff --git a/crates/streams/src/runtime.rs b/crates/streams/src/runtime.rs index 9db1ccf26..7db0b58b6 100644 --- a/crates/streams/src/runtime.rs +++ b/crates/streams/src/runtime.rs @@ -7,10 +7,12 @@ use crate::{ proto::{ AppendRecordsRequest, CloseScopeRequest, CloseStreamRequest, CloseWriterRequest, CloseWriterResponse, CloseWritersRequest, CloseWritersResponse, DescribeStreamRequest, - FreeResponse, FreeScopeRequest, FreeWriterRequest, FreeWriterResponse, MemoryStreamConfig, - OpenWriterRequest, OpenWriterResponse, ReadRecordsRequest, ShutdownRequest, - WorkflowAppendRequest, WriteRecordsRequest, WriteRecordsResponse, + FreeReaderRequest, FreeResponse, FreeScopeRequest, FreeWriterRequest, FreeWriterResponse, + MemoryStreamConfig, OpenReaderResponse, OpenWriterRequest, OpenWriterResponse, + ReadRecordsRequest, ShutdownRequest, WorkflowAppendRequest, WriteRecordsRequest, + WriteRecordsResponse, }, + reader::Readers, scope::Scopes, service::StreamService, store::{DEFAULT_RETENTION, DEFAULT_SAFETY_TTL, Store}, @@ -59,6 +61,7 @@ pub(crate) struct Shared { pub(crate) writers: Writers, pub(crate) scopes: Scopes, pub(crate) pending: Pending, + pub(crate) readers: Readers, shut_down: AtomicBool, /// Set when a `Shutdown` ends every scope and writer. Then the `Free` methods do nothing. finished: AtomicBool, @@ -134,6 +137,7 @@ impl StreamRuntime { writers: Writers::default(), scopes: Scopes::default(), pending: Pending::default(), + readers: Readers::default(), shut_down: AtomicBool::new(false), finished: AtomicBool::new(false), }); @@ -268,9 +272,11 @@ impl StreamRuntime { .await? .encode_to_vec(), "WorkflowClose" => shared.workflow_close(decode(input)?).await?.encode_to_vec(), + "ReadNext" => shared.read_next(decode(input)?).await?.encode_to_vec(), "Shutdown" => { decode::(input)?; shared.shut_down.store(true, Ordering::SeqCst); + shared.readers.clear(); let closed = shared.close_everything().await; shared.finished.store(true, Ordering::SeqCst); closed.encode_to_vec() @@ -310,6 +316,15 @@ impl StreamRuntime { "WorkflowAppendPending" => Ok(shared .workflow_append_pending(&decode::(input)?)? .encode_to_vec()), + "OpenReader" => { + let reader_id = shared.open_reader(decode(input)?)?; + Ok(OpenReaderResponse { reader_id }.encode_to_vec()) + } + "FreeReader" => { + let request = decode::(input)?; + shared.readers.free(request.reader_id)?; + Ok(FreeResponse {}.encode_to_vec()) + } "FreeScope" => { let request = decode::(input)?; shared.free_scope(request.scope_id)?; diff --git a/crates/streams/src/sdk.rs b/crates/streams/src/sdk.rs index 23d28b7fd..426263060 100644 --- a/crates/streams/src/sdk.rs +++ b/crates/streams/src/sdk.rs @@ -1,12 +1,14 @@ //! The static calls, which need no runtime object and no Store. These are the Producer of an -//! Activity attempt and the scoped Stream id. So every SDK derives them by one rule. +//! Activity attempt, the scoped Stream id and the withdrawal rule. So every SDK applies one rule. use crate::{ - Reason, StreamError, StreamResult, ids, + Reason, StreamError, StreamResult, + epoch::withdraws_producer, + ids, proto::{ ActivityContext, DeriveProducerRequest, DeriveProducerResponse, Producer, ProducerEpoch, ProducerPurpose, ScopedStreamIdRequest, ScopedStreamIdResponse, StreamProducerKind, - WorkflowTaskEpoch, + WithdrawsRequest, WithdrawsResponse, WorkflowTaskEpoch, }, runtime::decode, }; @@ -17,6 +19,18 @@ use prost::Message; pub fn call_static(method: &str, input: &[u8]) -> StreamResult> { match method { "DeriveProducer" => Ok(derive_producer(decode(input)?)?.encode_to_vec()), + "Withdraws" => { + let request = decode::(input)?; + let supersession = request + .supersession + .ok_or_else(|| StreamError::invalid("The request has no Superseded item"))?; + Ok(WithdrawsResponse { + withdraws: request + .producer + .is_some_and(|producer| withdraws_producer(&supersession, &producer)), + } + .encode_to_vec()) + } "ScopedStreamId" => { let request = decode::(input)?; Ok(ScopedStreamIdResponse { @@ -37,7 +51,7 @@ pub fn call_static(method: &str, input: &[u8]) -> StreamResult> { /// Whether [call_static] runs the method. pub(crate) fn is_static(method: &str) -> bool { - matches!(method, "DeriveProducer" | "ScopedStreamId") + matches!(method, "DeriveProducer" | "ScopedStreamId" | "Withdraws") } /// The Producer of an Activity attempt, from the facts that lang copies from its Activity info. A diff --git a/crates/streams/src/testing/mod.rs b/crates/streams/src/testing/mod.rs index a0efdde16..e37ff7106 100644 --- a/crates/streams/src/testing/mod.rs +++ b/crates/streams/src/testing/mod.rs @@ -3,6 +3,7 @@ pub(crate) mod support; +pub mod reader; pub mod scope; pub mod service; pub mod store; @@ -170,4 +171,9 @@ mod tests { async fn the_workflow_suite_passes_on_the_memory_store() { run(vec![Arc::new(MemoryBackend)], workflow::CASES).await; } + + #[tokio::test] + async fn the_reader_suite_passes_on_the_memory_store() { + run(vec![Arc::new(MemoryBackend)], reader::CASES).await; + } } diff --git a/crates/streams/src/testing/reader.rs b/crates/streams/src/testing/reader.rs new file mode 100644 index 000000000..82afb9dea --- /dev/null +++ b/crates/streams/src/testing/reader.rs @@ -0,0 +1,509 @@ +//! Readers with resume points and withdrawn offsets, through the runtime's byte call, as lang +//! sends them. +//! +//! Every Store runs the suite through its own runtime. + +use super::case; +use crate::{ + StreamError, StreamRuntime, + proto::{ + ActivityContext, AppendRecord, CloseStreamAsRequest, CloseStreamRequest, + CloseStreamResponse, CreateStreamRequest, CreateStreamResponse, FreeReaderRequest, + FreeResponse, OpenActivityScopeRequest, OpenClientScopeRequest, OpenReaderRequest, + OpenReaderResponse, OpenScopeResponse, Producer, ProducerEpoch, PublishRequest, + PublishResponse, ReadNextRequest, ReaderItem, ReaderPage, StreamProducerKind, + StreamReadStart, StreamRef, Supersession, WithdrawsRequest, WithdrawsResponse, + stream_record::Item, + }, +}; +use prost::Message; +use prost_types::Timestamp; +use std::{ + sync::Arc, + time::{Duration, Instant, SystemTime, UNIX_EPOCH}, +}; +use temporalio_common::protos::temporal::api::common::v1::Payload; + +static NEXT: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); + +struct Case { + ext: Arc, + namespace: String, + stream: String, +} + +fn unique() -> String { + let n = NEXT.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + let now = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos(); + format!("{now}-{n}") +} + +/// A case on each backend, with its Stream not created yet. +fn uncreated() -> Vec { + super::support::stream_runtimes() + .into_iter() + .map(|ext| { + let unique = unique(); + Case { + ext, + namespace: format!("ns-{unique}"), + stream: format!("s-{unique}"), + } + }) + .collect() +} + +async fn cases() -> Vec { + let cases = uncreated(); + for case in &cases { + case.create().await; + } + cases +} + +fn millis(ms: u64) -> Option { + let duration = Duration::from_millis(ms); + Some(prost_types::Duration { + seconds: duration.as_secs() as i64, + nanos: duration.subsec_nanos() as i32, + }) +} + +fn activity(attempt: i32) -> ActivityContext { + ActivityContext { + in_workflow: true, + workflow_id: "wf".into(), + workflow_run_id: "r1".into(), + activity_id: "a".into(), + attempt, + current_attempt_scheduled_time: Some(Timestamp { + seconds: 10 * attempt as i64, + nanos: 0, + }), + ..Default::default() + } +} + +fn texts(page: &ReaderPage) -> Vec { + page.items + .iter() + .filter_map(|item| match &item.record.as_ref().unwrap().item { + Some(Item::Data(data)) => { + Some(String::from_utf8(data.body.clone().unwrap().data).unwrap()) + } + _ => None, + }) + .collect() +} + +fn offset(item: &ReaderItem) -> i64 { + item.record.as_ref().unwrap().offset +} + +fn is_superseded(item: &ReaderItem) -> bool { + matches!( + item.record.as_ref().unwrap().item, + Some(Item::Superseded(_)) + ) +} + +impl Case { + async fn call( + &self, + method: &str, + request: Req, + ) -> Result { + let bytes = self.ext.call(method, &request.encode_to_vec()).await?; + Ok(Resp::decode(bytes.as_slice()).unwrap()) + } + + fn call_sync( + &self, + method: &str, + request: Req, + ) -> Result { + let bytes = self.ext.call_sync(method, &request.encode_to_vec())?; + Ok(Resp::decode(bytes.as_slice()).unwrap()) + } + + fn stream_ref(&self) -> StreamRef { + StreamRef { + id: self.stream.clone(), + ..Default::default() + } + } + + async fn create(&self) { + let _: CreateStreamResponse = self + .call( + "CreateStream", + CreateStreamRequest { + namespace: self.namespace.clone(), + stream_id: self.stream.clone(), + ..Default::default() + }, + ) + .await + .unwrap(); + } + + fn client(&self) -> u64 { + let opened: OpenScopeResponse = self + .call_sync( + "OpenClientScope", + OpenClientScopeRequest { + namespace: self.namespace.clone(), + identity: "me".into(), + ..Default::default() + }, + ) + .unwrap(); + opened.scope_id + } + + async fn activity(&self, attempt: i32) -> u64 { + let opened: OpenScopeResponse = self + .call( + "OpenActivityScope", + OpenActivityScopeRequest { + namespace: self.namespace.clone(), + context: Some(activity(attempt)), + streams: vec![self.stream_ref()], + }, + ) + .await + .unwrap(); + opened.scope_id + } + + async fn publish(&self, scope_id: u64, texts: &[&str]) -> PublishResponse { + self.call( + "Publish", + PublishRequest { + scope_id, + stream: Some(self.stream_ref()), + records: texts + .iter() + .map(|text| AppendRecord { + body: Some(Payload { + data: text.as_bytes().to_vec(), + ..Default::default() + }), + }) + .collect(), + }, + ) + .await + .unwrap() + } + + fn open(&self, change: impl FnOnce(&mut OpenReaderRequest)) -> Result { + let mut request = OpenReaderRequest { + namespace: self.namespace.clone(), + stream: Some(self.stream_ref()), + ..Default::default() + }; + change(&mut request); + let opened: OpenReaderResponse = self.call_sync("OpenReader", request)?; + Ok(opened.reader_id) + } + + async fn next(&self, reader_id: u64, max_wait_ms: u64) -> Result { + self.call( + "ReadNext", + ReadNextRequest { + reader_id, + max_wait: millis(max_wait_ms), + }, + ) + .await + } +} + +/// Each item carries the resume point after it. A read with no wait ends at the newest Record, +/// which the page after the last Record tells. +async fn each_item_carries_its_resume_point() { + for case in cases().await { + let client = case.client(); + let written = case.publish(client, &["a", "b", "c"]).await; + let reader = case.open(|_| {}).unwrap(); + let page = case.next(reader, 1000).await.unwrap(); + assert_eq!(texts(&page), ["a", "b", "c"]); + for item in &page.items { + assert_eq!(item.next_offset, offset(item) + 1); + } + assert_eq!(page.next_offset, written.last_offset + 1); + let after = case.next(reader, 1000).await.unwrap(); + assert!(after.items.is_empty() && after.done); + assert_eq!(after.next_offset, page.next_offset); + assert!(case.next(reader, 1000).await.unwrap().done); + } +} + +/// A Superseded item lists the offsets of the Records that this read gave and it withdraws. Its +/// resume point is its own offset. +async fn a_superseded_item_lists_what_it_withdraws_from_this_read() { + for case in cases().await { + let first = case.activity(1).await; + case.publish(first, &["a1", "a2"]).await; + let reader = case.open(|r| r.wait_new_records = true).unwrap(); + let page = case.next(reader, 1000).await.unwrap(); + let shown: Vec = page.items.iter().map(offset).collect(); + assert_eq!(shown.len(), 2); + let retry = case.activity(2).await; + case.publish(retry, &["b"]).await; + let mut items = vec![]; + while items.len() < 2 { + let page = case.next(reader, 1000).await.unwrap(); + assert!(!page.done); + items.extend(page.items); + } + assert!(is_superseded(&items[0])); + assert_eq!(items[0].withdrawn_offsets, shown); + assert_eq!(items[0].next_offset, offset(&items[0])); + assert!(items[1].withdrawn_offsets.is_empty()); + } +} + +/// A read with `last` starts with the newest Records. With a wait it then goes on after them. +async fn a_last_read_starts_with_the_newest_records() { + for case in cases().await { + let client = case.client(); + case.publish(client, &["1", "2", "3", "4"]).await; + let reader = case.open(|r| r.last = 2).unwrap(); + let page = case.next(reader, 1000).await.unwrap(); + assert_eq!(texts(&page), ["3", "4"]); + assert!(page.done); + let reader = case + .open(|r| { + r.last = 1; + r.wait_new_records = true; + }) + .unwrap(); + assert_eq!(texts(&case.next(reader, 1000).await.unwrap()), ["4"]); + case.publish(client, &["5"]).await; + assert_eq!(texts(&case.next(reader, 1000).await.unwrap()), ["5"]); + } +} + +/// A read from END gives only new Records. A quiet read returns an empty page at `max_wait`, with +/// the resume point after the newest Record. +async fn an_end_read_gives_only_new_records() { + for case in cases().await { + let client = case.client(); + let old = case.publish(client, &["old"]).await; + let reader = case + .open(|r| { + r.start = StreamReadStart::End as i32; + r.wait_new_records = true; + }) + .unwrap(); + let started = Instant::now(); + let quiet = case.next(reader, 300).await.unwrap(); + assert!(started.elapsed() >= Duration::from_millis(250)); + assert!(quiet.items.is_empty() && !quiet.done); + assert_eq!(quiet.next_offset, old.last_offset + 1); + case.publish(client, &["new"]).await; + assert_eq!(texts(&case.next(reader, 1000).await.unwrap()), ["new"]); + } +} + +/// A read waits for the Stream's create only as long as `create_wait`. +async fn a_read_waits_for_the_create_up_to_create_wait() { + for case in uncreated() { + let reader = case.open(|_| {}).unwrap(); + let error = case.next(reader, 1000).await.unwrap_err(); + assert_eq!(error.reason.token(), "stream_not_found"); + let reader = case + .open(|r| { + r.wait_new_records = true; + r.create_wait = millis(5000); + }) + .unwrap(); + let write = async { + tokio::time::sleep(Duration::from_millis(200)).await; + case.create().await; + let client = case.client(); + case.publish(client, &["x"]).await; + }; + let (page, ()) = tokio::join!(case.next(reader, 4000), write); + assert_eq!(texts(&page.unwrap()), ["x"]); + } +} + +/// The create wait starts at the first `ReadNext` that returns, not at the open or at a dropped +/// call. +async fn the_create_wait_starts_at_the_first_read() { + for case in uncreated() { + let reader = case + .open(|r| { + r.wait_new_records = true; + r.create_wait = millis(1000); + }) + .unwrap(); + tokio::select! { + biased; + _ = case.next(reader, 3000) => panic!("the read answered in its first poll"), + _ = std::future::ready(()) => {} + } + tokio::time::sleep(Duration::from_millis(1200)).await; + let write = async { + tokio::time::sleep(Duration::from_millis(200)).await; + case.create().await; + let client = case.client(); + case.publish(client, &["x"]).await; + }; + let (page, ()) = tokio::join!(case.next(reader, 3000), write); + assert_eq!(texts(&page.unwrap()), ["x"]); + } +} + +/// The close ends the read, as its last item. +async fn the_close_ends_the_read() { + for case in cases().await { + let client = case.client(); + case.publish(client, &["a"]).await; + let _: CloseStreamResponse = case + .call( + "CloseStreamAs", + CloseStreamAsRequest { + scope_id: client, + close: Some(CloseStreamRequest { + stream: Some(case.stream_ref()), + ..Default::default() + }), + }, + ) + .await + .unwrap(); + let reader = case.open(|r| r.wait_new_records = true).unwrap(); + let page = case.next(reader, 1000).await.unwrap(); + assert_eq!(texts(&page), ["a"]); + let last = page.items.last().unwrap(); + assert!(matches!( + last.record.as_ref().unwrap().item, + Some(Item::Close(_)) + )); + assert!(page.done); + assert_eq!(page.next_offset, offset(last) + 1); + } +} + +/// A dropped `ReadNext` leaves the reader as it was. +async fn a_dropped_read_leaves_the_reader_as_it_was() { + for case in cases().await { + let reader = case.open(|r| r.wait_new_records = true).unwrap(); + let dropped = tokio::time::timeout(Duration::from_millis(100), case.next(reader, 5000)); + assert!(dropped.await.is_err()); + let client = case.client(); + let written = case.publish(client, &["a"]).await; + let page = case.next(reader, 1000).await.unwrap(); + assert_eq!(texts(&page), ["a"]); + assert_eq!(page.next_offset, written.last_offset + 1); + } +} + +/// Bad options are refused at the open, and a freed reader is gone. +async fn bad_options_are_refused_and_a_freed_reader_is_gone() { + for case in cases().await { + let bad: [fn(&mut OpenReaderRequest); 4] = [ + |r| r.last = -1, + |r| r.from_offset = -1, + |r| { + r.last = 2; + r.from_offset = 5; + }, + |r| { + r.last = 2; + r.start = StreamReadStart::End as i32; + }, + ]; + for change in bad { + assert_eq!( + case.open(change).unwrap_err().reason.token(), + "stream_invalid" + ); + } + // A huge wait counts as a year, so it doesn't overflow. + let huge = Some(prost_types::Duration { + seconds: i64::MAX, + nanos: 0, + }); + let reader = case.open(|r| r.create_wait = huge).unwrap(); + let page: ReaderPage = case + .call( + "ReadNext", + ReadNextRequest { + reader_id: reader, + max_wait: huge, + }, + ) + .await + .unwrap(); + assert!(page.done); + let reader = case.open(|_| {}).unwrap(); + let _: FreeResponse = case + .call_sync("FreeReader", FreeReaderRequest { reader_id: reader }) + .unwrap(); + let error = case.next(reader, 100).await.unwrap_err(); + assert_eq!(error.reason.token(), "stream_unknown_writer"); + } +} + +/// The static `Withdraws` applies the withdrawal rule to a Producer of an earlier read. +async fn the_static_withdrawal_rule_matches_the_reader() { + for case in cases().await { + let producer = |attempt| Producer { + kind: StreamProducerKind::Activity as i32, + id: crate::ids::activity_producer_id("r1", "a", ""), + epoch: Some(ProducerEpoch { + attempt, + ..Default::default() + }), + run_id: "r1".into(), + }; + let supersession = |kind: StreamProducerKind| Supersession { + producer_kind: kind as i32, + producer_id: crate::ids::activity_producer_id("r1", "a", ""), + epoch: producer(2).epoch, + run_id: "r1".into(), + ..Default::default() + }; + for (kind, attempt, expected) in [ + (StreamProducerKind::Activity, 1, true), + (StreamProducerKind::Activity, 2, false), + (StreamProducerKind::Client, 1, false), + ] { + let answer: WithdrawsResponse = case + .call_sync( + "Withdraws", + WithdrawsRequest { + supersession: Some(supersession(kind)), + producer: Some(producer(attempt)), + }, + ) + .unwrap(); + assert_eq!(answer.withdraws, expected, "{kind:?} {attempt}"); + } + let error = crate::call_static("Withdraws", &WithdrawsRequest::default().encode_to_vec()) + .unwrap_err(); + assert_eq!(error.reason.token(), "stream_invalid"); + } +} + +/// The cases of the suite, in file order. +pub const CASES: &[super::Case] = &[ + case!(each_item_carries_its_resume_point), + case!(a_superseded_item_lists_what_it_withdraws_from_this_read), + case!(a_last_read_starts_with_the_newest_records), + case!(an_end_read_gives_only_new_records), + case!(a_read_waits_for_the_create_up_to_create_wait), + case!(the_create_wait_starts_at_the_first_read), + case!(the_close_ends_the_read), + case!(a_dropped_read_leaves_the_reader_as_it_was), + case!(bad_options_are_refused_and_a_freed_reader_is_gone), + case!(the_static_withdrawal_rule_matches_the_reader), +]; diff --git a/crates/streams/tests/vectors.rs b/crates/streams/tests/vectors.rs index e2e466b7e..429d54373 100644 --- a/crates/streams/tests/vectors.rs +++ b/crates/streams/tests/vectors.rs @@ -1,13 +1,14 @@ //! The shared test vectors, which `testdata/gen_vectors.py` writes from a second implementation. //! Every SDK that builds ids or reads Records runs the same files. +use prost::Message; use prost_types::Timestamp; use serde_json::Value; use temporalio_common::protos::temporal::api::common::v1::{ Payload, payload::ExternalPayloadDetails, }; use temporalio_streams::{ - derive_producer, + call_static, derive_producer, digest::batch_digest, epoch::{epoch_key, task_key, withdraws}, ids::{self, StreamKeys, fields}, @@ -15,7 +16,8 @@ use temporalio_streams::{ proto::{ ActivityContext, AppendRecord, DeriveProducerRequest, Producer, ProducerEpoch, ProducerPurpose, StreamData, StreamProducerKind, StreamRecord, Supersession, - WorkflowTaskAttempt, WorkflowTaskEpoch, stream_record::Item, + WithdrawsRequest, WithdrawsResponse, WorkflowTaskAttempt, WorkflowTaskEpoch, + stream_record::Item, }, }; @@ -257,24 +259,34 @@ fn the_appendix_e_withdrawals_match() { run_id: s["run_id"].as_str().unwrap_or_default().to_string(), }; let p = &case["record"]["producer"]; + let producer = Producer { + kind: kind(text(p, "kind")) as i32, + id: text(p, "id").to_string(), + epoch: p.get("epoch").map(epoch), + run_id: p["run_id"].as_str().unwrap_or_default().to_string(), + }; let record = StreamRecord { item: Some(Item::Data(StreamData { - producer: Some(Producer { - kind: kind(text(p, "kind")) as i32, - id: text(p, "id").to_string(), - epoch: p.get("epoch").map(epoch), - run_id: p["run_id"].as_str().unwrap_or_default().to_string(), - }), + producer: Some(producer.clone()), ..Default::default() })), ..Default::default() }; + let expected = case["withdraws"].as_bool().unwrap(); assert_eq!( withdraws(&supersession, &record), - case["withdraws"].as_bool().unwrap(), + expected, "{}", text(case, "name") ); + // The static method that lang calls for Records of an earlier read applies the same rule. + let request = WithdrawsRequest { + supersession: Some(supersession), + producer: Some(producer), + }; + let output = call_static("Withdraws", &request.encode_to_vec()).unwrap(); + let answer = WithdrawsResponse::decode(output.as_slice()).unwrap(); + assert_eq!(answer.withdraws, expected, "{}", text(case, "name")); } }