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 5ebfbd75e9..1a06a64791 100644 --- a/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto +++ b/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto @@ -40,9 +40,16 @@ 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 + // 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); @@ -367,6 +374,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 { @@ -402,6 +413,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 c8fcf84eba..a78ed897a8 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/crates/sdk-core/CHANGELOG.md @@ -58,6 +58,17 @@ 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: 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/sdk-core/src/streams.rs b/crates/sdk-core/src/streams.rs index 432f6158ba..341fb07c1c 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/Cargo.toml b/crates/streams/Cargo.toml index 450eecace0..adb0579970 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 = [ @@ -31,7 +32,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 = ["macros", "rt", "sync", "time"] } tracing = "0.1" [dependencies.temporalio-common] diff --git a/crates/streams/docs/redis.md b/crates/streams/docs/redis.md index aaf2d20f3e..1acf929ca3 100644 --- a/crates/streams/docs/redis.md +++ b/crates/streams/docs/redis.md @@ -104,7 +104,9 @@ expiry from the record at its cursor. It exists so tests run those fallbacks on Listing a chain's topics scans every primary, as a delete does, and needs `+scan`. **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, once more as a user with exactly the ACL above, and once in plain mode with the cases of the fallbacks. `tests/fallbacks.rs` adds faults on plain Redis too. The cross-implementation test also needs `STREAMS_BROOK_PY`, a Python SDK checkout whose 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..56f61c1a3d --- /dev/null +++ b/crates/streams/src/driver.rs @@ -0,0 +1,868 @@ +//! 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 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, + StreamNotification, StreamResult, StreamStore, WORKFLOW_OWNER_KIND, mint_cursor, + progress_counter, proto::ChainId, stream_hash, +}; +use std::{ + collections::{BTreeMap, HashMap}, + sync::{ + Arc, Mutex, + atomic::{AtomicBool, Ordering}, + }, + 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, 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, + /// 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: 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), + batch: 100, + } + } +} + +/// 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, + 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>, + /// 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 { + /// 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(), + 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: 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() { + 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(()); + } + }; + // 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. 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, + 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_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 + // 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)), + } + } + for (error, notification, ids) in failed { + self.settle_one(slot, error, vec![(notification, ids)], sent, reached); + } + } + } + } + + /// 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; + } + } + } + } + } + + /// 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!( + workflow_id = %entry.chain.workflow_id, + topic = %entry.topic, + 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); + 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, ids))| { + let chain = chain.into_chain(); + let hash = stream_hash( + &chain.namespace, + WORKFLOW_OWNER_KIND, + &chain.workflow_id, + &topic, + ); + let notification = StreamNotification { + position: mint_cursor(self.store.name(), &hash, &position), + chain, + topic, + counter, + close: None, + }; + (notification, ids) + }) + .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, + /// 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>, + /// 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] + 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())); + } + 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(()) + } + + 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 { + /// 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_in(namespace: &str, workflow_id: &str) -> ChainId { + ChainId { + 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_in(namespace, 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); + } + + 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 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()); + 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; + assert!( + consumer.ends_with(&format!("-{}", std::process::id())), + "{consumer}" + ); + } + + #[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/handle.rs b/crates/streams/src/handle.rs index 1b2645e714..6a79cf6a69 100644 --- a/crates/streams/src/handle.rs +++ b/crates/streams/src/handle.rs @@ -923,6 +923,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/lib.rs b/crates/streams/src/lib.rs index 08fae65935..1631834e67 100644 --- a/crates/streams/src/lib.rs +++ b/crates/streams/src/lib.rs @@ -10,6 +10,7 @@ mod close; mod cursor; +mod driver; mod error; mod handle; mod memory; @@ -28,6 +29,7 @@ mod subscribe; pub use self::redis::{DEFAULT_NOTIFY_DOORBELLS, RedisStore, RedisStoreOptions}; pub use close::{CHAIN_LOG, append_batch, close_chain, close_topic}; 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/notify.rs b/crates/streams/src/notify.rs index 419c71143b..f2381ab47e 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; diff --git a/crates/streams/src/service.rs b/crates/streams/src/service.rs index f8ea1c4a27..f918fe2481 100644 --- a/crates/streams/src/service.rs +++ b/crates/streams/src/service.rs @@ -7,8 +7,8 @@ use crate::{ AppendRequest, CloseRequest, DeleteOwnerRequest, DescribeStoreRequest, DescribeStoreResponse, FlushNotificationsRequest, FlushNotificationsResponse, LatestRequest, ReadRangeRequest, ReadRequest, ReleaseFloorRequest, RetireProducerRequest, - StreamFailure, StreamStoreConfig, SubscribeRequest, UnsubscribeRequest, - WaitForNotificationRequest, stream_store_config, + RunStreamDriverRequest, RunStreamDriverResponse, StreamFailure, StreamStoreConfig, + SubscribeRequest, UnsubscribeRequest, WaitForNotificationRequest, stream_store_config, }, subscribe::Subscriptions, }; @@ -183,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 @@ -244,6 +254,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, @@ -793,6 +824,68 @@ 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::default(), + )), + 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..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; @@ -96,7 +100,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") }) @@ -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(); 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(); +}