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 e0bfeb85e2..ab48869af7 100644 --- a/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto +++ b/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. diff --git a/crates/sdk-core-c-bridge/src/tests/mod.rs b/crates/sdk-core-c-bridge/src/tests/mod.rs index ba49907a18..95c3cabf12 100644 --- a/crates/sdk-core-c-bridge/src/tests/mod.rs +++ b/crates/sdk-core-c-bridge/src/tests/mod.rs @@ -508,6 +508,7 @@ fn memory_store_config() -> Vec { }; StreamStoreConfig { store: Some(stream_store_config::Store::Memory(MemoryStoreConfig {})), + ..Default::default() } .encode_to_vec() } diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index b968e32a50..63fb768661 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/crates/sdk-core/CHANGELOG.md @@ -56,6 +56,10 @@ relevant information. * Experimental: a read reports `SUPERSEDED` for a Workflow producer (`@`) 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 diff --git a/crates/streams/src/handle.rs b/crates/streams/src/handle.rs index d316fecb6f..557aea3634 100644 --- a/crates/streams/src/handle.rs +++ b/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 { 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> { + 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 { 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,6 +565,16 @@ 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(), @@ -519,15 +582,10 @@ impl Streams { }; 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] diff --git a/crates/streams/src/lib.rs b/crates/streams/src/lib.rs index 5bca4c6ccc..a945a23eec 100644 --- a/crates/streams/src/lib.rs +++ b/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::{ diff --git a/crates/streams/src/reader.rs b/crates/streams/src/reader.rs index f07a759f6c..ee00b6cb24 100644 --- a/crates/streams/src/reader.rs +++ b/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. diff --git a/crates/streams/src/service.rs b/crates/streams/src/service.rs index b67e6e15e3..55ffdfe23f 100644 --- a/crates/streams/src/service.rs +++ b/crates/streams/src/service.rs @@ -52,11 +52,13 @@ impl StreamService { )); } }; - Ok(Self::new( - store, - owner, - StreamsOptions::for_retention(retention), - )) + let mut options = StreamsOptions::for_retention(retention); + if let Some(grace) = config.owner_start_grace { + options.owner_start_grace = Duration::try_from(grace).map_err(|_| { + StreamError::refused("the stream store's owner start grace must not be negative") + })?; + } + Ok(Self::new(store, owner, options)) } /// Serves calls on `store`. @@ -216,6 +218,7 @@ mod tests { StreamService::connect( StreamStoreConfig { store: Some(stream_store_config::Store::Memory(MemoryStoreConfig {})), + ..Default::default() }, Arc::new(Running), ) @@ -453,10 +456,24 @@ mod tests { #[tokio::test] async fn a_config_that_names_no_store_is_refused() { - let error = StreamService::connect(StreamStoreConfig { store: None }, Arc::new(Running)) + let error = StreamService::connect(StreamStoreConfig::default(), Arc::new(Running)) + .await + .unwrap_err(); + assert_eq!(error.kind, StreamFailureKind::Refused); + } + + #[tokio::test] + async fn a_negative_owner_start_grace_is_refused() { + let mut config = StreamStoreConfig { + store: Some(stream_store_config::Store::Memory(MemoryStoreConfig {})), + owner_start_grace: Some(Default::default()), + }; + config.owner_start_grace.as_mut().unwrap().seconds = -1; + let error = StreamService::connect(config, Arc::new(Running)) .await .unwrap_err(); assert_eq!(error.kind, StreamFailureKind::Refused); + assert!(error.message.contains("owner start grace"), "{error}"); } #[cfg(not(feature = "redis"))] @@ -465,6 +482,7 @@ mod tests { let error = StreamService::connect( StreamStoreConfig { store: Some(stream_store_config::Store::Redis(Default::default())), + ..Default::default() }, Arc::new(Running), )