From a58f2375578b2348414a7eac0f8ec341616dd2c1 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Wed, 14 Oct 2026 09:10:11 -0700 Subject: [PATCH 1/4] Served `read_range` through the codec step. Replay of a positions read gets its records from the owner's endpoint. The answer has no byte budget, since it must give exactly what the read gave. --- .../temporal/sdk/core/streams/nexus.proto | 2 + .../nexus/temporal_streams.nexusrpc.yaml | 4 +- crates/streams/src/handle.rs | 2 +- crates/streams/src/nexus/mod.rs | 249 ++++++++++++++---- crates/streams/src/nexus/wire.rs | 7 - crates/streams/src/service.rs | 15 +- crates/streams/tests/nexus_contract.rs | 63 +++++ crates/streams/tests/nexus_service.rs | 117 ++++++++ crates/streams/tests/nexus_support/mod.rs | 18 ++ 9 files changed, 416 insertions(+), 61 deletions(-) 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/streams/nexus/temporal_streams.nexusrpc.yaml b/crates/streams/nexus/temporal_streams.nexusrpc.yaml index 99bc5e5a9a..9c05215b17 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 a844f0b17b..9a1cc47685 100644 --- a/crates/streams/src/handle.rs +++ b/crates/streams/src/handle.rs @@ -388,7 +388,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 2170e0b17d..2d86db4b8f 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)] @@ -336,20 +342,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. @@ -406,18 +401,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, @@ -438,6 +422,79 @@ 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)); + } + 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 the end, since only latest_only + // does, and it refuses a holder. + 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( @@ -554,6 +611,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), } } @@ -690,20 +752,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, @@ -769,6 +821,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), @@ -862,6 +993,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. fn address(namespace: &str, stream: &StreamRef) -> Result> { diff --git a/crates/streams/src/nexus/wire.rs b/crates/streams/src/nexus/wire.rs index 41895ae2a6..987336821a 100644 --- a/crates/streams/src/nexus/wire.rs +++ b/crates/streams/src/nexus/wire.rs @@ -104,13 +104,6 @@ pub(crate) struct ReadInput { #[derive(Debug, Clone, PartialEq, Eq, Deserialize)] #[serde(deny_unknown_fields)] -#[cfg_attr( - not(test), - expect( - dead_code, - reason = "the contract declares the model before Core serves it" - ) -)] pub(crate) struct ReadRangeInput { pub(crate) stream: StreamRef, pub(crate) chain_first_run_id: String, diff --git a/crates/streams/src/service.rs b/crates/streams/src/service.rs index 06be3617fa..06a960bcb6 100644 --- a/crates/streams/src/service.rs +++ b/crates/streams/src/service.rs @@ -238,6 +238,11 @@ impl StreamService { .read(decode::(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)?) @@ -1285,7 +1290,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 f313428853..f22b577dcc 100644 --- a/crates/streams/tests/nexus_contract.rs +++ b/crates/streams/tests/nexus_contract.rs @@ -253,6 +253,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 3297515c62..8296c4a4e0 100644 --- a/crates/streams/tests/nexus_service.rs +++ b/crates/streams/tests/nexus_service.rs @@ -1147,3 +1147,120 @@ 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")); + } +} diff --git a/crates/streams/tests/nexus_support/mod.rs b/crates/streams/tests/nexus_support/mod.rs index e805eb6c3e..bf60dfeb2f 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}) From 4c2c3b848750dd0fa35e9ab8772cee9fdcf44fe4 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Wed, 14 Oct 2026 09:12:50 -0700 Subject: [PATCH 2/4] Noted the Nexus floor holder and `read_range` in the changelog. --- crates/sdk-core/CHANGELOG.md | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index a97bb55d7f..64d58a5663 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/crates/sdk-core/CHANGELOG.md @@ -128,6 +128,11 @@ relevant information. newest record after the attach, or arms an empty stream, so a caller that only waits for progress hears of the records already there and of the first append. 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. ### Breaking Changes * Replay detects changes to the service or operation of scheduled Nexus operations, reporting From cd8f4cfaed6cd115ea395ac101cc29b3f8435f19 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Wed, 14 Oct 2026 09:41:10 -0700 Subject: [PATCH 3/4] Told a trimmed start apart from other records in a replay read. A range that began at the stream's start has no cursor to find gone. So a trim showed as a read that ended elsewhere, a mismatch, when it is a lost range. --- crates/streams/src/handle.rs | 41 +++++++++++++++++++++------ crates/streams/tests/nexus_service.rs | 39 +++++++++++++++++++++++++ 2 files changed, 72 insertions(+), 8 deletions(-) diff --git a/crates/streams/src/handle.rs b/crates/streams/src/handle.rs index 9a1cc47685..ed22fccece 100644 --- a/crates/streams/src/handle.rs +++ b/crates/streams/src/handle.rs @@ -525,15 +525,9 @@ impl Streams { request.after ))); } - if page.cursor != request.to { - return Err(mismatch(format!( - "the replayed read ends at {:?}, not at {:?}", - page.cursor, request.to - ))); - } - if page.records_digest != request.records_digest { + if page.cursor != request.to || page.records_digest != request.records_digest { // A read from the beginning has no position to check, so a trim since the read shows - // as other records. + // as other records, which end elsewhere or differ. if request.after.is_empty() && self .store @@ -546,6 +540,12 @@ impl Streams { .to_string(), )); } + if page.cursor != request.to { + return Err(mismatch(format!( + "the replayed read ends at {:?}, not at {:?}", + page.cursor, request.to + ))); + } return Err(mismatch( "the replayed records differ from the records the read got".to_string(), )); @@ -1829,6 +1829,31 @@ mod tests { assert!(error.message.starts_with("REC-01"), "{error}"); } + #[tokio::test] + async fn a_replay_read_from_a_trimmed_start_is_rec_01() { + let setup = setup(quick()); + setup.streams.append(append("p", 1, "1")).await.unwrap(); + setup.streams.append(append("p", 2, "2")).await.unwrap(); + let request = read_after("", &[], Duration::ZERO); + let page = setup.streams.read(request.clone()).await.unwrap(); + assert_eq!(page.start_cursor, ""); + // Later records stand where the range began, so the replay finds as many as it wants. + setup.store.truncate(&chain("run-1"), "out", 0); + for sequence in 3..=4 { + setup + .streams + .append(append("p", sequence, &sequence.to_string())) + .await + .unwrap(); + } + let error = setup + .streams + .read_range(replay_of(&request, &page)) + .await + .unwrap_err(); + assert_eq!(error.kind, StreamFailureKind::ReplayRangeLost, "{error}"); + } + #[tokio::test] async fn a_floor_keeps_what_a_positions_only_reader_read_until_it_is_released() { let setup = setup(quick()); diff --git a/crates/streams/tests/nexus_service.rs b/crates/streams/tests/nexus_service.rs index 8296c4a4e0..4462f3914f 100644 --- a/crates/streams/tests/nexus_service.rs +++ b/crates/streams/tests/nexus_service.rs @@ -1264,3 +1264,42 @@ async fn a_range_names_a_chain_of_the_workflow_and_a_count_a_read_can_answer() { 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 + ); +} From 9ed12d35b15b3fbbcc16009e0ef1f5157798ae6d Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 16 Oct 2026 06:34:24 -0700 Subject: [PATCH 4/4] Checked the floor holder of a Nexus replay read too. --- crates/streams/src/nexus/mod.rs | 9 +++++++-- crates/streams/tests/nexus_service.rs | 25 +++++++++++++++++++++++++ 2 files changed, 32 insertions(+), 2 deletions(-) diff --git a/crates/streams/src/nexus/mod.rs b/crates/streams/src/nexus/mod.rs index fbe1b24c39..d31e996e6d 100644 --- a/crates/streams/src/nexus/mod.rs +++ b/crates/streams/src/nexus/mod.rs @@ -462,6 +462,11 @@ impl NexusCalls { { 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, @@ -471,8 +476,8 @@ impl NexusCalls { 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 the end, since only latest_only - // does, and it refuses a holder. + // 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 { diff --git a/crates/streams/tests/nexus_service.rs b/crates/streams/tests/nexus_service.rs index 1bd5a32018..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());