diff --git a/crates/streams/nexus/temporal_streams.nexusrpc.yaml b/crates/streams/nexus/temporal_streams.nexusrpc.yaml index bab29e64c1..a62083d6b6 100644 --- a/crates/streams/nexus/temporal_streams.nexusrpc.yaml +++ b/crates/streams/nexus/temporal_streams.nexusrpc.yaml @@ -205,13 +205,6 @@ $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. floor_holder: type: string description: >- @@ -354,12 +347,6 @@ $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. replay: $ref: "#/$defs/ReplayRange" description: >- diff --git a/crates/streams/src/nexus/mod.rs b/crates/streams/src/nexus/mod.rs index d31e996e6d..482d32107f 100644 --- a/crates/streams/src/nexus/mod.rs +++ b/crates/streams/src/nexus/mod.rs @@ -331,7 +331,6 @@ impl NexusCalls { next_token: latest.cursor, done: false, state: None, - armed: false, replay: None, }), Err(error) => failed(failure::of_stream(&error)), @@ -730,7 +729,6 @@ impl Budget { next_token: read.response.cursor.clone(), done: false, state: Some(Base64::default()), - armed: false, replay: None, }); let state = crossing_size(read.state.encoded_len().min(MAX_STATE_BYTES)); @@ -796,8 +794,6 @@ fn finish_read( wires.push(wire); kept.push(record); } - // 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()) { @@ -825,7 +821,6 @@ fn finish_read( next_token, done, state: Some(Base64(state.encode_to_vec())), - armed, replay, }) } @@ -864,7 +859,6 @@ fn finish_read_range( next_token: response.cursor, done: false, state: None, - armed: false, replay: None, }) } @@ -996,7 +990,6 @@ fn read_failure(error: StreamError, state: Vec, positions: bool) -> NexusSte records_digest: Base64(records_digest(&[])), }), 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 987336821a..b390a7714c 100644 --- a/crates/streams/src/nexus/wire.rs +++ b/crates/streams/src/nexus/wire.rs @@ -97,8 +97,6 @@ pub(crate) struct ReadInput { #[serde(default)] pub(crate) state: Option, #[serde(default)] - pub(crate) arm: Option, - #[serde(default)] pub(crate) floor_holder: Option, } @@ -126,8 +124,6 @@ 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, #[serde(skip_serializing_if = "Option::is_none")] pub(crate) replay: Option, } @@ -276,7 +272,6 @@ mod tests { next_token: "c".to_string(), done: false, state: None, - armed: false, replay: Some(ReplayRange { chain_first_run_id: "run-1".to_string(), after_token: String::new(), diff --git a/crates/streams/tests/nexus_contract.rs b/crates/streams/tests/nexus_contract.rs index 3e361d7c17..0332745df9 100644 --- a/crates/streams/tests/nexus_contract.rs +++ b/crates/streams/tests/nexus_contract.rs @@ -58,8 +58,6 @@ 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), ("ReadInput", "floor_holder") => json!("ns/run-1/reader"), (model, member) => panic!("the contract added {model}.{member}; teach this test"), }; diff --git a/crates/streams/tests/nexus_service.rs b/crates/streams/tests/nexus_service.rs index 837e0ebf59..3bde7adb61 100644 --- a/crates/streams/tests/nexus_service.rs +++ b/crates/streams/tests/nexus_service.rs @@ -702,19 +702,13 @@ 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) { + pub(crate) async fn a_read_that_asks_to_arm_is_refused(store: Arc) { + // The contract has no arm, and a read refuses members it doesn't know. 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); + let mut input = read_input(&workflow(), ""); + input["arm"] = json!(true); + let refused = read(&calls, input).await.unwrap_err(); + assert_eq!(kind_of(&refused).0, "BAD_REQUEST"); } } @@ -755,7 +749,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, + a_read_that_asks_to_arm_is_refused, ); }; }