diff --git a/crates/streams/nexus/temporal_streams.nexusrpc.yaml b/crates/streams/nexus/temporal_streams.nexusrpc.yaml index e4ed1e5811..141771bcf6 100644 --- a/crates/streams/nexus/temporal_streams.nexusrpc.yaml +++ b/crates/streams/nexus/temporal_streams.nexusrpc.yaml @@ -189,6 +189,13 @@ $defs: producer attempts the reader already saw, so its superseded and stale marks are exact across calls and across handler Workers. A state that doesn't parse is refused as a cursor mistake. + arm: + type: boolean + description: >- + For a reader that sleeps until its caller hears of progress. When + the call waits its whole wait_ms and collects nothing, the endpoint + arms the stream, so the next append notifies the stream's callers. + A reader with no callback leaves it out. required: [stream] additionalProperties: false @@ -280,6 +287,12 @@ $defs: also when the answer was cut short by its byte budget. It holds at most 16 KiB, which is about 22 KiB as base64, so it can ride in a Workflow's History. A longer state is refused. + armed: + type: boolean + description: >- + The call asked to arm, collected nothing and armed the stream, so + the caller waits for progress. Never true with records or with done. + An empty answer without it falls back to waiting for progress too. required: [records, next_token, done] additionalProperties: true diff --git a/crates/streams/src/nexus/mod.rs b/crates/streams/src/nexus/mod.rs index 8bb999b6c1..213f6a2f38 100644 --- a/crates/streams/src/nexus/mod.rs +++ b/crates/streams/src/nexus/mod.rs @@ -307,6 +307,7 @@ impl NexusCalls { next_token: latest.cursor, done: false, state: None, + armed: false, }), Err(error) => failed(failure::of_stream(&error)), }; @@ -640,6 +641,7 @@ impl Budget { next_token: read.response.cursor.clone(), done: false, state: Some(Base64::default()), + armed: false, }); let state = crossing_size(read.state.encoded_len().min(MAX_STATE_BYTES)); Self { @@ -714,6 +716,8 @@ fn finish_read( wires.push(wire); kept.extend(stored); } + // Core's read no longer arms, so the contract's `armed` stays false until it leaves the IDL. + let armed = false; let (next_token, done) = if cut { // 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()) { @@ -734,6 +738,7 @@ fn finish_read( next_token, done, state: Some(Base64(state.encode_to_vec())), + armed, }) } @@ -807,6 +812,7 @@ fn read_failure(error: StreamError, state: Vec) -> NexusStep { next_token: cursor, done: false, state: (!state.is_empty()).then_some(Base64(state)), + armed: false, }), _ => failed(failure::of_stream(&error)), } diff --git a/crates/streams/src/nexus/wire.rs b/crates/streams/src/nexus/wire.rs index bc19392526..dc216e2ad6 100644 --- a/crates/streams/src/nexus/wire.rs +++ b/crates/streams/src/nexus/wire.rs @@ -96,6 +96,8 @@ pub(crate) struct ReadInput { /// Left as base64, so an oversized state is refused before it is decoded. #[serde(default)] pub(crate) state: Option, + #[serde(default)] + pub(crate) arm: Option, } #[derive(Debug, Clone, PartialEq, Eq, Serialize)] @@ -105,6 +107,8 @@ pub(crate) struct ReadOutput { pub(crate) done: bool, #[serde(skip_serializing_if = "Option::is_none")] pub(crate) state: Option, + #[serde(skip_serializing_if = "std::ops::Not::not")] + pub(crate) armed: bool, } #[derive(Debug, Clone, PartialEq, Eq, Serialize)] diff --git a/crates/streams/tests/nexus_contract.rs b/crates/streams/tests/nexus_contract.rs index e3d44dfe96..62bf3ac462 100644 --- a/crates/streams/tests/nexus_contract.rs +++ b/crates/streams/tests/nexus_contract.rs @@ -58,6 +58,8 @@ fn full_input(contract: &Value, name: &str, workflow_id: &str) -> Value { ("ReadInput", "wait_ms") => json!(0), ("ReadInput", "latest_only") => json!(false), ("ReadInput", "state") => Value::Null, + // So the answer carries `armed`, which the output check holds to the contract. + ("ReadInput", "arm") => json!(true), (model, member) => panic!("the contract added {model}.{member}; teach this test"), }; out.insert(member, value); diff --git a/crates/streams/tests/nexus_service.rs b/crates/streams/tests/nexus_service.rs index 863e0364f3..d7434c3b5a 100644 --- a/crates/streams/tests/nexus_service.rs +++ b/crates/streams/tests/nexus_service.rs @@ -699,6 +699,20 @@ mod cases { assert!(small.get("error").is_none()); assert_eq!(texts(&answer), ["small"]); } + fn arming(wf: &str, after: &str) -> Value { + let mut input = read_input(wf, after); + input["arm"] = json!(true); + input + } + + pub(crate) async fn a_read_that_asks_to_arm_is_never_armed(store: Arc) { + let calls = service(store, NexusOptions::default()); + let wf = workflow(); + let answer = read(&calls, arming(&wf, "")).await.unwrap(); + assert!(records(&answer).is_empty()); + assert!(answer.get("armed").is_none(), "{answer}"); + assert_eq!(answer["done"], false); + } } macro_rules! lanes { @@ -738,6 +752,7 @@ macro_rules! all_cases { a_record_above_the_budget_still_crosses_alone, an_append_above_the_record_limit_is_refused, a_stored_record_above_the_limit_crosses_as_an_error, + a_read_that_asks_to_arm_is_never_armed, ); }; }