diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index 17ddf342a..e7f3d29ce 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/crates/sdk-core/CHANGELOG.md @@ -158,6 +158,10 @@ relevant information. `{:!a::}`, which no Workflow id makes. `DeleteOwner` takes the Activity kind. `OwnerClient::describe` takes an `OwnerId`, and `OwnerDescription.status` is an `OwnerStatus`. A paused Workflow run now counts as running for a read pinned to it. +* Experimental: notifier calls name an Activity-owned stream by its kind, its Activity id and its + run. Notifications, closes, Workflow reader attaches, detaches and `WaitForNotification` polls + all do, and a stream driver turns an Activity log key into that reference. Their cursors hash + under the kind `activity`. ### 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 0771dc741..2db223dd4 100644 --- a/crates/sdk-core/src/streams.rs +++ b/crates/sdk-core/src/streams.rs @@ -170,8 +170,17 @@ fn poll_answer(response: PollStreamNotifierResponse) -> NotifierPoll { /// The notifier of `chain`'s stream on `topic`. 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. +/// its own. An Activity run is its own chain, so its notifier is keyed by the run. fn chain_reference(chain: &ChainId, topic: String) -> StreamReference { + if chain.owner_kind == proto::StreamOwnerKind::Activity as i32 { + return StreamReference { + owner_kind: StreamOwnerKind::Activity as i32, + activity_id: chain.activity_id.clone(), + run_id: chain.first_run_id.clone(), + topic, + ..Default::default() + }; + } StreamReference { owner_kind: StreamOwnerKind::Workflow as i32, workflow_id: chain.workflow_id.clone(), @@ -547,6 +556,34 @@ mod tests { assert!(request.close_result.is_none()); } + #[test] + fn a_notification_of_an_activity_stream_names_the_activity_run() { + let request = notify_request(StreamNotification { + chain: proto::ChainId { + namespace: "ns".to_string(), + first_run_id: "act-run".to_string(), + owner_kind: proto::StreamOwnerKind::Activity as i32, + activity_id: "gen".to_string(), + ..Default::default() + }, + topic: "tokens".to_string(), + position: "cursor".to_string(), + counter: 7, + close: Some(CloseOutcome::Failure(Default::default())), + }); + assert_eq!( + request.stream_ref, + Some(StreamReference { + owner_kind: StreamOwnerKind::Activity as i32, + workflow_id: String::new(), + run_id: "act-run".to_string(), + topic: "tokens".to_string(), + activity_id: "gen".to_string(), + }) + ); + assert!(request.close && request.close_failure.is_some()); + } + #[test] fn a_batch_of_notifications_goes_out_once_per_namespace() { let notification = |namespace: &str, workflow_id: &str, counter: i64| StreamNotification { diff --git a/crates/streams/src/driver.rs b/crates/streams/src/driver.rs index 3f1976464..163aa6d20 100644 --- a/crates/streams/src/driver.rs +++ b/crates/streams/src/driver.rs @@ -9,8 +9,8 @@ use crate::{ NotificationEntry, NotificationLog, NotifierClient, NotifyError, StreamError, - StreamNotification, StreamResult, StreamStore, WORKFLOW_OWNER_KIND, mint_cursor, - progress_counter, proto::ChainId, stream_hash, + StreamNotification, StreamResult, StreamStore, chain_owner, mint_cursor, progress_counter, + proto::ChainId, stream_hash, }; use std::{ collections::{BTreeMap, HashMap}, @@ -412,6 +412,7 @@ impl StreamDriver { let Some(counter) = progress_counter(&entry.position) else { tracing::warn!( workflow_id = %entry.chain.workflow_id, + activity_id = %entry.chain.activity_id, topic = %entry.topic, position = %entry.position, "No notifier counter comes from this position, so it isn't notified" @@ -435,12 +436,8 @@ impl StreamDriver { .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 (kind, id) = chain_owner(&chain); + let hash = stream_hash(&chain.namespace, kind, id, &topic); let notification = StreamNotification { position: mint_cursor(self.store.name(), &hash, &position), chain, @@ -454,9 +451,9 @@ impl StreamDriver { } } -/// A chain as a map key. +/// A chain as a map key: its namespace, Workflow id, first run, owner kind and Activity id. #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)] -struct ChainKey(String, String, String); +struct ChainKey(String, String, String, i32, String); impl ChainKey { fn of(chain: &ChainId) -> Self { @@ -464,6 +461,8 @@ impl ChainKey { chain.namespace.clone(), chain.workflow_id.clone(), chain.first_run_id.clone(), + chain.owner_kind, + chain.activity_id.clone(), ) } @@ -472,7 +471,8 @@ impl ChainKey { namespace: self.0, workflow_id: self.1, first_run_id: self.2, - ..Default::default() + owner_kind: self.3, + activity_id: self.4, } } } @@ -664,6 +664,46 @@ mod tests { ); } + #[tokio::test] + async fn an_activity_chains_entries_notify_the_activitys_stream() { + let store = Arc::new(MemoryStore::with_notification_log()); + let client = Arc::new(Recorder::default()); + let activity = ChainId { + namespace: "ns".to_string(), + first_run_id: "run-1".to_string(), + owner_kind: crate::proto::StreamOwnerKind::Activity as i32, + activity_id: "a".to_string(), + ..Default::default() + }; + // A Workflow with the same id and run writes another stream. + append(&store, "a", "out", 1).await; + let landed = store + .append(StoreAppendRequest { + chain: Some(activity.clone()), + topic: "out".to_string(), + producer_id: "p".to_string(), + attempt: 1, + sequence: 1, + digest: vec![1], + records: vec![b"record".to_vec(), b"record".to_vec()], + }) + .await + .unwrap(); + driver(&store, &client, "d1").drain(0).await.unwrap(); + let calls = client.calls.lock().unwrap().clone(); + assert_eq!(calls[0].len(), 2, "two streams, {calls:?}"); + let sent = calls[0].iter().find(|n| n.chain == activity).unwrap(); + assert_eq!(sent.counter, 2); + assert_eq!( + sent.position, + mint_cursor( + "memory", + &stream_hash("ns", crate::ACTIVITY_OWNER_KIND, "a", "out"), + &landed.last_position + ) + ); + } + #[tokio::test] async fn a_failed_call_leaves_the_entries_for_the_next_drain() { let store = Arc::new(MemoryStore::with_notification_log()); diff --git a/crates/streams/src/notify.rs b/crates/streams/src/notify.rs index 792d2981a..bf0056359 100644 --- a/crates/streams/src/notify.rs +++ b/crates/streams/src/notify.rs @@ -4,7 +4,7 @@ //! hint that there is something new to read. The records stay on the read path. use crate::{ - CloseOutcome, StreamResult, StreamStore, WORKFLOW_OWNER_KIND, mint_cursor, + CloseOutcome, StreamResult, StreamStore, chain_owner, mint_cursor, proto::{ ChainId, DeleteOwnerRequest, DeleteOwnerResponse, StoreAppendRequest, StoreAppendResponse, StoreLatestRequest, StoreLatestResponse, StoreReadRequest, StoreReadResponse, StoredClose, @@ -214,6 +214,8 @@ struct StreamKey { namespace: String, workflow_id: String, first_run_id: String, + owner_kind: i32, + activity_id: String, topic: String, } @@ -223,6 +225,8 @@ impl StreamKey { namespace: chain.namespace.clone(), workflow_id: chain.workflow_id.clone(), first_run_id: chain.first_run_id.clone(), + owner_kind: chain.owner_kind, + activity_id: chain.activity_id.clone(), topic: topic.to_string(), } } @@ -232,7 +236,8 @@ impl StreamKey { namespace: self.namespace.clone(), workflow_id: self.workflow_id.clone(), first_run_id: self.first_run_id.clone(), - ..Default::default() + owner_kind: self.owner_kind, + activity_id: self.activity_id.clone(), } } } @@ -343,6 +348,7 @@ impl Slot { Err(NotifyError::Failed(error)) if tries_left > 0 && !self.superseded() => { tracing::debug!( workflow_id = %self.key.workflow_id, + activity_id = %self.key.activity_id, topic = %self.key.topic, %error, "A stream notification failed; trying again in {delay:?}" @@ -351,6 +357,7 @@ impl Slot { Err(error) => { tracing::warn!( workflow_id = %self.key.workflow_id, + activity_id = %self.key.activity_id, topic = %self.key.topic, %error, "A stream notification failed. The next write tells the reader again" @@ -452,6 +459,7 @@ impl Notifier { Err(NotifyError::Refused(error) | NotifyError::Unsupported(error)) => { tracing::warn!( workflow_id = %chain.workflow_id, + activity_id = %chain.activity_id, topic, error, "The server refused the close of a stream's notifier" @@ -472,6 +480,7 @@ impl Notifier { } Err(NotifyError::Failed(error)) => tracing::warn!( workflow_id = %chain.workflow_id, + activity_id = %chain.activity_id, topic, error, "Could not close a stream's notifier; trying again in {delay:?}" @@ -685,13 +694,12 @@ impl NotifyingStore { } 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) + let (kind, id) = chain_owner(chain); + mint_cursor( + store.name(), + &stream_hash(&chain.namespace, kind, id, topic), + position, + ) } /// The cursor and counter that tell a reader about the record at `position`. @@ -706,6 +714,7 @@ fn notification_at( let Some(counter) = progress_counter(position) else { tracing::warn!( workflow_id = %chain.workflow_id, + activity_id = %chain.activity_id, topic, position, "No notifier counter comes from this position, so the write isn't notified" @@ -939,6 +948,17 @@ mod tests { } } + /// The chain of run `run_id` of standalone Activity `wf`, which shares the Workflow's id. + fn activity_chain(run_id: &str) -> ChainId { + ChainId { + namespace: "ns".to_string(), + first_run_id: run_id.to_string(), + owner_kind: crate::proto::StreamOwnerKind::Activity as i32, + activity_id: "wf".to_string(), + ..Default::default() + } + } + /// Lets every spawned send run until it waits. async fn settle() { for _ in 0..10 { @@ -1192,13 +1212,44 @@ mod tests { sent[0].position, mint_cursor( "memory", - &stream_hash("ns", WORKFLOW_OWNER_KIND, "wf", "t"), + &stream_hash("ns", crate::WORKFLOW_OWNER_KIND, "wf", "t"), &landed.last_position ), "the position is the cursor a reader resumes from" ); } + #[tokio::test] + async fn an_activity_append_notifies_the_activitys_stream() { + let recorder = Arc::new(Recorder::default()); + let (_, store) = notifying_store(&recorder); + // A Workflow with the same id writes too, and its stream is another one. + store + .append(append_request(&chain("run-1"), "t", 1)) + .await + .unwrap(); + let landed = store + .append(append_request(&activity_chain("run-1"), "t", 1)) + .await + .unwrap(); + store.flush_notifications().await; + let sent = recorder.sent.lock().unwrap().clone(); + assert_eq!(sent.len(), 2, "one notifier per owner, {sent:?}"); + let activity = sent + .iter() + .find(|n| n.chain == activity_chain("run-1")) + .unwrap(); + assert_eq!( + activity.position, + mint_cursor( + "memory", + &stream_hash("ns", crate::ACTIVITY_OWNER_KIND, "wf", "t"), + &landed.last_position + ), + "the cursor hashes under the Activity kind" + ); + } + /// A store whose appends land at a position no counter comes from. struct OddPositions(MemoryStore); @@ -1397,6 +1448,22 @@ mod tests { assert!(store.notifier().slots.lock().unwrap().is_empty()); } + #[tokio::test] + async fn an_activity_close_reaches_its_notifier_with_the_outcome() { + let (client, store) = sees_the_store(None); + let run = activity_chain("act-run"); + store.append(append_request(&run, "t", 1)).await.unwrap(); + store + .close_stream(&run, "t", CloseOutcome::Failure(gave_up()).into()) + .await + .unwrap(); + let sent = client.sent.lock().unwrap().clone(); + let close = sent.last().unwrap(); + assert_eq!(close.chain, run); + assert_eq!(close.close, Some(CloseOutcome::Failure(gave_up()))); + assert_eq!(close.counter, 2, "one above the newest record's counter"); + } + #[tokio::test] async fn a_second_close_tells_the_notifier_the_first() { // A rerun that closes again must not give waiters another outcome than readers get. diff --git a/crates/streams/src/subscribe.rs b/crates/streams/src/subscribe.rs index db49d5035..cd3f50d35 100644 --- a/crates/streams/src/subscribe.rs +++ b/crates/streams/src/subscribe.rs @@ -9,7 +9,7 @@ use crate::{ NotifierClient, NotifyError, OWNER_RECHECK, StreamError, StreamResult, Streams, StreamsOptions, WorkflowAttachment, - handle::{Remembered, address}, + handle::{Remembered, address, chain_of}, proto::{ ChainId, StreamAddress, SubscribeRequest, SubscribeResponse, UnsubscribeRequest, UnsubscribeResponse, WaitForNotificationRequest, WaitForNotificationResponse, @@ -41,7 +41,9 @@ impl Default for Subscriptions { #[derive(Debug, Clone, PartialEq, Eq, Hash)] struct SubscriptionKey { namespace: String, + owner_kind: i32, owner_workflow_id: String, + owner_activity_id: String, owner_run_id: String, topic: String, request_id: String, @@ -51,7 +53,9 @@ impl SubscriptionKey { fn new(stream: &StreamAddress, request_id: &str) -> Self { Self { namespace: stream.namespace.clone(), + owner_kind: stream.owner_kind, owner_workflow_id: stream.workflow_id.clone(), + owner_activity_id: stream.activity_id.clone(), owner_run_id: stream.run_id.clone(), topic: stream.topic.clone(), request_id: request_id.to_string(), @@ -179,12 +183,7 @@ impl Subscriptions { let client = self.client()?; let key = SubscriptionKey::new(stream, &request_id); let chain = match self.attached.get(&key) { - Some(attached) => crate::proto::ChainId { - namespace: stream.namespace.clone(), - workflow_id: stream.workflow_id.clone(), - first_run_id: attached.first_run_id, - ..Default::default() - }, + Some(attached) => chain_of(stream, &attached.first_run_id), None => streams.chain_now(stream).await?, }; client @@ -612,6 +611,53 @@ mod tests { assert_eq!(answer.chain_first_run_id, "chain-2"); } + #[tokio::test] + async fn a_workflow_reader_subscribes_to_a_stream_that_an_activity_owns() { + let setup = setup(); + let activity = StreamAddress { + owner_kind: StreamOwnerKind::Activity as i32, + workflow_id: String::new(), + activity_id: "owner".to_string(), + ..stream() + }; + let activity_chain = ChainId { + namespace: "ns".to_string(), + first_run_id: "chain-1".to_string(), + owner_kind: StreamOwnerKind::Activity as i32, + activity_id: "owner".to_string(), + ..Default::default() + }; + let request = SubscribeRequest { + stream: Some(activity.clone()), + ..subscribe_request("1") + }; + let answer = subscribe(&setup.service, request).await.unwrap(); + assert!(answer.attached); + assert_eq!(answer.chain_first_run_id, "chain-1"); + // A Workflow with the same id owns another stream, so its reader attaches apart. + assert!( + subscribe(&setup.service, subscribe_request("1")) + .await + .unwrap() + .attached + ); + let attached = setup.server.attached.lock().unwrap().clone(); + assert_eq!(attached[0].chain, activity_chain); + assert_eq!(attached[1].chain, chain("chain-1")); + let unsubscribe = UnsubscribeRequest { + stream: Some(activity), + run_id: "reader-run".to_string(), + subscription_id: "1".to_string(), + ..Default::default() + }; + setup + .service + .call("Unsubscribe", &unsubscribe.encode_to_vec()) + .await + .unwrap(); + assert_eq!(setup.server.detached.lock().unwrap()[0].0, activity_chain); + } + async fn wait( setup: &Setup, after_counter: i64, diff --git a/crates/streams/tests/driver.rs b/crates/streams/tests/driver.rs index 46ae80cc6..21c731f90 100644 --- a/crates/streams/tests/driver.rs +++ b/crates/streams/tests/driver.rs @@ -16,8 +16,9 @@ use std::{ }; use temporalio_streams::{ NotifierClient, NotifyError, RedisStore, RedisStoreOptions, StreamDriver, StreamDriverOptions, - StreamNotification, StreamStore, progress_counter, - proto::{ChainId, StoreAppendRequest}, + StreamNotification, StreamStore, chain_owner, mint_cursor, progress_counter, + proto::{ChainId, StoreAppendRequest, StreamOwnerKind}, + stream_hash, }; static IDS: AtomicUsize = AtomicUsize::new(0); @@ -34,10 +35,12 @@ fn unique(prefix: &str) -> String { ) } -/// What a notifier got: each stream's highest counter, and every call. +/// What a notifier got: each stream's highest counter and its position, by owner id and topic, +/// and every call. #[derive(Default)] struct Notifiers { newest: Mutex>, + positions: Mutex>, calls: AtomicUsize, fail: AtomicBool, } @@ -57,11 +60,17 @@ impl NotifierClient for Notifiers { } self.calls.fetch_add(1, Ordering::SeqCst); let mut newest = self.newest.lock().unwrap(); + let mut positions = self.positions.lock().unwrap(); for notification in notifications { - let counter = newest - .entry((notification.chain.workflow_id, notification.topic)) - .or_default(); - *counter = (*counter).max(notification.counter); + let key = ( + chain_owner(¬ification.chain).1.to_string(), + notification.topic, + ); + let counter = newest.entry(key.clone()).or_default(); + if notification.counter > *counter { + *counter = notification.counter; + positions.insert(key, notification.position); + } } Ok(()) } @@ -150,6 +159,84 @@ async fn write( expected } +/// Appends one batch to the runs of `streams` standalone Activities, `act-`, and answers +/// each Activity stream's newest counter and its cursor. +async fn write_activities( + store: &RedisStore, + streams: usize, +) -> BTreeMap<(String, String), (i64, String)> { + let mut expected = BTreeMap::new(); + for index in 0..streams { + let id = format!("act-{index}"); + let append = |chain: ChainId| StoreAppendRequest { + chain: Some(chain), + topic: "out".to_string(), + producer_id: "p".to_string(), + attempt: 1, + sequence: 1, + digest: vec![1], + records: vec![b"record".to_vec()], + }; + let landed = store + .append(append(ChainId { + namespace: "ns".to_string(), + first_run_id: "run-1".to_string(), + owner_kind: StreamOwnerKind::Activity as i32, + activity_id: id.clone(), + ..Default::default() + })) + .await + .unwrap(); + let position = mint_cursor( + store.name(), + &stream_hash("ns", "activity", &id, "out"), + &landed.last_position, + ); + expected.insert( + (id, "out".to_string()), + (progress_counter(&landed.last_position).unwrap(), position), + ); + } + expected +} + +async fn activity_streams_reach_their_notifiers(url: &str, cluster: bool) { + let _shared = CLUSTER.read().await; + let prefix = unique("driver-activity"); + 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_activities(&store, 100).await; + let counters = expected + .iter() + .map(|(stream, (counter, _))| (stream.clone(), *counter)) + .collect(); + notifiers.reach(&counters, Duration::from_secs(20)).await; + running.abort(); + let positions = notifiers.positions.lock().unwrap().clone(); + for (stream, (_, position)) in &expected { + assert_eq!(positions.get(stream), Some(position), "{stream:?}"); + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn activity_streams_on_a_server_reach_their_notifiers() { + let Some(url) = url("STREAMS_REDIS_URL") else { + return; + }; + activity_streams_reach_their_notifiers(&url, false).await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn activity_streams_in_a_cluster_reach_their_notifiers() { + let Some(url) = url("STREAMS_REDIS_CLUSTER_URL") else { + return; + }; + activity_streams_reach_their_notifiers(&url, true).await; +} + fn url(variable: &str) -> Option { let url = std::env::var(variable).ok(); if url.is_none() {