From 2258a2cd77c1279bef524a4c44a3e6dd7efc6509 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Tue, 13 Oct 2026 22:55:03 -0700 Subject: [PATCH 1/5] Attached and detached stream operation callers in Core. AttachCaller describes the Workflow fresh, mints the operation token and attaches the start's callback. DetachCaller decodes the token and detaches from the chain that the start attached to. Failures cross as a NexusFailure from the Nexus call table. --- .../temporal/sdk/core/streams/nexus.proto | 38 + crates/sdk-core/CHANGELOG.md | 9 + crates/sdk-core/src/streams.rs | 123 ++- .../nexus/operation_token_vectors.json | 77 ++ crates/streams/src/lib.rs | 7 +- crates/streams/src/nexus/mod.rs | 4 + crates/streams/src/nexus/operation.rs | 891 ++++++++++++++++++ crates/streams/src/notify.rs | 29 +- crates/streams/src/service.rs | 37 +- 9 files changed, 1175 insertions(+), 40 deletions(-) create mode 100644 crates/streams/nexus/operation_token_vectors.json create mode 100644 crates/streams/src/nexus/operation.rs 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 ffdaa94bba..3c89ae53c2 100644 --- a/crates/protos/protos/local/temporal/sdk/core/streams/nexus.proto +++ b/crates/protos/protos/local/temporal/sdk/core/streams/nexus.proto @@ -6,6 +6,7 @@ option ruby_package = "Temporalio::Internal::Bridge::Api::Streams"; import "google/protobuf/timestamp.proto"; import "temporal/api/common/v1/message.proto"; import "temporal/api/failure/v1/message.proto"; +import "temporal/sdk/core/streams/streams.proto"; // The stream service's Nexus operations, served by Core. Lang forwards each call's input in the // N0a contract's JSON transfer form, runs its payload codec when Core asks for it, and hands the @@ -20,6 +21,12 @@ service StreamNexusService { rpc NexusRead(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, + // and answers with the operation token, or the failure the start answers with. The server then delivers the stream's progress and its + // close result to the caller's callback. Idempotent by the start's request id. + rpc AttachCaller(AttachCallerRequest) returns (AttachCallerResponse); + // Detaches the caller a token's start attached, from the chain it attached to. + rpc DetachCaller(DetachCallerRequest) returns (DetachCallerResponse); } // One Nexus operation call. @@ -91,3 +98,34 @@ message NexusFailure { // application error type. temporal.api.failure.v1.Failure failure = 4; } + +message AttachCallerRequest { + // The stream the operation's start names. An empty run id attaches to the chain that holds the + // Workflow id now. + StreamAddress stream = 1; + // The start's request id. A retried start attaches once. + string request_id = 2; + // The start's callback URL and headers. An empty URL is refused, since the caller could never + // hear about the stream. + string callback_url = 3; + map callback_headers = 4; +} + +message AttachCallerResponse { + // The token the start answers with. It names the stream, the attach and the chain. + string operation_token = 1; + // Set when the attach failed, as the Nexus calls answer a failure, so lang hands it to its + // Nexus handler as it is. The token is then empty. + NexusFailure failure = 2; +} + +message DetachCallerRequest { + string namespace = 1; + // The token the start answered with. + string operation_token = 2; +} + +message DetachCallerResponse { + // Set when the detach failed, as the Nexus calls answer a failure. + NexusFailure failure = 1; +} diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index cbc2072a94..7d21bc6283 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/crates/sdk-core/CHANGELOG.md @@ -104,6 +104,15 @@ relevant information. in the same activations. Progress for an operation that is unknown, not started, already resolved, or below a counter already delivered is dropped. A task scheduled with progress ends a workflow task heartbeat chain, as one with stream wakes does. +* Experimental: the stream calls `AttachCaller` and `DetachCaller` serve a stream-returning Nexus + operation. `AttachCaller` resolves the stream's run chain, mints the operation token and + attaches the start's callback to the stream's notifier with the token and start time. + `DetachCaller` decodes the token and detaches from the chain the start attached to. The token + format is in `crates/streams/nexus/operation_token_vectors.json`, which every SDK's encoder and + decoder check. Both answer a failure in the response as a `NexusFailure`, from the same table as + the Nexus calls, so lang's handler raises it as it is. A mistake in the stream, an empty callback + URL or request id, and a token that names no stream are `BAD_REQUEST`. An attach for the latest + run describes the Workflow fresh, so it never lands on a chain that already ended. ### Breaking Changes * Replay detects changes to the service or operation of scheduled Nexus operations, reporting diff --git a/crates/sdk-core/src/streams.rs b/crates/sdk-core/src/streams.rs index ae55b97817..804ebe9298 100644 --- a/crates/sdk-core/src/streams.rs +++ b/crates/sdk-core/src/streams.rs @@ -4,7 +4,7 @@ use std::sync::Arc; use temporalio_client::Connection; use temporalio_common::protos::temporal::api::{ - common::v1::WorkflowExecution, + common::v1::{WorkflowExecution, callback}, enums::v1::StreamOwnerKind, stream::v1::{StreamReference, WorkflowWakeTarget}, workflowservice::v1::{ @@ -14,8 +14,8 @@ use temporalio_common::protos::temporal::api::{ }, }; use temporalio_streams::{ - CloseOutcome, NotifierClient, NotifierPoll, NotifyError, OwnerClient, OwnerDescription, - OwnerError, StreamNotification, WorkflowAttachment, proto::ChainId, + CallerAttachment, CloseOutcome, NotifierClient, NotifierPoll, NotifyError, OwnerClient, + OwnerDescription, OwnerError, StreamNotification, WorkflowAttachment, proto::ChainId, }; pub use temporalio_streams::{StreamError, StreamResult, StreamService, StreamStore, proto}; use tonic::IntoRequest; @@ -62,6 +62,15 @@ impl NotifierClient for ConnectionNotifierClient { .map_err(notify_error) } + async fn attach_caller(&self, attachment: CallerAttachment) -> Result<(), NotifyError> { + self.connection + .workflow_service() + .attach_stream_callback(caller_attach_request(attachment).into_request()) + .await + .map(|_| ()) + .map_err(notify_error) + } + async fn detach( &self, chain: ChainId, @@ -149,6 +158,21 @@ fn attach_request(attachment: WorkflowAttachment) -> AttachStreamCallbackRequest } } +fn caller_attach_request(attachment: CallerAttachment) -> AttachStreamCallbackRequest { + AttachStreamCallbackRequest { + namespace: attachment.chain.namespace.clone(), + stream_ref: Some(chain_reference(&attachment.chain, attachment.topic)), + request_id: attachment.request_id, + callback: Some(callback::Nexus { + url: attachment.callback_url, + header: attachment.callback_headers, + }), + operation_token: attachment.operation_token, + start_time: Some(attachment.start_time.into()), + ..Default::default() + } +} + fn notify_request(notification: StreamNotification) -> NotifyStreamRequest { NotifyStreamRequest { stream_ref: Some(chain_reference(¬ification.chain, notification.topic)), @@ -352,6 +376,45 @@ mod tests { assert!(request.close_result.is_none()); } + fn chain() -> ChainId { + ChainId { + namespace: "default".to_string(), + workflow_id: "owner-1".to_string(), + first_run_id: "chain-1".to_string(), + } + } + + #[test] + fn a_caller_attach_names_the_chains_notifier_and_carries_the_start() { + let start_time = std::time::UNIX_EPOCH + std::time::Duration::from_secs(1_700_000_000); + let request = caller_attach_request(CallerAttachment { + chain: chain(), + topic: "tokens".to_string(), + request_id: "attach-1".to_string(), + callback_url: "temporal://system".to_string(), + callback_headers: [("nexus-callback-token".to_string(), "abc".to_string())].into(), + operation_token: "token".to_string(), + start_time, + }); + assert_eq!(request.namespace, "default"); + assert_eq!( + request.stream_ref, + Some(StreamReference { + owner_kind: StreamOwnerKind::Workflow as i32, + workflow_id: "owner-1".to_string(), + run_id: "chain-1".to_string(), + topic: "tokens".to_string(), + }) + ); + assert_eq!(request.request_id, "attach-1"); + let callback = request.callback.unwrap(); + assert_eq!(callback.url, "temporal://system"); + assert_eq!(callback.header["nexus-callback-token"], "abc"); + assert_eq!(request.operation_token, "token"); + assert_eq!(request.start_time.unwrap().seconds, 1_700_000_000); + assert!(request.workflow_target.is_none(), "exactly one target"); + } + #[test] fn an_attach_names_the_chains_notifier_and_wakes_the_reader() { let request = attach_request(WorkflowAttachment { @@ -537,6 +600,8 @@ mod tests { stream: proto::StreamAddress, workflow_id: String, run_id: String, + /// The chain the caller's operation token names. + token_first_run_id: String, // The live tests share the callback listener's address. _turn: tokio::sync::MutexGuard<'static, ()>, } @@ -548,9 +613,8 @@ mod tests { /// `127.0.0.1:7899`. Answers `None` without one. async fn live() -> Option { use temporalio_common::protos::temporal::api::{ - common::v1::WorkflowType, - taskqueue::v1::TaskQueue, - workflowservice::v1::{AttachStreamCallbackRequest, StartWorkflowExecutionRequest}, + common::v1::WorkflowType, taskqueue::v1::TaskQueue, + workflowservice::v1::StartWorkflowExecutionRequest, }; let Ok(server) = std::env::var("STREAMS_NOTIFIER_SERVER") else { eprintln!("set STREAMS_NOTIFIER_SERVER to run the live notifier test"); @@ -587,30 +651,6 @@ mod tests { .into_inner(); // An earlier live test may have left deliveries behind. while deliveries().lock().unwrap().try_recv().is_ok() {} - connection - .workflow_service() - .attach_stream_callback( - AttachStreamCallbackRequest { - namespace: "default".to_string(), - stream_ref: Some(StreamReference { - owner_kind: StreamOwnerKind::Workflow as i32, - workflow_id: workflow_id.clone(), - run_id: started.run_id.clone(), - topic: "tokens".to_string(), - }), - request_id: "caller".to_string(), - callback: Some( - temporalio_common::protos::temporal::api::common::v1::callback::Nexus { - url: "http://127.0.0.1:7899/callback".to_string(), - ..Default::default() - }, - ), - ..Default::default() - } - .into_request(), - ) - .await - .unwrap(); let service = connect_stream_service( proto::StreamStoreConfig { store: Some(proto::stream_store_config::Store::Memory( @@ -630,12 +670,34 @@ mod tests { run_id: String::new(), topic: "tokens".to_string(), }; + use prost::Message; + let attached = proto::AttachCallerResponse::decode( + service + .call( + "AttachCaller", + &proto::AttachCallerRequest { + stream: Some(stream.clone()), + request_id: "caller".to_string(), + callback_url: "http://127.0.0.1:7899/callback".to_string(), + ..Default::default() + } + .encode_to_vec(), + ) + .await + .unwrap() + .as_slice(), + ) + .unwrap(); + let token = + temporalio_streams::OperationToken::decode("default", &attached.operation_token) + .unwrap(); Some(Live { connection, service, stream, workflow_id, run_id: started.run_id, + token_first_run_id: token.chain_first_run_id, _turn: turn, }) } @@ -713,6 +775,7 @@ mod tests { async fn a_live_notifier_hears_appends_as_progress_and_the_close_as_the_result() { use temporalio_common::protos::temporal::api::common::v1::Payload; let Some(live) = live().await else { return }; + assert_eq!(live.token_first_run_id, live.run_id); live.append_one().await; let (state, body) = live .close(proto::CloseRequest { diff --git a/crates/streams/nexus/operation_token_vectors.json b/crates/streams/nexus/operation_token_vectors.json new file mode 100644 index 0000000000..90b1ef636d --- /dev/null +++ b/crates/streams/nexus/operation_token_vectors.json @@ -0,0 +1,77 @@ +{ + "comment": "Token v2 of a stream-returning Nexus operation: base64url JSON, no padding, keys sorted, non-ASCII escaped, and no run_id for an unpinned ref. Core's encoder and every SDK's encoder write the vectors, and every decoder reads them. Decoders also read decode_only, the null run_id that SDK releases before Core wrote.", + "vectors": [ + { + "ref": { + "kind": "workflow", + "workflow_id": "owner-1", + "topic": "tokens" + }, + "attach": "attach-1", + "chain": "chain-1", + "token": "eyJhdHRhY2giOiJhdHRhY2gtMSIsImNoYWluIjoiY2hhaW4tMSIsInJlZiI6eyJraW5kIjoid29ya2Zsb3ciLCJ0b3BpYyI6InRva2VucyIsIndvcmtmbG93X2lkIjoib3duZXItMSJ9LCJ2IjoyfQ" + }, + { + "ref": { + "kind": "workflow", + "workflow_id": "owner-1", + "run_id": "run-2", + "topic": "tokens" + }, + "attach": "attach-1", + "chain": "run-1", + "token": "eyJhdHRhY2giOiJhdHRhY2gtMSIsImNoYWluIjoicnVuLTEiLCJyZWYiOnsia2luZCI6IndvcmtmbG93IiwicnVuX2lkIjoicnVuLTIiLCJ0b3BpYyI6InRva2VucyIsIndvcmtmbG93X2lkIjoib3duZXItMSJ9LCJ2IjoyfQ" + }, + { + "ref": { + "kind": "workflow", + "workflow_id": "order/42 \"quoted\"", + "topic": "t\\back\nline" + }, + "attach": "req-é", + "chain": "chain-ü", + "token": "eyJhdHRhY2giOiJyZXEtXHUwMGU5IiwiY2hhaW4iOiJjaGFpbi1cdTAwZmMiLCJyZWYiOnsia2luZCI6IndvcmtmbG93IiwidG9waWMiOiJ0XFxiYWNrXG5saW5lIiwid29ya2Zsb3dfaWQiOiJvcmRlci80MiBcInF1b3RlZFwiIn0sInYiOjJ9" + }, + { + "ref": { + "kind": "workflow", + "workflow_id": "日本語-wf", + "topic": "emoji-😀" + }, + "attach": "r", + "chain": "c", + "token": "eyJhdHRhY2giOiJyIiwiY2hhaW4iOiJjIiwicmVmIjp7ImtpbmQiOiJ3b3JrZmxvdyIsInRvcGljIjoiZW1vamktXHVkODNkXHVkZTAwIiwid29ya2Zsb3dfaWQiOiJcdTY1ZTVcdTY3MmNcdThhOWUtd2YifSwidiI6Mn0" + }, + { + "ref": { + "kind": "workflow", + "workflow_id": "delwf", + "topic": "t" + }, + "attach": "a", + "chain": "c", + "token": "eyJhdHRhY2giOiJhXHUwMDdmIiwiY2hhaW4iOiJjIiwicmVmIjp7ImtpbmQiOiJ3b3JrZmxvdyIsInRvcGljIjoidFx1MDA3ZiIsIndvcmtmbG93X2lkIjoiZGVsXHUwMDdmd2YifSwidiI6Mn0" + } + ], + "decode_only": [ + { + "ref": { + "kind": "workflow", + "workflow_id": "owner-1", + "run_id": null, + "topic": "tokens" + }, + "attach": "attach-1", + "chain": "chain-1", + "token": "eyJhdHRhY2giOiJhdHRhY2gtMSIsImNoYWluIjoiY2hhaW4tMSIsInJlZiI6eyJraW5kIjoid29ya2Zsb3ciLCJydW5faWQiOm51bGwsInRvcGljIjoidG9rZW5zIiwid29ya2Zsb3dfaWQiOiJvd25lci0xIn0sInYiOjJ9" + } + ], + "refused": [ + "", + "not-a-token", + "eyJ2IjogMn0", + "eyJ2IjogMSwgImF0dGFjaCI6ICJhdHRhY2gtMSIsICJjaGFpbiI6ICJjaGFpbi0xIiwgInJlZiI6IHsia2luZCI6ICJ3b3JrZmxvdyIsICJ3b3JrZmxvd19pZCI6ICJvd25lci0xIiwgInJ1bl9pZCI6IG51bGwsICJ0b3BpYyI6ICJ0b2tlbnMifX0=", + "eyJhdHRhY2giOiJhIiwiY2hhaW4iOiJjIiwiZXh0cmEiOjEsInJlZiI6eyJraW5kIjoid29ya2Zsb3ciLCJ0b3BpYyI6InQiLCJ3b3JrZmxvd19pZCI6IncifSwidiI6Mn0", + "eyJhdHRhY2giOiJhIiwiY2hhaW4iOiJjIiwicmVmIjp7ImN1cnNvciI6IngiLCJraW5kIjoid29ya2Zsb3ciLCJ0b3BpYyI6InQiLCJ3b3JrZmxvd19pZCI6IncifSwidiI6Mn0" + ] +} diff --git a/crates/streams/src/lib.rs b/crates/streams/src/lib.rs index a79e83cfc5..552fe79b7b 100644 --- a/crates/streams/src/lib.rs +++ b/crates/streams/src/lib.rs @@ -29,10 +29,11 @@ pub use cursor::{BEGINNING, END, cursor_position, mint_cursor, stream_hash}; pub use error::{StreamError, StreamResult}; pub use handle::{OWNER_RECHECK, Streams, StreamsOptions}; pub use memory::MemoryStore; +pub use nexus::{OPERATION_TOKEN_VERSION, OperationToken}; pub use notify::{ - Arm, CLOSE_RETRY_CAP, CLOSE_RETRY_FIRST, DEFAULT_MAX_NOTIFIERS, FLUSH_LIMIT, NOTIFIER_REFRESH, - Notifier, NotifierClient, NotifierPoll, NotifyError, NotifyingStore, StreamNotification, - WorkflowAttachment, progress_counter, + Arm, CLOSE_RETRY_CAP, CLOSE_RETRY_FIRST, CallerAttachment, DEFAULT_MAX_NOTIFIERS, FLUSH_LIMIT, + NOTIFIER_REFRESH, Notifier, NotifierClient, NotifierPoll, NotifyError, NotifyingStore, + StreamNotification, WorkflowAttachment, progress_counter, }; pub use owner::{OwnerClient, OwnerDescription, OwnerError}; pub use reader::{DEFAULT_MAX_RECORDS, MAX_TRACKED_PRODUCERS, ReadTarget, read_page}; diff --git a/crates/streams/src/nexus/mod.rs b/crates/streams/src/nexus/mod.rs index 00c40a88ae..e2262f8b6d 100644 --- a/crates/streams/src/nexus/mod.rs +++ b/crates/streams/src/nexus/mod.rs @@ -7,8 +7,12 @@ //! stateless, so a read goes on at any handler Worker. mod failure; +mod operation; mod wire; +pub use operation::{OPERATION_TOKEN_VERSION, OperationToken}; +pub(crate) use operation::{attach_caller, detach_caller}; + use crate::{ StreamError, Streams, handle::CarriedRead, diff --git a/crates/streams/src/nexus/operation.rs b/crates/streams/src/nexus/operation.rs new file mode 100644 index 0000000000..6edf7161c1 --- /dev/null +++ b/crates/streams/src/nexus/operation.rs @@ -0,0 +1,891 @@ +//! The callers of a stream-returning Nexus operation. +//! +//! The operation's start attaches the caller's callback to the stream's notifier on the server, +//! and answers with a token that names the stream. Its cancel detaches the callback the start +//! attached. The token format is shared with every SDK, which decodes it on the Workflow side, so +//! it must stay byte for byte what `crates/streams/nexus/operation_token_vectors.json` holds. + +use crate::{ + CallerAttachment, NotifierClient, NotifyError, StreamError, StreamResult, Streams, + proto::{ + AttachCallerRequest, AttachCallerResponse, ChainId, DetachCallerRequest, + DetachCallerResponse, StreamAddress, StreamOwnerKind, + }, +}; +use base64::Engine; +use serde::{Deserialize, Serialize}; +use std::{io, sync::Arc, time::SystemTime}; + +/// Attaches a start's caller to the stream's notifier and mints the operation token. +/// Answers a failure as the Nexus calls do, so lang's handler raises it with Core's one table. +pub(crate) async fn attach_caller( + streams: &Streams, + client: Option<&Arc>, + request: AttachCallerRequest, +) -> AttachCallerResponse { + match attach(streams, client, request).await { + Ok(operation_token) => AttachCallerResponse { + operation_token, + failure: None, + }, + Err(error) => AttachCallerResponse { + operation_token: String::new(), + failure: Some(super::failure::of_stream(&error)), + }, + } +} + +/// Detaches the caller a token's start attached. Answers a failure as the Nexus calls do. +pub(crate) async fn detach_caller( + client: Option<&Arc>, + request: DetachCallerRequest, +) -> DetachCallerResponse { + DetachCallerResponse { + failure: detach(client, request) + .await + .err() + .map(|error| super::failure::of_stream(&error)), + } +} + +async fn attach( + streams: &Streams, + client: Option<&Arc>, + request: AttachCallerRequest, +) -> StreamResult { + if request.callback_url.is_empty() { + return Err(bad_input( + "a stream-returning operation needs a caller callback; start it asynchronously", + )); + } + if request.request_id.is_empty() { + return Err(bad_input("an attach needs the start's request id")); + } + let client = client.ok_or_else(|| { + StreamError::unsupported("attaching a caller needs a client connected to the server") + })?; + let stream = checked(request.stream.as_ref())?; + // Fresh for the latest run, since the Workflow id may have moved on to a new chain since a + // read warmed the cache, and an attach to the ended chain would never hear the new stream. + let chain = streams.chain_now(stream).await?; + let operation_token = OperationToken { + stream: stream.clone(), + attach_request_id: request.request_id.clone(), + chain_first_run_id: chain.first_run_id.clone(), + } + .encode(); + // The server hands the token and start time back with progress and the completion, so a + // completion that beats the start's response still names the operation. + client + .attach_caller(CallerAttachment { + chain, + topic: stream.topic.clone(), + request_id: request.request_id, + callback_url: request.callback_url, + callback_headers: request.callback_headers, + operation_token: operation_token.clone(), + start_time: SystemTime::now(), + }) + .await + .map_err(caller_error)?; + Ok(operation_token) +} + +async fn detach( + client: Option<&Arc>, + request: DetachCallerRequest, +) -> StreamResult<()> { + if request.namespace.is_empty() { + return Err(bad_input("a detach needs its namespace")); + } + let token = OperationToken::decode(&request.namespace, &request.operation_token)?; + let client = client.ok_or_else(|| { + StreamError::unsupported("detaching a caller needs a client connected to the server") + })?; + // The server's detach changes nothing for an attach that's gone, so a retried cancel lands as + // the first did. + client + .detach( + token.chain(), + token.stream.topic.clone(), + token.attach_request_id, + ) + .await + .map_err(caller_error)?; + Ok(()) +} + +/// The stream an attach names, checked as the Nexus calls check theirs, so a mistake in it +/// 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 + ))); + } + 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", + )); + } + // 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) + .map_err(|reason| bad_input(&super::failure::without_values(&reason)))?; + Ok(stream) +} + +fn bad_input(message: &str) -> StreamError { + StreamError::invalid_argument(message) +} + +fn caller_error(error: NotifyError) -> StreamError { + match error { + NotifyError::Refused(message) => StreamError::refused(message), + NotifyError::Unsupported(message) => StreamError::unsupported(message), + NotifyError::Failed(message) => StreamError::storage(message), + } +} + +/// The token format version. A later release that carries more tells an old token apart by it. +pub const OPERATION_TOKEN_VERSION: u32 = 2; + +/// What a stream-returning operation's token carries. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct OperationToken { + /// The stream, as the operation's `open_stream` named it. + pub stream: StreamAddress, + /// The start's request id, which the attach was keyed by. + pub attach_request_id: String, + /// The first run of the chain the start attached to, so a cancel detaches from that chain + /// even after the Workflow id moved on to a new one. + pub chain_first_run_id: String, +} + +// Fields in name order, since SDKs encode with sorted keys and the bytes must match. The model is +// closed in the contract, so a token with another member is not one of ours. +#[derive(Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct TokenBody { + attach: String, + chain: String, + #[serde(rename = "ref")] + stream: TokenRef, + v: u32, +} + +#[derive(Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct TokenRef { + kind: String, + // An unpinned ref has none, as the generated SDK writers leave an unset member out. SDK + // releases before Core wrote null, which reads the same. + #[serde(default, skip_serializing_if = "Option::is_none")] + run_id: Option, + topic: String, + workflow_id: String, +} + +const WORKFLOW_KIND: &str = "workflow"; + +impl OperationToken { + /// The token's text: base64url JSON with sorted keys and no padding. + pub fn encode(&self) -> String { + let body = TokenBody { + attach: self.attach_request_id.clone(), + chain: self.chain_first_run_id.clone(), + stream: TokenRef { + kind: WORKFLOW_KIND.to_string(), + run_id: (!self.stream.run_id.is_empty()).then(|| self.stream.run_id.clone()), + topic: self.stream.topic.clone(), + workflow_id: self.stream.workflow_id.clone(), + }, + v: OPERATION_TOKEN_VERSION, + }; + let mut json = Vec::new(); + let mut serializer = serde_json::Serializer::with_formatter(&mut json, AsciiOnly); + body.serialize(&mut serializer) + .expect("a token body is plain strings"); + base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(json) + } + + /// Reads a token in `namespace`. Fails as an invalid argument when it isn't one a stream + /// operation issued. + pub fn decode(namespace: &str, token: &str) -> StreamResult { + let refuse = |why: &str| bad_input(&format!("not a stream operation token: {why}")); + let json = base64::engine::general_purpose::URL_SAFE_NO_PAD + .decode(token.trim_end_matches('=')) + .map_err(|_| refuse("not base64url"))?; + let version: serde_json::Value = + serde_json::from_slice(&json).map_err(|_| refuse("not JSON"))?; + if version.get("v").and_then(serde_json::Value::as_u64) + != Some(u64::from(OPERATION_TOKEN_VERSION)) + { + return Err(refuse("unknown token version")); + } + let body: TokenBody = + serde_json::from_value(version).map_err(|error| refuse(&error.to_string()))?; + if body.stream.kind != WORKFLOW_KIND { + return Err(refuse("the stream's owner is not a Workflow")); + } + Ok(Self { + stream: StreamAddress { + namespace: namespace.to_string(), + owner_kind: StreamOwnerKind::Workflow as i32, + workflow_id: body.stream.workflow_id, + run_id: body.stream.run_id.unwrap_or_default(), + topic: body.stream.topic, + }, + attach_request_id: body.attach, + chain_first_run_id: body.chain, + }) + } + + /// The chain the start attached to. + pub fn chain(&self) -> ChainId { + ChainId { + namespace: self.stream.namespace.clone(), + workflow_id: self.stream.workflow_id.clone(), + first_run_id: self.chain_first_run_id.clone(), + } + } +} + +/// Writes JSON with every character past ASCII, and DEL, as a `\u` escape, as Python's +/// `json.dumps` does by default. +struct AsciiOnly; + +impl serde_json::ser::Formatter for AsciiOnly { + fn write_string_fragment(&mut self, writer: &mut W, fragment: &str) -> io::Result<()> + where + W: ?Sized + io::Write, + { + for character in fragment.chars() { + if character.is_ascii() && character != '\u{7f}' { + writer.write_all(&[character as u8])?; + } else { + let mut units = [0u16; 2]; + for unit in character.encode_utf16(&mut units) { + write!(writer, "\\u{unit:04x}")?; + } + } + } + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::{ + MemoryStore, OwnerClient, OwnerDescription, OwnerError, StreamNotification, StreamService, + StreamsOptions, proto::NexusFailure, + }; + use prost::Message; + use std::{collections::HashMap, sync::Mutex, time::UNIX_EPOCH}; + use temporalio_common::protos::temporal::api::enums::v1::WorkflowExecutionStatus; + + /// Answers each Workflow id's chain as `chains` holds it, `chain-1` by default, and records + /// the runs it was asked about. + #[derive(Default)] + struct Owners { + chains: Mutex>, + described: Mutex>, + missing: std::sync::atomic::AtomicBool, + } + + #[async_trait::async_trait] + impl OwnerClient for Owners { + async fn describe( + &self, + _: &str, + workflow_id: &str, + run_id: &str, + ) -> Result { + self.described.lock().unwrap().push(run_id.to_string()); + if self.missing.load(std::sync::atomic::Ordering::SeqCst) { + return Err(OwnerError::NotFound(workflow_id.to_string())); + } + let first_run_id = self + .chains + .lock() + .unwrap() + .get(workflow_id) + .cloned() + .unwrap_or_else(|| "chain-1".to_string()); + Ok(OwnerDescription { + run_id: if run_id.is_empty() { + first_run_id.clone() + } else { + run_id.to_string() + }, + first_run_id, + status: WorkflowExecutionStatus::Running, + }) + } + } + + /// Records the attaches and detaches, and answers each with `answer`. + #[derive(Default)] + struct Notifier { + attached: Mutex>, + detached: Mutex>, + answer: Mutex>, + } + + #[async_trait::async_trait] + impl NotifierClient for Notifier { + async fn notify(&self, _: StreamNotification) -> Result<(), NotifyError> { + Ok(()) + } + + async fn attach_caller(&self, attachment: CallerAttachment) -> Result<(), NotifyError> { + self.attached.lock().unwrap().push(attachment); + self.answer.lock().unwrap().clone().map_or(Ok(()), Err) + } + + async fn detach( + &self, + chain: ChainId, + topic: String, + request_id: String, + ) -> Result<(), NotifyError> { + self.detached + .lock() + .unwrap() + .push((chain, topic, request_id)); + self.answer.lock().unwrap().clone().map_or(Ok(()), Err) + } + } + + struct Setup { + owners: Arc, + notifier: Arc, + service: StreamService, + } + + fn setup() -> Setup { + let owners = Arc::new(Owners::default()); + let notifier = Arc::new(Notifier::default()); + let service = StreamService::new( + Arc::new(MemoryStore::new()), + owners.clone(), + // The default recheck, so a start that read the owner cache would see a stale chain. + StreamsOptions::default(), + ) + .with_callers(notifier.clone()); + Setup { + owners, + notifier, + service, + } + } + + fn stream(run_id: &str) -> StreamAddress { + StreamAddress { + namespace: "default".to_string(), + owner_kind: StreamOwnerKind::Workflow as i32, + workflow_id: "owner-1".to_string(), + run_id: run_id.to_string(), + topic: "tokens".to_string(), + } + } + + fn attach_request(callback_url: &str) -> AttachCallerRequest { + AttachCallerRequest { + stream: Some(stream("")), + request_id: "attach-1".to_string(), + callback_url: callback_url.to_string(), + callback_headers: [("nexus-callback-token".to_string(), "abc".to_string())].into(), + } + } + + /// A failure as the start's Nexus handler raises it: its type, whether a retry may help, and + /// the class its message leads with. + #[derive(Debug, PartialEq)] + struct Raised { + kind: String, + retryable: bool, + class: String, + } + + fn raised(failure: NexusFailure) -> Raised { + Raised { + kind: failure.handler_error_type, + retryable: failure.retryable, + class: failure.message.split(':').next().unwrap().to_string(), + } + } + + fn expect(kind: &str, retryable: bool, class: &str) -> Raised { + Raised { + kind: kind.to_string(), + retryable, + class: class.to_string(), + } + } + + async fn attach( + service: &StreamService, + request: AttachCallerRequest, + ) -> Result { + let response = AttachCallerResponse::decode( + service + .call("AttachCaller", &request.encode_to_vec()) + .await + .unwrap() + .as_slice(), + ) + .unwrap(); + match response.failure { + Some(failure) => Err(raised(failure)), + None => Ok(response.operation_token), + } + } + + async fn detach(service: &StreamService, token: &str) -> Result<(), Raised> { + let response = DetachCallerResponse::decode( + service + .call( + "DetachCaller", + &DetachCallerRequest { + namespace: "default".to_string(), + operation_token: token.to_string(), + } + .encode_to_vec(), + ) + .await + .unwrap() + .as_slice(), + ) + .unwrap(); + response + .failure + .map_or(Ok(()), |failure| Err(raised(failure))) + } + + #[tokio::test] + async fn a_start_attaches_the_callers_callback_and_hands_back_the_ref() { + let setup = setup(); + let token = attach(&setup.service, attach_request("temporal://system")) + .await + .unwrap(); + assert_eq!( + OperationToken::decode("default", &token).unwrap().stream, + stream("") + ); + let attached = setup.notifier.attached.lock().unwrap().clone(); + let [attachment] = attached.as_slice() else { + panic!("one attach, not {attached:?}"); + }; + assert_eq!( + attachment.chain, + ChainId { + namespace: "default".to_string(), + workflow_id: "owner-1".to_string(), + first_run_id: "chain-1".to_string(), + }, + "the notifier is keyed by the chain's first run" + ); + assert_eq!(attachment.topic, "tokens"); + assert_eq!(attachment.request_id, "attach-1"); + assert_eq!(attachment.callback_url, "temporal://system"); + assert_eq!( + attachment.callback_headers, + [("nexus-callback-token".to_string(), "abc".to_string())].into() + ); + } + + #[tokio::test] + async fn the_attach_carries_the_start_token_and_time() { + let setup = setup(); + let before = SystemTime::now(); + let token = attach(&setup.service, attach_request("temporal://system")) + .await + .unwrap(); + let attachment = setup.notifier.attached.lock().unwrap()[0].clone(); + assert_eq!(attachment.operation_token, token); + assert!(attachment.start_time >= before && attachment.start_time <= SystemTime::now()); + assert!(attachment.start_time > UNIX_EPOCH); + } + + #[tokio::test] + async fn a_start_after_the_workflow_id_moved_to_a_new_chain_attaches_to_the_new_one() { + let setup = setup(); + let first = attach(&setup.service, attach_request("temporal://system")) + .await + .unwrap(); + // The first chain ended, and the app started a new one on the same Workflow id. + setup + .owners + .chains + .lock() + .unwrap() + .insert("owner-1".to_string(), "chain-2".to_string()); + let second = attach( + &setup.service, + AttachCallerRequest { + request_id: "attach-2".to_string(), + ..attach_request("temporal://system") + }, + ) + .await + .unwrap(); + assert_eq!( + OperationToken::decode("default", &first) + .unwrap() + .chain_first_run_id, + "chain-1" + ); + assert_eq!( + OperationToken::decode("default", &second) + .unwrap() + .chain_first_run_id, + "chain-2" + ); + assert_eq!( + setup.notifier.attached.lock().unwrap()[1] + .chain + .first_run_id, + "chain-2" + ); + } + + #[tokio::test] + async fn a_retried_start_gets_the_same_token() { + let setup = setup(); + let first = attach(&setup.service, attach_request("temporal://system")) + .await + .unwrap(); + let again = attach(&setup.service, attach_request("temporal://system")) + .await + .unwrap(); + assert_eq!(first, again); + let attached = setup.notifier.attached.lock().unwrap().clone(); + assert_eq!( + attached[0], + CallerAttachment { + start_time: attached[0].start_time, + ..attached[1].clone() + }, + "the server keys the attach by the request id, so the second changes nothing" + ); + } + + #[tokio::test] + async fn a_pinned_ref_attaches_by_its_chains_first_run() { + let setup = setup(); + setup + .owners + .chains + .lock() + .unwrap() + .insert("owner-1".to_string(), "run-1".to_string()); + let token = attach( + &setup.service, + AttachCallerRequest { + stream: Some(stream("run-2")), + ..attach_request("temporal://system") + }, + ) + .await + .unwrap(); + assert_eq!( + OperationToken::decode("default", &token).unwrap().stream, + stream("run-2") + ); + assert_eq!(*setup.owners.described.lock().unwrap(), ["run-2"]); + assert_eq!( + setup.notifier.attached.lock().unwrap()[0] + .chain + .first_run_id, + "run-1" + ); + } + + #[tokio::test] + async fn a_start_without_a_callback_is_refused() { + let setup = setup(); + let failure = attach(&setup.service, attach_request("")) + .await + .unwrap_err(); + assert_eq!(failure, expect("BAD_REQUEST", false, "ValueError")); + assert!(setup.notifier.attached.lock().unwrap().is_empty()); + assert!(setup.owners.described.lock().unwrap().is_empty()); + } + + #[tokio::test] + async fn a_start_whose_stream_names_no_topic_is_an_input_mistake() { + let setup = setup(); + for topic in ["", "a\u{0}b"] { + let failure = attach( + &setup.service, + AttachCallerRequest { + stream: Some(StreamAddress { + topic: topic.to_string(), + ..stream("") + }), + ..attach_request("temporal://system") + }, + ) + .await + .unwrap_err(); + assert_eq!( + failure, + expect("BAD_REQUEST", false, "ValueError"), + "{topic:?}" + ); + } + assert!(setup.owners.described.lock().unwrap().is_empty()); + } + + #[tokio::test] + async fn an_input_mistake_leaves_the_callers_value_out_of_its_message() { + let setup = setup(); + let response = AttachCallerResponse::decode( + setup + .service + .call( + "AttachCaller", + &AttachCallerRequest { + stream: Some(StreamAddress { + topic: "secret-value\u{1}".to_string(), + ..stream("") + }), + ..attach_request("temporal://system") + } + .encode_to_vec(), + ) + .await + .unwrap() + .as_slice(), + ) + .unwrap(); + let failure = response.failure.unwrap(); + assert_eq!(failure.handler_error_type, "BAD_REQUEST"); + assert!( + !failure.message.contains("secret-value"), + "{}", + failure.message + ); + } + + #[tokio::test] + async fn a_start_without_a_request_id_is_refused() { + let setup = setup(); + let failure = attach( + &setup.service, + AttachCallerRequest { + request_id: String::new(), + ..attach_request("temporal://system") + }, + ) + .await + .unwrap_err(); + assert_eq!(failure, expect("BAD_REQUEST", false, "ValueError")); + assert!(setup.notifier.attached.lock().unwrap().is_empty()); + } + + #[tokio::test] + async fn a_cancel_detaches_the_callback_it_attached() { + let setup = setup(); + let token = attach(&setup.service, attach_request("temporal://system")) + .await + .unwrap(); + detach(&setup.service, &token).await.unwrap(); + let attachment = setup.notifier.attached.lock().unwrap()[0].clone(); + assert_eq!( + *setup.notifier.detached.lock().unwrap(), + [( + attachment.chain, + "tokens".to_string(), + "attach-1".to_string() + )] + ); + } + + #[tokio::test] + async fn a_cancel_detaches_from_the_chain_it_attached_to() { + let setup = setup(); + let token = attach(&setup.service, attach_request("temporal://system")) + .await + .unwrap(); + // The Workflow id moved on to a new chain since the start. + setup + .owners + .chains + .lock() + .unwrap() + .insert("owner-1".to_string(), "chain-2".to_string()); + detach(&setup.service, &token).await.unwrap(); + assert_eq!( + setup.notifier.detached.lock().unwrap()[0].0.first_run_id, + "chain-1" + ); + assert_eq!( + setup.owners.described.lock().unwrap().len(), + 1, + "a cancel describes nothing" + ); + } + + #[tokio::test] + async fn a_cancel_with_a_token_that_names_no_stream_is_refused() { + let setup = setup(); + for token in vectors().refused { + let failure = detach(&setup.service, &token).await.unwrap_err(); + assert_eq!( + failure, + expect("BAD_REQUEST", false, "ValueError"), + "{token:?}" + ); + } + assert!(setup.notifier.detached.lock().unwrap().is_empty()); + } + + #[tokio::test] + async fn the_servers_answer_crosses_as_the_nexus_calls_cross_it() { + for (answer, crossed) in [ + ( + NotifyError::Failed("down".to_string()), + expect("UNAVAILABLE", true, "StreamStorageError"), + ), + ( + NotifyError::Refused("no".to_string()), + expect("INTERNAL", false, "StreamRefusedError"), + ), + ( + NotifyError::Unsupported("no notifier".to_string()), + expect("NOT_IMPLEMENTED", false, "StreamUnsupportedError"), + ), + ] { + let setup = setup(); + *setup.notifier.answer.lock().unwrap() = Some(answer.clone()); + let failure = attach(&setup.service, attach_request("temporal://system")) + .await + .unwrap_err(); + assert_eq!(failure, crossed); + *setup.notifier.answer.lock().unwrap() = Some(answer); + let token = OperationToken { + stream: stream(""), + attach_request_id: "attach-1".to_string(), + chain_first_run_id: "chain-1".to_string(), + } + .encode(); + assert_eq!(detach(&setup.service, &token).await.unwrap_err(), crossed); + } + } + + #[tokio::test] + async fn a_missing_owner_crosses_as_not_found() { + let setup = setup(); + setup + .owners + .missing + .store(true, std::sync::atomic::Ordering::SeqCst); + let failure = attach(&setup.service, attach_request("temporal://system")) + .await + .unwrap_err(); + assert_eq!(failure, expect("NOT_FOUND", false, "StreamNotFoundError")); + } + + #[tokio::test] + async fn a_service_without_a_client_cant_attach() { + let service = StreamService::new( + Arc::new(MemoryStore::new()), + Arc::new(Owners::default()), + StreamsOptions::default(), + ); + let failure = attach(&service, attach_request("temporal://system")) + .await + .unwrap_err(); + assert_eq!( + failure, + expect("NOT_IMPLEMENTED", false, "StreamUnsupportedError") + ); + } + + #[derive(Deserialize)] + struct Vectors { + vectors: Vec, + decode_only: Vec, + refused: Vec, + } + + #[derive(Deserialize)] + struct Vector { + #[serde(rename = "ref")] + stream: TokenRef, + attach: String, + chain: String, + token: String, + } + + fn vectors() -> Vectors { + serde_json::from_str(include_str!("../../nexus/operation_token_vectors.json")).unwrap() + } + + fn token_of(vector: &Vector) -> OperationToken { + OperationToken { + stream: StreamAddress { + namespace: "ns".to_string(), + owner_kind: StreamOwnerKind::Workflow as i32, + workflow_id: vector.stream.workflow_id.clone(), + run_id: vector.stream.run_id.clone().unwrap_or_default(), + topic: vector.stream.topic.clone(), + }, + attach_request_id: vector.attach.clone(), + chain_first_run_id: vector.chain.clone(), + } + } + + #[test] + fn a_token_encodes_as_the_shared_vectors_say() { + for vector in vectors().vectors { + assert_eq!(token_of(&vector).encode(), vector.token, "{}", vector.token); + } + } + + #[test] + fn a_token_decodes_as_the_shared_vectors_say() { + for vector in vectors().vectors { + assert_eq!( + OperationToken::decode("ns", &vector.token).unwrap(), + token_of(&vector) + ); + let padded = format!( + "{}{}", + vector.token, + "=".repeat((4 - vector.token.len() % 4) % 4) + ); + assert_eq!( + OperationToken::decode("ns", &padded).unwrap(), + token_of(&vector), + "padding is optional" + ); + } + } + + #[test] + fn a_token_from_before_core_still_decodes() { + for vector in vectors().decode_only { + assert_eq!( + OperationToken::decode("ns", &vector.token).unwrap(), + token_of(&vector) + ); + } + } + + #[test] + fn a_token_that_names_no_stream_is_refused() { + for token in vectors().refused { + let error = OperationToken::decode("ns", &token).unwrap_err(); + assert_eq!( + error.kind, + crate::proto::StreamFailureKind::InvalidArgument, + "{token:?}" + ); + } + } +} diff --git a/crates/streams/src/notify.rs b/crates/streams/src/notify.rs index ef884545be..adc517122c 100644 --- a/crates/streams/src/notify.rs +++ b/crates/streams/src/notify.rs @@ -115,8 +115,27 @@ pub struct NotifierPoll { pub close_failure: Option, } +/// One Nexus caller's callback, to attach to a stream's notifier. +#[derive(Debug, Clone, PartialEq)] +pub struct CallerAttachment { + /// The run chain whose stream it is. Its first run keys the notifier. + pub chain: ChainId, + /// The stream's topic. + pub topic: String, + /// The start's request id. The attach is idempotent by it, and a detach names it. + pub request_id: String, + /// Where the server delivers progress and the completion. + pub callback_url: String, + /// Headers the server sends with each delivery. + pub callback_headers: std::collections::HashMap, + /// The operation's token, which the server hands back with each delivery. + pub operation_token: String, + /// When the operation started, which the server hands back too. + pub start_time: std::time::SystemTime, +} + /// Sends notifications to the stream notifier on the server, attaches and detaches the -/// Workflows it wakes, and waits on it for clients. +/// Workflows it wakes and the Nexus callers it completes, and waits on it for clients. #[async_trait::async_trait] pub trait NotifierClient: Send + Sync { /// Sends one notification. @@ -130,6 +149,14 @@ pub trait NotifierClient: Send + Sync { )) } + /// Attaches a Nexus caller's callback to a stream's notifier. Idempotent by its request id. + async fn attach_caller(&self, attachment: CallerAttachment) -> Result<(), NotifyError> { + let _ = attachment; + Err(NotifyError::Unsupported( + "this client can't attach callers".to_string(), + )) + } + /// Detaches what the attach with `request_id` attached to `chain`'s stream on `topic`. async fn detach( &self, diff --git a/crates/streams/src/service.rs b/crates/streams/src/service.rs index eb88d205d1..0299794940 100644 --- a/crates/streams/src/service.rs +++ b/crates/streams/src/service.rs @@ -5,10 +5,10 @@ use crate::{ StreamResult, StreamStore, Streams, StreamsOptions, nexus::{NexusCalls, NexusOptions}, proto::{ - AppendRequest, CloseRequest, DeleteOwnerRequest, FlushNotificationsRequest, - FlushNotificationsResponse, LatestRequest, NexusCall, NexusResumeRequest, ReadRequest, - RetireProducerRequest, StreamFailure, StreamStoreConfig, SubscribeRequest, - UnsubscribeRequest, WaitForNotificationRequest, stream_store_config, + AppendRequest, AttachCallerRequest, CloseRequest, DeleteOwnerRequest, DetachCallerRequest, + FlushNotificationsRequest, FlushNotificationsResponse, LatestRequest, NexusCall, + NexusResumeRequest, ReadRequest, RetireProducerRequest, StreamFailure, StreamStoreConfig, + SubscribeRequest, UnsubscribeRequest, WaitForNotificationRequest, stream_store_config, }, subscribe::Subscriptions, }; @@ -32,6 +32,7 @@ pub struct StreamService { streams: Arc, subscriptions: Subscriptions, nexus: NexusCalls, + callers: Option>, } impl std::fmt::Debug for StreamService { @@ -54,7 +55,8 @@ impl StreamService { /// Like [StreamService::connect], closing each stream's notifier through `notifier` after /// the store closes the stream, sending progress too when the config turns it on, attaching - /// and detaching Workflow readers through it, and waiting on it for client readers. + /// and detaching Workflow readers and stream operation callers through it, and waiting on it + /// for client readers. pub async fn connect_with_notifier( config: StreamStoreConfig, owner: Arc, @@ -92,7 +94,9 @@ impl StreamService { }; let service = Self::new(store, owner, StreamsOptions::for_retention(retention)); Ok(match notifier { - Some(client) => service.with_subscriptions(client), + Some(client) => service + .with_subscriptions(client.clone()) + .with_callers(client), None => service, }) } @@ -110,6 +114,7 @@ impl StreamService { streams, subscriptions: Subscriptions::default(), nexus, + callers: None, } } @@ -119,6 +124,12 @@ impl StreamService { self } + /// Attaches and detaches stream operation callers through `client`. + pub fn with_callers(mut self, client: Arc) -> Self { + self.callers = Some(client); + self + } + /// Serves one call: `rpc` is a `StreamService` method name and `request` its serialized /// request. Answers with the serialized response, or the failure the call ended with. pub async fn call(&self, rpc: &str, request: &[u8]) -> Result, StreamFailure> { @@ -200,6 +211,20 @@ impl StreamService { .resume(decode::(rpc, request)?) .await .encode_to_vec()), + // Like the Nexus calls, a caller call answers its failure in the response. + "AttachCaller" => Ok(crate::nexus::attach_caller( + &self.streams, + self.callers.as_ref(), + decode::(rpc, request)?, + ) + .await + .encode_to_vec()), + "DetachCaller" => Ok(crate::nexus::detach_caller( + self.callers.as_ref(), + decode::(rpc, request)?, + ) + .await + .encode_to_vec()), other => Err(StreamError::unsupported(format!( "the stream service has no call {other:?}" ))), From 4842388b20939b7ddc2cb6810be1aab1919bb372 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Tue, 13 Oct 2026 22:45:14 -0700 Subject: [PATCH 2/5] Armed a stream from a Nexus read that waits and finds nothing. The Nexus read takes arm and answers armed, so a Nexus caller sleeps until its operation's progress says that the stream moved. --- .../nexus/temporal_streams.nexusrpc.yaml | 13 +++ crates/streams/src/nexus/mod.rs | 8 +- crates/streams/src/nexus/wire.rs | 4 + crates/streams/tests/nexus_contract.rs | 2 + crates/streams/tests/nexus_service.rs | 108 +++++++++++++++++- 5 files changed, 133 insertions(+), 2 deletions(-) diff --git a/crates/streams/nexus/temporal_streams.nexusrpc.yaml b/crates/streams/nexus/temporal_streams.nexusrpc.yaml index f34570d8c3..f160afec35 100644 --- a/crates/streams/nexus/temporal_streams.nexusrpc.yaml +++ b/crates/streams/nexus/temporal_streams.nexusrpc.yaml @@ -188,6 +188,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 @@ -279,6 +286,12 @@ $defs: also when the answer was cut short by its byte budget. It holds at most 64 KiB, which is about 87 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 a17e29a289..00c40a88ae 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)), }; @@ -340,7 +341,7 @@ impl NexusCalls { wait: wait.try_into().ok(), max_records: max_records as u32, state: state.clone(), - arm: false, + arm: input.arm.unwrap_or(false), }; let mut read = match self.streams.read_carried(request).await { Ok(read) => read, @@ -613,6 +614,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 { @@ -687,6 +689,8 @@ fn finish_read( wires.push(wire); kept.extend(stored); } + // An answer the budget cut holds records, so only a whole empty answer stays armed. + let armed = response.armed && !cut && wires.is_empty(); 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()) { @@ -707,6 +711,7 @@ fn finish_read( next_token, done, state: Some(Base64(state.encode_to_vec())), + armed, }) } @@ -780,6 +785,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 5c9ba45b6b..6cef74ee41 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 ce8790c061..300e9b6c98 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 38e9581c54..c4f7d80692 100644 --- a/crates/streams/tests/nexus_service.rs +++ b/crates/streams/tests/nexus_service.rs @@ -22,7 +22,7 @@ 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, StoreAppendRequest, StreamRecord, StreamRecordKind}, }; static IDS: AtomicUsize = AtomicUsize::new(0); @@ -633,6 +633,108 @@ 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 + } + + /// Appends one record to the stream through the store, answering the arm it reported. + async fn arm_reported(store: &Arc, wf: &str, sequence: i64) -> Option { + store + .append(StoreAppendRequest { + chain: Some(ChainId { + namespace: NAMESPACE.to_string(), + workflow_id: wf.to_string(), + first_run_id: "run-1".to_string(), + }), + topic: TOPIC.to_string(), + producer_id: "store".to_string(), + attempt: 1, + sequence, + digest: vec![sequence as u8], + records: vec![ + StreamRecord { + topic: TOPIC.to_string(), + ..Default::default() + } + .encode_to_vec(), + ], + }) + .await + .unwrap() + .arm + } + + pub(crate) async fn an_empty_read_that_asks_to_arm_answers_armed(store: Arc) { + let calls = service(store.clone(), NexusOptions::default()); + let wf = workflow(); + let answer = read(&calls, arming(&wf, "")).await.unwrap(); + assert!(records(&answer).is_empty()); + assert_eq!(answer["armed"], true, "{answer}"); + assert_eq!(answer["done"], false); + assert!( + arm_reported(&store, &wf, 1) + .await + .is_some_and(|arm| arm > 0) + ); + } + + pub(crate) async fn an_answer_with_records_never_says_armed(store: Arc) { + let calls = service(store.clone(), NexusOptions::default()); + let wf = workflow(); + write(&calls, &wf, 1, &["one"]).await; + let answer = read(&calls, arming(&wf, "")).await.unwrap(); + assert_eq!(texts(&answer), ["one"]); + assert!( + answer.get("armed").is_none_or(|armed| armed == false), + "{answer}" + ); + assert_eq!(arm_reported(&store, &wf, 1).await, Some(0)); + } + + pub(crate) async fn a_done_answer_never_arms(store: Arc) { + let wf = workflow(); + write( + &service(store.clone(), NexusOptions::default()), + &wf, + 1, + &["one"], + ) + .await; + let ended = NexusCalls::new( + Arc::new(Streams::new( + store, + Arc::new(Ended), + StreamsOptions::default(), + )), + NexusOptions::default(), + ); + let mut after = String::new(); + for _ in 0..10 { + let mut input = arming(&wf, &after); + input["wait_ms"] = json!(5000); + let answer = read(&ended, input).await.unwrap(); + if answer["done"] == true { + assert!( + answer.get("armed").is_none_or(|armed| armed == false), + "{answer}" + ); + return; + } + after = token(&answer).to_string(); + } + panic!("the read never said done"); + } + + pub(crate) async fn a_read_without_arm_never_arms(store: Arc) { + let calls = service(store.clone(), NexusOptions::default()); + let wf = workflow(); + let answer = read_after(&calls, &wf, "").await; + assert!(records(&answer).is_empty()); + assert!(answer.get("armed").is_none(), "{answer}"); + assert_eq!(arm_reported(&store, &wf, 1).await, Some(0)); + } } macro_rules! lanes { @@ -671,6 +773,10 @@ 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, + an_empty_read_that_asks_to_arm_answers_armed, + an_answer_with_records_never_says_armed, + a_done_answer_never_arms, + a_read_without_arm_never_arms, ); }; } From af5a195d1d1bab2e6026c25eb673f1538d977778 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Tue, 13 Oct 2026 22:50:10 -0700 Subject: [PATCH 3/5] Vendored the System Nexus WIT for the stream notifier RPCs. AttachStreamCallback, DetachStreamCallback and NotifyStream carry the exposed tag again, with the WIT and the model refs that the SDK generators read. It matches api moe/bridge/v2/n3-notifier-wit. --- crates/protos/protos/api_upstream/Makefile | 11 ++- .../nexus/temporal-proto-models-nexusrpc.yaml | 45 +++++++++ .../api_upstream/nexus/workflow-service.wit | 97 +++++++++++++++++++ .../api/workflowservice/v1/service.proto | 18 ++-- 4 files changed, 161 insertions(+), 10 deletions(-) diff --git a/crates/protos/protos/api_upstream/Makefile b/crates/protos/protos/api_upstream/Makefile index f6fcf899d8..39f8b6b25a 100644 --- a/crates/protos/protos/api_upstream/Makefile +++ b/crates/protos/protos/api_upstream/Makefile @@ -35,7 +35,12 @@ PROTO_PATHS = paths=source_relative:$(PROTO_OUT) OAPI_OUT := openapi OAPI3_PATH := .components.schemas.Payload -NEX_GEN ?= nex-gen +# nexgen escapes WIT keywords in field names and scaffolds proto maps and bytes in the +# forms its generators accept. Pinned by revision so every machine writes the same WIT. +NEXGEN_GIT ?= https://github.com/moetemp/nexgen +NEXGEN_REV ?= 6f46948678f764c68bb76f103ab43713047187b2 +NEXGEN_ROOT := $(STAMPDIR)/nexgen-$(NEXGEN_REV) +NEX_GEN := $(NEXGEN_ROOT)/bin/nexgen $(PROTO_OUT): mkdir $(PROTO_OUT) @@ -156,8 +161,8 @@ system-nexus-wit-install: @cd cmd/protoc-gen-system-nexus-wit && go install . nex-gen-install: - printf $(COLOR) "Install nex-gen if missing..." - command -v $(NEX_GEN) >/dev/null || cargo install nex-gen --version ^0.1 + printf $(COLOR) "Install nexgen $(NEXGEN_REV) if missing..." + command -v $(NEX_GEN) >/dev/null || CARGO_NET_GIT_FETCH_WITH_CLI=true cargo install --locked --git $(NEXGEN_GIT) --rev $(NEXGEN_REV) --features advanced --root $(NEXGEN_ROOT) nexgen ##### Clean ##### clean: diff --git a/crates/protos/protos/api_upstream/nexus/temporal-proto-models-nexusrpc.yaml b/crates/protos/protos/api_upstream/nexus/temporal-proto-models-nexusrpc.yaml index ebc74b68b8..e611f78575 100644 --- a/crates/protos/protos/api_upstream/nexus/temporal-proto-models-nexusrpc.yaml +++ b/crates/protos/protos/api_upstream/nexus/temporal-proto-models-nexusrpc.yaml @@ -17,3 +17,48 @@ services: $pythonRef: temporalio.api.workflowservice.v1.SignalWithStartWorkflowExecutionResponse $rubyRef: Temporalio::Api::WorkflowService::V1::SignalWithStartWorkflowExecutionResponse $typescriptRef: '@temporalio/api/workflowservice/v1.SignalWithStartWorkflowExecutionResponse' + AttachStreamCallback: + input: + $dotnetRef: Temporalio.Api.WorkflowService.V1.AttachStreamCallbackRequest + $goRef: go.temporal.io/api/workflowservice/v1.AttachStreamCallbackRequest + $javaRef: io.temporal.api.workflowservice.v1.AttachStreamCallbackRequest + $pythonRef: temporalio.api.workflowservice.v1.AttachStreamCallbackRequest + $rubyRef: Temporalio::Api::WorkflowService::V1::AttachStreamCallbackRequest + $typescriptRef: '@temporalio/api/workflowservice/v1.AttachStreamCallbackRequest' + output: + $dotnetRef: Temporalio.Api.WorkflowService.V1.AttachStreamCallbackResponse + $goRef: go.temporal.io/api/workflowservice/v1.AttachStreamCallbackResponse + $javaRef: io.temporal.api.workflowservice.v1.AttachStreamCallbackResponse + $pythonRef: temporalio.api.workflowservice.v1.AttachStreamCallbackResponse + $rubyRef: Temporalio::Api::WorkflowService::V1::AttachStreamCallbackResponse + $typescriptRef: '@temporalio/api/workflowservice/v1.AttachStreamCallbackResponse' + DetachStreamCallback: + input: + $dotnetRef: Temporalio.Api.WorkflowService.V1.DetachStreamCallbackRequest + $goRef: go.temporal.io/api/workflowservice/v1.DetachStreamCallbackRequest + $javaRef: io.temporal.api.workflowservice.v1.DetachStreamCallbackRequest + $pythonRef: temporalio.api.workflowservice.v1.DetachStreamCallbackRequest + $rubyRef: Temporalio::Api::WorkflowService::V1::DetachStreamCallbackRequest + $typescriptRef: '@temporalio/api/workflowservice/v1.DetachStreamCallbackRequest' + output: + $dotnetRef: Temporalio.Api.WorkflowService.V1.DetachStreamCallbackResponse + $goRef: go.temporal.io/api/workflowservice/v1.DetachStreamCallbackResponse + $javaRef: io.temporal.api.workflowservice.v1.DetachStreamCallbackResponse + $pythonRef: temporalio.api.workflowservice.v1.DetachStreamCallbackResponse + $rubyRef: Temporalio::Api::WorkflowService::V1::DetachStreamCallbackResponse + $typescriptRef: '@temporalio/api/workflowservice/v1.DetachStreamCallbackResponse' + NotifyStream: + input: + $dotnetRef: Temporalio.Api.WorkflowService.V1.NotifyStreamRequest + $goRef: go.temporal.io/api/workflowservice/v1.NotifyStreamRequest + $javaRef: io.temporal.api.workflowservice.v1.NotifyStreamRequest + $pythonRef: temporalio.api.workflowservice.v1.NotifyStreamRequest + $rubyRef: Temporalio::Api::WorkflowService::V1::NotifyStreamRequest + $typescriptRef: '@temporalio/api/workflowservice/v1.NotifyStreamRequest' + output: + $dotnetRef: Temporalio.Api.WorkflowService.V1.NotifyStreamResponse + $goRef: go.temporal.io/api/workflowservice/v1.NotifyStreamResponse + $javaRef: io.temporal.api.workflowservice.v1.NotifyStreamResponse + $pythonRef: temporalio.api.workflowservice.v1.NotifyStreamResponse + $rubyRef: Temporalio::Api::WorkflowService::V1::NotifyStreamResponse + $typescriptRef: '@temporalio/api/workflowservice/v1.NotifyStreamResponse' diff --git a/crates/protos/protos/api_upstream/nexus/workflow-service.wit b/crates/protos/protos/api_upstream/nexus/workflow-service.wit index caf3842a3b..fa0aa9aae2 100644 --- a/crates/protos/protos/api_upstream/nexus/workflow-service.wit +++ b/crates/protos/protos/api_upstream/nexus/workflow-service.wit @@ -138,4 +138,101 @@ interface workflow-service { signal-with-start-workflow: func( request: signal-with-start-workflow-request, ) -> signal-with-start-workflow-response; + /// @nexus.proto "temporal.api.enums.v1.StreamOwnerKind" + enum stream-owner-kind { + stream-owner-kind-unspecified, + stream-owner-kind-workflow, + } + + /// @nexus.proto "temporal.api.stream.v1.StreamReference" + record stream-reference { + owner-kind: option, + workflow-id: string, + /// @nexus.doc "The id of the first run in the owning Workflow's run chain. It is required, since the server refuses an empty run id." + run-id: option, + topic: string, + } + + /// @nexus.proto "temporal.api.common.v1.Callback.Nexus" + record callback-nexus { + url: string, + header: map, + } + + /// @nexus.proto "temporal.api.workflowservice.v1.AttachStreamCallbackRequest" + record attach-stream-callback-request { + /// @nexus.source python="workflow_namespace()" typescript="workflowNamespace()" go="workflow.GetInfo(ctx).Namespace" dotnet="TemporalWorkflowContext.WorkflowNamespace()" + /// @nexus.doc "Namespace of the stream." + namespace: string, + stream-ref: option, + request-id: string, + callback: option, + /// @nexus.omit + identity: placeholder, + operation-token: string, + start-time: option, + /// @nexus.omit + workflow-target: placeholder, + } + + /// @nexus.proto "temporal.api.workflowservice.v1.AttachStreamCallbackResponse" + record attach-stream-callback-response { + } + + attach-stream-callback: func( + request: attach-stream-callback-request, + ) -> attach-stream-callback-response; + /// @nexus.proto "temporal.api.workflowservice.v1.DetachStreamCallbackRequest" + record detach-stream-callback-request { + /// @nexus.source python="workflow_namespace()" typescript="workflowNamespace()" go="workflow.GetInfo(ctx).Namespace" dotnet="TemporalWorkflowContext.WorkflowNamespace()" + /// @nexus.doc "Namespace of the stream." + namespace: string, + stream-ref: option, + request-id: string, + /// @nexus.omit + identity: placeholder, + } + + /// @nexus.proto "temporal.api.workflowservice.v1.DetachStreamCallbackResponse" + record detach-stream-callback-response { + } + + detach-stream-callback: func( + request: detach-stream-callback-request, + ) -> detach-stream-callback-response; + use nexus:temporal-types/model@1.0.0.{ + payload, + }; + + /// @nexus.proto "temporal.api.workflowservice.v1.NotifyStreamRequest" + record notify-stream-request { + /// @nexus.source python="workflow_namespace()" typescript="workflowNamespace()" go="workflow.GetInfo(ctx).Namespace" dotnet="TemporalWorkflowContext.WorkflowNamespace()" + /// @nexus.doc "Namespace of the stream." + namespace: string, + stream-ref: option, + position: string, + counter: option, + metadata: map, + close: option, + close-result: option, + /// @nexus.omit + identity: placeholder, + close-failure: option, + } + + /// @nexus.proto "temporal.api.workflowservice.v1.NotifyStreamResponse" + record notify-stream-response { + } + + notify-stream: func( + request: notify-stream-request, + ) -> notify-stream-response; + /// @nexus.proto "google.protobuf.Timestamp" + record timestamp { + seconds: option, + nanos: option, + } + use nexus:temporal-types/model@1.0.0.{ + failure, + }; } diff --git a/crates/protos/protos/api_upstream/temporal/api/workflowservice/v1/service.proto b/crates/protos/protos/api_upstream/temporal/api/workflowservice/v1/service.proto index 7aec05b140..25cb8db62b 100644 --- a/crates/protos/protos/api_upstream/temporal/api/workflowservice/v1/service.proto +++ b/crates/protos/protos/api_upstream/temporal/api/workflowservice/v1/service.proto @@ -2045,8 +2045,9 @@ service WorkflowService { // operation returns a stream calls it with the caller's callback when the operation starts. // // (-- api-linter: core::0127::http-annotation=disabled - // aip.dev/not-precedent: SDKs reach the stream notifier through gRPC. --) + // aip.dev/not-precedent: SDKs reach the stream notifier through System Nexus or gRPC. --) rpc AttachStreamCallback (AttachStreamCallbackRequest) returns (AttachStreamCallbackResponse) { + option (nexusannotations.v1.operation).tags = "exposed"; } // DetachStreamCallback removes a callback from the notifier of a stream. The notifier completes @@ -2054,8 +2055,9 @@ service WorkflowService { // when the caller cancels the operation. // // (-- api-linter: core::0127::http-annotation=disabled - // aip.dev/not-precedent: SDKs reach the stream notifier through gRPC. --) + // aip.dev/not-precedent: SDKs reach the stream notifier through System Nexus or gRPC. --) rpc DetachStreamCallback (DetachStreamCallbackRequest) returns (DetachStreamCallbackResponse) { + option (nexusannotations.v1.operation).tags = "exposed"; } // NotifyStream tells the notifier of a stream that the stream moved, or that it closed. The @@ -2063,25 +2065,27 @@ service WorkflowService { // completes them. Producers fold their notifications and keep at most one call in flight. // // (-- api-linter: core::0127::http-annotation=disabled - // aip.dev/not-precedent: SDKs reach the stream notifier through gRPC. --) + // aip.dev/not-precedent: SDKs reach the stream notifier through System Nexus or gRPC. --) rpc NotifyStream (NotifyStreamRequest) returns (NotifyStreamResponse) { + option (nexusannotations.v1.operation).tags = "exposed"; } // DescribeStreamNotifier returns the state of the notifier of a stream and its callbacks. It is - // for operators and tools. + // for operators and tools, so it is not exposed through System Nexus. // // (-- api-linter: core::0127::http-annotation=disabled - // aip.dev/not-precedent: SDKs reach the stream notifier through gRPC. --) + // aip.dev/not-precedent: SDKs reach the stream notifier through System Nexus or gRPC. --) rpc DescribeStreamNotifier (DescribeStreamNotifierRequest) returns (DescribeStreamNotifierResponse) { } // PollStreamNotifier waits until the notifier of a stream has a counter above `after_counter`, // or the stream is closed. It returns with no change after a wait limit set by the server // (`streamnotifier.pollMaxWait`). A client that reads a stream calls it after a read that - // found nothing and armed the topic, in place of another blocking read on the store. + // found nothing and armed the topic, in place of another blocking read on the store. It is a + // long poll, so it is not exposed through System Nexus. // // (-- api-linter: core::0127::http-annotation=disabled - // aip.dev/not-precedent: SDKs reach the stream notifier through gRPC. --) + // aip.dev/not-precedent: SDKs reach the stream notifier through System Nexus or gRPC. --) rpc PollStreamNotifier (PollStreamNotifierRequest) returns (PollStreamNotifierResponse) { } } From 782a464a1aa0b532a9fba98f5e301322df67a5e8 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Tue, 13 Oct 2026 23:13:21 -0700 Subject: [PATCH 4/5] Named the token ref's chain by chain_run_id, as the contract does. The vectors take the renamed key, which sorts first in the ref. A token that still names run_id is refused, since the ref model is closed. --- crates/streams/nexus/operation_token_vectors.json | 13 +++++++------ crates/streams/src/nexus/operation.rs | 14 +++++++------- 2 files changed, 14 insertions(+), 13 deletions(-) diff --git a/crates/streams/nexus/operation_token_vectors.json b/crates/streams/nexus/operation_token_vectors.json index 90b1ef636d..0178de9ab0 100644 --- a/crates/streams/nexus/operation_token_vectors.json +++ b/crates/streams/nexus/operation_token_vectors.json @@ -1,5 +1,5 @@ { - "comment": "Token v2 of a stream-returning Nexus operation: base64url JSON, no padding, keys sorted, non-ASCII escaped, and no run_id for an unpinned ref. Core's encoder and every SDK's encoder write the vectors, and every decoder reads them. Decoders also read decode_only, the null run_id that SDK releases before Core wrote.", + "comment": "Token v2 of a stream-returning Nexus operation: base64url JSON, no padding, keys sorted, non-ASCII escaped, and no chain_run_id for a ref to the newest chain. Core's encoder and every SDK's encoder write the vectors, and every decoder reads them. Decoders also read decode_only, the null chain_run_id that SDK releases before Core wrote.", "vectors": [ { "ref": { @@ -15,12 +15,12 @@ "ref": { "kind": "workflow", "workflow_id": "owner-1", - "run_id": "run-2", + "chain_run_id": "run-2", "topic": "tokens" }, "attach": "attach-1", "chain": "run-1", - "token": "eyJhdHRhY2giOiJhdHRhY2gtMSIsImNoYWluIjoicnVuLTEiLCJyZWYiOnsia2luZCI6IndvcmtmbG93IiwicnVuX2lkIjoicnVuLTIiLCJ0b3BpYyI6InRva2VucyIsIndvcmtmbG93X2lkIjoib3duZXItMSJ9LCJ2IjoyfQ" + "token": "eyJhdHRhY2giOiJhdHRhY2gtMSIsImNoYWluIjoicnVuLTEiLCJyZWYiOnsiY2hhaW5fcnVuX2lkIjoicnVuLTIiLCJraW5kIjoid29ya2Zsb3ciLCJ0b3BpYyI6InRva2VucyIsIndvcmtmbG93X2lkIjoib3duZXItMSJ9LCJ2IjoyfQ" }, { "ref": { @@ -58,12 +58,12 @@ "ref": { "kind": "workflow", "workflow_id": "owner-1", - "run_id": null, + "chain_run_id": null, "topic": "tokens" }, "attach": "attach-1", "chain": "chain-1", - "token": "eyJhdHRhY2giOiJhdHRhY2gtMSIsImNoYWluIjoiY2hhaW4tMSIsInJlZiI6eyJraW5kIjoid29ya2Zsb3ciLCJydW5faWQiOm51bGwsInRvcGljIjoidG9rZW5zIiwid29ya2Zsb3dfaWQiOiJvd25lci0xIn0sInYiOjJ9" + "token": "eyJhdHRhY2giOiJhdHRhY2gtMSIsImNoYWluIjoiY2hhaW4tMSIsInJlZiI6eyJjaGFpbl9ydW5faWQiOm51bGwsImtpbmQiOiJ3b3JrZmxvdyIsInRvcGljIjoidG9rZW5zIiwid29ya2Zsb3dfaWQiOiJvd25lci0xIn0sInYiOjJ9" } ], "refused": [ @@ -72,6 +72,7 @@ "eyJ2IjogMn0", "eyJ2IjogMSwgImF0dGFjaCI6ICJhdHRhY2gtMSIsICJjaGFpbiI6ICJjaGFpbi0xIiwgInJlZiI6IHsia2luZCI6ICJ3b3JrZmxvdyIsICJ3b3JrZmxvd19pZCI6ICJvd25lci0xIiwgInJ1bl9pZCI6IG51bGwsICJ0b3BpYyI6ICJ0b2tlbnMifX0=", "eyJhdHRhY2giOiJhIiwiY2hhaW4iOiJjIiwiZXh0cmEiOjEsInJlZiI6eyJraW5kIjoid29ya2Zsb3ciLCJ0b3BpYyI6InQiLCJ3b3JrZmxvd19pZCI6IncifSwidiI6Mn0", - "eyJhdHRhY2giOiJhIiwiY2hhaW4iOiJjIiwicmVmIjp7ImN1cnNvciI6IngiLCJraW5kIjoid29ya2Zsb3ciLCJ0b3BpYyI6InQiLCJ3b3JrZmxvd19pZCI6IncifSwidiI6Mn0" + "eyJhdHRhY2giOiJhIiwiY2hhaW4iOiJjIiwicmVmIjp7ImN1cnNvciI6IngiLCJraW5kIjoid29ya2Zsb3ciLCJ0b3BpYyI6InQiLCJ3b3JrZmxvd19pZCI6IncifSwidiI6Mn0", + "eyJhdHRhY2giOiJhdHRhY2gtMSIsImNoYWluIjoicnVuLTEiLCJyZWYiOnsia2luZCI6IndvcmtmbG93IiwicnVuX2lkIjoicnVuLTIiLCJ0b3BpYyI6InRva2VucyIsIndvcmtmbG93X2lkIjoib3duZXItMSJ9LCJ2IjoyfQ" ] } diff --git a/crates/streams/src/nexus/operation.rs b/crates/streams/src/nexus/operation.rs index 6edf7161c1..e804ff8f53 100644 --- a/crates/streams/src/nexus/operation.rs +++ b/crates/streams/src/nexus/operation.rs @@ -178,11 +178,11 @@ struct TokenBody { #[derive(Serialize, Deserialize)] #[serde(deny_unknown_fields)] struct TokenRef { - kind: String, - // An unpinned ref has none, as the generated SDK writers leave an unset member out. SDK - // releases before Core wrote null, which reads the same. + // A ref to the newest chain has none, as the generated SDK writers leave an unset member out. + // SDK releases before Core wrote null, which reads the same. #[serde(default, skip_serializing_if = "Option::is_none")] - run_id: Option, + chain_run_id: Option, + kind: String, topic: String, workflow_id: String, } @@ -196,8 +196,8 @@ impl OperationToken { attach: self.attach_request_id.clone(), chain: self.chain_first_run_id.clone(), stream: TokenRef { + chain_run_id: (!self.stream.run_id.is_empty()).then(|| self.stream.run_id.clone()), kind: WORKFLOW_KIND.to_string(), - run_id: (!self.stream.run_id.is_empty()).then(|| self.stream.run_id.clone()), topic: self.stream.topic.clone(), workflow_id: self.stream.workflow_id.clone(), }, @@ -234,7 +234,7 @@ impl OperationToken { namespace: namespace.to_string(), owner_kind: StreamOwnerKind::Workflow as i32, workflow_id: body.stream.workflow_id, - run_id: body.stream.run_id.unwrap_or_default(), + run_id: body.stream.chain_run_id.unwrap_or_default(), topic: body.stream.topic, }, attach_request_id: body.attach, @@ -832,7 +832,7 @@ mod tests { namespace: "ns".to_string(), owner_kind: StreamOwnerKind::Workflow as i32, workflow_id: vector.stream.workflow_id.clone(), - run_id: vector.stream.run_id.clone().unwrap_or_default(), + run_id: vector.stream.chain_run_id.clone().unwrap_or_default(), topic: vector.stream.topic.clone(), }, attach_request_id: vector.attach.clone(), From 0f1bdd0eeaa28ed452e5976ad3222519b5ed3717 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 16 Oct 2026 06:25:15 -0700 Subject: [PATCH 5/5] Kept a retried stream-returning start on its first try's chain. A retry after a new chain took the Workflow id attached a second time and minted another token. --- crates/streams/src/nexus/operation.rs | 50 ++++++++++++++++++++++++--- crates/streams/src/service.rs | 4 +++ 2 files changed, 49 insertions(+), 5 deletions(-) diff --git a/crates/streams/src/nexus/operation.rs b/crates/streams/src/nexus/operation.rs index 8f6f6ee4f4..d33144e4ef 100644 --- a/crates/streams/src/nexus/operation.rs +++ b/crates/streams/src/nexus/operation.rs @@ -6,7 +6,9 @@ //! it must stay byte for byte what `crates/streams/nexus/operation_token_vectors.json` holds. use crate::{ - CallerAttachment, NotifierClient, NotifyError, StreamError, StreamResult, Streams, + CallerAttachment, NotifierClient, NotifyError, OWNER_RECHECK, StreamError, StreamResult, + Streams, + handle::{Remembered, chain_of}, proto::{ AttachCallerRequest, AttachCallerResponse, ChainId, DetachCallerRequest, DetachCallerResponse, StreamAddress, StreamOwnerKind, @@ -21,9 +23,10 @@ use std::{io, sync::Arc, time::SystemTime}; pub(crate) async fn attach_caller( streams: &Streams, client: Option<&Arc>, + chains: &Remembered, request: AttachCallerRequest, ) -> AttachCallerResponse { - match attach(streams, client, request).await { + match attach(streams, client, chains, request).await { Ok(operation_token) => AttachCallerResponse { operation_token, failure: None, @@ -51,6 +54,7 @@ pub(crate) async fn detach_caller( async fn attach( streams: &Streams, client: Option<&Arc>, + chains: &Remembered, request: AttachCallerRequest, ) -> StreamResult { if request.callback_url.is_empty() { @@ -65,9 +69,19 @@ async fn attach( StreamError::unsupported("attaching a caller needs a client connected to the server") })?; let stream = checked(request.stream.as_ref())?; - // Fresh for the latest run, since the Workflow id may have moved on to a new chain since a - // read warmed the cache, and an attach to the ended chain would never hear the new stream. - let chain = streams.chain_now(stream).await?; + // A retried start goes to the chain its first try attached to, so it mints the same token and + // the server keeps one attach. Otherwise fresh for the latest run, since the Workflow id may + // have moved on to a new chain since a read warmed the cache, and an attach to the ended + // chain would never hear the new stream. + let key = format!( + "{}\0{}\0{}\0{}", + stream.namespace, stream.workflow_id, stream.topic, request.request_id + ); + let chain = match chains.get(&key) { + Some(attached) => chain_of(stream, &attached.first_run_id), + None => streams.chain_now(stream).await?, + }; + let first_run_id = chain.first_run_id.clone(); let operation_token = OperationToken { stream: stream.clone(), attach_request_id: request.request_id.clone(), @@ -88,6 +102,7 @@ async fn attach( }) .await .map_err(caller_error)?; + chains.put(key, &first_run_id, OWNER_RECHECK); Ok(operation_token) } @@ -580,6 +595,31 @@ mod tests { ); } + #[tokio::test] + async fn a_retried_start_after_the_chain_moved_keeps_its_chain_and_token() { + let setup = setup(); + let first = attach(&setup.service, attach_request("temporal://system")) + .await + .unwrap(); + setup + .owners + .chains + .lock() + .unwrap() + .insert("owner-1".to_string(), "chain-2".to_string()); + let again = attach(&setup.service, attach_request("temporal://system")) + .await + .unwrap(); + assert_eq!(first, again); + let attached = setup.notifier.attached.lock().unwrap().clone(); + assert!( + attached + .iter() + .all(|attachment| attachment.chain.first_run_id == "chain-1"), + "{attached:?}" + ); + } + #[tokio::test] async fn a_pinned_ref_attaches_by_its_chains_first_run() { let setup = setup(); diff --git a/crates/streams/src/service.rs b/crates/streams/src/service.rs index 8018e612fc..ca7a26cc6c 100644 --- a/crates/streams/src/service.rs +++ b/crates/streams/src/service.rs @@ -40,6 +40,8 @@ pub struct StreamService { subscriptions: Subscriptions, nexus: NexusCalls, callers: Option>, + /// The chain each start's attach went to, by request id, so a retried start attaches there. + caller_chains: crate::handle::Remembered, /// What a driver archives into, when the config names an archive. archive: Option, } @@ -171,6 +173,7 @@ impl StreamService { subscriptions: Subscriptions::default(), nexus, callers: None, + caller_chains: crate::handle::Remembered::new(StreamsOptions::default().remembered), archive: None, } } @@ -310,6 +313,7 @@ impl StreamService { "AttachCaller" => Ok(crate::nexus::attach_caller( &self.streams, self.callers.as_ref(), + &self.caller_chains, decode::(rpc, request)?, ) .await