From 3c8d113aa263261e0699cee88465eab0f1822d79 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Wed, 14 Oct 2026 21:26:54 -0700 Subject: [PATCH] Attached a stream operation's caller as an idle callback. A reader reads first, so its operation needs to hear of progress only once it is caught up. The handler asks for one wake after a counter with the new WaitStreamCallback. --- crates/client/src/grpc.rs | 9 ++ .../nexus/temporal-proto-models-nexusrpc.yaml | 15 ++ .../api_upstream/nexus/workflow-service.wit | 20 +++ .../workflowservice/v1/request_response.proto | 23 +++ .../api/workflowservice/v1/service.proto | 12 ++ crates/sdk-core-c-bridge/src/client.rs | 3 + crates/sdk-core/CHANGELOG.md | 7 +- crates/sdk-core/src/streams.rs | 31 ++++ crates/streams/src/handle.rs | 5 - crates/streams/src/nexus/operation.rs | 146 ++---------------- crates/streams/src/notify.rs | 19 +++ crates/streams/src/service.rs | 16 +- 12 files changed, 149 insertions(+), 157 deletions(-) diff --git a/crates/client/src/grpc.rs b/crates/client/src/grpc.rs index 31b51dfbf7..df3b541c79 100644 --- a/crates/client/src/grpc.rs +++ b/crates/client/src/grpc.rs @@ -1896,6 +1896,15 @@ proxier! { r.extensions_mut().insert(labels); } ); + ( + wait_stream_callback, + WaitStreamCallbackRequest, + WaitStreamCallbackResponse, + |r| { + let labels = namespaced_request!(r); + r.extensions_mut().insert(labels); + } + ); ( notify_stream, NotifyStreamRequest, 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 e611f78575..5f535463e1 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 @@ -47,6 +47,21 @@ services: $pythonRef: temporalio.api.workflowservice.v1.DetachStreamCallbackResponse $rubyRef: Temporalio::Api::WorkflowService::V1::DetachStreamCallbackResponse $typescriptRef: '@temporalio/api/workflowservice/v1.DetachStreamCallbackResponse' + WaitStreamCallback: + input: + $dotnetRef: Temporalio.Api.WorkflowService.V1.WaitStreamCallbackRequest + $goRef: go.temporal.io/api/workflowservice/v1.WaitStreamCallbackRequest + $javaRef: io.temporal.api.workflowservice.v1.WaitStreamCallbackRequest + $pythonRef: temporalio.api.workflowservice.v1.WaitStreamCallbackRequest + $rubyRef: Temporalio::Api::WorkflowService::V1::WaitStreamCallbackRequest + $typescriptRef: '@temporalio/api/workflowservice/v1.WaitStreamCallbackRequest' + output: + $dotnetRef: Temporalio.Api.WorkflowService.V1.WaitStreamCallbackResponse + $goRef: go.temporal.io/api/workflowservice/v1.WaitStreamCallbackResponse + $javaRef: io.temporal.api.workflowservice.v1.WaitStreamCallbackResponse + $pythonRef: temporalio.api.workflowservice.v1.WaitStreamCallbackResponse + $rubyRef: Temporalio::Api::WorkflowService::V1::WaitStreamCallbackResponse + $typescriptRef: '@temporalio/api/workflowservice/v1.WaitStreamCallbackResponse' NotifyStream: input: $dotnetRef: Temporalio.Api.WorkflowService.V1.NotifyStreamRequest diff --git a/crates/protos/protos/api_upstream/nexus/workflow-service.wit b/crates/protos/protos/api_upstream/nexus/workflow-service.wit index fa0aa9aae2..5b847e5442 100644 --- a/crates/protos/protos/api_upstream/nexus/workflow-service.wit +++ b/crates/protos/protos/api_upstream/nexus/workflow-service.wit @@ -173,6 +173,7 @@ interface workflow-service { start-time: option, /// @nexus.omit workflow-target: placeholder, + idle: option, } /// @nexus.proto "temporal.api.workflowservice.v1.AttachStreamCallbackResponse" @@ -235,4 +236,23 @@ interface workflow-service { use nexus:temporal-types/model@1.0.0.{ failure, }; + /// @nexus.proto "temporal.api.workflowservice.v1.WaitStreamCallbackRequest" + record wait-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, + after-counter: option, + /// @nexus.omit + identity: placeholder, + } + + /// @nexus.proto "temporal.api.workflowservice.v1.WaitStreamCallbackResponse" + record wait-stream-callback-response { + } + + wait-stream-callback: func( + request: wait-stream-callback-request, + ) -> wait-stream-callback-response; } diff --git a/crates/protos/protos/api_upstream/temporal/api/workflowservice/v1/request_response.proto b/crates/protos/protos/api_upstream/temporal/api/workflowservice/v1/request_response.proto index 3d5f876aa3..d5755f19e9 100644 --- a/crates/protos/protos/api_upstream/temporal/api/workflowservice/v1/request_response.proto +++ b/crates/protos/protos/api_upstream/temporal/api/workflowservice/v1/request_response.proto @@ -3719,11 +3719,34 @@ message AttachStreamCallbackRequest { // Workflow gets each change of the stream as a `StreamWake` on its next Workflow Task. Exactly // one of `callback` and `workflow_target` is set. temporal.api.stream.v1.WorkflowWakeTarget workflow_target = 8; + // Attaches the Nexus callback idle. An idle callback gets no progress until + // WaitStreamCallback asks for one wake, and it goes back to idle after that delivery. It still + // gets the completion when the stream closes. Not allowed with `workflow_target`. + bool idle = 9; } message AttachStreamCallbackResponse { } +message WaitStreamCallbackRequest { + string namespace = 1; + temporal.api.stream.v1.StreamReference stream_ref = 2; + // The request ID the idle callback was attached with. + string request_id = 3; + // The notifier delivers progress to the callback once, when its counter is above this value. + // If the counter is above it already, the delivery goes out at once. A reader passes the + // counter of the position its last read reached. + // + // (-- api-linter: core::0140::prepositions=disabled + // aip.dev/not-precedent: "after" names the bound the counter must pass. --) + int64 after_counter = 4; + // The identity of the caller. + string identity = 5; +} + +message WaitStreamCallbackResponse { +} + message DetachStreamCallbackRequest { string namespace = 1; temporal.api.stream.v1.StreamReference stream_ref = 2; 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 0e5ff8bb2b..a26f7474ea 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 @@ -2061,6 +2061,18 @@ service WorkflowService { option (nexusannotations.v1.operation).tags = "exposed"; } + // WaitStreamCallback asks the notifier of a stream to wake one idle callback once, when the + // notifier's counter passes `after_counter`. If it passed already, the wake goes out at once. + // After the delivery the callback is idle again. The handler of a stream operation calls it + // for its caller when a read finds nothing, so a busy stream that the caller keeps up with + // sends the caller nothing. Waiting on a callback that is not idle changes nothing. + // + // (-- api-linter: core::0127::http-annotation=disabled + // aip.dev/not-precedent: SDKs reach the stream notifier through System Nexus or gRPC. --) + rpc WaitStreamCallback (WaitStreamCallbackRequest) returns (WaitStreamCallbackResponse) { + option (nexusannotations.v1.operation).tags = "exposed"; + } + // NotifyStream tells the notifier of a stream that the stream moved, or that it closed. The // notifier passes the latest notification on to every attached callback as progress, and on close // completes them. Producers fold their notifications and keep at most one call in flight. diff --git a/crates/sdk-core-c-bridge/src/client.rs b/crates/sdk-core-c-bridge/src/client.rs index 36dbed41d9..4671586341 100644 --- a/crates/sdk-core-c-bridge/src/client.rs +++ b/crates/sdk-core-c-bridge/src/client.rs @@ -716,6 +716,9 @@ async fn call_workflow_service( "DetachStreamCallback" => { rpc_call_on_trait!(client, call, WorkflowService, detach_stream_callback) } + "WaitStreamCallback" => { + rpc_call_on_trait!(client, call, WorkflowService, wait_stream_callback) + } "ExecuteMultiOperation" => { rpc_call_on_trait!(client, call, WorkflowService, execute_multi_operation) } diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index 97988208a5..63b40493c3 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/crates/sdk-core/CHANGELOG.md @@ -139,9 +139,10 @@ relevant information. 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. -* Experimental: with `notify_on_append`, `AttachCaller` tells the stream's notifier about the - newest record after the attach, so a caller that only waits for progress hears of the records - already there. An empty stream needs nothing, since every append notifies. The API protos carry the +* Experimental: `AttachCaller` attaches the caller's callback as an idle one, which hears no + progress until a wait after a counter and still gets the completion. A reader reads first, so a + busy one costs no deliveries. `NotifierClient::wait_callback` sends the new + `WaitStreamCallback`, and the C bridge takes it as a raw call. The API protos carry the `NexusHandler` progress fields and `nexus/notification-service.wit`. * Experimental: the Nexus `read` takes `floor_holder`. With it the read holds the floor and answers `replay`, the range of the records that crossed after the byte cut, with Core's digest diff --git a/crates/sdk-core/src/streams.rs b/crates/sdk-core/src/streams.rs index 51508317fc..5cfd7c0c1d 100644 --- a/crates/sdk-core/src/streams.rs +++ b/crates/sdk-core/src/streams.rs @@ -11,6 +11,7 @@ use temporalio_common::protos::temporal::api::{ AttachStreamCallbackRequest, DescribeWorkflowExecutionRequest, DescribeWorkflowExecutionResponse, DetachStreamCallbackRequest, NotifyStreamRequest, NotifyStreamsRequest, PollStreamNotifierRequest, PollStreamNotifierResponse, + WaitStreamCallbackRequest, }, }; use temporalio_streams::{ @@ -85,6 +86,30 @@ impl NotifierClient for ConnectionNotifierClient { .map_err(notify_error) } + async fn wait_callback( + &self, + chain: ChainId, + topic: String, + request_id: String, + after_counter: i64, + ) -> Result<(), NotifyError> { + self.connection + .workflow_service() + .wait_stream_callback( + WaitStreamCallbackRequest { + namespace: chain.namespace.clone(), + stream_ref: Some(chain_reference(&chain, topic)), + request_id, + after_counter, + identity: self.connection.identity().to_string(), + } + .into_request(), + ) + .await + .map(|_| ()) + .map_err(notify_error) + } + async fn detach( &self, chain: ChainId, @@ -183,6 +208,7 @@ fn caller_attach_request(attachment: CallerAttachment) -> AttachStreamCallbackRe }), operation_token: attachment.operation_token, start_time: Some(attachment.start_time.into()), + idle: attachment.idle, ..Default::default() } } @@ -436,6 +462,7 @@ mod tests { callback_headers: [("nexus-callback-token".to_string(), "abc".to_string())].into(), operation_token: "token".to_string(), start_time, + idle: true, }); assert_eq!(request.namespace, "default"); assert_eq!( @@ -454,6 +481,10 @@ mod tests { 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"); + assert!( + request.idle, + "the reader wakes the callback when it is caught up" + ); } #[test] diff --git a/crates/streams/src/handle.rs b/crates/streams/src/handle.rs index a60114c678..37732870a3 100644 --- a/crates/streams/src/handle.rs +++ b/crates/streams/src/handle.rs @@ -660,11 +660,6 @@ impl Streams { } } - /// The store the handles serve. - pub(crate) fn store(&self) -> &dyn StreamStore { - self.store.as_ref() - } - /// The cursor of a stream's newest record, empty when it holds none. pub async fn latest(&self, request: LatestRequest) -> StreamResult { let stream = address(request.stream.as_ref())?; diff --git a/crates/streams/src/nexus/operation.rs b/crates/streams/src/nexus/operation.rs index 4e2f11f5ec..8a3c1e6969 100644 --- a/crates/streams/src/nexus/operation.rs +++ b/crates/streams/src/nexus/operation.rs @@ -1,7 +1,9 @@ //! 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 +//! The operation's start attaches the caller's callback to the stream's notifier on the server as +//! an idle one, and answers with a token that names the stream. An idle callback hears nothing of +//! the stream's progress until its reader, caught up, asks to be woken after a counter. It still +//! gets the completion when the stream closes. 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. @@ -21,10 +23,9 @@ use std::{io, sync::Arc, time::SystemTime}; pub(crate) async fn attach_caller( streams: &Streams, client: Option<&Arc>, - announce: Option<&Arc>, request: AttachCallerRequest, ) -> AttachCallerResponse { - match attach(streams, client, announce, request).await { + match attach(streams, client, request).await { Ok(operation_token) => AttachCallerResponse { operation_token, failure: None, @@ -52,7 +53,6 @@ pub(crate) async fn detach_caller( async fn attach( streams: &Streams, client: Option<&Arc>, - announce: Option<&Arc>, request: AttachCallerRequest, ) -> StreamResult { if request.callback_url.is_empty() { @@ -87,55 +87,14 @@ async fn attach( callback_headers: request.callback_headers, operation_token: operation_token.clone(), start_time: SystemTime::now(), + // A reader reads first, and a busy one never waits, so it costs no deliveries. + idle: true, }) .await .map_err(caller_error)?; - if let Some(notifier) = announce { - announce_newest(streams, notifier, chain, &stream.topic).await; - } Ok(operation_token) } -/// Notifies the newest record on `topic`, so a caller that just attached hears of the records -/// already there. The server keeps a notifier only once a caller attached, so the notifications -/// sent before were dropped. It goes through the notifier the appends use, so a failed send is -/// tried again and a flush waits for it. A notification the retries can't send is lost as an -/// append's is, and the caller still hears of every record from the next append or the close. -/// -/// An empty stream needs nothing, since every append notifies. -async fn announce_newest( - streams: &Streams, - notifier: &crate::Notifier, - chain: ChainId, - topic: &str, -) { - let latest = streams - .store() - .latest(crate::proto::StoreLatestRequest { - chain: Some(chain.clone()), - topic: topic.to_string(), - }) - .await; - let position = match latest { - Ok(latest) if latest.position.is_empty() => return, - Ok(latest) => latest.position, - Err(error) => { - tracing::warn!( - workflow_id = %chain.workflow_id, - topic, - %error, - "Could not find the newest record to tell a new caller about" - ); - return; - } - }; - if let Some((cursor, counter)) = - crate::notify::notification_at(streams.store(), &chain, topic, &position) - { - notifier.notify(&chain, topic, cursor, counter); - } -} - async fn detach( client: Option<&Arc>, request: DetachCallerRequest, @@ -549,97 +508,16 @@ mod tests { .encode_to_vec() } - /// `setup()` with the announcement on, sent through Core's folding notifier. - fn announcing() -> Setup { - let mut setup = setup(); - let folding = Arc::new(crate::Notifier::new(setup.notifier.clone(), 16)); - setup.service = setup.service.with_attach_progress(Some(folding)); - setup - } - - /// Waits for the background sends of the folding notifier. - async fn notified(notifier: &Notifier, count: usize) -> Vec { - for _ in 0..50 { - if notifier.notified.lock().unwrap().len() >= count { - break; - } - tokio::time::sleep(std::time::Duration::from_millis(50)).await; - } - notifier.notified.lock().unwrap().clone() - } - #[tokio::test] - async fn an_attach_announces_the_records_already_there_when_progress_is_on() { - let setup = announcing(); - setup - .service - .call("Append", &append(&["a", "b"])) - .await - .unwrap(); - attach(&setup.service, attach_request("temporal://system")) - .await - .unwrap(); - let notified = notified(&setup.notifier, 1).await; - let [notification] = notified.as_slice() else { - panic!("one notification after the attach, not {notified:?}"); - }; - assert_eq!( - notification.chain, - setup.notifier.attached.lock().unwrap()[0].chain - ); - assert_eq!(notification.topic, "tokens"); - // The memory store's second record is at offset 1, so its counter is 2. - assert_eq!(notification.counter, 2); - assert!( - notification.position.ends_with(":1"), - "{}", - notification.position - ); - assert!(notification.close.is_none()); - } - - #[tokio::test] - async fn a_failed_announcement_is_sent_again_as_an_append_notification_is() { - // A quiet stream gets no other notification, so a lost announcement would leave a - // caller that only waits for progress with nothing until the close. - let setup = announcing(); - setup.service.call("Append", &append(&["a"])).await.unwrap(); - setup - .notifier - .notify_failures - .store(1, std::sync::atomic::Ordering::SeqCst); - attach(&setup.service, attach_request("temporal://system")) - .await - .unwrap(); - let notified = notified(&setup.notifier, 1).await; - assert_eq!(notified.len(), 1, "the retry went out: {notified:?}"); - assert_eq!(notified[0].counter, 1); - } - - #[tokio::test] - async fn an_attach_to_a_stream_with_records_announces_its_newest_record() { - let setup = announcing(); - setup.service.call("Append", &append(&["a"])).await.unwrap(); - attach(&setup.service, attach_request("temporal://system")) - .await - .unwrap(); - assert_eq!(notified(&setup.notifier, 1).await.len(), 1); - } - - #[tokio::test] - async fn an_attach_announces_nothing_on_an_empty_stream_or_with_progress_off() { - let setup = announcing(); - attach(&setup.service, attach_request("temporal://system")) - .await - .unwrap(); - assert!(notified(&setup.notifier, 1).await.is_empty()); - - let setup = self::setup(); + async fn a_start_attaches_an_idle_callback_and_announces_nothing() { + let setup = setup(); setup.service.call("Append", &append(&["a"])).await.unwrap(); attach(&setup.service, attach_request("temporal://system")) .await .unwrap(); - assert!(notified(&setup.notifier, 1).await.is_empty()); + let attached = setup.notifier.attached.lock().unwrap().clone(); + assert!(attached.iter().all(|a| a.idle), "{attached:?}"); + assert!(setup.notifier.notified.lock().unwrap().is_empty()); } #[tokio::test] diff --git a/crates/streams/src/notify.rs b/crates/streams/src/notify.rs index 8385d09259..2a1d077e61 100644 --- a/crates/streams/src/notify.rs +++ b/crates/streams/src/notify.rs @@ -128,6 +128,9 @@ pub struct CallerAttachment { pub operation_token: String, /// When the operation started, which the server hands back too. pub start_time: std::time::SystemTime, + /// The callback hears no progress until a wait after a counter, and still gets the + /// completion. + pub idle: bool, } /// Sends notifications to the stream notifier on the server, attaches and detaches the @@ -165,6 +168,22 @@ pub trait NotifierClient: Send + Sync { )) } + /// Wakes the idle callback that the attach with `request_id` attached to `chain`'s stream on + /// `topic` once, when the notifier's counter is above `after_counter`, at once when it is + /// already. A newer wait replaces an older one. + async fn wait_callback( + &self, + chain: ChainId, + topic: String, + request_id: String, + after_counter: i64, + ) -> Result<(), NotifyError> { + let _ = (chain, topic, request_id, after_counter); + Err(NotifyError::Unsupported( + "this client can't wake 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 9e8e80352d..c66c40c727 100644 --- a/crates/streams/src/service.rs +++ b/crates/streams/src/service.rs @@ -35,7 +35,6 @@ pub struct StreamService { subscriptions: Subscriptions, nexus: NexusCalls, callers: Option>, - attach_progress: Option>, } impl std::fmt::Debug for StreamService { @@ -117,8 +116,7 @@ impl StreamService { Ok(match notifier { Some(client) => service .with_subscriptions(client.clone()) - .with_callers(client) - .with_attach_progress(folding.filter(|_| config.notify_on_append)), + .with_callers(client), None => service, }) } @@ -137,7 +135,6 @@ impl StreamService { subscriptions: Subscriptions::default(), nexus, callers: None, - attach_progress: None, } } @@ -153,16 +150,6 @@ impl StreamService { self } - /// After a caller's attach, notifies the stream's newest record, so the new caller hears of - /// the records already there. A caller that only waits for progress would otherwise hear - /// nothing until the next append. It goes through `notifier`, the one that sends the appends' - /// notifications, so it is retried and flushed as they are. An empty stream needs nothing, - /// since every append notifies. `None` turns it off, as the config's progress setting does. - pub fn with_attach_progress(mut self, notifier: Option>) -> Self { - self.attach_progress = notifier; - self - } - /// The store, for the Workers that append Workflow output to it. pub fn store(&self) -> Arc { self.store.clone() @@ -274,7 +261,6 @@ impl StreamService { "AttachCaller" => Ok(crate::nexus::attach_caller( &self.streams, self.callers.as_ref(), - self.attach_progress.as_ref(), decode::(rpc, request)?, ) .await