From f9c0ccb16b03bb5d96d799ef0164aab2238fd471 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Wed, 14 Oct 2026 16:32:23 -0700 Subject: [PATCH 1/3] Vendored NotifyStreams and sent driver notifications with it. A driver of a notification log tells many notifiers at once. One call names one namespace, so Core sends one per namespace. --- crates/client/src/grpc.rs | 9 ++ .../api_upstream/openapi/openapiv2.json | 2 +- .../api_upstream/openapi/openapiv3.yaml | 2 +- .../temporal/api/stream/v1/message.proto | 15 ++- .../workflowservice/v1/request_response.proto | 14 ++- .../api/workflowservice/v1/service.proto | 22 ++++- crates/sdk-core-c-bridge/src/client.rs | 3 + crates/sdk-core/src/streams.rs | 91 ++++++++++++++++++- 8 files changed, 149 insertions(+), 9 deletions(-) diff --git a/crates/client/src/grpc.rs b/crates/client/src/grpc.rs index ef9e212e41..9b9b88f6e3 100644 --- a/crates/client/src/grpc.rs +++ b/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, diff --git a/crates/protos/protos/api_upstream/openapi/openapiv2.json b/crates/protos/protos/api_upstream/openapi/openapiv2.json index a677fc013c..ce4d4aa584 100644 --- a/crates/protos/protos/api_upstream/openapi/openapiv2.json +++ b/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", diff --git a/crates/protos/protos/api_upstream/openapi/openapiv3.yaml b/crates/protos/protos/api_upstream/openapi/openapiv3.yaml index 92e4433895..9f11cceca4 100644 --- a/crates/protos/protos/api_upstream/openapi/openapiv3.yaml +++ b/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 diff --git a/crates/protos/protos/api_upstream/temporal/api/stream/v1/message.proto b/crates/protos/protos/api_upstream/temporal/api/stream/v1/message.proto index a179a71aad..2576cdb42f 100644 --- a/crates/protos/protos/api_upstream/temporal/api/stream/v1/message.proto +++ b/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,16 @@ 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 metadata = 4; +} 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 aaa5bca8bc..62f2fcf7e4 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 @@ -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,15 @@ 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 { +} 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 7aec05b140..1d38044e09 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 @@ -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,23 @@ 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. A call that fails may have applied some entries. Sending + // the batch 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) { + } } diff --git a/crates/sdk-core-c-bridge/src/client.rs b/crates/sdk-core-c-bridge/src/client.rs index 516a04ad9c..cb43f6f688 100644 --- a/crates/sdk-core-c-bridge/src/client.rs +++ b/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" => { diff --git a/crates/sdk-core/src/streams.rs b/crates/sdk-core/src/streams.rs index ae55b97817..809e031fae 100644 --- a/crates/sdk-core/src/streams.rs +++ b/crates/sdk-core/src/streams.rs @@ -1,16 +1,16 @@ //! Streams for lang: one process-wide stream service over the configured store, asking the //! client's server about streams' owners. -use std::sync::Arc; +use std::{collections::BTreeMap, sync::Arc}; use temporalio_client::Connection; 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::{ @@ -53,6 +53,20 @@ impl NotifierClient for ConnectionNotifierClient { .map_err(notify_error) } + async fn notify_streams( + &self, + notifications: Vec, + ) -> 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() @@ -149,6 +163,33 @@ fn attach_request(attachment: WorkflowAttachment) -> AttachStreamCallbackRequest } } +/// One `NotifyStreams` request per namespace, since a call names one. +fn notify_streams_requests( + notifications: Vec, + identity: &str, +) -> Vec { + let mut by_namespace: BTreeMap> = BTreeMap::new(); + for notification in notifications { + by_namespace + .entry(notification.chain.namespace.clone()) + .or_default() + .push(v1::StreamNotification { + stream_ref: Some(chain_reference(¬ification.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(¬ification.chain, notification.topic)), @@ -352,6 +393,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::>(), + ) + }) + .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 { From 943ef40dff6eed017a33a3da54b37f113f88d3d3 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Wed, 14 Oct 2026 16:32:23 -0700 Subject: [PATCH 2/3] Added a notification log to the store contract and the memory store. A store that writes one entry per append in the same step lets a driver notify for every producer. Producers then send nothing, and a store without a log keeps the producer-side notify. --- .../temporal/sdk/core/streams/streams.proto | 4 + crates/sdk-core/CHANGELOG.md | 7 + crates/streams/src/lib.rs | 2 + crates/streams/src/memory.rs | 141 ++++++++++++++++- crates/streams/src/notify.rs | 20 ++- crates/streams/src/notifylog.rs | 62 ++++++++ crates/streams/src/service.rs | 59 ++++++- crates/streams/src/store.rs | 7 + crates/streams/tests/conformance.rs | 147 +++++++++++++++++- 9 files changed, 442 insertions(+), 7 deletions(-) create mode 100644 crates/streams/src/notifylog.rs diff --git a/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto b/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto index 2993b4f8ab..efa809cc82 100644 --- a/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto +++ b/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. diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index b63aed0bdc..2417ecd4a2 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/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 diff --git a/crates/streams/src/lib.rs b/crates/streams/src/lib.rs index d578386618..97455aaacf 100644 --- a/crates/streams/src/lib.rs +++ b/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::{ 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, diff --git a/crates/streams/src/memory.rs b/crates/streams/src/memory.rs index d733a1307d..9877027401 100644 --- a/crates/streams/src/memory.rs +++ b/crates/streams/src/memory.rs @@ -6,7 +6,7 @@ //! position keeps naming the same record after [MemoryStore::truncate] drops older ones. use crate::{ - StreamError, StreamResult, StreamStore, + NotificationEntry, NotificationLog, StreamError, StreamResult, StreamStore, proto::{ ChainId, DeleteOwnerRequest, DeleteOwnerResponse, StoreAppendRequest, StoreAppendResponse, StoreLatestRequest, StoreLatestResponse, StoreReadRequest, StoreReadResponse, StoredClose, @@ -15,7 +15,7 @@ use crate::{ }; use std::{ collections::{HashMap, VecDeque}, - sync::Mutex, + sync::{Arc, Mutex}, time::Duration, }; use tokio::{sync::watch, time::Instant}; @@ -88,6 +88,7 @@ pub struct MemoryStore { chains: Mutex>, /// Bumped on every write, so a waiting read checks again. written: watch::Sender, + log: Option>, } impl Default for MemoryStore { @@ -95,16 +96,144 @@ impl Default for MemoryStore { Self { chains: Mutex::default(), written: watch::Sender::new(0), + log: None, } } } +/// A notification log in memory: one partition and one doorbell. +#[derive(Debug)] +struct MemoryLog { + state: Mutex, + /// Bumped on every entry, so a waiting driver looks again. + rung: watch::Sender, +} + +#[derive(Debug, Default)] +struct LogState { + next_id: u64, + items: VecDeque, +} + +#[derive(Debug)] +struct LogItem { + entry: NotificationEntry, + /// The consumer that took the entry and when, until an ack drops it. + taken: Option<(String, Instant)>, +} + +impl MemoryLog { + fn push(&self, chain: &ChainId, topic: &str, position: String) { + let mut state = self.state.lock().unwrap(); + state.next_id += 1; + let id = state.next_id.to_string(); + state.items.push_back(LogItem { + entry: NotificationEntry { + id, + chain: chain.clone(), + topic: topic.to_string(), + position, + }, + taken: None, + }); + drop(state); + self.rung.send_modify(|rung| *rung += 1); + } + + fn holds_entries(&self) -> bool { + !self.state.lock().unwrap().items.is_empty() + } +} + +#[async_trait::async_trait] +impl NotificationLog for MemoryLog { + fn partitions(&self) -> u32 { + 1 + } + + fn doorbells(&self) -> u32 { + 1 + } + + async fn wait_doorbell( + &self, + _doorbell: u32, + _consumer: &str, + block: Duration, + ) -> StreamResult> { + let mut rung = self.rung.subscribe(); + if !self.holds_entries() { + let _ = tokio::time::timeout(block, rung.changed()).await; + } + Ok(if self.holds_entries() { + vec![0] + } else { + vec![] + }) + } + + async fn take( + &self, + _partition: u32, + consumer: &str, + claim_idle: Duration, + max: usize, + ) -> StreamResult> { + let now = Instant::now(); + let mut state = self.state.lock().unwrap(); + let mut taken = vec![]; + for item in state.items.iter_mut() { + if taken.len() >= max { + break; + } + let free = match &item.taken { + None => true, + Some((owner, _)) if owner == consumer => true, + Some((_, at)) => now.duration_since(*at) >= claim_idle, + }; + if free { + item.taken = Some((consumer.to_string(), now)); + taken.push(item.entry.clone()); + } + } + Ok(taken) + } + + async fn ack(&self, _partition: u32, ids: &[String]) -> StreamResult<()> { + self.state + .lock() + .unwrap() + .items + .retain(|item| !ids.contains(&item.entry.id)); + Ok(()) + } + + async fn busy(&self) -> StreamResult> { + Ok(if self.holds_entries() { + vec![0] + } else { + vec![] + }) + } +} + impl MemoryStore { /// An empty store. pub fn new() -> Self { Self::default() } + /// An empty store whose appends write a notification log in the same step. + pub fn with_notification_log() -> Self { + Self { + log: Some(Arc::new(MemoryLog { + state: Mutex::default(), + rung: watch::Sender::new(0), + })), + ..Self::default() + } + } + /// Drops all but the newest `keep` records of a topic, standing in for a store's retention. /// It keeps the records a positions-only reader holds, from the position it holds on. pub fn truncate(&self, chain: &ChainId, topic: &str, keep: usize) { @@ -212,6 +341,10 @@ impl StreamStore for MemoryStore { NAME } + fn notification_log(&self) -> Option> { + self.log.clone().map(|log| log as Arc) + } + async fn append(&self, request: StoreAppendRequest) -> StreamResult { if request.records.is_empty() { return Err(StreamError::refused("an append needs at least one record")); @@ -279,6 +412,10 @@ impl StreamStore for MemoryStore { digest: request.digest, }, ); + // Under the store's lock, so the entry lands in the same step as the records. + if let (Some(log), Some(chain)) = (&self.log, &request.chain) { + log.push(chain, &request.topic, last.to_string()); + } drop(chains); self.wrote(); Ok(StoreAppendResponse { diff --git a/crates/streams/src/notify.rs b/crates/streams/src/notify.rs index e322abdd07..4b253c9144 100644 --- a/crates/streams/src/notify.rs +++ b/crates/streams/src/notify.rs @@ -118,6 +118,18 @@ pub trait NotifierClient: Send + Sync { /// Sends one notification. async fn notify(&self, notification: StreamNotification) -> Result<(), NotifyError>; + /// Sends notifications for many streams in one call, as a driver of a notification log does. + /// A client without a batch call sends them one at a time. + async fn notify_streams( + &self, + notifications: Vec, + ) -> Result<(), NotifyError> { + for notification in notifications { + self.notify(notification).await?; + } + Ok(()) + } + /// Attaches a Workflow's reader to a stream's notifier. Idempotent by its request id. async fn attach_workflow(&self, attachment: WorkflowAttachment) -> Result<(), NotifyError> { let _ = attachment; @@ -593,7 +605,8 @@ impl NotifierClient for UnlessUnsupported { } /// A store that closes each stream's notifier after the store closes the stream, and with -/// progress on, notifies it after every append that lands. +/// progress on, notifies it after every append that lands. A store with a notification log +/// notifies through its driver, so its appends send nothing here. /// /// Wraps the store a process hands to its stream calls, so every write path notifies without /// knowing about notifiers. It forwards every trait method by name, since a @@ -683,6 +696,7 @@ impl StreamStore for NotifyingStore { let topic = request.topic.clone(); let landed = self.inner.append(request).await?; if self.progress + && self.inner.notification_log().is_none() && let Some(chain) = &chain { self.notify_at(chain, &topic, &landed.last_position); @@ -806,6 +820,10 @@ impl StreamStore for NotifyingStore { async fn flush_notifications(&self) { self.notifier.flush().await; } + + fn notification_log(&self) -> Option> { + self.inner.notification_log() + } } #[cfg(test)] diff --git a/crates/streams/src/notifylog.rs b/crates/streams/src/notifylog.rs new file mode 100644 index 0000000000..e607c4995c --- /dev/null +++ b/crates/streams/src/notifylog.rs @@ -0,0 +1,62 @@ +//! The notification log: one entry per append, written by the store in the same step as the +//! append, which a driver reads and hands to the server's stream notifiers. +//! +//! With a log, a producer sends nothing to the server, and readers still hear of every append +//! once a driver runs. Entries go to a fixed set of partitions, so drivers can share the work, +//! and a doorbell tells a waiting driver which partition took its first entry. Delivery is at +//! least once. A repeat carries a counter the notifier has seen and delivers nothing. + +use crate::{StreamResult, proto::ChainId}; +use std::time::Duration; + +/// One append, as the log keeps it until a driver hands it to the stream's notifier. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct NotificationEntry { + /// The entry's id in its partition, which the ack names. + pub id: String, + /// The run chain whose stream moved. + pub chain: ChainId, + /// The stream's topic. + pub topic: String, + /// The store position of the append's last record. + pub position: String, +} + +/// The notification log of a store, for the drivers that read it. +/// +/// Every driver reads as one consumer group, so an entry goes to one consumer at a time. An entry +/// stays until a driver acks it. A consumer that stops without an ack leaves its entries pending, +/// and after `claim_idle` another consumer takes them. +#[async_trait::async_trait] +pub trait NotificationLog: Send + Sync { + /// How many partitions the log has. One stream's entries always land in the same one. + fn partitions(&self) -> u32; + + /// How many doorbells the log rings. Partition `p` rings doorbell `p % doorbells`. + fn doorbells(&self) -> u32; + + /// Waits up to `block` for doorbell `doorbell`, as `consumer`, and answers the partitions it + /// rang. A ring can be lost, so a driver also sweeps with [NotificationLog::busy]. + async fn wait_doorbell( + &self, + doorbell: u32, + consumer: &str, + block: Duration, + ) -> StreamResult>; + + /// At most `max` entries of `partition` for `consumer`: the entries another consumer took and + /// left unacked for `claim_idle`, this consumer's own unacked ones, and new ones. + async fn take( + &self, + partition: u32, + consumer: &str, + claim_idle: Duration, + max: usize, + ) -> StreamResult>; + + /// Acks and drops entries of `partition` once their notifications landed. + async fn ack(&self, partition: u32, ids: &[String]) -> StreamResult<()>; + + /// The partitions that hold entries now, acked or not. + async fn busy(&self) -> StreamResult>; +} diff --git a/crates/streams/src/service.rs b/crates/streams/src/service.rs index 4504469842..dadcaebe36 100644 --- a/crates/streams/src/service.rs +++ b/crates/streams/src/service.rs @@ -60,7 +60,12 @@ impl StreamService { ) -> StreamResult { let (store, retention): (Arc, Duration) = match config.store { Some(stream_store_config::Store::Memory(_)) => { - (Arc::new(crate::MemoryStore::new()), DEFAULT_RETENTION) + let store = if config.notification_log { + crate::MemoryStore::with_notification_log() + } else { + crate::MemoryStore::new() + }; + (Arc::new(store), DEFAULT_RETENTION) } Some(stream_store_config::Store::Redis(redis)) => connect_redis(redis).await?, None => { @@ -69,6 +74,12 @@ impl StreamService { )); } }; + if config.notification_log && store.notification_log().is_none() { + return Err(StreamError::refused(format!( + "the {} store keeps no notification log", + store.name() + ))); + } if config.notify_on_append && notifier.is_none() { return Err(StreamError::refused( "stream notifications need a client to send them with", @@ -664,6 +675,52 @@ mod tests { assert_eq!(counters.last(), Some(&2), "{counters:?}"); } + #[tokio::test] + async fn with_a_notification_log_an_append_writes_an_entry_and_sends_nothing() { + let recorder = Arc::new(Recorder::default()); + let service = StreamService::connect_with_notifier( + StreamStoreConfig { + store: Some(stream_store_config::Store::Memory(MemoryStoreConfig {})), + notify_on_append: true, + notification_log: true, + ..Default::default() + }, + Arc::new(Running), + Some(recorder.clone()), + ) + .await + .unwrap(); + service.call("Append", &append(1, &["a"])).await.unwrap(); + flush(&service).await; + assert!(recorder.0.lock().unwrap().is_empty(), "the driver notifies"); + let log = service.store().notification_log().unwrap(); + assert_eq!(log.busy().await.unwrap(), [0]); + } + + #[cfg(feature = "redis")] + #[tokio::test] + async fn a_store_without_a_notification_log_refuses_one() { + let Ok(url) = std::env::var("STREAMS_REDIS_URL") else { + return; + }; + let error = StreamService::connect( + StreamStoreConfig { + store: Some(stream_store_config::Store::Redis( + crate::proto::RedisStoreConfig { + urls: vec![url], + ..Default::default() + }, + )), + notification_log: true, + ..Default::default() + }, + Arc::new(Running), + ) + .await + .unwrap_err(); + assert_eq!(error.kind, StreamFailureKind::Refused); + } + #[tokio::test] async fn notifications_are_off_by_default() { let recorder = Arc::new(Recorder::default()); diff --git a/crates/streams/src/store.rs b/crates/streams/src/store.rs index 73059f08e2..cc9732f879 100644 --- a/crates/streams/src/store.rs +++ b/crates/streams/src/store.rs @@ -209,6 +209,13 @@ pub trait StreamStore: Send + Sync { /// Waits until the notifications this store sent are out. A store that sends none returns at /// once. async fn flush_notifications(&self) {} + + /// The store's notification log, when each append writes an entry to it in the same step. + /// Producers then send no notifications and a driver reads the log. A store without one + /// keeps the default, and each producer notifies after its append. + fn notification_log(&self) -> Option> { + None + } } #[cfg(test)] diff --git a/crates/streams/tests/conformance.rs b/crates/streams/tests/conformance.rs index 4189b6c761..a426f9bed6 100644 --- a/crates/streams/tests/conformance.rs +++ b/crates/streams/tests/conformance.rs @@ -15,9 +15,9 @@ use std::{ }; use temporalio_common::protos::temporal::api::common::v1::Payload; use temporalio_streams::{ - BEGINNING, END, MemoryStore, Notifier, NotifierClient, NotifyError, NotifyingStore, ReadTarget, - StreamNotification, StreamResult, StreamStore, mint_cursor, proto::*, read_page, - stored_append_record, stream_hash, + BEGINNING, END, MemoryStore, NotificationEntry, NotificationLog, Notifier, NotifierClient, + NotifyError, NotifyingStore, ReadTarget, StreamNotification, StreamResult, StreamStore, + mint_cursor, proto::*, read_page, stored_append_record, stream_hash, }; #[async_trait::async_trait] @@ -74,6 +74,10 @@ async fn memory() -> Option { Some(Memory(Arc::new(MemoryStore::new()))) } +async fn memory_with_log() -> Option { + Some(Memory(Arc::new(MemoryStore::with_notification_log()))) +} + /// Takes every notification and sends nothing. struct Quiet; @@ -1419,6 +1423,124 @@ mod cases { ["1"] ); } + + /// The store's log. A case runs only on a store that keeps one. + fn log_of(case: &dyn Case) -> Arc { + case.store() + .notification_log() + .expect("the notification log cases run on stores that keep a log") + } + + /// Takes the entries of `chain` from every busy partition, as `consumer`. + async fn entries_of( + log: &dyn NotificationLog, + chain: &ChainId, + consumer: &str, + claim_idle: Duration, + ) -> Vec<(u32, NotificationEntry)> { + let mut found = vec![]; + for partition in log.busy().await.unwrap() { + for entry in log + .take(partition, consumer, claim_idle, 1000) + .await + .unwrap() + { + if &entry.chain == chain { + found.push((partition, entry)); + } + } + } + found + } + + pub(crate) async fn an_append_writes_one_entry_with_its_last_position(case: &dyn Case) { + let log = log_of(case); + let chain = case.chain(); + let mut producer = Producer::new(case, &chain, OUT, "p", 1); + let landed = producer.append(&["1", "2"]).await; + // A repeat of the newest batch writes nothing again. + producer.sequence = 1; + assert_eq!(producer.append(&["1", "2"]).await, landed); + let entries = entries_of(log.as_ref(), &chain, "a", Duration::ZERO).await; + let [(partition, entry)] = entries.as_slice() else { + panic!("one append gives one entry: {entries:?}"); + }; + assert!(*partition < log.partitions()); + assert_eq!( + (entry.topic.as_str(), entry.position.as_str()), + (OUT, landed.last_position.as_str()) + ); + } + + pub(crate) async fn an_entry_stays_until_its_ack(case: &dyn Case) { + let log = log_of(case); + let chain = case.chain(); + Producer::new(case, &chain, OUT, "p", 1) + .append(&["1"]) + .await; + let first = entries_of(log.as_ref(), &chain, "a", Duration::ZERO).await; + assert_eq!(first.len(), 1); + // A consumer that took an entry and didn't ack it gets it again. + let again = entries_of(log.as_ref(), &chain, "a", Duration::ZERO).await; + assert_eq!(again, first); + let (partition, entry) = &first[0]; + log.ack(*partition, std::slice::from_ref(&entry.id)) + .await + .unwrap(); + assert!( + entries_of(log.as_ref(), &chain, "a", Duration::ZERO) + .await + .is_empty() + ); + } + + pub(crate) async fn another_consumer_claims_an_entry_only_after_claim_idle(case: &dyn Case) { + let log = log_of(case); + let chain = case.chain(); + Producer::new(case, &chain, OUT, "p", 1) + .append(&["1"]) + .await; + let hour = Duration::from_secs(3600); + assert_eq!(entries_of(log.as_ref(), &chain, "a", hour).await.len(), 1); + assert!( + entries_of(log.as_ref(), &chain, "b", hour).await.is_empty(), + "a live consumer keeps what it took" + ); + tokio::time::sleep(Duration::from_millis(20)).await; + let claimed = entries_of(log.as_ref(), &chain, "b", Duration::from_millis(10)).await; + assert_eq!(claimed.len(), 1, "a stopped consumer's entry moves on"); + } + + pub(crate) async fn a_refused_append_writes_no_entry(case: &dyn Case) { + let log = log_of(case); + let store = case.store(); + let chain = case.chain(); + store.close_topic(&chain, OUT).await.unwrap(); + let refused = Producer::new(case, &chain, OUT, "p", 1) + .send(&[data("1")]) + .await; + assert!(refused.is_err()); + assert!( + entries_of(log.as_ref(), &chain, "a", Duration::ZERO) + .await + .is_empty() + ); + } + + pub(crate) async fn the_doorbell_of_a_partition_with_entries_rings(case: &dyn Case) { + let log = log_of(case); + let chain = case.chain(); + Producer::new(case, &chain, OUT, "p", 1) + .append(&["1"]) + .await; + let entries = entries_of(log.as_ref(), &chain, "a", Duration::ZERO).await; + let partition = entries[0].0; + let rung = log + .wait_doorbell(partition % log.doorbells(), "a", Duration::from_millis(100)) + .await + .unwrap(); + assert!(rung.contains(&partition), "{rung:?}"); + } } macro_rules! conformance { @@ -1479,13 +1601,32 @@ macro_rules! store_cases { }; } +/// The notification log, for a store that keeps one. +macro_rules! log_cases { + ($case:expr) => { + conformance!($case; + an_append_writes_one_entry_with_its_last_position, + an_entry_stays_until_its_ack, + another_consumer_claims_an_entry_only_after_claim_idle, + a_refused_append_writes_no_entry, + the_doorbell_of_a_partition_with_entries_rings, + ); + }; +} + mod memory { store_cases!(super::memory()); conformance!(super::memory(); deleting_an_owner_drops_every_chain_of_it); } +mod memory_with_log { + store_cases!(super::memory_with_log()); + log_cases!(super::memory_with_log()); +} + mod notifying_memory { store_cases!(super::notifying(super::memory())); + log_cases!(super::notifying(super::memory_with_log())); conformance!(super::notifying(super::memory()); deleting_an_owner_drops_every_chain_of_it, ); From 64871840738841c92b4ea8aab7f8fc8b1e3dc90e Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 16 Oct 2026 08:27:12 -0700 Subject: [PATCH 3/3] Vendored the per-entry `NotifyStreams` results. The server answers each entry, so a driver can settle one bad entry without the others. --- .../api_upstream/temporal/api/stream/v1/message.proto | 9 +++++++++ .../api/workflowservice/v1/request_response.proto | 3 +++ .../temporal/api/workflowservice/v1/service.proto | 6 ++++-- 3 files changed, 16 insertions(+), 2 deletions(-) diff --git a/crates/protos/protos/api_upstream/temporal/api/stream/v1/message.proto b/crates/protos/protos/api_upstream/temporal/api/stream/v1/message.proto index 2576cdb42f..e92037ad87 100644 --- a/crates/protos/protos/api_upstream/temporal/api/stream/v1/message.proto +++ b/crates/protos/protos/api_upstream/temporal/api/stream/v1/message.proto @@ -104,3 +104,12 @@ message StreamNotification { // Small details passed on to the callbacks. At most 2 KiB in total, counted as UTF-8 bytes. map 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; +} 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 62f2fcf7e4..6fd209726e 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 @@ -3824,4 +3824,7 @@ message NotifyStreamsRequest { } 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; } 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 1d38044e09..f995f4abf8 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 @@ -2095,8 +2095,10 @@ service WorkflowService { // 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. A call that fails may have applied some entries. Sending - // the batch again is safe, since a notifier ignores a counter at or below the one it holds. + // 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. --)