diff --git a/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto b/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto index 1f9eb63830..9cb4576f15 100644 --- a/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto +++ b/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto @@ -167,6 +167,10 @@ message ReadRequest { uint32 max_records = 4; // The `state` of the previous answer of the same read. Empty on its first call. bytes state = 5; + // For a reader that sleeps on the stream's notifier. When the read waits its whole `wait` and + // finds nothing, the store arms the stream in the same step, so the next append notifies the + // notifier. A reader that doesn't wait on the notifier leaves it unset. + bool arm = 6; } // A later attempt of a producer started writing. Readers synthesize it, and no store holds it. @@ -201,6 +205,13 @@ message ReadResponse { // The owner's chain ended and the read delivered every record the stream retains. No later // call returns a record. bool done = 4; + // The read asked to arm, found nothing and armed the stream, so the next append notifies the + // stream's notifier. Never set with records or with `done`. + bool armed = 5; + // Set with `armed`: the notification counter of the read's position, zero for an empty + // stream. A notification with a higher counter names a record after the position, so the + // reader waits for a counter above this one. One that landed before the wait started counts. + int64 wake_after = 6; } message LatestRequest { diff --git a/crates/streams/src/handle.rs b/crates/streams/src/handle.rs index 40791da4e4..e49faead7a 100644 --- a/crates/streams/src/handle.rs +++ b/crates/streams/src/handle.rs @@ -255,8 +255,9 @@ impl Streams { /// the interval over. So many readers parked on a quiet stream don't load the server. Once /// the owner ended, or the store marked the stream closed, the read delivers what is left /// and answers `done`. A read pinned to a run ends when that run closes, even if the chain - /// continued as new. A read whose owner Temporal doesn't know yet answers empty within - /// [StreamsOptions::owner_start_grace], so it can start before its owner does. + /// continued as new. A read with `arm` that waits its whole `wait` and finds nothing arms the + /// stream and answers `armed`. A read whose owner Temporal doesn't know yet answers empty + /// within [StreamsOptions::owner_start_grace], so it can start before its owner does. pub async fn read(&self, request: ReadRequest) -> StreamResult { let stream = address(request.stream.as_ref())?; let mut state = ReadState::decode_from(&request.state)?; @@ -302,6 +303,9 @@ impl Streams { remaining.min(until_check) }; page_request.wait = wait.try_into().ok(); + // Only the pass that waits out the read's own wait arms, so a shorter pass before an + // owner check doesn't wake the caller for a gap shorter than the reader chose. + page_request.arm = request.arm && !state.owner_ended && remaining <= until_check; let (mut page, closed) = read_with(&*self.store, &target, &page_request, &mut state).await?; page_request.after = page.cursor.clone(); @@ -317,6 +321,8 @@ impl Streams { } if state.owner_ended { page.done = true; + page.armed = false; + page.wake_after = 0; break page; } if unix_ms() >= state.next_check_ms { @@ -1625,6 +1631,83 @@ mod tests { panic!("the read never ended"); } + #[tokio::test] + async fn only_the_pass_that_waits_out_the_read_arms() { + // Owner checks every 20 ms cut the wait into passes. One arm for the whole wait keeps a + // gap shorter than the reader's wait from costing a notification. + let setup = setup(quick()); + let page = setup + .streams + .read(ReadRequest { + arm: true, + ..read_after("", &[], Duration::from_millis(300)) + }) + .await + .unwrap(); + assert!(page.armed && page.records.is_empty() && !page.done); + assert_eq!(page.wake_after, 0, "an empty stream wakes on any counter"); + assert_eq!( + setup + .store + .newest_or_arm(&chain("run-1"), "out") + .await + .unwrap(), + crate::NewestOrArm::Armed(2), + "the read armed once" + ); + } + + #[tokio::test] + async fn an_armed_read_waits_for_a_counter_above_its_position() { + let setup = setup(StreamsOptions::default()); + setup.streams.append(append("p", 1, "1")).await.unwrap(); + let landed = setup.streams.append(append("p", 2, "2")).await.unwrap(); + let page = setup + .streams + .read(ReadRequest { + arm: true, + ..read_after(&landed.last_cursor, &[], Duration::from_millis(20)) + }) + .await + .unwrap(); + assert!(page.armed); + let at = |cursor: &str| { + crate::cursor_position( + cursor, + setup.store.name(), + &setup.streams.hash(&stream("out")), + ) + .unwrap() + .unwrap() + .to_string() + }; + assert_eq!( + Some(page.wake_after), + crate::progress_counter(&at(&landed.last_cursor)) + ); + let next = setup.streams.append(append("p", 3, "3")).await.unwrap(); + assert!( + crate::progress_counter(&at(&next.last_cursor)).unwrap() > page.wake_after, + "the next record's notification wakes the reader" + ); + } + + #[tokio::test] + async fn a_read_that_finds_its_owner_ended_answers_done_unarmed() { + let setup = setup(quick()); + setup.owners.end("wf", WorkflowExecutionStatus::Completed); + let page = setup + .streams + .read(ReadRequest { + arm: true, + ..read_after("", &[], Duration::from_secs(5)) + }) + .await + .unwrap(); + assert!(page.done); + assert!(!page.armed); + } + #[tokio::test] async fn a_read_ends_when_its_owner_chain_ends_and_marks_it_closed() { let setup = setup(quick()); diff --git a/crates/streams/src/reader.rs b/crates/streams/src/reader.rs index 200d8a3e8c..caa81dfb01 100644 --- a/crates/streams/src/reader.rs +++ b/crates/streams/src/reader.rs @@ -7,6 +7,7 @@ use crate::{ BEGINNING, END, StreamError, StreamResult, StreamStore, cursor_position, mint_cursor, + progress_counter, proto::{ ChainId, ReadRecord, ReadRequest, ReadResponse, StoreLatestRequest, StoreReadRequest, StreamRecord, StreamRecordKind, Supersession, read_record, @@ -305,16 +306,16 @@ pub(crate) async fn read_with( .read(StoreReadRequest { chain: Some(target.chain.clone()), topic: target.topic.clone(), - after_position: position, + after_position: position.clone(), wait: request.wait, max_records: match request.max_records { 0 => DEFAULT_MAX_RECORDS, max => max, }, - arm: false, + arm: request.arm, }) .await?; - let closed = page.closed; + let (closed, armed) = (page.closed, page.armed); let mut records = Vec::with_capacity(page.records.len()); let mut previous = start; for stored in page.records { @@ -360,6 +361,14 @@ pub(crate) async fn read_with( cursor: previous, state: Vec::new(), done: false, + armed, + // An armed read found nothing after `position`, so a notification counter above that + // position's counter can only come from a newer record. + wake_after: if armed { + progress_counter(&position).unwrap_or(0) + } else { + 0 + }, }, closed, ))