[Jetty] Kept and checked streams that a standalone Activity owns.

crates/sdk-core/CHANGELOG.md

@@ -149,6 +149,15 @@ relevant information.
   `owner_kind` and `activity_id`. `ActivityProducer` gains `attempt_started_time`. The vendored
   API's `StreamReference` gains `activity_id` and its owner kind gains `ACTIVITY`. Core still
   refuses the Activity kind as `UNSUPPORTED`.
+* Experimental: a standalone Activity owns streams. A `StreamAddress` of the `ACTIVITY` kind names
+  the Activity id and, optionally, its run. One Activity run is a chain of one run, so every chain
+  rule holds with the run as `first_run_id`. Core checks the owner with
+  `DescribeActivityExecution`. Any terminal status ends the run's streams, and retries, backoff,
+  pause and reset keep them open. A new run of the Activity id ends the earlier run's streams.
+  Cursors hash under the kind `activity`. The Redis keys of an Activity run have the hash tag
+  `{<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.
 
 ### Breaking Changes
 * Replay detects changes to the service or operation of scheduled Nexus operations, reporting

crates/sdk-core/src/streams.rs

@@ -10,14 +10,15 @@ use temporalio_common::protos::temporal::api::{
     enums::v1::StreamOwnerKind,
     stream::v1::{self, StreamReference, WorkflowWakeTarget},
     workflowservice::v1::{
-        AttachStreamCallbackRequest, DescribeWorkflowExecutionRequest,
+        AttachStreamCallbackRequest, DescribeActivityExecutionRequest,
+        DescribeActivityExecutionResponse, DescribeWorkflowExecutionRequest,
         DescribeWorkflowExecutionResponse, DetachStreamCallbackRequest, NotifyStreamRequest,
         NotifyStreamsRequest, PollStreamNotifierRequest, PollStreamNotifierResponse,
     },
 };
 use temporalio_streams::{
     CloseOutcome, NotifierClient, NotifierPoll, NotifyError, OwnerClient, OwnerDescription,
-    OwnerError, StreamNotification, WorkflowAttachment, proto::ChainId,
+    OwnerError, OwnerId, OwnerStatus, StreamNotification, WorkflowAttachment, proto::ChainId,
 };
 pub use temporalio_streams::{StreamError, StreamResult, StreamService, StreamStore, proto};
 use tonic::IntoRequest;
@@ -282,25 +283,46 @@ impl OwnerClient for ConnectionOwnerClient {
     async fn describe(
         &self,
         namespace: &str,
-        workflow_id: &str,
+        owner: OwnerId<'_>,
         run_id: &str,
     ) -> Result<OwnerDescription, OwnerError> {
-        let response = self
-            .connection
-            .workflow_service()
-            .describe_workflow_execution(
-                DescribeWorkflowExecutionRequest {
-                    namespace: namespace.to_string(),
-                    execution: Some(WorkflowExecution {
-                        workflow_id: workflow_id.to_string(),
-                        run_id: run_id.to_string(),
-                    }),
-                }
-                .into_request(),
-            )
-            .await
-            .map_err(owner_error)?;
-        owner_description(response.into_inner())
+        match owner {
+            OwnerId::Workflow(workflow_id) => {
+                let response = self
+                    .connection
+                    .workflow_service()
+                    .describe_workflow_execution(
+                        DescribeWorkflowExecutionRequest {
+                            namespace: namespace.to_string(),
+                            execution: Some(WorkflowExecution {
+                                workflow_id: workflow_id.to_string(),
+                                run_id: run_id.to_string(),
+                            }),
+                        }
+                        .into_request(),
+                    )
+                    .await
+                    .map_err(owner_error)?;
+                owner_description(response.into_inner())
+            }
+            OwnerId::Activity(activity_id) => {
+                let response = self
+                    .connection
+                    .workflow_service()
+                    .describe_activity_execution(
+                        DescribeActivityExecutionRequest {
+                            namespace: namespace.to_string(),
+                            activity_id: activity_id.to_string(),
+                            run_id: run_id.to_string(),
+                            ..Default::default()
+                        }
+                        .into_request(),
+                    )
+                    .await
+                    .map_err(owner_error)?;
+                activity_owner_description(response.into_inner())
+            }
+        }
     }
 }
 
@@ -342,12 +364,36 @@ fn owner_description(
         info.first_run_id.clone()
     };
     Ok(OwnerDescription {
-        status: info.status(),
+        status: OwnerStatus::of_workflow(info.status()),
         run_id,
         first_run_id,
     })
 }
 
+/// A standalone Activity run is a chain of one run, so it is its own first run.
+fn activity_owner_description(
+    response: DescribeActivityExecutionResponse,
+) -> Result<OwnerDescription, OwnerError> {
+    let info = response
+        .info
+        .ok_or_else(|| OwnerError::Failed("the server described no Activity".to_string()))?;
+    let run_id = if response.run_id.is_empty() {
+        info.run_id.clone()
+    } else {
+        response.run_id
+    };
+    if run_id.is_empty() {
+        return Err(OwnerError::Failed(
+            "the server described an Activity with no run id".to_string(),
+        ));
+    }
+    Ok(OwnerDescription {
+        status: OwnerStatus::of_activity(info.status()),
+        first_run_id: run_id.clone(),
+        run_id,
+    })
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -377,9 +423,52 @@ mod tests {
             OwnerDescription {
                 run_id: "run-2".to_string(),
                 first_run_id: "run-1".to_string(),
-                status: WorkflowExecutionStatus::ContinuedAsNew,
+                status: OwnerStatus::ContinuedAsNew,
+            }
+        );
+    }
+
+    #[test]
+    fn an_activity_run_is_its_own_chain_and_any_terminal_status_ends_it() {
+        use temporalio_common::protos::temporal::api::{
+            activity::v1::ActivityExecutionInfo, enums::v1::ActivityExecutionStatus,
+        };
+        let described = |status: ActivityExecutionStatus| DescribeActivityExecutionResponse {
+            run_id: "act-run".to_string(),
+            info: Some(ActivityExecutionInfo {
+                activity_id: "gen".to_string(),
+                run_id: "act-run".to_string(),
+                status: status as i32,
+                ..Default::default()
+            }),
+            ..Default::default()
+        };
+        assert_eq!(
+            activity_owner_description(described(ActivityExecutionStatus::Paused)).unwrap(),
+            OwnerDescription {
+                run_id: "act-run".to_string(),
+                first_run_id: "act-run".to_string(),
+                status: OwnerStatus::Running,
             }
         );
+        for status in [
+            ActivityExecutionStatus::Completed,
+            ActivityExecutionStatus::Failed,
+            ActivityExecutionStatus::Canceled,
+            ActivityExecutionStatus::Terminated,
+            ActivityExecutionStatus::TimedOut,
+        ] {
+            assert!(
+                activity_owner_description(described(status))
+                    .unwrap()
+                    .chain_ended(),
+                "{status:?}"
+            );
+        }
+        assert!(matches!(
+            activity_owner_description(DescribeActivityExecutionResponse::default()),
+            Err(OwnerError::Failed(_))
+        ));
     }
 
     #[test]

crates/streams/docs/redis.md

@@ -71,9 +71,15 @@ slides the expiry of the log and its meta, so a stream dies a retention after it
 and nobody cleans up when the Workflow closes. The meta outlives its log by a 30-day grace, as
 the tombstone that tells a log that expired from one that never existed.
 
-**Deletes.** Deleting an owner's streams scans every primary for the Workflow id's keys. `SCAN`
-makes no promise about keys that move to another node while it runs, so a delete during a
-cluster reshard can miss some. Run it again once the reshard is done.
+**Activity owners.** A standalone Activity run is a chain of one run. Its keys have the hash tag
+`{<namespace>:!a:<activity id>:<run id>}`, and every other part of the layout is the same. Each
+part escapes `!`, so no Workflow id makes that tag, and a Workflow's delete pattern never matches
+an Activity's keys. The tag starts with the namespace, so the documented ACL covers it.
+
+**Deletes.** Deleting an owner's streams scans every primary for the owner id's keys, across all
+the Workflow's chains or all the Activity's runs. `SCAN` makes no promise about keys that move to
+another node while it runs, so a delete during a cluster reshard can miss some. Run it again once
+the reshard is done.
 
 **Server.** Redis 7.0 or later is required, and the store refuses an older server when it
 connects. At the same moment it reads `maxmemory-policy` on every primary and logs a warning