[Bridge] Attached stream operation callers idle.

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

crates/protos/protos/api_upstream/nexus/workflow-service.wit

@@ -173,6 +173,7 @@ interface workflow-service {
     start-time: option<timestamp>,
     /// @nexus.omit
     workflow-target: placeholder,
+    idle: option<bool>,
   }
 
   /// @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<stream-reference>,
+    request-id: string,
+    after-counter: option<s64>,
+    /// @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;
 }

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

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]

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<LatestResponse> {
         let stream = address(request.stream.as_ref())?;

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<dyn NotifierClient>>,
-    announce: Option<&Arc<crate::Notifier>>,
     chains: &Remembered<String>,
     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<dyn NotifierClient>>,
-    announce: Option<&Arc<crate::Notifier>>,
     chains: &Remembered<String>,
     request: AttachCallerRequest,
 ) -> StreamResult<String> {
@@ -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<dyn NotifierClient>>,
     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<StreamNotification> {
-        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]

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

crates/streams/src/service.rs

@@ -40,7 +40,6 @@ pub struct StreamService {
     subscriptions: Subscriptions,
     nexus: NexusCalls,
     callers: Option<Arc<dyn NotifierClient>>,
-    attach_progress: Option<Arc<Notifier>>,
     /// The chain each start's attach went to, by request id, so a retried start attaches there.
     caller_chains: crate::handle::Remembered<String>,
     /// 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<Arc<Notifier>>) -> Self {
-        self.attach_progress = notifier;
-        self
-    }
-
     /// The store, for the Workers that append Workflow output to it.
     pub fn store(&self) -> Arc<dyn StreamStore> {
         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::<AttachCallerRequest>(rpc, request)?,
             )