[Bridge] Notified a stream only while a reader armed it.

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

@@ -38,11 +38,13 @@ 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. It turns on progress only. Closes reach the notifier either way. A
-  // server without the stream notifier costs one call, and then the process stops notifying.
+  // Tells each stream's notifier on the server when the stream moves while a reader sleeps on
+  // it: after an append or a Workflow batch that becomes visible, when a reader armed the stream
+  // with an empty read. That's what makes a caller of a stream-returning Nexus operation see
+  // progress. A busy stream, whose readers find records on each read, costs no notification. Off
+  // by default, since most streams have no such caller. It turns on progress only. Closes reach
+  // the notifier either way. 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;

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

@@ -1296,11 +1296,32 @@ mod tests {
             }
         }
 
+        /// Arms topic `t` of the batches' chain, as a reader that sleeps on progress does, so its
+        /// output notifies.
+        async fn armed(store: &temporalio_streams::MemoryStore) {
+            let read = store
+                .read(StoreReadRequest {
+                    chain: Some(ChainId {
+                        namespace: "ns".to_string(),
+                        workflow_id: "wf".to_string(),
+                        first_run_id: "run".to_string(),
+                    }),
+                    topic: "t".to_string(),
+                    arm: true,
+                    ..Default::default()
+                })
+                .await
+                .unwrap();
+            assert!(read.armed);
+        }
+
         #[tokio::test]
         async fn an_append_notifies_each_topic_it_made_visible() {
             let recorder = Arc::new(Recorder::default());
+            let memory = Arc::new(temporalio_streams::MemoryStore::new());
+            armed(&memory).await;
             let notifying = Arc::new(temporalio_streams::NotifyingStore::new(
-                Arc::new(MemoryStore::new()),
+                memory,
                 Arc::new(temporalio_streams::Notifier::new(recorder.clone(), 10)),
             ));
             let store = OutputStore::new(notifying, "ns".to_string(), Default::default());
@@ -1313,9 +1334,11 @@ mod tests {
         }
 
         /// A Worker's output store over memory that notifies `recorder`.
-        fn notifying_output(recorder: &Arc<Recorder>) -> OutputStore {
+        async fn notifying_output(recorder: &Arc<Recorder>) -> OutputStore {
+            let memory = Arc::new(temporalio_streams::MemoryStore::new());
+            armed(&memory).await;
             let notifying = Arc::new(temporalio_streams::NotifyingStore::new(
-                Arc::new(temporalio_streams::MemoryStore::new()),
+                memory,
                 Arc::new(temporalio_streams::Notifier::new(recorder.clone(), 10)),
             ));
             OutputStore::new(notifying, "ns".to_string(), Default::default())
@@ -1373,7 +1396,7 @@ mod tests {
         #[tokio::test]
         async fn a_workflow_close_ends_the_stream_on_the_server_after_its_output() {
             let recorder = Arc::new(Recorder::default());
-            let store = notifying_output(&recorder);
+            let store = notifying_output(&recorder).await;
             store.append(&batch("t")).await.unwrap();
             store.settle_closes(
                 "run",
@@ -1394,7 +1417,7 @@ mod tests {
         #[tokio::test]
         async fn a_close_replay_proves_ends_the_stream_on_the_server() {
             let recorder = Arc::new(Recorder::default());
-            let store = notifying_output(&recorder);
+            let store = notifying_output(&recorder).await;
             store.append(&batch("t")).await.unwrap();
             // The answer to the completion was lost, so the close waits for replay.
             store.settle_closes(

crates/streams/src/lib.rs

@@ -30,7 +30,7 @@ pub use handle::{OWNER_RECHECK, Streams, StreamsOptions};
 pub use memory::MemoryStore;
 pub use nexus::{OPERATION_TOKEN_VERSION, OperationToken};
 pub use notify::{
-    CLOSE_RETRY_CAP, CLOSE_RETRY_FIRST, CallerAttachment, DEFAULT_MAX_NOTIFIERS, FLUSH_LIMIT,
+    Arm, CLOSE_RETRY_CAP, CLOSE_RETRY_FIRST, CallerAttachment, DEFAULT_MAX_NOTIFIERS, FLUSH_LIMIT,
     Notifier, NotifierClient, NotifyError, NotifyingStore, StreamNotification, progress_counter,
 };
 pub use owner::{OwnerClient, OwnerDescription, OwnerError};