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 54b7084051..32b8468c69 100644 --- a/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto +++ b/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto @@ -49,6 +49,10 @@ message StreamStoreConfig { 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; + // With `notify_on_append`, notifies after every append, whether or not a reader armed the + // stream. For a deployment whose readers can't arm, + // such as a hand-written reader over HTTP. Off by default. + bool notify_every_append = 5; // How long a read waits for a Workflow that Temporal doesn't know yet, counted from the read's // first owner check. Until then the read answers empty, so a reader can start before its // owner does. After it the read fails as NOT_FOUND. Unset waits 60 seconds, and zero fails at diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index 44f78e2239..95a142a712 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/crates/sdk-core/CHANGELOG.md @@ -41,10 +41,19 @@ relevant information. * Experimental: the C bridge has `temporal_core_stream_store_new`, `_call` and `_free` for the stream service, built with the Redis store. * Experimental: with `StreamStoreConfig.notify_on_append`, Core tells each stream's notifier on - the server when the stream moves, after every append. That's how a reader that waits on the - notifier wakes. 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). + the server when the stream moves while a reader sleeps on it. That's how a reader that waits on + the notifier wakes. A reader arms the stream with a read that sets the new `ReadRequest.arm` + and waits its whole wait with no record. The answer says `armed` and carries `wake_after`, the + counter of the read's position, and the next append notifies once. Core clears the arm once the + notification lands, and a failed one keeps it for the next append. A busy stream, whose readers + find records on each read, sends no notification, apart from one refresh a day that keeps its + notifier from the idle timeout. `notify_every_append` restores a notification after every + append, for readers that can't arm. 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: `StreamStore` has `disarm` and `newest_or_arm`, and its read and append messages + carry the arm. The Redis store keeps it in each topic's meta as `arm` and `arm_seq`, which + changes the append script hash. * 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/notify.rs b/crates/streams/src/notify.rs index 3fc45f68e2..d6793d4a36 100644 --- a/crates/streams/src/notify.rs +++ b/crates/streams/src/notify.rs @@ -672,6 +672,7 @@ pub struct NotifyingStore { inner: Arc, notifier: Arc, progress: bool, + every_append: bool, } impl NotifyingStore { @@ -681,6 +682,7 @@ impl NotifyingStore { inner, notifier, progress: true, + every_append: false, } } @@ -691,6 +693,13 @@ impl NotifyingStore { self } + /// With progress on, notifies after every write whether or not a reader armed the stream, for + /// readers that can't arm. + pub fn with_every_append(mut self, every_append: bool) -> Self { + self.every_append = every_append; + self + } + /// The notifier this store notifies through. pub fn notifier(&self) -> &Arc { &self.notifier @@ -706,6 +715,9 @@ impl NotifyingStore { /// sent, and otherwise the arm to clear once the notification lands. A store that can't tell /// reports no arm, and its writes notify without one. fn owed(&self, arm: Option) -> Option> { + if self.every_append { + return Some(None); + } match arm { Some(0) => None, Some(generation) => Some(Some(Arm { diff --git a/crates/streams/src/service.rs b/crates/streams/src/service.rs index ae1f6562f7..35dca41fd6 100644 --- a/crates/streams/src/service.rs +++ b/crates/streams/src/service.rs @@ -78,7 +78,8 @@ impl StreamService { }; Arc::new( NotifyingStore::new(store, Arc::new(Notifier::new(client, max_notifiers))) - .with_progress(config.notify_on_append), + .with_progress(config.notify_on_append) + .with_every_append(config.notify_every_append), ) } None => store, @@ -620,6 +621,43 @@ mod tests { assert_eq!(sent[0].counter, 2); } + #[tokio::test] + async fn with_every_append_on_an_append_with_nothing_armed_notifies() { + 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, + notify_every_append: true, + ..Default::default() + }, + Arc::new(Running), + Some(recorder.clone()), + ) + .await + .unwrap(); + service.call("Append", &append(1, &["a"])).await.unwrap(); + service.call("Append", &append(2, &["b"])).await.unwrap(); + flush(&service).await; + let counters: Vec = recorder + .0 + .lock() + .unwrap() + .iter() + .map(|n| n.counter) + .collect(); + assert_eq!(counters.last(), Some(&2), "{counters:?}"); + } + + #[tokio::test] + async fn without_every_append_an_append_with_nothing_armed_sends_nothing() { + let recorder = Arc::new(Recorder::default()); + let service = service_notifying(true, recorder.clone()).await; + service.call("Append", &append(1, &["a"])).await.unwrap(); + flush(&service).await; + assert!(recorder.0.lock().unwrap().is_empty()); + } + #[tokio::test] async fn notifications_are_off_by_default() { let recorder = Arc::new(Recorder::default());