diff --git a/crates/protos/protos/local/temporal/sdk/core/streams/nexus.proto b/crates/protos/protos/local/temporal/sdk/core/streams/nexus.proto index 3c89ae53c2..8ceabee1bc 100644 --- a/crates/protos/protos/local/temporal/sdk/core/streams/nexus.proto +++ b/crates/protos/protos/local/temporal/sdk/core/streams/nexus.proto @@ -19,6 +19,8 @@ service StreamNexusService { rpc NexusAppend(NexusCall) returns (NexusStep); // The stream service's Nexus `read` operation, over its JSON transfer form. rpc NexusRead(NexusCall) returns (NexusStep); + // The stream service's Nexus `read_range` operation, over its JSON transfer form. + rpc NexusReadRange(NexusCall) returns (NexusStep); // Goes on with a call that asked for codec work. rpc NexusResume(NexusResumeRequest) returns (NexusStep); // Attaches the caller of a stream-returning Nexus operation's start to the stream's notifier, diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index e883093713..d72e559bb3 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/crates/sdk-core/CHANGELOG.md @@ -180,6 +180,11 @@ relevant information. newest record after the attach, so a caller that only waits for progress hears of the records already there. An empty stream needs nothing, since every append notifies. The API protos carry the `NexusHandler` progress fields and `nexus/notification-service.wit`. +* Experimental: the Nexus `read` takes `floor_holder`. With it the read holds the floor and + answers `replay`, the range of the records that crossed after the byte cut, with Core's digest + over them. The new Nexus `read_range` operation, `NexusReadRange` for lang, reads that range + again through the codec step and checks the digest. Its answer has no byte budget and carries + only the records and `next_token`. It checks that the chain it names is one of the Workflow's. * Experimental: a Workflow publishes stream records with `WorkflowOutputStreamCommit` in its activation completion. The Worker appends them to its `stream_store` before it reports the completion, so readers see them while the Workflow Task runs, and nothing goes to History. The diff --git a/crates/streams/nexus/temporal_streams.nexusrpc.yaml b/crates/streams/nexus/temporal_streams.nexusrpc.yaml index 9dcdbdd741..bab29e64c1 100644 --- a/crates/streams/nexus/temporal_streams.nexusrpc.yaml +++ b/crates/streams/nexus/temporal_streams.nexusrpc.yaml @@ -62,7 +62,9 @@ services: owner and has no byte budget, since it must answer exactly what the read answered. Records with other content are refused as StreamReplayMismatchError, and records the store no longer holds as - StreamReplayRangeLostError. + StreamReplayRangeLostError. The answer holds the records and + next_token only, since the caller keeps the read's state, done and + replay. input: { $ref: "#/$defs/ReadRangeInput" } output: { $ref: "#/$defs/ReadOutput" } diff --git a/crates/streams/src/handle.rs b/crates/streams/src/handle.rs index 8da58056fc..8578e7814f 100644 --- a/crates/streams/src/handle.rs +++ b/crates/streams/src/handle.rs @@ -462,7 +462,7 @@ impl Streams { /// Refuses a carried state whose chain is not one of the Workflow's. A caller outside the /// process could otherwise point a read at another run's chain by editing its state. Asked /// once per chain in a process. - async fn check_carried_chain( + pub(crate) async fn check_carried_chain( &self, stream: &StreamAddress, first_run_id: &str, diff --git a/crates/streams/src/nexus/mod.rs b/crates/streams/src/nexus/mod.rs index 8c5e2f71d9..d31e996e6d 100644 --- a/crates/streams/src/nexus/mod.rs +++ b/crates/streams/src/nexus/mod.rs @@ -1,10 +1,10 @@ //! The stream service's Nexus operations, served by Core. //! -//! Lang forwards each `append` and `read` call as N0a's JSON transfer form and gets back a -//! [NexusStep]. A call whose bodies need lang's payload codec answers first with [CodecWork], and -//! lang resumes it with the results. Core runs every rule of the contract, so each SDK's handler -//! keeps only its authorizer and its codec. The state a reader carries makes the endpoint -//! stateless, so a read goes on at any handler Worker. +//! Lang forwards each `append`, `read` and `read_range` call as N0a's JSON transfer form and gets +//! back a [NexusStep]. A call whose bodies need lang's payload codec answers first with +//! [CodecWork], and lang resumes it with the results. Core runs every rule of the contract, so +//! each SDK's handler keeps only its authorizer and its codec. The state a reader carries makes +//! the endpoint stateless, so a read goes on at any handler Worker. mod failure; mod operation; @@ -18,9 +18,9 @@ use crate::{ handle::CarriedRead, proto::{ AppendRecord, AppendRequest, CodecResult, CodecWork, LatestRequest, NamedProducer, - NexusCall, NexusFailure, NexusResumeRequest, NexusStep, ReadRecord, ReadRequest, - StreamAddress, StreamOwnerKind, StreamRecord, StreamRecordKind, append_request::Producer, - codec_result, nexus_step, read_record, + NexusCall, NexusFailure, NexusResumeRequest, NexusStep, ReadRangeRequest, ReadRecord, + ReadRequest, ReadResponse, StreamAddress, StreamOwnerKind, StreamRecord, StreamRecordKind, + append_request::Producer, codec_result, nexus_step, read_record, }, reader::{ReadState, records_digest}, }; @@ -37,8 +37,8 @@ use std::{ use temporalio_common::protos::temporal::api::common::v1::Payload; use tokio::time::Instant; use wire::{ - AppendInput, AppendOutput, Base64, ReadInput, ReadOutput, RecordWire, ReplayRange, StreamRef, - Supersession, crossing_size, json_size, + AppendInput, AppendOutput, Base64, ReadInput, ReadOutput, ReadRangeInput, RecordWire, + ReplayRange, StreamRef, Supersession, crossing_size, json_size, }; /// The most bytes one read answer takes as JSON past its first record, with its tokens and state. @@ -145,6 +145,12 @@ enum Work { /// The call named a floor holder, so the answer carries its replay range. positions: bool, }, + ReadRange { + response: Box, + limits: Limits, + /// As for [Work::Read]. + bodies: Vec, + }, } #[derive(Debug, Clone, Copy)] @@ -339,20 +345,9 @@ impl NexusCalls { )) }); let wait = shortened(Duration::from_millis(wait_ms as u64), deadline); - let state = match input.state { - None => Vec::new(), - Some(text) if text.len() > crossing_size(MAX_STATE_BYTES) => { - return failed(failure::of_stream(&StreamError::cursor(format!( - "the read state takes {} bytes, above the {} of any state a read answers, so \ - no earlier call of this read returned it", - text.len(), - crossing_size(MAX_STATE_BYTES) - )))); - } - Some(text) => match wire::base64_bytes(&text) { - Some(state) => state, - None => return failed(failure::input_mistake("state must be base64-encoded")), - }, + let state = match carried_state(input.state) { + Ok(state) => state, + Err(failure) => return failed(*failure), }; // A ref's run only picks the chain, so the read names the chain by its state and follows // it past that run's end. @@ -409,18 +404,7 @@ impl NexusCalls { } let cut = keep < read.response.records.len(); read.response.records.truncate(keep); - let mut bodies = Vec::new(); - let mut payloads = Vec::new(); - let mut bytes = 0; - for (index, record) in read.response.records.iter_mut().enumerate() { - if let Some(read_record::Record::Stored(stored)) = &mut record.record - && let Some(body) = stored.body.take() - { - bytes += body.encoded_len(); - bodies.push(index); - payloads.push(body); - } - } + let (bodies, payloads, bytes) = take_bodies(&mut read.response.records); let work = Work::Read { read: Box::new(read), limits, @@ -441,6 +425,84 @@ impl NexusCalls { ) } + /// Serves the `read_range` operation: the records of an earlier read that named a floor + /// holder, read again and checked against that read's digest. + /// + /// The answer has no byte budget, since it must hold exactly the records the read answered, + /// and those fit an answer once. It carries no state and no replay, because the caller keeps + /// the read's own. The chain comes from the caller, so it is checked to be one of the + /// Workflow's, as a carried read state is. + pub async fn read_range(&self, call: NexusCall) -> NexusStep { + let limits = Limits::of(&call); + let input: ReadRangeInput = match serde_json::from_slice(&call.input) { + Ok(input) => input, + Err(error) => return failed(failure::input_mistake(input_error("read_range", error))), + }; + let mut address = match address(&call.namespace, &input.stream) { + Ok(address) => address, + Err(failure) => return failed(*failure), + }; + // The chain names the run chain, so the ref's run picks nothing here. + address.run_id.clear(); + if !(0..=MAX_RECORDS).contains(&input.stored_records) { + return failed(failure::input_mistake(format!( + "stored_records is from 0 to {MAX_RECORDS}, not {}", + input.stored_records + ))); + } + let state = match carried_state(input.state) { + Ok(state) => state, + Err(failure) => return failed(*failure), + }; + if !input.chain_first_run_id.is_empty() + && let Err(error) = self + .streams + .check_carried_chain(&address, &input.chain_first_run_id) + .await + { + return failed(failure::of_stream(&error)); + } + if let Some(holder) = &input.floor_holder + && let Err(failure) = check_holder(holder, &input.after_token) + { + return failed(*failure); + } + let request = ReadRangeRequest { + stream: Some(address), + chain_first_run_id: input.chain_first_run_id, + after: input.after_token, + to: input.to_token, + stored_records: input.stored_records as u32, + state, + records_digest: input.records_digest.0, + floor_holder: input.floor_holder.unwrap_or_default(), + // A read that names a floor holder never starts at `$end`, since the read refuses it. + // The caller reads latest_only first and starts at its cursor. + start_from_end: false, + }; + let mut response = match self.streams.read_range(request).await { + Ok(response) => response, + Err(error) => return failed(failure::of_stream(&error)), + }; + let (bodies, payloads, bytes) = take_bodies(&mut response.records); + let work = Work::ReadRange { + response: Box::new(response), + limits, + bodies, + }; + if payloads.is_empty() { + return self.finish(work, Vec::new()).await; + } + self.ask_codec( + work, + bytes, + false, + &call.namespace, + &input.stream.workflow_id, + payloads, + ) + } + /// `address` without its run, and a state that names the chain the run picked. A state the /// caller passed back already names its chain. async fn chain_by_state( @@ -557,6 +619,11 @@ impl NexusCalls { cut, positions, } => finish_read(*read, limits, &bodies, cut, positions, results), + Work::ReadRange { + response, + limits, + bodies, + } => finish_read_range(*response, limits, &bodies, results), } } @@ -694,20 +761,10 @@ fn finish_read( positions: bool, results: Vec, ) -> NexusStep { - if results.len() != bodies.len() { - return failed(wrong_count(results.len(), bodies.len())); - } - let mut decoded: HashMap> = HashMap::new(); - for (index, result) in bodies.iter().zip(results) { - decoded.insert( - *index, - match result.result { - Some(codec_result::Result::Payload(body)) => Ok(body), - Some(codec_result::Result::Error(error)) => Err(error), - None => Err("the payload codec answered nothing".to_string()), - }, - ); - } + let mut decoded = match decoded(bodies, results) { + Ok(decoded) => decoded, + Err(failure) => return failed(*failure), + }; let mut budget = Budget::new(limits, &read); let CarriedRead { response, @@ -773,6 +830,85 @@ fn finish_read( }) } +/// A replay's answer. Each record crosses as the read's did, with no budget to cut it. +fn finish_read_range( + response: ReadResponse, + limits: Limits, + bodies: &[usize], + results: Vec, +) -> NexusStep { + let mut decoded = match decoded(bodies, results) { + Ok(decoded) => decoded, + Err(failure) => return failed(*failure), + }; + let records = response + .records + .into_iter() + .enumerate() + .filter_map(|(index, record)| match record.record { + Some(read_record::Record::Superseded(superseded)) => { + Some(superseded_wire(record.cursor, &superseded)) + } + Some(read_record::Record::Stored(stored)) => Some(record_wire( + record.cursor, + &stored, + decoded.remove(&index), + record.stale, + limits, + )), + None => None, + }) + .collect(); + output(&ReadOutput { + records, + next_token: response.cursor, + done: false, + state: None, + armed: false, + replay: None, + }) +} + +/// Takes the bodies out of `records` for lang's codec, answering which records had one, the +/// bodies, and their bytes. +fn take_bodies(records: &mut [ReadRecord]) -> (Vec, Vec, usize) { + let mut bodies = Vec::new(); + let mut payloads = Vec::new(); + let mut bytes = 0; + for (index, record) in records.iter_mut().enumerate() { + if let Some(read_record::Record::Stored(stored)) = &mut record.record + && let Some(body) = stored.body.take() + { + bytes += body.encoded_len(); + bodies.push(index); + payloads.push(body); + } + } + (bodies, payloads, bytes) +} + +/// Lang's codec results by the index of the record each body came from. +fn decoded( + bodies: &[usize], + results: Vec, +) -> Result>, Box> { + if results.len() != bodies.len() { + return Err(Box::new(wrong_count(results.len(), bodies.len()))); + } + Ok(bodies + .iter() + .zip(results) + .map(|(index, result)| { + let body = match result.result { + Some(codec_result::Result::Payload(body)) => Ok(body), + Some(codec_result::Result::Error(error)) => Err(error), + None => Err("the payload codec answered nothing".to_string()), + }; + (*index, body) + }) + .collect()) +} + fn stored_of(record: &ReadRecord) -> Option<&StreamRecord> { match &record.record { Some(read_record::Record::Stored(stored)) => Some(stored), @@ -866,6 +1002,24 @@ fn read_failure(error: StreamError, state: Vec, positions: bool) -> NexusSte } } +/// The state a call carries, decoded. One longer than any state a read answers is refused before +/// it is decoded. +fn carried_state(state: Option) -> Result, Box> { + match state { + None => Ok(Vec::new()), + Some(text) if text.len() > crossing_size(MAX_STATE_BYTES) => { + Err(Box::new(failure::of_stream(&StreamError::cursor(format!( + "the read state takes {} bytes, above the {} of any state a read answers, so no \ + earlier call of this read returned it", + text.len(), + crossing_size(MAX_STATE_BYTES) + ))))) + } + Some(text) => wire::base64_bytes(&text) + .ok_or_else(|| Box::new(failure::input_mistake("state must be base64-encoded"))), + } +} + /// The stream address a call names. An owner kind this release can't open is refused before any /// codec work. /// Checks a caller's floor holder. A Nexus caller holds as `//(rpc, request)?) .await .encode_to_vec()), + "NexusReadRange" => Ok(self + .nexus + .read_range(decode::(rpc, request)?) + .await + .encode_to_vec()), "NexusResume" => Ok(self .nexus .resume(decode::(rpc, request)?) @@ -1733,7 +1738,13 @@ mod tests { .unwrap(); let step = NexusStep::decode(answer.as_slice()).unwrap(); assert!(matches!(step.step, Some(nexus_step::Step::Failure(_)))); - let invalid = service.call("NexusRead", b"\xff").await.unwrap_err(); - assert_eq!(invalid.kind, StreamFailureKind::InvalidArgument as i32); + for rpc in ["NexusRead", "NexusReadRange"] { + let invalid = service.call(rpc, b"\xff").await.unwrap_err(); + assert_eq!( + invalid.kind, + StreamFailureKind::InvalidArgument as i32, + "{rpc}" + ); + } } } diff --git a/crates/streams/tests/nexus_contract.rs b/crates/streams/tests/nexus_contract.rs index ba849f8782..3e361d7c17 100644 --- a/crates/streams/tests/nexus_contract.rs +++ b/crates/streams/tests/nexus_contract.rs @@ -265,6 +265,69 @@ async fn every_declared_input_member_is_accepted_and_nothing_else() { } } +#[tokio::test] +async fn every_read_range_member_is_accepted_and_nothing_else() { + let contract = contract(); + let calls = service(memory(), NexusOptions::default()); + let wf = "contract-read-range"; + append( + &calls, + json!({"stream": stream(wf, "out"), "producer_id": "p", "attempt": 1, "sequence": 1, + "payloads": [body("a")]}), + ) + .await + .unwrap(); + let first = read( + &calls, + json!({"stream": stream(wf, "out"), "floor_holder": "ns/run-1/reader"}), + ) + .await + .unwrap(); + append( + &calls, + json!({"stream": stream(wf, "out"), "producer_id": "p", "attempt": 1, "sequence": 2, + "payloads": [body("b")]}), + ) + .await + .unwrap(); + let second = read( + &calls, + json!({"stream": stream(wf, "out"), "after_token": first["next_token"], + "state": first["state"], "floor_holder": "ns/run-1/reader"}), + ) + .await + .unwrap(); + let mut full = replay_input(stream(wf, "out"), &second, first["state"].clone()); + full["stream"]["chain_run_id"] = Value::Null; + full["floor_holder"] = json!("ns/run-1/reader"); + let mut declared = properties(model(&contract, "ReadRangeInput")); + let mut sent: Vec = full.as_object().unwrap().keys().cloned().collect(); + declared.sort(); + sent.sort(); + assert_eq!( + sent, declared, + "the contract changed ReadRangeInput; teach this test" + ); + let answer = read_range(&calls, full.clone()).await.unwrap(); + check_output(&contract, "ReadOutput", &answer); + assert_eq!(answer["records"], second["records"]); + let mut unknown = full.clone(); + unknown["surprise"] = json!(1); + let failure = read_range(&calls, unknown).await.unwrap_err(); + assert!(is_contract_mistake(&failure), "{}", failure.message); + assert_eq!(failure.handler_error_type, "BAD_REQUEST"); + for member in required(model(&contract, "ReadRangeInput")) { + let mut missing = full.clone(); + missing.as_object_mut().unwrap().remove(&member); + let failure = read_range(&calls, missing).await.unwrap_err(); + assert!( + is_contract_mistake(&failure), + "{member}: {}", + failure.message + ); + } +} + #[tokio::test] async fn every_output_member_is_declared() { let contract = contract(); diff --git a/crates/streams/tests/nexus_service.rs b/crates/streams/tests/nexus_service.rs index 3743f77c78..837e0ebf59 100644 --- a/crates/streams/tests/nexus_service.rs +++ b/crates/streams/tests/nexus_service.rs @@ -988,6 +988,31 @@ fn base64_of(bytes: &[u8]) -> String { base64::engine::general_purpose::STANDARD.encode(bytes) } +#[tokio::test] +async fn a_holding_read_from_the_end_starts_at_latest_only_and_replays() { + let (streams, calls) = layered(memory()); + let wf = workflow(); + write(&calls, &wf, 1, &["one"]).await; + let mut latest = read_input(&wf, ""); + latest.as_object_mut().unwrap().remove("after_token"); + latest["latest_only"] = json!(true); + let end = read(&calls, latest).await.unwrap(); + append(&calls, append_input(&wf, "p", 2, 1, &["two"])) + .await + .unwrap(); + let live = read(&calls, holding(&wf, token(&end))).await.unwrap(); + assert_eq!(values(&live), ["SUPERSEDED p 1->2", "p@2 two"]); + let replayed = streams + .read_range(replay_of(&wf, &live, &[])) + .await + .unwrap(); + assert_eq!( + replayed.records.len(), + 2, + "the replay sees what the live read saw" + ); +} + #[tokio::test] async fn a_range_replays_with_the_state_the_read_sent() { let (streams, calls) = layered(memory()); @@ -1094,3 +1119,159 @@ async fn a_floor_holder_is_at_most_512_bytes_and_never_beside_latest_only() { assert_eq!(kind_of(&refused), ("BAD_REQUEST", false, "ValueError")); assert!(refused.message.contains("floor"), "{}", refused.message); } + +/// A cut answer and the answer after it, with a supersession, each read again through +/// `read_range`, cross the same way the read's did. +async fn a_range_replays_as_it_crossed(store: Arc) { + let calls = service(store, NexusOptions::default()); + let wf = workflow(); + let big = "x".repeat(300 * 1024); + write(&calls, &wf, 1, &[&format!("1{big}"), &format!("2{big}")]).await; + append(&calls, append_input(&wf, "p", 2, 1, &[&format!("3{big}")])) + .await + .unwrap(); + let first = read(&calls, holding(&wf, "")).await.unwrap(); + assert_eq!(records(&first).len(), 2, "the budget cut the answer"); + let mut input = holding(&wf, token(&first)); + input["state"] = first["state"].clone(); + let second = read(&calls, input).await.unwrap(); + assert_eq!(values(&second)[0], "SUPERSEDED p 1->2"); + for (answer, state) in [(&first, Value::Null), (&second, first["state"].clone())] { + let replayed = read_range(&calls, replay_input(stream(&wf, TOPIC), answer, state)) + .await + .unwrap(); + assert_eq!(replayed["records"], answer["records"]); + assert_eq!(replayed["next_token"], answer["next_token"]); + assert_eq!(replayed["done"], false); + assert!(replayed.get("state").is_none() && replayed.get("replay").is_none()); + } +} + +#[tokio::test] +async fn a_range_replays_as_it_crossed_on_memory() { + a_range_replays_as_it_crossed(memory()).await; +} + +#[cfg(feature = "redis")] +#[tokio::test] +async fn a_range_replays_as_it_crossed_on_redis() { + let Some(store) = on_redis().await else { + return; + }; + a_range_replays_as_it_crossed(store).await; +} + +/// A held read of two records on `wf`, through `calls`. +async fn held_read(calls: &NexusCalls, wf: &str) -> Value { + write(calls, wf, 1, &["one", "two"]).await; + read(calls, holding(wf, "")).await.unwrap() +} + +#[tokio::test] +async fn a_range_with_other_records_is_a_mismatch() { + let calls = service(memory(), NexusOptions::default()); + let wf = workflow(); + let answer = held_read(&calls, &wf).await; + let mut other_digest = replay_input(stream(&wf, TOPIC), &answer, Value::Null); + other_digest["records_digest"] = json!(base64_of(&[0; 32])); + let mut other_count = replay_input(stream(&wf, TOPIC), &answer, Value::Null); + other_count["stored_records"] = json!(1); + for input in [other_digest, other_count] { + let refused = read_range(&calls, input).await.unwrap_err(); + assert_eq!( + kind_of(&refused), + ("BAD_REQUEST", false, "StreamReplayMismatchError") + ); + } +} + +#[tokio::test] +async fn a_range_the_store_dropped_is_lost_and_a_replay_holds_it_again() { + let memory = Arc::new(MemoryStore::new()); + let (streams, calls) = layered(memory.clone()); + let wf = workflow(); + let answer = held_read(&calls, &wf).await; + let release = |holder: &str| temporalio_streams::proto::ReleaseFloorRequest { + stream: replay_of(&wf, &answer, &[]).stream, + chain_first_run_id: "run-1".to_string(), + floor_holder: holder.to_string(), + }; + streams.release_floor(release(HOLDER)).await.unwrap(); + // A replay that names a holder holds the range again, as after a reset. + let mut renewing = replay_input(stream(&wf, TOPIC), &answer, Value::Null); + renewing["floor_holder"] = json!("ns/reset-run/1"); + read_range(&calls, renewing).await.unwrap(); + memory.truncate(&chain_of(&wf), TOPIC, 0); + let input = replay_input(stream(&wf, TOPIC), &answer, Value::Null); + let replayed = read_range(&calls, input.clone()).await.unwrap(); + assert_eq!(replayed["records"], answer["records"]); + streams + .release_floor(release("ns/reset-run/1")) + .await + .unwrap(); + memory.truncate(&chain_of(&wf), TOPIC, 0); + let refused = read_range(&calls, input).await.unwrap_err(); + assert_eq!( + kind_of(&refused), + ("BAD_REQUEST", false, "StreamReplayRangeLostError") + ); +} + +#[tokio::test] +async fn a_range_names_a_chain_of_the_workflow_and_a_count_a_read_can_answer() { + let calls = service(memory(), NexusOptions::default()); + let wf = workflow(); + let answer = held_read(&calls, &wf).await; + let mut other_chain = replay_input(stream(&wf, TOPIC), &answer, Value::Null); + other_chain["chain_first_run_id"] = json!("run-9"); + let refused = read_range(&calls, other_chain).await.unwrap_err(); + assert_eq!( + kind_of(&refused), + ("BAD_REQUEST", false, "StreamCursorError") + ); + for count in [-1, 1001] { + let mut input = replay_input(stream(&wf, TOPIC), &answer, Value::Null); + input["stored_records"] = json!(count); + let refused = read_range(&calls, input).await.unwrap_err(); + assert_eq!(kind_of(&refused), ("BAD_REQUEST", false, "ValueError")); + } +} + +/// A reader's first range starts at the stream's start, so it has no cursor to find gone. Once +/// its hold lapsed and retention trimmed the range, later records stand where it began. +#[cfg(feature = "redis")] +#[tokio::test] +async fn a_range_from_a_trimmed_start_is_lost_on_redis() { + let Ok(url) = std::env::var("STREAMS_REDIS_URL") else { + eprintln!("set STREAMS_REDIS_URL to run the Redis cases"); + return; + }; + let mut options = temporalio_streams::RedisStoreOptions::new(url); + options.key_prefix = workflow(); + options.retention = Duration::from_millis(300); + options.floor_lease = Some(Duration::from_millis(100)); + let store = Arc::new( + temporalio_streams::RedisStore::connect(options) + .await + .unwrap(), + ); + let calls = service(store, NexusOptions::default()); + let wf = workflow(); + let answer = held_read(&calls, &wf).await; + assert_eq!(answer["replay"]["after_token"], ""); + tokio::time::sleep(Duration::from_millis(600)).await; + write(&calls, &wf, 3, &["three"]).await; + write(&calls, &wf, 4, &["four"]).await; + let refused = read_range( + &calls, + replay_input(stream(&wf, TOPIC), &answer, Value::Null), + ) + .await + .unwrap_err(); + assert_eq!( + kind_of(&refused), + ("BAD_REQUEST", false, "StreamReplayRangeLostError"), + "{}", + refused.message + ); +} diff --git a/crates/streams/tests/nexus_support/mod.rs b/crates/streams/tests/nexus_support/mod.rs index 868fd4b0f6..091e57dfba 100644 --- a/crates/streams/tests/nexus_support/mod.rs +++ b/crates/streams/tests/nexus_support/mod.rs @@ -171,6 +171,24 @@ pub(crate) async fn read(calls: &NexusCalls, input: Value) -> Result Result { + drive(calls, calls.read_range(call(input)).await, Codec::Sealing).await +} + +/// The `read_range` input that replays `answer`, the answer of a read that sent `state`. +pub(crate) fn replay_input(stream: Value, answer: &Value, state: Value) -> Value { + let replay = &answer["replay"]; + serde_json::json!({ + "stream": stream, + "chain_first_run_id": replay["chain_first_run_id"], + "after_token": replay["after_token"], + "to_token": answer["next_token"], + "stored_records": replay["stored_records"], + "state": state, + "records_digest": replay["records_digest"], + }) +} + /// A stream reference to `topic` of Workflow `workflow_id`. pub(crate) fn stream(workflow_id: &str, topic: &str) -> Value { serde_json::json!({"kind": "workflow", "workflow_id": workflow_id, "topic": topic})