[Brook] Gave a reader a grace before its owner starts.

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

@@ -36,6 +36,11 @@ message StreamStoreConfig {
     RedisStoreConfig redis = 1;
     MemoryStoreConfig memory = 2;
   }
+  // 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
+  // once.
+  google.protobuf.Duration owner_start_grace = 6;
 }
 
 // Streams in a Redis or Redis Cluster that Core connects to itself.

crates/sdk-core-c-bridge/src/tests/mod.rs

@@ -508,6 +508,7 @@ fn memory_store_config() -> Vec<u8> {
     };
     StreamStoreConfig {
         store: Some(stream_store_config::Store::Memory(MemoryStoreConfig {})),
+        ..Default::default()
     }
     .encode_to_vec()
 }

crates/sdk-core/CHANGELOG.md

@@ -56,6 +56,10 @@ relevant information.
 * Experimental: a read reports `SUPERSEDED` for a Workflow producer (`@<run id>`) only when its
   attempt keeps the high bits (the same task) and raises the low 16 bits (a rerun). A higher
   prefix is the next task and reads with no supersession.
+* Experimental: a read of a Workflow that Temporal doesn't know yet answers empty for
+  `StreamStoreConfig.owner_start_grace` (60 seconds unless set, counted from the read's first
+  owner check), so a reader can start before its owner. After the grace it fails as `NOT_FOUND`
+  as before. Zero turns the grace off.
 * Experimental: a Workflow Task append that the store refuses as closed drops that topic's
   records, logs a warning, counts `workflow_stream_output_dropped`, and the task goes on. A
   retry would be refused the same way, also when the run closed the topic itself before its task

crates/streams/src/handle.rs

@@ -4,15 +4,15 @@
 //! where the contract says so, and mints and checks cursors. Stores only move records.
 
 use crate::{
-    ReadTarget, StreamError, StreamResult, StreamStore, WORKFLOW_OWNER_KIND, activity_producer_id,
-    check_append_digest, mint_cursor,
+    BEGINNING, END, ReadTarget, StreamError, StreamResult, StreamStore, WORKFLOW_OWNER_KIND,
+    activity_producer_id, check_append_digest, mint_cursor,
     owner::{OwnerClient, OwnerDescription, OwnerError},
     proto::{
         ActivityProducer, AppendRequest, AppendResponse, ChainId, CloseRequest, CloseResponse,
         DeleteOwnerRequest, DeleteOwnerResponse, LatestRequest, LatestResponse, NamedProducer,
         ReadRequest, ReadResponse, RetireProducerRequest, RetireProducerResponse,
-        StoreAppendRequest, StoreLatestRequest, StreamAddress, StreamOwnerKind, append_request,
-        retire_producer_request,
+        StoreAppendRequest, StoreLatestRequest, StreamAddress, StreamFailureKind, StreamOwnerKind,
+        append_request, retire_producer_request,
     },
     reader::{ReadState, read_with},
     stored_append_record, stream_hash,
@@ -29,6 +29,9 @@ use tokio::time::Instant;
 /// The longest a producer writes without asking whether its owner's chain ended.
 pub const OWNER_RECHECK: Duration = Duration::from_secs(60);
 
+/// How long a read waits by default for an owner that Temporal doesn't know yet.
+pub const OWNER_START_GRACE: Duration = Duration::from_secs(60);
+
 /// How the stream layer behaves.
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct StreamsOptions {
@@ -41,6 +44,9 @@ pub struct StreamsOptions {
     pub owner_check_min: Duration,
     /// The longest an idle read waits between two such questions.
     pub owner_check_max: Duration,
+    /// How long after its first owner check a read takes a Workflow that Temporal doesn't know
+    /// as one that has not started yet, and answers empty. Zero fails the read at once.
+    pub owner_start_grace: Duration,
 }
 
 impl Default for StreamsOptions {
@@ -50,6 +56,7 @@ impl Default for StreamsOptions {
             remembered: 10_000,
             owner_check_min: Duration::from_millis(500),
             owner_check_max: Duration::from_secs(5),
+            owner_start_grace: OWNER_START_GRACE,
         }
     }
 }
@@ -248,23 +255,37 @@ impl Streams {
     /// the interval over. So many readers parked on a quiet stream don't load the server. Once
     /// the owner ended, or the store marked the stream closed, the read delivers what is left
     /// and answers `done`. A read pinned to a run ends when that run closes, even if the chain
-    /// continued as new.
+    /// continued as new. A read whose owner Temporal doesn't know yet answers empty within
+    /// [StreamsOptions::owner_start_grace], so it can start before its owner does.
     pub async fn read(&self, request: ReadRequest) -> StreamResult<ReadResponse> {
         let stream = address(request.stream.as_ref())?;
         let mut state = ReadState::decode_from(&request.state)?;
+        let wait = request
+            .wait
+            .and_then(|wait| Duration::try_from(wait).ok())
+            .unwrap_or_default();
+        let deadline = Instant::now() + wait;
         if state.first_run_id.is_empty() {
-            state.first_run_id = self.chain(stream).await?.first_run_id;
+            let Some(chain) = self.started_chain(stream, &mut state, deadline).await? else {
+                // The owner has not started, so its stream holds nothing and its end is its start.
+                let cursor = if request.after == END {
+                    BEGINNING.to_string()
+                } else {
+                    request.after.clone()
+                };
+                return Ok(ReadResponse {
+                    cursor,
+                    state: state.encode_to_vec(),
+                    ..Default::default()
+                });
+            };
+            state.first_run_id = chain.first_run_id;
         }
         let target = ReadTarget {
             chain: chain_of(stream, &state.first_run_id),
             topic: stream.topic.clone(),
             stream_hash: self.hash(stream),
         };
-        let wait = request
-            .wait
-            .and_then(|wait| Duration::try_from(wait).ok())
-            .unwrap_or_default();
-        let deadline = Instant::now() + wait;
         let minimum = self.options.owner_check_min;
         let mut page_request = request.clone();
         let response = loop {
@@ -319,19 +340,51 @@ impl Streams {
         })
     }
 
+    /// The chain a read follows, or `None` when Temporal doesn't know its owner yet, within
+    /// [StreamsOptions::owner_start_grace] of the read's first owner check. It asks again as
+    /// an idle read does until `deadline`. After the grace the read fails as not found.
+    async fn started_chain(
+        &self,
+        stream: &StreamAddress,
+        state: &mut ReadState,
+        deadline: Instant,
+    ) -> StreamResult<Option<ChainId>> {
+        let mut interval = self.options.owner_check_min;
+        loop {
+            let error = match self.chain(stream).await {
+                Ok(chain) => {
+                    state.owner_missing_since_ms = 0;
+                    return Ok(Some(chain));
+                }
+                Err(error) if error.kind == StreamFailureKind::NotFound => error,
+                Err(error) => return Err(error),
+            };
+            let now = unix_ms();
+            if state.owner_missing_since_ms == 0 {
+                state.owner_missing_since_ms = now;
+            }
+            let missing_for = now.saturating_sub(state.owner_missing_since_ms);
+            if missing_for >= millis(self.options.owner_start_grace) {
+                return Err(error);
+            }
+            let remaining = deadline.saturating_duration_since(Instant::now());
+            if remaining.is_zero() {
+                return Ok(None);
+            }
+            tokio::time::sleep(interval.min(remaining)).await;
+            interval = (interval * 2).min(self.options.owner_check_max);
+        }
+    }
+
     /// Whether the owner a read follows ended. Following the chain, a run that continued as
     /// new is not the end, and a new chain on the Workflow id is.
     async fn owner_ended(&self, stream: &StreamAddress, chain: &ChainId) -> StreamResult<bool> {
         let key = owner_key(stream);
         match self.describe(&key, &key.run_id).await {
-            Err(OwnerError::NotFound(_)) => {
-                // The chain came from this Workflow, so History has since dropped it, and
-                // nothing more will be written for it.
-                if key.run_id.is_empty() {
-                    self.store.close_chain(chain).await?;
-                }
-                Ok(true)
-            }
+            // The chain came from this Workflow, so History has since dropped it, or a failover
+            // hides it for a while. A reader only ends its own read, and producers close the
+            // chain.
+            Err(OwnerError::NotFound(_)) => Ok(self.gone_for_good(chain)),
             Err(error) => Err(describe_failed(&error)),
             Ok(described) if !key.run_id.is_empty() => {
                 Ok(described.status != WorkflowExecutionStatus::Running)
@@ -512,22 +565,27 @@ impl Streams {
     /// at least [StreamsOptions::owner_recheck] after the first. The first answer fails the call
     /// as a storage failure, which the caller retries.
     async fn gone(&self, chain: &ChainId) -> StreamError {
+        if self.gone_for_good(chain) {
+            self.closed(chain, "is gone from History").await
+        } else {
+            missing(chain)
+        }
+    }
+
+    /// Whether Temporal answered that `chain`'s owner is gone twice, at least
+    /// [StreamsOptions::owner_recheck] apart, with no answer that holds it in between.
+    fn gone_for_good(&self, chain: &ChainId) -> bool {
         let key = OwnerKey {
             namespace: chain.namespace.clone(),
             workflow_id: chain.workflow_id.clone(),
             run_id: String::new(),
         };
         let recheck = self.options.owner_recheck;
         match self.missing.get(&key) {
-            Some(seen)
-                if seen.first_run_id == chain.first_run_id && seen.at.elapsed() >= recheck =>
-            {
-                self.closed(chain, "is gone from History").await
-            }
-            Some(seen) if seen.first_run_id == chain.first_run_id => missing(chain),
+            Some(seen) if seen.first_run_id == chain.first_run_id => seen.at.elapsed() >= recheck,
             _ => {
                 self.missing.put(key, &chain.first_run_id, recheck * 2);
-                missing(chain)
+                false
             }
         }
     }
@@ -1436,6 +1494,100 @@ mod tests {
         }
     }
 
+    /// A setup whose Workflow `wf` has not started yet, with an owner start grace of `grace`.
+    fn before_owner_starts(grace: Duration) -> Setup {
+        let store = Arc::new(MemoryStore::new());
+        let owners = Arc::new(Owners::default());
+        let options = StreamsOptions {
+            owner_start_grace: grace,
+            ..quick()
+        };
+        let streams = Streams::new(store.clone(), owners.clone(), options);
+        Setup {
+            store,
+            owners,
+            streams,
+        }
+    }
+
+    #[tokio::test]
+    async fn a_read_before_its_owner_starts_answers_empty_until_the_owner_writes() {
+        let setup = before_owner_starts(Duration::from_secs(60));
+        let empty = setup
+            .streams
+            .read(read_after(END, &[], Duration::ZERO))
+            .await
+            .unwrap();
+        assert!(empty.records.is_empty() && !empty.done);
+        assert_eq!(
+            empty.cursor, BEGINNING,
+            "the end of a stream that holds nothing is its start"
+        );
+        setup.owners.run("wf", "run-1", "run-1");
+        setup.streams.append(append("p", 1, "1")).await.unwrap();
+        let read = setup
+            .streams
+            .read(read_after(&empty.cursor, &empty.state, Duration::ZERO))
+            .await
+            .unwrap();
+        assert_eq!(read.records.len(), 1, "the first record is not missed");
+    }
+
+    #[tokio::test]
+    async fn a_waiting_read_gets_the_first_record_of_an_owner_that_starts_later() {
+        let setup = before_owner_starts(Duration::from_secs(60));
+        let owners = setup.owners.clone();
+        let streams = Arc::new(setup.streams);
+        let writer = streams.clone();
+        let late_start = tokio::spawn(async move {
+            tokio::time::sleep(Duration::from_millis(300)).await;
+            owners.run("wf", "run-1", "run-1");
+            writer.append(append("p", 1, "1")).await.unwrap();
+        });
+        let read = streams
+            .read(read_after("", &[], Duration::from_secs(5)))
+            .await
+            .unwrap();
+        late_start.await.unwrap();
+        assert_eq!(read.records.len(), 1);
+    }
+
+    #[tokio::test]
+    async fn a_read_of_an_owner_that_never_starts_is_not_found_after_the_grace() {
+        let setup = before_owner_starts(Duration::from_millis(200));
+        let empty = setup
+            .streams
+            .read(read_after("", &[], Duration::from_millis(50)))
+            .await
+            .unwrap();
+        assert!(empty.records.is_empty());
+        let error = setup
+            .streams
+            .read(read_after("", &empty.state, Duration::from_secs(5)))
+            .await
+            .unwrap_err();
+        assert_eq!(error.kind, StreamFailureKind::NotFound);
+        // The grace counts from the read's first owner check, so a new read waits again.
+        assert!(
+            setup
+                .streams
+                .read(read_after("", &[], Duration::ZERO))
+                .await
+                .is_ok()
+        );
+    }
+
+    #[tokio::test]
+    async fn no_grace_fails_a_read_of_a_missing_owner_at_once() {
+        let setup = before_owner_starts(Duration::ZERO);
+        let error = setup
+            .streams
+            .read(read_after("", &[], Duration::from_secs(5)))
+            .await
+            .unwrap_err();
+        assert_eq!(error.kind, StreamFailureKind::NotFound);
+    }
+
     /// Reads until the read says it is done, or fails after `limit` calls.
     async fn read_to_end(streams: &Streams, wait: Duration) -> (usize, ReadResponse) {
         let mut delivered = 0;
@@ -1547,7 +1699,10 @@ mod tests {
 
     #[tokio::test]
     async fn rule_20_3_a_read_ends_when_history_no_longer_holds_its_owner() {
-        let setup = setup(quick());
+        let setup = setup(StreamsOptions {
+            owner_recheck: Duration::from_millis(200),
+            ..quick()
+        });
         setup.streams.append(append("p", 1, "1")).await.unwrap();
         let first = setup
             .streams
@@ -1567,7 +1722,45 @@ mod tests {
             .await
             .unwrap();
         assert!(page.done);
-        assert!(chain_closed(&setup.store, "run-1").await);
+        // The reader's check ends its own read only. A producer's check closes the chain.
+        assert!(!chain_closed(&setup.store, "run-1").await);
+    }
+
+    #[tokio::test]
+    async fn a_read_goes_on_after_one_not_found_from_its_owner_check() {
+        let setup = setup(quick());
+        setup.streams.append(append("p", 1, "1")).await.unwrap();
+        let first = setup
+            .streams
+            .read(read_after("", &[], Duration::ZERO))
+            .await
+            .unwrap();
+        setup
+            .owners
+            .fail("wf", OwnerError::NotFound("workflow not found".to_string()));
+        let page = setup
+            .streams
+            .read(read_after(
+                &first.cursor,
+                &first.state,
+                Duration::from_millis(300),
+            ))
+            .await
+            .unwrap();
+        assert!(!page.done);
+        setup.owners.run("wf", "run-1", "run-1");
+        setup.streams.append(append("p", 2, "2")).await.unwrap();
+        let page = setup
+            .streams
+            .read(read_after(
+                &page.cursor,
+                &page.state,
+                Duration::from_secs(5),
+            ))
+            .await
+            .unwrap();
+        assert_eq!(page.records.len(), 1);
+        assert!(!chain_closed(&setup.store, "run-1").await);
     }
 
     #[tokio::test]

crates/streams/src/lib.rs

@@ -24,7 +24,7 @@ mod store;
 pub use self::redis::{RedisStore, RedisStoreOptions};
 pub use cursor::{BEGINNING, END, cursor_position, mint_cursor, stream_hash};
 pub use error::{StreamError, StreamResult};
-pub use handle::{OWNER_RECHECK, Streams, StreamsOptions};
+pub use handle::{OWNER_RECHECK, OWNER_START_GRACE, Streams, StreamsOptions};
 pub use memory::MemoryStore;
 pub use owner::{OwnerClient, OwnerDescription, OwnerError};
 pub use reader::{

crates/streams/src/reader.rs

@@ -96,6 +96,10 @@ pub(crate) struct ReadState {
     watermark_known: bool,
     #[prost(uint64, tag = "10")]
     clock: u64,
+    /// When the read first found no owner, in milliseconds since the Unix epoch, or 0 once it
+    /// found one. The owner start grace counts from it.
+    #[prost(uint64, tag = "11")]
+    pub(crate) owner_missing_since_ms: u64,
 }
 
 /// What a read knows of one producer.