From ef9153afa8e9b89f3c69c738c1b65215202cd7cf Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sat, 10 Oct 2026 21:25:57 -0700 Subject: [PATCH 01/11] Notified each stream's notifier after appends and visible batches. A caller of a stream-returning Nexus operation sees progress only through the notifier. One call in flight per stream, folded, with counters from store positions. --- crates/streams/Cargo.toml | 3 +- crates/streams/src/lib.rs | 5 + crates/streams/src/notify.rs | 926 +++++++++++++++++++++++++++++++++++ crates/streams/src/store.rs | 4 + 4 files changed, 937 insertions(+), 1 deletion(-) create mode 100644 crates/streams/src/notify.rs diff --git a/crates/streams/Cargo.toml b/crates/streams/Cargo.toml index 2a01324de..13b23667a 100644 --- a/crates/streams/Cargo.toml +++ b/crates/streams/Cargo.toml @@ -20,6 +20,7 @@ redis = ["dep:redis", "dep:crc16"] [dependencies] async-trait = "0.1" crc16 = { version = "0.4", optional = true } +lru = "0.18" prost = { workspace = true } redis = { version = "1.7", optional = true, default-features = false, features = [ "cluster-async", @@ -30,7 +31,7 @@ redis = { version = "1.7", optional = true, default-features = false, features = ] } sha2 = "0.10" thiserror = { workspace = true } -tokio = { version = "1.47", default-features = false, features = ["sync", "time"] } +tokio = { version = "1.47", default-features = false, features = ["rt", "sync", "time"] } tracing = "0.1" [dependencies.temporalio-common] diff --git a/crates/streams/src/lib.rs b/crates/streams/src/lib.rs index 32713f4e2..2fe3090d9 100644 --- a/crates/streams/src/lib.rs +++ b/crates/streams/src/lib.rs @@ -13,6 +13,7 @@ mod cursor; mod error; mod handle; mod memory; +mod notify; mod owner; mod reader; mod record; @@ -27,6 +28,10 @@ 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 memory::MemoryStore; +pub use notify::{ + DEFAULT_MAX_NOTIFIERS, FLUSH_LIMIT, Notifier, NotifierClient, NotifyError, NotifyingStore, + StreamNotification, progress_counter, +}; pub use owner::{OwnerClient, OwnerDescription, OwnerError}; pub use reader::{DEFAULT_MAX_RECORDS, ReadTarget, read_page}; pub use record::{ diff --git a/crates/streams/src/notify.rs b/crates/streams/src/notify.rs new file mode 100644 index 000000000..0e74b7029 --- /dev/null +++ b/crates/streams/src/notify.rs @@ -0,0 +1,926 @@ +//! Telling a stream's notifier on the server that the stream moved or closed. +//! +//! The notifier holds the callbacks of every caller a stream-returning Nexus operation attached. A +//! notification is a hint that there is something new to read. The records stay on the read path. + +use crate::{ + StreamResult, StreamStore, WORKFLOW_OWNER_KIND, mint_cursor, + proto::{ + ChainId, DeleteOwnerRequest, DeleteOwnerResponse, PendingStage, PromoteOutcome, + PromoteResult, StageRef, StagedBatch, StoreAppendRequest, StoreAppendResponse, + StoreLatestRequest, StoreLatestResponse, StoreReadRequest, StoreReadResponse, + }, + stream_hash, +}; +use std::{ + num::NonZeroUsize, + sync::{Arc, Mutex}, + time::Duration, +}; +use temporalio_common::protos::temporal::api::common::v1::Payload; +use tokio::sync::watch; + +/// How many streams a process keeps a notifier for when its configuration doesn't say. +pub const DEFAULT_MAX_NOTIFIERS: usize = 1000; + +/// How long a stopping Worker waits for the notifications still out. +pub const FLUSH_LIMIT: Duration = Duration::from_secs(10); + +// Redis entry ids are `-`. Twenty bits hold the sequence, far more entries than one +// stream takes in a millisecond, and the milliseconds stay below 2^43 until the year 2248, so the +// packed value fits a positive i64. +const REDIS_SEQUENCE_BITS: u32 = 20; + +/// One notification for one stream's notifier. +#[derive(Debug, Clone, PartialEq)] +pub struct StreamNotification { + /// The run chain whose stream moved. Its first run keys the notifier. + pub chain: ChainId, + /// The stream's topic. + pub topic: String, + /// The cursor of the newest record the notification covers. + pub position: String, + /// Grows with the stream. The notifier keeps the highest and a caller drops a lower one. + pub counter: i64, + /// Set on the close, which completes every attached operation with it. + pub close_result: Option, +} + +/// Why a notifier didn't take a notification. +#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)] +pub enum NotifyError { + /// The server will never take it, so trying again can't help. + #[error("the server refused the notification: {0}")] + Refused(String), + /// It may take it later. + #[error("the notification failed: {0}")] + Failed(String), +} + +/// Sends notifications to the stream notifier on the server. +#[async_trait::async_trait] +pub trait NotifierClient: Send + Sync { + /// Sends one notification. + async fn notify(&self, notification: StreamNotification) -> Result<(), NotifyError>; +} + +/// The notifier counter for a store position. +/// +/// It must grow with the stream, and producers in different processes must agree on it. So it +/// comes from the store's own position, not from a clock. A memory position, an offset, gives +/// `offset + 1`. A Redis entry id `-` gives `ms << 20 | seq`, with a sequence beyond 20 +/// bits held at the largest, which keeps the order non-decreasing. No position gives zero. Two +/// records past the 20-bit sequence in one millisecond get one counter, and the notifier drops +/// the second as a repeat. The next notification tells the reader again. +pub fn progress_counter(position: &str) -> Option { + if position.is_empty() { + return Some(0); + } + match position.split_once('-') { + Some((milliseconds, sequence)) => { + let milliseconds: i64 = milliseconds.parse().ok()?; + let sequence: u64 = sequence.parse().ok()?; + let largest = (1u64 << REDIS_SEQUENCE_BITS) - 1; + let sequence = i64::try_from(sequence.min(largest)).ok()?; + milliseconds + .checked_mul(1 << REDIS_SEQUENCE_BITS) + .map(|shifted| shifted | sequence) + .filter(|counter| *counter >= 0) + } + None => position.parse::().ok()?.checked_add(1), + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +struct StreamKey { + namespace: String, + workflow_id: String, + first_run_id: String, + topic: String, +} + +impl StreamKey { + fn new(chain: &ChainId, topic: &str) -> Self { + Self { + namespace: chain.namespace.clone(), + workflow_id: chain.workflow_id.clone(), + first_run_id: chain.first_run_id.clone(), + topic: topic.to_string(), + } + } + + fn chain(&self) -> ChainId { + ChainId { + namespace: self.namespace.clone(), + workflow_id: self.workflow_id.clone(), + first_run_id: self.first_run_id.clone(), + } + } +} + +#[derive(Debug, Default)] +struct SlotState { + pending: Option<(String, i64)>, + sending: bool, + closed: bool, +} + +/// One stream's notifications: folded, with one call in flight. +#[derive(Debug)] +struct Slot { + key: StreamKey, + state: Mutex, + idle: watch::Sender, +} + +impl Slot { + fn new(key: StreamKey) -> Arc { + Arc::new(Self { + key, + state: Mutex::default(), + idle: watch::Sender::new(true), + }) + } + + /// 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) { + let mut state = self.state.lock().unwrap(); + if state.closed { + return; + } + if state + .pending + .as_ref() + .is_none_or(|(_, held)| counter >= *held) + { + state.pending = Some((position, counter)); + } + if !state.sending { + state.sending = true; + self.idle.send_replace(false); + tokio::spawn(self.clone().send_pending(client.clone())); + } + } + + async fn send_pending(self: Arc, client: Arc) { + loop { + let (position, counter) = { + let mut state = self.state.lock().unwrap(); + match state.pending.take() { + Some(next) => next, + None => { + state.sending = false; + self.idle.send_replace(true); + return; + } + } + }; + let notification = StreamNotification { + chain: self.key.chain(), + topic: self.key.topic.clone(), + position, + counter, + close_result: None, + }; + // A notification is only a hint: the next one tells the reader again, and the reader + // reads from its own cursor. + if let Err(error) = client.notify(notification).await { + tracing::warn!( + workflow_id = %self.key.workflow_id, + topic = %self.key.topic, + %error, + "A stream notification failed. The next one tells the reader again" + ); + } + } + } + + /// Stops new notifications and waits for the one in flight, so a close is the last thing the + /// notifier hears from this process. + async fn close(&self) { + self.state.lock().unwrap().closed = true; + self.wait_idle().await; + } + + async fn wait_idle(&self) { + let mut idle = self.idle.subscribe(); + // The sender lives as long as the slot, so the wait can't fail. + let _ = idle.wait_for(|idle| *idle).await; + } + + fn is_idle(&self) -> bool { + *self.idle.borrow() + } +} + +/// Notifies the stream notifiers of the streams a process writes to. +/// +/// Keeps one slot per stream, and drops a slot when its stream closes, when its chain ends, or +/// when it's the least recently used of more than the limit. A dropped slot still sends what it +/// holds, and [Notifier::flush] waits for it. +pub struct Notifier { + client: Arc, + slots: Mutex>>, + retired: Mutex>>, +} + +impl std::fmt::Debug for Notifier { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("Notifier") + .field("slots", &self.slots.lock().unwrap().len()) + .finish_non_exhaustive() + } +} + +impl Notifier { + /// Notifies through `client`, keeping at most `max_notifiers` streams' slots. + pub fn new(client: Arc, max_notifiers: usize) -> Self { + Self { + client, + slots: Mutex::new(lru::LruCache::new( + NonZeroUsize::new(max_notifiers).unwrap_or(NonZeroUsize::MIN), + )), + retired: Mutex::default(), + } + } + + /// 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.slot(StreamKey::new(chain, topic)) + .notify(&self.client, position, counter); + } + + /// Closes the stream on the server with `result`, after the notification in flight, and + /// drops its slot. `counter` must rank above the stream's last notification. + pub async fn close( + &self, + chain: &ChainId, + topic: &str, + position: String, + counter: i64, + result: Payload, + ) -> Result<(), NotifyError> { + let key = StreamKey::new(chain, topic); + let held = self.slots.lock().unwrap().pop(&key); + if let Some(slot) = &held { + slot.close().await; + } + self.client + .notify(StreamNotification { + chain: chain.clone(), + topic: topic.to_string(), + position, + counter, + close_result: Some(result), + }) + .await + } + + /// Drops the slots of `chain`'s streams, once the chain ended. + pub fn forget_chain(&self, chain: &ChainId) { + let mut slots = self.slots.lock().unwrap(); + let ended: Vec = slots + .iter() + .map(|(key, _)| key) + .filter(|key| key.chain() == *chain) + .cloned() + .collect(); + let dropped: Vec> = ended.iter().filter_map(|key| slots.pop(key)).collect(); + drop(slots); + self.retire(dropped); + } + + /// Drops the slot of one stream, once it closed. + pub fn forget_stream(&self, chain: &ChainId, topic: &str) { + let dropped = self + .slots + .lock() + .unwrap() + .pop(&StreamKey::new(chain, topic)); + self.retire(dropped); + } + + /// Waits until no notification is out or waiting. + pub async fn flush(&self) { + let slots: Vec> = self + .slots + .lock() + .unwrap() + .iter() + .map(|(_, slot)| slot.clone()) + .chain(self.retired.lock().unwrap().iter().cloned()) + .collect(); + for slot in slots { + slot.wait_idle().await; + } + self.retired.lock().unwrap().retain(|slot| !slot.is_idle()); + } + + fn slot(&self, key: StreamKey) -> Arc { + let mut slots = self.slots.lock().unwrap(); + if let Some(slot) = slots.get(&key) { + return slot.clone(); + } + let slot = Slot::new(key.clone()); + let evicted = slots.push(key, slot.clone()).map(|(_, slot)| slot); + drop(slots); + self.retire(evicted); + slot + } + + fn retire(&self, dropped: impl IntoIterator>) { + let mut retired = self.retired.lock().unwrap(); + retired.retain(|slot| !slot.is_idle()); + retired.extend(dropped.into_iter().filter(|slot| !slot.is_idle())); + } +} + +/// A store that notifies each stream's notifier after an append lands and after a Workflow's +/// staged output becomes visible. +/// +/// Wraps the store a process hands to its stream calls and its Workers, so every write path +/// notifies without knowing about notifiers. +pub struct NotifyingStore { + inner: Arc, + notifier: Arc, +} + +impl NotifyingStore { + /// Wraps `inner`, notifying through `notifier`. + pub fn new(inner: Arc, notifier: Arc) -> Self { + Self { inner, notifier } + } + + /// The notifier this store notifies through. + pub fn notifier(&self) -> &Arc { + &self.notifier + } + + fn cursor(&self, chain: &ChainId, topic: &str, position: &str) -> String { + let hash = stream_hash( + &chain.namespace, + WORKFLOW_OWNER_KIND, + &chain.workflow_id, + topic, + ); + mint_cursor(self.inner.name(), &hash, position) + } + + fn notify_at(&self, chain: &ChainId, topic: &str, position: &str) { + // The write landed, so a notification that can't go must not turn it into an error the + // caller might retry with other content. + let Some(counter) = progress_counter(position) else { + tracing::warn!( + workflow_id = %chain.workflow_id, + topic, + position, + "No notifier counter comes from this position, so the write isn't notified" + ); + return; + }; + let cursor = self.cursor(chain, topic, position); + self.notifier.notify(chain, topic, cursor, counter); + } + + async fn notify_latest(&self, chain: &ChainId, topic: &str) { + // A promotion has no position of its own, so the newest record's is told. + match self + .inner + .latest(StoreLatestRequest { + chain: Some(chain.clone()), + topic: topic.to_string(), + }) + .await + { + Ok(latest) if !latest.position.is_empty() => { + self.notify_at(chain, topic, &latest.position) + } + Ok(_) => {} + Err(error) => tracing::warn!( + workflow_id = %chain.workflow_id, + topic, + %error, + "Could not find the newest record to notify" + ), + } + } +} + +#[async_trait::async_trait] +impl StreamStore for NotifyingStore { + fn name(&self) -> &str { + self.inner.name() + } + + async fn append(&self, request: StoreAppendRequest) -> StreamResult { + let chain = request.chain.clone(); + let topic = request.topic.clone(); + let landed = self.inner.append(request).await?; + if let Some(chain) = &chain { + self.notify_at(chain, &topic, &landed.last_position); + } + Ok(landed) + } + + async fn read(&self, request: StoreReadRequest) -> StreamResult { + self.inner.read(request).await + } + + async fn latest(&self, request: StoreLatestRequest) -> StreamResult { + self.inner.latest(request).await + } + + async fn record_at( + &self, + chain: &ChainId, + topic: &str, + position: &str, + ) -> StreamResult>> { + self.inner.record_at(chain, topic, position).await + } + + async fn stage(&self, batch: StagedBatch) -> StreamResult<()> { + self.inner.stage(batch).await + } + + async fn promote(&self, stage: &StageRef) -> StreamResult { + let result = self.inner.promote(stage).await?; + if result.outcome() == PromoteOutcome::Promoted + && let Some(chain) = &stage.chain + { + for topic in &stage.topics { + self.notify_latest(chain, topic).await; + } + } + Ok(result) + } + + async fn abort(&self, stage: &StageRef) -> StreamResult<()> { + self.inner.abort(stage).await + } + + async fn close_chain(&self, chain: &ChainId) -> StreamResult<()> { + self.inner.close_chain(chain).await?; + self.notifier.forget_chain(chain); + Ok(()) + } + + async fn close_topic(&self, chain: &ChainId, topic: &str) -> StreamResult<()> { + self.inner.close_topic(chain, topic).await?; + self.notifier.forget_stream(chain, topic); + Ok(()) + } + + async fn pending_stages(&self, chain: &ChainId) -> StreamResult> { + self.inner.pending_stages(chain).await + } + + async fn delete_owner(&self, request: DeleteOwnerRequest) -> StreamResult { + self.inner.delete_owner(request).await + } + + async fn flush_notifications(&self) { + self.notifier.flush().await; + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::{ + MemoryStore, + proto::{StagedRecord, StoreAppendRequest}, + }; + use std::sync::atomic::{AtomicBool, Ordering}; + use tokio::sync::Semaphore; + + /// Records what it was sent. While `holding`, each call waits for a permit, so a test + /// decides when the call in flight returns. + struct Recorder { + sent: Mutex>, + holding: AtomicBool, + gate: Semaphore, + fail: AtomicBool, + } + + impl Default for Recorder { + fn default() -> Self { + Self { + sent: Mutex::default(), + holding: AtomicBool::new(false), + gate: Semaphore::new(0), + fail: AtomicBool::new(false), + } + } + } + + impl Recorder { + fn holding() -> Arc { + let recorder = Arc::new(Self::default()); + recorder.holding.store(true, Ordering::SeqCst); + recorder + } + + fn sent(&self) -> Vec<(String, i64)> { + self.sent + .lock() + .unwrap() + .iter() + .map(|n| (n.position.clone(), n.counter)) + .collect() + } + + fn release(&self, calls: usize) { + self.gate.add_permits(calls); + } + } + + #[async_trait::async_trait] + impl NotifierClient for Recorder { + async fn notify(&self, notification: StreamNotification) -> Result<(), NotifyError> { + self.sent.lock().unwrap().push(notification); + if self.holding.load(Ordering::SeqCst) { + self.gate.acquire().await.unwrap().forget(); + } + if self.fail.load(Ordering::SeqCst) { + return Err(NotifyError::Failed("down".to_string())); + } + Ok(()) + } + } + + fn chain(first_run_id: &str) -> ChainId { + ChainId { + namespace: "ns".to_string(), + workflow_id: "wf".to_string(), + first_run_id: first_run_id.to_string(), + } + } + + /// Lets every spawned send run until it waits. + async fn settle() { + for _ in 0..10 { + tokio::task::yield_now().await; + } + } + + #[test] + fn counters_come_from_the_stores_position() { + assert_eq!(progress_counter(""), Some(0)); + assert_eq!(progress_counter("0"), Some(1)); + assert_eq!(progress_counter("41"), Some(42)); + assert_eq!( + progress_counter("1700000000000-3"), + Some((1_700_000_000_000 << 20) | 3) + ); + assert_eq!( + progress_counter("5-2000000"), + Some((5 << 20) | ((1 << 20) - 1)), + "a sequence past 20 bits is held at the largest" + ); + assert!(progress_counter("1-1") < progress_counter("1-2")); + assert!(progress_counter("1-2") < progress_counter("2-0")); + assert_eq!(progress_counter("not a position"), None); + assert_eq!(progress_counter("9223372036854775807-0"), None, "past i64"); + } + + #[tokio::test] + async fn notifications_fold_with_one_call_in_flight() { + let recorder = Recorder::holding(); + let notifier = Notifier::new(recorder.clone(), 10); + notifier.notify(&chain("run-1"), "t", "c1".to_string(), 1); + settle().await; + notifier.notify(&chain("run-1"), "t", "c3".to_string(), 3); + notifier.notify(&chain("run-1"), "t", "c2".to_string(), 2); + settle().await; + assert_eq!( + recorder.sent(), + vec![("c1".to_string(), 1)], + "one call in flight" + ); + recorder.release(2); + notifier.flush().await; + assert_eq!( + recorder.sent(), + vec![("c1".to_string(), 1), ("c3".to_string(), 3)], + "the burst folds into one call with the highest counter" + ); + } + + #[tokio::test] + async fn a_failed_notification_leaves_the_next_one_to_tell_the_reader() { + let recorder = Arc::new(Recorder::default()); + recorder.fail.store(true, Ordering::SeqCst); + let notifier = Notifier::new(recorder.clone(), 10); + notifier.notify(&chain("run-1"), "t", "c1".to_string(), 1); + notifier.flush().await; + recorder.fail.store(false, Ordering::SeqCst); + notifier.notify(&chain("run-1"), "t", "c2".to_string(), 2); + notifier.flush().await; + assert_eq!( + recorder.sent(), + vec![("c1".to_string(), 1), ("c2".to_string(), 2)] + ); + } + + #[tokio::test] + async fn close_waits_for_the_call_in_flight_and_carries_the_result() { + let recorder = Recorder::holding(); + let notifier = Arc::new(Notifier::new(recorder.clone(), 10)); + notifier.notify(&chain("run-1"), "t", "c1".to_string(), 1); + settle().await; + let closing = { + let notifier = notifier.clone(); + tokio::spawn(async move { + let result = Payload { + data: b"summary".to_vec(), + ..Default::default() + }; + notifier + .close(&chain("run-1"), "t", "c1".to_string(), 2, result) + .await + }) + }; + settle().await; + assert_eq!( + recorder.sent().len(), + 1, + "the close waits for the call in flight" + ); + recorder.release(2); + closing.await.unwrap().unwrap(); + let sent = recorder.sent.lock().unwrap().clone(); + assert_eq!(sent.len(), 2); + assert_eq!(sent[1].counter, 2); + assert_eq!(sent[1].close_result.as_ref().unwrap().data, b"summary"); + } + + #[tokio::test] + async fn a_notification_after_close_sends_nothing() { + let recorder = Arc::new(Recorder::default()); + let notifier = Notifier::new(recorder.clone(), 10); + 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); + notifier.flush().await; + assert_eq!(recorder.sent(), vec![("c1".to_string(), 1)]); + } + + #[tokio::test] + async fn notifiers_beyond_the_cap_are_dropped_least_recent_first() { + let recorder = Recorder::holding(); + let notifier = Notifier::new(recorder.clone(), 2); + notifier.notify(&chain("run-1"), "a", "a1".to_string(), 1); + notifier.notify(&chain("run-1"), "b", "b1".to_string(), 1); + settle().await; + notifier.notify(&chain("run-1"), "a", "a2".to_string(), 2); + notifier.notify(&chain("run-1"), "c", "c1".to_string(), 1); + let kept: Vec = notifier + .slots + .lock() + .unwrap() + .iter() + .map(|(key, _)| key.topic.clone()) + .collect(); + assert_eq!(kept, vec!["c", "a"], "b was the least recently used"); + recorder.release(10); + notifier.flush().await; + let mut sent = recorder.sent(); + sent.sort(); + assert_eq!( + sent, + vec![ + ("a1".to_string(), 1), + ("a2".to_string(), 2), + ("b1".to_string(), 1), + ("c1".to_string(), 1) + ], + "a dropped notifier still sends what it holds" + ); + assert!(notifier.retired.lock().unwrap().is_empty()); + } + + #[tokio::test] + async fn an_ended_chain_drops_its_notifiers() { + let recorder = Arc::new(Recorder::default()); + let notifier = Notifier::new(recorder.clone(), 10); + notifier.notify(&chain("run-1"), "a", "a1".to_string(), 1); + notifier.notify(&chain("run-1"), "b", "b1".to_string(), 1); + notifier.notify(&chain("run-9"), "a", "x1".to_string(), 1); + notifier.forget_chain(&chain("run-1")); + notifier.forget_stream(&chain("run-9"), "a"); + assert_eq!(notifier.slots.lock().unwrap().len(), 0); + notifier.flush().await; + assert_eq!( + recorder.sent().len(), + 3, + "the dropped ones still sent what they held" + ); + } + + #[tokio::test] + async fn flush_waits_for_every_notification_in_flight() { + let recorder = Recorder::holding(); + let notifier = Arc::new(Notifier::new(recorder.clone(), 10)); + notifier.notify(&chain("run-1"), "a", "a1".to_string(), 1); + notifier.notify(&chain("run-1"), "b", "b1".to_string(), 1); + let flushing = { + let notifier = notifier.clone(); + tokio::spawn(async move { notifier.flush().await }) + }; + settle().await; + assert!(!flushing.is_finished()); + recorder.release(1); + settle().await; + assert!(!flushing.is_finished(), "one is still out"); + recorder.release(1); + flushing.await.unwrap(); + } + + fn append_request(chain: &ChainId, topic: &str, sequence: i64) -> StoreAppendRequest { + StoreAppendRequest { + chain: Some(chain.clone()), + topic: topic.to_string(), + producer_id: "p".to_string(), + attempt: 1, + sequence, + digest: vec![sequence as u8], + records: vec![b"record".to_vec()], + } + } + + fn notifying_store(recorder: &Arc) -> (Arc, NotifyingStore) { + let inner = Arc::new(MemoryStore::new()); + let notifier = Arc::new(Notifier::new(recorder.clone(), 10)); + (inner.clone(), NotifyingStore::new(inner, notifier)) + } + + #[tokio::test] + async fn an_append_notifies_the_chain_it_wrote_to() { + let recorder = Arc::new(Recorder::default()); + let (_, store) = notifying_store(&recorder); + let landed = store + .append(append_request(&chain("run-1"), "t", 1)) + .await + .unwrap(); + store.flush_notifications().await; + let sent = recorder.sent.lock().unwrap().clone(); + assert_eq!(sent.len(), 1); + assert_eq!(sent[0].chain, chain("run-1")); + assert_eq!(sent[0].topic, "t"); + assert_eq!( + sent[0].counter, + progress_counter(&landed.last_position).unwrap() + ); + assert_eq!( + sent[0].position, + mint_cursor( + "memory", + &stream_hash("ns", WORKFLOW_OWNER_KIND, "wf", "t"), + &landed.last_position + ), + "the position is the cursor a reader resumes from" + ); + } + + /// A store whose appends land at a position no counter comes from. + struct OddPositions(MemoryStore); + + #[async_trait::async_trait] + impl StreamStore for OddPositions { + fn name(&self) -> &str { + "odd" + } + async fn append(&self, r: StoreAppendRequest) -> StreamResult { + self.0.append(r).await?; + Ok(StoreAppendResponse { + first_position: "odd".to_string(), + last_position: "odd".to_string(), + }) + } + 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 { + self.0.promote(s).await + } + 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) -> StreamResult<()> { + self.0.close_topic(c, t).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_position_without_a_counter_does_not_fail_the_append() { + let recorder = Arc::new(Recorder::default()); + let notifier = Arc::new(Notifier::new(recorder.clone(), 10)); + let store = NotifyingStore::new(Arc::new(OddPositions(MemoryStore::new())), notifier); + store + .append(append_request(&chain("run-1"), "t", 1)) + .await + .unwrap(); + store.flush_notifications().await; + assert!(recorder.sent().is_empty()); + } + + fn staged(chain: &ChainId, token: &str, topics: &[&str]) -> StagedBatch { + StagedBatch { + chain: Some(chain.clone()), + run_id: chain.first_run_id.clone(), + token: token.to_string(), + history_floor_event_id: 3, + records: topics + .iter() + .map(|topic| StagedRecord { + topic: topic.to_string(), + record: b"record".to_vec(), + }) + .collect(), + } + } + + fn stage_ref(chain: &ChainId, token: &str, topics: &[&str]) -> StageRef { + StageRef { + chain: Some(chain.clone()), + token: token.to_string(), + topics: topics.iter().map(|topic| topic.to_string()).collect(), + } + } + + #[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 run = chain("run-1"); + store + .stage(staged(&run, "token", &["a", "b", "a"])) + .await + .unwrap(); + store.flush_notifications().await; + assert!(recorder.sent().is_empty(), "a stage isn't visible yet"); + store + .promote(&stage_ref(&run, "token", &["a", "b"])) + .await + .unwrap(); + store.flush_notifications().await; + let sent = recorder.sent.lock().unwrap().clone(); + let topics: Vec<(&str, i64)> = sent.iter().map(|n| (n.topic.as_str(), n.counter)).collect(); + assert_eq!( + topics, + vec![("a", 2), ("b", 1)], + "each topic's newest record" + ); + + store + .promote(&stage_ref(&run, "token", &["a", "b"])) + .await + .unwrap(); + store.stage(staged(&run, "aborted", &["a"])).await.unwrap(); + store + .abort(&stage_ref(&run, "aborted", &["a"])) + .await + .unwrap(); + store.flush_notifications().await; + assert_eq!( + recorder.sent().len(), + 2, + "a repeated promotion and an abort make nothing visible" + ); + } + + #[tokio::test] + async fn a_closed_streams_notifier_is_dropped() { + let recorder = Arc::new(Recorder::default()); + let (_, store) = notifying_store(&recorder); + let run = chain("run-1"); + store.append(append_request(&run, "a", 1)).await.unwrap(); + store.append(append_request(&run, "b", 1)).await.unwrap(); + store.close_topic(&run, "a").await.unwrap(); + assert_eq!(store.notifier().slots.lock().unwrap().len(), 1); + store.close_chain(&run).await.unwrap(); + assert_eq!(store.notifier().slots.lock().unwrap().len(), 0); + } +} diff --git a/crates/streams/src/store.rs b/crates/streams/src/store.rs index b2c41cd85..4222b978c 100644 --- a/crates/streams/src/store.rs +++ b/crates/streams/src/store.rs @@ -75,6 +75,10 @@ pub trait StreamStore: Send + Sync { /// Deletes every stream and stage of one owner, across all its chains. async fn delete_owner(&self, request: DeleteOwnerRequest) -> StreamResult; + + /// Waits until the notifications this store sent are out. A store that sends none returns at + /// once. + async fn flush_notifications(&self) {} } #[cfg(test)] From 80d9ac47adad716667a2231006a6ee37d9b40842 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sat, 10 Oct 2026 21:25:57 -0700 Subject: [PATCH 02/11] Turned stream notifications on from the store config. Most streams have no Nexus caller, so notifications stay off unless the config asks. --- .../temporal/sdk/core/streams/streams.proto | 14 ++ crates/streams/src/service.rs | 133 +++++++++++++++++- 2 files changed, 142 insertions(+), 5 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 e5e613d79..6563f5c1e 100644 --- a/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto +++ b/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto @@ -24,6 +24,9 @@ service StreamService { rpc Close(CloseRequest) returns (CloseResponse); // Deletes every stream of one owner. rpc DeleteOwner(DeleteOwnerRequest) returns (DeleteOwnerResponse); + // Waits until the stream notifications this process sent are out. Call it before the process + // stops, since a notification is sent in the background. + rpc FlushNotifications(FlushNotificationsRequest) returns (FlushNotificationsResponse); } // Which store holds a process's streams, and how to reach it. Lang builds one store from it and @@ -33,6 +36,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. Needs a server with the stream notifier. + 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; } // Streams in a Redis or Redis Cluster that Core connects to itself. @@ -190,6 +200,10 @@ message CloseRequest { message CloseResponse {} +message FlushNotificationsRequest {} + +message FlushNotificationsResponse {} + message DeleteOwnerRequest { string namespace = 1; StreamOwnerKind owner_kind = 2; diff --git a/crates/streams/src/service.rs b/crates/streams/src/service.rs index d1bab5dc2..bf1fdfe52 100644 --- a/crates/streams/src/service.rs +++ b/crates/streams/src/service.rs @@ -1,10 +1,12 @@ //! The stream calls lang makes, served in process. use crate::{ - OwnerClient, StreamError, StreamResult, StreamStore, Streams, StreamsOptions, + DEFAULT_MAX_NOTIFIERS, Notifier, NotifierClient, NotifyingStore, OwnerClient, StreamError, + StreamResult, StreamStore, Streams, StreamsOptions, proto::{ - AppendRequest, CloseRequest, DeleteOwnerRequest, LatestRequest, ReadRequest, StreamFailure, - StreamStoreConfig, stream_store_config, + AppendRequest, CloseRequest, DeleteOwnerRequest, FlushNotificationsRequest, + FlushNotificationsResponse, LatestRequest, ReadRequest, StreamFailure, StreamStoreConfig, + stream_store_config, }, }; use prost::Message; @@ -31,10 +33,21 @@ impl std::fmt::Debug for StreamService { } impl StreamService { - /// Connects to the store `config` names, asking `owner` about streams' owners. + /// Connects to the store `config` names, asking `owner` about streams' owners. Refuses a + /// config that turns notifications on, since there's no client to send them with. pub async fn connect( config: StreamStoreConfig, owner: Arc, + ) -> StreamResult { + Self::connect_with_notifier(config, owner, None).await + } + + /// Like [StreamService::connect], sending stream notifications through `notifier` when the + /// config turns them on. + pub async fn connect_with_notifier( + config: StreamStoreConfig, + owner: Arc, + notifier: Option>, ) -> StreamResult { let (store, retention): (Arc, Duration) = match config.store { Some(stream_store_config::Store::Memory(_)) => { @@ -47,6 +60,21 @@ impl StreamService { )); } }; + let store: Arc = if config.notify_on_append { + let client = notifier.ok_or_else(|| { + StreamError::refused("stream notifications need a client to send them with") + })?; + let max_notifiers = match config.max_notifiers { + 0 => DEFAULT_MAX_NOTIFIERS, + max => max as usize, + }; + Arc::new(NotifyingStore::new( + store, + Arc::new(Notifier::new(client, max_notifiers)), + )) + } else { + store + }; Ok(Self::new( store, owner, @@ -98,6 +126,11 @@ impl StreamService { .delete_owner(decode::(rpc, request)?) .await .map(|r| r.encode_to_vec()), + "FlushNotifications" => { + decode::(rpc, request)?; + self.store.flush_notifications().await; + Ok(FlushNotificationsResponse {}.encode_to_vec()) + } other => Err(StreamError::unsupported(format!( "the stream service has no call {other:?}" ))), @@ -193,6 +226,7 @@ mod tests { StreamService::connect( StreamStoreConfig { store: Some(stream_store_config::Store::Memory(MemoryStoreConfig {})), + ..Default::default() }, Arc::new(Running), ) @@ -318,7 +352,7 @@ 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); @@ -330,6 +364,7 @@ mod tests { let error = StreamService::connect( StreamStoreConfig { store: Some(stream_store_config::Store::Redis(Default::default())), + ..Default::default() }, Arc::new(Running), ) @@ -337,4 +372,92 @@ mod tests { .unwrap_err(); assert_eq!(error.kind, StreamFailureKind::Unsupported); } + + /// Records the notifications it was sent. + #[derive(Default)] + struct Recorder(std::sync::Mutex>); + + #[async_trait::async_trait] + impl NotifierClient for Recorder { + async fn notify( + &self, + notification: crate::StreamNotification, + ) -> Result<(), crate::NotifyError> { + self.0.lock().unwrap().push(notification); + Ok(()) + } + } + + async fn service_notifying(notify_on_append: bool, recorder: Arc) -> StreamService { + StreamService::connect_with_notifier( + StreamStoreConfig { + store: Some(stream_store_config::Store::Memory(MemoryStoreConfig {})), + notify_on_append, + ..Default::default() + }, + Arc::new(Running), + Some(recorder), + ) + .await + .unwrap() + } + + async fn flush(service: &StreamService) { + let response = service + .call( + "FlushNotifications", + &FlushNotificationsRequest {}.encode_to_vec(), + ) + .await + .unwrap(); + FlushNotificationsResponse::decode(response.as_slice()).unwrap(); + } + + #[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; + let appended = AppendResponse::decode( + service + .call("Append", &append(1, &["a", "b"])) + .await + .unwrap() + .as_slice(), + ) + .unwrap(); + flush(&service).await; + let sent = recorder.0.lock().unwrap().clone(); + assert_eq!(sent.len(), 1); + assert_eq!( + sent[0].chain.first_run_id, "run-1", + "the chain the append wrote to" + ); + assert_eq!(sent[0].topic, "out"); + assert_eq!(sent[0].position, appended.last_cursor); + assert_eq!(sent[0].counter, 2); + } + + #[tokio::test] + async fn notifications_are_off_by_default() { + let recorder = Arc::new(Recorder::default()); + let service = service_notifying(false, recorder.clone()).await; + service.call("Append", &append(1, &["a"])).await.unwrap(); + flush(&service).await; + assert!(recorder.0.lock().unwrap().is_empty()); + } + + #[tokio::test] + async fn notifications_without_a_client_are_refused() { + let error = StreamService::connect( + StreamStoreConfig { + store: Some(stream_store_config::Store::Memory(MemoryStoreConfig {})), + notify_on_append: true, + ..Default::default() + }, + Arc::new(Running), + ) + .await + .unwrap_err(); + assert_eq!(error.kind, StreamFailureKind::Refused); + } } From 18dc4580881656fe1072fabf2caa12f0974f0920 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sat, 10 Oct 2026 21:25:57 -0700 Subject: [PATCH 03/11] Sent notifications through Core's client and flushed them at stop. A notification goes out in the background, so a Worker waits a short while for it. --- crates/sdk-core/src/streams.rs | 135 +++++++++++++++++- crates/sdk-core/src/worker/workflow/mod.rs | 4 + .../src/worker/workflow/stream_output.rs | 123 ++++++++++++++++ 3 files changed, 257 insertions(+), 5 deletions(-) diff --git a/crates/sdk-core/src/streams.rs b/crates/sdk-core/src/streams.rs index 9c4220091..50606c4b7 100644 --- a/crates/sdk-core/src/streams.rs +++ b/crates/sdk-core/src/streams.rs @@ -5,20 +5,83 @@ use std::sync::Arc; use temporalio_client::Connection; use temporalio_common::protos::temporal::api::{ common::v1::WorkflowExecution, - workflowservice::v1::{DescribeWorkflowExecutionRequest, DescribeWorkflowExecutionResponse}, + enums::v1::StreamOwnerKind, + stream::v1::StreamReference, + workflowservice::v1::{ + DescribeWorkflowExecutionRequest, DescribeWorkflowExecutionResponse, NotifyStreamRequest, + }, +}; +use temporalio_streams::{ + NotifierClient, NotifyError, OwnerClient, OwnerDescription, OwnerError, StreamNotification, }; -use temporalio_streams::{OwnerClient, OwnerDescription, OwnerError}; pub use temporalio_streams::{StreamError, StreamResult, StreamService, StreamStore, proto}; use tonic::IntoRequest; /// Connects to the store `config` names. The service asks `connection`'s server about streams' -/// owners, and its [StreamService::store] goes to each Worker's `stream_store`. +/// owners, sends the stream notifications the config turns on to it, and its +/// [StreamService::store] goes to each Worker's `stream_store`. pub async fn connect_stream_service( config: proto::StreamStoreConfig, connection: Connection, ) -> StreamResult> { - let owner = Arc::new(ConnectionOwnerClient { connection }); - Ok(Arc::new(StreamService::connect(config, owner).await?)) + let owner = Arc::new(ConnectionOwnerClient { + connection: connection.clone(), + }); + let notifier = Arc::new(ConnectionNotifierClient { connection }); + Ok(Arc::new( + StreamService::connect_with_notifier(config, owner, Some(notifier)).await?, + )) +} + +/// Sends stream notifications to a Temporal server's stream notifier through a client +/// connection. +struct ConnectionNotifierClient { + connection: Connection, +} + +#[async_trait::async_trait] +impl NotifierClient for ConnectionNotifierClient { + async fn notify(&self, notification: StreamNotification) -> Result<(), NotifyError> { + self.connection + .workflow_service() + .notify_stream(notify_request(notification).into_request()) + .await + .map(|_| ()) + .map_err(notify_error) + } +} + +fn notify_request(notification: StreamNotification) -> NotifyStreamRequest { + NotifyStreamRequest { + namespace: notification.chain.namespace, + // The run chain's first run keys the notifier, so one notifier serves the stream across + // Continue-as-New and a new chain on the same Workflow id gets its own. + stream_ref: Some(StreamReference { + owner_kind: StreamOwnerKind::Workflow as i32, + workflow_id: notification.chain.workflow_id, + run_id: notification.chain.first_run_id, + topic: notification.topic, + }), + position: notification.position, + counter: notification.counter, + close: notification.close_result.is_some(), + close_result: notification.close_result, + ..Default::default() + } +} + +/// Answers that mean the server will never take the notification, so a retry can't help. +fn notify_error(status: tonic::Status) -> NotifyError { + use tonic::Code; + match status.code() { + Code::Unimplemented + | Code::InvalidArgument + | Code::NotFound + | Code::PermissionDenied + | Code::Unauthenticated + | Code::FailedPrecondition => NotifyError::Refused(status.to_string()), + _ => NotifyError::Failed(status.to_string()), + } } /// Asks a Temporal server about owners through a client connection. @@ -144,4 +207,66 @@ mod tests { Err(OwnerError::Failed(_)) )); } + + #[test] + fn a_notification_names_the_stream_by_its_chains_first_run() { + let request = notify_request(StreamNotification { + chain: proto::ChainId { + namespace: "ns".to_string(), + workflow_id: "wf".to_string(), + first_run_id: "run-1".to_string(), + }, + topic: "tokens".to_string(), + position: "cursor".to_string(), + counter: 7, + close_result: None, + }); + assert_eq!(request.namespace, "ns"); + assert_eq!( + request.stream_ref, + Some(StreamReference { + owner_kind: StreamOwnerKind::Workflow as i32, + workflow_id: "wf".to_string(), + run_id: "run-1".to_string(), + topic: "tokens".to_string(), + }) + ); + assert_eq!((request.position.as_str(), request.counter), ("cursor", 7)); + assert!(!request.close); + assert!(request.close_result.is_none()); + } + + #[test] + fn only_answers_a_retry_cant_change_refuse_a_notification() { + for code in [ + tonic::Code::Unimplemented, + tonic::Code::InvalidArgument, + tonic::Code::NotFound, + tonic::Code::PermissionDenied, + tonic::Code::Unauthenticated, + tonic::Code::FailedPrecondition, + ] { + assert!( + matches!( + notify_error(tonic::Status::new(code, "")), + NotifyError::Refused(_) + ), + "{code:?}" + ); + } + for code in [ + tonic::Code::Unavailable, + tonic::Code::DeadlineExceeded, + tonic::Code::ResourceExhausted, + tonic::Code::Internal, + ] { + assert!( + matches!( + notify_error(tonic::Status::new(code, "")), + NotifyError::Failed(_) + ), + "{code:?}" + ); + } + } } diff --git a/crates/sdk-core/src/worker/workflow/mod.rs b/crates/sdk-core/src/worker/workflow/mod.rs index 32e8a2291..5908a632a 100644 --- a/crates/sdk-core/src/worker/workflow/mod.rs +++ b/crates/sdk-core/src/worker/workflow/mod.rs @@ -841,6 +841,10 @@ impl Workflows { anyhow!("Error joining workflow processing thread: {as_str:?}") })?; } + #[cfg(feature = "streams")] + if let Some(store) = &self.output_store { + store.flush_notifications().await; + } Ok(()) } diff --git a/crates/sdk-core/src/worker/workflow/stream_output.rs b/crates/sdk-core/src/worker/workflow/stream_output.rs index e8a51c0a7..09c8410c7 100644 --- a/crates/sdk-core/src/worker/workflow/stream_output.rs +++ b/crates/sdk-core/src/worker/workflow/stream_output.rs @@ -300,6 +300,20 @@ impl OutputStore { .map_err(|error| error.to_string()) } + /// Waits a short while for the stream notifications still out, since they're sent in the + /// background and a process that stops right after its Worker would drop them. + pub(crate) async fn flush_notifications(&self) { + if tokio::time::timeout( + temporalio_streams::FLUSH_LIMIT, + self.store.flush_notifications(), + ) + .await + .is_err() + { + warn!("Stream notifications were still out when the Worker stopped"); + } + } + async fn promote(&self, stage: temporalio_streams::proto::StageRef) { use temporalio_streams::proto::PromoteOutcome; match self.store.promote(&stage).await { @@ -787,6 +801,115 @@ mod tests { ) -> StreamResult { Err(StreamError::unsupported("delete_owner")) } + async fn flush_notifications(&self) { + self.0.lock().push("flush".to_string()); + } + } + + /// A store whose notifications never go out. + struct Stuck(Log); + + #[async_trait::async_trait] + impl StreamStore for Stuck { + fn name(&self) -> &str { + "stuck" + } + async fn append(&self, r: StoreAppendRequest) -> StreamResult { + self.0.append(r).await + } + 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 { + self.0.promote(s).await + } + 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) -> StreamResult<()> { + self.0.close_topic(c, t).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 + } + async fn flush_notifications(&self) { + std::future::pending::<()>().await; + } + } + + /// Records the stream notifications it was sent. + #[derive(Default)] + struct Recorder(Mutex>); + + #[async_trait::async_trait] + impl temporalio_streams::NotifierClient for Recorder { + async fn notify( + &self, + notification: temporalio_streams::StreamNotification, + ) -> Result<(), temporalio_streams::NotifyError> { + self.0.lock().push(notification); + Ok(()) + } + } + + #[tokio::test] + async fn an_accepted_completion_notifies_each_topic_it_made_visible() { + let recorder = Arc::new(Recorder::default()); + let notifying = Arc::new(temporalio_streams::NotifyingStore::new( + Arc::new(temporalio_streams::MemoryStore::new()), + Arc::new(temporalio_streams::Notifier::new(recorder.clone(), 10)), + )); + let store = OutputStore::new(notifying.clone(), "ns".to_string()); + store.stage(&[batch("a")]).await.unwrap(); + assert!(recorder.0.lock().is_empty(), "staged output isn't visible"); + store + .settle("run", vec![batch("a")], CompletionOutcome::Accepted) + .await; + store.flush_notifications().await; + let sent = recorder.0.lock().clone(); + assert_eq!(sent.len(), 1); + assert_eq!((sent[0].topic.as_str(), sent[0].counter), ("t", 1)); + assert_eq!(sent[0].chain.first_run_id, "run"); + } + + #[tokio::test] + async fn a_stopping_worker_waits_for_its_stream_notifications() { + let log = Arc::new(Log::default()); + OutputStore::new(log.clone(), "ns".to_string()) + .flush_notifications() + .await; + assert_eq!(*log.0.lock(), ["flush"]); + } + + #[tokio::test(start_paused = true)] + async fn a_stopping_worker_waits_for_its_stream_notifications_only_so_long() { + let store = OutputStore::new(Arc::new(Stuck(Log::default())), "ns".to_string()); + let started = tokio::time::Instant::now(); + store.flush_notifications().await; + assert_eq!(started.elapsed(), temporalio_streams::FLUSH_LIMIT); } fn proven(token: &str) -> ProvenStage { From 7eb2cb500e5cbdcadb0629c501a9a5fa76b18c1f Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sat, 10 Oct 2026 21:30:55 -0700 Subject: [PATCH 04/11] Tested notification counters against Redis entry ids. --- crates/streams/tests/redis_store.rs | 88 +++++++++++++++++++++++++++++ 1 file changed, 88 insertions(+) diff --git a/crates/streams/tests/redis_store.rs b/crates/streams/tests/redis_store.rs index c2d554856..7d4869846 100644 --- a/crates/streams/tests/redis_store.rs +++ b/crates/streams/tests/redis_store.rs @@ -1109,3 +1109,91 @@ async fn rule_20_4_a_malformed_entry_fails_with_its_cursor() { .unwrap(); assert_eq!(after.records.len(), 1); } + +/// Records the stream notifications it was sent. +#[derive(Default)] +struct Recorder(std::sync::Mutex>); + +#[async_trait::async_trait] +impl temporalio_streams::NotifierClient for Recorder { + async fn notify( + &self, + notification: temporalio_streams::StreamNotification, + ) -> Result<(), temporalio_streams::NotifyError> { + self.0.lock().unwrap().push(notification); + Ok(()) + } +} + +#[tokio::test] +async fn notification_counters_come_from_redis_entry_ids_and_grow_with_the_stream() { + let Some(setup) = setup().await else { return }; + let recorder = Arc::new(Recorder::default()); + let store = temporalio_streams::NotifyingStore::new( + Arc::new(setup.store), + Arc::new(temporalio_streams::Notifier::new(recorder.clone(), 10)), + ); + let chain = chain(); + let mut positions = vec![]; + for sequence in [1, 3] { + let digest = [sequence as u8]; + let landed = store + .append(append(&chain, "p", sequence, 2, &digest)) + .await + .unwrap(); + positions.push(landed.last_position); + store.flush_notifications().await; + } + store + .stage(StagedBatch { + chain: Some(chain.clone()), + run_id: "run-1".to_string(), + token: "token".to_string(), + history_floor_event_id: 3, + records: vec![StagedRecord { + topic: "events".to_string(), + record: record("events", "workflow", 1), + }], + }) + .await + .unwrap(); + let promoted = store + .promote(&StageRef { + chain: Some(chain.clone()), + token: "token".to_string(), + topics: vec!["events".to_string()], + }) + .await + .unwrap(); + assert_eq!(promoted.outcome(), PromoteOutcome::Promoted); + store.flush_notifications().await; + let latest = store + .latest(StoreLatestRequest { + chain: Some(chain.clone()), + topic: "events".to_string(), + }) + .await + .unwrap() + .position; + positions.push(latest); + + let counters: Vec = recorder + .0 + .lock() + .unwrap() + .iter() + .map(|n| n.counter) + .collect(); + let expected: Vec = positions + .iter() + .map(|position| { + let (milliseconds, sequence) = position.split_once('-').expect("an entry id"); + (milliseconds.parse::().unwrap() << 20) | sequence.parse::().unwrap() + }) + .collect(); + assert_eq!(counters, expected); + assert!( + counters.windows(2).all(|pair| pair[0] < pair[1]), + "{counters:?}" + ); +} From b12a967aec0ef4d225766988d34b0c2284070a94 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sat, 10 Oct 2026 23:38:52 -0700 Subject: [PATCH 05/11] Capped a notification flush at 10 seconds. A notifier close retries until the server takes it, so a flush against a server that's down hung the stopping process. --- .../temporal/sdk/core/streams/streams.proto | 4 +- crates/streams/src/service.rs | 95 ++++++++++++++++++- 2 files changed, 96 insertions(+), 3 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 78721f374..1892f94b5 100644 --- a/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto +++ b/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto @@ -24,8 +24,8 @@ service StreamService { rpc Close(CloseRequest) returns (CloseResponse); // Deletes every stream of one owner. rpc DeleteOwner(DeleteOwnerRequest) returns (DeleteOwnerResponse); - // Waits until the stream notifications this process sent are out. Call it before the process - // stops, since a notification is sent in the background. + // Waits until the stream notifications this process sent are out, for at most 10 seconds. Call + // it before the process stops, since a notification is sent in the background. rpc FlushNotifications(FlushNotificationsRequest) returns (FlushNotificationsResponse); } diff --git a/crates/streams/src/service.rs b/crates/streams/src/service.rs index 892fa44d4..124c7c55a 100644 --- a/crates/streams/src/service.rs +++ b/crates/streams/src/service.rs @@ -128,7 +128,10 @@ impl StreamService { .map(|r| r.encode_to_vec()), "FlushNotifications" => { decode::(rpc, request)?; - self.store.flush_notifications().await; + // A close retries until the server takes it, so a stopping process waits only + // so long for one against a server that's down. + let _ = tokio::time::timeout(crate::FLUSH_LIMIT, self.store.flush_notifications()) + .await; Ok(FlushNotificationsResponse {}.encode_to_vec()) } other => Err(StreamError::unsupported(format!( @@ -497,4 +500,94 @@ mod tests { .unwrap_err(); assert_eq!(error.kind, StreamFailureKind::Refused); } + + /// A store whose notifications never go out, as when the server is down and a close retries. + struct NeverFlushes(crate::MemoryStore); + + #[async_trait::async_trait] + impl StreamStore for NeverFlushes { + fn name(&self) -> &str { + self.0.name() + } + async fn append( + &self, + r: crate::proto::StoreAppendRequest, + ) -> StreamResult { + self.0.append(r).await + } + async fn read( + &self, + r: crate::proto::StoreReadRequest, + ) -> StreamResult { + self.0.read(r).await + } + async fn latest( + &self, + r: crate::proto::StoreLatestRequest, + ) -> StreamResult { + self.0.latest(r).await + } + async fn record_at( + &self, + c: &crate::proto::ChainId, + t: &str, + p: &str, + ) -> StreamResult>> { + self.0.record_at(c, t, p).await + } + async fn stage(&self, b: crate::proto::StagedBatch) -> StreamResult<()> { + self.0.stage(b).await + } + async fn promote( + &self, + s: &crate::proto::StageRef, + ) -> StreamResult { + self.0.promote(s).await + } + async fn abort(&self, s: &crate::proto::StageRef) -> StreamResult<()> { + self.0.abort(s).await + } + async fn close_chain(&self, c: &crate::proto::ChainId) -> StreamResult<()> { + self.0.close_chain(c).await + } + async fn close_topic( + &self, + c: &crate::proto::ChainId, + t: &str, + r: Option, + ) -> StreamResult<()> { + self.0.close_topic(c, t, r).await + } + async fn pending_stages( + &self, + c: &crate::proto::ChainId, + ) -> StreamResult> { + self.0.pending_stages(c).await + } + async fn delete_owner( + &self, + r: DeleteOwnerRequest, + ) -> StreamResult { + self.0.delete_owner(r).await + } + async fn flush_notifications(&self) { + std::future::pending::<()>().await; + } + } + + #[tokio::test(start_paused = true)] + async fn a_flush_waits_only_so_long_for_notifications_that_never_go_out() { + let service = StreamService::new( + Arc::new(NeverFlushes(crate::MemoryStore::new())), + Arc::new(Running), + StreamsOptions::default(), + ); + let started = tokio::time::Instant::now(); + flush(&service).await; + assert_eq!( + started.elapsed(), + crate::FLUSH_LIMIT, + "a stopping process mustn't hang on a close retrying against a server that's down" + ); + } } From f24f72bd23257233bc8eae60e140050ed57dcba9 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sun, 11 Oct 2026 00:35:37 -0700 Subject: [PATCH 06/11] Made notifications retry, skip stock servers and leave the report path. A quiet stream kept its reader parked after one failed notification, and each promotion waited for a latest read per topic. A server without the notifier now costs one call. --- crates/streams/src/notify.rs | 405 +++++++++++++++++++++++++++++------ 1 file changed, 341 insertions(+), 64 deletions(-) diff --git a/crates/streams/src/notify.rs b/crates/streams/src/notify.rs index 74d183beb..27a15850d 100644 --- a/crates/streams/src/notify.rs +++ b/crates/streams/src/notify.rs @@ -13,8 +13,12 @@ use crate::{ stream_hash, }; use std::{ + future::Future, num::NonZeroUsize, - sync::{Arc, Mutex}, + sync::{ + Arc, Mutex, + atomic::{AtomicBool, Ordering}, + }, time::Duration, }; use temporalio_common::protos::temporal::api::common::v1::Payload; @@ -26,6 +30,11 @@ pub const DEFAULT_MAX_NOTIFIERS: usize = 1000; /// How long a stopping Worker waits for the notifications still out. pub const FLUSH_LIMIT: Duration = Duration::from_secs(10); +// A stream that goes quiet after a failed notification leaves its reader parked until the next +// write, so the last one is tried again a few times. A newer one replaces the retry. +const NOTIFY_RETRIES: u32 = 2; +const NOTIFY_RETRY_FIRST: Duration = Duration::from_millis(250); + // Redis entry ids are `-`. Twenty bits hold the sequence, far more entries than one // stream takes in a millisecond, and the milliseconds stay below 2^43 until the year 2248, so the // packed value fits a positive i64. @@ -55,6 +64,9 @@ pub enum NotifyError { /// It may take it later. #[error("the notification failed: {0}")] Failed(String), + /// The server has no stream notifier. + #[error("the server has no stream notifier: {0}")] + Unsupported(String), } /// Sends notifications to the stream notifier on the server. @@ -122,10 +134,12 @@ impl StreamKey { struct SlotState { pending: Option<(String, i64)>, sending: bool, + /// Positions still being looked up, which the slot sends once found. + resolving: usize, closed: bool, } -/// One stream's notifications: folded, with one call in flight. +/// One stream's notifications: folded, with one call in flight per slot. #[derive(Debug)] struct Slot { key: StreamKey, @@ -146,6 +160,42 @@ impl Slot { /// 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) { let mut state = self.state.lock().unwrap(); + self.hold(&mut state, client, position, counter); + } + + /// 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 + F: Future> + Send + 'static, + { + { + let mut state = self.state.lock().unwrap(); + if state.closed { + return; + } + state.resolving += 1; + self.idle.send_replace(false); + } + let (slot, client) = (self.clone(), client.clone()); + tokio::spawn(async move { + let found = lookup.await; + let mut state = slot.state.lock().unwrap(); + state.resolving -= 1; + if let Some((position, counter)) = found { + slot.hold(&mut state, &client, position, counter); + } + slot.mark_idle_if_done(&state); + }); + } + + fn hold( + self: &Arc, + state: &mut SlotState, + client: &Arc, + position: String, + counter: i64, + ) { if state.closed { return; } @@ -163,6 +213,12 @@ impl Slot { } } + fn mark_idle_if_done(&self, state: &SlotState) { + if !state.sending && state.resolving == 0 { + self.idle.send_replace(true); + } + } + async fn send_pending(self: Arc, client: Arc) { loop { let (position, counter) = { @@ -171,7 +227,7 @@ impl Slot { Some(next) => next, None => { state.sending = false; - self.idle.send_replace(true); + self.mark_idle_if_done(&state); return; } } @@ -183,19 +239,50 @@ impl Slot { counter, close_result: None, }; - // A notification is only a hint: the next one tells the reader again, and the reader - // reads from its own cursor. - if let Err(error) = client.notify(notification).await { - tracing::warn!( - workflow_id = %self.key.workflow_id, - topic = %self.key.topic, - %error, - "A stream notification failed. The next one tells the reader again" - ); + self.send(&client, notification).await; + } + } + + async fn send(&self, client: &Arc, notification: StreamNotification) { + 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, + Err(NotifyError::Failed(error)) if tries_left > 0 && !self.superseded() => { + tracing::debug!( + workflow_id = %self.key.workflow_id, + topic = %self.key.topic, + %error, + "A stream notification failed; trying again in {delay:?}" + ); + } + Err(error) => { + tracing::warn!( + workflow_id = %self.key.workflow_id, + topic = %self.key.topic, + %error, + "A stream notification failed. The next write tells the reader again" + ); + return; + } + } + tokio::time::sleep(delay).await; + delay *= 2; + if self.superseded() { + return; } } } + /// Whether something newer will reach the notifier anyway: a later notification, or the + /// close. + fn superseded(&self) -> bool { + let state = self.state.lock().unwrap(); + state.pending.is_some() || state.closed + } + /// Stops new notifications and waits for the one in flight, so a close is the last thing the /// notifier hears from this process. async fn close(&self) { @@ -237,7 +324,10 @@ impl Notifier { /// Notifies through `client`, keeping at most `max_notifiers` streams' slots. pub fn new(client: Arc, max_notifiers: usize) -> Self { Self { - client, + client: Arc::new(UnlessUnsupported { + client, + unsupported: AtomicBool::new(false), + }), slots: Mutex::new(lru::LruCache::new( NonZeroUsize::new(max_notifiers).unwrap_or(NonZeroUsize::MIN), )), @@ -252,6 +342,16 @@ impl Notifier { .notify(&self.client, position, counter); } + /// 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) + where + F: Future> + Send + 'static, + { + self.slot(StreamKey::new(chain, topic)) + .notify_found(&self.client, lookup); + } + /// Closes the stream on the server with `result`, after the notification in flight, and /// drops its slot. `counter` must rank above the stream's last notification. pub async fn close( @@ -337,6 +437,35 @@ impl Notifier { } } +/// Stops every call once the server answers that it has no stream notifier. The answer can't +/// change while the process runs, and each call would cost a round trip. +struct UnlessUnsupported { + client: Arc, + unsupported: AtomicBool, +} + +#[async_trait::async_trait] +impl NotifierClient for UnlessUnsupported { + async fn notify(&self, notification: StreamNotification) -> Result<(), NotifyError> { + if self.unsupported.load(Ordering::Relaxed) { + return Ok(()); + } + match self.client.notify(notification).await { + // No caller can be attached to a notifier the server doesn't have. + Err(NotifyError::Unsupported(error)) => { + if !self.unsupported.swap(true, Ordering::Relaxed) { + tracing::debug!( + %error, + "The server has no stream notifier, so this process stops notifying" + ); + } + Ok(()) + } + other => other, + } + } +} + /// A store that notifies each stream's notifier after an append lands and after a Workflow's /// staged output becomes visible. /// @@ -358,56 +487,77 @@ impl NotifyingStore { &self.notifier } - fn cursor(&self, chain: &ChainId, topic: &str, position: &str) -> String { - let hash = stream_hash( - &chain.namespace, - WORKFLOW_OWNER_KIND, - &chain.workflow_id, - topic, - ); - mint_cursor(self.inner.name(), &hash, position) - } - fn notify_at(&self, chain: &ChainId, topic: &str, position: &str) { - // The write landed, so a notification that can't go must not turn it into an error the - // caller might retry with other content. - let Some(counter) = progress_counter(position) else { - tracing::warn!( - workflow_id = %chain.workflow_id, - topic, - position, - "No notifier counter comes from this position, so the write isn't notified" - ); - return; - }; - let cursor = self.cursor(chain, topic, position); - self.notifier.notify(chain, topic, cursor, counter); + if let Some((cursor, counter)) = + notification_at(self.inner.as_ref(), chain, topic, position) + { + self.notifier.notify(chain, topic, cursor, counter); + } } - async fn notify_latest(&self, chain: &ChainId, topic: &str) { - // A promotion has no position of its own, so the newest record's is told. - match self - .inner - .latest(StoreLatestRequest { - chain: Some(chain.clone()), - topic: topic.to_string(), - }) - .await - { - Ok(latest) if !latest.position.is_empty() => { - self.notify_at(chain, topic, &latest.position) + /// 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) { + let inner = self.inner.clone(); + let (found_chain, found_topic) = (chain.clone(), topic.to_string()); + self.notifier.notify_found(chain, topic, async move { + let (chain, topic) = (found_chain, found_topic); + match inner + .latest(StoreLatestRequest { + chain: Some(chain.clone()), + topic: topic.clone(), + }) + .await + { + Ok(latest) if !latest.position.is_empty() => { + notification_at(inner.as_ref(), &chain, &topic, &latest.position) + } + Ok(_) => None, + Err(error) => { + tracing::warn!( + workflow_id = %chain.workflow_id, + topic, + %error, + "Could not find the newest record to notify" + ); + None + } } - Ok(_) => {} - Err(error) => tracing::warn!( - workflow_id = %chain.workflow_id, - topic, - %error, - "Could not find the newest record to notify" - ), - } + }); } } +fn cursor_of(store: &dyn StreamStore, chain: &ChainId, topic: &str, position: &str) -> String { + let hash = stream_hash( + &chain.namespace, + WORKFLOW_OWNER_KIND, + &chain.workflow_id, + topic, + ); + mint_cursor(store.name(), &hash, position) +} + +/// The cursor and counter that tell a reader about the record at `position`. +fn notification_at( + store: &dyn StreamStore, + chain: &ChainId, + topic: &str, + position: &str, +) -> Option<(String, i64)> { + // The write landed, so a notification that can't go must not turn it into an error the + // caller might retry with other content. + let Some(counter) = progress_counter(position) else { + tracing::warn!( + workflow_id = %chain.workflow_id, + topic, + position, + "No notifier counter comes from this position, so the write isn't notified" + ); + return None; + }; + Some((cursor_of(store, chain, topic, position), counter)) +} + #[async_trait::async_trait] impl StreamStore for NotifyingStore { fn name(&self) -> &str { @@ -451,7 +601,7 @@ impl StreamStore for NotifyingStore { && let Some(chain) = &stage.chain { for topic in &stage.topics { - self.notify_latest(chain, topic).await; + self.notify_latest(chain, topic); } } Ok(result) @@ -498,7 +648,7 @@ mod tests { MemoryStore, proto::{StagedRecord, StoreAppendRequest}, }; - use std::sync::atomic::{AtomicBool, Ordering}; + use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use tokio::sync::Semaphore; /// Records what it was sent. While `holding`, each call waits for a permit, so a test @@ -508,6 +658,9 @@ mod tests { holding: AtomicBool, gate: Semaphore, fail: AtomicBool, + /// How many of the next calls fail. + failures: AtomicUsize, + unsupported: AtomicBool, } impl Default for Recorder { @@ -517,6 +670,8 @@ mod tests { holding: AtomicBool::new(false), gate: Semaphore::new(0), fail: AtomicBool::new(false), + failures: AtomicUsize::new(0), + unsupported: AtomicBool::new(false), } } } @@ -549,7 +704,14 @@ mod tests { if self.holding.load(Ordering::SeqCst) { self.gate.acquire().await.unwrap().forget(); } - if self.fail.load(Ordering::SeqCst) { + if self.unsupported.load(Ordering::SeqCst) { + return Err(NotifyError::Unsupported("unimplemented".to_string())); + } + let failing = self + .failures + .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |n| n.checked_sub(1)) + .is_ok(); + if failing || self.fail.load(Ordering::SeqCst) { return Err(NotifyError::Failed("down".to_string())); } Ok(()) @@ -614,14 +776,36 @@ mod tests { ); } - #[tokio::test] - async fn a_failed_notification_leaves_the_next_one_to_tell_the_reader() { + #[tokio::test(start_paused = true)] + async fn a_failed_last_notification_is_tried_twice_more() { let recorder = Arc::new(Recorder::default()); - recorder.fail.store(true, Ordering::SeqCst); + recorder.failures.store(1, Ordering::SeqCst); let notifier = Notifier::new(recorder.clone(), 10); notifier.notify(&chain("run-1"), "t", "c1".to_string(), 1); notifier.flush().await; - recorder.fail.store(false, Ordering::SeqCst); + assert_eq!( + recorder.sent(), + vec![("c1".to_string(), 1), ("c1".to_string(), 1)], + "a quiet stream gets its last notification again" + ); + + recorder.fail.store(true, Ordering::SeqCst); + notifier.notify(&chain("run-1"), "t", "c2".to_string(), 2); + notifier.flush().await; + assert_eq!( + recorder.sent().len(), + 5, + "two more tries, then it's dropped" + ); + } + + #[tokio::test(start_paused = true)] + async fn a_newer_notification_goes_instead_of_a_retry() { + let recorder = Arc::new(Recorder::default()); + recorder.failures.store(1, Ordering::SeqCst); + let notifier = Notifier::new(recorder.clone(), 10); + notifier.notify(&chain("run-1"), "t", "c1".to_string(), 1); + settle().await; notifier.notify(&chain("run-1"), "t", "c2".to_string(), 2); notifier.flush().await; assert_eq!( @@ -630,6 +814,29 @@ mod tests { ); } + #[tokio::test] + async fn a_server_without_the_notifier_costs_one_call() { + let recorder = Arc::new(Recorder::default()); + recorder.unsupported.store(true, Ordering::SeqCst); + let notifier = Notifier::new(recorder.clone(), 10); + notifier.notify(&chain("run-1"), "a", "a1".to_string(), 1); + notifier.flush().await; + notifier.notify(&chain("run-1"), "b", "b1".to_string(), 1); + notifier.notify(&chain("run-2"), "a", "a1".to_string(), 1); + notifier.flush().await; + notifier + .close( + &chain("run-1"), + "a", + "a1".to_string(), + 2, + Payload::default(), + ) + .await + .unwrap(); + assert_eq!(recorder.sent().len(), 1); + } + #[tokio::test] async fn close_waits_for_the_call_in_flight_and_carries_the_result() { let recorder = Recorder::holding(); @@ -892,7 +1099,9 @@ mod tests { .unwrap(); store.flush_notifications().await; let sent = recorder.sent.lock().unwrap().clone(); - let topics: Vec<(&str, i64)> = sent.iter().map(|n| (n.topic.as_str(), n.counter)).collect(); + let mut topics: Vec<(&str, i64)> = + sent.iter().map(|n| (n.topic.as_str(), n.counter)).collect(); + topics.sort(); assert_eq!( topics, vec![("a", 2), ("b", 1)], @@ -916,6 +1125,74 @@ mod tests { ); } + /// A memory store whose `latest` waits for a permit. + struct HeldLatest(MemoryStore, Semaphore); + + #[async_trait::async_trait] + impl StreamStore for HeldLatest { + fn name(&self) -> &str { + self.0.name() + } + async fn append(&self, r: StoreAppendRequest) -> StreamResult { + self.0.append(r).await + } + async fn read(&self, r: StoreReadRequest) -> StreamResult { + self.0.read(r).await + } + async fn latest(&self, r: StoreLatestRequest) -> StreamResult { + self.1.acquire().await.unwrap().forget(); + 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 { + self.0.promote(s).await + } + 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_promotion_returns_before_it_reads_the_newest_positions() { + let recorder = Arc::new(Recorder::default()); + let inner = Arc::new(HeldLatest(MemoryStore::new(), Semaphore::new(0))); + let store = + NotifyingStore::new(inner.clone(), Arc::new(Notifier::new(recorder.clone(), 10))); + let run = chain("run-1"); + store + .stage(staged(&run, "token", &["a", "b"])) + .await + .unwrap(); + tokio::time::timeout( + Duration::from_secs(5), + store.promote(&stage_ref(&run, "token", &["a", "b"])), + ) + .await + .expect("the promotion waited for the newest positions") + .unwrap(); + assert!(recorder.sent().is_empty()); + inner.1.add_permits(2); + store.flush_notifications().await; + assert_eq!(recorder.sent().len(), 2); + } + #[tokio::test] async fn a_closed_streams_notifier_is_dropped() { let recorder = Arc::new(Recorder::default()); From b2fe7fb4378416f14468f5ac6eaaf8475a8a9116 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sun, 11 Oct 2026 00:35:37 -0700 Subject: [PATCH 07/11] Kept retrying a notifier call while its namespace fails over. --- crates/sdk-core/src/streams.rs | 56 ++++++++++++++++++++++++++++++++-- 1 file changed, 53 insertions(+), 3 deletions(-) diff --git a/crates/sdk-core/src/streams.rs b/crates/sdk-core/src/streams.rs index 883a4ca72..32109a7c9 100644 --- a/crates/sdk-core/src/streams.rs +++ b/crates/sdk-core/src/streams.rs @@ -76,8 +76,11 @@ fn notify_request(notification: StreamNotification) -> NotifyStreamRequest { fn notify_error(status: tonic::Status) -> NotifyError { use tonic::Code; match status.code() { - Code::Unimplemented - | Code::InvalidArgument + Code::Unimplemented => NotifyError::Unsupported(status.to_string()), + Code::FailedPrecondition if namespace_in_transition(&status) => { + NotifyError::Failed(status.to_string()) + } + Code::InvalidArgument | Code::NotFound | Code::PermissionDenied | Code::Unauthenticated @@ -86,6 +89,24 @@ fn notify_error(status: tonic::Status) -> NotifyError { } } +/// A namespace that's failing over answers `FailedPrecondition` too, but only until the +/// redirection sees the active cluster, so a close mustn't stop on it. +fn namespace_in_transition(status: &tonic::Status) -> bool { + use temporalio_common::protos::{ + google::rpc::Status as RpcStatus, + temporal::api::errordetails::v1::{ + NamespaceInvalidStateFailure, NamespaceNotActiveFailure, + }, + }; + let Ok(details) = ::decode(status.details()) else { + return false; + }; + details.details.iter().any(|detail| { + detail.to_msg::().is_ok() + || detail.to_msg::().is_ok() + }) +} + /// Asks a Temporal server about owners through a client connection. struct ConnectionOwnerClient { connection: Connection, @@ -296,8 +317,11 @@ mod tests { #[test] fn only_answers_a_retry_cant_change_refuse_a_notification() { + assert!(matches!( + notify_error(tonic::Status::new(tonic::Code::Unimplemented, "")), + NotifyError::Unsupported(_) + )); for code in [ - tonic::Code::Unimplemented, tonic::Code::InvalidArgument, tonic::Code::NotFound, tonic::Code::PermissionDenied, @@ -328,6 +352,32 @@ mod tests { } } + fn failed_precondition_with(detail: &T) -> tonic::Status { + let status = temporalio_common::protos::google::rpc::Status { + code: tonic::Code::FailedPrecondition as i32, + message: "namespace".to_string(), + details: vec![prost_types::Any::from_msg(detail).unwrap()], + }; + tonic::Status::with_details( + tonic::Code::FailedPrecondition, + "namespace", + prost::Message::encode_to_vec(&status).into(), + ) + } + + #[test] + fn a_namespace_in_failover_fails_a_notification_for_now() { + use temporalio_common::protos::temporal::api::errordetails::v1::{ + NamespaceInvalidStateFailure, NamespaceNotActiveFailure, + }; + for status in [ + failed_precondition_with(&NamespaceNotActiveFailure::default()), + failed_precondition_with(&NamespaceInvalidStateFailure::default()), + ] { + assert!(matches!(notify_error(status), NotifyError::Failed(_))); + } + } + fn event(event_id: i64) -> HistoryEvent { HistoryEvent { event_id, From a27b4e993929193eff800810d1765e95bcd9bcf6 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sun, 11 Oct 2026 00:35:43 -0700 Subject: [PATCH 08/11] Added changelog entries for stream notifications. --- .../local/temporal/sdk/core/streams/streams.proto | 8 +++++--- crates/sdk-core/CHANGELOG.md | 11 +++++++++++ 2 files changed, 16 insertions(+), 3 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 1892f94b5..0542a632c 100644 --- a/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto +++ b/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto @@ -24,8 +24,9 @@ service StreamService { rpc Close(CloseRequest) returns (CloseResponse); // Deletes every stream of one owner. rpc DeleteOwner(DeleteOwnerRequest) returns (DeleteOwnerResponse); - // Waits until the stream notifications this process sent are out, for at most 10 seconds. Call - // it before the process stops, since a notification is sent in the background. + // Waits until the stream notifications this process sent are out, for at most 10 seconds. A + // notification goes in the background, so a process that holds only a client calls it before it + // stops. A Worker flushes at stop on its own. rpc FlushNotifications(FlushNotificationsRequest) returns (FlushNotificationsResponse); } @@ -39,7 +40,8 @@ message StreamStoreConfig { // 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. Needs a server with the stream notifier. + // notification costs a call. 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/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index 639bc717f..ee7f6ddaf 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/crates/sdk-core/CHANGELOG.md @@ -67,6 +67,17 @@ relevant information. Worker's `stream_store`, so a Workflow's output and outside producers share it. * Experimental: the C bridge has `temporal_core_stream_store_new`, `_call` and `_free` for the stream service, built with the Redis store, and a `stream_store` field on its worker options. +* Experimental: with `StreamStoreConfig.notify_on_append`, Core tells each stream's notifier on + the server when the stream moves, after every append and every Workflow batch that becomes + visible. That's how a caller of a stream-returning Nexus operation sees progress. Notifications + for one stream fold into one call in flight, a failed last notification is tried twice more, + and a notification never fails the write. `max_notifiers` caps how many streams the process + keeps a notifier for (1000 when zero). +* Experimental: the new `FlushNotifications` stream call waits up to 10 seconds for the + notifications still out. A Worker flushes at stop on its own, so only a process that holds just + a client needs it. +* Experimental: a server without the stream notifier answers the first notifier call with + `UNIMPLEMENTED`. Core then stops notifier calls in that process and logs it once at debug. ### Fixed * Worker heartbeats now report correct task-slot and poller counts when using buffered or custom From 441ea7c9693b3c1e461eb33531bfddd7e3b707f0 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sun, 11 Oct 2026 01:00:39 -0700 Subject: [PATCH 09/11] Forwarded trimmed through the notifying store. The wrapper answered with the trait default, so an empty read from the beginning missed a trim in every process with notifications. The conformance suite and repair now run through it. --- crates/streams/src/handle.rs | 34 ++++++++ crates/streams/src/notify.rs | 7 +- crates/streams/tests/conformance.rs | 119 ++++++++++++++++------------ 3 files changed, 110 insertions(+), 50 deletions(-) diff --git a/crates/streams/src/handle.rs b/crates/streams/src/handle.rs index 11c93318d..0e4a583d4 100644 --- a/crates/streams/src/handle.rs +++ b/crates/streams/src/handle.rs @@ -1624,6 +1624,40 @@ mod tests { assert_eq!(described, ["run-1", "run-2"]); } + /// Takes every notification and sends nothing. + struct Quiet; + + #[async_trait::async_trait] + impl crate::NotifierClient for Quiet { + async fn notify(&self, _: crate::StreamNotification) -> Result<(), crate::NotifyError> { + Ok(()) + } + } + + #[tokio::test] + async fn a_reader_repairs_through_the_notifying_store() { + // Every process with a client wraps its store, so repair must work through the wrapper. + let setup = setup(quick()); + let notifying = Arc::new(crate::NotifyingStore::new( + setup.store.clone(), + Arc::new(crate::Notifier::new(Arc::new(Quiet), 10)), + )); + let streams = Streams::new(notifying, setup.owners.clone(), quick()); + left_behind(&setup.store, "run-1", "t2", 6, "2").await; + left_behind(&setup.store, "run-1", "t1", 3, "1").await; + history( + &setup.owners, + "run-1", + vec![marker(5, "t1", 3), marker(8, "t2", 6)], + ); + let page = streams + .read(read_after("", &[], Duration::ZERO)) + .await + .unwrap(); + assert_eq!(values(&page), ["1", "2"]); + assert!(pending(&setup.store).await.is_empty()); + } + #[tokio::test] async fn rule_24_3_a_stage_history_has_not_decided_holds_back_the_ones_after_it() { let setup = setup(quick()); diff --git a/crates/streams/src/notify.rs b/crates/streams/src/notify.rs index 27a15850d..5cbc760b8 100644 --- a/crates/streams/src/notify.rs +++ b/crates/streams/src/notify.rs @@ -470,7 +470,8 @@ impl NotifierClient for UnlessUnsupported { /// staged output becomes visible. /// /// Wraps the store a process hands to its stream calls and its Workers, so every write path -/// notifies without knowing about notifiers. +/// notifies without knowing about notifiers. It forwards every trait method by name, since a +/// method left to its default would answer for the inner store. pub struct NotifyingStore { inner: Arc, notifier: Arc, @@ -591,6 +592,10 @@ impl StreamStore for NotifyingStore { self.inner.record_at(chain, topic, position).await } + async fn trimmed(&self, chain: &ChainId, topic: &str) -> StreamResult> { + self.inner.trimmed(chain, topic).await + } + async fn stage(&self, batch: StagedBatch) -> StreamResult<()> { self.inner.stage(batch).await } diff --git a/crates/streams/tests/conformance.rs b/crates/streams/tests/conformance.rs index 54a5df747..4a9409686 100644 --- a/crates/streams/tests/conformance.rs +++ b/crates/streams/tests/conformance.rs @@ -15,8 +15,9 @@ use std::{ }; use temporalio_common::protos::temporal::api::common::v1::Payload; use temporalio_streams::{ - BEGINNING, END, MemoryStore, ReadTarget, StreamResult, StreamStore, mint_cursor, proto::*, - read_page, stored_append_record, stored_output_record, stream_hash, + BEGINNING, END, MemoryStore, Notifier, NotifierClient, NotifyError, NotifyingStore, ReadTarget, + StreamNotification, StreamResult, StreamStore, mint_cursor, proto::*, read_page, + stored_append_record, stored_output_record, stream_hash, }; #[async_trait::async_trait] @@ -65,6 +66,49 @@ async fn memory() -> Option { Some(Memory(Arc::new(MemoryStore::new()))) } +/// Takes every notification and sends nothing. +struct Quiet; + +#[async_trait::async_trait] +impl NotifierClient for Quiet { + async fn notify(&self, _: StreamNotification) -> Result<(), NotifyError> { + Ok(()) + } +} + +/// Another case's store behind `NotifyingStore`, which every process with a client runs, so a +/// method the wrapper doesn't forward fails the contract. +struct Notifying { + inner: C, + store: Arc, +} + +async fn notifying( + inner: impl std::future::Future>, +) -> Option> { + let inner = inner.await?; + let store = Arc::new(NotifyingStore::new( + inner.store(), + Arc::new(Notifier::new(Arc::new(Quiet), 10)), + )); + Some(Notifying { inner, store }) +} + +#[async_trait::async_trait] +impl Case for Notifying { + fn store(&self) -> Arc { + self.store.clone() + } + + fn chain(&self) -> ChainId { + self.inner.chain() + } + + async fn drop_oldest(&self, chain: &ChainId, topic: &str, keep: usize) { + self.inner.drop_oldest(chain, topic, keep).await; + } +} + #[cfg(feature = "redis")] mod on_redis { use super::*; @@ -1281,51 +1325,8 @@ macro_rules! conformance { }; } -mod memory { - conformance!(super::memory(); - append_read_roundtrip, - a_batch_lands_in_order_and_names_its_last_record, - an_empty_append_is_refused, - an_append_without_a_digest_is_refused, - latest_is_empty_on_an_empty_topic, - a_cursor_resumes_strictly_after_its_record, - latest_positions_a_reader_at_the_end, - end_reads_only_what_arrives_after_the_read_starts, - a_read_waits_no_longer_than_asked, - dropping_a_waiting_read_releases_it, - beginning_starts_at_the_oldest_record_still_held, - a_reader_that_falls_behind_retention_is_told, - an_empty_read_from_the_beginning_learns_of_a_trim, - an_empty_read_from_the_beginning_with_nothing_trimmed_goes_on, - a_retried_append_returns_the_original_positions, - a_retry_with_other_ciphertext_still_deduplicates, - a_divergent_retry_is_refused, - a_sequence_below_the_newest_is_refused, - rule_9_1_a_batch_inside_the_held_batch_is_refused, - rule_64_1_a_gap_in_the_sequence_is_accepted, - producers_and_attempts_dedupe_apart, - a_new_attempt_supersedes_the_old_one, - a_read_resumed_before_a_new_attempt_reports_it, - a_record_that_does_not_parse_fails_with_its_cursor, - finish_is_a_record_of_its_own, - topics_and_chains_are_apart, - a_cursor_from_another_stream_or_store_is_refused, - a_closed_chain_refuses_new_batches_but_answers_a_repeat, - a_closed_topic_leaves_the_others_open, - a_closed_topic_answers_a_repeat_and_ends_reads, - a_stage_is_invisible_until_promoted, - a_repeated_stage_holds_its_records_once, - an_aborted_stage_never_lands, - a_promotion_naming_too_few_topics_writes_nothing, - a_promotion_lands_on_a_closed_chain, - rule_19_3_a_reset_run_writes_on_and_keeps_the_chain_open, - deleting_an_owner_drops_every_chain_of_it, - ); -} - /// Every case but owner deletes, which need a rule of their own under the documented ACL. -#[cfg(feature = "redis")] -macro_rules! redis_cases { +macro_rules! store_cases { ($case:expr) => { conformance!($case; append_read_roundtrip, @@ -1368,9 +1369,21 @@ macro_rules! redis_cases { }; } +mod memory { + store_cases!(super::memory()); + conformance!(super::memory(); deleting_an_owner_drops_every_chain_of_it); +} + +mod notifying_memory { + store_cases!(super::notifying(super::memory())); + conformance!(super::notifying(super::memory()); + deleting_an_owner_drops_every_chain_of_it, + ); +} + #[cfg(feature = "redis")] mod standalone_redis { - redis_cases!(super::on_redis::redis("STREAMS_REDIS_URL", false)); + store_cases!(super::on_redis::redis("STREAMS_REDIS_URL", false)); conformance!(super::on_redis::redis("STREAMS_REDIS_URL", false); deleting_an_owner_drops_every_chain_of_it, ); @@ -1378,16 +1391,24 @@ mod standalone_redis { #[cfg(feature = "redis")] mod cluster_redis { - redis_cases!(super::on_redis::redis("STREAMS_REDIS_CLUSTER_URL", true)); + store_cases!(super::on_redis::redis("STREAMS_REDIS_CLUSTER_URL", true)); // A Workflow id's chains hash to different slots, so the delete reaches every primary. conformance!(super::on_redis::redis("STREAMS_REDIS_CLUSTER_URL", true); deleting_an_owner_drops_every_chain_of_it, ); } +#[cfg(feature = "redis")] +mod notifying_redis { + store_cases!(super::notifying(super::on_redis::redis( + "STREAMS_REDIS_URL", + false + ))); +} + #[cfg(feature = "redis")] mod documented_acl_redis { - redis_cases!(super::on_redis::redis_as_documented_user(&[])); + store_cases!(super::on_redis::redis_as_documented_user(&[])); // The page names the one rule a delete needs beyond the others. conformance!(super::on_redis::redis_as_documented_user(&["+scan"]); deleting_an_owner_drops_every_chain_of_it, From 75e741f2fc84b4440e04a95d4c40e39a8087452c Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sun, 11 Oct 2026 01:43:39 -0700 Subject: [PATCH 10/11] Forwarded newest_ever and dropped_after through the notifying store. Left to their defaults, an empty read from the beginning missed a trim again in every process with a client. --- crates/streams/src/notify.rs | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/crates/streams/src/notify.rs b/crates/streams/src/notify.rs index 82847400b..e0f536a1f 100644 --- a/crates/streams/src/notify.rs +++ b/crates/streams/src/notify.rs @@ -592,6 +592,19 @@ impl StreamStore for NotifyingStore { self.inner.record_at(chain, topic, position).await } + async fn newest_ever(&self, chain: &ChainId, topic: &str) -> StreamResult { + self.inner.newest_ever(chain, topic).await + } + + async fn dropped_after( + &self, + chain: &ChainId, + topic: &str, + mark: &str, + ) -> StreamResult> { + self.inner.dropped_after(chain, topic, mark).await + } + async fn stage(&self, batch: StagedBatch) -> StreamResult<()> { self.inner.stage(batch).await } From 61cec2b13154ae800c837455ecc76df48e0454c9 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sun, 11 Oct 2026 15:20:40 -0700 Subject: [PATCH 11/11] Tested that a Nexus append notifies through the notifying store. --- crates/streams/src/service.rs | 41 +++++++++++++++++++++++++++++++++++ 1 file changed, 41 insertions(+) diff --git a/crates/streams/src/service.rs b/crates/streams/src/service.rs index 9a8893b9c..d2a49b6f3 100644 --- a/crates/streams/src/service.rs +++ b/crates/streams/src/service.rs @@ -683,6 +683,47 @@ mod tests { ); } + #[tokio::test] + async fn a_nexus_append_notifies_its_streams_notifier() { + use crate::proto::{NexusCall, NexusStep, nexus_step}; + let recorder = Arc::new(Recorder::default()); + let service = service_notifying(true, recorder.clone()).await; + let input = serde_json::json!({ + "stream": {"kind": "workflow", "workflow_id": "wf", "topic": "t"}, + "producer_id": "p", + "attempt": 1, + "sequence": 1, + "finish": true, + }); + let answer = service + .call( + "NexusAppend", + &NexusCall { + input: serde_json::to_vec(&input).unwrap(), + namespace: "ns".to_string(), + ..Default::default() + } + .encode_to_vec(), + ) + .await + .unwrap(); + let step = NexusStep::decode(answer.as_slice()).unwrap(); + let Some(nexus_step::Step::Output(output)) = step.step else { + panic!("a finish needs no codec work, so it answers at once"); + }; + let output: serde_json::Value = serde_json::from_slice(&output).unwrap(); + flush(&service).await; + let sent = recorder.0.lock().unwrap().clone(); + assert_eq!( + sent.len(), + 1, + "the Nexus append goes through the notifying store" + ); + assert_eq!(sent[0].chain.first_run_id, "run-1"); + assert_eq!(sent[0].topic, "t"); + assert_eq!(sent[0].position, output["cursor"].as_str().unwrap()); + } + #[tokio::test] async fn the_nexus_calls_go_through_the_one_call_entry() { use crate::proto::{NexusCall, NexusResumeRequest, NexusStep, nexus_step};