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 36bc429db..5242d5013 100644 --- a/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto +++ b/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto @@ -24,6 +24,10 @@ service StreamService { rpc Close(CloseRequest) returns (CloseResponse); // Deletes every stream of one owner. rpc DeleteOwner(DeleteOwnerRequest) returns (DeleteOwnerResponse); + // Waits until the stream notifications this process sent are out, for at most 10 seconds. A + // notification goes in the background, so a process that holds only a client calls it before it + // stops. A Worker flushes at stop on its own. + rpc FlushNotifications(FlushNotificationsRequest) returns (FlushNotificationsResponse); } // Which store holds a process's streams, and how to reach it. Lang builds one store from it and @@ -33,6 +37,14 @@ message StreamStoreConfig { RedisStoreConfig redis = 1; MemoryStoreConfig memory = 2; } + // Tells each stream's notifier on the server when the stream moves, after every append and + // every Workflow batch that becomes visible. That's what makes a caller of a stream-returning + // Nexus operation see progress. Off by default, since most streams have no such caller and each + // notification costs a call. A server without the stream notifier costs one call, and then the + // process stops notifying. + bool notify_on_append = 3; + // How many streams to keep a notifier for, dropping the least recently used. Zero uses 1000. + uint32 max_notifiers = 4; } // Streams in a Redis or Redis Cluster that Core connects to itself. @@ -198,6 +210,10 @@ message CloseRequest { message CloseResponse {} +message FlushNotificationsRequest {} + +message FlushNotificationsResponse {} + message DeleteOwnerRequest { string namespace = 1; StreamOwnerKind owner_kind = 2; diff --git a/crates/sdk-core-c-bridge/src/tests/mod.rs b/crates/sdk-core-c-bridge/src/tests/mod.rs index c5ffb5650..069c48539 100644 --- a/crates/sdk-core-c-bridge/src/tests/mod.rs +++ b/crates/sdk-core-c-bridge/src/tests/mod.rs @@ -508,6 +508,7 @@ fn memory_store_config() -> Vec { }; StreamStoreConfig { store: Some(stream_store_config::Store::Memory(MemoryStoreConfig {})), + ..Default::default() } .encode_to_vec() } diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index 89cec05d5..90606aa7f 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/crates/sdk-core/CHANGELOG.md @@ -68,6 +68,17 @@ relevant information. Worker's `stream_store`, so a Workflow's output and outside producers share it. * Experimental: the C bridge has `temporal_core_stream_store_new`, `_call` and `_free` for the stream service, built with the Redis store, and a `stream_store` field on its worker options. +* Experimental: with `StreamStoreConfig.notify_on_append`, Core tells each stream's notifier on + the server when the stream moves, after every append and every Workflow batch that becomes + visible. That's how a caller of a stream-returning Nexus operation sees progress. Notifications + for one stream fold into one call in flight, a failed last notification is tried twice more, + and a notification never fails the write. `max_notifiers` caps how many streams the process + keeps a notifier for (1000 when zero). +* Experimental: the new `FlushNotifications` stream call waits up to 10 seconds for the + notifications still out. A Worker flushes at stop on its own, so only a process that holds just + a client needs it. +* Experimental: a server without the stream notifier answers the first notifier call with + `UNIMPLEMENTED`. Core then stops notifier calls in that process and logs it once at debug. ### Breaking Changes * Replay detects changes to the service or operation of scheduled Nexus operations, reporting diff --git a/crates/sdk-core/src/streams.rs b/crates/sdk-core/src/streams.rs index 9c4220091..5c93c9e3d 100644 --- a/crates/sdk-core/src/streams.rs +++ b/crates/sdk-core/src/streams.rs @@ -5,20 +5,104 @@ use std::sync::Arc; use temporalio_client::Connection; use temporalio_common::protos::temporal::api::{ common::v1::WorkflowExecution, - workflowservice::v1::{DescribeWorkflowExecutionRequest, DescribeWorkflowExecutionResponse}, + enums::v1::StreamOwnerKind, + stream::v1::StreamReference, + workflowservice::v1::{ + DescribeWorkflowExecutionRequest, DescribeWorkflowExecutionResponse, NotifyStreamRequest, + }, +}; +use temporalio_streams::{ + NotifierClient, NotifyError, OwnerClient, OwnerDescription, OwnerError, StreamNotification, }; -use temporalio_streams::{OwnerClient, OwnerDescription, OwnerError}; pub use temporalio_streams::{StreamError, StreamResult, StreamService, StreamStore, proto}; use tonic::IntoRequest; /// Connects to the store `config` names. The service asks `connection`'s server about streams' -/// owners, and its [StreamService::store] goes to each Worker's `stream_store`. +/// owners, sends the stream notifications the config turns on to it, and its +/// [StreamService::store] goes to each Worker's `stream_store`. pub async fn connect_stream_service( config: proto::StreamStoreConfig, connection: Connection, ) -> StreamResult> { - let owner = Arc::new(ConnectionOwnerClient { connection }); - Ok(Arc::new(StreamService::connect(config, owner).await?)) + let owner = Arc::new(ConnectionOwnerClient { + connection: connection.clone(), + }); + let notifier = Arc::new(ConnectionNotifierClient { connection }); + Ok(Arc::new( + StreamService::connect_with_notifier(config, owner, Some(notifier)).await?, + )) +} + +/// Sends stream notifications to a Temporal server's stream notifier through a client +/// connection. +struct ConnectionNotifierClient { + connection: Connection, +} + +#[async_trait::async_trait] +impl NotifierClient for ConnectionNotifierClient { + async fn notify(&self, notification: StreamNotification) -> Result<(), NotifyError> { + self.connection + .workflow_service() + .notify_stream(notify_request(notification).into_request()) + .await + .map(|_| ()) + .map_err(notify_error) + } +} + +fn notify_request(notification: StreamNotification) -> NotifyStreamRequest { + NotifyStreamRequest { + namespace: notification.chain.namespace, + // The run chain's first run keys the notifier, so one notifier serves the stream across + // Continue-as-New and a new chain on the same Workflow id gets its own. + stream_ref: Some(StreamReference { + owner_kind: StreamOwnerKind::Workflow as i32, + workflow_id: notification.chain.workflow_id, + run_id: notification.chain.first_run_id, + topic: notification.topic, + }), + position: notification.position, + counter: notification.counter, + close: notification.close_result.is_some(), + close_result: notification.close_result, + ..Default::default() + } +} + +/// Answers that mean the server will never take the notification, so a retry can't help. +fn notify_error(status: tonic::Status) -> NotifyError { + use tonic::Code; + match status.code() { + Code::Unimplemented => NotifyError::Unsupported(status.to_string()), + Code::FailedPrecondition if namespace_in_transition(&status) => { + NotifyError::Failed(status.to_string()) + } + Code::InvalidArgument + | Code::NotFound + | Code::PermissionDenied + | Code::Unauthenticated + | Code::FailedPrecondition => NotifyError::Refused(status.to_string()), + _ => NotifyError::Failed(status.to_string()), + } +} + +/// A namespace that's failing over answers `FailedPrecondition` too, but only until the +/// redirection sees the active cluster, so a close mustn't stop on it. +fn namespace_in_transition(status: &tonic::Status) -> bool { + use temporalio_common::protos::{ + google::rpc::Status as RpcStatus, + temporal::api::errordetails::v1::{ + NamespaceInvalidStateFailure, NamespaceNotActiveFailure, + }, + }; + let Ok(details) = ::decode(status.details()) else { + return false; + }; + details.details.iter().any(|detail| { + detail.to_msg::().is_ok() + || detail.to_msg::().is_ok() + }) } /// Asks a Temporal server about owners through a client connection. @@ -144,4 +228,95 @@ mod tests { Err(OwnerError::Failed(_)) )); } + + #[test] + fn a_notification_names_the_stream_by_its_chains_first_run() { + let request = notify_request(StreamNotification { + chain: proto::ChainId { + namespace: "ns".to_string(), + workflow_id: "wf".to_string(), + first_run_id: "run-1".to_string(), + }, + topic: "tokens".to_string(), + position: "cursor".to_string(), + counter: 7, + close_result: None, + }); + assert_eq!(request.namespace, "ns"); + assert_eq!( + request.stream_ref, + Some(StreamReference { + owner_kind: StreamOwnerKind::Workflow as i32, + workflow_id: "wf".to_string(), + run_id: "run-1".to_string(), + topic: "tokens".to_string(), + }) + ); + assert_eq!((request.position.as_str(), request.counter), ("cursor", 7)); + assert!(!request.close); + assert!(request.close_result.is_none()); + } + + #[test] + fn only_answers_a_retry_cant_change_refuse_a_notification() { + assert!(matches!( + notify_error(tonic::Status::new(tonic::Code::Unimplemented, "")), + NotifyError::Unsupported(_) + )); + for code in [ + tonic::Code::InvalidArgument, + tonic::Code::NotFound, + tonic::Code::PermissionDenied, + tonic::Code::Unauthenticated, + tonic::Code::FailedPrecondition, + ] { + assert!( + matches!( + notify_error(tonic::Status::new(code, "")), + NotifyError::Refused(_) + ), + "{code:?}" + ); + } + for code in [ + tonic::Code::Unavailable, + tonic::Code::DeadlineExceeded, + tonic::Code::ResourceExhausted, + tonic::Code::Internal, + ] { + assert!( + matches!( + notify_error(tonic::Status::new(code, "")), + NotifyError::Failed(_) + ), + "{code:?}" + ); + } + } + + fn failed_precondition_with(detail: &T) -> tonic::Status { + let status = temporalio_common::protos::google::rpc::Status { + code: tonic::Code::FailedPrecondition as i32, + message: "namespace".to_string(), + details: vec![prost_types::Any::from_msg(detail).unwrap()], + }; + tonic::Status::with_details( + tonic::Code::FailedPrecondition, + "namespace", + prost::Message::encode_to_vec(&status).into(), + ) + } + + #[test] + fn a_namespace_in_failover_fails_a_notification_for_now() { + use temporalio_common::protos::temporal::api::errordetails::v1::{ + NamespaceInvalidStateFailure, NamespaceNotActiveFailure, + }; + for status in [ + failed_precondition_with(&NamespaceNotActiveFailure::default()), + failed_precondition_with(&NamespaceInvalidStateFailure::default()), + ] { + assert!(matches!(notify_error(status), NotifyError::Failed(_))); + } + } } diff --git a/crates/sdk-core/src/worker/workflow/mod.rs b/crates/sdk-core/src/worker/workflow/mod.rs index c5299329d..42ba61a8c 100644 --- a/crates/sdk-core/src/worker/workflow/mod.rs +++ b/crates/sdk-core/src/worker/workflow/mod.rs @@ -843,6 +843,10 @@ impl Workflows { anyhow!("Error joining workflow processing thread: {as_str:?}") })?; } + #[cfg(feature = "streams")] + if let Some(store) = &self.output_store { + store.flush_notifications().await; + } Ok(()) } diff --git a/crates/sdk-core/src/worker/workflow/stream_output.rs b/crates/sdk-core/src/worker/workflow/stream_output.rs index d253610ec..2d71a5693 100644 --- a/crates/sdk-core/src/worker/workflow/stream_output.rs +++ b/crates/sdk-core/src/worker/workflow/stream_output.rs @@ -402,6 +402,20 @@ impl OutputStore { self.store.close_chain(&self.chain(chain)).await } + /// Waits a short while for the stream notifications still out, since they're sent in the + /// background and a process that stops right after its Worker would drop them. + pub(crate) async fn flush_notifications(&self) { + if tokio::time::timeout( + temporalio_streams::FLUSH_LIMIT, + self.store.flush_notifications(), + ) + .await + .is_err() + { + warn!("Stream notifications were still out when the Worker stopped"); + } + } + fn chain(&self, chain: &RunChain) -> temporalio_streams::proto::ChainId { temporalio_streams::proto::ChainId { namespace: self.namespace.clone(), @@ -1028,4 +1042,144 @@ mod tests { assert!(closed(&store, &memory, &["a"]).await.is_empty()); } } + + #[cfg(feature = "streams")] + mod notifications { + use super::{super::*, data}; + use parking_lot::Mutex; + use std::sync::Arc; + use temporalio_streams::{ + MemoryStore, StreamResult, StreamStore, + proto::{ + ChainId, DeleteOwnerRequest, DeleteOwnerResponse, Payload, StoreAppendRequest, + StoreAppendResponse, StoreLatestRequest, StoreLatestResponse, StoreReadRequest, + StoreReadResponse, + }, + }; + + /// A memory store whose notifications either log their flush or never go out. + #[derive(Default)] + struct Flushing { + inner: MemoryStore, + flushes: Mutex, + stuck: bool, + } + + #[async_trait::async_trait] + impl StreamStore for Flushing { + fn name(&self) -> &str { + "flushing" + } + async fn append(&self, r: StoreAppendRequest) -> StreamResult { + self.inner.append(r).await + } + async fn read(&self, r: StoreReadRequest) -> StreamResult { + self.inner.read(r).await + } + async fn latest(&self, r: StoreLatestRequest) -> StreamResult { + self.inner.latest(r).await + } + async fn record_at( + &self, + c: &ChainId, + t: &str, + p: &str, + ) -> StreamResult>> { + self.inner.record_at(c, t, p).await + } + async fn close_chain(&self, c: &ChainId) -> StreamResult<()> { + self.inner.close_chain(c).await + } + async fn close_topic( + &self, + c: &ChainId, + t: &str, + r: Option, + ) -> StreamResult<()> { + self.inner.close_topic(c, t, r).await + } + async fn delete_owner( + &self, + r: DeleteOwnerRequest, + ) -> StreamResult { + self.inner.delete_owner(r).await + } + async fn flush_notifications(&self) { + if self.stuck { + std::future::pending::<()>().await; + } + *self.flushes.lock() += 1; + } + } + + /// Records the stream notifications it was sent. + #[derive(Default)] + struct Recorder(Mutex>); + + #[async_trait::async_trait] + impl temporalio_streams::NotifierClient for Recorder { + async fn notify( + &self, + notification: temporalio_streams::StreamNotification, + ) -> Result<(), temporalio_streams::NotifyError> { + self.0.lock().push(notification); + Ok(()) + } + } + + fn batch(topic: &str) -> OutputBatch { + OutputBatch { + workflow_id: "wf".to_string(), + run_id: "run".to_string(), + first_run_id: "run".to_string(), + attempt: 1, + topics: vec![TopicOutput { + topic: topic.to_string(), + sequence: 1, + digest: vec![1; 32], + records: vec![data(topic, 1, 3)], + }], + } + } + + #[tokio::test] + async fn an_append_notifies_each_topic_it_made_visible() { + let recorder = Arc::new(Recorder::default()); + let notifying = Arc::new(temporalio_streams::NotifyingStore::new( + Arc::new(MemoryStore::new()), + Arc::new(temporalio_streams::Notifier::new(recorder.clone(), 10)), + )); + let store = OutputStore::new(notifying, "ns".to_string(), Default::default()); + store.append(&batch("t")).await.unwrap(); + store.flush_notifications().await; + let sent = recorder.0.lock().clone(); + assert_eq!(sent.len(), 1); + assert_eq!((sent[0].topic.as_str(), sent[0].counter), ("t", 1)); + assert_eq!(sent[0].chain.first_run_id, "run"); + } + + #[tokio::test] + async fn a_stopping_worker_waits_for_its_stream_notifications() { + let flushing = Arc::new(Flushing::default()); + OutputStore::new(flushing.clone(), "ns".to_string(), Default::default()) + .flush_notifications() + .await; + assert_eq!(*flushing.flushes.lock(), 1); + } + + #[tokio::test(start_paused = true)] + async fn a_stopping_worker_waits_for_its_stream_notifications_only_so_long() { + let store = OutputStore::new( + Arc::new(Flushing { + stuck: true, + ..Default::default() + }), + "ns".to_string(), + Default::default(), + ); + let started = tokio::time::Instant::now(); + store.flush_notifications().await; + assert_eq!(started.elapsed(), temporalio_streams::FLUSH_LIMIT); + } + } } diff --git a/crates/streams/Cargo.toml b/crates/streams/Cargo.toml index 4b3567b04..ce381142c 100644 --- a/crates/streams/Cargo.toml +++ b/crates/streams/Cargo.toml @@ -21,6 +21,7 @@ redis = ["dep:redis", "dep:crc16"] async-trait = "0.1" base64 = "0.23" crc16 = { version = "0.4", optional = true } +lru = "0.18" prost = { workspace = true } redis = { version = "1.7", optional = true, default-features = false, features = [ "cluster-async", diff --git a/crates/streams/src/lib.rs b/crates/streams/src/lib.rs index 33cfbd3c7..3e587f5fb 100644 --- a/crates/streams/src/lib.rs +++ b/crates/streams/src/lib.rs @@ -13,6 +13,7 @@ mod error; mod handle; mod memory; pub mod nexus; +mod notify; mod owner; mod reader; mod record; @@ -27,6 +28,10 @@ pub use cursor::{BEGINNING, END, cursor_position, mint_cursor, stream_hash}; pub use error::{StreamError, StreamResult}; pub use handle::{OWNER_RECHECK, Streams, StreamsOptions}; pub use memory::MemoryStore; +pub use notify::{ + DEFAULT_MAX_NOTIFIERS, FLUSH_LIMIT, Notifier, NotifierClient, NotifyError, NotifyingStore, + StreamNotification, progress_counter, +}; pub use owner::{OwnerClient, OwnerDescription, OwnerError}; pub use reader::{DEFAULT_MAX_RECORDS, MAX_TRACKED_PRODUCERS, ReadTarget, read_page}; pub use record::{ diff --git a/crates/streams/src/notify.rs b/crates/streams/src/notify.rs new file mode 100644 index 000000000..bea80526c --- /dev/null +++ b/crates/streams/src/notify.rs @@ -0,0 +1,1016 @@ +//! Telling a stream's notifier on the server that the stream moved or closed. +//! +//! The notifier holds the callbacks of every caller a stream-returning Nexus operation attached. A +//! notification is a hint that there is something new to read. The records stay on the read path. + +use crate::{ + StreamResult, StreamStore, WORKFLOW_OWNER_KIND, mint_cursor, + proto::{ + ChainId, DeleteOwnerRequest, DeleteOwnerResponse, StoreAppendRequest, StoreAppendResponse, + StoreLatestRequest, StoreLatestResponse, StoreReadRequest, StoreReadResponse, + }, + stream_hash, +}; +use std::{ + future::Future, + num::NonZeroUsize, + sync::{ + Arc, Mutex, + atomic::{AtomicBool, Ordering}, + }, + time::Duration, +}; +use temporalio_common::protos::temporal::api::common::v1::Payload; +use tokio::sync::watch; + +/// How many streams a process keeps a notifier for when its configuration doesn't say. +pub const DEFAULT_MAX_NOTIFIERS: usize = 1000; + +/// How long a stopping Worker waits for the notifications still out. +pub const FLUSH_LIMIT: Duration = Duration::from_secs(10); + +// A stream that goes quiet after a failed notification leaves its reader parked until the next +// write, so the last one is tried again a few times. A newer one replaces the retry. +const NOTIFY_RETRIES: u32 = 2; +const NOTIFY_RETRY_FIRST: Duration = Duration::from_millis(250); + +// Redis entry ids are `-`. Twenty bits hold the sequence, far more entries than one +// stream takes in a millisecond, and the milliseconds stay below 2^43 until the year 2248, so the +// packed value fits a positive i64. +const REDIS_SEQUENCE_BITS: u32 = 20; + +/// One notification for one stream's notifier. +#[derive(Debug, Clone, PartialEq)] +pub struct StreamNotification { + /// The run chain whose stream moved. Its first run keys the notifier. + pub chain: ChainId, + /// The stream's topic. + pub topic: String, + /// The cursor of the newest record the notification covers. + pub position: String, + /// Grows with the stream. The notifier keeps the highest and a caller drops a lower one. + pub counter: i64, + /// Set on the close, which completes every attached operation with it. + pub close_result: Option, +} + +/// Why a notifier didn't take a notification. +#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)] +pub enum NotifyError { + /// The server will never take it, so trying again can't help. + #[error("the server refused the notification: {0}")] + Refused(String), + /// It may take it later. + #[error("the notification failed: {0}")] + Failed(String), + /// The server has no stream notifier. + #[error("the server has no stream notifier: {0}")] + Unsupported(String), +} + +/// Sends notifications to the stream notifier on the server. +#[async_trait::async_trait] +pub trait NotifierClient: Send + Sync { + /// Sends one notification. + async fn notify(&self, notification: StreamNotification) -> Result<(), NotifyError>; +} + +/// The notifier counter for a store position. +/// +/// It must grow with the stream, and producers in different processes must agree on it. So it +/// comes from the store's own position, not from a clock. A memory position, an offset, gives +/// `offset + 1`. A Redis entry id `-` gives `ms << 20 | seq`, with a sequence beyond 20 +/// bits held at the largest, which keeps the order non-decreasing. No position gives zero. Two +/// records past the 20-bit sequence in one millisecond get one counter, and the notifier drops +/// the second as a repeat. The next notification tells the reader again. +pub fn progress_counter(position: &str) -> Option { + if position.is_empty() { + return Some(0); + } + match position.split_once('-') { + Some((milliseconds, sequence)) => { + let milliseconds: i64 = milliseconds.parse().ok()?; + let sequence: u64 = sequence.parse().ok()?; + let largest = (1u64 << REDIS_SEQUENCE_BITS) - 1; + let sequence = i64::try_from(sequence.min(largest)).ok()?; + milliseconds + .checked_mul(1 << REDIS_SEQUENCE_BITS) + .map(|shifted| shifted | sequence) + .filter(|counter| *counter >= 0) + } + None => position.parse::().ok()?.checked_add(1), + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +struct StreamKey { + namespace: String, + workflow_id: String, + first_run_id: String, + topic: String, +} + +impl StreamKey { + fn new(chain: &ChainId, topic: &str) -> Self { + Self { + namespace: chain.namespace.clone(), + workflow_id: chain.workflow_id.clone(), + first_run_id: chain.first_run_id.clone(), + topic: topic.to_string(), + } + } + + fn chain(&self) -> ChainId { + ChainId { + namespace: self.namespace.clone(), + workflow_id: self.workflow_id.clone(), + first_run_id: self.first_run_id.clone(), + } + } +} + +#[derive(Debug, Default)] +struct SlotState { + pending: Option<(String, i64)>, + sending: bool, + /// Positions still being looked up, which the slot sends once found. + resolving: usize, + closed: bool, +} + +/// One stream's notifications: folded, with one call in flight per slot. +#[derive(Debug)] +struct Slot { + key: StreamKey, + state: Mutex, + idle: watch::Sender, +} + +impl Slot { + fn new(key: StreamKey) -> Arc { + Arc::new(Self { + key, + state: Mutex::default(), + idle: watch::Sender::new(true), + }) + } + + /// Records the newest position and returns at once. While a call is out, later + /// notifications fold into the one that goes next, so a burst costs at most two calls. + fn notify(self: &Arc, client: &Arc, position: String, counter: i64) { + let mut state = self.state.lock().unwrap(); + self.hold(&mut state, client, position, counter); + } + + /// Like [Slot::notify] with the position `lookup` finds, without waiting for it. The slot + /// counts as busy until the lookup ends, so a close or a flush waits for it. + fn notify_found(self: &Arc, client: &Arc, lookup: F) + where + F: Future> + Send + 'static, + { + { + let mut state = self.state.lock().unwrap(); + if state.closed { + return; + } + state.resolving += 1; + self.idle.send_replace(false); + } + let (slot, client) = (self.clone(), client.clone()); + tokio::spawn(async move { + let found = lookup.await; + let mut state = slot.state.lock().unwrap(); + state.resolving -= 1; + if let Some((position, counter)) = found { + slot.hold(&mut state, &client, position, counter); + } + slot.mark_idle_if_done(&state); + }); + } + + fn hold( + self: &Arc, + state: &mut SlotState, + client: &Arc, + position: String, + counter: i64, + ) { + if state.closed { + return; + } + if state + .pending + .as_ref() + .is_none_or(|(_, held)| counter >= *held) + { + state.pending = Some((position, counter)); + } + if !state.sending { + state.sending = true; + self.idle.send_replace(false); + tokio::spawn(self.clone().send_pending(client.clone())); + } + } + + fn mark_idle_if_done(&self, state: &SlotState) { + if !state.sending && state.resolving == 0 { + self.idle.send_replace(true); + } + } + + async fn send_pending(self: Arc, client: Arc) { + loop { + let (position, counter) = { + let mut state = self.state.lock().unwrap(); + match state.pending.take() { + Some(next) => next, + None => { + state.sending = false; + self.mark_idle_if_done(&state); + return; + } + } + }; + let notification = StreamNotification { + chain: self.key.chain(), + topic: self.key.topic.clone(), + position, + counter, + close_result: None, + }; + self.send(&client, notification).await; + } + } + + async fn send(&self, client: &Arc, notification: StreamNotification) { + let mut delay = NOTIFY_RETRY_FIRST; + for tries_left in (0..=NOTIFY_RETRIES).rev() { + // A notification is only a hint, and the reader reads from its own cursor, so a + // notification that can't go never fails the write. + match client.notify(notification.clone()).await { + Ok(()) => return, + Err(NotifyError::Failed(error)) if tries_left > 0 && !self.superseded() => { + tracing::debug!( + workflow_id = %self.key.workflow_id, + topic = %self.key.topic, + %error, + "A stream notification failed; trying again in {delay:?}" + ); + } + Err(error) => { + tracing::warn!( + workflow_id = %self.key.workflow_id, + topic = %self.key.topic, + %error, + "A stream notification failed. The next write tells the reader again" + ); + return; + } + } + tokio::time::sleep(delay).await; + delay *= 2; + if self.superseded() { + return; + } + } + } + + /// Whether something newer will reach the notifier anyway: a later notification, or the + /// close. + fn superseded(&self) -> bool { + let state = self.state.lock().unwrap(); + state.pending.is_some() || state.closed + } + + /// Stops new notifications and waits for the one in flight, so a close is the last thing the + /// notifier hears from this process. + async fn close(&self) { + self.state.lock().unwrap().closed = true; + self.wait_idle().await; + } + + async fn wait_idle(&self) { + let mut idle = self.idle.subscribe(); + // The sender lives as long as the slot, so the wait can't fail. + let _ = idle.wait_for(|idle| *idle).await; + } + + fn is_idle(&self) -> bool { + *self.idle.borrow() + } +} + +/// Notifies the stream notifiers of the streams a process writes to. +/// +/// Keeps one slot per stream, and drops a slot when its stream closes, when its chain ends, or +/// when it's the least recently used of more than the limit. A dropped slot still sends what it +/// holds, and [Notifier::flush] waits for it. +pub struct Notifier { + client: Arc, + slots: Mutex>>, + retired: Mutex>>, +} + +impl std::fmt::Debug for Notifier { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("Notifier") + .field("slots", &self.slots.lock().unwrap().len()) + .finish_non_exhaustive() + } +} + +impl Notifier { + /// Notifies through `client`, keeping at most `max_notifiers` streams' slots. + pub fn new(client: Arc, max_notifiers: usize) -> Self { + Self { + client: Arc::new(UnlessUnsupported { + client, + unsupported: AtomicBool::new(false), + }), + slots: Mutex::new(lru::LruCache::new( + NonZeroUsize::new(max_notifiers).unwrap_or(NonZeroUsize::MIN), + )), + retired: Mutex::default(), + } + } + + /// Tells the notifier of `chain`'s stream on `topic` that it moved to `position`. Never + /// waits, and never fails: a notification that can't go is logged and dropped. + pub fn notify(&self, chain: &ChainId, topic: &str, position: String, counter: i64) { + self.slot(StreamKey::new(chain, topic)) + .notify(&self.client, position, counter); + } + + /// Like [Notifier::notify] with the cursor and counter `lookup` finds, without waiting for + /// it. A close of the stream and [Notifier::flush] wait for the lookup. + pub fn notify_found(&self, chain: &ChainId, topic: &str, lookup: F) + where + F: Future> + Send + 'static, + { + self.slot(StreamKey::new(chain, topic)) + .notify_found(&self.client, lookup); + } + + /// Closes the stream on the server with `result`, after the notification in flight, and + /// drops its slot. `counter` must rank above the stream's last notification. + pub async fn close( + &self, + chain: &ChainId, + topic: &str, + position: String, + counter: i64, + result: Payload, + ) -> Result<(), NotifyError> { + let key = StreamKey::new(chain, topic); + let held = self.slots.lock().unwrap().pop(&key); + if let Some(slot) = &held { + slot.close().await; + } + self.client + .notify(StreamNotification { + chain: chain.clone(), + topic: topic.to_string(), + position, + counter, + close_result: Some(result), + }) + .await + } + + /// Drops the slots of `chain`'s streams, once the chain ended. + pub fn forget_chain(&self, chain: &ChainId) { + let mut slots = self.slots.lock().unwrap(); + let ended: Vec = slots + .iter() + .map(|(key, _)| key) + .filter(|key| key.chain() == *chain) + .cloned() + .collect(); + let dropped: Vec> = ended.iter().filter_map(|key| slots.pop(key)).collect(); + drop(slots); + self.retire(dropped); + } + + /// Drops the slot of one stream, once it closed. + pub fn forget_stream(&self, chain: &ChainId, topic: &str) { + let dropped = self + .slots + .lock() + .unwrap() + .pop(&StreamKey::new(chain, topic)); + self.retire(dropped); + } + + /// Waits until no notification is out or waiting. + pub async fn flush(&self) { + let slots: Vec> = self + .slots + .lock() + .unwrap() + .iter() + .map(|(_, slot)| slot.clone()) + .chain(self.retired.lock().unwrap().iter().cloned()) + .collect(); + for slot in slots { + slot.wait_idle().await; + } + self.retired.lock().unwrap().retain(|slot| !slot.is_idle()); + } + + fn slot(&self, key: StreamKey) -> Arc { + let mut slots = self.slots.lock().unwrap(); + if let Some(slot) = slots.get(&key) { + return slot.clone(); + } + let slot = Slot::new(key.clone()); + let evicted = slots.push(key, slot.clone()).map(|(_, slot)| slot); + drop(slots); + self.retire(evicted); + slot + } + + fn retire(&self, dropped: impl IntoIterator>) { + let mut retired = self.retired.lock().unwrap(); + retired.retain(|slot| !slot.is_idle()); + retired.extend(dropped.into_iter().filter(|slot| !slot.is_idle())); + } +} + +/// Stops every call once the server answers that it has no stream notifier. The answer can't +/// change while the process runs, and each call would cost a round trip. +struct UnlessUnsupported { + client: Arc, + unsupported: AtomicBool, +} + +#[async_trait::async_trait] +impl NotifierClient for UnlessUnsupported { + async fn notify(&self, notification: StreamNotification) -> Result<(), NotifyError> { + if self.unsupported.load(Ordering::Relaxed) { + return Ok(()); + } + match self.client.notify(notification).await { + // No caller can be attached to a notifier the server doesn't have. + Err(NotifyError::Unsupported(error)) => { + if !self.unsupported.swap(true, Ordering::Relaxed) { + tracing::debug!( + %error, + "The server has no stream notifier, so this process stops notifying" + ); + } + Ok(()) + } + other => other, + } + } +} + +/// A store that notifies each stream's notifier after an append lands, a Workflow's output +/// included. +/// +/// Wraps the store a process hands to its stream calls and its Workers, so every write path +/// notifies without knowing about notifiers. It forwards every trait method by name, since a +/// method left to its default would answer for the inner store. +pub struct NotifyingStore { + inner: Arc, + notifier: Arc, +} + +impl NotifyingStore { + /// Wraps `inner`, notifying through `notifier`. + pub fn new(inner: Arc, notifier: Arc) -> Self { + Self { inner, notifier } + } + + /// The notifier this store notifies through. + pub fn notifier(&self) -> &Arc { + &self.notifier + } + + fn notify_at(&self, chain: &ChainId, topic: &str, position: &str) { + if let Some((cursor, counter)) = + notification_at(self.inner.as_ref(), chain, topic, position) + { + self.notifier.notify(chain, topic, cursor, counter); + } + } +} + +fn cursor_of(store: &dyn StreamStore, chain: &ChainId, topic: &str, position: &str) -> String { + let hash = stream_hash( + &chain.namespace, + WORKFLOW_OWNER_KIND, + &chain.workflow_id, + topic, + ); + mint_cursor(store.name(), &hash, position) +} + +/// The cursor and counter that tell a reader about the record at `position`. +fn notification_at( + store: &dyn StreamStore, + chain: &ChainId, + topic: &str, + position: &str, +) -> Option<(String, i64)> { + // The write landed, so a notification that can't go must not turn it into an error the + // caller might retry with other content. + let Some(counter) = progress_counter(position) else { + tracing::warn!( + workflow_id = %chain.workflow_id, + topic, + position, + "No notifier counter comes from this position, so the write isn't notified" + ); + return None; + }; + Some((cursor_of(store, chain, topic, position), counter)) +} + +#[async_trait::async_trait] +impl StreamStore for NotifyingStore { + fn name(&self) -> &str { + self.inner.name() + } + + async fn append(&self, request: StoreAppendRequest) -> StreamResult { + let chain = request.chain.clone(); + let topic = request.topic.clone(); + let landed = self.inner.append(request).await?; + if let Some(chain) = &chain { + self.notify_at(chain, &topic, &landed.last_position); + } + Ok(landed) + } + + async fn read(&self, request: StoreReadRequest) -> StreamResult { + self.inner.read(request).await + } + + async fn latest(&self, request: StoreLatestRequest) -> StreamResult { + self.inner.latest(request).await + } + + async fn record_at( + &self, + chain: &ChainId, + topic: &str, + position: &str, + ) -> StreamResult>> { + self.inner.record_at(chain, topic, position).await + } + + async fn newest_ever(&self, chain: &ChainId, topic: &str) -> StreamResult { + self.inner.newest_ever(chain, topic).await + } + + async fn dropped_after( + &self, + chain: &ChainId, + topic: &str, + mark: &str, + ) -> StreamResult> { + self.inner.dropped_after(chain, topic, mark).await + } + + async fn close_chain(&self, chain: &ChainId) -> StreamResult<()> { + self.inner.close_chain(chain).await?; + self.notifier.forget_chain(chain); + Ok(()) + } + + async fn close_topic( + &self, + chain: &ChainId, + topic: &str, + result: Option, + ) -> StreamResult<()> { + self.inner.close_topic(chain, topic, result).await?; + self.notifier.forget_stream(chain, topic); + Ok(()) + } + + async fn delete_owner(&self, request: DeleteOwnerRequest) -> StreamResult { + self.inner.delete_owner(request).await + } + + async fn flush_notifications(&self) { + self.notifier.flush().await; + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::{MemoryStore, proto::StoreAppendRequest}; + use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; + use tokio::sync::Semaphore; + + /// Records what it was sent. While `holding`, each call waits for a permit, so a test + /// decides when the call in flight returns. + struct Recorder { + sent: Mutex>, + holding: AtomicBool, + gate: Semaphore, + fail: AtomicBool, + /// How many of the next calls fail. + failures: AtomicUsize, + unsupported: AtomicBool, + } + + impl Default for Recorder { + fn default() -> Self { + Self { + sent: Mutex::default(), + holding: AtomicBool::new(false), + gate: Semaphore::new(0), + fail: AtomicBool::new(false), + failures: AtomicUsize::new(0), + unsupported: AtomicBool::new(false), + } + } + } + + impl Recorder { + fn holding() -> Arc { + let recorder = Arc::new(Self::default()); + recorder.holding.store(true, Ordering::SeqCst); + recorder + } + + fn sent(&self) -> Vec<(String, i64)> { + self.sent + .lock() + .unwrap() + .iter() + .map(|n| (n.position.clone(), n.counter)) + .collect() + } + + fn release(&self, calls: usize) { + self.gate.add_permits(calls); + } + } + + #[async_trait::async_trait] + impl NotifierClient for Recorder { + async fn notify(&self, notification: StreamNotification) -> Result<(), NotifyError> { + self.sent.lock().unwrap().push(notification); + if self.holding.load(Ordering::SeqCst) { + self.gate.acquire().await.unwrap().forget(); + } + if self.unsupported.load(Ordering::SeqCst) { + return Err(NotifyError::Unsupported("unimplemented".to_string())); + } + let failing = self + .failures + .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |n| n.checked_sub(1)) + .is_ok(); + if failing || self.fail.load(Ordering::SeqCst) { + return Err(NotifyError::Failed("down".to_string())); + } + Ok(()) + } + } + + fn chain(first_run_id: &str) -> ChainId { + ChainId { + namespace: "ns".to_string(), + workflow_id: "wf".to_string(), + first_run_id: first_run_id.to_string(), + } + } + + /// Lets every spawned send run until it waits. + async fn settle() { + for _ in 0..10 { + tokio::task::yield_now().await; + } + } + + #[test] + fn counters_come_from_the_stores_position() { + assert_eq!(progress_counter(""), Some(0)); + assert_eq!(progress_counter("0"), Some(1)); + assert_eq!(progress_counter("41"), Some(42)); + assert_eq!( + progress_counter("1700000000000-3"), + Some((1_700_000_000_000 << 20) | 3) + ); + assert_eq!( + progress_counter("5-2000000"), + Some((5 << 20) | ((1 << 20) - 1)), + "a sequence past 20 bits is held at the largest" + ); + assert!(progress_counter("1-1") < progress_counter("1-2")); + assert!(progress_counter("1-2") < progress_counter("2-0")); + assert_eq!(progress_counter("not a position"), None); + assert_eq!(progress_counter("9223372036854775807-0"), None, "past i64"); + } + + #[tokio::test] + async fn notifications_fold_with_one_call_in_flight() { + let recorder = Recorder::holding(); + let notifier = Notifier::new(recorder.clone(), 10); + notifier.notify(&chain("run-1"), "t", "c1".to_string(), 1); + settle().await; + notifier.notify(&chain("run-1"), "t", "c3".to_string(), 3); + notifier.notify(&chain("run-1"), "t", "c2".to_string(), 2); + settle().await; + assert_eq!( + recorder.sent(), + vec![("c1".to_string(), 1)], + "one call in flight" + ); + recorder.release(2); + notifier.flush().await; + assert_eq!( + recorder.sent(), + vec![("c1".to_string(), 1), ("c3".to_string(), 3)], + "the burst folds into one call with the highest counter" + ); + } + + #[tokio::test(start_paused = true)] + async fn a_failed_last_notification_is_tried_twice_more() { + let recorder = Arc::new(Recorder::default()); + recorder.failures.store(1, Ordering::SeqCst); + let notifier = Notifier::new(recorder.clone(), 10); + notifier.notify(&chain("run-1"), "t", "c1".to_string(), 1); + notifier.flush().await; + assert_eq!( + recorder.sent(), + vec![("c1".to_string(), 1), ("c1".to_string(), 1)], + "a quiet stream gets its last notification again" + ); + + recorder.fail.store(true, Ordering::SeqCst); + notifier.notify(&chain("run-1"), "t", "c2".to_string(), 2); + notifier.flush().await; + assert_eq!( + recorder.sent().len(), + 5, + "two more tries, then it's dropped" + ); + } + + #[tokio::test(start_paused = true)] + async fn a_newer_notification_goes_instead_of_a_retry() { + let recorder = Arc::new(Recorder::default()); + recorder.failures.store(1, Ordering::SeqCst); + let notifier = Notifier::new(recorder.clone(), 10); + notifier.notify(&chain("run-1"), "t", "c1".to_string(), 1); + settle().await; + notifier.notify(&chain("run-1"), "t", "c2".to_string(), 2); + notifier.flush().await; + assert_eq!( + recorder.sent(), + vec![("c1".to_string(), 1), ("c2".to_string(), 2)] + ); + } + + #[tokio::test] + async fn a_server_without_the_notifier_costs_one_call() { + let recorder = Arc::new(Recorder::default()); + recorder.unsupported.store(true, Ordering::SeqCst); + let notifier = Notifier::new(recorder.clone(), 10); + notifier.notify(&chain("run-1"), "a", "a1".to_string(), 1); + notifier.flush().await; + notifier.notify(&chain("run-1"), "b", "b1".to_string(), 1); + notifier.notify(&chain("run-2"), "a", "a1".to_string(), 1); + notifier.flush().await; + notifier + .close( + &chain("run-1"), + "a", + "a1".to_string(), + 2, + Payload::default(), + ) + .await + .unwrap(); + assert_eq!(recorder.sent().len(), 1); + } + + #[tokio::test] + async fn close_waits_for_the_call_in_flight_and_carries_the_result() { + let recorder = Recorder::holding(); + let notifier = Arc::new(Notifier::new(recorder.clone(), 10)); + notifier.notify(&chain("run-1"), "t", "c1".to_string(), 1); + settle().await; + let closing = { + let notifier = notifier.clone(); + tokio::spawn(async move { + let result = Payload { + data: b"summary".to_vec(), + ..Default::default() + }; + notifier + .close(&chain("run-1"), "t", "c1".to_string(), 2, result) + .await + }) + }; + settle().await; + assert_eq!( + recorder.sent().len(), + 1, + "the close waits for the call in flight" + ); + recorder.release(2); + closing.await.unwrap().unwrap(); + let sent = recorder.sent.lock().unwrap().clone(); + assert_eq!(sent.len(), 2); + assert_eq!(sent[1].counter, 2); + assert_eq!(sent[1].close_result.as_ref().unwrap().data, b"summary"); + } + + #[tokio::test] + async fn a_notification_after_close_sends_nothing() { + let recorder = Arc::new(Recorder::default()); + let notifier = Notifier::new(recorder.clone(), 10); + notifier.notify(&chain("run-1"), "t", "c1".to_string(), 1); + let slot = notifier.slot(StreamKey::new(&chain("run-1"), "t")); + slot.close().await; + slot.notify(¬ifier.client, "c2".to_string(), 2); + notifier.flush().await; + assert_eq!(recorder.sent(), vec![("c1".to_string(), 1)]); + } + + #[tokio::test] + async fn notifiers_beyond_the_cap_are_dropped_least_recent_first() { + let recorder = Recorder::holding(); + let notifier = Notifier::new(recorder.clone(), 2); + notifier.notify(&chain("run-1"), "a", "a1".to_string(), 1); + notifier.notify(&chain("run-1"), "b", "b1".to_string(), 1); + settle().await; + notifier.notify(&chain("run-1"), "a", "a2".to_string(), 2); + notifier.notify(&chain("run-1"), "c", "c1".to_string(), 1); + let kept: Vec = notifier + .slots + .lock() + .unwrap() + .iter() + .map(|(key, _)| key.topic.clone()) + .collect(); + assert_eq!(kept, vec!["c", "a"], "b was the least recently used"); + recorder.release(10); + notifier.flush().await; + let mut sent = recorder.sent(); + sent.sort(); + assert_eq!( + sent, + vec![ + ("a1".to_string(), 1), + ("a2".to_string(), 2), + ("b1".to_string(), 1), + ("c1".to_string(), 1) + ], + "a dropped notifier still sends what it holds" + ); + assert!(notifier.retired.lock().unwrap().is_empty()); + } + + #[tokio::test] + async fn an_ended_chain_drops_its_notifiers() { + let recorder = Arc::new(Recorder::default()); + let notifier = Notifier::new(recorder.clone(), 10); + notifier.notify(&chain("run-1"), "a", "a1".to_string(), 1); + notifier.notify(&chain("run-1"), "b", "b1".to_string(), 1); + notifier.notify(&chain("run-9"), "a", "x1".to_string(), 1); + notifier.forget_chain(&chain("run-1")); + notifier.forget_stream(&chain("run-9"), "a"); + assert_eq!(notifier.slots.lock().unwrap().len(), 0); + notifier.flush().await; + assert_eq!( + recorder.sent().len(), + 3, + "the dropped ones still sent what they held" + ); + } + + #[tokio::test] + async fn flush_waits_for_every_notification_in_flight() { + let recorder = Recorder::holding(); + let notifier = Arc::new(Notifier::new(recorder.clone(), 10)); + notifier.notify(&chain("run-1"), "a", "a1".to_string(), 1); + notifier.notify(&chain("run-1"), "b", "b1".to_string(), 1); + let flushing = { + let notifier = notifier.clone(); + tokio::spawn(async move { notifier.flush().await }) + }; + settle().await; + assert!(!flushing.is_finished()); + recorder.release(1); + settle().await; + assert!(!flushing.is_finished(), "one is still out"); + recorder.release(1); + flushing.await.unwrap(); + } + + fn append_request(chain: &ChainId, topic: &str, sequence: i64) -> StoreAppendRequest { + StoreAppendRequest { + chain: Some(chain.clone()), + topic: topic.to_string(), + producer_id: "p".to_string(), + attempt: 1, + sequence, + digest: vec![sequence as u8], + records: vec![b"record".to_vec()], + } + } + + fn notifying_store(recorder: &Arc) -> (Arc, NotifyingStore) { + let inner = Arc::new(MemoryStore::new()); + let notifier = Arc::new(Notifier::new(recorder.clone(), 10)); + (inner.clone(), NotifyingStore::new(inner, notifier)) + } + + #[tokio::test] + async fn an_append_notifies_the_chain_it_wrote_to() { + let recorder = Arc::new(Recorder::default()); + let (_, store) = notifying_store(&recorder); + let landed = store + .append(append_request(&chain("run-1"), "t", 1)) + .await + .unwrap(); + store.flush_notifications().await; + let sent = recorder.sent.lock().unwrap().clone(); + assert_eq!(sent.len(), 1); + assert_eq!(sent[0].chain, chain("run-1")); + assert_eq!(sent[0].topic, "t"); + assert_eq!( + sent[0].counter, + progress_counter(&landed.last_position).unwrap() + ); + assert_eq!( + sent[0].position, + mint_cursor( + "memory", + &stream_hash("ns", WORKFLOW_OWNER_KIND, "wf", "t"), + &landed.last_position + ), + "the position is the cursor a reader resumes from" + ); + } + + /// A store whose appends land at a position no counter comes from. + struct OddPositions(MemoryStore); + + #[async_trait::async_trait] + impl StreamStore for OddPositions { + fn name(&self) -> &str { + "odd" + } + async fn append(&self, r: StoreAppendRequest) -> StreamResult { + self.0.append(r).await?; + Ok(StoreAppendResponse { + first_position: "odd".to_string(), + last_position: "odd".to_string(), + }) + } + async fn read(&self, r: StoreReadRequest) -> StreamResult { + self.0.read(r).await + } + async fn latest(&self, r: StoreLatestRequest) -> StreamResult { + self.0.latest(r).await + } + async fn record_at(&self, c: &ChainId, t: &str, p: &str) -> StreamResult>> { + self.0.record_at(c, t, p).await + } + async fn close_chain(&self, c: &ChainId) -> StreamResult<()> { + self.0.close_chain(c).await + } + async fn close_topic(&self, c: &ChainId, t: &str, r: Option) -> StreamResult<()> { + self.0.close_topic(c, t, r).await + } + async fn delete_owner(&self, r: DeleteOwnerRequest) -> StreamResult { + self.0.delete_owner(r).await + } + } + + #[tokio::test] + async fn a_position_without_a_counter_does_not_fail_the_append() { + let recorder = Arc::new(Recorder::default()); + let notifier = Arc::new(Notifier::new(recorder.clone(), 10)); + let store = NotifyingStore::new(Arc::new(OddPositions(MemoryStore::new())), notifier); + store + .append(append_request(&chain("run-1"), "t", 1)) + .await + .unwrap(); + store.flush_notifications().await; + assert!(recorder.sent().is_empty()); + } + + #[tokio::test] + async fn a_closed_streams_notifier_is_dropped() { + let recorder = Arc::new(Recorder::default()); + let (_, store) = notifying_store(&recorder); + let run = chain("run-1"); + store.append(append_request(&run, "a", 1)).await.unwrap(); + store.append(append_request(&run, "b", 1)).await.unwrap(); + store.close_topic(&run, "a", None).await.unwrap(); + assert_eq!(store.notifier().slots.lock().unwrap().len(), 1); + store.close_chain(&run).await.unwrap(); + assert_eq!(store.notifier().slots.lock().unwrap().len(), 0); + } +} diff --git a/crates/streams/src/service.rs b/crates/streams/src/service.rs index e45645cb4..733ce915e 100644 --- a/crates/streams/src/service.rs +++ b/crates/streams/src/service.rs @@ -1,11 +1,13 @@ //! The stream calls lang makes, served in process. use crate::{ - OwnerClient, StreamError, StreamResult, StreamStore, Streams, StreamsOptions, + DEFAULT_MAX_NOTIFIERS, Notifier, NotifierClient, NotifyingStore, OwnerClient, StreamError, + StreamResult, StreamStore, Streams, StreamsOptions, nexus::{NexusCalls, NexusOptions}, proto::{ - AppendRequest, CloseRequest, DeleteOwnerRequest, LatestRequest, NexusCall, - NexusResumeRequest, ReadRequest, StreamFailure, StreamStoreConfig, stream_store_config, + AppendRequest, CloseRequest, DeleteOwnerRequest, FlushNotificationsRequest, + FlushNotificationsResponse, LatestRequest, NexusCall, NexusResumeRequest, ReadRequest, + StreamFailure, StreamStoreConfig, stream_store_config, }, }; use prost::Message; @@ -38,10 +40,21 @@ impl std::fmt::Debug for StreamService { } impl StreamService { - /// Connects to the store `config` names, asking `owner` about streams' owners. + /// Connects to the store `config` names, asking `owner` about streams' owners. Refuses a + /// config that turns notifications on, since there's no client to send them with. pub async fn connect( config: StreamStoreConfig, owner: Arc, + ) -> StreamResult { + Self::connect_with_notifier(config, owner, None).await + } + + /// Like [StreamService::connect], sending stream notifications through `notifier` when the + /// config turns them on. + pub async fn connect_with_notifier( + config: StreamStoreConfig, + owner: Arc, + notifier: Option>, ) -> StreamResult { let (store, retention): (Arc, Duration) = match config.store { Some(stream_store_config::Store::Memory(_)) => { @@ -54,6 +67,21 @@ impl StreamService { )); } }; + let store: Arc = if config.notify_on_append { + let client = notifier.ok_or_else(|| { + StreamError::refused("stream notifications need a client to send them with") + })?; + let max_notifiers = match config.max_notifiers { + 0 => DEFAULT_MAX_NOTIFIERS, + max => max as usize, + }; + Arc::new(NotifyingStore::new( + store, + Arc::new(Notifier::new(client, max_notifiers)), + )) + } else { + store + }; Ok(Self::new( store, owner, @@ -110,6 +138,14 @@ impl StreamService { .delete_owner(decode::(rpc, request)?) .await .map(|r| r.encode_to_vec()), + "FlushNotifications" => { + decode::(rpc, request)?; + // A close retries until the server takes it, so a stopping process waits only + // so long for one against a server that's down. + let _ = tokio::time::timeout(crate::FLUSH_LIMIT, self.store.flush_notifications()) + .await; + Ok(FlushNotificationsResponse {}.encode_to_vec()) + } // A Nexus call answers with a step, its failure included, so lang hands the failure // to its Nexus handler as it is. "NexusAppend" => Ok(self @@ -235,6 +271,7 @@ mod tests { StreamService::connect( StreamStoreConfig { store: Some(stream_store_config::Store::Memory(MemoryStoreConfig {})), + ..Default::default() }, Arc::new(Running), ) @@ -436,7 +473,7 @@ mod tests { #[tokio::test] async fn a_config_that_names_no_store_is_refused() { - let error = StreamService::connect(StreamStoreConfig { store: None }, Arc::new(Running)) + let error = StreamService::connect(StreamStoreConfig::default(), Arc::new(Running)) .await .unwrap_err(); assert_eq!(error.kind, StreamFailureKind::Refused); @@ -448,6 +485,7 @@ mod tests { let error = StreamService::connect( StreamStoreConfig { store: Some(stream_store_config::Store::Redis(Default::default())), + ..Default::default() }, Arc::new(Running), ) @@ -456,6 +494,207 @@ mod tests { assert_eq!(error.kind, StreamFailureKind::Unsupported); } + /// Records the notifications it was sent. + #[derive(Default)] + struct Recorder(std::sync::Mutex>); + + #[async_trait::async_trait] + impl NotifierClient for Recorder { + async fn notify( + &self, + notification: crate::StreamNotification, + ) -> Result<(), crate::NotifyError> { + self.0.lock().unwrap().push(notification); + Ok(()) + } + } + + async fn service_notifying(notify_on_append: bool, recorder: Arc) -> StreamService { + StreamService::connect_with_notifier( + StreamStoreConfig { + store: Some(stream_store_config::Store::Memory(MemoryStoreConfig {})), + notify_on_append, + ..Default::default() + }, + Arc::new(Running), + Some(recorder), + ) + .await + .unwrap() + } + + async fn flush(service: &StreamService) { + let response = service + .call( + "FlushNotifications", + &FlushNotificationsRequest {}.encode_to_vec(), + ) + .await + .unwrap(); + FlushNotificationsResponse::decode(response.as_slice()).unwrap(); + } + + #[tokio::test] + async fn an_append_notifies_its_streams_notifier_once_notifications_are_on() { + let recorder = Arc::new(Recorder::default()); + let service = service_notifying(true, recorder.clone()).await; + let appended = AppendResponse::decode( + service + .call("Append", &append(1, &["a", "b"])) + .await + .unwrap() + .as_slice(), + ) + .unwrap(); + flush(&service).await; + let sent = recorder.0.lock().unwrap().clone(); + assert_eq!(sent.len(), 1); + assert_eq!( + sent[0].chain.first_run_id, "run-1", + "the chain the append wrote to" + ); + assert_eq!(sent[0].topic, "out"); + assert_eq!(sent[0].position, appended.last_cursor); + assert_eq!(sent[0].counter, 2); + } + + #[tokio::test] + async fn notifications_are_off_by_default() { + let recorder = Arc::new(Recorder::default()); + let service = service_notifying(false, recorder.clone()).await; + service.call("Append", &append(1, &["a"])).await.unwrap(); + flush(&service).await; + assert!(recorder.0.lock().unwrap().is_empty()); + } + + #[tokio::test] + async fn notifications_without_a_client_are_refused() { + let error = StreamService::connect( + StreamStoreConfig { + store: Some(stream_store_config::Store::Memory(MemoryStoreConfig {})), + notify_on_append: true, + ..Default::default() + }, + Arc::new(Running), + ) + .await + .unwrap_err(); + assert_eq!(error.kind, StreamFailureKind::Refused); + } + + /// A store whose notifications never go out, as when the server is down and a close retries. + struct NeverFlushes(crate::MemoryStore); + + #[async_trait::async_trait] + impl StreamStore for NeverFlushes { + fn name(&self) -> &str { + self.0.name() + } + async fn append( + &self, + r: crate::proto::StoreAppendRequest, + ) -> StreamResult { + self.0.append(r).await + } + async fn read( + &self, + r: crate::proto::StoreReadRequest, + ) -> StreamResult { + self.0.read(r).await + } + async fn latest( + &self, + r: crate::proto::StoreLatestRequest, + ) -> StreamResult { + self.0.latest(r).await + } + async fn record_at( + &self, + c: &crate::proto::ChainId, + t: &str, + p: &str, + ) -> StreamResult>> { + self.0.record_at(c, t, p).await + } + async fn close_chain(&self, c: &crate::proto::ChainId) -> StreamResult<()> { + self.0.close_chain(c).await + } + async fn close_topic( + &self, + c: &crate::proto::ChainId, + t: &str, + r: Option, + ) -> StreamResult<()> { + self.0.close_topic(c, t, r).await + } + async fn delete_owner( + &self, + r: DeleteOwnerRequest, + ) -> StreamResult { + self.0.delete_owner(r).await + } + async fn flush_notifications(&self) { + std::future::pending::<()>().await; + } + } + + #[tokio::test(start_paused = true)] + async fn a_flush_waits_only_so_long_for_notifications_that_never_go_out() { + let service = StreamService::new( + Arc::new(NeverFlushes(crate::MemoryStore::new())), + Arc::new(Running), + StreamsOptions::default(), + ); + let started = tokio::time::Instant::now(); + flush(&service).await; + assert_eq!( + started.elapsed(), + crate::FLUSH_LIMIT, + "a stopping process mustn't hang on a close retrying against a server that's down" + ); + } + + #[tokio::test] + async fn a_nexus_append_notifies_its_streams_notifier() { + use crate::proto::{NexusCall, NexusStep, nexus_step}; + let recorder = Arc::new(Recorder::default()); + let service = service_notifying(true, recorder.clone()).await; + let input = serde_json::json!({ + "stream": {"kind": "workflow", "workflow_id": "wf", "topic": "t"}, + "producer_id": "p", + "attempt": 1, + "sequence": 1, + "finish": true, + }); + let answer = service + .call( + "NexusAppend", + &NexusCall { + input: serde_json::to_vec(&input).unwrap(), + namespace: "ns".to_string(), + ..Default::default() + } + .encode_to_vec(), + ) + .await + .unwrap(); + let step = NexusStep::decode(answer.as_slice()).unwrap(); + let Some(nexus_step::Step::Output(output)) = step.step else { + panic!("a finish needs no codec work, so it answers at once"); + }; + let output: serde_json::Value = serde_json::from_slice(&output).unwrap(); + flush(&service).await; + let sent = recorder.0.lock().unwrap().clone(); + assert_eq!( + sent.len(), + 1, + "the Nexus append goes through the notifying store" + ); + assert_eq!(sent[0].chain.first_run_id, "run-1"); + assert_eq!(sent[0].topic, "t"); + assert_eq!(sent[0].position, output["cursor"].as_str().unwrap()); + } + #[tokio::test] async fn the_nexus_calls_go_through_the_one_call_entry() { use crate::proto::{NexusCall, NexusResumeRequest, NexusStep, nexus_step}; diff --git a/crates/streams/src/store.rs b/crates/streams/src/store.rs index fc0978f77..97a60ab35 100644 --- a/crates/streams/src/store.rs +++ b/crates/streams/src/store.rs @@ -86,6 +86,10 @@ pub trait StreamStore: Send + Sync { /// Deletes every stream of one owner, across all its chains. async fn delete_owner(&self, request: DeleteOwnerRequest) -> StreamResult; + + /// Waits until the notifications this store sent are out. A store that sends none returns at + /// once. + async fn flush_notifications(&self) {} } #[cfg(test)] diff --git a/crates/streams/tests/conformance.rs b/crates/streams/tests/conformance.rs index ff6176a59..4d2d05be2 100644 --- a/crates/streams/tests/conformance.rs +++ b/crates/streams/tests/conformance.rs @@ -15,8 +15,9 @@ use std::{ }; use temporalio_common::protos::temporal::api::common::v1::Payload; use temporalio_streams::{ - BEGINNING, END, MemoryStore, ReadTarget, StreamResult, StreamStore, mint_cursor, proto::*, - read_page, stored_append_record, stored_output_record, stream_hash, workflow_producer_id, + BEGINNING, END, MemoryStore, Notifier, NotifierClient, NotifyError, NotifyingStore, ReadTarget, + StreamNotification, StreamResult, StreamStore, mint_cursor, proto::*, read_page, + stored_append_record, stored_output_record, stream_hash, workflow_producer_id, }; #[async_trait::async_trait] @@ -73,6 +74,53 @@ async fn memory() -> Option { Some(Memory(Arc::new(MemoryStore::new()))) } +/// Takes every notification and sends nothing. +struct Quiet; + +#[async_trait::async_trait] +impl NotifierClient for Quiet { + async fn notify(&self, _: StreamNotification) -> Result<(), NotifyError> { + Ok(()) + } +} + +/// Another case's store behind `NotifyingStore`, which every process with a client runs, so a +/// method the wrapper doesn't forward fails the contract. +struct Notifying { + inner: C, + store: Arc, +} + +async fn notifying( + inner: impl std::future::Future>, +) -> Option> { + let inner = inner.await?; + let store = Arc::new(NotifyingStore::new( + inner.store(), + Arc::new(Notifier::new(Arc::new(Quiet), 10)), + )); + Some(Notifying { inner, store }) +} + +#[async_trait::async_trait] +impl Case for Notifying { + fn store(&self) -> Arc { + self.store.clone() + } + + fn chain(&self) -> ChainId { + self.inner.chain() + } + + async fn drop_oldest(&self, chain: &ChainId, topic: &str, keep: usize) { + self.inner.drop_oldest(chain, topic, keep).await; + } + + async fn expire_log(&self, chain: &ChainId, topic: &str) { + self.inner.expire_log(chain, topic).await; + } +} + #[cfg(feature = "redis")] mod on_redis { use super::*; @@ -1217,48 +1265,8 @@ macro_rules! conformance { }; } -mod memory { - conformance!(super::memory(); - append_read_roundtrip, - a_batch_lands_in_order_and_names_its_last_record, - an_empty_append_is_refused, - an_append_without_a_digest_is_refused, - latest_is_empty_on_an_empty_topic, - a_cursor_resumes_strictly_after_its_record, - latest_positions_a_reader_at_the_end, - end_reads_only_what_arrives_after_the_read_starts, - a_read_waits_no_longer_than_asked, - dropping_a_waiting_read_releases_it, - beginning_starts_at_the_oldest_record_still_held, - a_reader_that_falls_behind_retention_is_told, - an_empty_read_from_the_beginning_learns_of_a_trim, - an_empty_read_from_the_beginning_with_nothing_trimmed_goes_on, - rule_11_2_an_empty_read_goes_on_when_a_quiet_log_comes_back, - a_retried_append_returns_the_original_positions, - a_retry_with_other_ciphertext_still_deduplicates, - a_divergent_retry_is_refused, - a_sequence_below_the_newest_is_refused, - rule_9_1_a_batch_inside_the_held_batch_is_refused, - rule_64_1_a_gap_in_the_sequence_is_accepted, - producers_and_attempts_dedupe_apart, - a_new_attempt_supersedes_the_old_one, - a_read_resumed_before_a_new_attempt_reports_it, - a_workflow_rerun_supersedes_its_failed_attempt, - a_record_that_does_not_parse_fails_with_its_cursor, - finish_is_a_record_of_its_own, - topics_and_chains_are_apart, - a_cursor_from_another_stream_or_store_is_refused, - a_closed_chain_refuses_new_batches_but_answers_a_repeat, - a_closed_topic_leaves_the_others_open, - a_closed_topic_answers_a_repeat_and_ends_reads, - a_refused_append_leaves_no_trace_of_its_topic, - deleting_an_owner_drops_every_chain_of_it, - ); -} - /// Every case but owner deletes, which need a rule of their own under the documented ACL. -#[cfg(feature = "redis")] -macro_rules! redis_cases { +macro_rules! store_cases { ($case:expr) => { conformance!($case; append_read_roundtrip, @@ -1298,9 +1306,21 @@ macro_rules! redis_cases { }; } +mod memory { + store_cases!(super::memory()); + conformance!(super::memory(); deleting_an_owner_drops_every_chain_of_it); +} + +mod notifying_memory { + store_cases!(super::notifying(super::memory())); + conformance!(super::notifying(super::memory()); + deleting_an_owner_drops_every_chain_of_it, + ); +} + #[cfg(feature = "redis")] mod standalone_redis { - redis_cases!(super::on_redis::redis("STREAMS_REDIS_URL", false)); + store_cases!(super::on_redis::redis("STREAMS_REDIS_URL", false)); conformance!(super::on_redis::redis("STREAMS_REDIS_URL", false); deleting_an_owner_drops_every_chain_of_it, ); @@ -1308,16 +1328,24 @@ mod standalone_redis { #[cfg(feature = "redis")] mod cluster_redis { - redis_cases!(super::on_redis::redis("STREAMS_REDIS_CLUSTER_URL", true)); + store_cases!(super::on_redis::redis("STREAMS_REDIS_CLUSTER_URL", true)); // A Workflow id's chains hash to different slots, so the delete reaches every primary. conformance!(super::on_redis::redis("STREAMS_REDIS_CLUSTER_URL", true); deleting_an_owner_drops_every_chain_of_it, ); } +#[cfg(feature = "redis")] +mod notifying_redis { + store_cases!(super::notifying(super::on_redis::redis( + "STREAMS_REDIS_URL", + false + ))); +} + #[cfg(feature = "redis")] mod documented_acl_redis { - redis_cases!(super::on_redis::redis_as_documented_user( + store_cases!(super::on_redis::redis_as_documented_user( "STREAMS_REDIS_URL", false, &[] @@ -1331,7 +1359,7 @@ mod documented_acl_redis { #[cfg(feature = "redis")] mod documented_acl_cluster_redis { // A cluster client asks the topology first, so the documented rules must allow it. - redis_cases!(super::on_redis::redis_as_documented_user( + store_cases!(super::on_redis::redis_as_documented_user( "STREAMS_REDIS_CLUSTER_URL", true, &[] diff --git a/crates/streams/tests/redis_store.rs b/crates/streams/tests/redis_store.rs index 6a2da4092..681cc68da 100644 --- a/crates/streams/tests/redis_store.rs +++ b/crates/streams/tests/redis_store.rs @@ -1447,3 +1447,72 @@ async fn a_delete_reaches_every_primary_of_a_cluster() { assert!(keys.is_empty(), "{keys:?}"); } } + +/// Records the stream notifications it was sent. +#[derive(Default)] +struct Recorder(std::sync::Mutex>); + +#[async_trait::async_trait] +impl temporalio_streams::NotifierClient for Recorder { + async fn notify( + &self, + notification: temporalio_streams::StreamNotification, + ) -> Result<(), temporalio_streams::NotifyError> { + self.0.lock().unwrap().push(notification); + Ok(()) + } +} + +#[tokio::test] +async fn notification_counters_come_from_redis_entry_ids_and_grow_with_the_stream() { + let Some(setup) = setup().await else { return }; + let recorder = Arc::new(Recorder::default()); + let store = temporalio_streams::NotifyingStore::new( + Arc::new(setup.store), + Arc::new(temporalio_streams::Notifier::new(recorder.clone(), 10)), + ); + let chain = chain(); + let mut positions = vec![]; + for sequence in [1, 3] { + let digest = [sequence as u8]; + let landed = store + .append(append(&chain, "p", sequence, 2, &digest)) + .await + .unwrap(); + positions.push(landed.last_position); + store.flush_notifications().await; + } + // A Workflow's output is an append as its run's producer, so it notifies the same way. + let landed = store + .append(append( + &chain, + &temporalio_streams::workflow_producer_id("run-1"), + 1, + 1, + &[9], + )) + .await + .unwrap(); + positions.push(landed.last_position); + store.flush_notifications().await; + + let counters: Vec = recorder + .0 + .lock() + .unwrap() + .iter() + .map(|n| n.counter) + .collect(); + let expected: Vec = positions + .iter() + .map(|position| { + let (milliseconds, sequence) = position.split_once('-').expect("an entry id"); + (milliseconds.parse::().unwrap() << 20) | sequence.parse::().unwrap() + }) + .collect(); + assert_eq!(counters, expected); + assert!( + counters.windows(2).all(|pair| pair[0] < pair[1]), + "{counters:?}" + ); +}