diff --git a/crates/streams/src/nexus/mod.rs b/crates/streams/src/nexus/mod.rs index be24566cc..8c5e2f71d 100644 --- a/crates/streams/src/nexus/mod.rs +++ b/crates/streams/src/nexus/mod.rs @@ -22,7 +22,7 @@ use crate::{ StreamAddress, StreamOwnerKind, StreamRecord, StreamRecordKind, append_request::Producer, codec_result, nexus_step, read_record, }, - reader::ReadState, + reader::{ReadState, records_digest}, }; use prost::Message; use sha2::{Digest, Sha256}; @@ -37,13 +37,14 @@ use std::{ use temporalio_common::protos::temporal::api::common::v1::Payload; use tokio::time::Instant; use wire::{ - AppendInput, AppendOutput, Base64, ReadInput, ReadOutput, RecordWire, StreamRef, Supersession, - crossing_size, json_size, + AppendInput, AppendOutput, Base64, ReadInput, ReadOutput, 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. -/// A read answer is a sync Nexus result recorded in the caller's History, so it stays well below -/// the server's blob size limit (2 MiB by default). +/// A Workflow caller records a read answer in its History, unless its read names a floor holder +/// and it keeps only the range. So an answer stays well below the server's blob size limit +/// (2 MiB by default). pub const DEFAULT_MAX_ANSWER_BYTES: u64 = 1 << 20; /// The most bytes one record may take in an answer. One record must fit an answer on its own, so /// a larger one can never be read through the service. @@ -141,6 +142,8 @@ enum Work { bodies: Vec, /// The answer's byte budget left out records of the page before the codec step. cut: bool, + /// The call named a floor holder, so the answer carries its replay range. + positions: bool, }, } @@ -269,7 +272,8 @@ impl NexusCalls { /// The answer carries the read's state as it stood after the last record kept, also when the /// byte budget cut it short, so a reader's marks stay exact across calls and Workers. A record /// the store's reader can't parse crosses as an error at its cursor, which the reader goes on - /// past. + /// past. A call with a floor holder holds the records after its start for that holder, and its + /// answer carries the range of the records that crossed, so replay can read them again. pub async fn read(&self, call: NexusCall) -> NexusStep { let limits = Limits::of(&call); let input: ReadInput = match serde_json::from_slice(&call.input) { @@ -293,12 +297,22 @@ impl NexusCalls { ))); } let after = input.after_token.unwrap_or_default(); + let floor_holder = input.floor_holder.unwrap_or_default(); + if let Err(failure) = check_holder(&floor_holder, &after) { + return failed(*failure); + } if input.latest_only.unwrap_or(false) { if !after.is_empty() { return failed(failure::input_mistake( "latest_only starts a read at the newest record; it takes no after_token", )); } + if !floor_holder.is_empty() { + return failed(failure::input_mistake( + "latest_only answers no records, so it holds no floor; start the read that \ + holds one at the cursor latest_only answers", + )); + } return match self .streams .latest(LatestRequest { @@ -352,9 +366,10 @@ impl NexusCalls { wait: wait.try_into().ok(), max_records: max_records as u32, state: state.clone(), - floor_holder: String::new(), + floor_holder: floor_holder.clone(), subscriber: None, }; + let positions = !floor_holder.is_empty(); let mut read = match self.streams.read_carried(request).await { Ok(read) => read, Err(error) => { @@ -363,7 +378,7 @@ impl NexusCalls { } else { state }; - return read_failure(error, state); + return read_failure(error, state, positions); } }; // Cut by the stored bodies first, so the codec decodes only what can cross. A codec that @@ -411,6 +426,7 @@ impl NexusCalls { limits, bodies, cut, + positions, }; if payloads.is_empty() { return self.finish(work, Vec::new()).await; @@ -539,7 +555,8 @@ impl NexusCalls { limits, bodies, cut, - } => finish_read(*read, limits, &bodies, cut, results), + positions, + } => finish_read(*read, limits, &bodies, cut, positions, results), } } @@ -674,6 +691,7 @@ fn finish_read( limits: Limits, bodies: &[usize], cut: bool, + positions: bool, results: Vec, ) -> NexusStep { if results.len() != bodies.len() { @@ -698,29 +716,28 @@ fn finish_read( } = read; let mut cut = cut; let mut wires: Vec = Vec::with_capacity(response.records.len()); - let mut kept: Vec = Vec::new(); + // The records that crossed, without their bodies, for the state at a cut and the digest. + let mut kept: Vec = Vec::with_capacity(response.records.len()); for (index, record) in response.records.into_iter().enumerate() { - let ReadRecord { - cursor, - record, - stale, - } = record; - let (wire, stored) = match record { + let wire = match &record.record { Some(read_record::Record::Superseded(superseded)) => { - (superseded_wire(cursor, &superseded), None) - } - Some(read_record::Record::Stored(stored)) => { - let wire = record_wire(cursor, &stored, decoded.remove(&index), stale, limits); - (wire, Some(stored)) + superseded_wire(record.cursor.clone(), superseded) } + Some(read_record::Record::Stored(stored)) => record_wire( + record.cursor.clone(), + stored, + decoded.remove(&index), + record.stale, + limits, + ), None => continue, }; - if !budget.admits(json_size(&wire), stored.is_some()) { + if !budget.admits(json_size(&wire), wire.superseded.is_none()) { cut = true; break; } wires.push(wire); - kept.extend(stored); + kept.push(record); } // Core's read no longer arms, so the contract's `armed` stays false until it leaves the IDL. let armed = false; @@ -728,8 +745,9 @@ fn finish_read( // A supersession belongs to the record after it, so it goes into the next answer too. if wires.last().is_some_and(|wire| wire.superseded.is_some()) { wires.pop(); + kept.pop(); } - state = state.at_cut(&before, &kept); + state = state.at_cut(&before, kept.iter().filter_map(stored_of)); let last = wires .last() .map(|wire| wire.token.clone()) @@ -739,16 +757,29 @@ fn finish_read( (response.cursor, response.done) }; state.fit_within(MAX_STATE_BYTES); + let replay = positions.then(|| ReplayRange { + chain_first_run_id: response.chain_first_run_id, + after_token: response.start_cursor, + stored_records: kept.iter().filter_map(stored_of).count() as i64, + records_digest: Base64(records_digest(&kept)), + }); output(&ReadOutput { records: wires, next_token, done, state: Some(Base64(state.encode_to_vec())), armed, - replay: None, + replay, }) } +fn stored_of(record: &ReadRecord) -> Option<&StreamRecord> { + match &record.record { + Some(read_record::Record::Stored(stored)) => Some(stored), + _ => None, + } +} + fn superseded_wire(token: String, superseded: &crate::proto::Supersession) -> RecordWire { RecordWire { token, @@ -808,19 +839,28 @@ fn record_wire( } /// How a read's failure crosses. A record the store's reader can't parse ends no read: it -/// crosses as an error at its cursor, and the reader goes on past it with the state it had. -fn read_failure(error: StreamError, state: Vec) -> NexusStep { +/// crosses as an error at its cursor, and the reader goes on past it with the state it had. With +/// `positions` the answer's range is empty at that cursor, since a replay can't read the record +/// either, and the reader keeps the error itself. +fn read_failure(error: StreamError, state: Vec, positions: bool) -> NexusStep { match (error.kind, error.cursor.clone()) { (crate::proto::StreamFailureKind::Record, Some(cursor)) => output(&ReadOutput { records: vec![RecordWire { error: Some(failure::capped(&error.message)), ..stored_wire(cursor.clone(), Base64::default(), false) }], - next_token: cursor, + next_token: cursor.clone(), done: false, + replay: positions.then(|| ReplayRange { + chain_first_run_id: ReadState::decode_from(&state) + .map(|state| state.first_run_id) + .unwrap_or_default(), + after_token: cursor, + stored_records: 0, + records_digest: Base64(records_digest(&[])), + }), state: (!state.is_empty()).then_some(Base64(state)), armed: false, - replay: None, }), _ => failed(failure::of_stream(&error)), } @@ -828,6 +868,29 @@ fn read_failure(error: StreamError, state: Vec) -> NexusStep { /// 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 `//`, so callers in other namespaces can't share or release a holder. A positions-only read +/// can't start at `$end`, since its replay would look up the record there, which the live read +/// did not. The caller reads `latest_only` first and starts at the cursor it answers. +fn check_holder(floor_holder: &str, after: &str) -> Result<(), Box> { + if floor_holder.is_empty() { + return Ok(()); + } + let parts: Vec<&str> = floor_holder.splitn(3, '/').collect(); + if parts.len() != 3 || parts.iter().any(|part| part.is_empty()) { + return Err(Box::new(failure::input_mistake( + "a floor holder is //", + ))); + } + if after == crate::END { + return Err(Box::new(failure::input_mistake( + "a read that holds a floor starts at a cursor, not at $end; read latest_only first \ + and start at the cursor it answers", + ))); + } + Ok(()) +} + fn address(namespace: &str, stream: &StreamRef) -> Result> { if stream.kind != "workflow" { return Err(Box::new(failure::of_stream(&StreamError::unsupported( diff --git a/crates/streams/tests/nexus_contract.rs b/crates/streams/tests/nexus_contract.rs index 0c584bede..ba849f878 100644 --- a/crates/streams/tests/nexus_contract.rs +++ b/crates/streams/tests/nexus_contract.rs @@ -341,6 +341,11 @@ async fn every_output_member_is_declared() { "no answer carried {member}" ); } + let mut holding = json!({"stream": stream(wf, "out")}); + holding["floor_holder"] = json!("ns/run-1/reader"); + let held = read(&calls, holding).await.unwrap(); + check_output(&contract, "ReadOutput", &held); + check_output(&contract, "ReplayRange", &held["replay"]); let latest = read( &calls, json!({"stream": stream(wf, "out"), "latest_only": true}), diff --git a/crates/streams/tests/nexus_service.rs b/crates/streams/tests/nexus_service.rs index d7434c3b5..3743f77c7 100644 --- a/crates/streams/tests/nexus_service.rs +++ b/crates/streams/tests/nexus_service.rs @@ -22,7 +22,10 @@ use temporalio_common::protos::temporal::api::{ use temporalio_streams::{ MemoryStore, OwnerClient, OwnerDescription, OwnerError, StreamStore, Streams, StreamsOptions, nexus::{NexusCalls, NexusOptions}, - proto::{ChainId, NexusFailure, StreamRecord, StreamRecordKind}, + proto::{ + ChainId, NexusFailure, ReadRangeRequest, ReadRequest, StoreAppendRequest, StreamAddress, + StreamOwnerKind, StreamRecord, StreamRecordKind, + }, }; static IDS: AtomicUsize = AtomicUsize::new(0); @@ -807,3 +810,287 @@ async fn an_expired_cursor_is_told_apart() { ("BAD_REQUEST", false, "StreamExpiredError") ); } + +const HOLDER: &str = "ns/reader-run/1"; + +fn holding(wf: &str, after: &str) -> Value { + let mut input = read_input(wf, after); + input["floor_holder"] = json!(HOLDER); + input +} + +fn chain_of(wf: &str) -> ChainId { + ChainId { + namespace: NAMESPACE.to_string(), + workflow_id: wf.to_string(), + first_run_id: "run-1".to_string(), + ..Default::default() + } +} + +/// The Nexus calls over `store`, and the stream layer they share. +fn layered(store: Arc) -> (Arc, NexusCalls) { + let streams = Arc::new(Streams::new( + store, + Arc::new(Running), + StreamsOptions::default(), + )); + let calls = NexusCalls::new(streams.clone(), NexusOptions::default()); + (streams, calls) +} + +/// Core's replay read of the range `answer` carries, for the read that sent `state`. +fn replay_of(wf: &str, answer: &Value, state: &[u8]) -> ReadRangeRequest { + use base64::Engine; + let replay = &answer["replay"]; + ReadRangeRequest { + stream: Some(StreamAddress { + namespace: NAMESPACE.to_string(), + owner_kind: StreamOwnerKind::Workflow as i32, + workflow_id: wf.to_string(), + run_id: String::new(), + topic: TOPIC.to_string(), + ..Default::default() + }), + chain_first_run_id: replay["chain_first_run_id"].as_str().unwrap().to_string(), + after: replay["after_token"].as_str().unwrap().to_string(), + to: token(answer).to_string(), + stored_records: replay["stored_records"].as_u64().unwrap() as u32, + state: state.to_vec(), + records_digest: base64::engine::general_purpose::STANDARD + .decode(replay["records_digest"].as_str().unwrap()) + .unwrap(), + floor_holder: String::new(), + start_from_end: false, + } +} + +#[tokio::test] +async fn a_read_without_a_floor_holder_answers_no_range() { + let calls = service(memory(), NexusOptions::default()); + let wf = workflow(); + write(&calls, &wf, 1, &["one"]).await; + let answer = read_after(&calls, &wf, "").await; + assert!(answer.get("replay").is_none(), "{answer}"); +} + +#[tokio::test] +async fn a_read_with_a_floor_holder_holds_what_it_read_against_retention() { + let memory = Arc::new(MemoryStore::new()); + let calls = service(memory.clone(), NexusOptions::default()); + let held = workflow(); + let free = workflow(); + let mut starts = Vec::new(); + for wf in [&held, &free] { + let first = write(&calls, wf, 1, &["one"]).await; + write(&calls, wf, 2, &["two", "three"]).await; + starts.push(first); + } + let answer = read(&calls, holding(&held, &starts[0])).await.unwrap(); + assert_eq!(texts(&answer), ["two", "three"]); + let replay = &answer["replay"]; + assert_eq!(replay["chain_first_run_id"], "run-1"); + assert_eq!(replay["after_token"], starts[0].as_str()); + assert_eq!(replay["stored_records"], 2); + read_after(&calls, &free, &starts[1]).await; + for wf in [&held, &free] { + memory.truncate(&chain_of(wf), TOPIC, 0); + } + assert_eq!( + texts(&read_after(&calls, &held, &starts[0]).await), + ["two", "three"], + "the floor kept the records after the read's start" + ); + let refused = read(&calls, read_input(&free, &starts[1])) + .await + .unwrap_err(); + assert_eq!( + kind_of(&refused), + ("BAD_REQUEST", false, "StreamExpiredError") + ); +} + +#[cfg(feature = "redis")] +#[tokio::test] +async fn a_read_with_a_floor_holder_holds_what_it_read_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(200); + options.floor_lease = Some(Duration::from_secs(60)); + let store = Arc::new( + temporalio_streams::RedisStore::connect(options) + .await + .unwrap(), + ); + let calls = service(store, NexusOptions::default()); + let wf = workflow(); + let first = write(&calls, &wf, 1, &["one"]).await; + write(&calls, &wf, 2, &["two", "three"]).await; + let answer = read(&calls, holding(&wf, &first)).await.unwrap(); + assert_eq!(answer["replay"]["stored_records"], 2); + // Past the retention, an append's trim would drop the records without the floor. + tokio::time::sleep(Duration::from_millis(400)).await; + write(&calls, &wf, 4, &["four"]).await; + assert_eq!( + texts(&read_after(&calls, &wf, &first).await), + ["two", "three", "four"] + ); +} + +#[tokio::test] +async fn a_cut_answer_ranges_over_the_records_that_crossed() { + let (streams, calls) = layered(memory()); + let wf = workflow(); + let big = "x".repeat(300 * 1024); + write(&calls, &wf, 1, &[&format!("1{big}"), &format!("2{big}")]).await; + // The newer attempt's supersession fits the budget, but its record doesn't, so the cut + // leaves the supersession for the next answer. + append(&calls, append_input(&wf, "p", 2, 1, &[&format!("3{big}")])) + .await + .unwrap(); + let answer = read(&calls, holding(&wf, "")).await.unwrap(); + assert_eq!(records(&answer).len(), 2, "the budget cut the answer"); + assert!( + records(&answer) + .iter() + .all(|r| r.get("superseded").is_none()) + ); + let replay = &answer["replay"]; + assert_eq!(replay["after_token"], ""); + assert_eq!(replay["stored_records"], 2); + let replayed = streams + .read_range(replay_of(&wf, &answer, &[])) + .await + .unwrap(); + assert_eq!(replayed.records.len(), 2); + assert_eq!(replayed.cursor, token(&answer)); + // Core's page went on past the cut, so its digest is not the answer's. + let page = streams + .read(ReadRequest { + stream: replay_of(&wf, &answer, &[]).stream, + ..Default::default() + }) + .await + .unwrap(); + assert_eq!(page.records.len(), 4); + assert_ne!( + json!(base64_of(&page.records_digest)), + replay["records_digest"] + ); +} + +fn base64_of(bytes: &[u8]) -> String { + use base64::Engine; + base64::engine::general_purpose::STANDARD.encode(bytes) +} + +#[tokio::test] +async fn a_range_replays_with_the_state_the_read_sent() { + let (streams, calls) = layered(memory()); + let wf = workflow(); + write(&calls, &wf, 1, &["one"]).await; + let first = read(&calls, holding(&wf, "")).await.unwrap(); + append(&calls, append_input(&wf, "p", 2, 1, &["two"])) + .await + .unwrap(); + let mut input = holding(&wf, token(&first)); + input["state"] = first["state"].clone(); + let second = read(&calls, input).await.unwrap(); + assert_eq!(values(&second), ["SUPERSEDED p 1->2", "p@2 two"]); + let state = { + use base64::Engine; + base64::engine::general_purpose::STANDARD + .decode(first["state"].as_str().unwrap()) + .unwrap() + }; + let replayed = streams + .read_range(replay_of(&wf, &second, &state)) + .await + .unwrap(); + assert_eq!( + replayed.records.len(), + 2, + "the supersession comes out again" + ); +} + +#[tokio::test] +async fn a_record_the_reader_cannot_parse_has_an_empty_range_at_its_cursor() { + let memory = Arc::new(MemoryStore::new()); + let (streams, calls) = layered(memory.clone()); + let wf = workflow(); + memory + .append(StoreAppendRequest { + chain: Some(chain_of(&wf)), + topic: TOPIC.to_string(), + producer_id: "raw".to_string(), + attempt: 1, + sequence: 1, + digest: vec![1; 32], + records: vec![b"\xff".to_vec()], + }) + .await + .unwrap(); + let answer = read(&calls, holding(&wf, "")).await.unwrap(); + let [wire] = records(&answer).as_slice() else { + panic!("expected one record, got {answer}"); + }; + assert!(wire.get("error").is_some(), "{answer}"); + let replay = &answer["replay"]; + assert_eq!(replay["after_token"], token(&answer)); + assert_eq!(replay["stored_records"], 0); + assert_eq!(replay["chain_first_run_id"], "run-1"); + let replayed = streams + .read_range(replay_of(&wf, &answer, &[])) + .await + .unwrap(); + assert!(replayed.records.is_empty()); +} + +#[tokio::test] +async fn a_floor_holder_names_the_caller_and_starts_at_a_cursor() { + let calls = service(memory(), NexusOptions::default()); + let wf = workflow(); + write(&calls, &wf, 1, &["one"]).await; + for holder in ["reader-run/1", "ns//1", "archive"] { + let mut input = read_input(&wf, ""); + input["floor_holder"] = json!(holder); + let refused = read(&calls, input).await.unwrap_err(); + assert_eq!( + kind_of(&refused), + ("BAD_REQUEST", false, "ValueError"), + "{holder}" + ); + } + let refused = read(&calls, holding(&wf, "$end")).await.unwrap_err(); + assert_eq!(kind_of(&refused), ("BAD_REQUEST", false, "ValueError")); + assert!( + refused.message.contains("latest_only"), + "{}", + refused.message + ); +} + +#[tokio::test] +async fn a_floor_holder_is_at_most_512_bytes_and_never_beside_latest_only() { + let calls = service(memory(), NexusOptions::default()); + let wf = workflow(); + write(&calls, &wf, 1, &["one"]).await; + let mut input = read_input(&wf, ""); + input["floor_holder"] = json!(format!("ns/run/{}", "h".repeat(505))); + read(&calls, input.clone()).await.unwrap(); + input["floor_holder"] = json!(format!("ns/run/{}", "h".repeat(506))); + let refused = read(&calls, input).await.unwrap_err(); + assert_eq!(kind_of(&refused), ("BAD_REQUEST", false, "ValueError")); + assert!(refused.message.contains("512"), "{}", refused.message); + let mut latest = holding(&wf, ""); + latest.as_object_mut().unwrap().remove("after_token"); + latest["latest_only"] = json!(true); + let refused = read(&calls, latest).await.unwrap_err(); + assert_eq!(kind_of(&refused), ("BAD_REQUEST", false, "ValueError")); + assert!(refused.message.contains("floor"), "{}", refused.message); +}