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/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index 1cb58bf9a8..b8f9d9e33b 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/crates/sdk-core/CHANGELOG.md @@ -176,9 +176,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 d6ac995cd2..8eb56d4067 100644 --- a/crates/sdk-core/src/streams.rs +++ b/crates/sdk-core/src/streams.rs @@ -233,6 +233,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() } } @@ -681,6 +682,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!( @@ -700,6 +702,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 8578e7814f..0788013b6d 100644 --- a/crates/streams/src/handle.rs +++ b/crates/streams/src/handle.rs @@ -747,11 +747,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 276959c368..5d18f46eb0 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. @@ -23,11 +25,10 @@ use std::{io, sync::Arc, time::SystemTime}; pub(crate) async fn attach_caller( streams: &Streams, client: Option<&Arc>, - announce: Option<&Arc>, chains: &Remembered, request: AttachCallerRequest, ) -> AttachCallerResponse { - match attach(streams, client, announce, chains, request).await { + match attach(streams, client, chains, request).await { Ok(operation_token) => AttachCallerResponse { operation_token, failure: None, @@ -55,7 +56,6 @@ pub(crate) async fn detach_caller( async fn attach( streams: &Streams, client: Option<&Arc>, - announce: Option<&Arc>, chains: &Remembered, request: AttachCallerRequest, ) -> StreamResult { @@ -101,56 +101,15 @@ 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; - } chains.put(key, &first_run_id, OWNER_RECHECK); 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, @@ -571,97 +530,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 fc266ef1ae..3e849cce7a 100644 --- a/crates/streams/src/notify.rs +++ b/crates/streams/src/notify.rs @@ -133,6 +133,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 diff --git a/crates/streams/src/service.rs b/crates/streams/src/service.rs index 1907ac9a4e..d522c5115a 100644 --- a/crates/streams/src/service.rs +++ b/crates/streams/src/service.rs @@ -40,7 +40,6 @@ pub struct StreamService { subscriptions: Subscriptions, nexus: NexusCalls, callers: Option>, - attach_progress: 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. @@ -155,8 +154,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, }) } @@ -175,7 +173,6 @@ impl StreamService { subscriptions: Subscriptions::default(), nexus, callers: None, - attach_progress: None, caller_chains: crate::handle::Remembered::new(StreamsOptions::default().remembered), archive: None, } @@ -193,16 +190,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() @@ -331,7 +318,6 @@ impl StreamService { "AttachCaller" => Ok(crate::nexus::attach_caller( &self.streams, self.callers.as_ref(), - self.attach_progress.as_ref(), &self.caller_chains, decode::(rpc, request)?, )