From 907562c31a49c1d2bdceaea497ce55cdab8b5f76 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Wed, 14 Oct 2026 17:05:51 -0700 Subject: [PATCH 1/5] Added a stream driver that reads the notification log. A customer-run driver hands every append's notification to the server, so producers send none. It acks an entry only once the notify call landed, and a sweep finds a lost doorbell ring. --- .../temporal/sdk/core/streams/streams.proto | 19 + crates/sdk-core/CHANGELOG.md | 6 + crates/streams/Cargo.toml | 2 +- crates/streams/docs/redis.md | 4 +- crates/streams/scripts/redis-test-servers.sh | 13 +- crates/streams/src/driver.rs | 478 ++++++++++++++ crates/streams/src/lib.rs | 2 + crates/streams/src/service.rs | 96 ++- crates/streams/src/subscribe.rs | 2 +- crates/streams/tests/driver.rs | 591 ++++++++++++++++++ 10 files changed, 1203 insertions(+), 10 deletions(-) create mode 100644 crates/streams/src/driver.rs create mode 100644 crates/streams/tests/driver.rs 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 67a2bd802f..2ca8ab6a90 100644 --- a/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto +++ b/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto @@ -43,6 +43,10 @@ service StreamService { // wait limit (`streamnotifier.pollMaxWait`). The reader then reads again. After the wait limit, // that read finds records whose notification was lost, or comes back empty again. rpc WaitForNotification(WaitForNotificationRequest) returns (WaitForNotificationResponse); + // Runs a stream driver over the store's notification log until the caller cancels the call. + // It answers only when the server takes no stream notifications, or when the store keeps no + // log. Each driver of one store needs its own consumer name. + rpc RunStreamDriver(RunStreamDriverRequest) returns (RunStreamDriverResponse); // Reads again the exact range an earlier read answered, for a positions-only Workflow reader on // replay, and checks it against that read's digest. Waits for nothing and checks no owner. rpc ReadRange(ReadRangeRequest) returns (ReadResponse); @@ -389,6 +393,21 @@ message WaitForNotificationResponse { temporal.api.failure.v1.Failure close_failure = 6; } +message RunStreamDriverRequest { + // The driver's name in the log's consumer group, unique among the store's drivers. + string consumer = 1; + // How often the driver looks at every slot that holds entries. Unset uses 30 seconds. + google.protobuf.Duration sweep_interval = 2; + // How long another driver's unacked entries wait before this one takes them. Unset uses 60 + // seconds. + google.protobuf.Duration claim_idle = 3; + // The most entries one read takes, and so the most streams one notify call names. Zero uses + // 100. + uint32 batch = 4; +} + +message RunStreamDriverResponse {} + message DeleteOwnerRequest { string namespace = 1; StreamOwnerKind owner_kind = 2; diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index c68427473c..8462f668ff 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/crates/sdk-core/CHANGELOG.md @@ -58,6 +58,12 @@ relevant information. `:{}:n`, and Core rings one of `notify_doorbells` (16 when zero) doorbell streams when that stream takes its first entry. `testdata/redis/notify_slots.json` pins the slot tags for other SDKs. Drivers read as the consumer group `streams-driver`. +* Experimental: the new `RunStreamDriver` stream call runs a stream driver until the caller + cancels it. The driver blocks on each doorbell and sweeps every busy slot every + `sweep_interval` (30 seconds). It folds each drain to the newest counter per stream, sends one + `NotifyStreams` call per namespace, and acks the entries only once the call landed. It takes + entries another driver left unacked for `claim_idle` (60 seconds). Each driver needs its own + `consumer`. `StreamDriver` runs the same in Rust. * Experimental: the new `FlushNotifications` stream call waits up to 10 seconds for the notifications still out. A process calls it before it stops. * Experimental: a server without the stream notifier answers the first notifier call with diff --git a/crates/streams/Cargo.toml b/crates/streams/Cargo.toml index 450eecace0..010105d424 100644 --- a/crates/streams/Cargo.toml +++ b/crates/streams/Cargo.toml @@ -31,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/docs/redis.md b/crates/streams/docs/redis.md index b7f40fdd07..91bdf0c53c 100644 --- a/crates/streams/docs/redis.md +++ b/crates/streams/docs/redis.md @@ -93,7 +93,9 @@ write, and `+xreadgroup +xautoclaim +xack +xdel +xgroup|create +xlen +xadd +clus `+config|get` only reads `maxmemory-policy`, and the store goes on when a server refuses it. **Tests.** The Redis tests run when `STREAMS_REDIS_URL` names a standalone server and -`STREAMS_REDIS_CLUSTER_URL` a cluster seed. `scripts/redis-test-servers.sh` starts both. The +`STREAMS_REDIS_CLUSTER_URL` a cluster seed. `scripts/redis-test-servers.sh` starts both, the +cluster with one replica per primary. `STREAMS_REDIS_CLUSTER_ADMIN=1` lets the driver tests move +a slot and fail a primary over, so set it only for a cluster of the tests' own. The conformance suite runs on each, and once more as a user with exactly the ACL above. The cross-implementation test also needs `STREAMS_BROOK_PY`, a Python SDK checkout whose provider predates the move. diff --git a/crates/streams/scripts/redis-test-servers.sh b/crates/streams/scripts/redis-test-servers.sh index a5d27198e5..70ff4ed7c2 100755 --- a/crates/streams/scripts/redis-test-servers.sh +++ b/crates/streams/scripts/redis-test-servers.sh @@ -1,14 +1,17 @@ #!/usr/bin/env bash # Starts the Redis servers the stream store tests run against: one standalone server, and a -# cluster of three primaries. Uses a local redis-server when there is one, and docker on a Linux -# host otherwise, since a cluster's nodes must reach each other on the addresses they announce. +# cluster of three primaries with one replica each, so the driver tests can fail a primary over. +# Uses a local redis-server when there is one, and docker on a Linux host otherwise, since a +# cluster's nodes must reach each other on the addresses they announce. STREAMS_CLUSTER_BASE moves +# the six cluster ports. # # STREAMS_REDIS_URL=redis://localhost:6384/0 # STREAMS_REDIS_CLUSTER_URL=redis://localhost:7101/0 set -euo pipefail STANDALONE=${STREAMS_STANDALONE_PORT:-6384} -CLUSTER=(7101 7102 7103) +BASE=${STREAMS_CLUSTER_BASE:-7101} +CLUSTER=($BASE $((BASE + 1)) $((BASE + 2)) $((BASE + 3)) $((BASE + 4)) $((BASE + 5))) DATA=${STREAMS_REDIS_DATA:-${TMPDIR:-/tmp}/streams-redis} mkdir -p "$DATA" @@ -33,7 +36,7 @@ for port in "${CLUSTER[@]}"; do start "$port" --cluster-enabled yes --cluster-config-file nodes.conf done for _ in $(seq 50); do - redis-cli -p "${CLUSTER[2]}" ping >/dev/null 2>&1 && break + redis-cli -p "${CLUSTER[5]}" ping >/dev/null 2>&1 && break sleep 0.1 done if ! redis-cli -p "${CLUSTER[0]}" cluster info | grep -q cluster_state:ok; then @@ -41,7 +44,7 @@ if ! redis-cli -p "${CLUSTER[0]}" cluster info | grep -q cluster_state:ok; then for port in "${CLUSTER[@]}"; do nodes+=("127.0.0.1:$port") done - redis-cli --cluster create "${nodes[@]}" --cluster-replicas 0 --cluster-yes >/dev/null + redis-cli --cluster create "${nodes[@]}" --cluster-replicas 1 --cluster-yes >/dev/null for _ in $(seq 50); do redis-cli -p "${CLUSTER[0]}" cluster info | grep -q cluster_state:ok && break sleep 0.1 diff --git a/crates/streams/src/driver.rs b/crates/streams/src/driver.rs new file mode 100644 index 0000000000..4a8a364230 --- /dev/null +++ b/crates/streams/src/driver.rs @@ -0,0 +1,478 @@ +//! The stream driver: reads a store's notification log and tells the server's stream notifiers. +//! +//! A driver blocks on each doorbell, drains each slot a doorbell names, and sweeps every slot +//! that holds entries now and then, since a ring can be lost. It folds a drain's entries to the +//! newest counter of each stream, sends them in one call, and acks them only once the call +//! landed. A failed call leaves them pending, so this driver or another one sends them again. +//! Delivery is at least once, and a notifier ignores a counter it has seen. + +use crate::{ + NotificationEntry, NotificationLog, NotifierClient, NotifyError, StreamError, + StreamNotification, StreamResult, StreamStore, WORKFLOW_OWNER_KIND, mint_cursor, + progress_counter, proto::ChainId, stream_hash, +}; +use std::{ + collections::{BTreeMap, HashMap}, + sync::{Arc, Mutex}, + time::Duration, +}; +use tokio::task::JoinSet; + +/// How a driver reads the log. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct StreamDriverOptions { + /// The driver's name in the log's consumer group. Each driver needs its own. + pub consumer: String, + /// How often the driver looks at every slot that holds entries. + pub sweep_interval: Duration, + /// How long another driver's unacked entries wait before this one takes them. + pub claim_idle: Duration, + /// How long one doorbell read blocks. + pub doorbell_wait: Duration, + /// The most entries one take reads, and so the most streams one notify call names. + pub batch: usize, +} + +impl Default for StreamDriverOptions { + fn default() -> Self { + Self { + consumer: "driver".to_string(), + sweep_interval: Duration::from_secs(30), + claim_idle: Duration::from_secs(60), + doorbell_wait: Duration::from_secs(5), + batch: 100, + } + } +} + +/// Waits before a slot whose notify call failed is drained again. +const RETRY_WAIT: Duration = Duration::from_secs(1); + +/// A driver over one store's notification log. +pub struct StreamDriver { + store: Arc, + log: Arc, + client: Arc, + options: StreamDriverOptions, + /// The slots a drain runs on now, so the doorbells and the sweep don't drain one twice. Set + /// when another task wanted to drain the slot meanwhile, so the drain looks once more. + draining: Mutex>, +} + +impl StreamDriver { + /// A driver for `store`, which must keep a notification log, sending through `client`. + pub fn new( + store: Arc, + client: Arc, + options: StreamDriverOptions, + ) -> StreamResult { + let log = store.notification_log().ok_or_else(|| { + StreamError::refused(format!( + "the {} store keeps no notification log, so it needs no driver", + store.name() + )) + })?; + if options.consumer.is_empty() { + return Err(StreamError::invalid_argument( + "a driver needs a consumer name", + )); + } + Ok(Self { + store, + log, + client, + options, + draining: Mutex::default(), + }) + } + + /// Runs until the returned future is dropped. It only ends early when the server can't take + /// notifications at all. + pub async fn run(self: Arc) -> StreamResult<()> { + let mut tasks = JoinSet::new(); + for doorbell in 0..self.log.doorbells() { + let driver = self.clone(); + tasks.spawn(async move { driver.listen(doorbell).await }); + } + let driver = self.clone(); + tasks.spawn(async move { driver.sweep().await }); + // Dropping the set aborts every task, so a caller stops the driver by dropping this. + match tasks.join_next().await { + Some(Ok(result)) => result, + Some(Err(error)) => Err(StreamError::storage(format!( + "a stream driver task stopped: {error}" + ))), + None => Ok(()), + } + } + + async fn listen(&self, doorbell: u32) -> StreamResult<()> { + loop { + let slots = match self + .log + .wait_doorbell(doorbell, &self.options.consumer, self.options.doorbell_wait) + .await + { + Ok(slots) => slots, + Err(error) => { + tracing::warn!(doorbell, %error, "A stream driver could not read a doorbell"); + tokio::time::sleep(RETRY_WAIT).await; + continue; + } + }; + for slot in slots { + self.drain(slot).await?; + } + } + } + + async fn sweep(&self) -> StreamResult<()> { + loop { + match self.log.busy().await { + Ok(slots) => { + for slot in slots { + self.drain(slot).await?; + } + } + Err(error) => { + tracing::warn!(%error, "A stream driver could not sweep the notification log"); + } + } + tokio::time::sleep(self.options.sweep_interval).await; + } + } + + /// Sends everything `slot` holds, one batch at a time, unless another task drains it now. + /// Fails only when the server can't take notifications at all. + async fn drain(&self, slot: u32) -> StreamResult<()> { + { + let mut draining = self.draining.lock().unwrap(); + if let Some(again) = draining.get_mut(&slot) { + *again = true; + return Ok(()); + } + draining.insert(slot, false); + } + loop { + let drained = self.drain_slot(slot).await; + let mut draining = self.draining.lock().unwrap(); + if drained.is_ok() && draining.get(&slot) == Some(&true) { + draining.insert(slot, false); + continue; + } + draining.remove(&slot); + return drained; + } + } + + async fn drain_slot(&self, slot: u32) -> StreamResult<()> { + loop { + let entries = match self + .log + .take( + slot, + &self.options.consumer, + self.options.claim_idle, + self.options.batch, + ) + .await + { + Ok(entries) if entries.is_empty() => return Ok(()), + Ok(entries) => entries, + Err(error) => { + tracing::warn!(slot, %error, "A stream driver could not read a slot"); + return Ok(()); + } + }; + let ids: Vec = entries.iter().map(|entry| entry.id.clone()).collect(); + match self.client.notify_streams(self.fold(entries)).await { + Ok(()) => { + if let Err(error) = self.log.ack(slot, &ids).await { + // The entries come again, and the notifier drops the repeat. + tracing::warn!(slot, %error, "A stream driver could not ack its entries"); + return Ok(()); + } + } + Err(NotifyError::Unsupported(message)) => { + return Err(StreamError::unsupported(format!( + "the server takes no stream notifications: {message}" + ))); + } + Err(error) => { + // Left pending, so the next drain sends them again. + tracing::warn!(slot, %error, "A stream driver's notify call failed"); + tokio::time::sleep(RETRY_WAIT).await; + return Ok(()); + } + } + } + } + + /// One notification per stream, at the newest counter the entries hold for it. + fn fold(&self, entries: Vec) -> Vec { + let mut newest: BTreeMap<(ChainKey, String), (i64, String)> = BTreeMap::new(); + for entry in entries { + let Some(counter) = progress_counter(&entry.position) else { + tracing::warn!( + workflow_id = %entry.chain.workflow_id, + topic = %entry.topic, + position = %entry.position, + "No notifier counter comes from this position, so it isn't notified" + ); + continue; + }; + let key = (ChainKey::of(&entry.chain), entry.topic); + match newest.get(&key) { + Some((held, _)) if *held >= counter => {} + _ => { + newest.insert(key, (counter, entry.position)); + } + } + } + newest + .into_iter() + .map(|((chain, topic), (counter, position))| { + let chain = chain.into_chain(); + let hash = stream_hash( + &chain.namespace, + WORKFLOW_OWNER_KIND, + &chain.workflow_id, + &topic, + ); + StreamNotification { + position: mint_cursor(self.store.name(), &hash, &position), + chain, + topic, + counter, + close: None, + } + }) + .collect() + } +} + +/// A chain as a map key. +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)] +struct ChainKey(String, String, String); + +impl ChainKey { + fn of(chain: &ChainId) -> Self { + Self( + chain.namespace.clone(), + chain.workflow_id.clone(), + chain.first_run_id.clone(), + ) + } + + fn into_chain(self) -> ChainId { + ChainId { + namespace: self.0, + workflow_id: self.1, + first_run_id: self.2, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::{MemoryStore, proto::StoreAppendRequest}; + use std::sync::atomic::{AtomicBool, Ordering}; + + /// Records each notify call, and fails them while `fail` is set. + #[derive(Default)] + struct Recorder { + calls: Mutex>>, + fail: AtomicBool, + unsupported: AtomicBool, + } + + #[async_trait::async_trait] + impl NotifierClient for Recorder { + async fn notify(&self, _: StreamNotification) -> Result<(), NotifyError> { + unreachable!("a driver sends batches") + } + + async fn notify_streams( + &self, + notifications: Vec, + ) -> Result<(), NotifyError> { + if self.unsupported.load(Ordering::SeqCst) { + return Err(NotifyError::Unsupported("no notifier".to_string())); + } + if self.fail.load(Ordering::SeqCst) { + return Err(NotifyError::Failed("the server is down".to_string())); + } + self.calls.lock().unwrap().push(notifications); + Ok(()) + } + } + + impl Recorder { + /// Each stream's notifications, as `(workflow id, topic, counter)`, in call order. + fn sent(&self) -> Vec<(String, String, i64)> { + self.calls + .lock() + .unwrap() + .iter() + .flatten() + .map(|n| (n.chain.workflow_id.clone(), n.topic.clone(), n.counter)) + .collect() + } + } + + fn chain(workflow_id: &str) -> ChainId { + ChainId { + namespace: "ns".to_string(), + workflow_id: workflow_id.to_string(), + first_run_id: "run-1".to_string(), + } + } + + async fn append(store: &MemoryStore, workflow_id: &str, topic: &str, sequence: i64) { + store + .append(StoreAppendRequest { + chain: Some(chain(workflow_id)), + topic: topic.to_string(), + producer_id: "p".to_string(), + attempt: 1, + sequence, + digest: vec![sequence as u8], + records: vec![b"record".to_vec()], + }) + .await + .unwrap(); + } + + fn driver(store: &Arc, client: &Arc, consumer: &str) -> StreamDriver { + driver_claiming(store, client, consumer, Duration::from_secs(60)) + } + + fn driver_claiming( + store: &Arc, + client: &Arc, + consumer: &str, + claim_idle: Duration, + ) -> StreamDriver { + StreamDriver::new( + store.clone(), + client.clone(), + StreamDriverOptions { + consumer: consumer.to_string(), + claim_idle, + ..Default::default() + }, + ) + .unwrap() + } + + #[tokio::test] + async fn a_drain_sends_each_streams_newest_counter_once_then_acks() { + let store = Arc::new(MemoryStore::with_notification_log()); + let client = Arc::new(Recorder::default()); + for sequence in 1..=3 { + append(&store, "a", "out", sequence).await; + } + append(&store, "b", "out", 1).await; + let driver = driver(&store, &client, "d1"); + driver.drain(0).await.unwrap(); + // Memory positions are offsets, and the counter is the offset plus one. + assert_eq!( + client.sent(), + [ + ("a".to_string(), "out".to_string(), 3), + ("b".to_string(), "out".to_string(), 1) + ] + ); + assert_eq!(client.calls.lock().unwrap().len(), 1, "one call per batch"); + let position = client.calls.lock().unwrap()[0][0].position.clone(); + assert!(position.starts_with("memory:"), "a cursor"); + assert!( + store + .notification_log() + .unwrap() + .busy() + .await + .unwrap() + .is_empty() + ); + } + + #[tokio::test] + async fn a_failed_call_leaves_the_entries_for_the_next_drain() { + let store = Arc::new(MemoryStore::with_notification_log()); + let client = Arc::new(Recorder::default()); + append(&store, "a", "out", 1).await; + let driver = driver(&store, &client, "d1"); + client.fail.store(true, Ordering::SeqCst); + driver.drain(0).await.unwrap(); + assert_eq!(store.notification_log().unwrap().busy().await.unwrap(), [0]); + client.fail.store(false, Ordering::SeqCst); + driver.drain(0).await.unwrap(); + assert_eq!(client.sent(), [("a".to_string(), "out".to_string(), 1)]); + } + + #[tokio::test] + async fn a_stopped_drivers_entries_move_to_another_driver() { + let store = Arc::new(MemoryStore::with_notification_log()); + let client = Arc::new(Recorder::default()); + append(&store, "a", "out", 1).await; + client.fail.store(true, Ordering::SeqCst); + driver(&store, &client, "stopped").drain(0).await.unwrap(); + client.fail.store(false, Ordering::SeqCst); + driver(&store, &client, "other").drain(0).await.unwrap(); + assert!( + client.sent().is_empty(), + "the first driver's hold is still fresh" + ); + driver_claiming(&store, &client, "other", Duration::from_millis(10)) + .drain(0) + .await + .unwrap(); + assert_eq!(client.sent(), [("a".to_string(), "out".to_string(), 1)]); + } + + #[tokio::test] + async fn a_running_driver_wakes_on_the_doorbell() { + let store = Arc::new(MemoryStore::with_notification_log()); + let client = Arc::new(Recorder::default()); + let running = tokio::spawn(Arc::new(driver(&store, &client, "d1")).run()); + append(&store, "a", "out", 1).await; + tokio::time::timeout(Duration::from_secs(5), async { + while client.sent().is_empty() { + tokio::time::sleep(Duration::from_millis(5)).await; + } + }) + .await + .expect("the doorbell woke the driver"); + running.abort(); + } + + #[tokio::test] + async fn a_server_without_notifiers_stops_the_driver() { + let store = Arc::new(MemoryStore::with_notification_log()); + let client = Arc::new(Recorder::default()); + client.unsupported.store(true, Ordering::SeqCst); + append(&store, "a", "out", 1).await; + let error = tokio::time::timeout( + Duration::from_secs(5), + Arc::new(driver(&store, &client, "d1")).run(), + ) + .await + .unwrap() + .unwrap_err(); + assert_eq!(error.kind, crate::proto::StreamFailureKind::Unsupported); + } + + #[test] + fn a_store_without_a_log_needs_no_driver() { + let error = StreamDriver::new( + Arc::new(MemoryStore::new()), + Arc::new(Recorder::default()), + StreamDriverOptions::default(), + ) + .err() + .unwrap(); + assert_eq!(error.kind, crate::proto::StreamFailureKind::Refused); + } +} diff --git a/crates/streams/src/lib.rs b/crates/streams/src/lib.rs index b440c9a991..ea6570e7c1 100644 --- a/crates/streams/src/lib.rs +++ b/crates/streams/src/lib.rs @@ -9,6 +9,7 @@ //! `StreamService`, and Core's Worker appends a Workflow's own output through the same trait. mod cursor; +mod driver; mod error; mod handle; mod memory; @@ -26,6 +27,7 @@ mod subscribe; #[cfg(feature = "redis")] pub use self::redis::{DEFAULT_NOTIFY_DOORBELLS, RedisStore, RedisStoreOptions}; pub use cursor::{BEGINNING, END, cursor_position, mint_cursor, stream_hash}; +pub use driver::{StreamDriver, StreamDriverOptions}; pub use error::{StreamError, StreamResult}; pub use handle::{OWNER_RECHECK, OWNER_START_GRACE, Streams, StreamsOptions}; pub use memory::MemoryStore; diff --git a/crates/streams/src/service.rs b/crates/streams/src/service.rs index e7c6863a3c..18aa0de077 100644 --- a/crates/streams/src/service.rs +++ b/crates/streams/src/service.rs @@ -6,8 +6,9 @@ use crate::{ proto::{ AppendRequest, CloseRequest, DeleteOwnerRequest, FlushNotificationsRequest, FlushNotificationsResponse, LatestRequest, ReadRangeRequest, ReadRequest, - ReleaseFloorRequest, RetireProducerRequest, StreamFailure, StreamStoreConfig, - SubscribeRequest, UnsubscribeRequest, WaitForNotificationRequest, stream_store_config, + ReleaseFloorRequest, RetireProducerRequest, RunStreamDriverRequest, + RunStreamDriverResponse, StreamFailure, StreamStoreConfig, SubscribeRequest, + UnsubscribeRequest, WaitForNotificationRequest, stream_store_config, }, subscribe::Subscriptions, }; @@ -182,6 +183,16 @@ impl StreamService { .retire_producer(decode::(rpc, request)?) .await .map(|r| r.encode_to_vec()), + "RunStreamDriver" => { + let request = decode::(rpc, request)?; + let driver = crate::StreamDriver::new( + self.store.clone(), + self.subscriptions.client()?.clone(), + driver_options(request)?, + )?; + Arc::new(driver).run().await?; + Ok(RunStreamDriverResponse {}.encode_to_vec()) + } "FlushNotifications" => { decode::(rpc, request)?; // A close retries until the server takes it, so a stopping process waits only @@ -235,6 +246,27 @@ fn decode(rpc: &str, request: &[u8]) -> Result StreamResult { + let mut options = crate::StreamDriverOptions { + consumer: request.consumer, + ..Default::default() + }; + if let Some(interval) = request.sweep_interval { + options.sweep_interval = Duration::try_from(interval) + .ok() + .filter(|interval| !interval.is_zero()) + .ok_or_else(|| StreamError::invalid_argument("a sweep interval must be positive"))?; + } + if let Some(idle) = request.claim_idle { + options.claim_idle = Duration::try_from(idle) + .map_err(|_| StreamError::invalid_argument("a claim idle time must not be negative"))?; + } + if request.batch > 0 { + options.batch = request.batch as usize; + } + Ok(options) +} + #[cfg(feature = "redis")] async fn connect_redis( config: crate::proto::RedisStoreConfig, @@ -736,6 +768,66 @@ mod tests { } } + #[tokio::test] + async fn a_driver_runs_as_a_call_until_its_caller_drops_it() { + let recorder = Arc::new(Recorder::default()); + let service = Arc::new( + StreamService::connect_with_notifier( + StreamStoreConfig { + store: Some(stream_store_config::Store::Memory(MemoryStoreConfig {})), + notify_on_append: true, + notification_log: true, + ..Default::default() + }, + Arc::new(Running), + Some(recorder.clone()), + ) + .await + .unwrap(), + ); + let driving = service.clone(); + let driver = tokio::spawn(async move { + driving + .call( + "RunStreamDriver", + &RunStreamDriverRequest { + consumer: "d1".to_string(), + ..Default::default() + } + .encode_to_vec(), + ) + .await + }); + service.call("Append", &append(1, &["a"])).await.unwrap(); + tokio::time::timeout(Duration::from_secs(5), async { + while recorder.0.lock().unwrap().is_empty() { + tokio::time::sleep(Duration::from_millis(5)).await; + } + }) + .await + .expect("the driver notified the append"); + assert_eq!(recorder.0.lock().unwrap()[0].counter, 1); + assert!(!driver.is_finished(), "a driver runs until it is dropped"); + driver.abort(); + } + + #[tokio::test] + async fn a_driver_needs_a_store_with_a_log() { + let service = service_notifying(true, Arc::new(Recorder::default())).await; + let failure = service + .call( + "RunStreamDriver", + &RunStreamDriverRequest { + consumer: "d1".to_string(), + ..Default::default() + } + .encode_to_vec(), + ) + .await + .unwrap_err(); + assert_eq!(failure.kind, StreamFailureKind::Refused as i32); + } + #[tokio::test] async fn notifications_are_off_by_default() { let recorder = Arc::new(Recorder::default()); diff --git a/crates/streams/src/subscribe.rs b/crates/streams/src/subscribe.rs index 464a54a4da..d1f135684c 100644 --- a/crates/streams/src/subscribe.rs +++ b/crates/streams/src/subscribe.rs @@ -96,7 +96,7 @@ impl Subscriptions { } } - fn client(&self) -> StreamResult<&Arc> { + pub(crate) fn client(&self) -> StreamResult<&Arc> { self.client.as_ref().ok_or_else(|| { StreamError::unsupported("a Workflow reader needs a client connected to the server") }) diff --git a/crates/streams/tests/driver.rs b/crates/streams/tests/driver.rs new file mode 100644 index 0000000000..c6d06d070c --- /dev/null +++ b/crates/streams/tests/driver.rs @@ -0,0 +1,591 @@ +//! The stream driver over a Redis notification log, on a standalone server and on a cluster. +//! +//! Set `STREAMS_REDIS_URL` and `STREAMS_REDIS_CLUSTER_URL` to run these. The cluster cases that +//! move slots and fail a primary over need `STREAMS_REDIS_CLUSTER_ADMIN`, since they change the +//! cluster, and replicas for the failover. + +#![cfg(feature = "redis")] + +use std::{ + collections::{BTreeMap, HashMap}, + sync::{ + Arc, Mutex, + atomic::{AtomicBool, AtomicUsize, Ordering}, + }, + time::Duration, +}; +use temporalio_streams::{ + NotifierClient, NotifyError, RedisStore, RedisStoreOptions, StreamDriver, StreamDriverOptions, + StreamNotification, StreamStore, progress_counter, + proto::{ChainId, StoreAppendRequest}, +}; + +static IDS: AtomicUsize = AtomicUsize::new(0); + +/// The cases that move a slot or fail a primary over hold it alone, the others share it, since a +/// cluster change makes appends fail for a moment. +static CLUSTER: tokio::sync::RwLock<()> = tokio::sync::RwLock::const_new(()); + +fn unique(prefix: &str) -> String { + format!( + "{prefix}-{}-{}", + std::process::id(), + IDS.fetch_add(1, Ordering::Relaxed) + ) +} + +/// What a notifier got: each stream's highest counter, and every call. +#[derive(Default)] +struct Notifiers { + newest: Mutex>, + calls: AtomicUsize, + fail: AtomicBool, +} + +#[async_trait::async_trait] +impl NotifierClient for Notifiers { + async fn notify(&self, _: StreamNotification) -> Result<(), NotifyError> { + unreachable!("a driver sends batches") + } + + async fn notify_streams( + &self, + notifications: Vec, + ) -> Result<(), NotifyError> { + if self.fail.load(Ordering::SeqCst) { + return Err(NotifyError::Failed("the server is down".to_string())); + } + self.calls.fetch_add(1, Ordering::SeqCst); + let mut newest = self.newest.lock().unwrap(); + for notification in notifications { + let counter = newest + .entry((notification.chain.workflow_id, notification.topic)) + .or_default(); + *counter = (*counter).max(notification.counter); + } + Ok(()) + } +} + +impl Notifiers { + /// Waits a bounded time until every stream in `expected` reached its counter. + async fn reach(&self, expected: &BTreeMap<(String, String), i64>, within: Duration) { + let reached = tokio::time::timeout(within, async { + loop { + { + let newest = self.newest.lock().unwrap(); + if expected + .iter() + .all(|(stream, counter)| newest.get(stream) >= Some(counter)) + { + return; + } + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + }) + .await; + if reached.is_err() { + let newest = self.newest.lock().unwrap(); + let missing: Vec<_> = expected + .iter() + .filter(|(stream, counter)| newest.get(*stream) < Some(*counter)) + .collect(); + panic!("{} streams were not notified: {missing:?}", missing.len()); + } + } +} + +async fn store(url: &str, cluster: bool, prefix: &str) -> Arc { + let mut options = RedisStoreOptions::new(url); + options.cluster = cluster; + options.key_prefix = prefix.to_string(); + options.notification_log = true; + options.notify_doorbells = 4; + Arc::new(RedisStore::connect(options).await.unwrap()) +} + +fn options(consumer: &str) -> StreamDriverOptions { + StreamDriverOptions { + consumer: consumer.to_string(), + sweep_interval: Duration::from_millis(300), + claim_idle: Duration::from_millis(200), + doorbell_wait: Duration::from_millis(200), + batch: 50, + } +} + +/// Appends one batch to `streams` chains, each in its own slot most likely, and answers each +/// stream's newest counter. +async fn write( + store: &RedisStore, + streams: usize, + sequence: i64, +) -> BTreeMap<(String, String), i64> { + let mut expected = BTreeMap::new(); + for index in 0..streams { + let workflow_id = format!("wf-{index}"); + let landed = store + .append(StoreAppendRequest { + chain: Some(ChainId { + namespace: "ns".to_string(), + workflow_id: workflow_id.clone(), + first_run_id: "run-1".to_string(), + }), + topic: "out".to_string(), + producer_id: "p".to_string(), + attempt: 1, + sequence, + digest: vec![sequence as u8], + records: vec![b"record".to_vec()], + }) + .await + .unwrap(); + expected.insert( + (workflow_id, "out".to_string()), + progress_counter(&landed.last_position).unwrap(), + ); + } + expected +} + +fn url(variable: &str) -> Option { + let url = std::env::var(variable).ok(); + if url.is_none() { + eprintln!("set {variable} to run the stream driver tests"); + } + url +} + +async fn streams_in_every_slot_reach_their_notifiers(url: &str, cluster: bool) { + let _shared = CLUSTER.read().await; + let prefix = unique("driver-spread"); + let store = store(url, cluster, &prefix).await; + let notifiers = Arc::new(Notifiers::default()); + let driver = + Arc::new(StreamDriver::new(store.clone(), notifiers.clone(), options("d1")).unwrap()); + let running = tokio::spawn(driver.run()); + let expected = write(&store, 300, 1).await; + notifiers.reach(&expected, Duration::from_secs(20)).await; + // A second batch moves every counter on. + let expected = write(&store, 300, 2).await; + notifiers.reach(&expected, Duration::from_secs(20)).await; + running.abort(); + let log = store.notification_log().unwrap(); + assert!( + log.busy().await.unwrap().is_empty(), + "every entry was acked" + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn streams_on_a_server_reach_their_notifiers() { + let Some(url) = url("STREAMS_REDIS_URL") else { + return; + }; + streams_in_every_slot_reach_their_notifiers(&url, false).await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn streams_in_every_slot_of_a_cluster_reach_their_notifiers() { + let Some(url) = url("STREAMS_REDIS_CLUSTER_URL") else { + return; + }; + streams_in_every_slot_reach_their_notifiers(&url, true).await; +} + +/// The four doorbells of a store under `prefix`, which needs no percent-encoding. +fn bell_keys(prefix: &str) -> Vec { + let tags: Vec = + serde_json::from_str(include_str!("../testdata/redis/notify_slots.json")).unwrap(); + tags[..4] + .iter() + .map(|tag| format!("{prefix}:{{{tag}}}:bell")) + .collect() +} + +async fn delete(url: &str, cluster: bool, keys: &[String]) { + for key in keys { + if cluster { + let mut conn = redis::cluster::ClusterClient::new(vec![url]) + .unwrap() + .get_async_connection() + .await + .unwrap(); + let _: () = redis::cmd("DEL") + .arg(key) + .query_async(&mut conn) + .await + .unwrap(); + } else { + let mut conn = redis::Client::open(url) + .unwrap() + .get_multiplexed_async_connection() + .await + .unwrap(); + let _: () = redis::cmd("DEL") + .arg(key) + .query_async(&mut conn) + .await + .unwrap(); + } + } +} + +async fn a_lost_ring_is_found_by_the_sweep(url: &str, cluster: bool) { + let _shared = CLUSTER.read().await; + let prefix = unique("driver-sweep"); + let store = store(url, cluster, &prefix).await; + let expected = write(&store, 20, 1).await; + // The rings are gone, as when Core stopped between the append and its ring. + delete(url, cluster, &bell_keys(&prefix)).await; + let notifiers = Arc::new(Notifiers::default()); + let driver = + Arc::new(StreamDriver::new(store.clone(), notifiers.clone(), options("d1")).unwrap()); + let running = tokio::spawn(driver.run()); + // One sweep interval and the time one sweep takes. + notifiers.reach(&expected, Duration::from_secs(5)).await; + running.abort(); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_lost_ring_on_a_server_is_found_by_the_sweep() { + let Some(url) = url("STREAMS_REDIS_URL") else { + return; + }; + a_lost_ring_is_found_by_the_sweep(&url, false).await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_lost_ring_on_a_cluster_is_found_by_the_sweep() { + let Some(url) = url("STREAMS_REDIS_CLUSTER_URL") else { + return; + }; + a_lost_ring_is_found_by_the_sweep(&url, true).await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_stopped_drivers_entries_move_to_another_driver() { + let Some(url) = url("STREAMS_REDIS_CLUSTER_URL") else { + return; + }; + let _shared = CLUSTER.read().await; + let prefix = unique("driver-claim"); + let store = store(&url, true, &prefix).await; + let failing = Arc::new(Notifiers::default()); + failing.fail.store(true, Ordering::SeqCst); + let stopped = tokio::spawn( + Arc::new(StreamDriver::new(store.clone(), failing.clone(), options("stopped")).unwrap()) + .run(), + ); + let expected = write(&store, 30, 1).await; + // The first driver takes the entries and can't send them, then stops. + tokio::time::sleep(Duration::from_millis(500)).await; + stopped.abort(); + let notifiers = Arc::new(Notifiers::default()); + let other = tokio::spawn( + Arc::new(StreamDriver::new(store.clone(), notifiers.clone(), options("other")).unwrap()) + .run(), + ); + notifiers.reach(&expected, Duration::from_secs(10)).await; + other.abort(); +} + +/// The cluster to move slots on and fail over, when the test may change it. +fn admin_url() -> Option { + let admin = std::env::var("STREAMS_REDIS_CLUSTER_ADMIN").ok()?; + url("STREAMS_REDIS_CLUSTER_URL").filter(|_| admin == "1") +} + +/// One node of the cluster, as `CLUSTER NODES` lists it. +#[derive(Debug, Clone)] +struct Node { + id: String, + url: String, + primary_id: Option, +} + +async fn node_list(url: &str) -> Vec { + let mut conn = redis::Client::open(url) + .unwrap() + .get_multiplexed_async_connection() + .await + .unwrap(); + let text: String = redis::cmd("CLUSTER") + .arg("NODES") + .query_async(&mut conn) + .await + .unwrap(); + text.lines() + .map(|line| { + let parts: Vec<&str> = line.split_whitespace().collect(); + let address = parts[1].split('@').next().unwrap(); + Node { + id: parts[0].to_string(), + url: format!("redis://{address}"), + primary_id: parts[2].contains("slave").then(|| parts[3].to_string()), + } + }) + .collect() +} + +async fn on(url: &str, cmd: &redis::Cmd) -> redis::Value { + let mut conn = redis::Client::open(url) + .unwrap() + .get_multiplexed_async_connection() + .await + .unwrap(); + cmd.query_async(&mut conn).await.unwrap() +} + +/// A key's Cluster slot, as Redis computes it from the hash tag. +fn keyslot(key: &str) -> u16 { + let tag = key + .split_once('{') + .and_then(|(_, rest)| rest.split_once('}')) + .map_or(key, |(tag, _)| tag); + crc16::State::::calculate(tag.as_bytes()) % 16384 +} + +/// The primary that serves `slot`. +async fn owner_of(url: &str, slot: u16) -> Node { + let mut conn = redis::Client::open(url) + .unwrap() + .get_multiplexed_async_connection() + .await + .unwrap(); + let ranges: Vec<(u16, u16, (String, u16))> = { + let reply: redis::Value = redis::cmd("CLUSTER") + .arg("SLOTS") + .query_async(&mut conn) + .await + .unwrap(); + let redis::Value::Array(ranges) = reply else { + panic!("{reply:?}") + }; + ranges + .into_iter() + .map(|range| { + let redis::Value::Array(range) = range else { + panic!() + }; + let start: u16 = redis::from_redis_value(range[0].clone()).unwrap(); + let end: u16 = redis::from_redis_value(range[1].clone()).unwrap(); + let redis::Value::Array(node) = range[2].clone() else { + panic!() + }; + let host: String = redis::from_redis_value(node[0].clone()).unwrap(); + let port: u16 = redis::from_redis_value(node[1].clone()).unwrap(); + (start, end, (host, port)) + }) + .collect() + }; + let (_, _, (host, port)) = ranges + .into_iter() + .find(|(start, end, _)| (*start..=*end).contains(&slot)) + .unwrap(); + let url_of = format!("redis://{host}:{port}"); + node_list(url) + .await + .into_iter() + .find(|node| node.url == url_of) + .unwrap() +} + +/// Workflow ids whose chains land in `slot`, found by asking the cluster. +fn ids_in_slot(prefix: &str, slot: u16, count: usize) -> Vec { + let mut found = vec![]; + let mut index = 0; + while found.len() < count { + let workflow_id = format!("wf-{index}"); + if keyslot(&format!("{prefix}:{{ns:{workflow_id}:run-1}}")) == slot { + found.push(workflow_id); + } + index += 1; + } + found +} + +/// Appends to each of `ids` again and again until `stop`, retrying a batch that a slot move +/// refused, and answers each stream's newest counter. +async fn traffic( + store: Arc, + ids: Vec, + stop: Arc, +) -> BTreeMap<(String, String), i64> { + let mut expected = BTreeMap::new(); + let mut sequence = 1; + while !stop.load(Ordering::SeqCst) { + for workflow_id in &ids { + let request = StoreAppendRequest { + chain: Some(ChainId { + namespace: "ns".to_string(), + workflow_id: workflow_id.clone(), + first_run_id: "run-1".to_string(), + }), + topic: "out".to_string(), + producer_id: "p".to_string(), + attempt: 1, + sequence, + digest: vec![(sequence % 251) as u8, (sequence / 251) as u8], + records: vec![b"record".to_vec()], + }; + // The same batch again dedupes, so a retry after an unknown outcome is safe. + let landed = loop { + match store.append(request.clone()).await { + Ok(landed) => break landed, + Err(_) => tokio::time::sleep(Duration::from_millis(20)).await, + } + }; + expected.insert( + (workflow_id.clone(), "out".to_string()), + progress_counter(&landed.last_position).unwrap(), + ); + } + sequence += 1; + tokio::time::sleep(Duration::from_millis(10)).await; + } + expected +} + +/// Moves `slot` from its primary to another one, key by key, as `redis-cli --cluster reshard` +/// does. +async fn move_slot(url: &str, slot: u16) { + let from = owner_of(url, slot).await; + let to = node_list(url) + .await + .into_iter() + .find(|node| node.primary_id.is_none() && node.id != from.id) + .unwrap(); + on( + &to.url, + redis::cmd("CLUSTER") + .arg("SETSLOT") + .arg(slot) + .arg("IMPORTING") + .arg(&from.id), + ) + .await; + on( + &from.url, + redis::cmd("CLUSTER") + .arg("SETSLOT") + .arg(slot) + .arg("MIGRATING") + .arg(&to.id), + ) + .await; + let (host, port) = to + .url + .trim_start_matches("redis://") + .rsplit_once(':') + .unwrap(); + loop { + let keys: Vec = redis::from_redis_value( + on( + &from.url, + redis::cmd("CLUSTER") + .arg("GETKEYSINSLOT") + .arg(slot) + .arg(100), + ) + .await, + ) + .unwrap(); + if keys.is_empty() { + break; + } + on( + &from.url, + redis::cmd("MIGRATE") + .arg(host) + .arg(port) + .arg("") + .arg(0) + .arg(5000) + .arg("KEYS") + .arg(&keys), + ) + .await; + tokio::time::sleep(Duration::from_millis(20)).await; + } + for node in node_list(url).await { + if node.primary_id.is_none() { + on( + &node.url, + redis::cmd("CLUSTER") + .arg("SETSLOT") + .arg(slot) + .arg("NODE") + .arg(&to.id), + ) + .await; + } + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_slot_moved_under_traffic_loses_no_notification() { + let Some(url) = admin_url() else { return }; + let _alone = CLUSTER.write().await; + let prefix = unique("driver-move"); + let store = store(&url, true, &prefix).await; + let slot = keyslot(&format!("{prefix}:{{ns:wf-0:run-1}}")); + let ids = ids_in_slot(&prefix, slot, 5); + let notifiers = Arc::new(Notifiers::default()); + let running = tokio::spawn( + Arc::new(StreamDriver::new(store.clone(), notifiers.clone(), options("d1")).unwrap()).run(), + ); + let stop = Arc::new(AtomicBool::new(false)); + let writer = tokio::spawn(traffic(store.clone(), ids, stop.clone())); + tokio::time::sleep(Duration::from_millis(300)).await; + move_slot(&url, slot).await; + tokio::time::sleep(Duration::from_millis(300)).await; + stop.store(true, Ordering::SeqCst); + let expected = writer.await.unwrap(); + notifiers.reach(&expected, Duration::from_secs(20)).await; + running.abort(); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_primary_failover_under_traffic_still_notifies_every_stream() { + let Some(url) = admin_url() else { return }; + let _alone = CLUSTER.write().await; + let prefix = unique("driver-failover"); + let store = store(&url, true, &prefix).await; + let notifiers = Arc::new(Notifiers::default()); + let running = tokio::spawn( + Arc::new(StreamDriver::new(store.clone(), notifiers.clone(), options("d1")).unwrap()).run(), + ); + let ids: Vec = (0..40).map(|index| format!("wf-{index}")).collect(); + let stop = Arc::new(AtomicBool::new(false)); + let writer = tokio::spawn(traffic(store.clone(), ids, stop.clone())); + tokio::time::sleep(Duration::from_millis(300)).await; + // The replica of the primary that serves slot 0 takes over. + let primary = owner_of(&url, 0).await; + let replica = node_list(&url) + .await + .into_iter() + .find(|node| node.primary_id.as_deref() == Some(primary.id.as_str())) + .expect("each primary has a replica"); + // A manual failover gives up after a few seconds when the replica can't catch up in time, + // so it is asked again. + tokio::time::timeout(Duration::from_secs(60), async { + loop { + on(&replica.url, redis::cmd("CLUSTER").arg("FAILOVER")).await; + for _ in 0..60 { + if owner_of(&url, 0).await.id == replica.id { + return; + } + tokio::time::sleep(Duration::from_millis(100)).await; + } + } + }) + .await + .expect("the replica took over"); + tokio::time::sleep(Duration::from_millis(500)).await; + stop.store(true, Ordering::SeqCst); + let expected = writer.await.unwrap(); + notifiers.reach(&expected, Duration::from_secs(30)).await; + running.abort(); +} From 5aa933010b9bcc587cf4f49980be7a826b7920e1 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Wed, 14 Oct 2026 17:10:56 -0700 Subject: [PATCH 2/5] Read a stream once more after a subscribe or a poll starts its notifier. A notification for a stream with no notifier is dropped, so a record from before the subscribe woke no one. The subscribe answers the newest counter, and a client wait rereads once its poll is out. --- .../temporal/sdk/core/streams/streams.proto | 11 +- crates/sdk-core/CHANGELOG.md | 5 + crates/streams/Cargo.toml | 2 +- crates/streams/src/handle.rs | 22 ++++ crates/streams/src/subscribe.rs | 122 +++++++++++++++++- 5 files changed, 153 insertions(+), 9 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 2ca8ab6a90..df86d7b891 100644 --- a/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto +++ b/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto @@ -40,8 +40,11 @@ service StreamService { rpc Unsubscribe(UnsubscribeRequest) returns (UnsubscribeResponse); // Waits on the stream's notifier, for a client reader whose read came back empty. Returns when // the notifier's counter is above `after_counter`, when the stream closed, or at the server's - // wait limit (`streamnotifier.pollMaxWait`). The reader then reads again. After the wait limit, - // that read finds records whose notification was lost, or comes back empty again. + // wait limit (`streamnotifier.pollMaxWait`). The poll starts the notifier, and Core reads the + // stream once more just after it sends the poll, since a record that landed before the notifier + // existed woke no one. A newer record there answers at once. The reader then reads again. + // After the wait limit, that read finds records whose notification was lost, or comes back + // empty again. rpc WaitForNotification(WaitForNotificationRequest) returns (WaitForNotificationResponse); // Runs a stream driver over the store's notification log until the caller cancels the call. // It answers only when the server takes no stream notifications, or when the store keeps no @@ -358,6 +361,10 @@ message SubscribeResponse { string chain_first_run_id = 1; // Whether this call sent the attach. False when this process attached the reader before. bool attached = 2; + // The notification counter of the stream's newest record, read after the attach landed, zero + // for an empty stream. A record that landed before the notifier existed woke no one, so a + // reader whose last read stopped below this counter reads once more before it sleeps. + int64 newest_counter = 3; } message UnsubscribeRequest { diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index 8462f668ff..beee0237cd 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/crates/sdk-core/CHANGELOG.md @@ -64,6 +64,11 @@ relevant information. `NotifyStreams` call per namespace, and acks the entries only once the call landed. It takes entries another driver left unacked for `claim_idle` (60 seconds). Each driver needs its own `consumer`. `StreamDriver` runs the same in Rust. +* Experimental: a reader reads first and subscribes only when a read comes back empty, since a + notification for a stream with no notifier is dropped. `SubscribeResponse.newest_counter` names + the newest record after the attach landed, so a reader whose read stopped below it reads once + more before it sleeps. `WaitForNotification` reads the stream once more just after it sends the + poll, and answers at once for a record that landed before the notifier existed. * Experimental: the new `FlushNotifications` stream call waits up to 10 seconds for the notifications still out. A process calls it before it stops. * Experimental: a server without the stream notifier answers the first notifier call with diff --git a/crates/streams/Cargo.toml b/crates/streams/Cargo.toml index 010105d424..45281988f5 100644 --- a/crates/streams/Cargo.toml +++ b/crates/streams/Cargo.toml @@ -31,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 = ["rt", "sync", "time"] } +tokio = { version = "1.47", default-features = false, features = ["macros", "rt", "sync", "time"] } tracing = "0.1" [dependencies.temporalio-common] diff --git a/crates/streams/src/handle.rs b/crates/streams/src/handle.rs index 053c8deb26..0a74385d63 100644 --- a/crates/streams/src/handle.rs +++ b/crates/streams/src/handle.rs @@ -844,6 +844,28 @@ impl Streams { fn cursor(&self, stream: &StreamAddress, position: &str) -> String { mint_cursor(self.store.name(), &self.hash(stream), position) } + + /// The notification counter and the cursor of the newest record of `stream` on `chain`, or + /// zero and an empty cursor when it holds none. + pub(crate) async fn newest( + &self, + stream: &StreamAddress, + chain: &ChainId, + ) -> StreamResult<(i64, String)> { + let position = self + .store + .latest(StoreLatestRequest { + chain: Some(chain.clone()), + topic: stream.topic.clone(), + }) + .await? + .position; + if position.is_empty() { + return Ok((0, String::new())); + } + let counter = crate::progress_counter(&position).unwrap_or(0); + Ok((counter, self.cursor(stream, &position))) + } } /// The address of a stream this release keeps, checked. diff --git a/crates/streams/src/subscribe.rs b/crates/streams/src/subscribe.rs index d1f135684c..8326d9b001 100644 --- a/crates/streams/src/subscribe.rs +++ b/crates/streams/src/subscribe.rs @@ -11,11 +11,15 @@ use crate::{ WorkflowAttachment, handle::{Remembered, address}, proto::{ - StreamAddress, SubscribeRequest, SubscribeResponse, UnsubscribeRequest, + ChainId, StreamAddress, SubscribeRequest, SubscribeResponse, UnsubscribeRequest, UnsubscribeResponse, WaitForNotificationRequest, WaitForNotificationResponse, }, }; -use std::sync::Arc; +use std::{sync::Arc, time::Duration}; + +/// How long after it sends a poll Core reads the stream once more. The poll starts the stream's +/// notifier, and a record that lands later notifies it. +const REREAD_AFTER: Duration = Duration::from_millis(200); /// The longest subscription id the server takes. const MAX_SUBSCRIPTION_ID_BYTES: usize = 256; @@ -123,7 +127,12 @@ impl Subscriptions { ); let key = SubscriptionKey::new(stream, &request_id); if let Some(attached) = self.attached.get(&key) { + let chain = ChainId { + first_run_id: attached.first_run_id.clone(), + ..streams.chain(stream).await? + }; return Ok(SubscribeResponse { + newest_counter: streams.newest(stream, &chain).await?.0, chain_first_run_id: attached.first_run_id, attached: false, }); @@ -144,9 +153,12 @@ impl Subscriptions { .await .map_err(subscribe_error)?; self.attached.put(key, &chain.first_run_id, OWNER_RECHECK); + // Read after the attach landed, so a record after this one notifies the reader. + let newest_counter = streams.newest(stream, &chain).await?.0; Ok(SubscribeResponse { chain_first_run_id: chain.first_run_id, attached: true, + newest_counter, }) } @@ -193,10 +205,28 @@ impl Subscriptions { let stream = address(request.stream.as_ref())?; let client = self.client()?; let chain = streams.chain(stream).await?; - let polled = client - .poll(chain, stream.topic.clone(), request.after_counter) - .await - .map_err(subscribe_error)?; + let poll = client.poll(chain.clone(), stream.topic.clone(), request.after_counter); + tokio::pin!(poll); + // The poll starts the notifier, so a record from before it woke no one. One more read + // once the poll is out finds it. + let reread = async { + tokio::time::sleep(REREAD_AFTER).await; + streams.newest(stream, &chain).await + }; + let polled = tokio::select! { + polled = &mut poll => polled, + newest = reread => match newest { + Ok((counter, position)) if counter > request.after_counter => { + return Ok(WaitForNotificationResponse { + counter, + position, + ..Default::default() + }); + } + _ => poll.await, + }, + } + .map_err(subscribe_error)?; let Some(polled) = polled else { return Ok(WaitForNotificationResponse { timed_out: true, @@ -254,6 +284,8 @@ mod tests { answer: Mutex>, polled: Mutex>, poll_answer: Mutex, NotifyError>>>, + /// How long the poll takes before it answers. + poll_delay: Mutex, } #[async_trait::async_trait] @@ -293,6 +325,8 @@ mod tests { .lock() .unwrap() .push((chain, topic, after_counter)); + let delay = *self.poll_delay.lock().unwrap(); + tokio::time::sleep(delay).await; self.poll_answer .lock() .unwrap() @@ -596,6 +630,82 @@ mod tests { .map(|answer| WaitForNotificationResponse::decode(answer.as_slice()).unwrap()) } + /// Appends `count` records to the chain the owner names now. + async fn append_records(setup: &Setup, count: i64) { + use crate::StreamStore; + let first_run_id = setup.owner.0.lock().unwrap().clone(); + setup + .store + .append(crate::proto::StoreAppendRequest { + chain: Some(ChainId { + namespace: "ns".to_string(), + workflow_id: "owner".to_string(), + first_run_id, + }), + topic: "tokens".to_string(), + producer_id: "p".to_string(), + attempt: 1, + sequence: 1, + digest: vec![1], + records: (0..count).map(|_| b"record".to_vec()).collect(), + }) + .await + .unwrap(); + } + + #[tokio::test] + async fn a_subscription_names_the_newest_counter_it_found_after_the_attach() { + let setup = setup(); + let empty = subscribe(&setup.service, subscribe_request("1")) + .await + .unwrap(); + assert_eq!(empty.newest_counter, 0, "an empty stream"); + append_records(&setup, 2).await; + // A record from before the attach woke no one, so the reader reads once more. + let again = subscribe(&setup.service, subscribe_request("1")) + .await + .unwrap(); + assert!(!again.attached); + assert_eq!(again.newest_counter, 2); + let fresh = subscribe(&setup.service, subscribe_request("2")) + .await + .unwrap(); + assert!(fresh.attached); + assert_eq!(fresh.newest_counter, 2); + } + + #[tokio::test] + async fn a_wait_answers_at_once_for_a_record_from_before_the_poll() { + let setup = setup(); + append_records(&setup, 1).await; + *setup.server.poll_delay.lock().unwrap() = Duration::from_secs(30); + let started = tokio::time::Instant::now(); + let answer = wait(&setup, 0, Ok(None)).await.unwrap(); + assert!(started.elapsed() < Duration::from_secs(5)); + assert!(!answer.timed_out && !answer.closed); + assert_eq!(answer.counter, 1); + assert!(answer.position.starts_with("memory:"), "{answer:?}"); + } + + #[tokio::test] + async fn a_wait_with_nothing_newer_waits_for_the_poll() { + let setup = setup(); + append_records(&setup, 1).await; + *setup.server.poll_delay.lock().unwrap() = Duration::from_millis(400); + let answer = wait( + &setup, + 1, + Ok(Some(NotifierPoll { + counter: 2, + position: "cursor-2".to_string(), + ..Default::default() + })), + ) + .await + .unwrap(); + assert_eq!((answer.counter, answer.position.as_str()), (2, "cursor-2")); + } + #[tokio::test] async fn a_wait_asks_the_notifier_of_the_streams_chain_for_a_counter_above_the_read() { let setup = setup(); From a344566fc9d010e6b4866c36684c06348f9bedbd Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 16 Oct 2026 04:50:22 -0700 Subject: [PATCH 3/5] Named a driver after its host and process by default. Two drivers left at the default name `driver` took each other's entries. --- crates/streams/Cargo.toml | 1 + crates/streams/src/driver.rs | 18 ++++++++++++++++-- 2 files changed, 17 insertions(+), 2 deletions(-) diff --git a/crates/streams/Cargo.toml b/crates/streams/Cargo.toml index 010105d424..2961175cfa 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 } +gethostname = "1.0.2" lru = "0.18" prost = { workspace = true } redis = { version = "1.7", optional = true, default-features = false, features = [ diff --git a/crates/streams/src/driver.rs b/crates/streams/src/driver.rs index 4a8a364230..37471205a2 100644 --- a/crates/streams/src/driver.rs +++ b/crates/streams/src/driver.rs @@ -21,7 +21,8 @@ use tokio::task::JoinSet; /// How a driver reads the log. #[derive(Debug, Clone, PartialEq, Eq)] pub struct StreamDriverOptions { - /// The driver's name in the log's consumer group. Each driver needs its own. + /// The driver's name in the log's consumer group. Each driver needs its own, and the default + /// is the host name and the process id. pub consumer: String, /// How often the driver looks at every slot that holds entries. pub sweep_interval: Duration, @@ -36,7 +37,11 @@ pub struct StreamDriverOptions { impl Default for StreamDriverOptions { fn default() -> Self { Self { - consumer: "driver".to_string(), + consumer: format!( + "{}-{}", + gethostname::gethostname().to_string_lossy(), + std::process::id() + ), sweep_interval: Duration::from_secs(30), claim_idle: Duration::from_secs(60), doorbell_wait: Duration::from_secs(5), @@ -464,6 +469,15 @@ mod tests { assert_eq!(error.kind, crate::proto::StreamFailureKind::Unsupported); } + #[test] + fn the_default_consumer_names_the_host_and_the_process() { + let consumer = StreamDriverOptions::default().consumer; + assert!( + consumer.ends_with(&format!("-{}", std::process::id())), + "{consumer}" + ); + } + #[test] fn a_store_without_a_log_needs_no_driver() { let error = StreamDriver::new( From 37cc51c8685894b23732a4e1625911408a917583 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 16 Oct 2026 04:50:22 -0700 Subject: [PATCH 4/5] Kept one refused notification from stopping a driver slot. The driver sends one call per namespace, and a refused call again per stream. It acks what stays refused, and an entry that keeps failing while others land is dropped after ten tries. One namespace with no notifiers no longer stops the driver. --- crates/streams/src/driver.rs | 359 +++++++++++++++++++++++++++++++---- 1 file changed, 324 insertions(+), 35 deletions(-) diff --git a/crates/streams/src/driver.rs b/crates/streams/src/driver.rs index 37471205a2..95894a41c4 100644 --- a/crates/streams/src/driver.rs +++ b/crates/streams/src/driver.rs @@ -2,9 +2,10 @@ //! //! A driver blocks on each doorbell, drains each slot a doorbell names, and sweeps every slot //! that holds entries now and then, since a ring can be lost. It folds a drain's entries to the -//! newest counter of each stream, sends them in one call, and acks them only once the call -//! landed. A failed call leaves them pending, so this driver or another one sends them again. -//! Delivery is at least once, and a notifier ignores a counter it has seen. +//! newest counter of each stream, sends them in one call per namespace, and acks them only once +//! the call landed. A failed call leaves them pending, so this driver or another one sends them +//! again. Delivery is at least once, and a notifier ignores a counter it has seen. A call the +//! server refuses goes again one stream at a time, so one bad entry holds back no other stream. use crate::{ NotificationEntry, NotificationLog, NotifierClient, NotifyError, StreamError, @@ -13,7 +14,10 @@ use crate::{ }; use std::{ collections::{BTreeMap, HashMap}, - sync::{Arc, Mutex}, + sync::{ + Arc, Mutex, + atomic::{AtomicBool, Ordering}, + }, time::Duration, }; use tokio::task::JoinSet; @@ -53,6 +57,30 @@ impl Default for StreamDriverOptions { /// Waits before a slot whose notify call failed is drained again. const RETRY_WAIT: Duration = Duration::from_secs(1); +/// How many times one entry's call may fail while the server takes other streams' calls before +/// the entry is dropped. +const MAX_ENTRY_FAILURES: u32 = 10; + +/// How many entries' failures a driver counts at most. +const MAX_TRACKED_FAILURES: usize = 10_000; + +/// What one drain did with its entries. +#[derive(Default)] +struct Sent { + /// The entries to ack, since they landed or never will. + acked: Vec, + /// Some entries are left for the next drain. + left_pending: bool, + /// A namespace answered that the server takes no notifications. + unsupported: bool, +} + +impl Sent { + fn ack(&mut self, ids: impl IntoIterator) { + self.acked.extend(ids); + } +} + /// A driver over one store's notification log. pub struct StreamDriver { store: Arc, @@ -62,6 +90,11 @@ pub struct StreamDriver { /// The slots a drain runs on now, so the doorbells and the sweep don't drain one twice. Set /// when another task wanted to drain the slot meanwhile, so the drain looks once more. draining: Mutex>, + /// Whether any call landed, so a server with no notifiers at all is told from one namespace + /// without them. + delivered: AtomicBool, + /// How often each entry's call failed while the server took other calls. + failures: Mutex>, } impl StreamDriver { @@ -88,11 +121,13 @@ impl StreamDriver { client, options, draining: Mutex::default(), + delivered: AtomicBool::new(false), + failures: Mutex::default(), }) } /// Runs until the returned future is dropped. It only ends early when the server can't take - /// notifications at all. + /// notifications at all: every namespace it sent for answered so, and no call ever landed. pub async fn run(self: Arc) -> StreamResult<()> { let mut tasks = JoinSet::new(); for doorbell in 0..self.log.doorbells() { @@ -189,33 +224,179 @@ impl StreamDriver { return Ok(()); } }; - let ids: Vec = entries.iter().map(|entry| entry.id.clone()).collect(); - match self.client.notify_streams(self.fold(entries)).await { - Ok(()) => { - if let Err(error) = self.log.ack(slot, &ids).await { - // The entries come again, and the notifier drops the repeat. - tracing::warn!(slot, %error, "A stream driver could not ack its entries"); - return Ok(()); + // A server call names one namespace, and one namespace's refusal must not hold back + // the others in the slot. + let mut by_namespace: BTreeMap> = BTreeMap::new(); + for entry in entries { + by_namespace + .entry(entry.chain.namespace.clone()) + .or_default() + .push(entry); + } + let mut sent = Sent::default(); + for (namespace, entries) in by_namespace { + self.send_namespace(slot, &namespace, entries, &mut sent) + .await; + } + if !sent.acked.is_empty() { + if let Err(error) = self.log.ack(slot, &sent.acked).await { + // The entries come again, and the notifier drops the repeat. + tracing::warn!(slot, %error, "A stream driver could not ack its entries"); + return Ok(()); + } + let mut failures = self.failures.lock().unwrap(); + for id in &sent.acked { + failures.remove(id); + } + } + if sent.unsupported && !self.delivered.load(Ordering::Relaxed) { + return Err(StreamError::unsupported( + "the server takes no stream notifications", + )); + } + if sent.left_pending { + // Left pending, so the next drain sends them again. + tokio::time::sleep(RETRY_WAIT).await; + return Ok(()); + } + } + } + + /// Sends one namespace's entries in one call. When the server refuses the call, each stream + /// goes on its own, so one bad entry holds back no other stream. + async fn send_namespace( + &self, + slot: u32, + namespace: &str, + entries: Vec, + sent: &mut Sent, + ) { + let (streams, unsendable): (Vec<_>, Vec<_>) = self + .fold(entries) + .into_iter() + .partition(|(notification, _)| notification.counter > 0); + sent.ack(unsendable.into_iter().flat_map(|(_, ids)| ids)); + if streams.is_empty() { + return; + } + let notifications = streams.iter().map(|(n, _)| n.clone()).collect(); + match self.client.notify_streams(notifications).await { + Ok(()) => { + self.delivered.store(true, Ordering::Relaxed); + sent.ack(streams.into_iter().flat_map(|(_, ids)| ids)); + } + Err(NotifyError::Unsupported(error)) => { + // A server with no notifiers answers this for every namespace. Only one namespace + // answering it, the others still get theirs. + tracing::warn!( + slot, + namespace, + error, + "A namespace takes no stream notifications" + ); + sent.unsupported = true; + sent.ack(streams.into_iter().flat_map(|(_, ids)| ids)); + } + Err(error) if streams.len() == 1 => { + self.settle_one(slot, error, streams, sent, false); + } + Err(_) => { + let mut reached = false; + let mut failed = vec![]; + for (notification, ids) in streams { + match self.client.notify_streams(vec![notification.clone()]).await { + Ok(()) => { + self.delivered.store(true, Ordering::Relaxed); + reached = true; + sent.ack(ids); + } + Err(error) => failed.push((error, notification, ids)), } } - Err(NotifyError::Unsupported(message)) => { - return Err(StreamError::unsupported(format!( - "the server takes no stream notifications: {message}" - ))); + for (error, notification, ids) in failed { + self.settle_one(slot, error, vec![(notification, ids)], sent, reached); } - Err(error) => { - // Left pending, so the next drain sends them again. - tracing::warn!(slot, %error, "A stream driver's notify call failed"); - tokio::time::sleep(RETRY_WAIT).await; - return Ok(()); + } + } + } + + /// What becomes of one stream's entries whose call failed. A refusal won't change on a retry, + /// so it is logged and acked. A failure is tried again, but only [MAX_ENTRY_FAILURES] times + /// while the server takes other streams' calls, since then the server is up and this entry + /// is the one it can't take. + fn settle_one( + &self, + slot: u32, + error: NotifyError, + streams: Vec<(StreamNotification, Vec)>, + sent: &mut Sent, + server_reached: bool, + ) { + for (notification, ids) in streams { + let workflow_id = notification.chain.workflow_id.as_str(); + let topic = notification.topic.as_str(); + match &error { + NotifyError::Unsupported(message) => { + tracing::warn!(slot, workflow_id, topic, message, "Dropped a notification"); + sent.unsupported = true; + sent.ack(ids); + } + NotifyError::Refused(message) => { + tracing::warn!( + slot, + workflow_id, + topic, + message, + "The server refused a stream notification, so it is dropped" + ); + sent.ack(ids); + } + NotifyError::Failed(message) => { + let failures = if server_reached { + self.failed_again(&ids) + } else { + 0 + }; + if failures >= MAX_ENTRY_FAILURES { + tracing::warn!( + slot, + workflow_id, + topic, + message, + "A stream notification failed {failures} times while others landed, \ + so it is dropped" + ); + sent.ack(ids); + } else { + tracing::warn!(slot, workflow_id, topic, message, "A notify call failed"); + sent.left_pending = true; + } } } } } - /// One notification per stream, at the newest counter the entries hold for it. - fn fold(&self, entries: Vec) -> Vec { - let mut newest: BTreeMap<(ChainKey, String), (i64, String)> = BTreeMap::new(); + /// Counts one more failure for each entry, and answers the highest count. + fn failed_again(&self, ids: &[String]) -> u32 { + let mut failures = self.failures.lock().unwrap(); + // Entries another driver acked since are never seen again, so the map is bounded. + if failures.len() > MAX_TRACKED_FAILURES { + failures.clear(); + } + ids.iter() + .map(|id| { + let count = failures.entry(id.clone()).or_default(); + *count += 1; + *count + }) + .max() + .unwrap_or(0) + } + + /// One notification per stream, at the newest counter the entries hold for it, with the ids + /// of every entry it stands for. + fn fold(&self, entries: Vec) -> Vec<(StreamNotification, Vec)> { + let mut newest: BTreeMap<(ChainKey, String), (i64, String, Vec)> = BTreeMap::new(); for entry in entries { let Some(counter) = progress_counter(&entry.position) else { tracing::warn!( @@ -224,19 +405,24 @@ impl StreamDriver { position = %entry.position, "No notifier counter comes from this position, so it isn't notified" ); + // No call will take it, so it is acked with the stream's other entries, if any. + newest + .entry((ChainKey::of(&entry.chain), entry.topic)) + .or_insert((0, String::new(), vec![])) + .2 + .push(entry.id); continue; }; let key = (ChainKey::of(&entry.chain), entry.topic); - match newest.get(&key) { - Some((held, _)) if *held >= counter => {} - _ => { - newest.insert(key, (counter, entry.position)); - } + let held = newest.entry(key).or_insert((0, String::new(), vec![])); + held.2.push(entry.id); + if held.0 < counter { + (held.0, held.1) = (counter, entry.position); } } newest .into_iter() - .map(|((chain, topic), (counter, position))| { + .map(|((chain, topic), (counter, position, ids))| { let chain = chain.into_chain(); let hash = stream_hash( &chain.namespace, @@ -244,13 +430,14 @@ impl StreamDriver { &chain.workflow_id, &topic, ); - StreamNotification { + let notification = StreamNotification { position: mint_cursor(self.store.name(), &hash, &position), chain, topic, counter, close: None, - } + }; + (notification, ids) }) .collect() } @@ -290,6 +477,12 @@ mod tests { calls: Mutex>>, fail: AtomicBool, unsupported: AtomicBool, + /// A call that names this Workflow id is refused whole, as the server does. + refuse_workflow: Mutex>, + /// A call that names this Workflow id fails whole. + fail_workflow: Mutex>, + /// A call for this namespace answers that it takes no notifications. + unsupported_namespace: Mutex>, } #[async_trait::async_trait] @@ -308,6 +501,21 @@ mod tests { if self.fail.load(Ordering::SeqCst) { return Err(NotifyError::Failed("the server is down".to_string())); } + let names = |field: &Mutex>, of: fn(&StreamNotification) -> &str| { + let field = field.lock().unwrap(); + notifications + .iter() + .any(|n| field.as_deref() == Some(of(n))) + }; + if names(&self.unsupported_namespace, |n| &n.chain.namespace) { + return Err(NotifyError::Unsupported("off here".to_string())); + } + if names(&self.refuse_workflow, |n| &n.chain.workflow_id) { + return Err(NotifyError::Refused("no such owner".to_string())); + } + if names(&self.fail_workflow, |n| &n.chain.workflow_id) { + return Err(NotifyError::Failed("internal".to_string())); + } self.calls.lock().unwrap().push(notifications); Ok(()) } @@ -326,18 +534,28 @@ mod tests { } } - fn chain(workflow_id: &str) -> ChainId { + fn chain_in(namespace: &str, workflow_id: &str) -> ChainId { ChainId { - namespace: "ns".to_string(), + namespace: namespace.to_string(), workflow_id: workflow_id.to_string(), first_run_id: "run-1".to_string(), } } async fn append(store: &MemoryStore, workflow_id: &str, topic: &str, sequence: i64) { + append_in(store, "ns", workflow_id, topic, sequence).await; + } + + async fn append_in( + store: &MemoryStore, + namespace: &str, + workflow_id: &str, + topic: &str, + sequence: i64, + ) { store .append(StoreAppendRequest { - chain: Some(chain(workflow_id)), + chain: Some(chain_in(namespace, workflow_id)), topic: topic.to_string(), producer_id: "p".to_string(), attempt: 1, @@ -469,6 +687,77 @@ mod tests { assert_eq!(error.kind, crate::proto::StreamFailureKind::Unsupported); } + async fn log_is_empty(store: &MemoryStore) -> bool { + store + .notification_log() + .unwrap() + .busy() + .await + .unwrap() + .is_empty() + } + + #[tokio::test] + async fn one_refused_entry_holds_back_no_other_stream() { + let store = Arc::new(MemoryStore::with_notification_log()); + let client = Arc::new(Recorder::default()); + *client.refuse_workflow.lock().unwrap() = Some("bad".to_string()); + append(&store, "bad", "out", 1).await; + append(&store, "good", "out", 1).await; + driver(&store, &client, "d1").drain(0).await.unwrap(); + assert_eq!(client.sent(), [("good".to_string(), "out".to_string(), 1)]); + assert!(log_is_empty(&store).await, "the refused entry is acked too"); + } + + #[tokio::test] + async fn a_namespace_without_notifiers_stops_neither_the_driver_nor_the_others() { + let store = Arc::new(MemoryStore::with_notification_log()); + let client = Arc::new(Recorder::default()); + *client.unsupported_namespace.lock().unwrap() = Some("off".to_string()); + append_in(&store, "off", "a", "out", 1).await; + append_in(&store, "on", "b", "out", 1).await; + driver(&store, &client, "d1").drain(0).await.unwrap(); + assert_eq!(client.sent(), [("b".to_string(), "out".to_string(), 1)]); + assert!(log_is_empty(&store).await); + } + + #[tokio::test(start_paused = true)] + async fn an_entry_that_keeps_failing_while_others_land_is_dropped() { + let store = Arc::new(MemoryStore::with_notification_log()); + let client = Arc::new(Recorder::default()); + *client.fail_workflow.lock().unwrap() = Some("bad".to_string()); + append(&store, "bad", "out", 1).await; + let driver = driver(&store, &client, "d1"); + for sequence in 1..=MAX_ENTRY_FAILURES as i64 { + assert!(!log_is_empty(&store).await, "kept after {sequence} drains"); + append(&store, "good", "out", sequence).await; + driver.drain(0).await.unwrap(); + } + assert!(log_is_empty(&store).await); + assert_eq!( + client.sent().last(), + Some(&( + "good".to_string(), + "out".to_string(), + MAX_ENTRY_FAILURES as i64 + )) + ); + } + + #[tokio::test(start_paused = true)] + async fn a_failing_server_drops_nothing() { + let store = Arc::new(MemoryStore::with_notification_log()); + let client = Arc::new(Recorder::default()); + append(&store, "a", "out", 1).await; + append(&store, "b", "out", 1).await; + client.fail.store(true, Ordering::SeqCst); + let driver = driver(&store, &client, "d1"); + for _ in 0..=MAX_ENTRY_FAILURES { + driver.drain(0).await.unwrap(); + } + assert!(!log_is_empty(&store).await); + } + #[test] fn the_default_consumer_names_the_host_and_the_process() { let consumer = StreamDriverOptions::default().consumer; From 16c98bc222b3da3050a139ee253bcae71a579d39 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 16 Oct 2026 08:30:16 -0700 Subject: [PATCH 5/5] Settled each `NotifyStreams` entry by the server's answer for it. A refused entry is logged and acked, and a failed one stays for the next drain. It counts toward its drop only while other entries land. --- crates/sdk-core/src/streams.rs | 34 ++++++++++-- crates/streams/src/driver.rs | 99 +++++++++++++++++++++++++++++++--- crates/streams/src/notify.rs | 12 +++++ 3 files changed, 136 insertions(+), 9 deletions(-) diff --git a/crates/sdk-core/src/streams.rs b/crates/sdk-core/src/streams.rs index cd6c6775ea..033f6f099a 100644 --- a/crates/sdk-core/src/streams.rs +++ b/crates/sdk-core/src/streams.rs @@ -59,14 +59,42 @@ impl NotifierClient for ConnectionNotifierClient { &self, notifications: Vec, ) -> Result<(), NotifyError> { + self.notify_streams_each(notifications) + .await? + .into_iter() + .collect() + } + + async fn notify_streams_each( + &self, + notifications: Vec, + ) -> Result>, NotifyError> { + let namespaces: Vec<_> = notifications + .iter() + .map(|n| n.chain.namespace.clone()) + .collect(); + let mut results: Vec<_> = namespaces.iter().map(|_| Ok(())).collect(); for request in notify_streams_requests(notifications, self.connection.identity()) { - self.connection + // A request keeps the order of its namespace's notifications. + let indexes: Vec<_> = (0..namespaces.len()) + .filter(|&index| namespaces[index] == request.namespace) + .collect(); + let response = self + .connection .workflow_service() .notify_streams(request.into_request()) .await - .map_err(notify_error)?; + .map_err(notify_error)? + .into_inner(); + // A server that answers no results applied every entry. + for (index, result) in indexes.into_iter().zip(response.results) { + if result.code != tonic::Code::Ok as i32 { + let code = tonic::Code::from(result.code); + results[index] = Err(notify_error(tonic::Status::new(code, result.message))); + } + } } - Ok(()) + Ok(results) } async fn attach_workflow(&self, attachment: WorkflowAttachment) -> Result<(), NotifyError> { diff --git a/crates/streams/src/driver.rs b/crates/streams/src/driver.rs index 95894a41c4..56f61c1a3d 100644 --- a/crates/streams/src/driver.rs +++ b/crates/streams/src/driver.rs @@ -262,8 +262,9 @@ impl StreamDriver { } } - /// Sends one namespace's entries in one call. When the server refuses the call, each stream - /// goes on its own, so one bad entry holds back no other stream. + /// Sends one namespace's entries in one call. The server answers each stream, and a stream it + /// refused or failed is settled on its own. When the server refuses the whole call, each + /// stream goes on its own, so one bad entry holds back no other stream. async fn send_namespace( &self, slot: u32, @@ -280,10 +281,20 @@ impl StreamDriver { return; } let notifications = streams.iter().map(|(n, _)| n.clone()).collect(); - match self.client.notify_streams(notifications).await { - Ok(()) => { - self.delivered.store(true, Ordering::Relaxed); - sent.ack(streams.into_iter().flat_map(|(_, ids)| ids)); + match self.client.notify_streams_each(notifications).await { + Ok(results) => { + // A server that took some entries is up, so the ones it failed count toward + // their drop. A server that failed them all may be busy, so they wait. + let reached = results.iter().any(Result::is_ok); + if reached { + self.delivered.store(true, Ordering::Relaxed); + } + for (stream, result) in streams.into_iter().zip(results) { + match result { + Ok(()) => sent.ack(stream.1), + Err(error) => self.settle_one(slot, error, vec![stream], sent, reached), + } + } } Err(NotifyError::Unsupported(error)) => { // A server with no notifiers answers this for every namespace. Only one namespace @@ -483,6 +494,10 @@ mod tests { fail_workflow: Mutex>, /// A call for this namespace answers that it takes no notifications. unsupported_namespace: Mutex>, + /// The server answers an entry for this Workflow id as refused, and takes the others. + refuse_entry: Mutex>, + /// The server answers an entry for this Workflow id as failed, and takes the others. + fail_entry: Mutex>, } #[async_trait::async_trait] @@ -519,6 +534,32 @@ mod tests { self.calls.lock().unwrap().push(notifications); Ok(()) } + + async fn notify_streams_each( + &self, + notifications: Vec, + ) -> Result>, NotifyError> { + let refused = self.refuse_entry.lock().unwrap().clone(); + let failed = self.fail_entry.lock().unwrap().clone(); + let results: Vec<_> = notifications + .iter() + .map(|n| match Some(n.chain.workflow_id.as_str()) { + id if id == refused.as_deref() => { + Err(NotifyError::Refused("no such owner".to_string())) + } + id if id == failed.as_deref() => Err(NotifyError::Failed("busy".to_string())), + _ => Ok(()), + }) + .collect(); + let landed = notifications + .into_iter() + .zip(&results) + .filter(|(_, result)| result.is_ok()) + .map(|(notification, _)| notification) + .collect(); + self.notify_streams(landed).await?; + Ok(results) + } } impl Recorder { @@ -709,6 +750,52 @@ mod tests { assert!(log_is_empty(&store).await, "the refused entry is acked too"); } + #[tokio::test] + async fn an_entry_the_answer_refuses_is_acked_with_no_second_call() { + let store = Arc::new(MemoryStore::with_notification_log()); + let client = Arc::new(Recorder::default()); + *client.refuse_entry.lock().unwrap() = Some("bad".to_string()); + append(&store, "bad", "out", 1).await; + append(&store, "good", "out", 1).await; + driver(&store, &client, "d1").drain(0).await.unwrap(); + assert_eq!(client.calls.lock().unwrap().len(), 1); + assert_eq!(client.sent(), [("good".to_string(), "out".to_string(), 1)]); + assert!(log_is_empty(&store).await, "the refused entry is acked too"); + } + + #[tokio::test(start_paused = true)] + async fn an_entry_the_answer_fails_stays_while_no_other_lands() { + let store = Arc::new(MemoryStore::with_notification_log()); + let client = Arc::new(Recorder::default()); + *client.fail_entry.lock().unwrap() = Some("bad".to_string()); + append(&store, "bad", "out", 1).await; + let driver = driver(&store, &client, "d1"); + for _ in 0..=MAX_ENTRY_FAILURES { + driver.drain(0).await.unwrap(); + } + assert!(!log_is_empty(&store).await); + *client.fail_entry.lock().unwrap() = None; + driver.drain(0).await.unwrap(); + assert_eq!(client.sent(), [("bad".to_string(), "out".to_string(), 1)]); + assert!(log_is_empty(&store).await); + } + + #[tokio::test(start_paused = true)] + async fn an_entry_the_answer_keeps_failing_while_others_land_is_dropped() { + let store = Arc::new(MemoryStore::with_notification_log()); + let client = Arc::new(Recorder::default()); + *client.fail_entry.lock().unwrap() = Some("bad".to_string()); + append(&store, "bad", "out", 1).await; + let driver = driver(&store, &client, "d1"); + for sequence in 1..=MAX_ENTRY_FAILURES as i64 { + assert!(!log_is_empty(&store).await, "kept after {sequence} drains"); + append(&store, "good", "out", sequence).await; + driver.drain(0).await.unwrap(); + } + assert!(log_is_empty(&store).await); + assert!(client.sent().iter().all(|(id, _, _)| id == "good")); + } + #[tokio::test] async fn a_namespace_without_notifiers_stops_neither_the_driver_nor_the_others() { let store = Arc::new(MemoryStore::with_notification_log()); diff --git a/crates/streams/src/notify.rs b/crates/streams/src/notify.rs index 8d66f71c84..dc71b699db 100644 --- a/crates/streams/src/notify.rs +++ b/crates/streams/src/notify.rs @@ -133,6 +133,18 @@ pub trait NotifierClient: Send + Sync { Ok(()) } + /// [Self::notify_streams], with one outcome per notification in the same order. The call + /// fails only as a whole. A client whose server answers only for the whole call answers each + /// notification as landed when the call does. + async fn notify_streams_each( + &self, + notifications: Vec, + ) -> Result>, NotifyError> { + let count = notifications.len(); + self.notify_streams(notifications).await?; + Ok((0..count).map(|_| Ok(())).collect()) + } + /// Attaches a Workflow's reader to a stream's notifier. Idempotent by its request id. async fn attach_workflow(&self, attachment: WorkflowAttachment) -> Result<(), NotifyError> { let _ = attachment;