diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index b53a33df68..17ddf342ae 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/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 + `{:!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. ### 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 7ad6d5d592..0771dc741a 100644 --- a/crates/sdk-core/src/streams.rs +++ b/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 { - 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 { + 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] diff --git a/crates/streams/docs/redis.md b/crates/streams/docs/redis.md index 0d4b3d91a6..85d7bb35a8 100644 --- a/crates/streams/docs/redis.md +++ b/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 +`{:!a::}`, 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 diff --git a/crates/streams/src/handle.rs b/crates/streams/src/handle.rs index 2b0d185caa..e38fa6a726 100644 --- a/crates/streams/src/handle.rs +++ b/crates/streams/src/handle.rs @@ -5,8 +5,8 @@ use crate::{ BEGINNING, END, MAX_CLOSE_BYTES, ReadTarget, StreamError, StreamResult, StreamStore, - WORKFLOW_OWNER_KIND, activity_producer_id, check_append_digest, cursor_position, mint_cursor, - owner::{OwnerClient, OwnerDescription, OwnerError}, + activity_producer_id, chain_owner, check_append_digest, cursor_position, mint_cursor, + owner::{OwnerClient, OwnerDescription, OwnerError, OwnerId, OwnerStatus}, proto::{ ActivityProducer, AppendRequest, AppendResponse, ChainId, CloseRequest, CloseResponse, DeleteOwnerRequest, DeleteOwnerResponse, LatestRequest, LatestResponse, NamedProducer, @@ -24,7 +24,6 @@ use std::{ sync::{Arc, Mutex}, time::Duration, }; -use temporalio_common::protos::temporal::api::enums::v1::WorkflowExecutionStatus; use tokio::time::Instant; /// The longest `floor_holder` a read takes. @@ -82,10 +81,32 @@ impl StreamsOptions { #[derive(Debug, Clone, PartialEq, Eq, Hash)] struct OwnerKey { namespace: String, - workflow_id: String, + activity: bool, + /// The Workflow id or the Activity id. + id: String, run_id: String, } +impl OwnerKey { + fn owner(&self) -> OwnerId<'_> { + if self.activity { + OwnerId::Activity(&self.id) + } else { + OwnerId::Workflow(&self.id) + } + } + + /// The owner as error texts name it. + fn label(&self) -> String { + let kind = if self.activity { + "Activity" + } else { + "Workflow" + }; + format!("{kind} {:?}", self.id) + } +} + #[derive(Debug, Clone, PartialEq, Eq, Hash)] struct ProducerKey { owner: OwnerKey, @@ -190,9 +211,10 @@ impl Streams { /// Appends one batch for one producer attempt. /// /// The first batch for an owner, and the first after [StreamsOptions::owner_recheck], asks - /// whether the owner's chain ended. All producers of the owner share that check. If it did, or Temporal no longer holds the owner, - /// or a new chain took the Workflow id, the chain is marked closed and the batch is refused - /// as closed. + /// whether the owner's chain ended. All producers of the owner share that check. If it did, or + /// Temporal no longer holds the owner, or a new chain took the owner id, the chain is marked + /// closed and the batch is refused as closed. A standalone Activity run is a chain of one + /// run, so a new run of the Activity id ends the earlier run's streams. pub async fn append(&self, request: AppendRequest) -> StreamResult { let stream = address(request.stream.as_ref())?; let (producer_id, attempt) = match &request.producer { @@ -578,14 +600,12 @@ impl Streams { async fn owner_ended(&self, stream: &StreamAddress, chain: &ChainId) -> StreamResult { let key = owner_key(stream); match self.describe(&key, &key.run_id).await { - // The chain came from this Workflow, so History has since dropped it, or a failover + // The chain came from this owner, so Temporal has since dropped it, or a failover // hides it for a while. A reader only ends its own read, and producers close the // chain. Err(OwnerError::NotFound(_)) => Ok(self.gone_for_good(chain)), Err(error) => Err(describe_failed(&error)), - Ok(described) if !key.run_id.is_empty() => { - Ok(described.status != WorkflowExecutionStatus::Running) - } + Ok(described) if !key.run_id.is_empty() => Ok(described.status != OwnerStatus::Running), Ok(latest) if !latest.chain_ended() && latest.first_run_id == chain.first_run_id => { Ok(false) } @@ -698,22 +718,26 @@ impl Streams { &self, request: DeleteOwnerRequest, ) -> StreamResult { - if request.owner_kind != StreamOwnerKind::Workflow as i32 { - return Err(StreamError::unsupported(format!( - "this release keeps streams of Workflows only, not owner kind {}", - request.owner_kind - ))); + let (id, other) = match StreamOwnerKind::try_from(request.owner_kind) { + Ok(StreamOwnerKind::Workflow) => (&request.workflow_id, &request.activity_id), + Ok(StreamOwnerKind::Activity) => (&request.activity_id, &request.workflow_id), + _ => return Err(unsupported_kind(request.owner_kind)), + }; + if request.namespace.is_empty() || id.is_empty() { + return Err(StreamError::invalid_argument( + "a delete needs the namespace and the owner's id of its kind", + )); } - if request.namespace.is_empty() || request.workflow_id.is_empty() { + if !other.is_empty() { return Err(StreamError::invalid_argument( - "a delete needs the namespace and the owner's Workflow id", + "a delete names the owner's id of its kind only", )); } self.store.delete_owner(request).await } /// The chain a producer writes on. A producer attempt stays on the chain it began on, and - /// is refused once a new chain took the Workflow id. + /// is refused once a new chain took the owner id. async fn producer_chain(&self, key: ProducerKey) -> StreamResult { let chain = self.owner_chain(&key.owner).await?; if let Some(began) = self.producers.get(&key) @@ -741,11 +765,11 @@ impl Streams { // The latest run names the chain and tells whether it ended, in one call. let latest = self.describe(owner, "").await; if let (Err(OwnerError::NotFound(_)), Some(checked)) = (&latest, &known) { - // The chain came from this Workflow, so History has since dropped it. + // The chain came from this owner, so Temporal has since dropped it. let gone = chain_of_key(owner, &checked.first_run_id); return Err(self.gone(&gone).await); } - let latest = latest.map_err(not_found)?; + let latest = latest.map_err(|error| not_found(owner, error))?; let chain = chain_of_key(owner, &first_run_id(owner, &latest)?); if latest.chain_ended() { return Err(self.closed(&chain, "has closed").await); @@ -760,10 +784,10 @@ impl Streams { let pinned = self .describe(owner, &owner.run_id) .await - .map_err(not_found)?; + .map_err(|error| not_found(owner, error))?; let chain = chain_of_key(owner, &first_run_id(owner, &pinned)?); // A running run is its chain's latest, so it answers the end check too. - if pinned.status != WorkflowExecutionStatus::Running { + if pinned.status != OwnerStatus::Running { self.refuse_if_ended(&chain).await?; } chain @@ -773,15 +797,11 @@ impl Streams { } /// Marks the chain closed and refuses when its latest run ended, when Temporal no longer - /// holds it, or when a new chain took its Workflow id. + /// holds it, or when a new chain took its owner id. async fn refuse_if_ended(&self, chain: &ChainId) -> StreamResult<()> { - let key = OwnerKey { - namespace: chain.namespace.clone(), - workflow_id: chain.workflow_id.clone(), - run_id: String::new(), - }; + let key = chain_owner_key(chain); match self.describe(&key, "").await { - // The chain came from this Workflow, so History has since dropped it. + // The chain came from this owner, so Temporal has since dropped it. Err(OwnerError::NotFound(_)) => Err(self.gone(chain).await), Err(error) => Err(describe_failed(&error)), Ok(latest) if latest.chain_ended() || latest.first_run_id != chain.first_run_id => { @@ -796,7 +816,7 @@ impl Streams { /// as a storage failure, which the caller retries. async fn gone(&self, chain: &ChainId) -> StreamError { if self.gone_for_good(chain) { - self.closed(chain, "is gone from History").await + self.closed(chain, gone_text(&chain_owner_key(chain))).await } else { missing(chain) } @@ -805,11 +825,7 @@ impl Streams { /// Whether Temporal answered that `chain`'s owner is gone twice, at least /// [StreamsOptions::owner_recheck] apart, with no answer that holds it in between. fn gone_for_good(&self, chain: &ChainId) -> bool { - let key = OwnerKey { - namespace: chain.namespace.clone(), - workflow_id: chain.workflow_id.clone(), - run_id: String::new(), - }; + let key = chain_owner_key(chain); let recheck = self.options.owner_recheck; match self.missing.get(&key) { Some(seen) if seen.first_run_id == chain.first_run_id => seen.at.elapsed() >= recheck, @@ -825,8 +841,8 @@ impl Streams { return error; } StreamError::closed(format!( - "the Workflow {:?} that owns this stream {how}", - chain.workflow_id + "the {} that owns this stream {how}", + chain_owner_key(chain).label() )) } @@ -838,7 +854,10 @@ impl Streams { if !key.run_id.is_empty() { return self.chain(stream).await; } - let described = self.describe(&key, "").await.map_err(not_found)?; + let described = self + .describe(&key, "") + .await + .map_err(|error| not_found(&key, error))?; let first_run_id = first_run_id(&key, &described)?; self.chains .put(key, &first_run_id, self.options.owner_recheck); @@ -849,13 +868,16 @@ impl Streams { pub(crate) async fn chain(&self, stream: &StreamAddress) -> StreamResult { let key = owner_key(stream); // A pinned run's chain never changes. The latest run's chain changes when a new chain - // takes the Workflow id. + // takes the owner id. if let Some(checked) = self.chains.get(&key) && (!key.run_id.is_empty() || checked.at.elapsed() < self.options.owner_recheck) { return Ok(chain_of(stream, &checked.first_run_id)); } - let described = self.describe(&key, &key.run_id).await.map_err(not_found)?; + let described = self + .describe(&key, &key.run_id) + .await + .map_err(|error| not_found(&key, error))?; let first_run_id = first_run_id(&key, &described)?; self.chains .put(key, &first_run_id, self.options.owner_recheck); @@ -869,25 +891,21 @@ impl Streams { ) -> Result { let described = self .owner - .describe(&owner.namespace, &owner.workflow_id, run_id) + .describe(&owner.namespace, owner.owner(), run_id) .await; if described.is_ok() { self.missing.forget(&OwnerKey { - namespace: owner.namespace.clone(), - workflow_id: owner.workflow_id.clone(), run_id: String::new(), + ..owner.clone() }); } described } fn hash(&self, stream: &StreamAddress) -> String { - stream_hash( - &stream.namespace, - WORKFLOW_OWNER_KIND, - &stream.workflow_id, - &stream.topic, - ) + let chain = chain_of(stream, ""); + let (kind, id) = chain_owner(&chain); + stream_hash(&stream.namespace, kind, id, &stream.topic) } fn cursor(&self, stream: &StreamAddress, position: &str) -> String { @@ -921,21 +939,44 @@ impl Streams { pub(crate) fn address(stream: Option<&StreamAddress>) -> StreamResult<&StreamAddress> { let stream = stream.ok_or_else(|| StreamError::invalid_argument("a stream call needs its stream"))?; - if stream.owner_kind != StreamOwnerKind::Workflow as i32 { - return Err(StreamError::unsupported(format!( - "this release keeps streams of Workflows only, not owner kind {}", - stream.owner_kind - ))); - } - if stream.namespace.is_empty() || stream.workflow_id.is_empty() { - return Err(StreamError::invalid_argument( - "a stream needs its namespace and its owner's Workflow id", - )); + match StreamOwnerKind::try_from(stream.owner_kind) { + Ok(StreamOwnerKind::Workflow) => { + if stream.namespace.is_empty() || stream.workflow_id.is_empty() { + return Err(StreamError::invalid_argument( + "a stream needs its namespace and its owner's Workflow id", + )); + } + if !stream.activity_id.is_empty() { + return Err(StreamError::invalid_argument( + "a stream that a Workflow owns names no Activity id", + )); + } + } + Ok(StreamOwnerKind::Activity) => { + if stream.namespace.is_empty() || stream.activity_id.is_empty() { + return Err(StreamError::invalid_argument( + "a stream needs its namespace and its owner's Activity id", + )); + } + if !stream.workflow_id.is_empty() { + return Err(StreamError::invalid_argument( + "a stream that an Activity owns names no Workflow id", + )); + } + } + _ => return Err(unsupported_kind(stream.owner_kind)), } check_topic(&stream.topic)?; Ok(stream) } +fn unsupported_kind(owner_kind: i32) -> StreamError { + StreamError::unsupported(format!( + "this release keeps streams of Workflows and standalone Activities only, not owner kind \ + {owner_kind}" + )) +} + /// Refuses a topic no store keys reliably, by the rules Core's Workflow path also applies. /// The producer id and attempt a named producer or an Activity attempt writes as. fn producer( @@ -1007,44 +1048,83 @@ fn millis(duration: Duration) -> u64 { } fn owner_key(stream: &StreamAddress) -> OwnerKey { + let activity = stream.owner_kind == StreamOwnerKind::Activity as i32; OwnerKey { namespace: stream.namespace.clone(), - workflow_id: stream.workflow_id.clone(), + activity, + id: if activity { + stream.activity_id.clone() + } else { + stream.workflow_id.clone() + }, run_id: stream.run_id.clone(), } } -fn chain_of(stream: &StreamAddress, first_run_id: &str) -> ChainId { +/// The latest run of the owner of `chain`. +fn chain_owner_key(chain: &ChainId) -> OwnerKey { + let activity = chain.owner_kind == StreamOwnerKind::Activity as i32; + OwnerKey { + namespace: chain.namespace.clone(), + activity, + id: chain_owner(chain).1.to_string(), + run_id: String::new(), + } +} + +/// The chain of `stream`'s owner that starts at `first_run_id`. +pub(crate) fn chain_of(stream: &StreamAddress, first_run_id: &str) -> ChainId { chain_of_key(&owner_key(stream), first_run_id) } fn chain_of_key(owner: &OwnerKey, first_run_id: &str) -> ChainId { - ChainId { - namespace: owner.namespace.clone(), - workflow_id: owner.workflow_id.clone(), - first_run_id: first_run_id.to_string(), - ..Default::default() + // A Workflow chain keeps the kind unset, as every chain from before the Activity kind does, + // so the two compare equal. + if owner.activity { + ChainId { + namespace: owner.namespace.clone(), + first_run_id: first_run_id.to_string(), + owner_kind: StreamOwnerKind::Activity as i32, + activity_id: owner.id.clone(), + ..Default::default() + } + } else { + ChainId { + namespace: owner.namespace.clone(), + workflow_id: owner.id.clone(), + first_run_id: first_run_id.to_string(), + ..Default::default() + } + } +} + +/// How a refusal says that Temporal no longer holds the owner. +fn gone_text(owner: &OwnerKey) -> &'static str { + if owner.activity { + "is gone from Temporal" + } else { + "is gone from History" } } fn first_run_id(owner: &OwnerKey, described: &OwnerDescription) -> StreamResult { if described.first_run_id.is_empty() { - // The Workflow's own output is keyed by the chain's first run, so an empty one would - // split the stream in two without an error. + // The owner's own output is keyed by the chain's first run, so an empty one would split + // the stream in two without an error. return Err(StreamError::unsupported(format!( - "Temporal did not report the first run id of Workflow {:?}, which streams are \ - keyed by", - owner.workflow_id + "Temporal did not report the first run id of {}, which streams are keyed by", + owner.label() ))); } Ok(described.first_run_id.clone()) } -fn not_found(error: OwnerError) -> StreamError { +fn not_found(owner: &OwnerKey, error: OwnerError) -> StreamError { match error { OwnerError::NotFound(message) => StreamError::not_found(format!( - "the Workflow that owns the stream was not found, and its stream is keyed by its \ - run chain: {message}" + "the {} that owns the stream was not found, and its stream is keyed by its run \ + chain: {message}", + owner.label() )), error => describe_failed(&error), } @@ -1052,9 +1132,9 @@ fn not_found(error: OwnerError) -> StreamError { fn missing(chain: &ChainId) -> StreamError { StreamError::storage(format!( - "Temporal doesn't hold the Workflow {:?} that owns this stream for now, so the call \ - fails until it holds it again or answers the same after a recheck", - chain.workflow_id + "Temporal doesn't hold the owner {:?} of this stream for now, so the call fails until \ + it holds it again or answers the same after a recheck", + chain_owner_key(chain).id )) } @@ -1078,10 +1158,12 @@ mod tests { use sha2::{Digest, Sha256}; use std::sync::atomic::{AtomicUsize, Ordering}; use temporalio_common::protos::temporal::api::{ - common::v1::Payload, enums::v1::WorkflowExecutionStatus, + common::v1::Payload, + enums::v1::{ActivityExecutionStatus, WorkflowExecutionStatus}, }; - /// Temporal as the tests need it: the latest run of each Workflow id, and pinned runs. + /// Temporal as the tests need it: the latest run of each owner, and pinned runs. A Workflow + /// owner is keyed by its id, and an Activity owner by `activity/`. #[derive(Default)] struct Owners { latest: Mutex>>, @@ -1094,7 +1176,7 @@ mod tests { let described = OwnerDescription { run_id: run_id.to_string(), first_run_id: first_run_id.to_string(), - status: WorkflowExecutionStatus::Running, + status: OwnerStatus::Running, }; self.runs .lock() @@ -1106,8 +1188,28 @@ mod tests { .insert(workflow_id.to_string(), Ok(described)); } + /// Starts run `run_id` of Activity `activity_id`, which is its own chain. + fn activity_run(&self, activity_id: &str, run_id: &str) { + self.run(&format!("activity/{activity_id}"), run_id, run_id); + } + fn end(&self, workflow_id: &str, status: WorkflowExecutionStatus) { - if let Some(Ok(latest)) = self.latest.lock().unwrap().get_mut(workflow_id) { + self.end_as(workflow_id, OwnerStatus::of_workflow(status)); + } + + fn end_activity(&self, activity_id: &str, status: ActivityExecutionStatus) { + let status = OwnerStatus::of_activity(status); + let key = format!("activity/{activity_id}"); + if let Some(Ok(latest)) = self.latest.lock().unwrap().get(&key) + && let Some(run) = self.runs.lock().unwrap().get_mut(&latest.run_id) + { + run.status = status; + } + self.end_as(&key, status); + } + + fn end_as(&self, key: &str, status: OwnerStatus) { + if let Some(Ok(latest)) = self.latest.lock().unwrap().get_mut(key) { latest.status = status; } } @@ -1129,10 +1231,14 @@ mod tests { async fn describe( &self, _namespace: &str, - workflow_id: &str, + owner: OwnerId<'_>, run_id: &str, ) -> Result { self.describes.fetch_add(1, Ordering::Relaxed); + let workflow_id = &match owner { + OwnerId::Workflow(id) => id.to_string(), + OwnerId::Activity(id) => format!("activity/{id}"), + }; if !run_id.is_empty() { return self .runs @@ -1619,16 +1725,36 @@ mod tests { OwnerDescription { run_id: String::new(), first_run_id: String::new(), - status, + status: OwnerStatus::of_workflow(status), } .chain_ended() }; for status in [Completed, Failed, Canceled, Terminated, TimedOut] { assert!(ended(status), "{status:?}"); } - for status in [Running, ContinuedAsNew, Unspecified] { + for status in [Running, ContinuedAsNew, Unspecified, Paused] { assert!(!ended(status), "{status:?}"); } + assert_eq!(OwnerStatus::of_workflow(Paused), OwnerStatus::Running); + } + + #[test] + fn only_a_terminal_activity_status_ends_its_run() { + use ActivityExecutionStatus::*; + for status in [Completed, Failed, Canceled, Terminated, TimedOut] { + assert_eq!( + OwnerStatus::of_activity(status), + OwnerStatus::Ended, + "{status:?}" + ); + } + for status in [Running, Paused, Unspecified] { + assert_eq!( + OwnerStatus::of_activity(status), + OwnerStatus::Running, + "{status:?}" + ); + } } #[tokio::test] @@ -2267,7 +2393,7 @@ mod tests { .unwrap() .get_mut("run-1") .unwrap() - .status = WorkflowExecutionStatus::ContinuedAsNew; + .status = OwnerStatus::ContinuedAsNew; setup.owners.run("wf", "run-2", "run-1"); let mut request = read_after("", &[], Duration::from_secs(5)); request.stream.as_mut().unwrap().run_id = "run-1".to_string(); @@ -2559,4 +2685,336 @@ mod tests { } assert_eq!(setup.owners.describes(), 0); } + + /// A stream of standalone Activity `act`, whose run `act-run-1` is running. + fn activity_stream(topic: &str) -> StreamAddress { + StreamAddress { + namespace: "ns".to_string(), + owner_kind: StreamOwnerKind::Activity as i32, + activity_id: "act".to_string(), + topic: topic.to_string(), + ..Default::default() + } + } + + fn activity_chain(run_id: &str) -> ChainId { + ChainId { + namespace: "ns".to_string(), + first_run_id: run_id.to_string(), + owner_kind: StreamOwnerKind::Activity as i32, + activity_id: "act".to_string(), + ..Default::default() + } + } + + fn activity_setup() -> Setup { + let setup = setup(quick()); + setup.owners.activity_run("act", "act-run-1"); + setup + } + + fn activity_append(sequence: i64, value: &str) -> AppendRequest { + AppendRequest { + stream: Some(activity_stream("out")), + producer: Some(Producer::Activity(ActivityProducer { + run_id: "act-run-1".to_string(), + activity_id: "act".to_string(), + attempt: 1, + ..Default::default() + })), + ..append("", sequence, value) + } + } + + fn activity_read(after: &str, state: &[u8], wait: Duration) -> ReadRequest { + ReadRequest { + stream: Some(activity_stream("out")), + ..read_after(after, state, wait) + } + } + + async fn read_activity_to_end( + streams: &Streams, + request: ReadRequest, + ) -> (usize, ReadResponse) { + let mut delivered = 0; + let mut request = request; + for _ in 0..50 { + let page = streams.read(request.clone()).await.unwrap(); + delivered += page.records.len(); + if page.done { + return (delivered, page); + } + request.after = page.cursor; + request.state = page.state; + } + panic!("the read never ended"); + } + + async fn activity_chain_closed(store: &MemoryStore, run_id: &str) -> bool { + store + .read(StoreReadRequest { + chain: Some(activity_chain(run_id)), + topic: "out".to_string(), + max_records: 1, + ..Default::default() + }) + .await + .unwrap() + .closed + } + + #[tokio::test] + async fn an_activity_writes_its_own_stream_on_the_chain_of_its_run() { + let setup = activity_setup(); + let landed = setup.streams.append(activity_append(1, "1")).await.unwrap(); + let held = setup + .store + .latest(StoreLatestRequest { + chain: Some(activity_chain("act-run-1")), + topic: "out".to_string(), + }) + .await + .unwrap(); + assert_eq!(held.position, "0"); + let read = setup + .streams + .read(activity_read("", &[], Duration::ZERO)) + .await + .unwrap(); + assert_eq!(read.records.len(), 1); + assert_eq!(read.records[0].cursor, landed.last_cursor); + assert_eq!(read.chain_first_run_id, "act-run-1"); + // A Workflow with the same id owns other streams, and their cursors don't mix. + setup.owners.run("act", "run-1", "run-1"); + let mut workflow = stream("out"); + workflow.workflow_id = "act".to_string(); + let workflow_read = setup + .streams + .read(ReadRequest { + stream: Some(workflow.clone()), + ..Default::default() + }) + .await + .unwrap(); + assert!(workflow_read.records.is_empty()); + let error = setup + .streams + .read(ReadRequest { + stream: Some(workflow), + after: landed.last_cursor, + ..Default::default() + }) + .await + .unwrap_err(); + assert_eq!(error.kind, StreamFailureKind::Cursor, "{error}"); + } + + #[tokio::test] + async fn an_activity_address_names_the_activity_only() { + let setup = activity_setup(); + let mut no_id = activity_stream("out"); + no_id.activity_id.clear(); + let mut with_workflow = activity_stream("out"); + with_workflow.workflow_id = "wf".to_string(); + let mut workflow_with_activity = stream("out"); + workflow_with_activity.activity_id = "act".to_string(); + for bad in [no_id, with_workflow, workflow_with_activity] { + assert_eq!( + kind( + setup + .streams + .latest(LatestRequest { stream: Some(bad) }) + .await + ), + StreamFailureKind::InvalidArgument + ); + } + assert_eq!(setup.owners.describes(), 0); + } + + #[tokio::test] + async fn an_activity_read_ends_done_when_the_run_reaches_a_terminal_status() { + let setup = activity_setup(); + setup.streams.append(activity_append(1, "1")).await.unwrap(); + setup + .owners + .end_activity("act", ActivityExecutionStatus::Failed); + let (delivered, end) = read_activity_to_end( + &setup.streams, + activity_read("", &[], Duration::from_secs(5)), + ) + .await; + assert_eq!(delivered, 1); + assert!(end.done); + assert!(activity_chain_closed(&setup.store, "act-run-1").await); + } + + #[tokio::test] + async fn an_activity_read_that_retries_and_pauses_stays_open() { + let setup = activity_setup(); + setup + .owners + .end_activity("act", ActivityExecutionStatus::Paused); + let idle = setup + .streams + .read(activity_read("", &[], Duration::from_millis(200))) + .await + .unwrap(); + assert!(!idle.done); + assert!(!activity_chain_closed(&setup.store, "act-run-1").await); + } + + #[tokio::test] + async fn an_activity_read_before_the_run_exists_waits_the_grace() { + let setup = before_owner_starts(Duration::from_secs(60)); + let empty = setup + .streams + .read(activity_read(END, &[], Duration::ZERO)) + .await + .unwrap(); + assert!(empty.records.is_empty() && !empty.done); + setup.owners.activity_run("act", "act-run-1"); + setup.streams.append(activity_append(1, "1")).await.unwrap(); + let read = setup + .streams + .read(activity_read(&empty.cursor, &empty.state, Duration::ZERO)) + .await + .unwrap(); + assert_eq!(read.records.len(), 1); + let late = before_owner_starts(Duration::ZERO); + let error = late + .streams + .read(activity_read("", &[], Duration::from_secs(5))) + .await + .unwrap_err(); + assert_eq!(error.kind, StreamFailureKind::NotFound); + assert!(error.message.contains("Activity \"act\""), "{error}"); + } + + #[tokio::test(start_paused = true)] + async fn an_activity_producer_is_refused_once_the_run_ended() { + let setup = activity_setup(); + setup.streams.append(activity_append(1, "1")).await.unwrap(); + setup + .owners + .end_activity("act", ActivityExecutionStatus::Completed); + tokio::time::advance(OWNER_RECHECK + Duration::from_secs(1)).await; + let error = setup + .streams + .append(activity_append(2, "2")) + .await + .unwrap_err(); + assert_eq!(error.kind, StreamFailureKind::Closed); + assert!(error.message.contains("Activity \"act\""), "{error}"); + assert!(activity_chain_closed(&setup.store, "act-run-1").await); + } + + #[tokio::test(start_paused = true)] + async fn a_new_run_of_the_activity_id_ends_the_earlier_runs_streams() { + let setup = activity_setup(); + setup.streams.append(activity_append(1, "1")).await.unwrap(); + setup.owners.activity_run("act", "act-run-2"); + tokio::time::advance(OWNER_RECHECK + Duration::from_secs(1)).await; + assert_eq!( + kind(setup.streams.append(activity_append(2, "2")).await), + StreamFailureKind::Closed + ); + assert!(activity_chain_closed(&setup.store, "act-run-1").await); + // A producer of the new run writes on the new run's chain. + let mut next = activity_append(1, "3"); + if let Some(Producer::Activity(activity)) = &mut next.producer { + activity.run_id = "act-run-2".to_string(); + } + setup.streams.append(next).await.unwrap(); + assert!(!activity_chain_closed(&setup.store, "act-run-2").await); + } + + #[tokio::test] + async fn an_activity_read_pinned_to_a_run_reads_that_run() { + let setup = activity_setup(); + setup.streams.append(activity_append(1, "1")).await.unwrap(); + setup.owners.activity_run("act", "act-run-2"); + let mut pinned = activity_read("", &[], Duration::ZERO); + pinned.stream.as_mut().unwrap().run_id = "act-run-1".to_string(); + let page = setup.streams.read(pinned).await.unwrap(); + assert_eq!(page.records.len(), 1); + assert_eq!(page.chain_first_run_id, "act-run-1"); + } + + #[tokio::test] + async fn an_activity_read_follows_the_newest_run_at_its_start_until_a_newer_run() { + let setup = activity_setup(); + setup.streams.append(activity_append(1, "1")).await.unwrap(); + let first = setup + .streams + .read(activity_read("", &[], Duration::ZERO)) + .await + .unwrap(); + assert_eq!(first.chain_first_run_id, "act-run-1"); + setup.owners.activity_run("act", "act-run-2"); + let (delivered, end) = read_activity_to_end( + &setup.streams, + activity_read(&first.cursor, &first.state, Duration::from_secs(5)), + ) + .await; + assert_eq!(delivered, 0); + assert!(end.done); + assert!(activity_chain_closed(&setup.store, "act-run-1").await); + } + + #[tokio::test] + async fn an_activity_owner_delete_removes_only_that_activitys_streams() { + let setup = activity_setup(); + setup.streams.append(activity_append(1, "1")).await.unwrap(); + setup.streams.append(append("p", 1, "1")).await.unwrap(); + let request = DeleteOwnerRequest { + namespace: "ns".to_string(), + owner_kind: StreamOwnerKind::Activity as i32, + activity_id: "act".to_string(), + ..Default::default() + }; + assert!( + setup + .streams + .delete_owner(request.clone()) + .await + .unwrap() + .deleted + > 0 + ); + let gone = setup + .store + .latest(StoreLatestRequest { + chain: Some(activity_chain("act-run-1")), + topic: "out".to_string(), + }) + .await + .unwrap(); + assert_eq!(gone.position, ""); + let kept = setup + .store + .latest(StoreLatestRequest { + chain: Some(chain("run-1")), + topic: "out".to_string(), + }) + .await + .unwrap(); + assert_eq!(kept.position, "0"); + for bad in [ + DeleteOwnerRequest { + activity_id: String::new(), + ..request.clone() + }, + DeleteOwnerRequest { + workflow_id: "wf".to_string(), + ..request.clone() + }, + ] { + assert_eq!( + kind(setup.streams.delete_owner(bad).await), + StreamFailureKind::InvalidArgument + ); + } + } } diff --git a/crates/streams/src/lib.rs b/crates/streams/src/lib.rs index f59204617f..e98add792b 100644 --- a/crates/streams/src/lib.rs +++ b/crates/streams/src/lib.rs @@ -2,11 +2,12 @@ //! Stream stores for Temporal Workflows. //! -//! A stream is a topic of a Workflow, kept in a store outside Temporal and keyed by the -//! Workflow's run chain. This crate holds what every store and every SDK share: the -//! [StreamStore] trait a store implements, the cursor format, the record helpers that stamp -//! records, and the error every call fails with. Lang reaches a store through Core's -//! `StreamService`, and Core's Worker appends a Workflow's own output through the same trait. +//! A stream is a topic of a Workflow or a standalone Activity, kept in a store outside Temporal +//! and keyed by the owner's run chain. An Activity run is a chain of one run. This crate holds +//! what every store and every SDK share: the [StreamStore] trait a store implements, the cursor +//! format, the record helpers that stamp records, and the error every call fails with. Lang +//! reaches a store through Core's `StreamService`, and Core's Worker appends a Workflow's own +//! output through the same trait. mod cursor; mod driver; @@ -37,13 +38,13 @@ pub use notify::{ WorkflowAttachment, progress_counter, }; pub use notifylog::{NotificationEntry, NotificationLog}; -pub use owner::{OwnerClient, OwnerDescription, OwnerError}; +pub use owner::{OwnerClient, OwnerDescription, OwnerError, OwnerId, OwnerStatus}; pub use reader::{ DEFAULT_MAX_RECORDS, MAX_STATE_BYTES, MAX_TRACKED_PRODUCERS, ReadTarget, read_page, }; pub use record::{ - WORKFLOW_OWNER_KIND, activity_producer_id, check_append_digest, stored_append_record, - workflow_batch_digest, workflow_producer_id, + ACTIVITY_OWNER_KIND, WORKFLOW_OWNER_KIND, activity_producer_id, chain_owner, + check_append_digest, stored_append_record, workflow_batch_digest, workflow_producer_id, }; pub use service::StreamService; pub use store::{ diff --git a/crates/streams/src/memory.rs b/crates/streams/src/memory.rs index f23cca4c53..b2fb9f056e 100644 --- a/crates/streams/src/memory.rs +++ b/crates/streams/src/memory.rs @@ -6,7 +6,8 @@ //! position keeps naming the same record after [MemoryStore::truncate] drops older ones. use crate::{ - NotificationEntry, NotificationLog, StreamError, StreamResult, StreamStore, + ACTIVITY_OWNER_KIND, NotificationEntry, NotificationLog, StreamError, StreamResult, + StreamStore, WORKFLOW_OWNER_KIND, chain_owner, proto::{ ChainId, DeleteOwnerRequest, DeleteOwnerResponse, StoreAppendRequest, StoreAppendResponse, StoreLatestRequest, StoreLatestResponse, StoreReadRequest, StoreReadResponse, StoredClose, @@ -22,12 +23,15 @@ use tokio::{sync::watch, time::Instant}; const NAME: &str = "memory"; -type ChainKey = (String, String, String); +/// A chain's namespace, owner kind, owner id and first run. +type ChainKey = (String, &'static str, String, String); fn chain_key(chain: &ChainId) -> ChainKey { + let (kind, id) = chain_owner(chain); ( chain.namespace.clone(), - chain.workflow_id.clone(), + kind, + id.to_string(), chain.first_run_id.clone(), ) } @@ -593,16 +597,21 @@ impl StreamStore for MemoryStore { } async fn delete_owner(&self, request: DeleteOwnerRequest) -> StreamResult { - if request.owner_kind != StreamOwnerKind::Workflow as i32 { - return Err(StreamError::unsupported(format!( - "this release keeps streams of Workflows only, not owner kind {}", - request.owner_kind - ))); - } + let (owner_kind, owner_id) = match StreamOwnerKind::try_from(request.owner_kind) { + Ok(StreamOwnerKind::Workflow) => (WORKFLOW_OWNER_KIND, &request.workflow_id), + Ok(StreamOwnerKind::Activity) => (ACTIVITY_OWNER_KIND, &request.activity_id), + _ => { + return Err(StreamError::unsupported(format!( + "this release keeps streams of Workflows and standalone Activities only, not \ + owner kind {}", + request.owner_kind + ))); + } + }; let mut chains = self.chains.lock().unwrap(); let mut deleted = 0; - chains.retain(|(namespace, workflow_id, _), chain| { - let owned = *namespace == request.namespace && *workflow_id == request.workflow_id; + chains.retain(|(namespace, kind, id, _), chain| { + let owned = *namespace == request.namespace && *kind == owner_kind && id == owner_id; if owned { deleted += 1 + chain.topics.len() as u64; } diff --git a/crates/streams/src/owner.rs b/crates/streams/src/owner.rs index 37bd22ec4d..14ff04eaf5 100644 --- a/crates/streams/src/owner.rs +++ b/crates/streams/src/owner.rs @@ -1,41 +1,79 @@ //! What the stream layer asks Temporal about a stream's owner. //! -//! A store can't see a Workflow close, so the layer above it asks: to find the chain a stream is +//! A store can't see an owner close, so the layer above it asks: to find the chain a stream is //! keyed by, to refuse a producer once the chain ended, and to end a read. Core implements this //! over its client, and tests use a fake. -use temporalio_common::protos::temporal::api::enums::v1::WorkflowExecutionStatus; +use temporalio_common::protos::temporal::api::enums::v1::{ + ActivityExecutionStatus, WorkflowExecutionStatus, +}; -/// One run of a Workflow, as Temporal describes it. +/// The owner of a stream, by kind and id. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum OwnerId<'a> { + /// A Workflow id. + Workflow(&'a str), + /// A standalone Activity id. + Activity(&'a str), +} + +/// How an owner's run stands, as the stream layer needs it. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum OwnerStatus { + /// The run has not reached a terminal status. + Running, + /// The run closed, and its successor in the chain writes on. + ContinuedAsNew, + /// The run closed and ended its chain. + Ended, +} + +impl OwnerStatus { + /// The status of a Workflow run. A paused run is running, since it can resume and write. + pub fn of_workflow(status: WorkflowExecutionStatus) -> Self { + use WorkflowExecutionStatus::*; + match status { + Completed | Failed | Canceled | Terminated | TimedOut => Self::Ended, + ContinuedAsNew => Self::ContinuedAsNew, + Unspecified | Running | Paused => Self::Running, + } + } + + /// The status of a standalone Activity run. Retries, backoff, pause and reset keep the run, so + /// only a terminal status ends it. + pub fn of_activity(status: ActivityExecutionStatus) -> Self { + use ActivityExecutionStatus::*; + match status { + Completed | Failed | Canceled | Terminated | TimedOut => Self::Ended, + Unspecified | Running | Paused => Self::Running, + } + } +} + +/// One run of an owner, as Temporal describes it. #[derive(Debug, Clone, PartialEq, Eq)] pub struct OwnerDescription { /// The run described. pub run_id: String, - /// The first run of the chain the run belongs to. Streams are keyed by it. + /// The first run of the chain the run belongs to. Streams are keyed by it. An Activity run is + /// its own chain. pub first_run_id: String, /// How the run stands. - pub status: WorkflowExecutionStatus, + pub status: OwnerStatus, } impl OwnerDescription { /// Whether the run ended its chain. A run that continued as new did not, since its successor /// writes on. pub fn chain_ended(&self) -> bool { - matches!( - self.status, - WorkflowExecutionStatus::Completed - | WorkflowExecutionStatus::Failed - | WorkflowExecutionStatus::Canceled - | WorkflowExecutionStatus::Terminated - | WorkflowExecutionStatus::TimedOut - ) + self.status == OwnerStatus::Ended } } /// Why a describe failed. #[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)] pub enum OwnerError { - /// Temporal holds no such Workflow or run, or no longer does. + /// Temporal holds no such owner or run, or no longer does. #[error("not found: {0}")] NotFound(String), /// Temporal could not answer. @@ -46,12 +84,11 @@ pub enum OwnerError { /// Temporal, as the stream layer asks it about owners. #[async_trait::async_trait] pub trait OwnerClient: Send + Sync { - /// Describes run `run_id` of `workflow_id`, or the Workflow id's latest run when `run_id` is - /// empty. + /// Describes run `run_id` of `owner`, or the owner's latest run when `run_id` is empty. async fn describe( &self, namespace: &str, - workflow_id: &str, + owner: OwnerId<'_>, run_id: &str, ) -> Result; } diff --git a/crates/streams/src/record.rs b/crates/streams/src/record.rs index 94d975ce90..6ca174e1b3 100644 --- a/crates/streams/src/record.rs +++ b/crates/streams/src/record.rs @@ -2,7 +2,9 @@ use crate::{ StreamError, StreamResult, - proto::{ActivityProducer, AppendRecord, StreamRecord, StreamRecordKind}, + proto::{ + ActivityProducer, AppendRecord, ChainId, StreamOwnerKind, StreamRecord, StreamRecordKind, + }, }; use temporalio_common::{ protos::temporal::api::common::v1::Payload, @@ -12,6 +14,18 @@ use temporalio_common::{ /// The owner kind a Workflow's streams hash under in their cursors. pub const WORKFLOW_OWNER_KIND: &str = "workflow"; +/// The owner kind a standalone Activity's streams hash under in their cursors. +pub const ACTIVITY_OWNER_KIND: &str = "activity"; + +/// The owner kind a chain's streams hash under, and the owner's id. +pub fn chain_owner(chain: &ChainId) -> (&'static str, &str) { + if chain.owner_kind == StreamOwnerKind::Activity as i32 { + (ACTIVITY_OWNER_KIND, &chain.activity_id) + } else { + (WORKFLOW_OWNER_KIND, &chain.workflow_id) + } +} + const HASH_LENGTH: usize = 32; /// The producer id an Activity attempt writes as. diff --git a/crates/streams/src/redis/keys.rs b/crates/streams/src/redis/keys.rs index 8634cd1bc9..a96568728a 100644 --- a/crates/streams/src/redis/keys.rs +++ b/crates/streams/src/redis/keys.rs @@ -11,6 +11,10 @@ //! :{::}:chain the close flag //! ``` //! +//! A standalone Activity run is a chain of one run. Its tag has the literal `!a` after the +//! namespace, `{:!a::}`, and the same keys follow it. Each part +//! escapes `!`, so no Workflow id makes that tag. +//! //! The notification log lives apart from the chains, one stream per Cluster slot and a few //! doorbells, each under the hash tag of a slot (see `slots.rs`). A chain's appends go to the //! stream of the chain's own slot, so the append script writes both: @@ -25,7 +29,11 @@ //! two streams share a key. use super::{conn::slot, slots}; -use crate::proto::ChainId; +use crate::proto::{ChainId, StreamOwnerKind}; + +/// The part after the namespace that marks an Activity chain's tag. [part] escapes `!`, so no +/// Workflow id makes it. +const ACTIVITY_MARK: &str = "!a"; /// Percent-encodes every byte but the unreserved ones, as Python's `quote(text, safe="")` does. pub(crate) fn part(text: &str) -> String { @@ -48,12 +56,16 @@ pub(crate) struct ChainKeys { impl ChainKeys { pub(crate) fn new(prefix: &str, chain: &ChainId) -> Self { + let owner = if chain.owner_kind == StreamOwnerKind::Activity as i32 { + format!("{ACTIVITY_MARK}:{}", part(&chain.activity_id)) + } else { + part(&chain.workflow_id) + }; Self { base: format!( - "{}:{{{}:{}:{}}}", + "{}:{{{}:{owner}:{}}}", part(prefix), part(&chain.namespace), - part(&chain.workflow_id), part(&chain.first_run_id) ), } @@ -99,20 +111,24 @@ pub(crate) fn doorbell(prefix: &str, doorbell: u32, doorbells: u32) -> String { pub(crate) fn chain_of_log(prefix: &str, key: &str) -> Option<(ChainId, String)> { let rest = key.strip_prefix(&format!("{}:{{", part(prefix)))?; let (tag, topic) = rest.split_once("}:t:")?; - let mut parts = tag.split(':'); - let (namespace, workflow_id, first_run_id) = (parts.next()?, parts.next()?, parts.next()?); - if parts.next().is_some() { - return None; - } - Some(( - ChainId { + let parts: Vec<&str> = tag.split(':').collect(); + let chain = match parts[..] { + [namespace, workflow_id, first_run_id] if workflow_id != ACTIVITY_MARK => ChainId { namespace: unpart(namespace)?, workflow_id: unpart(workflow_id)?, first_run_id: unpart(first_run_id)?, ..Default::default() }, - unpart(topic)?, - )) + [namespace, ACTIVITY_MARK, activity_id, run_id] => ChainId { + namespace: unpart(namespace)?, + first_run_id: unpart(run_id)?, + owner_kind: StreamOwnerKind::Activity as i32, + activity_id: unpart(activity_id)?, + ..Default::default() + }, + _ => return None, + }; + Some((chain, unpart(topic)?)) } /// Undoes [part]. @@ -148,6 +164,16 @@ pub(crate) fn owner_pattern(prefix: &str, namespace: &str, workflow_id: &str) -> ) } +/// The pattern that matches every key of one standalone Activity id, across all its runs. +pub(crate) fn activity_owner_pattern(prefix: &str, namespace: &str, activity_id: &str) -> String { + format!( + "{}:{{{}:{ACTIVITY_MARK}:{}:*", + part(prefix), + part(namespace), + part(activity_id) + ) +} + #[cfg(test)] mod tests { use super::*; @@ -157,22 +183,34 @@ mod tests { case[field].as_str().unwrap() } + fn case_chain(case: &Value) -> ChainId { + if case["owner_kind"] == "activity" { + ChainId { + namespace: text(case, "namespace").to_string(), + first_run_id: text(case, "first_run_id").to_string(), + owner_kind: StreamOwnerKind::Activity as i32, + activity_id: text(case, "activity_id").to_string(), + ..Default::default() + } + } else { + ChainId { + namespace: text(case, "namespace").to_string(), + workflow_id: text(case, "workflow_id").to_string(), + first_run_id: text(case, "first_run_id").to_string(), + ..Default::default() + } + } + } + #[test] fn keys_match_the_python_sdk() { // Streams written by an SDK before the store moved into Core must stay readable. let cases: Vec = serde_json::from_str(include_str!("../../testdata/redis/keys.json")).unwrap(); assert!(cases.len() >= 4); + assert!(cases.iter().any(|case| case["owner_kind"] == "activity")); for case in &cases { - let keys = ChainKeys::new( - text(case, "prefix"), - &ChainId { - namespace: text(case, "namespace").to_string(), - workflow_id: text(case, "workflow_id").to_string(), - first_run_id: text(case, "first_run_id").to_string(), - ..Default::default() - }, - ); + let keys = ChainKeys::new(text(case, "prefix"), &case_chain(case)); let topic = text(case, "topic"); assert_eq!(keys.log(topic), text(case, "log")); assert_eq!(keys.meta(topic), text(case, "meta")); @@ -185,12 +223,7 @@ mod tests { let cases: Vec = serde_json::from_str(include_str!("../../testdata/redis/keys.json")).unwrap(); for case in &cases { - let chain = ChainId { - namespace: text(case, "namespace").to_string(), - workflow_id: text(case, "workflow_id").to_string(), - first_run_id: text(case, "first_run_id").to_string(), - ..Default::default() - }; + let chain = case_chain(case); let prefix = text(case, "prefix"); let keys = ChainKeys::new(prefix, &chain); let topic = text(case, "topic"); @@ -204,6 +237,8 @@ mod tests { } assert_eq!(chain_of_log("p", "q:{ns:wf:run}:t:out"), None); assert_eq!(chain_of_log("p", "p:{ns:wf}:t:out"), None); + assert_eq!(chain_of_log("p", "p:{ns:!b:act:run}:t:out"), None); + assert_eq!(chain_of_log("p", "p:{ns:!a:act}:t:out"), None); } #[test] @@ -240,6 +275,44 @@ mod tests { assert_eq!(owner_pattern("p", "ns", "a*b"), "p:{ns:a%2Ab:*"); } + #[test] + fn an_activity_chain_keeps_apart_from_every_workflow() { + let activity = |id: &str| ChainId { + namespace: "ns".to_string(), + first_run_id: "run-1".to_string(), + owner_kind: StreamOwnerKind::Activity as i32, + activity_id: id.to_string(), + ..Default::default() + }; + let workflow = |id: &str| ChainId { + namespace: "ns".to_string(), + workflow_id: id.to_string(), + first_run_id: "run-1".to_string(), + ..Default::default() + }; + let matcher = + |pattern: String| move |key: &str| key.starts_with(pattern.trim_end_matches('*')); + for id in ["wf", "!a", "a", "!a:wf"] { + let keys = ChainKeys::new("p", &activity(id)); + for workflow_id in [id, "!a", "a", "%21a"] { + assert!( + !matcher(owner_pattern("p", "ns", workflow_id))(&keys.chain()), + "{id}" + ); + assert_ne!(keys, ChainKeys::new("p", &workflow(workflow_id)), "{id}"); + } + let own = matcher(activity_owner_pattern("p", "ns", id)); + assert!(own(&keys.chain()) && own(&keys.log("out")), "{id}"); + assert!(!own(&ChainKeys::new("p", &workflow(id)).chain()), "{id}"); + // The `:` after the id keeps `wf` from matching the runs of `wf2`. + assert!( + !own(&ChainKeys::new("p", &activity(&format!("{id}2"))).chain()), + "{id}" + ); + } + assert_eq!(activity_owner_pattern("p", "ns", "a*b"), "p:{ns:!a:a%2Ab:*"); + } + #[test] fn the_doorbells_spread_over_the_slot_range() { let slots: Vec = (0..16) diff --git a/crates/streams/src/redis/mod.rs b/crates/streams/src/redis/mod.rs index 16be29d880..40daf6db61 100644 --- a/crates/streams/src/redis/mod.rs +++ b/crates/streams/src/redis/mod.rs @@ -17,7 +17,7 @@ use crate::{ }; use conn::{BlockingPool, Shared}; use errors::{Call, Mapped}; -use keys::{ChainKeys, owner_pattern, session_field}; +use keys::{ChainKeys, activity_owner_pattern, owner_pattern, session_field}; use prost::Message; use redis::{Cmd, Script, Value, streams::StreamReadReply}; use std::{ @@ -807,15 +807,23 @@ impl StreamStore for RedisStore { } async fn delete_owner(&self, request: DeleteOwnerRequest) -> StreamResult { - if request.owner_kind != StreamOwnerKind::Workflow as i32 { - return Err(StreamError::unsupported(format!( - "this release keeps streams of Workflows only, not owner kind {}", - request.owner_kind - ))); - } - // A Workflow id's chains hash to different slots, so every primary is scanned. Some keys + let pattern = match StreamOwnerKind::try_from(request.owner_kind) { + Ok(StreamOwnerKind::Workflow) => { + owner_pattern(&self.prefix, &request.namespace, &request.workflow_id) + } + Ok(StreamOwnerKind::Activity) => { + activity_owner_pattern(&self.prefix, &request.namespace, &request.activity_id) + } + _ => { + return Err(StreamError::unsupported(format!( + "this release keeps streams of Workflows and standalone Activities only, not \ + owner kind {}", + request.owner_kind + ))); + } + }; + // An owner id's chains hash to different slots, so every primary is scanned. Some keys // may be gone when a later step fails, which is why a failure leaves the outcome unknown. - let pattern = owner_pattern(&self.prefix, &request.namespace, &request.workflow_id); let mut deleted = 0; for node in self.shared.primaries().await.mapped(Call::Write)? { let mut cursor = "0".to_string(); diff --git a/crates/streams/src/service.rs b/crates/streams/src/service.rs index 40a6210ef0..d93fbfcec3 100644 --- a/crates/streams/src/service.rs +++ b/crates/streams/src/service.rs @@ -326,16 +326,14 @@ fn duration(seconds: i64, nanos: i32, name: &str) -> StreamResult { mod tests { use super::*; use crate::{ - OwnerDescription, OwnerError, + OwnerDescription, OwnerError, OwnerId, OwnerStatus, proto::{ AppendRecord, AppendResponse, LatestResponse, MemoryStoreConfig, NamedProducer, ReadResponse, StreamAddress, StreamFailureKind, StreamOwnerKind, StreamRecordKind, append_request, read_record, }, }; - use temporalio_common::protos::temporal::api::{ - common::v1::Payload, enums::v1::WorkflowExecutionStatus, - }; + use temporalio_common::protos::temporal::api::common::v1::Payload; /// Every Workflow's latest run is `run-1`, the first of its chain, and still running. struct Running; @@ -345,13 +343,13 @@ mod tests { async fn describe( &self, _: &str, - _: &str, + _: OwnerId<'_>, _: &str, ) -> Result { Ok(OwnerDescription { run_id: "run-1".to_string(), first_run_id: "run-1".to_string(), - status: WorkflowExecutionStatus::Running, + status: OwnerStatus::Running, }) } } @@ -938,7 +936,7 @@ mod tests { async fn describe( &self, _: &str, - _: &str, + _: OwnerId<'_>, _: &str, ) -> Result { let ended = self.0.load(std::sync::atomic::Ordering::SeqCst); @@ -946,9 +944,9 @@ mod tests { run_id: "run-1".to_string(), first_run_id: "run-1".to_string(), status: if ended { - WorkflowExecutionStatus::Completed + OwnerStatus::Ended } else { - WorkflowExecutionStatus::Running + OwnerStatus::Running }, }) } diff --git a/crates/streams/src/subscribe.rs b/crates/streams/src/subscribe.rs index da2745c28f..db49d50359 100644 --- a/crates/streams/src/subscribe.rs +++ b/crates/streams/src/subscribe.rs @@ -249,13 +249,12 @@ impl Subscriptions { mod tests { use super::*; use crate::{ - MemoryStore, NotifierPoll, OwnerClient, OwnerDescription, OwnerError, StreamNotification, - StreamService, StreamsOptions, + MemoryStore, NotifierPoll, OwnerClient, OwnerDescription, OwnerError, OwnerId, OwnerStatus, + StreamNotification, StreamService, StreamsOptions, proto::{ChainId, StreamFailureKind, StreamOwnerKind}, }; use prost::Message; use std::sync::Mutex; - use temporalio_common::protos::temporal::api::enums::v1::WorkflowExecutionStatus; /// The owner's chain is `first_run_id`, which a test moves to a new chain. struct Owner(Mutex); @@ -265,14 +264,14 @@ mod tests { async fn describe( &self, _: &str, - _: &str, + _: OwnerId<'_>, _: &str, ) -> Result { let first_run_id = self.0.lock().unwrap().clone(); Ok(OwnerDescription { run_id: first_run_id.clone(), first_run_id, - status: WorkflowExecutionStatus::Running, + status: OwnerStatus::Running, }) } } diff --git a/crates/streams/testdata/redis/keys.json b/crates/streams/testdata/redis/keys.json index e6deb56472..23f823bbd4 100644 --- a/crates/streams/testdata/redis/keys.json +++ b/crates/streams/testdata/redis/keys.json @@ -48,5 +48,49 @@ "log": "pre%20fix:{n%25s:a~b.c_d-e:r}:t:x%20y", "meta": "pre%20fix:{n%25s:a~b.c_d-e:r}:t:x%20y:meta", "chain": "pre%20fix:{n%25s:a~b.c_d-e:r}:chain" + }, + { + "prefix": "temporal-streams", + "namespace": "default", + "owner_kind": "activity", + "activity_id": "gen-1", + "first_run_id": "run-1", + "topic": "tokens", + "log": "temporal-streams:{default:!a:gen-1:run-1}:t:tokens", + "meta": "temporal-streams:{default:!a:gen-1:run-1}:t:tokens:meta", + "chain": "temporal-streams:{default:!a:gen-1:run-1}:chain" + }, + { + "prefix": "p{x}", + "namespace": "ns:a", + "owner_kind": "activity", + "activity_id": "act:1{b}!", + "first_run_id": "r:1", + "topic": "t/1", + "log": "p%7Bx%7D:{ns%3Aa:!a:act%3A1%7Bb%7D%21:r%3A1}:t:t%2F1", + "meta": "p%7Bx%7D:{ns%3Aa:!a:act%3A1%7Bb%7D%21:r%3A1}:t:t%2F1:meta", + "chain": "p%7Bx%7D:{ns%3Aa:!a:act%3A1%7Bb%7D%21:r%3A1}:chain" + }, + { + "prefix": "temporal-streams", + "namespace": "ns", + "owner_kind": "activity", + "activity_id": "!a", + "first_run_id": "run", + "topic": "out", + "log": "temporal-streams:{ns:!a:%21a:run}:t:out", + "meta": "temporal-streams:{ns:!a:%21a:run}:t:out:meta", + "chain": "temporal-streams:{ns:!a:%21a:run}:chain" + }, + { + "prefix": "temporal-streams", + "namespace": "ns", + "owner_kind": "activity", + "activity_id": "wörk é", + "first_run_id": "01a1-xyz", + "topic": "tópic", + "log": "temporal-streams:{ns:!a:w%C3%B6rk%20%C3%A9:01a1-xyz}:t:t%C3%B3pic", + "meta": "temporal-streams:{ns:!a:w%C3%B6rk%20%C3%A9:01a1-xyz}:t:t%C3%B3pic:meta", + "chain": "temporal-streams:{ns:!a:w%C3%B6rk%20%C3%A9:01a1-xyz}:chain" } ] diff --git a/crates/streams/tests/conformance.rs b/crates/streams/tests/conformance.rs index 54618b4f75..fbcf47c355 100644 --- a/crates/streams/tests/conformance.rs +++ b/crates/streams/tests/conformance.rs @@ -17,7 +17,7 @@ use temporalio_common::protos::temporal::api::common::v1::Payload; use temporalio_streams::{ BEGINNING, END, MemoryStore, NotificationEntry, NotificationLog, Notifier, NotifierClient, NotifyError, NotifyingStore, ReadTarget, StreamNotification, StreamResult, StreamStore, - mint_cursor, proto::*, read_page, stored_append_record, stream_hash, + chain_owner, mint_cursor, proto::*, read_page, stored_append_record, stream_hash, }; #[async_trait::async_trait] @@ -126,6 +126,47 @@ impl Case for Notifying { } } +/// Another case whose chains are standalone Activity runs, so every case runs on the Activity +/// keys too. +struct Activity(C); + +async fn activity( + inner: impl std::future::Future>, +) -> Option> { + Some(Activity(inner.await?)) +} + +/// The chain of standalone Activity run `run_id` of `activity_id`. +fn activity_chain(namespace: &str, activity_id: &str, run_id: &str) -> ChainId { + ChainId { + namespace: namespace.to_string(), + first_run_id: run_id.to_string(), + owner_kind: StreamOwnerKind::Activity as i32, + activity_id: activity_id.to_string(), + ..Default::default() + } +} + +#[async_trait::async_trait] +impl Case for Activity { + fn store(&self) -> Arc { + self.0.store() + } + + fn chain(&self) -> ChainId { + let chain = self.0.chain(); + activity_chain(&chain.namespace, &chain.workflow_id, &chain.first_run_id) + } + + async fn drop_oldest(&self, chain: &ChainId, topic: &str, keep: usize) { + self.0.drop_oldest(chain, topic, keep).await; + } + + async fn expire_log(&self, chain: &ChainId, topic: &str) { + self.0.expire_log(chain, topic).await; + } +} + #[cfg(feature = "redis")] mod on_redis { use super::*; @@ -177,11 +218,15 @@ mod on_redis { } pub(crate) fn log_key(prefix: &str, chain: &ChainId, topic: &str) -> String { + let owner = if chain.owner_kind == StreamOwnerKind::Activity as i32 { + format!("!a:{}", part(&chain.activity_id)) + } else { + part(&chain.workflow_id) + }; format!( - "{}:{{{}:{}:{}}}:t:{}", + "{}:{{{}:{owner}:{}}}:t:{}", part(prefix), part(&chain.namespace), - part(&chain.workflow_id), part(&chain.first_run_id), part(topic) ) @@ -523,7 +568,10 @@ fn target(chain: &ChainId, topic: &str) -> ReadTarget { ReadTarget { chain: chain.clone(), topic: topic.to_string(), - stream_hash: stream_hash(&chain.namespace, "workflow", &chain.workflow_id, topic), + stream_hash: { + let (kind, id) = chain_owner(chain); + stream_hash(&chain.namespace, kind, id, topic) + }, } } @@ -1436,18 +1484,40 @@ mod cases { ..chain.clone() }; let bystander = case.chain(); + let (kind_name, id) = chain_owner(&chain); + let activity = kind_name == "activity"; + // An owner of the other kind with the same id keeps its streams. + let twin = if activity { + ChainId { + namespace: chain.namespace.clone(), + workflow_id: id.to_string(), + first_run_id: chain.first_run_id.clone(), + ..Default::default() + } + } else { + activity_chain(&chain.namespace, id, &chain.first_run_id) + }; let landed = Producer::new(case, &chain, OUT, "p", 1) .append(&["1"]) .await; Producer::new(case, &next, OUT, "p", 1).append(&["1"]).await; - Producer::new(case, &bystander, OUT, "p", 1) - .append(&["1"]) - .await; - let request = DeleteOwnerRequest { - namespace: chain.namespace.clone(), - owner_kind: StreamOwnerKind::Workflow as i32, - workflow_id: chain.workflow_id.clone(), - ..Default::default() + for kept in [&bystander, &twin] { + Producer::new(case, kept, OUT, "p", 1).append(&["1"]).await; + } + let request = if activity { + DeleteOwnerRequest { + namespace: chain.namespace.clone(), + owner_kind: StreamOwnerKind::Activity as i32, + activity_id: id.to_string(), + ..Default::default() + } + } else { + DeleteOwnerRequest { + namespace: chain.namespace.clone(), + owner_kind: StreamOwnerKind::Workflow as i32, + workflow_id: id.to_string(), + ..Default::default() + } }; assert!(store.delete_owner(request.clone()).await.unwrap().deleted > 0); assert_eq!(store.delete_owner(request).await.unwrap().deleted, 0); @@ -1458,10 +1528,12 @@ mod cases { kind(read(case, &chain, OUT, &after).await), StreamFailureKind::NotFound ); - assert_eq!( - values(&read(case, &bystander, OUT, BEGINNING).await.unwrap()), - ["1"] - ); + for kept in [&bystander, &twin] { + assert_eq!( + values(&read(case, kept, OUT, BEGINNING).await.unwrap()), + ["1"] + ); + } } /// The store's log. A case runs only on a store that keeps one. @@ -1664,6 +1736,12 @@ mod memory_with_log { log_cases!(super::memory_with_log()); } +mod activity_memory { + store_cases!(super::activity(super::memory())); + log_cases!(super::activity(super::memory_with_log())); + conformance!(super::activity(super::memory()); deleting_an_owner_drops_every_chain_of_it); +} + mod notifying_memory { store_cases!(super::notifying(super::memory())); log_cases!(super::notifying(super::memory_with_log())); @@ -1689,6 +1767,53 @@ mod cluster_redis { ); } +#[cfg(feature = "redis")] +mod activity_standalone_redis { + store_cases!(super::activity(super::on_redis::redis( + "STREAMS_REDIS_URL", + false + ))); + log_cases!(super::activity(super::on_redis::redis_with_log( + "STREAMS_REDIS_URL", + false + ))); + conformance!(super::activity(super::on_redis::redis("STREAMS_REDIS_URL", false)); + deleting_an_owner_drops_every_chain_of_it, + ); +} + +#[cfg(feature = "redis")] +mod activity_documented_acl_redis { + // The documented key rule covers an Activity's keys, since they start with the namespace. + store_cases!(super::activity(super::on_redis::redis_as_documented_user( + "STREAMS_REDIS_URL", + false, + &[] + ))); + conformance!(super::activity(super::on_redis::redis_as_documented_user( + "STREAMS_REDIS_URL", + false, + &["+scan"] + )); + deleting_an_owner_drops_every_chain_of_it, + ); +} + +#[cfg(feature = "redis")] +mod activity_cluster_redis { + store_cases!(super::activity(super::on_redis::redis( + "STREAMS_REDIS_CLUSTER_URL", + true + ))); + log_cases!(super::activity(super::on_redis::redis_with_log( + "STREAMS_REDIS_CLUSTER_URL", + true + ))); + conformance!(super::activity(super::on_redis::redis("STREAMS_REDIS_CLUSTER_URL", true)); + deleting_an_owner_drops_every_chain_of_it, + ); +} + #[cfg(feature = "redis")] mod standalone_redis_with_log { store_cases!(super::on_redis::redis_with_log("STREAMS_REDIS_URL", false)); diff --git a/crates/streams/tests/cross_impl.rs b/crates/streams/tests/cross_impl.rs index eef5e36e84..9f2900a4c8 100644 --- a/crates/streams/tests/cross_impl.rs +++ b/crates/streams/tests/cross_impl.rs @@ -18,12 +18,10 @@ use std::{ }, time::Duration, }; -use temporalio_common::protos::temporal::api::{ - common::v1::Payload, enums::v1::WorkflowExecutionStatus, -}; +use temporalio_common::protos::temporal::api::common::v1::Payload; use temporalio_streams::{ - OwnerClient, OwnerDescription, OwnerError, ReadTarget, RedisStore, RedisStoreOptions, - StreamStore, Streams, StreamsOptions, mint_cursor, + OwnerClient, OwnerDescription, OwnerError, OwnerId, OwnerStatus, ReadTarget, RedisStore, + RedisStoreOptions, StreamStore, Streams, StreamsOptions, mint_cursor, proto::{ AppendRecord, AppendRequest, ChainId, NamedProducer, ReadRequest, StoreAppendRequest, StreamAddress, StreamFailureKind, StreamOwnerKind, StreamRecordKind, @@ -272,13 +270,13 @@ impl OwnerClient for Running { async fn describe( &self, _namespace: &str, - _workflow_id: &str, + _owner: OwnerId<'_>, _run_id: &str, ) -> Result { Ok(OwnerDescription { run_id: FIRST_RUN.to_string(), first_run_id: FIRST_RUN.to_string(), - status: WorkflowExecutionStatus::Running, + status: OwnerStatus::Running, }) } } diff --git a/crates/streams/tests/redis_store.rs b/crates/streams/tests/redis_store.rs index d5785404ea..4aac4e51b3 100644 --- a/crates/streams/tests/redis_store.rs +++ b/crates/streams/tests/redis_store.rs @@ -761,13 +761,13 @@ impl temporalio_streams::OwnerClient for OneOwner { async fn describe( &self, _namespace: &str, - _workflow_id: &str, + _owner: temporalio_streams::OwnerId<'_>, _run_id: &str, ) -> Result { Ok(temporalio_streams::OwnerDescription { run_id: "run-1".to_string(), first_run_id: "run-1".to_string(), - status: *self.status.lock().unwrap(), + status: temporalio_streams::OwnerStatus::of_workflow(*self.status.lock().unwrap()), }) } }