[Bridge] Added the reader arm to the stream store contract.

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

@@ -294,6 +294,9 @@ message StoreAppendRequest {
 message StoreAppendResponse {
   string first_position = 1;
   string last_position = 2;
+  // The topic's arm generation as the append found it, zero when no reader armed the topic.
+  // Unset when the store can't tell, and Core then notifies after the append.
+  optional uint64 arm = 3;
 }
 
 message StoreReadRequest {
@@ -303,6 +306,9 @@ message StoreReadRequest {
   string after_position = 3;
   google.protobuf.Duration wait = 4;
   uint32 max_records = 5;
+  // When the read ends with no record and the topic is open, arm the topic in the same step that
+  // finds it empty, so the next append reports the arm. A store that can't arm ignores it.
+  bool arm = 6;
 }
 
 message StoredRecord {
@@ -315,6 +321,9 @@ message StoreReadResponse {
   repeated StoredRecord records = 1;
   // The chain or this topic is marked closed.
   bool closed = 2;
+  // The read asked to arm, found no record and armed the topic. Never set with records or with
+  // `closed`.
+  bool armed = 3;
 }
 
 message StoreLatestRequest {

crates/streams/src/lib.rs

@@ -40,7 +40,7 @@ pub use record::{
     stored_output_record, workflow_producer_id,
 };
 pub use service::StreamService;
-pub use store::{CloseOutcome, StreamStore, empty_close_result};
+pub use store::{CloseOutcome, NewestOrArm, StreamStore, empty_close_result};
 
 /// The protos stores and lang exchange.
 pub mod proto {

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,
+    NewestOrArm, StreamError, StreamResult, StreamStore,
     proto::{
         ChainId, DeleteOwnerRequest, DeleteOwnerResponse, StoreAppendRequest, StoreAppendResponse,
         StoreLatestRequest, StoreLatestResponse, StoreReadRequest, StoreReadResponse, StoredRecord,
@@ -72,6 +72,29 @@ impl Topic {
 struct Chain {
     topics: HashMap<String, Topic>,
     closed: bool,
+    /// Kept apart from the topics, so an arm on a topic with no records leaves it unknown to
+    /// reads and to the topic list.
+    arms: HashMap<String, Arm>,
+}
+
+/// A topic's arm: the generation a reader set, zero once cleared, and the last one handed out.
+#[derive(Debug, Default)]
+struct Arm {
+    generation: u64,
+    last: u64,
+}
+
+impl Chain {
+    fn arm(&self, topic: &str) -> u64 {
+        self.arms.get(topic).map_or(0, |arm| arm.generation)
+    }
+
+    fn set_arm(&mut self, topic: &str) -> u64 {
+        let arm = self.arms.entry(topic.to_string()).or_default();
+        arm.last += 1;
+        arm.generation = arm.last;
+        arm.generation
+    }
 }
 
 /// A [StreamStore] in process memory. For tests only.
@@ -118,9 +141,33 @@ impl MemoryStore {
         self.written.send_modify(|version| *version += 1);
     }
 
-    /// One look at what a read gets now: `None` when it should wait for more.
-    fn read_now(&self, request: &StoreReadRequest) -> StreamResult<Option<StoreReadResponse>> {
-        let chains = self.chains.lock().unwrap();
+    /// One look at what a read gets now: `None` when it should wait for more. With `arm`, a look
+    /// that finds nothing arms the topic instead.
+    fn read_now(
+        &self,
+        request: &StoreReadRequest,
+        arm: bool,
+    ) -> StreamResult<Option<StoreReadResponse>> {
+        let mut chains = self.chains.lock().unwrap();
+        let found = Self::look(&chains, request)?;
+        if found.is_some() || !arm {
+            return Ok(found);
+        }
+        // Under the lock the look held, so no append lands between finding nothing and the arm.
+        chains
+            .entry(chain_key(chain_of(request.chain.as_ref())?))
+            .or_default()
+            .set_arm(&request.topic);
+        Ok(Some(StoreReadResponse {
+            armed: true,
+            ..Default::default()
+        }))
+    }
+
+    fn look(
+        chains: &HashMap<ChainKey, Chain>,
+        request: &StoreReadRequest,
+    ) -> StreamResult<Option<StoreReadResponse>> {
         let chain = chains.get(&chain_key(chain_of(request.chain.as_ref())?));
         let topic = chain.and_then(|chain| chain.topics.get(&request.topic));
         let closed = chain.is_some_and(|chain| chain.closed) || topic.is_some_and(|t| t.closed);
@@ -134,8 +181,8 @@ impl MemoryStore {
                 )));
             }
             return Ok(closed.then(|| StoreReadResponse {
-                records: Vec::new(),
                 closed,
+                ..Default::default()
             }));
         };
         let start = after.map_or(topic.base, |after| after + 1);
@@ -159,7 +206,13 @@ impl MemoryStore {
                 record: topic.at(offset).cloned().unwrap_or_default(),
             })
             .collect();
-        Ok((!records.is_empty() || closed).then_some(StoreReadResponse { records, closed }))
+        Ok(
+            (!records.is_empty() || closed).then_some(StoreReadResponse {
+                records,
+                closed,
+                armed: false,
+            }),
+        )
     }
 }
 
@@ -222,6 +275,7 @@ impl StreamStore for MemoryStore {
                 return Ok(StoreAppendResponse {
                     first_position: held.first.to_string(),
                     last_position: last.to_string(),
+                    arm: Some(chain.arm(&request.topic)),
                 });
             }
             let next = held.sequence + held.count;
@@ -241,6 +295,7 @@ impl StreamStore for MemoryStore {
                 "the Workflow that owns this stream has closed",
             ));
         }
+        let arm = chain.arm(&request.topic);
         let topic = chain.topics.entry(request.topic.clone()).or_default();
         let count = request.records.len() as i64;
         let mut positions = request.records.into_iter().map(|record| topic.push(record));
@@ -260,21 +315,22 @@ impl StreamStore for MemoryStore {
         Ok(StoreAppendResponse {
             first_position: first.to_string(),
             last_position: last.to_string(),
+            arm: Some(arm),
         })
     }
 
     async fn read(&self, request: StoreReadRequest) -> StreamResult<StoreReadResponse> {
         let deadline = Instant::now() + wait_of(&request);
         let mut written = self.written.subscribe();
         loop {
-            if let Some(response) = self.read_now(&request)? {
+            if let Some(response) = self.read_now(&request, false)? {
                 return Ok(response);
             }
             if tokio::time::timeout_at(deadline, written.changed())
                 .await
                 .is_err()
             {
-                return Ok(StoreReadResponse::default());
+                return Ok(self.read_now(&request, request.arm)?.unwrap_or_default());
             }
         }
     }
@@ -383,6 +439,27 @@ impl StreamStore for MemoryStore {
             .unwrap_or_default())
     }
 
+    async fn disarm(&self, chain: &ChainId, topic: &str, generation: u64) -> StreamResult<()> {
+        let mut chains = self.chains.lock().unwrap();
+        if let Some(arm) = chains
+            .get_mut(&chain_key(chain))
+            .and_then(|chain| chain.arms.get_mut(topic))
+            && arm.generation == generation
+        {
+            arm.generation = 0;
+        }
+        Ok(())
+    }
+
+    async fn newest_or_arm(&self, chain: &ChainId, topic: &str) -> StreamResult<NewestOrArm> {
+        let mut chains = self.chains.lock().unwrap();
+        let chain = chains.entry(chain_key(chain)).or_default();
+        if let Some(held) = chain.topics.get(topic).filter(|t| !t.records.is_empty()) {
+            return Ok(NewestOrArm::Newest((held.head() - 1).to_string()));
+        }
+        Ok(NewestOrArm::Armed(chain.set_arm(topic)))
+    }
+
     async fn delete_owner(&self, request: DeleteOwnerRequest) -> StreamResult<DeleteOwnerResponse> {
         if request.owner_kind != StreamOwnerKind::Workflow as i32 {
             return Err(StreamError::unsupported(format!(

crates/streams/src/notify.rs

@@ -4,7 +4,7 @@
 //! notification is a hint that there is something new to read. The records stay on the read path.
 
 use crate::{
-    CloseOutcome, StreamResult, StreamStore, WORKFLOW_OWNER_KIND, mint_cursor,
+    CloseOutcome, NewestOrArm, StreamResult, StreamStore, WORKFLOW_OWNER_KIND, mint_cursor,
     proto::{
         ChainId, DeleteOwnerRequest, DeleteOwnerResponse, StoreAppendRequest, StoreAppendResponse,
         StoreLatestRequest, StoreLatestResponse, StoreReadRequest, StoreReadResponse,
@@ -831,6 +831,14 @@ impl StreamStore for NotifyingStore {
     async fn flush_notifications(&self) {
         self.notifier.flush().await;
     }
+
+    async fn disarm(&self, chain: &ChainId, topic: &str, generation: u64) -> StreamResult<()> {
+        self.inner.disarm(chain, topic, generation).await
+    }
+
+    async fn newest_or_arm(&self, chain: &ChainId, topic: &str) -> StreamResult<NewestOrArm> {
+        self.inner.newest_or_arm(chain, topic).await
+    }
 }
 
 #[cfg(test)]
@@ -1188,6 +1196,7 @@ mod tests {
             Ok(StoreAppendResponse {
                 first_position: "odd".to_string(),
                 last_position: "odd".to_string(),
+                arm: None,
             })
         }
         async fn read(&self, r: StoreReadRequest) -> StreamResult<StoreReadResponse> {

crates/streams/src/reader.rs

@@ -334,6 +334,7 @@ pub(crate) async fn read_with(
                 0 => DEFAULT_MAX_RECORDS,
                 max => max,
             },
+            arm: false,
         })
         .await?;
     let closed = page.closed;
@@ -493,7 +494,7 @@ mod tests {
                 .collect();
             Ok(StoreReadResponse {
                 records,
-                closed: false,
+                ..Default::default()
             })
         }
 

crates/streams/src/redis/mod.rs

@@ -465,6 +465,7 @@ impl StreamStore for RedisStore {
         Ok(StoreAppendResponse {
             first_position,
             last_position,
+            arm: None,
         })
     }
 
@@ -509,6 +510,7 @@ impl StreamStore for RedisStore {
         Ok(StoreReadResponse {
             records: found,
             closed: state.chain_closed || state.topic_closed,
+            armed: false,
         })
     }
 

crates/streams/src/store.rs

@@ -58,6 +58,7 @@ pub trait StreamStore: Send + Sync {
     /// positions and writes nothing. The same first sequence with another digest fails as
     /// producer divergent, and a first sequence below the end of the newest batch as producer
     /// stale. A closed chain or topic refuses a new batch as closed, but still answers a repeat.
+    /// The answer, a repeat's too, reports the topic's arm, so a lost notification goes again.
     async fn append(&self, request: StoreAppendRequest) -> StreamResult<StoreAppendResponse>;
 
     /// Reads the records of a topic after a position, waiting up to the request's `wait` for one
@@ -66,6 +67,10 @@ pub trait StreamStore: Send + Sync {
     /// Fails as expired when the position names a record retention dropped, or when retention
     /// dropped records the read had not delivered yet. Fails as not found when nothing is known
     /// about the topic and the position is not empty.
+    ///
+    /// A read that asks to arm and ends with no record on an open topic arms the topic in the
+    /// same step that finds it empty. The next append then reports the arm, and the producer
+    /// notifies the topic's callers.
     async fn read(&self, request: StoreReadRequest) -> StreamResult<StoreReadResponse>;
 
     /// The position of the newest record of a topic, or empty when it holds none.
@@ -167,6 +172,43 @@ 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) {}
+
+    /// Clears the arm of a topic when it still holds `generation`, once the notification that
+    /// generation asked for landed. A reader that armed again since holds a higher generation,
+    /// and its arm stays for the next append. A store that can't arm keeps the default.
+    async fn disarm(&self, chain: &ChainId, topic: &str, generation: u64) -> StreamResult<()> {
+        let _ = (chain, topic, generation);
+        Ok(())
+    }
+
+    /// The position of the newest record of a topic, or, when it holds none, an arm set in the
+    /// same step that finds it empty, so the first append reports the arm. A store that can't
+    /// arm keeps the default, which answers [NewestOrArm::Unarmed] for an empty topic.
+    async fn newest_or_arm(&self, chain: &ChainId, topic: &str) -> StreamResult<NewestOrArm> {
+        let position = self
+            .latest(StoreLatestRequest {
+                chain: Some(chain.clone()),
+                topic: topic.to_string(),
+            })
+            .await?
+            .position;
+        Ok(if position.is_empty() {
+            NewestOrArm::Unarmed
+        } else {
+            NewestOrArm::Newest(position)
+        })
+    }
+}
+
+/// What [StreamStore::newest_or_arm] found.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum NewestOrArm {
+    /// The position of the topic's newest record.
+    Newest(String),
+    /// The topic held no record, and the store armed it with this generation.
+    Armed(u64),
+    /// The topic held no record, and the store can't arm.
+    Unarmed,
 }
 
 #[cfg(test)]