[Bridge] Notified a stream's notifier after appends and visible batches.

crates/protos/protos/local/temporal/sdk/core/streams/streams.proto

@@ -24,6 +24,10 @@ service StreamService {
   rpc Close(CloseRequest) returns (CloseResponse);
   // Deletes every stream of one owner.
   rpc DeleteOwner(DeleteOwnerRequest) returns (DeleteOwnerResponse);
+  // Waits until the stream notifications this process sent are out, for at most 10 seconds. A
+  // notification goes in the background, so a process that holds only a client calls it before it
+  // stops. A Worker flushes at stop on its own.
+  rpc FlushNotifications(FlushNotificationsRequest) returns (FlushNotificationsResponse);
 }
 
 // Which store holds a process's streams, and how to reach it. Lang builds one store from it and
@@ -33,6 +37,14 @@ message StreamStoreConfig {
     RedisStoreConfig redis = 1;
     MemoryStoreConfig memory = 2;
   }
+  // Tells each stream's notifier on the server when the stream moves, after every append and
+  // every Workflow batch that becomes visible. That's what makes a caller of a stream-returning
+  // Nexus operation see progress. Off by default, since most streams have no such caller and each
+  // notification costs a call. A server without the stream notifier costs one call, and then the
+  // process stops notifying.
+  bool notify_on_append = 3;
+  // How many streams to keep a notifier for, dropping the least recently used. Zero uses 1000.
+  uint32 max_notifiers = 4;
 }
 
 // Streams in a Redis or Redis Cluster that Core connects to itself.
@@ -198,6 +210,10 @@ message CloseRequest {
 
 message CloseResponse {}
 
+message FlushNotificationsRequest {}
+
+message FlushNotificationsResponse {}
+
 message DeleteOwnerRequest {
   string namespace = 1;
   StreamOwnerKind owner_kind = 2;

crates/sdk-core-c-bridge/src/tests/mod.rs

@@ -508,6 +508,7 @@ fn memory_store_config() -> Vec<u8> {
     };
     StreamStoreConfig {
         store: Some(stream_store_config::Store::Memory(MemoryStoreConfig {})),
+        ..Default::default()
     }
     .encode_to_vec()
 }

crates/sdk-core/CHANGELOG.md

@@ -68,6 +68,17 @@ relevant information.
   Worker's `stream_store`, so a Workflow's output and outside producers share it.
 * Experimental: the C bridge has `temporal_core_stream_store_new`, `_call` and `_free` for the
   stream service, built with the Redis store, and a `stream_store` field on its worker options.
+* Experimental: with `StreamStoreConfig.notify_on_append`, Core tells each stream's notifier on
+  the server when the stream moves, after every append and every Workflow batch that becomes
+  visible. That's how a caller of a stream-returning Nexus operation sees progress. Notifications
+  for one stream fold into one call in flight, a failed last notification is tried twice more,
+  and a notification never fails the write. `max_notifiers` caps how many streams the process
+  keeps a notifier for (1000 when zero).
+* Experimental: the new `FlushNotifications` stream call waits up to 10 seconds for the
+  notifications still out. A Worker flushes at stop on its own, so only a process that holds just
+  a client needs it.
+* Experimental: a server without the stream notifier answers the first notifier call with
+  `UNIMPLEMENTED`. Core then stops notifier calls in that process and logs it once at debug.
 
 ### Breaking Changes
 * Replay detects changes to the service or operation of scheduled Nexus operations, reporting

crates/sdk-core/src/streams.rs

@@ -5,20 +5,104 @@ use std::sync::Arc;
 use temporalio_client::Connection;
 use temporalio_common::protos::temporal::api::{
     common::v1::WorkflowExecution,
-    workflowservice::v1::{DescribeWorkflowExecutionRequest, DescribeWorkflowExecutionResponse},
+    enums::v1::StreamOwnerKind,
+    stream::v1::StreamReference,
+    workflowservice::v1::{
+        DescribeWorkflowExecutionRequest, DescribeWorkflowExecutionResponse, NotifyStreamRequest,
+    },
+};
+use temporalio_streams::{
+    NotifierClient, NotifyError, OwnerClient, OwnerDescription, OwnerError, StreamNotification,
 };
-use temporalio_streams::{OwnerClient, OwnerDescription, OwnerError};
 pub use temporalio_streams::{StreamError, StreamResult, StreamService, StreamStore, proto};
 use tonic::IntoRequest;
 
 /// Connects to the store `config` names. The service asks `connection`'s server about streams'
-/// owners, and its [StreamService::store] goes to each Worker's `stream_store`.
+/// owners, sends the stream notifications the config turns on to it, and its
+/// [StreamService::store] goes to each Worker's `stream_store`.
 pub async fn connect_stream_service(
     config: proto::StreamStoreConfig,
     connection: Connection,
 ) -> StreamResult<Arc<StreamService>> {
-    let owner = Arc::new(ConnectionOwnerClient { connection });
-    Ok(Arc::new(StreamService::connect(config, owner).await?))
+    let owner = Arc::new(ConnectionOwnerClient {
+        connection: connection.clone(),
+    });
+    let notifier = Arc::new(ConnectionNotifierClient { connection });
+    Ok(Arc::new(
+        StreamService::connect_with_notifier(config, owner, Some(notifier)).await?,
+    ))
+}
+
+/// Sends stream notifications to a Temporal server's stream notifier through a client
+/// connection.
+struct ConnectionNotifierClient {
+    connection: Connection,
+}
+
+#[async_trait::async_trait]
+impl NotifierClient for ConnectionNotifierClient {
+    async fn notify(&self, notification: StreamNotification) -> Result<(), NotifyError> {
+        self.connection
+            .workflow_service()
+            .notify_stream(notify_request(notification).into_request())
+            .await
+            .map(|_| ())
+            .map_err(notify_error)
+    }
+}
+
+fn notify_request(notification: StreamNotification) -> NotifyStreamRequest {
+    NotifyStreamRequest {
+        namespace: notification.chain.namespace,
+        // The run chain's first run keys the notifier, so one notifier serves the stream across
+        // Continue-as-New and a new chain on the same Workflow id gets its own.
+        stream_ref: Some(StreamReference {
+            owner_kind: StreamOwnerKind::Workflow as i32,
+            workflow_id: notification.chain.workflow_id,
+            run_id: notification.chain.first_run_id,
+            topic: notification.topic,
+        }),
+        position: notification.position,
+        counter: notification.counter,
+        close: notification.close_result.is_some(),
+        close_result: notification.close_result,
+        ..Default::default()
+    }
+}
+
+/// Answers that mean the server will never take the notification, so a retry can't help.
+fn notify_error(status: tonic::Status) -> NotifyError {
+    use tonic::Code;
+    match status.code() {
+        Code::Unimplemented => NotifyError::Unsupported(status.to_string()),
+        Code::FailedPrecondition if namespace_in_transition(&status) => {
+            NotifyError::Failed(status.to_string())
+        }
+        Code::InvalidArgument
+        | Code::NotFound
+        | Code::PermissionDenied
+        | Code::Unauthenticated
+        | Code::FailedPrecondition => NotifyError::Refused(status.to_string()),
+        _ => NotifyError::Failed(status.to_string()),
+    }
+}
+
+/// A namespace that's failing over answers `FailedPrecondition` too, but only until the
+/// redirection sees the active cluster, so a close mustn't stop on it.
+fn namespace_in_transition(status: &tonic::Status) -> bool {
+    use temporalio_common::protos::{
+        google::rpc::Status as RpcStatus,
+        temporal::api::errordetails::v1::{
+            NamespaceInvalidStateFailure, NamespaceNotActiveFailure,
+        },
+    };
+    let Ok(details) = <RpcStatus as prost::Message>::decode(status.details()) else {
+        return false;
+    };
+    details.details.iter().any(|detail| {
+        detail.to_msg::<NamespaceNotActiveFailure>().is_ok()
+            || detail.to_msg::<NamespaceInvalidStateFailure>().is_ok()
+    })
 }
 
 /// Asks a Temporal server about owners through a client connection.
@@ -144,4 +228,95 @@ mod tests {
             Err(OwnerError::Failed(_))
         ));
     }
+
+    #[test]
+    fn a_notification_names_the_stream_by_its_chains_first_run() {
+        let request = notify_request(StreamNotification {
+            chain: proto::ChainId {
+                namespace: "ns".to_string(),
+                workflow_id: "wf".to_string(),
+                first_run_id: "run-1".to_string(),
+            },
+            topic: "tokens".to_string(),
+            position: "cursor".to_string(),
+            counter: 7,
+            close_result: None,
+        });
+        assert_eq!(request.namespace, "ns");
+        assert_eq!(
+            request.stream_ref,
+            Some(StreamReference {
+                owner_kind: StreamOwnerKind::Workflow as i32,
+                workflow_id: "wf".to_string(),
+                run_id: "run-1".to_string(),
+                topic: "tokens".to_string(),
+            })
+        );
+        assert_eq!((request.position.as_str(), request.counter), ("cursor", 7));
+        assert!(!request.close);
+        assert!(request.close_result.is_none());
+    }
+
+    #[test]
+    fn only_answers_a_retry_cant_change_refuse_a_notification() {
+        assert!(matches!(
+            notify_error(tonic::Status::new(tonic::Code::Unimplemented, "")),
+            NotifyError::Unsupported(_)
+        ));
+        for code in [
+            tonic::Code::InvalidArgument,
+            tonic::Code::NotFound,
+            tonic::Code::PermissionDenied,
+            tonic::Code::Unauthenticated,
+            tonic::Code::FailedPrecondition,
+        ] {
+            assert!(
+                matches!(
+                    notify_error(tonic::Status::new(code, "")),
+                    NotifyError::Refused(_)
+                ),
+                "{code:?}"
+            );
+        }
+        for code in [
+            tonic::Code::Unavailable,
+            tonic::Code::DeadlineExceeded,
+            tonic::Code::ResourceExhausted,
+            tonic::Code::Internal,
+        ] {
+            assert!(
+                matches!(
+                    notify_error(tonic::Status::new(code, "")),
+                    NotifyError::Failed(_)
+                ),
+                "{code:?}"
+            );
+        }
+    }
+
+    fn failed_precondition_with<T: prost::Message + prost::Name>(detail: &T) -> tonic::Status {
+        let status = temporalio_common::protos::google::rpc::Status {
+            code: tonic::Code::FailedPrecondition as i32,
+            message: "namespace".to_string(),
+            details: vec![prost_types::Any::from_msg(detail).unwrap()],
+        };
+        tonic::Status::with_details(
+            tonic::Code::FailedPrecondition,
+            "namespace",
+            prost::Message::encode_to_vec(&status).into(),
+        )
+    }
+
+    #[test]
+    fn a_namespace_in_failover_fails_a_notification_for_now() {
+        use temporalio_common::protos::temporal::api::errordetails::v1::{
+            NamespaceInvalidStateFailure, NamespaceNotActiveFailure,
+        };
+        for status in [
+            failed_precondition_with(&NamespaceNotActiveFailure::default()),
+            failed_precondition_with(&NamespaceInvalidStateFailure::default()),
+        ] {
+            assert!(matches!(notify_error(status), NotifyError::Failed(_)));
+        }
+    }
 }

crates/sdk-core/src/worker/workflow/mod.rs

@@ -843,6 +843,10 @@ impl Workflows {
                 anyhow!("Error joining workflow processing thread: {as_str:?}")
             })?;
         }
+        #[cfg(feature = "streams")]
+        if let Some(store) = &self.output_store {
+            store.flush_notifications().await;
+        }
         Ok(())
     }