From b10ac0531265c32c0ccd806aab06ff3d40191516 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Thu, 15 Oct 2026 00:27:15 -0700 Subject: [PATCH] Served streams that a standalone Activity owns over Nexus. Every Bridge path keys on Core's address and chain, which already take the Activity kind. A wake token now matches its read on the whole owner, since every Activity ref has the same empty Workflow id. --- crates/sdk-core/CHANGELOG.md | 7 + crates/streams/src/handle.rs | 4 +- crates/streams/src/nexus/mod.rs | 81 ++++----- crates/streams/src/nexus/operation.rs | 69 +++++++- crates/streams/tests/nexus_service.rs | 200 +++++++++++++++++++++- crates/streams/tests/nexus_support/mod.rs | 10 ++ 6 files changed, 316 insertions(+), 55 deletions(-) diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index 30aa13ff92..8f671cc9e5 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/crates/sdk-core/CHANGELOG.md @@ -211,6 +211,13 @@ relevant information. Activity id, keeps version 2, and leaves `activity_id` out of a Workflow's ref, so a Workflow's token keeps its bytes. `operation_token_vectors.json` has Activity tokens. `CodecWork` gains `owner_kind` and `activity_id`. The Nexus calls still refuse an Activity ref. +* Experimental: the stream service's Nexus calls serve streams that a standalone Activity owns. + `append`, `read`, `read_range`, `latest_only`, the floor holder, `AttachCaller` with + `wake_after_counter`, `DetachCaller` and the read-first wake all take an Activity ref. A read's + `wake_token` must name the read's owner kind, Workflow id, Activity id and topic, so a token of + one Activity never wakes another Activity's reader on the same topic. `CodecWork` names the + Activity, so lang's codec context is the namespace and the Activity id. A ref of kind + `activity` with a Workflow id, or of kind `workflow` with an Activity id, is a bad request. ### Breaking Changes * Replay detects changes to the service or operation of scheduled Nexus operations, reporting diff --git a/crates/streams/src/handle.rs b/crates/streams/src/handle.rs index 9aabe6f9ff..014d25ebd7 100644 --- a/crates/streams/src/handle.rs +++ b/crates/streams/src/handle.rs @@ -447,8 +447,8 @@ impl Streams { Ok(described) if described.first_run_id != first_run_id => { return Err(StreamError::cursor(format!( "the read state names run {first_run_id:?}, which doesn't start a run chain \ - of Workflow {:?}", - stream.workflow_id + of {}", + key.label() ))); } // A chain Temporal no longer holds leaves nothing another chain's reader could diff --git a/crates/streams/src/nexus/mod.rs b/crates/streams/src/nexus/mod.rs index 33f5899823..d31b22aa80 100644 --- a/crates/streams/src/nexus/mod.rs +++ b/crates/streams/src/nexus/mod.rs @@ -272,14 +272,7 @@ impl NexusCalls { if finish { return self.finish(work, Vec::new()).await; } - self.ask_codec( - work, - bytes, - true, - &call.namespace, - &input.stream.workflow_id, - parsed, - ) + self.ask_codec(work, bytes, true, &call.namespace, &input.stream, parsed) } /// Serves the `read` operation: the records after the caller's cursor, waiting for some. @@ -315,8 +308,12 @@ impl NexusCalls { let floor_holder = input.floor_holder.unwrap_or_default(); let wake = match &input.wake_token { Some(token) => match OperationToken::decode(&call.namespace, token) { + // A token of another owner with the same topic must not wake this reader, and + // every Activity's ref has the same empty Workflow id. Ok(token) - if token.stream.workflow_id == address.workflow_id + if token.stream.owner_kind == address.owner_kind + && token.stream.workflow_id == address.workflow_id + && token.stream.activity_id == address.activity_id && token.stream.topic == address.topic => { Some(token) @@ -449,14 +446,7 @@ impl NexusCalls { if payloads.is_empty() { return self.finish(work, Vec::new()).await; } - self.ask_codec( - work, - bytes, - false, - &call.namespace, - &input.stream.workflow_id, - payloads, - ) + self.ask_codec(work, bytes, false, &call.namespace, &input.stream, payloads) } /// Asks the notifier to wake the operation callback that `token` names once the stream moves @@ -476,12 +466,7 @@ impl NexusCalls { })?; client .wait_callback( - crate::proto::ChainId { - namespace: address.namespace.clone(), - workflow_id: address.workflow_id.clone(), - first_run_id: token.chain_first_run_id.clone(), - ..Default::default() - }, + crate::handle::chain_of(address, &token.chain_first_run_id), address.topic.clone(), token.attach_request_id.clone(), read.response.wake_after, @@ -565,14 +550,7 @@ impl NexusCalls { if payloads.is_empty() { return self.finish(work, Vec::new()).await; } - self.ask_codec( - work, - bytes, - false, - &call.namespace, - &input.stream.workflow_id, - payloads, - ) + self.ask_codec(work, bytes, false, &call.namespace, &input.stream, payloads) } /// `address` without its run, and a state that names the chain the run picked. A state the @@ -629,7 +607,7 @@ impl NexusCalls { bytes: usize, encode: bool, namespace: &str, - workflow_id: &str, + stream: &StreamRef, payloads: Vec, ) -> NexusStep { let continuation = self.next.fetch_add(1, Ordering::Relaxed); @@ -665,10 +643,13 @@ impl NexusCalls { step: Some(nexus_step::Step::Codec(CodecWork { encode, namespace: namespace.to_string(), - workflow_id: workflow_id.to_string(), + workflow_id: stream.workflow_id.clone(), payloads, continuation, - ..Default::default() + // The address check already refused any other kind. + owner_kind: operation::owner_kind(&stream.kind).unwrap_or(StreamOwnerKind::Workflow) + as i32, + activity_id: stream.activity_id.clone().unwrap_or_default(), })), } } @@ -1089,31 +1070,45 @@ fn carried_state(state: Option) -> Result, Box> { /// 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> { - if stream.kind != "workflow" { + let Some(owner_kind) = operation::owner_kind(&stream.kind) else { return Err(Box::new(failure::of_stream(&StreamError::unsupported( format!( - "this release opens streams of Workflows only, not of kind {}", + "this release opens streams of Workflows and standalone Activities only, not of \ + kind {}", failure::named(&stream.kind) ), )))); - } + }; if let Err(reason) = temporalio_common::streams::check_topic(&stream.topic) { return Err(Box::new(failure::input_mistake(failure::without_values( &reason, )))); } - if stream.workflow_id.is_empty() { - return Err(Box::new(failure::input_mistake( - "a stream needs its owner's Workflow id", - ))); + let activity_id = stream.activity_id.clone().unwrap_or_default(); + let mistake = match owner_kind { + StreamOwnerKind::Activity if activity_id.is_empty() => { + Some("a stream that an Activity owns needs its owner's Activity id") + } + StreamOwnerKind::Activity if !stream.workflow_id.is_empty() => { + Some("a stream that an Activity owns names an empty Workflow id") + } + StreamOwnerKind::Activity => None, + _ if stream.workflow_id.is_empty() => Some("a stream needs its owner's Workflow id"), + _ if stream.activity_id.is_some() => { + Some("a stream that a Workflow owns names no Activity id") + } + _ => None, + }; + if let Some(mistake) = mistake { + return Err(Box::new(failure::input_mistake(mistake))); } Ok(StreamAddress { namespace: namespace.to_string(), - owner_kind: StreamOwnerKind::Workflow as i32, + owner_kind: owner_kind as i32, workflow_id: stream.workflow_id.clone(), run_id: stream.chain_run_id.clone().unwrap_or_default(), topic: stream.topic.clone(), - ..Default::default() + activity_id, }) } diff --git a/crates/streams/src/nexus/operation.rs b/crates/streams/src/nexus/operation.rs index 30993a3e76..177c02e95e 100644 --- a/crates/streams/src/nexus/operation.rs +++ b/crates/streams/src/nexus/operation.rs @@ -136,16 +136,26 @@ async fn detach( /// crosses as the caller's input mistake. fn checked(stream: Option<&StreamAddress>) -> StreamResult<&StreamAddress> { let stream = stream.ok_or_else(|| bad_input("an attach needs its stream"))?; - if stream.owner_kind != StreamOwnerKind::Workflow as i32 { - return Err(StreamError::unsupported(format!( - "this release opens streams of Workflows only, not owner kind {}", - stream.owner_kind + let (owner_id, other_id, kind) = match StreamOwnerKind::try_from(stream.owner_kind) { + Ok(StreamOwnerKind::Workflow) => (&stream.workflow_id, &stream.activity_id, "Workflow"), + Ok(StreamOwnerKind::Activity) => (&stream.activity_id, &stream.workflow_id, "Activity"), + _ => { + return Err(StreamError::unsupported(format!( + "this release opens streams of Workflows and standalone Activities only, not \ + owner kind {}", + stream.owner_kind + ))); + } + }; + if stream.namespace.is_empty() || owner_id.is_empty() { + return Err(bad_input(&format!( + "a stream needs its namespace and its owner's {kind} id" ))); } - if stream.namespace.is_empty() || stream.workflow_id.is_empty() { - return Err(bad_input( - "a stream needs its namespace and its owner's Workflow id", - )); + if !other_id.is_empty() { + return Err(bad_input(&format!( + "a stream that a {kind} owns names only its {kind} id" + ))); } // The topic is the caller's own value, so the message names the rule and not the topic. temporalio_common::streams::check_topic(&stream.topic) @@ -612,6 +622,49 @@ mod tests { ); } + #[tokio::test] + async fn an_activity_stream_start_attaches_waits_and_detaches_on_the_activity_run() { + let setup = setup(); + let activity = StreamAddress { + owner_kind: StreamOwnerKind::Activity as i32, + workflow_id: String::new(), + activity_id: "tool-1".to_string(), + ..stream("") + }; + let start = AttachCallerRequest { + stream: Some(activity.clone()), + wake_after_counter: Some(0), + ..attach_request("temporal://system") + }; + let token = attach(&setup.service, start).await.unwrap(); + let run = ChainId { + namespace: "default".to_string(), + first_run_id: "chain-1".to_string(), + owner_kind: StreamOwnerKind::Activity as i32, + activity_id: "tool-1".to_string(), + ..Default::default() + }; + assert_eq!(setup.notifier.attached.lock().unwrap()[0].chain, run); + assert_eq!(setup.notifier.waits.lock().unwrap()[0].0, run); + let decoded = OperationToken::decode("default", &token).unwrap(); + assert_eq!(decoded.stream, activity); + assert_eq!(decoded.chain(), run); + detach(&setup.service, &token).await.unwrap(); + assert_eq!(setup.notifier.detached.lock().unwrap()[0].0, run); + // An Activity's address names no Workflow id. + let refused = AttachCallerRequest { + stream: Some(StreamAddress { + workflow_id: "owner-1".to_string(), + ..activity + }), + ..attach_request("temporal://system") + }; + assert_eq!( + attach(&setup.service, refused).await.unwrap_err().kind, + "BAD_REQUEST" + ); + } + #[tokio::test] async fn a_start_attaches_an_idle_callback_and_announces_nothing() { let setup = setup(); diff --git a/crates/streams/tests/nexus_service.rs b/crates/streams/tests/nexus_service.rs index 893a18daec..d210094293 100644 --- a/crates/streams/tests/nexus_service.rs +++ b/crates/streams/tests/nexus_service.rs @@ -159,6 +159,22 @@ fn operation_token(wf: &str) -> String { .encode() } +/// The token of Activity `id`'s stream operation, attached as `attach-1` on its run `run-1`. +fn activity_operation_token(id: &str) -> String { + OperationToken { + stream: StreamAddress { + namespace: NAMESPACE.to_string(), + owner_kind: StreamOwnerKind::Activity as i32, + activity_id: id.to_string(), + topic: TOPIC.to_string(), + ..Default::default() + }, + attach_request_id: "attach-1".to_string(), + chain_first_run_id: "run-1".to_string(), + } + .encode() +} + fn waking(wf: &str, after: &str) -> Value { let mut input = read_input(wf, after); input["wake_token"] = json!(operation_token(wf)); @@ -533,7 +549,7 @@ mod cases { let calls = service(store, NexusOptions::default()); let refused = read( &calls, - json!({"stream": {"kind": "activity", "workflow_id": "wf", "topic": "t"}}), + json!({"stream": {"kind": "nexus", "workflow_id": "wf", "topic": "t"}}), ) .await .unwrap_err(); @@ -543,7 +559,137 @@ mod cases { kind_of(&refused), ("NOT_IMPLEMENTED", false, "StreamUnsupportedError") ); - assert!(refused.message.contains("activity"), "{}", refused.message); + assert!(refused.message.contains("nexus"), "{}", refused.message); + // An Activity's ref names the Activity and an empty Workflow id, and a Workflow's ref no + // Activity. + for stream in [ + json!({"kind": "activity", "workflow_id": "wf", "activity_id": "a", "topic": "t"}), + json!({"kind": "activity", "workflow_id": "", "topic": "t"}), + json!({"kind": "activity", "workflow_id": "", "activity_id": "", "topic": "t"}), + json!({"kind": "workflow", "workflow_id": "wf", "activity_id": "a", "topic": "t"}), + ] { + let refused = read(&calls, json!({ "stream": stream.clone() })) + .await + .unwrap_err(); + assert_eq!(kind_of(&refused).0, "BAD_REQUEST", "{stream}"); + } + } + + pub(crate) async fn an_activity_stream_appends_and_reads_apart_from_a_workflow( + store: Arc, + ) { + let calls = service(store, NexusOptions::default()); + let id = workflow(); + let activity = |texts: &[&str], sequence: i64| { + let mut input = append_input(&id, "p", 1, sequence, texts); + input["stream"] = activity_stream(&id, TOPIC); + input + }; + append(&calls, activity(&["one", "two"], 1)).await.unwrap(); + // A Workflow with the same id owns another stream. + write(&calls, &id, 1, &["workflow"]).await; + let answer = read(&calls, json!({"stream": activity_stream(&id, TOPIC)})) + .await + .unwrap(); + assert_eq!(texts(&answer), ["one", "two"]); + assert_eq!(texts(&read_after(&calls, &id, "").await), ["workflow"]); + let latest = read( + &calls, + json!({"stream": activity_stream(&id, TOPIC), "latest_only": true}), + ) + .await + .unwrap(); + assert_eq!(latest["next_token"], answer["next_token"]); + // A cursor of the Activity's stream is not one of the Workflow's. + let refused = read(&calls, read_input(&id, token(&answer))) + .await + .unwrap_err(); + assert_eq!(kind_of(&refused).2, "StreamCursorError"); + } + + pub(crate) async fn an_activity_read_says_done_once_the_run_ends(store: Arc) { + let id = workflow(); + let mut input = append_input(&id, "p", 1, 1, &["one"]); + input["stream"] = activity_stream(&id, TOPIC); + append(&service(store.clone(), NexusOptions::default()), input) + .await + .unwrap(); + let ended = NexusCalls::new( + Arc::new(Streams::new( + store, + Arc::new(Ended), + StreamsOptions::default(), + )), + NexusOptions::default(), + ); + let mut input = json!({"stream": activity_stream(&id, TOPIC), "wait_ms": 2000}); + let mut seen = vec![]; + for _ in 0..10 { + let answer = read(&ended, input.clone()).await.unwrap(); + seen.extend(texts(&answer)); + if answer["done"] == true { + assert_eq!(seen, ["one"]); + return; + } + input["after_token"] = answer["next_token"].clone(); + input["state"] = answer["state"].clone(); + } + panic!("the read never ended"); + } + + pub(crate) async fn a_wake_token_of_another_activity_is_refused(store: Arc) { + let waker = Arc::new(Waker::default()); + let calls = service(store, NexusOptions::default()).with_callers(waker.clone()); + let id = workflow(); + // Every Activity's ref has the same empty Workflow id, so only the Activity id tells + // the two streams apart. + for (stream, token) in [ + ( + activity_stream(&id, TOPIC), + activity_operation_token("other"), + ), + (activity_stream(&id, TOPIC), operation_token("")), + (stream(&id, TOPIC), activity_operation_token(&id)), + ] { + let input = json!({"stream": stream, "wake_token": token}); + let refused = read(&calls, input).await.unwrap_err(); + assert_eq!(kind_of(&refused).0, "BAD_REQUEST", "{stream}"); + } + assert!(waker.waits.lock().unwrap().is_empty()); + } + + pub(crate) async fn an_empty_activity_read_wakes_its_operation_on_the_activity_run( + store: Arc, + ) { + let waker = Arc::new(Waker::default()); + let calls = service(store, NexusOptions::default()).with_callers(waker.clone()); + let id = workflow(); + let mut input = append_input(&id, "p", 1, 1, &["one"]); + input["stream"] = activity_stream(&id, TOPIC); + let last = append(&calls, input).await.unwrap()["cursor"].clone(); + let input = json!({ + "stream": activity_stream(&id, TOPIC), + "after_token": last, + "wake_token": activity_operation_token(&id), + }); + let caught_up = read(&calls, input).await.unwrap(); + assert!(records(&caught_up).is_empty()); + let waits = waker.waits.lock().unwrap().clone(); + let [(chain, topic, request_id, after)] = waits.as_slice() else { + panic!("one wait, not {waits:?}"); + }; + assert_eq!( + chain, + &ChainId { + namespace: NAMESPACE.to_string(), + first_run_id: "run-1".to_string(), + owner_kind: StreamOwnerKind::Activity as i32, + activity_id: id.clone(), + ..Default::default() + } + ); + assert_eq!((topic.as_str(), request_id.as_str()), (TOPIC, "attach-1")); + assert!(*after > 0); } pub(crate) async fn a_read_says_done_once_the_owner_ends(store: Arc) { @@ -915,6 +1061,10 @@ macro_rules! all_cases { a_record_that_lands_before_the_wait_is_in_the_answer, a_positions_reader_waits_and_rereads_too, a_wake_token_of_another_stream_is_refused, + an_activity_stream_appends_and_reads_apart_from_a_workflow, + an_activity_read_says_done_once_the_run_ends, + a_wake_token_of_another_activity_is_refused, + an_empty_activity_read_wakes_its_operation_on_the_activity_run, ); }; } @@ -1385,3 +1535,49 @@ async fn a_range_from_a_trimmed_start_is_lost_on_redis() { refused.message ); } + +#[tokio::test] +async fn an_activity_range_replays_and_names_only_the_activity_run() { + let calls = service(memory(), NexusOptions::default()); + let id = workflow(); + let mut input = append_input(&id, "p", 1, 1, &["one", "two"]); + input["stream"] = activity_stream(&id, TOPIC); + append(&calls, input).await.unwrap(); + let answer = read( + &calls, + json!({"stream": activity_stream(&id, TOPIC), "floor_holder": HOLDER}), + ) + .await + .unwrap(); + assert_eq!(answer["replay"]["chain_first_run_id"], "run-1"); + let replay = replay_input(activity_stream(&id, TOPIC), &answer, Value::Null); + let replayed = read_range(&calls, replay.clone()).await.unwrap(); + assert_eq!(replayed["records"], answer["records"]); + let mut other_run = replay; + other_run["chain_first_run_id"] = json!("run-9"); + let refused = read_range(&calls, other_run).await.unwrap_err(); + assert_eq!( + kind_of(&refused), + ("BAD_REQUEST", false, "StreamCursorError") + ); +} + +#[tokio::test] +async fn an_activity_streams_codec_work_names_the_activity() { + let calls = service(memory(), NexusOptions::default()); + let mut input = append_input("unused", "p", 1, 1, &["one"]); + input["stream"] = activity_stream("tool-1", TOPIC); + let step = calls.append(call(input)).await; + let Some(temporalio_streams::proto::nexus_step::Step::Codec(work)) = step.step else { + panic!("an append asks for codec work first, not {step:?}"); + }; + assert_eq!( + ( + work.namespace.as_str(), + work.owner_kind, + work.workflow_id.as_str(), + work.activity_id.as_str() + ), + (NAMESPACE, StreamOwnerKind::Activity as i32, "", "tool-1") + ); +} diff --git a/crates/streams/tests/nexus_support/mod.rs b/crates/streams/tests/nexus_support/mod.rs index 091e57dfba..5d6a7869e0 100644 --- a/crates/streams/tests/nexus_support/mod.rs +++ b/crates/streams/tests/nexus_support/mod.rs @@ -189,6 +189,16 @@ pub(crate) fn replay_input(stream: Value, answer: &Value, state: Value) -> Value }) } +/// A stream reference to `topic` of standalone Activity `activity_id`, on its newest run. +pub(crate) fn activity_stream(activity_id: &str, topic: &str) -> Value { + serde_json::json!({ + "kind": "activity", + "workflow_id": "", + "activity_id": activity_id, + "topic": topic, + }) +} + /// 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})