From 9fc263ae8de6a302591932b9edc6392c7b60e326 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Wed, 14 Oct 2026 23:26:44 -0700 Subject: [PATCH] Sent notifier calls for streams that an Activity owns. The notifier keys an Activity stream by its Activity id and run, so every notify, close, attach and poll names that reference, and cursors hash under the Activity kind. --- crates/sdk-core/CHANGELOG.md | 4 ++ crates/sdk-core/src/streams.rs | 39 +++++++++++- crates/streams/src/driver.rs | 62 ++++++++++++++++---- crates/streams/src/notify.rs | 87 +++++++++++++++++++++++---- crates/streams/src/subscribe.rs | 60 ++++++++++++++++--- crates/streams/tests/driver.rs | 101 +++++++++++++++++++++++++++++--- 6 files changed, 317 insertions(+), 36 deletions(-) diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index 17ddf342ae..e7f3d29ce3 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 7329f7a3c0..5bb506e9ff 100644 --- a/crates/sdk-core/src/streams.rs +++ b/crates/sdk-core/src/streams.rs @@ -140,8 +140,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(), @@ -485,6 +494,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 d396b69a5e..ef2b6325ac 100644 --- a/crates/streams/src/driver.rs +++ b/crates/streams/src/driver.rs @@ -8,8 +8,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}, @@ -215,6 +215,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" @@ -233,12 +234,8 @@ impl StreamDriver { .into_iter() .map(|((chain, topic), (counter, position))| { let chain = chain.into_chain(); - let hash = stream_hash( - &chain.namespace, - WORKFLOW_OWNER_KIND, - &chain.workflow_id, - &topic, - ); + let (kind, id) = chain_owner(&chain); + let hash = stream_hash(&chain.namespace, kind, id, &topic); StreamNotification { position: mint_cursor(self.store.name(), &hash, &position), chain, @@ -251,9 +248,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 { @@ -261,6 +258,8 @@ impl ChainKey { chain.namespace.clone(), chain.workflow_id.clone(), chain.first_run_id.clone(), + chain.owner_kind, + chain.activity_id.clone(), ) } @@ -269,7 +268,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, } } } @@ -400,6 +400,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 982d6b6fa4..c3d2970eb3 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, @@ -199,6 +199,8 @@ struct StreamKey { namespace: String, workflow_id: String, first_run_id: String, + owner_kind: i32, + activity_id: String, topic: String, } @@ -208,6 +210,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(), } } @@ -217,7 +221,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(), } } } @@ -328,6 +333,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:?}" @@ -336,6 +342,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" @@ -435,6 +442,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" @@ -443,6 +451,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:?}" @@ -656,13 +665,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`. @@ -677,6 +685,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" @@ -910,6 +919,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 { @@ -1163,13 +1183,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); @@ -1368,6 +1419,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 db49d50359..cd3f50d350 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 46ae80cc67..21c731f90c 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() {