[Jetty] Added a notification log to the store contract.

crates/client/src/grpc.rs

@@ -1905,6 +1905,15 @@ proxier! {
             r.extensions_mut().insert(labels);
         }
     );
+    (
+        notify_streams,
+        NotifyStreamsRequest,
+        NotifyStreamsResponse,
+        |r| {
+            let labels = namespaced_request!(r);
+            r.extensions_mut().insert(labels);
+        }
+    );
     (
         describe_stream_notifier,
         DescribeStreamNotifierRequest,

crates/protos/protos/api_upstream/openapi/openapiv2.json

@@ -20063,7 +20063,7 @@
         "counter": {
           "type": "string",
           "format": "int64",
-          "description": "The highest counter the notifier has seen. A reader that armed its read at a counter wakes\nwhen this value is above it. Zero when the notifier has seen no notification."
+          "description": "The highest counter the notifier has seen. A reader that subscribed after a counter wakes\nwhen this value is above it. Zero when the notifier has seen no notification."
         },
         "metadata": {
           "type": "object",

crates/protos/protos/api_upstream/openapi/openapiv3.yaml

@@ -17692,7 +17692,7 @@ components:
         counter:
           type: string
           description: |-
-            The highest counter the notifier has seen. A reader that armed its read at a counter wakes
+            The highest counter the notifier has seen. A reader that subscribed after a counter wakes
              when this value is above it. Zero when the notifier has seen no notification.
         metadata:
           type: object

crates/protos/protos/api_upstream/temporal/api/stream/v1/message.proto

@@ -72,7 +72,7 @@ message StreamWake {
     string subscription_id = 1;
     // The position that came with `counter`, in the producer's terms.
     string position = 2;
-    // The highest counter the notifier has seen. A reader that armed its read at a counter wakes
+    // The highest counter the notifier has seen. A reader that subscribed after a counter wakes
     // when this value is above it. Zero when the notifier has seen no notification.
     int64 counter = 3;
     // The metadata that came with `counter`.
@@ -91,3 +91,25 @@ message StreamWake {
     // reader attaches again under a new request ID.
     bool resubscribe = 9;
 }
+
+// One stream's entry in a NotifyStreams request: the same fields as a NotifyStream request that
+// does not close the stream.
+message StreamNotification {
+    StreamReference stream_ref = 1;
+    // Where the stream stands, in the producer's terms. At most 1 KiB of UTF-8.
+    string position = 2;
+    // The newest counter of the stream. Must be positive. The notifier ignores a counter at or
+    // below the one it holds.
+    int64 counter = 3;
+    // Small details passed on to the callbacks. At most 2 KiB in total, counted as UTF-8 bytes.
+    map<string, string> metadata = 4;
+}
+
+// The answer to one entry of a NotifyStreams request.
+message StreamNotificationResult {
+    // The gRPC status code of the entry, as a NotifyStream call for it would answer. Zero (OK)
+    // when the server applied the entry, or dropped it since the stream has no notifier.
+    int32 code = 1;
+    // Why the server refused the entry. Empty when `code` is zero.
+    string message = 2;
+}

crates/protos/protos/api_upstream/temporal/api/workflowservice/v1/request_response.proto

@@ -3788,7 +3788,7 @@ message PollStreamNotifierRequest {
     string namespace = 1;
     temporal.api.stream.v1.StreamReference stream_ref = 2;
     // The poll returns when the notifier's counter is above this value, or when the stream is
-    // closed. A reader passes the counter of the position where it armed its read, so a
+    // closed. A reader passes the counter of the position its last read reached, so a
     // notification that landed before the poll makes it return at once.
     //
     // (-- api-linter: core::0140::prepositions=disabled
@@ -3813,3 +3813,18 @@ message PollStreamNotifierResponse {
     // The failure the stream closed with, when `closed` is set and it closed with one.
     temporal.api.failure.v1.Failure close_failure = 6;
 }
+
+message NotifyStreamsRequest {
+    string namespace = 1;
+    // One entry per stream. The server rejects a request with more than
+    // `streamnotifier.maxNotifyBatch` entries, or with two entries for one stream.
+    repeated temporal.api.stream.v1.StreamNotification notifications = 2;
+    // The identity of the caller, such as a stream driver.
+    string identity = 3;
+}
+
+message NotifyStreamsResponse {
+    // One result per entry of the request, in the same order. Empty only from a server that
+    // fails the whole call on a bad entry, so an empty list on success means every entry applied.
+    repeated temporal.api.stream.v1.StreamNotificationResult results = 1;
+}

crates/protos/protos/api_upstream/temporal/api/workflowservice/v1/service.proto

@@ -2042,7 +2042,8 @@ service WorkflowService {
 
     // AttachStreamCallback attaches a callback to the notifier of a stream. The callback receives the
     // stream's progress while it is open and one completion when it closes. A Nexus handler whose
-    // operation returns a stream calls it with the caller's callback when the operation starts.
+    // operation returns a stream calls it with the caller's callback when the operation starts. It
+    // starts the notifier when the stream has none.
     //
     // (-- api-linter: core::0127::http-annotation=disabled
     //     aip.dev/not-precedent: SDKs reach the stream notifier through gRPC. --)
@@ -2062,6 +2063,10 @@ service WorkflowService {
     // 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.
     //
+    // A notification that does not close the stream starts nothing. When the stream has no
+    // notifier, nobody waits on it, and the server drops the notification. A close starts a notifier
+    // when the stream has none, so a callback that attaches later still gets the close outcome.
+    //
     // (-- api-linter: core::0127::http-annotation=disabled
     //     aip.dev/not-precedent: SDKs reach the stream notifier through gRPC. --)
     rpc NotifyStream (NotifyStreamRequest) returns (NotifyStreamResponse) {
@@ -2078,10 +2083,25 @@ service WorkflowService {
     // PollStreamNotifier waits until the notifier of a stream has a counter above `after_counter`,
     // or the stream is closed. It returns with no change after a wait limit set by the server
     // (`streamnotifier.pollMaxWait`). A client that reads a stream calls it after a read that
-    // found nothing and armed the topic, in place of another blocking read on the store.
+    // found nothing, in place of another blocking read on the store. It starts the notifier when the
+    // stream has none. The client reads the stream once more after the poll starts, since a
+    // notification that came before the notifier existed was dropped.
     //
     // (-- api-linter: core::0127::http-annotation=disabled
     //     aip.dev/not-precedent: SDKs reach the stream notifier through gRPC. --)
     rpc PollStreamNotifier (PollStreamNotifierRequest) returns (PollStreamNotifierResponse) {
     }
+
+    // NotifyStreams tells the notifiers of many streams in one namespace that they moved. It acts as
+    // one NotifyStream without close per entry. A stream with no notifier drops its entry, as
+    // NotifyStream does. A driver that reads the store's notification log calls it with the newest
+    // counter of each stream in a batch. The server checks and applies each entry on its own and
+    // answers each in `results`, so a bad entry doesn't hold up the others. The call fails only for
+    // the request as a whole, such as an empty batch or one that is too large. Sending an entry
+    // again is safe, since a notifier ignores a counter at or below the one it holds.
+    //
+    // (-- api-linter: core::0127::http-annotation=disabled
+    //     aip.dev/not-precedent: SDKs reach the stream notifier through gRPC. --)
+    rpc NotifyStreams (NotifyStreamsRequest) returns (NotifyStreamsResponse) {
+    }
 }

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

@@ -73,6 +73,10 @@ message StreamStoreConfig {
   // owner does. After it the read fails as NOT_FOUND. Unset waits 60 seconds, and zero fails at
   // once.
   google.protobuf.Duration owner_start_grace = 6;
+  // Each append writes a notification entry in the same store step, and producers send no
+  // notifications. A `StreamDriver` reads the entries and notifies the server, so readers wake
+  // only while one runs. Off by default. A store without a notification log refuses it.
+  bool notification_log = 7;
 }
 
 // Streams in a Redis or Redis Cluster that Core connects to itself.

crates/sdk-core-c-bridge/src/client.rs

@@ -814,6 +814,9 @@ async fn call_workflow_service(
         "NotifyStream" => {
             rpc_call_on_trait!(client, call, WorkflowService, notify_stream)
         }
+        "NotifyStreams" => {
+            rpc_call_on_trait!(client, call, WorkflowService, notify_streams)
+        }
         "PatchSchedule" => rpc_call_on_trait!(client, call, WorkflowService, patch_schedule),
         "PauseActivity" => rpc_call_on_trait!(client, call, WorkflowService, pause_activity),
         "PauseActivityExecution" => {

crates/sdk-core/CHANGELOG.md

@@ -46,6 +46,13 @@ relevant information.
   and the reader waits for a counter above it. 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: a stream store can keep a notification log, one entry per append written in the
+  same store step. With `StreamStoreConfig.notification_log`, producers send no notifications and
+  a driver reads the log through `StreamStore::notification_log`, a `NotificationLog` of
+  partitions, doorbells, consumer-group takes, claims after an idle time, acks and a sweep. The
+  memory store keeps one, and a store without one refuses the setting. `NotifierClient` gains
+  `notify_streams`, which Core's client sends as one `NotifyStreams` call per namespace, and the C
+  bridge takes `NotifyStreams` as a raw call.
 * Experimental: the new `FlushNotifications` stream call waits up to 10 seconds for the
   notifications still out. A process calls it before it stops.
 * Experimental: a server without the stream notifier answers the first notifier call with

crates/sdk-core/src/streams.rs

@@ -2,17 +2,17 @@
 //! client's server about streams' owners.
 
 use prost::Message;
-use std::sync::Arc;
+use std::{collections::BTreeMap, sync::Arc};
 use temporalio_client::Connection;
 use temporalio_common::protos::google::rpc::Status as RpcStatus;
 use temporalio_common::protos::temporal::api::{
     common::v1::WorkflowExecution,
     enums::v1::StreamOwnerKind,
-    stream::v1::{StreamReference, WorkflowWakeTarget},
+    stream::v1::{self, StreamReference, WorkflowWakeTarget},
     workflowservice::v1::{
         AttachStreamCallbackRequest, DescribeWorkflowExecutionRequest,
         DescribeWorkflowExecutionResponse, DetachStreamCallbackRequest, NotifyStreamRequest,
-        PollStreamNotifierRequest, PollStreamNotifierResponse,
+        NotifyStreamsRequest, PollStreamNotifierRequest, PollStreamNotifierResponse,
     },
 };
 use temporalio_streams::{
@@ -55,6 +55,20 @@ impl NotifierClient for ConnectionNotifierClient {
             .map_err(notify_error)
     }
 
+    async fn notify_streams(
+        &self,
+        notifications: Vec<StreamNotification>,
+    ) -> Result<(), NotifyError> {
+        for request in notify_streams_requests(notifications, self.connection.identity()) {
+            self.connection
+                .workflow_service()
+                .notify_streams(request.into_request())
+                .await
+                .map_err(notify_error)?;
+        }
+        Ok(())
+    }
+
     async fn attach_workflow(&self, attachment: WorkflowAttachment) -> Result<(), NotifyError> {
         self.connection
             .workflow_service()
@@ -151,6 +165,33 @@ fn attach_request(attachment: WorkflowAttachment) -> AttachStreamCallbackRequest
     }
 }
 
+/// One `NotifyStreams` request per namespace, since a call names one.
+fn notify_streams_requests(
+    notifications: Vec<StreamNotification>,
+    identity: &str,
+) -> Vec<NotifyStreamsRequest> {
+    let mut by_namespace: BTreeMap<String, Vec<v1::StreamNotification>> = BTreeMap::new();
+    for notification in notifications {
+        by_namespace
+            .entry(notification.chain.namespace.clone())
+            .or_default()
+            .push(v1::StreamNotification {
+                stream_ref: Some(chain_reference(&notification.chain, notification.topic)),
+                position: notification.position,
+                counter: notification.counter,
+                ..Default::default()
+            });
+    }
+    by_namespace
+        .into_iter()
+        .map(|(namespace, notifications)| NotifyStreamsRequest {
+            namespace,
+            notifications,
+            identity: identity.to_string(),
+        })
+        .collect()
+}
+
 fn notify_request(notification: StreamNotification) -> NotifyStreamRequest {
     NotifyStreamRequest {
         stream_ref: Some(chain_reference(&notification.chain, notification.topic)),
@@ -386,6 +427,50 @@ mod tests {
         assert!(request.close_result.is_none());
     }
 
+    #[test]
+    fn a_batch_of_notifications_goes_out_once_per_namespace() {
+        let notification = |namespace: &str, workflow_id: &str, counter: i64| StreamNotification {
+            chain: proto::ChainId {
+                namespace: namespace.to_string(),
+                workflow_id: workflow_id.to_string(),
+                first_run_id: "run-1".to_string(),
+            },
+            topic: "tokens".to_string(),
+            position: format!("cursor-{counter}"),
+            counter,
+            close: None,
+        };
+        let requests = notify_streams_requests(
+            vec![
+                notification("b", "wf-1", 1),
+                notification("a", "wf-2", 2),
+                notification("b", "wf-3", 3),
+            ],
+            "driver",
+        );
+        let shape: Vec<_> = requests
+            .iter()
+            .map(|r| {
+                (
+                    r.namespace.as_str(),
+                    r.identity.as_str(),
+                    r.notifications
+                        .iter()
+                        .map(|n| n.counter)
+                        .collect::<Vec<_>>(),
+                )
+            })
+            .collect();
+        assert_eq!(
+            shape,
+            [("a", "driver", vec![2]), ("b", "driver", vec![1, 3])]
+        );
+        let first = &requests[1].notifications[0];
+        assert_eq!(first.position, "cursor-1");
+        assert_eq!(first.stream_ref.as_ref().unwrap().workflow_id, "wf-1");
+        assert_eq!(first.stream_ref.as_ref().unwrap().run_id, "run-1");
+    }
+
     #[test]
     fn an_attach_names_the_chains_notifier_and_wakes_the_reader() {
         let request = attach_request(WorkflowAttachment {

crates/streams/src/lib.rs

@@ -13,6 +13,7 @@ mod error;
 mod handle;
 mod memory;
 mod notify;
+mod notifylog;
 mod owner;
 mod reader;
 mod record;
@@ -33,6 +34,7 @@ pub use notify::{
     Notifier, NotifierClient, NotifierPoll, NotifyError, NotifyingStore, StreamNotification,
     WorkflowAttachment, progress_counter,
 };
+pub use notifylog::{NotificationEntry, NotificationLog};
 pub use owner::{OwnerClient, OwnerDescription, OwnerError};
 pub use reader::{
     DEFAULT_MAX_RECORDS, MAX_STATE_BYTES, MAX_TRACKED_PRODUCERS, ReadTarget, read_page,