From 2258a2cd77c1279bef524a4c44a3e6dd7efc6509 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Tue, 13 Oct 2026 22:55:03 -0700 Subject: [PATCH 1/3] 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 782a464a1aa0b532a9fba98f5e301322df67a5e8 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Tue, 13 Oct 2026 23:13:21 -0700 Subject: [PATCH 2/3] 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 3/3] 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