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 ae6ae53b8..6c1864622 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 2b663c852..a333c8914 100644 --- a/crates/sdk-core/src/worker/workflow/stream_output.rs +++ b/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) -> 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()) @@ -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( diff --git a/crates/streams/src/lib.rs b/crates/streams/src/lib.rs index 5f63564b7..903a43f07 100644 --- a/crates/streams/src/lib.rs +++ b/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}; diff --git a/crates/streams/src/notify.rs b/crates/streams/src/notify.rs index 7f8b30baf..ecb4352c4 100644 --- a/crates/streams/src/notify.rs +++ b/crates/streams/src/notify.rs @@ -182,9 +182,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, @@ -210,15 +243,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, { { @@ -235,7 +278,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); }); @@ -247,16 +290,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; @@ -273,7 +326,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, @@ -291,17 +348,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); + } } } - async fn send(&self, client: &Arc, notification: StreamNotification) { + /// 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" + ); + } + } + + /// 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, @@ -317,15 +402,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 @@ -434,18 +520,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 @@ -592,7 +691,9 @@ 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, a Workflow's output included. +/// progress on, notifies it after an append lands, a Workflow's output included, 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 @@ -631,11 +732,26 @@ 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); } } } @@ -683,8 +799,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) } @@ -1079,7 +1196,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)]); } @@ -1158,7 +1275,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 @@ -1235,8 +1353,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(); @@ -1263,7 +1383,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 @@ -1659,4 +1780,181 @@ 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 a_workflows_appends_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; + for topic in ["a", "b"] { + store + .append(StoreAppendRequest { + producer_id: crate::workflow_producer_id("run-1"), + ..append_request(&run, topic, 1) + }) + .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 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 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.append(append_request(&run, "b", 1)).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 7b8c066d5..f0491161c 100644 --- a/crates/streams/src/service.rs +++ b/crates/streams/src/service.rs @@ -577,10 +577,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"])) @@ -790,6 +815,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 @@ -926,6 +952,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 148244a95..791a72a61 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 }) } @@ -1489,7 +1504,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, ); @@ -1520,7 +1535,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 6c6822217..b70fe9d1d 100644 --- a/crates/streams/tests/redis_store.rs +++ b/crates/streams/tests/redis_store.rs @@ -1609,8 +1609,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)) @@ -1619,6 +1625,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; // A Workflow's output is an append as its run's producer, so it notifies the same way. let landed = store .append(append(