[Jetty] Sent notifier calls for streams that an Activity owns.

crates/sdk-core/CHANGELOG.md

@@ -158,6 +158,10 @@ relevant information.
   `{<namespace>:!a:<activity id>:<run id>}`, 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

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 {

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,16 +451,18 @@ 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 {
         Self(
             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());

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.