[Jetty] Added notify_every_append for readers that can't arm.

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

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

crates/streams/src/notify.rs

@@ -672,6 +672,7 @@ pub struct NotifyingStore {
     inner: Arc<dyn StreamStore>,
     notifier: Arc<Notifier>,
     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<Notifier> {
         &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<u64>) -> Option<Option<Arm>> {
+        if self.every_append {
+            return Some(None);
+        }
         match arm {
             Some(0) => None,
             Some(generation) => Some(Some(Arm {

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<i64> = 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());