From 6583597ed0461b89847366dc654649e847b0540e Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Mon, 12 Oct 2026 14:09:17 -0700 Subject: [PATCH] Notified a stream only while a reader armed it. An append or promotion notifies when the store reports an arm, or can't tell, and clears the arm once the notification lands. A failed send keeps the arm, so the next write notifies again. --- .../temporal/sdk/core/streams/streams.proto | 12 +- .../src/worker/workflow/stream_output.rs | 43 +- crates/streams/src/lib.rs | 2 +- crates/streams/src/notify.rs | 392 ++++++++++++++++-- crates/streams/src/service.rs | 27 ++ crates/streams/tests/conformance.rs | 27 +- crates/streams/tests/redis_store.rs | 9 +- 7 files changed, 455 insertions(+), 57 deletions(-) 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 57be55ecf..64fbb3f98 100644 --- a/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto +++ b/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; diff --git a/crates/sdk-core/src/worker/workflow/stream_output.rs b/crates/sdk-core/src/worker/workflow/stream_output.rs index 3c93e5d02..0a604b3b7 100644 --- a/crates/sdk-core/src/worker/workflow/stream_output.rs +++ b/crates/sdk-core/src/worker/workflow/stream_output.rs @@ -1603,11 +1603,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_accepted_completion_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(temporalio_streams::MemoryStore::new()), + memory, Arc::new(temporalio_streams::Notifier::new(recorder.clone(), 10)), )); let store = OutputStore::new(notifying.clone(), "ns".to_string(), Default::default()); @@ -1624,9 +1645,11 @@ mod tests { } /// A Worker's output store over memory that notifies `recorder`. - fn notifying_output(recorder: &Arc) -> OutputStore { + async fn notifying_output(recorder: &Arc) -> 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()) @@ -1659,7 +1682,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; let mut closed = batch("a"); closed.closes = vec![closing("t", "done")]; store.stage(std::slice::from_ref(&closed)).await.unwrap(); @@ -1739,11 +1762,13 @@ mod tests { #[tokio::test] async fn a_retried_promotion_still_ends_the_stream_on_the_server_after_its_output() { let recorder = Arc::new(Recorder::default()); + let inner = Arc::new(FailsFirstPromote( + temporalio_streams::MemoryStore::new(), + Mutex::new(true), + )); + armed(&inner.0).await; let notifying = Arc::new(temporalio_streams::NotifyingStore::new( - Arc::new(FailsFirstPromote( - temporalio_streams::MemoryStore::new(), - Mutex::new(true), - )), + inner, Arc::new(temporalio_streams::Notifier::new(recorder.clone(), 10)), )); let store = OutputStore::new(notifying, "ns".to_string(), Default::default()); @@ -1773,7 +1798,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.stage(&[batch("a")]).await.unwrap(); let mut replayed = proven("a"); replayed.closes = vec![closing("t", "done")]; diff --git a/crates/streams/src/lib.rs b/crates/streams/src/lib.rs index b802b2c6f..37ca5eed5 100644 --- a/crates/streams/src/lib.rs +++ b/crates/streams/src/lib.rs @@ -32,7 +32,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}; diff --git a/crates/streams/src/notify.rs b/crates/streams/src/notify.rs index d12a100ac..27c108caf 100644 --- a/crates/streams/src/notify.rs +++ b/crates/streams/src/notify.rs @@ -183,9 +183,42 @@ impl StreamKey { } } +/// An arm a notification answers. The slot clears it in the store once the notification lands, +/// so a send that fails for good leaves the arm for the next append to report. +#[derive(Clone)] +pub struct Arm { + /// The generation the store reported. + pub generation: u64, + /// The store that holds the arm. + pub store: Arc, +} + +impl std::fmt::Debug for Arm { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("Arm") + .field("generation", &self.generation) + .finish_non_exhaustive() + } +} + +/// The newer of two arms, which a disarm of the older would leave set. +fn newer(a: Option, b: Option) -> Option { + match (a, b) { + (Some(a), Some(b)) => Some(if b.generation > a.generation { b } else { a }), + (a, b) => a.or(b), + } +} + +#[derive(Debug)] +struct Pending { + position: String, + counter: i64, + arm: Option, +} + #[derive(Debug, Default)] struct SlotState { - pending: Option<(String, i64)>, + pending: Option, sending: bool, /// Positions still being looked up, which the slot sends once found. resolving: usize, @@ -211,15 +244,25 @@ impl Slot { /// Records the newest position and returns at once. While a call is out, later /// notifications fold into the one that goes next, so a burst costs at most two calls. - fn notify(self: &Arc, client: &Arc, position: String, counter: i64) { + fn notify( + self: &Arc, + client: &Arc, + position: String, + counter: i64, + arm: Option, + ) { let mut state = self.state.lock().unwrap(); - self.hold(&mut state, client, position, counter); + self.hold(&mut state, client, position, counter, arm); } /// Like [Slot::notify] with the position `lookup` finds, without waiting for it. The slot /// counts as busy until the lookup ends, so a close or a flush waits for it. - fn notify_found(self: &Arc, client: &Arc, lookup: F) - where + fn notify_found( + self: &Arc, + client: &Arc, + arm: Option, + lookup: F, + ) where F: Future> + Send + 'static, { { @@ -236,7 +279,7 @@ impl Slot { let mut state = slot.state.lock().unwrap(); state.resolving -= 1; if let Some((position, counter)) = found { - slot.hold(&mut state, &client, position, counter); + slot.hold(&mut state, &client, position, counter, arm); } slot.mark_idle_if_done(&state); }); @@ -248,16 +291,26 @@ impl Slot { client: &Arc, position: String, counter: i64, + arm: Option, ) { if state.closed { return; } - if state - .pending - .as_ref() - .is_none_or(|(_, held)| counter >= *held) - { - state.pending = Some((position, counter)); + match &mut state.pending { + Some(held) => { + if counter >= held.counter { + held.position = position; + held.counter = counter; + } + held.arm = newer(held.arm.take(), arm); + } + None => { + state.pending = Some(Pending { + position, + counter, + arm, + }) + } } if !state.sending { state.sending = true; @@ -274,7 +327,11 @@ impl Slot { async fn send_pending(self: Arc, client: Arc) { loop { - let (position, counter) = { + let Pending { + position, + counter, + arm, + } = { let mut state = self.state.lock().unwrap(); match state.pending.take() { Some(next) => next, @@ -292,17 +349,45 @@ impl Slot { counter, close: None, }; - self.send(&client, notification).await; + if self.send(&client, notification).await { + if let Some(arm) = arm { + self.disarm(arm).await; + } + } else if let Some(held) = self.state.lock().unwrap().pending.as_mut() { + // The newer notification answers this one's arm too. + held.arm = newer(held.arm.take(), arm); + } + } + } + + /// Clears the arm a landed notification answered, unless a reader armed again since. + async fn disarm(&self, arm: Arm) { + if let Err(error) = arm + .store + .disarm(&self.key.chain(), &self.key.topic, arm.generation) + .await + { + tracing::warn!( + workflow_id = %self.key.workflow_id, + topic = %self.key.topic, + %error, + "Could not clear a stream's arm, so the next write notifies again" + ); } } - async fn send(&self, client: &Arc, notification: StreamNotification) { + /// Sends one notification, trying a failed one again. Answers whether it landed. + async fn send( + &self, + client: &Arc, + notification: StreamNotification, + ) -> bool { let mut delay = NOTIFY_RETRY_FIRST; for tries_left in (0..=NOTIFY_RETRIES).rev() { // A notification is only a hint, and the reader reads from its own cursor, so a // notification that can't go never fails the write. match client.notify(notification.clone()).await { - Ok(()) => return, + Ok(()) => return true, Err(NotifyError::Failed(error)) if tries_left > 0 && !self.superseded() => { tracing::debug!( workflow_id = %self.key.workflow_id, @@ -318,15 +403,16 @@ impl Slot { %error, "A stream notification failed. The next write tells the reader again" ); - return; + return false; } } tokio::time::sleep(delay).await; delay *= 2; if self.superseded() { - return; + return false; } } + false } /// Whether something newer will reach the notifier anyway: a later notification, or the @@ -435,18 +521,31 @@ impl Notifier { /// Tells the notifier of `chain`'s stream on `topic` that it moved to `position`. Never /// waits, and never fails: a notification that can't go is logged and dropped. pub fn notify(&self, chain: &ChainId, topic: &str, position: String, counter: i64) { + self.notify_armed(chain, topic, position, counter, None); + } + + /// Like [Notifier::notify], for a write that found `arm` set. Once the notification lands, + /// the arm is cleared, unless a reader armed again since. + pub fn notify_armed( + &self, + chain: &ChainId, + topic: &str, + position: String, + counter: i64, + arm: Option, + ) { self.slot(StreamKey::new(chain, topic)) - .notify(&self.client, position, counter); + .notify(&self.client, position, counter, arm); } - /// Like [Notifier::notify] with the cursor and counter `lookup` finds, without waiting for - /// it. A close of the stream and [Notifier::flush] wait for the lookup. - pub fn notify_found(&self, chain: &ChainId, topic: &str, lookup: F) + /// Like [Notifier::notify_armed] with the cursor and counter `lookup` finds, without waiting + /// for it. A close of the stream and [Notifier::flush] wait for the lookup. + pub fn notify_found(&self, chain: &ChainId, topic: &str, arm: Option, lookup: F) where F: Future> + Send + 'static, { self.slot(StreamKey::new(chain, topic)) - .notify_found(&self.client, lookup); + .notify_found(&self.client, arm, lookup); } /// Closes the stream on the server with `outcome`, after the notification in flight, and @@ -594,7 +693,8 @@ impl NotifierClient for UnlessUnsupported { /// A store that closes each stream's notifier after the store closes the stream, and with /// progress on, notifies it after an append lands and after a Workflow's staged output becomes -/// visible. +/// visible, when the store reports that a reader armed the stream or can't tell. A landed +/// notification clears the arm it answered. /// /// Wraps the store a process hands to its stream calls and its Workers, so every write path /// notifies without knowing about notifiers. It forwards every trait method by name, since a @@ -633,20 +733,35 @@ impl NotifyingStore { close_notifier_in_background(&self.inner, &self.notifier, chain, topic, outcome); } - fn notify_at(&self, chain: &ChainId, topic: &str, position: &str) { + /// What a write that found `arm` owes: `None` when no reader armed the topic, so nothing is + /// 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) -> Option> { + match arm { + Some(0) => None, + Some(generation) => Some(Some(Arm { + generation, + store: self.inner.clone(), + })), + None => Some(None), + } + } + + fn notify_at(&self, chain: &ChainId, topic: &str, position: &str, arm: Option) { if let Some((cursor, counter)) = notification_at(self.inner.as_ref(), chain, topic, position) { - self.notifier.notify(chain, topic, cursor, counter); + self.notifier + .notify_armed(chain, topic, cursor, counter, arm); } } /// Notifies the newest record of `chain`'s stream on `topic` without waiting for the read, /// since a promotion sits on the Worker's report path and has no position of its own. - fn notify_latest(&self, chain: &ChainId, topic: &str) { + fn notify_latest(&self, chain: &ChainId, topic: &str, arm: Option) { let inner = self.inner.clone(); let (found_chain, found_topic) = (chain.clone(), topic.to_string()); - self.notifier.notify_found(chain, topic, async move { + self.notifier.notify_found(chain, topic, arm, async move { let (chain, topic) = (found_chain, found_topic); match inner .latest(StoreLatestRequest { @@ -716,8 +831,9 @@ impl StreamStore for NotifyingStore { let landed = self.inner.append(request).await?; if self.progress && let Some(chain) = &chain + && let Some(arm) = self.owed(landed.arm) { - self.notify_at(chain, &topic, &landed.last_position); + self.notify_at(chain, &topic, &landed.last_position, arm); } Ok(landed) } @@ -763,7 +879,9 @@ impl StreamStore for NotifyingStore { && let Some(chain) = &stage.chain { for topic in &stage.topics { - self.notify_latest(chain, topic); + if let Some(arm) = self.owed(result.arms.get(topic).copied()) { + self.notify_latest(chain, topic, arm); + } } } Ok(result) @@ -1140,7 +1258,7 @@ mod tests { notifier.notify(&chain("run-1"), "t", "c1".to_string(), 1); let slot = notifier.slot(StreamKey::new(&chain("run-1"), "t")); slot.close().await; - slot.notify(¬ifier.client, "c2".to_string(), 2); + slot.notify(¬ifier.client, "c2".to_string(), 2, None); notifier.flush().await; assert_eq!(recorder.sent(), vec![("c1".to_string(), 1)]); } @@ -1219,7 +1337,8 @@ mod tests { #[tokio::test] async fn an_append_notifies_the_chain_it_wrote_to() { let recorder = Arc::new(Recorder::default()); - let (_, store) = notifying_store(&recorder); + let (inner, store) = notifying_store(&recorder); + arm(inner.as_ref(), &chain("run-1"), "t").await; let landed = store .append(append_request(&chain("run-1"), "t", 1)) .await @@ -1332,8 +1451,10 @@ mod tests { #[tokio::test] async fn a_workflows_batch_notifies_each_of_its_topics_once_visible() { let recorder = Arc::new(Recorder::default()); - let (_, store) = notifying_store(&recorder); + let (inner, store) = notifying_store(&recorder); let run = chain("run-1"); + arm(inner.as_ref(), &run, "a").await; + arm(inner.as_ref(), &run, "b").await; store .stage(staged(&run, "token", &["a", "b", "a"])) .await @@ -1423,6 +1544,8 @@ mod tests { let store = NotifyingStore::new(inner.clone(), Arc::new(Notifier::new(recorder.clone(), 10))); let run = chain("run-1"); + arm(&inner.0, &run, "a").await; + arm(&inner.0, &run, "b").await; store .stage(staged(&run, "token", &["a", "b"])) .await @@ -1443,8 +1566,10 @@ mod tests { #[tokio::test] async fn a_closed_streams_notifier_is_dropped() { let recorder = Arc::new(Recorder::default()); - let (_, store) = notifying_store(&recorder); + let (inner, store) = notifying_store(&recorder); let run = chain("run-1"); + arm(inner.as_ref(), &run, "a").await; + arm(inner.as_ref(), &run, "b").await; store.append(append_request(&run, "a", 1)).await.unwrap(); store.append(append_request(&run, "b", 1)).await.unwrap(); store.close_topic(&run, "a", None).await.unwrap(); @@ -1471,7 +1596,8 @@ mod tests { #[tokio::test] async fn an_ended_chain_closes_only_its_own_streams() { let recorder = Arc::new(Recorder::default()); - let (_, store) = notifying_store(&recorder); + let (inner, store) = notifying_store(&recorder); + arm(inner.as_ref(), &chain("run-9"), "a").await; store .append(append_request(&chain("run-1"), "a", 1)) .await @@ -1867,4 +1993,200 @@ mod tests { store.flush_notifications().await; assert_eq!(client.at.lock().unwrap().len(), 2, "the retry lands"); } + + /// Arms `topic` as a reader that found nothing after the newest record does. + async fn arm(store: &dyn StreamStore, chain: &ChainId, topic: &str) { + let after = store + .latest(StoreLatestRequest { + chain: Some(chain.clone()), + topic: topic.to_string(), + }) + .await + .unwrap() + .position; + let read = store + .read(StoreReadRequest { + chain: Some(chain.clone()), + topic: topic.to_string(), + after_position: after, + arm: true, + ..Default::default() + }) + .await + .unwrap(); + assert!(read.armed); + } + + /// The arm an append straight to `store` reports, which sends no notification. + async fn arm_now(store: &MemoryStore, chain: &ChainId, topic: &str, sequence: i64) -> u64 { + let mut request = append_request(chain, topic, sequence); + request.producer_id = "probe".to_string(); + store.append(request).await.unwrap().arm.unwrap() + } + + #[tokio::test] + async fn an_append_with_nothing_armed_sends_nothing() { + let recorder = Arc::new(Recorder::default()); + let (_, store) = notifying_store(&recorder); + store + .append(append_request(&chain("run-1"), "t", 1)) + .await + .unwrap(); + store.flush_notifications().await; + assert!(recorder.sent().is_empty(), "no reader sleeps on the stream"); + assert!(store.notifier().slots.lock().unwrap().is_empty()); + } + + #[tokio::test] + async fn a_send_disarms_only_after_it_lands() { + let recorder = Recorder::holding(); + let (inner, store) = notifying_store(&recorder); + let run = chain("run-1"); + arm(inner.as_ref(), &run, "t").await; + store.append(append_request(&run, "t", 1)).await.unwrap(); + settle().await; + assert_eq!(recorder.sent().len(), 1, "the notification is in flight"); + let armed = arm_now(&inner, &run, "t", 1).await; + assert!(armed > 0, "the arm stays while the notification is out"); + recorder.release(1); + store.flush_notifications().await; + assert_eq!( + arm_now(&inner, &run, "t", 2).await, + 0, + "cleared once it landed" + ); + } + + #[tokio::test] + async fn a_disarm_keeps_an_arm_a_reader_set_while_the_send_was_out() { + let recorder = Recorder::holding(); + let (inner, store) = notifying_store(&recorder); + let run = chain("run-1"); + arm(inner.as_ref(), &run, "t").await; + store.append(append_request(&run, "t", 1)).await.unwrap(); + settle().await; + // The reader woke from another notification, read everything and armed again. + arm(inner.as_ref(), &run, "t").await; + recorder.release(1); + store.flush_notifications().await; + assert!(arm_now(&inner, &run, "t", 1).await > 0); + } + + #[tokio::test(start_paused = true)] + async fn a_failed_send_keeps_the_arm_for_the_next_append() { + let recorder = Arc::new(Recorder::default()); + recorder.fail.store(true, Ordering::SeqCst); + let (inner, store) = notifying_store(&recorder); + let run = chain("run-1"); + arm(inner.as_ref(), &run, "t").await; + store.append(append_request(&run, "t", 1)).await.unwrap(); + store.flush_notifications().await; + assert_eq!(recorder.sent().len(), 3, "three tries, then it's dropped"); + recorder.fail.store(false, Ordering::SeqCst); + store.append(append_request(&run, "t", 2)).await.unwrap(); + store.flush_notifications().await; + assert_eq!(recorder.sent().len(), 4, "the next append notifies again"); + assert_eq!(arm_now(&inner, &run, "t", 1).await, 0); + } + + #[tokio::test] + async fn promotions_notify_only_armed_topics() { + let recorder = Arc::new(Recorder::default()); + let (inner, store) = notifying_store(&recorder); + let run = chain("run-1"); + arm(inner.as_ref(), &run, "a").await; + store + .stage(staged(&run, "token", &["a", "b"])) + .await + .unwrap(); + store + .promote(&stage_ref(&run, "token", &["a", "b"])) + .await + .unwrap(); + store.flush_notifications().await; + let topics: Vec = recorder + .sent + .lock() + .unwrap() + .iter() + .map(|n| n.topic.clone()) + .collect(); + assert_eq!(topics, ["a"]); + assert_eq!(arm_now(&inner, &run, "a", 1).await, 0); + } + + /// A memory store that reports no arm, as a store that can't arm does. + struct CantTell(MemoryStore); + + #[async_trait::async_trait] + impl StreamStore for CantTell { + fn name(&self) -> &str { + self.0.name() + } + async fn append(&self, r: StoreAppendRequest) -> StreamResult { + let landed = self.0.append(r).await?; + Ok(StoreAppendResponse { + arm: None, + ..landed + }) + } + async fn read(&self, r: StoreReadRequest) -> StreamResult { + self.0.read(r).await + } + async fn latest(&self, r: StoreLatestRequest) -> StreamResult { + self.0.latest(r).await + } + async fn record_at(&self, c: &ChainId, t: &str, p: &str) -> StreamResult>> { + self.0.record_at(c, t, p).await + } + async fn stage(&self, b: StagedBatch) -> StreamResult<()> { + self.0.stage(b).await + } + async fn promote(&self, s: &StageRef) -> StreamResult { + let promoted = self.0.promote(s).await?; + Ok(PromoteResult { + arms: Default::default(), + ..promoted + }) + } + async fn abort(&self, s: &StageRef) -> StreamResult<()> { + self.0.abort(s).await + } + async fn close_chain(&self, c: &ChainId) -> StreamResult<()> { + self.0.close_chain(c).await + } + async fn close_topic(&self, c: &ChainId, t: &str, r: Option) -> StreamResult<()> { + self.0.close_topic(c, t, r).await + } + async fn pending_stages(&self, c: &ChainId) -> StreamResult> { + self.0.pending_stages(c).await + } + async fn delete_owner(&self, r: DeleteOwnerRequest) -> StreamResult { + self.0.delete_owner(r).await + } + } + + #[tokio::test] + async fn a_store_that_cant_tell_still_notifies() { + let recorder = Arc::new(Recorder::default()); + let notifier = Arc::new(Notifier::new(recorder.clone(), 10)); + let store = NotifyingStore::new(Arc::new(CantTell(MemoryStore::new())), notifier); + let run = chain("run-1"); + store.append(append_request(&run, "t", 1)).await.unwrap(); + store.stage(staged(&run, "token", &["b"])).await.unwrap(); + store + .promote(&stage_ref(&run, "token", &["b"])) + .await + .unwrap(); + store.flush_notifications().await; + let mut topics: Vec = recorder + .sent + .lock() + .unwrap() + .iter() + .map(|n| n.topic.clone()) + .collect(); + topics.sort(); + assert_eq!(topics, ["b", "t"]); + } } diff --git a/crates/streams/src/service.rs b/crates/streams/src/service.rs index 6c0cae801..8a02ca838 100644 --- a/crates/streams/src/service.rs +++ b/crates/streams/src/service.rs @@ -588,10 +588,35 @@ mod tests { FlushNotificationsResponse::decode(response.as_slice()).unwrap(); } + /// Reads `topic` with `arm`, as a reader that sleeps until its caller hears of progress. + async fn armed_read(service: &StreamService, topic: &str) { + let read = ReadResponse::decode( + service + .call( + "Read", + &ReadRequest { + stream: Some(StreamAddress { + topic: topic.to_string(), + ..address() + }), + arm: true, + ..Default::default() + } + .encode_to_vec(), + ) + .await + .unwrap() + .as_slice(), + ) + .unwrap(); + assert!(read.armed); + } + #[tokio::test] async fn an_append_notifies_its_streams_notifier_once_notifications_are_on() { let recorder = Arc::new(Recorder::default()); let service = service_notifying(true, recorder.clone()).await; + armed_read(&service, "out").await; let appended = AppendResponse::decode( service .call("Append", &append(1, &["a", "b"])) @@ -812,6 +837,7 @@ mod tests { async fn a_close_with_a_result_closes_the_store_and_then_the_notifier() { let recorder = Arc::new(Recorder::default()); let service = service_notifying(true, recorder.clone()).await; + armed_read(&service, "out").await; service .call("Append", &append(1, &["a", "b"])) .await @@ -966,6 +992,7 @@ mod tests { use crate::proto::{NexusCall, NexusStep, nexus_step}; let recorder = Arc::new(Recorder::default()); let service = service_notifying(true, recorder.clone()).await; + armed_read(&service, "t").await; let input = serde_json::json!({ "stream": {"kind": "workflow", "workflow_id": "wf", "topic": "t"}, "producer_id": "p", diff --git a/crates/streams/tests/conformance.rs b/crates/streams/tests/conformance.rs index fce1c997b..498f772f1 100644 --- a/crates/streams/tests/conformance.rs +++ b/crates/streams/tests/conformance.rs @@ -94,12 +94,27 @@ struct Notifying { async fn notifying( inner: impl std::future::Future>, +) -> Option> { + notifying_with(inner, true).await +} + +/// [notifying] with progress off, for the arm cases, since with progress on a landed +/// notification clears the arm the case checks. +async fn notifying_quietly( + inner: impl std::future::Future>, +) -> Option> { + notifying_with(inner, false).await +} + +async fn notifying_with( + inner: impl std::future::Future>, + progress: bool, ) -> Option> { let inner = inner.await?; - let store = Arc::new(NotifyingStore::new( - inner.store(), - Arc::new(Notifier::new(Arc::new(Quiet), 10)), - )); + let store = Arc::new( + NotifyingStore::new(inner.store(), Arc::new(Notifier::new(Arc::new(Quiet), 10))) + .with_progress(progress), + ); Some(Notifying { inner, store }) } @@ -1862,7 +1877,7 @@ mod memory { mod notifying_memory { store_cases!(super::notifying(super::memory())); - arm_cases!(super::notifying(super::memory())); + arm_cases!(super::notifying_quietly(super::memory())); conformance!(super::notifying(super::memory()); deleting_an_owner_drops_every_chain_of_it, ); @@ -1893,7 +1908,7 @@ mod notifying_redis { "STREAMS_REDIS_URL", false ))); - arm_cases!(super::notifying(super::on_redis::redis( + arm_cases!(super::notifying_quietly(super::on_redis::redis( "STREAMS_REDIS_URL", false ))); diff --git a/crates/streams/tests/redis_store.rs b/crates/streams/tests/redis_store.rs index 76a00e3f7..5a14bd0b4 100644 --- a/crates/streams/tests/redis_store.rs +++ b/crates/streams/tests/redis_store.rs @@ -1811,8 +1811,14 @@ async fn notification_counters_come_from_redis_entry_ids_and_grow_with_the_strea Arc::new(temporalio_streams::Notifier::new(recorder.clone(), 10)), ); let chain = chain(); - let mut positions = vec![]; + let mut positions: Vec = vec![]; + // A reader sleeps before each write, and each landed notification clears its arm. + let rearm = async |after: &str| { + let read = store.read(arming_read(&chain, after)).await.unwrap(); + assert!(read.armed); + }; for sequence in [1, 3] { + rearm(positions.last().map_or("", String::as_str)).await; let digest = [sequence as u8]; let landed = store .append(append(&chain, "p", sequence, 2, &digest)) @@ -1821,6 +1827,7 @@ async fn notification_counters_come_from_redis_entry_ids_and_grow_with_the_strea positions.push(landed.last_position); store.flush_notifications().await; } + rearm(&positions[1]).await; store .stage(StagedBatch { chain: Some(chain.clone()),