From e7933918a3f89292e064ab13d0efa703ff456403 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Wed, 14 Oct 2026 23:30:22 -0700 Subject: [PATCH 1/5] Ordered Activity producer attempts by their start time. A reset sets the Temporal attempt back to 1, so the reset attempt's first batch was dropped or refused. The attempt start time grows across retries and resets, so it orders them. --- crates/sdk-core/CHANGELOG.md | 6 + crates/streams/src/handle.rs | 162 +++++++++++++++++- crates/streams/src/lib.rs | 5 +- crates/streams/src/record.rs | 36 ++++ .../testdata/streams/activity_attempts.json | 80 +++++++++ crates/streams/tests/vectors.rs | 33 ++++ 6 files changed, 318 insertions(+), 4 deletions(-) create mode 100644 crates/streams/testdata/streams/activity_attempts.json diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index e7f3d29ce3..6f31cb35be 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/crates/sdk-core/CHANGELOG.md @@ -162,6 +162,12 @@ relevant information. 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`. +* Experimental: an `ActivityProducer` with `attempt_started_time` writes as the attempt + `started_unix_ms << 16 | min(attempt, 65535)`. A retry and a reset attempt start later, so they + supersede the earlier attempt, also after a reset sets the Temporal attempt back to 1. An + attempt that writes on after a newer one started reads as stale. Without the field the attempt + is the Temporal attempt as before. `testdata/streams/activity_attempts.json` in + `temporalio-streams` pins the formula for other SDKs. ### Breaking Changes * Replay detects changes to the service or operation of scheduled Nexus operations, reporting diff --git a/crates/streams/src/handle.rs b/crates/streams/src/handle.rs index 146774a83d..69d392756a 100644 --- a/crates/streams/src/handle.rs +++ b/crates/streams/src/handle.rs @@ -5,7 +5,8 @@ use crate::{ BEGINNING, END, MAX_CLOSE_BYTES, ReadTarget, StreamError, StreamResult, StreamStore, - activity_producer_id, chain_owner, check_append_digest, cursor_position, mint_cursor, + activity_attempt, 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, @@ -946,7 +947,7 @@ fn producer( } (named.producer_id.clone(), named.attempt) } - (None, Some(activity)) => (activity_producer_id(activity), i64::from(activity.attempt)), + (None, Some(activity)) => (activity_producer_id(activity), activity_attempt(activity)?), (None, None) => return Err(StreamError::invalid_argument("a call needs its producer")), }; if attempt < 1 { @@ -2817,4 +2818,161 @@ mod tests { ); } } + + /// The start of 2026-10-14 in Unix milliseconds. + const DAY_START_MS: i64 = 1_791_936_000_000; + + /// Attempt `attempt` of Activity `fetch` of run `run-1` of `wf`, started `after_ms` past + /// [DAY_START_MS], writing `value` at `sequence` on the Workflow's stream. + fn timed(attempt: i32, after_ms: i64, sequence: i64, value: &str) -> AppendRequest { + let mut activity = ActivityProducer { + workflow_id: "wf".to_string(), + run_id: "run-1".to_string(), + activity_id: "fetch".to_string(), + attempt, + ..Default::default() + }; + let started = activity + .attempt_started_time + .get_or_insert_with(Default::default); + let ms = DAY_START_MS + after_ms; + started.seconds = ms / 1000; + started.nanos = (ms % 1000) as i32 * 1_000_000; + AppendRequest { + producer: Some(Producer::Activity(activity)), + ..append("", sequence, value) + } + } + + fn key(attempt: i64, after_ms: i64) -> i64 { + (DAY_START_MS + after_ms) << 16 | attempt + } + + /// The read's records, as each stored record's value and attempt, or a supersession's two + /// attempts with an empty value. + async fn shape( + setup: &Setup, + after: &str, + state: &[u8], + ) -> (Vec<(String, i64, i64, bool)>, ReadResponse) { + let page = setup + .streams + .read(read_after(after, state, Duration::ZERO)) + .await + .unwrap(); + let shape = page + .records + .iter() + .map(|record| match &record.record { + Some(read_record::Record::Stored(stored)) => ( + String::from_utf8(stored.body.clone().unwrap().data).unwrap(), + stored.attempt, + 0, + record.stale, + ), + Some(read_record::Record::Superseded(superseded)) => ( + String::new(), + superseded.previous_attempt, + superseded.attempt, + record.stale, + ), + None => panic!("an empty record"), + }) + .collect(); + (shape, page) + } + + #[tokio::test] + async fn a_timed_activity_attempt_writes_its_start_and_attempt_as_its_key() { + let setup = setup(quick()); + setup.streams.append(timed(3, 0, 1, "a")).await.unwrap(); + let (records, _) = shape(&setup, "", &[]).await; + assert_eq!(records, [("a".to_string(), key(3, 0), 0, false)]); + let mut late = timed(70_000, 0, 1, "b"); + if let Some(Producer::Activity(activity)) = &mut late.producer { + activity.activity_id = "other".to_string(); + } + setup.streams.append(late).await.unwrap(); + let (records, _) = shape(&setup, "", &[]).await; + assert_eq!(records[1].1, key(crate::MAX_KEYED_ACTIVITY_ATTEMPT, 0)); + let before_1970 = timed(1, -DAY_START_MS - 1000, 1, "c"); + assert_eq!( + kind(setup.streams.append(before_1970).await), + StreamFailureKind::InvalidArgument + ); + } + + #[tokio::test] + async fn an_activity_retry_supersedes_the_earlier_attempt() { + let setup = setup(quick()); + setup.streams.append(timed(1, 0, 1, "a1")).await.unwrap(); + setup + .streams + .append(timed(2, 1_000, 1, "b1")) + .await + .unwrap(); + let (records, _) = shape(&setup, "", &[]).await; + assert_eq!( + records, + [ + ("a1".to_string(), key(1, 0), 0, false), + (String::new(), key(1, 0), key(2, 1_000), false), + ("b1".to_string(), key(2, 1_000), 0, false), + ] + ); + } + + #[tokio::test] + async fn a_reset_activity_attempt_supersedes_with_no_divergence_and_no_drop() { + let setup = setup(quick()); + for (attempt, after_ms) in [(1, 0), (2, 1_000), (3, 2_000)] { + setup + .streams + .append(timed(attempt, after_ms, 1, &format!("attempt {attempt}"))) + .await + .unwrap(); + } + // A reset sets the Temporal attempt back to 1, and the reset attempt starts later. + setup + .streams + .append(timed(1, 3_000, 1, "reset")) + .await + .unwrap(); + let (records, _) = shape(&setup, "", &[]).await; + assert_eq!( + records.last().unwrap(), + &("reset".to_string(), key(1, 3_000), 0, false) + ); + assert_eq!( + records[records.len() - 2], + (String::new(), key(3, 2_000), key(1, 3_000), false) + ); + // With no start time the reset attempt would repeat attempt 1 with other content. + let mut untimed = timed(1, 0, 1, "reset"); + if let Some(Producer::Activity(activity)) = &mut untimed.producer { + activity.attempt_started_time = None; + } + setup.streams.append(untimed.clone()).await.unwrap(); + untimed.digest = vec![1; 32]; + assert_eq!( + kind(setup.streams.append(untimed).await), + StreamFailureKind::ProducerDivergent + ); + } + + #[tokio::test] + async fn an_activity_attempt_that_writes_after_a_newer_one_started_is_stale() { + let setup = setup(quick()); + setup.streams.append(timed(1, 0, 1, "a1")).await.unwrap(); + setup + .streams + .append(timed(2, 1_000, 1, "b1")) + .await + .unwrap(); + let (_, page) = shape(&setup, "", &[]).await; + // The first attempt timed out but keeps writing. + setup.streams.append(timed(1, 0, 2, "a2")).await.unwrap(); + let (records, _) = shape(&setup, &page.cursor, &page.state).await; + assert_eq!(records, [("a2".to_string(), key(1, 0), 0, true)]); + } } diff --git a/crates/streams/src/lib.rs b/crates/streams/src/lib.rs index 525700109e..140e401a54 100644 --- a/crates/streams/src/lib.rs +++ b/crates/streams/src/lib.rs @@ -43,8 +43,9 @@ pub use reader::{ DEFAULT_MAX_RECORDS, MAX_STATE_BYTES, MAX_TRACKED_PRODUCERS, ReadTarget, read_page, }; pub use record::{ - ACTIVITY_OWNER_KIND, WORKFLOW_OWNER_KIND, activity_producer_id, chain_owner, - check_append_digest, stored_append_record, workflow_batch_digest, workflow_producer_id, + ACTIVITY_OWNER_KIND, MAX_KEYED_ACTIVITY_ATTEMPT, WORKFLOW_OWNER_KIND, activity_attempt, + activity_producer_id, chain_owner, check_append_digest, stored_append_record, + workflow_batch_digest, workflow_producer_id, }; pub use service::StreamService; pub use store::{CloseOutcome, MAX_CLOSE_BYTES, StreamStore, empty_close_result}; diff --git a/crates/streams/src/record.rs b/crates/streams/src/record.rs index 6ca174e1b3..a271f3eb30 100644 --- a/crates/streams/src/record.rs +++ b/crates/streams/src/record.rs @@ -36,6 +36,42 @@ pub fn activity_producer_id(activity: &ActivityProducer) -> String { format!("{}@{}", activity.activity_id, activity.run_id) } +/// The largest Temporal attempt an Activity attempt key holds. A higher attempt keeps this one. +pub const MAX_KEYED_ACTIVITY_ATTEMPT: i64 = 0xFFFF; + +/// The attempt an Activity attempt writes as. +/// +/// With `attempt_started_time`, it is `started_unix_ms << 16 | min(attempt, 65535)`. A retry and +/// a reset attempt start later, so they write a higher attempt and supersede the earlier one, even +/// though a reset sets the Temporal attempt back to 1. An attempt that writes on after a newer one +/// started reads as stale. With no start time, it is the Temporal attempt, as a lang that sends +/// none expects. Fails as an invalid argument for an attempt below one or a start time outside +/// what the key holds. +pub fn activity_attempt(activity: &ActivityProducer) -> StreamResult { + let attempt = i64::from(activity.attempt); + if attempt < 1 { + return Err(StreamError::invalid_argument(format!( + "a producer's attempt starts at one, not {attempt}" + ))); + } + let Some(started) = &activity.attempt_started_time else { + return Ok(attempt); + }; + let unix_ms = started + .seconds + .checked_mul(1000) + .and_then(|ms| ms.checked_add(i64::from(started.nanos.div_euclid(1_000_000)))) + .filter(|ms| (1..1 << 47).contains(ms)) + .ok_or_else(|| { + StreamError::invalid_argument(format!( + "an Activity attempt's start time must be after 1970 and before the year 6429, \ + not {}s {}ns", + started.seconds, started.nanos + )) + })?; + Ok(unix_ms << 16 | attempt.min(MAX_KEYED_ACTIVITY_ATTEMPT)) +} + /// The producer id a Workflow run writes its own records as. /// /// It is an Activity's id form with no Activity id. An Activity id is never empty, so no Activity diff --git a/crates/streams/testdata/streams/activity_attempts.json b/crates/streams/testdata/streams/activity_attempts.json new file mode 100644 index 0000000000..c16ae8c07d --- /dev/null +++ b/crates/streams/testdata/streams/activity_attempts.json @@ -0,0 +1,80 @@ +[ + { + "name": "first attempt", + "attempt": 1, + "started_time": { + "seconds": 1791936000, + "nanos": 0 + }, + "key": "117436317696000001" + }, + { + "name": "a retry starts later and keys higher", + "attempt": 2, + "started_time": { + "seconds": 1791936002, + "nanos": 500000000 + }, + "key": "117436317859840002" + }, + { + "name": "the start time keeps whole milliseconds only", + "attempt": 3, + "started_time": { + "seconds": 1791936005, + "nanos": 999999 + }, + "key": "117436318023680003" + }, + { + "name": "a reset attempt 1 starts later than attempt 3 and keys higher", + "attempt": 1, + "started_time": { + "seconds": 1791936009, + "nanos": 0 + }, + "key": "117436318285824001" + }, + { + "name": "an attempt past 65535 keys as 65535", + "attempt": 70000, + "started_time": { + "seconds": 1791936010, + "nanos": 1000000 + }, + "key": "117436318351491071" + }, + { + "name": "no start time writes the Temporal attempt", + "attempt": 4, + "started_time": null, + "key": "4" + }, + { + "name": "an attempt below one is refused", + "attempt": 0, + "started_time": { + "seconds": 1791936000, + "nanos": 0 + }, + "key": null + }, + { + "name": "a start time at the epoch is refused", + "attempt": 1, + "started_time": { + "seconds": 0, + "nanos": 0 + }, + "key": null + }, + { + "name": "a start time before the epoch is refused", + "attempt": 1, + "started_time": { + "seconds": -1, + "nanos": 0 + }, + "key": null + } +] diff --git a/crates/streams/tests/vectors.rs b/crates/streams/tests/vectors.rs index 9d3eb615db..7715bb42d8 100644 --- a/crates/streams/tests/vectors.rs +++ b/crates/streams/tests/vectors.rs @@ -29,3 +29,36 @@ fn cursors_match_the_python_sdk() { } } } + +#[test] +fn activity_attempt_keys_match_the_vectors() { + // Every SDK reports the attempt a record carries, so each one composes it the same way. + use temporalio_streams::{activity_attempt, proto::ActivityProducer}; + let cases: Vec = + serde_json::from_str(include_str!("../testdata/streams/activity_attempts.json")).unwrap(); + assert!(cases.iter().any(|case| case["key"].is_null())); + for case in &cases { + let mut activity = ActivityProducer { + activity_id: "act".to_string(), + run_id: "run".to_string(), + attempt: case["attempt"].as_i64().unwrap() as i32, + ..Default::default() + }; + if let Some(started) = case["started_time"].as_object() { + let time = activity + .attempt_started_time + .get_or_insert_with(Default::default); + time.seconds = started["seconds"].as_i64().unwrap(); + time.nanos = started["nanos"].as_i64().unwrap() as i32; + } + let name = case["name"].as_str().unwrap(); + match case["key"].as_str() { + Some(key) => assert_eq!( + activity_attempt(&activity).unwrap().to_string(), + key, + "{name}" + ), + None => assert!(activity_attempt(&activity).is_err(), "{name}"), + } + } +} From fa579ac767fefd1287ef7110477ed4fd2b9b9dab Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Wed, 14 Oct 2026 23:11:55 -0700 Subject: [PATCH 2/5] Vendored the Activity owner kind and added it to the stream protos. A standalone Activity owns streams in Jetty. The API names it by kind and Activity id, and the stream protos carry it with the attempt start time that orders Activity attempts. --- .../temporal/api/enums/v1/stream.proto | 6 +++++ .../temporal/api/stream/v1/message.proto | 12 ++++++---- .../temporal/sdk/core/streams/streams.proto | 24 +++++++++++++++++-- crates/sdk-core-c-bridge/src/tests/mod.rs | 2 ++ crates/sdk-core/CHANGELOG.md | 5 ++++ .../src/core_tests/external_streams.rs | 1 + crates/sdk-core/src/streams.rs | 9 +++++++ .../src/worker/workflow/stream_output.rs | 1 + crates/streams/src/driver.rs | 2 ++ crates/streams/src/handle.rs | 5 ++++ crates/streams/src/memory.rs | 2 ++ crates/streams/src/notify.rs | 2 ++ crates/streams/src/reader.rs | 1 + crates/streams/src/record.rs | 1 + crates/streams/src/redis/keys.rs | 4 ++++ crates/streams/src/redis/mod.rs | 1 + crates/streams/src/service.rs | 3 +++ crates/streams/src/subscribe.rs | 4 ++++ crates/streams/tests/conformance.rs | 3 +++ crates/streams/tests/cross_impl.rs | 2 ++ crates/streams/tests/driver.rs | 2 ++ crates/streams/tests/redis_store.rs | 8 +++++++ 22 files changed, 94 insertions(+), 6 deletions(-) diff --git a/crates/protos/protos/api_upstream/temporal/api/enums/v1/stream.proto b/crates/protos/protos/api_upstream/temporal/api/enums/v1/stream.proto index 4a079844ce..6c2dfa80c5 100644 --- a/crates/protos/protos/api_upstream/temporal/api/enums/v1/stream.proto +++ b/crates/protos/protos/api_upstream/temporal/api/enums/v1/stream.proto @@ -16,4 +16,10 @@ enum StreamOwnerKind { // failure. If the Workflow's run chain ends with no close and callbacks are attached, the // stream notifier closes the stream with a failure within `streamnotifier.ownerCheckInterval`. STREAM_OWNER_KIND_WORKFLOW = 1; + // A standalone Activity owns the stream. The Activity closes the stream itself, with a result + // or a failure. If the Activity run reaches a terminal status with no close and callbacks + // are attached, the stream notifier closes the stream within + // `streamnotifier.ownerCheckInterval`. A completed run gives no result. A failed, canceled, + // terminated or timed-out run gives the run's own failure. + STREAM_OWNER_KIND_ACTIVITY = 2; } diff --git a/crates/protos/protos/api_upstream/temporal/api/stream/v1/message.proto b/crates/protos/protos/api_upstream/temporal/api/stream/v1/message.proto index 2576cdb42f..755fa7c7bb 100644 --- a/crates/protos/protos/api_upstream/temporal/api/stream/v1/message.proto +++ b/crates/protos/protos/api_upstream/temporal/api/stream/v1/message.proto @@ -18,14 +18,18 @@ import "temporal/api/failure/v1/message.proto"; // Names one stream within a namespace: its owner and its topic. message StreamReference { temporal.api.enums.v1.StreamOwnerKind owner_kind = 1; - // The owning Workflow, for `STREAM_OWNER_KIND_WORKFLOW`. + // The owning Workflow, for `STREAM_OWNER_KIND_WORKFLOW`. Empty for other kinds. string workflow_id = 2; - // The id of the first run in the owning Workflow's run chain. The stream notifier is keyed by - // it, so one notifier serves the stream across the chain's runs, and a stream that reuses the - // Workflow id in a new chain gets a notifier of its own. An empty run id is refused. + // For `STREAM_OWNER_KIND_WORKFLOW`, the id of the first run in the owning Workflow's run + // chain. The stream notifier is keyed by it, so one notifier serves the stream across the + // chain's runs, and a stream that reuses the Workflow id in a new chain gets a notifier of its + // own. For `STREAM_OWNER_KIND_ACTIVITY`, the id of the owning Activity run. An empty run id is + // refused. string run_id = 3; // The stream's topic within its owner. string topic = 4; + // The owning standalone Activity, for `STREAM_OWNER_KIND_ACTIVITY`. Empty for other kinds. + string activity_id = 5; } // A callback attached to a stream notifier, as DescribeStreamNotifier reports it. diff --git a/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto b/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto index df86d7b891..2aacf8c76d 100644 --- a/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto +++ b/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto @@ -4,6 +4,7 @@ package coresdk.streams; option ruby_package = "Temporalio::Internal::Bridge::Api::Streams"; import "google/protobuf/duration.proto"; +import "google/protobuf/timestamp.proto"; import "temporal/api/common/v1/message.proto"; import "temporal/api/failure/v1/message.proto"; import "temporal/sdk/streams/v1/message.proto"; @@ -120,25 +121,37 @@ message MemoryStoreConfig {} enum StreamOwnerKind { STREAM_OWNER_KIND_UNSPECIFIED = 0; STREAM_OWNER_KIND_WORKFLOW = 1; + // A standalone Activity run. It is a chain of one run, so its streams end with the run. + STREAM_OWNER_KIND_ACTIVITY = 2; } // One stream, a topic of an owner in a namespace. message StreamAddress { string namespace = 1; StreamOwnerKind owner_kind = 2; + // The owning Workflow. Empty for an Activity owner. string workflow_id = 3; // Pins the stream to the run chain that holds this run. Empty follows the chain that holds the - // Workflow id when the call starts. + // Workflow id when the call starts. For an Activity owner, the Activity run, and empty names + // the newest run of the Activity id when the call starts. string run_id = 4; string topic = 5; + // The owning standalone Activity. Empty for a Workflow owner. + string activity_id = 6; } // A Workflow's run chain. Its streams are keyed by its first run, so they follow -// Continue-as-New, and a new chain on the same Workflow id gets streams of its own. +// Continue-as-New, and a new chain on the same Workflow id gets streams of its own. A standalone +// Activity run is a chain of one run, with the run as `first_run_id`. message ChainId { string namespace = 1; + // Empty for an Activity owner. string workflow_id = 2; string first_run_id = 3; + // `UNSPECIFIED` is a Workflow owner, so a chain from before the Activity kind stays valid. + StreamOwnerKind owner_kind = 4; + // Empty for a Workflow owner. + string activity_id = 5; } // A producer that names itself. @@ -156,6 +169,10 @@ message ActivityProducer { string run_id = 2; string activity_id = 3; int32 attempt = 4; + // When the attempt started, as the Activity task gives it. With it, Core writes the attempt as + // `started_unix_ms << 16 | min(attempt, 65535)`, so a later attempt and a reset attempt both + // supersede an earlier one. Unset writes `attempt` as it is. + google.protobuf.Timestamp attempt_started_time = 5; } // One record lang appends. @@ -418,7 +435,10 @@ message RunStreamDriverResponse {} message DeleteOwnerRequest { string namespace = 1; StreamOwnerKind owner_kind = 2; + // The Workflow whose streams go, for a Workflow owner. string workflow_id = 3; + // The standalone Activity whose streams go, across all its runs, for an Activity owner. + string activity_id = 4; } message DeleteOwnerResponse { diff --git a/crates/sdk-core-c-bridge/src/tests/mod.rs b/crates/sdk-core-c-bridge/src/tests/mod.rs index 95c3cabf12..f96ecf862d 100644 --- a/crates/sdk-core-c-bridge/src/tests/mod.rs +++ b/crates/sdk-core-c-bridge/src/tests/mod.rs @@ -537,6 +537,7 @@ fn test_stream_store_appends_and_reads_through_the_c_bridge() { workflow_id: "wf".to_string(), run_id: String::new(), topic: "out".to_string(), + ..Default::default() }; let appended = context @@ -632,6 +633,7 @@ fn test_a_cancelled_stream_read_says_it_only_stopped_reading() { workflow_id: "wf".to_string(), run_id: String::new(), topic: "out".to_string(), + ..Default::default() }), after: "$end".to_string(), wait: Some(std::time::Duration::from_secs(30).try_into().unwrap()), diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index a78ed897a8..b53a33df68 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/crates/sdk-core/CHANGELOG.md @@ -144,6 +144,11 @@ relevant information. retry would be refused the same way, also when the run closed the topic itself before its task reran. Any other store failure still fails the task. Local Activities that a completion with stream output schedules start only once that output is in the store. +* Experimental: the stream protos name a standalone Activity owner. `StreamOwnerKind` gains + `ACTIVITY`, `StreamAddress` and `DeleteOwnerRequest` gain `activity_id`, and `ChainId` gains + `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`. ### Breaking Changes * Replay detects changes to the service or operation of scheduled Nexus operations, reporting diff --git a/crates/sdk-core/src/core_tests/external_streams.rs b/crates/sdk-core/src/core_tests/external_streams.rs index d239cc3da4..added0df89 100644 --- a/crates/sdk-core/src/core_tests/external_streams.rs +++ b/crates/sdk-core/src/core_tests/external_streams.rs @@ -1153,6 +1153,7 @@ async fn a_topic_closed_before_the_publish_drops_its_records_and_the_task_goes_o namespace: crate::test_help::NAMESPACE.to_string(), workflow_id: "fakeid".to_string(), first_run_id: String::new(), + ..Default::default() }; let appended_at_completion: Arc>> = Default::default(); let history = timer_history(); diff --git a/crates/sdk-core/src/streams.rs b/crates/sdk-core/src/streams.rs index 809e031fae..9472b81590 100644 --- a/crates/sdk-core/src/streams.rs +++ b/crates/sdk-core/src/streams.rs @@ -146,6 +146,7 @@ fn chain_reference(chain: &ChainId, topic: String) -> StreamReference { workflow_id: chain.workflow_id.clone(), run_id: chain.first_run_id.clone(), topic, + ..Default::default() } } @@ -372,6 +373,7 @@ mod tests { namespace: "ns".to_string(), workflow_id: "wf".to_string(), first_run_id: "run-1".to_string(), + ..Default::default() }, topic: "tokens".to_string(), position: "cursor".to_string(), @@ -386,6 +388,7 @@ mod tests { workflow_id: "wf".to_string(), run_id: "run-1".to_string(), topic: "tokens".to_string(), + ..Default::default() }) ); assert_eq!((request.position.as_str(), request.counter), ("cursor", 7)); @@ -400,6 +403,7 @@ mod tests { namespace: namespace.to_string(), workflow_id: workflow_id.to_string(), first_run_id: "run-1".to_string(), + ..Default::default() }, topic: "tokens".to_string(), position: format!("cursor-{counter}"), @@ -444,6 +448,7 @@ mod tests { namespace: "ns".to_string(), workflow_id: "owner".to_string(), first_run_id: "chain-1".to_string(), + ..Default::default() }, topic: "tokens".to_string(), request_id: "jetty/reader-run/1".to_string(), @@ -459,6 +464,7 @@ mod tests { workflow_id: "owner".to_string(), run_id: "chain-1".to_string(), topic: "tokens".to_string(), + ..Default::default() }) ); assert_eq!(request.request_id, "jetty/reader-run/1"); @@ -682,6 +688,7 @@ mod tests { workflow_id: workflow_id.clone(), run_id: started.run_id.clone(), topic: "tokens".to_string(), + ..Default::default() }), request_id: "caller".to_string(), callback: Some( @@ -714,6 +721,7 @@ mod tests { workflow_id: workflow_id.clone(), run_id: String::new(), topic: "tokens".to_string(), + ..Default::default() }; Some(Live { connection, @@ -855,6 +863,7 @@ mod tests { workflow_id: live.workflow_id.clone(), run_id: live.run_id.clone(), topic: "tokens".to_string(), + ..Default::default() }), } .into_request(), diff --git a/crates/sdk-core/src/worker/workflow/stream_output.rs b/crates/sdk-core/src/worker/workflow/stream_output.rs index faa6afb1da..0d995f0850 100644 --- a/crates/sdk-core/src/worker/workflow/stream_output.rs +++ b/crates/sdk-core/src/worker/workflow/stream_output.rs @@ -137,6 +137,7 @@ impl OutputStore { namespace: self.namespace.clone(), workflow_id: workflow_id.to_string(), first_run_id: first_run_id.to_string(), + ..Default::default() } } diff --git a/crates/streams/src/driver.rs b/crates/streams/src/driver.rs index 4a8a364230..d396b69a5e 100644 --- a/crates/streams/src/driver.rs +++ b/crates/streams/src/driver.rs @@ -269,6 +269,7 @@ impl ChainKey { namespace: self.0, workflow_id: self.1, first_run_id: self.2, + ..Default::default() } } } @@ -326,6 +327,7 @@ mod tests { namespace: "ns".to_string(), workflow_id: workflow_id.to_string(), first_run_id: "run-1".to_string(), + ..Default::default() } } diff --git a/crates/streams/src/handle.rs b/crates/streams/src/handle.rs index 0a74385d63..9bd025d33b 100644 --- a/crates/streams/src/handle.rs +++ b/crates/streams/src/handle.rs @@ -944,6 +944,7 @@ fn chain_of_key(owner: &OwnerKey, first_run_id: &str) -> ChainId { namespace: owner.namespace.clone(), workflow_id: owner.workflow_id.clone(), first_run_id: first_run_id.to_string(), + ..Default::default() } } @@ -1088,6 +1089,7 @@ mod tests { workflow_id: "wf".to_string(), run_id: String::new(), topic: topic.to_string(), + ..Default::default() } } @@ -1120,6 +1122,7 @@ mod tests { namespace: "ns".to_string(), workflow_id: "wf".to_string(), first_run_id: first_run_id.to_string(), + ..Default::default() } } @@ -1183,6 +1186,7 @@ mod tests { run_id: "run-1".to_string(), activity_id: "fetch".to_string(), attempt: 2, + ..Default::default() })); setup.streams.append(request).await.unwrap(); let read = setup @@ -2261,6 +2265,7 @@ mod tests { namespace: "ns".to_string(), owner_kind: StreamOwnerKind::Workflow as i32, workflow_id: "wf".to_string(), + ..Default::default() }; assert!( setup diff --git a/crates/streams/src/memory.rs b/crates/streams/src/memory.rs index 9877027401..85d46b9867 100644 --- a/crates/streams/src/memory.rs +++ b/crates/streams/src/memory.rs @@ -620,6 +620,7 @@ mod tests { namespace: "ns".to_string(), workflow_id: "wf".to_string(), first_run_id: "run-1".to_string(), + ..Default::default() } } @@ -680,6 +681,7 @@ mod tests { namespace: "ns".to_string(), owner_kind: StreamOwnerKind::Unspecified as i32, workflow_id: "wf".to_string(), + ..Default::default() }) .await .unwrap_err(); diff --git a/crates/streams/src/notify.rs b/crates/streams/src/notify.rs index 4b253c9144..982d6b6fa4 100644 --- a/crates/streams/src/notify.rs +++ b/crates/streams/src/notify.rs @@ -217,6 +217,7 @@ impl StreamKey { namespace: self.namespace.clone(), workflow_id: self.workflow_id.clone(), first_run_id: self.first_run_id.clone(), + ..Default::default() } } } @@ -905,6 +906,7 @@ mod tests { namespace: "ns".to_string(), workflow_id: "wf".to_string(), first_run_id: first_run_id.to_string(), + ..Default::default() } } diff --git a/crates/streams/src/reader.rs b/crates/streams/src/reader.rs index 1649c2a182..7d0396a9b4 100644 --- a/crates/streams/src/reader.rs +++ b/crates/streams/src/reader.rs @@ -546,6 +546,7 @@ mod tests { namespace: "ns".to_string(), workflow_id: "wf".to_string(), first_run_id: "run-1".to_string(), + ..Default::default() }, topic: "out".to_string(), stream_hash: stream_hash("ns", "workflow", "wf", "out"), diff --git a/crates/streams/src/record.rs b/crates/streams/src/record.rs index 1d01d0b958..94d975ce90 100644 --- a/crates/streams/src/record.rs +++ b/crates/streams/src/record.rs @@ -188,6 +188,7 @@ mod tests { run_id: "run-1".to_string(), activity_id: "fetch".to_string(), attempt: 3, + ..Default::default() }; assert_eq!(activity_producer_id(&activity), "fetch@run-1"); } diff --git a/crates/streams/src/redis/keys.rs b/crates/streams/src/redis/keys.rs index 5928da8488..16328c9534 100644 --- a/crates/streams/src/redis/keys.rs +++ b/crates/streams/src/redis/keys.rs @@ -106,6 +106,7 @@ pub(crate) fn chain_of_log(prefix: &str, key: &str) -> Option<(ChainId, String)> namespace: unpart(namespace)?, workflow_id: unpart(workflow_id)?, first_run_id: unpart(first_run_id)?, + ..Default::default() }, unpart(topic)?, )) @@ -166,6 +167,7 @@ mod tests { 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 topic = text(case, "topic"); @@ -184,6 +186,7 @@ mod tests { 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 prefix = text(case, "prefix"); let keys = ChainKeys::new(prefix, &chain); @@ -218,6 +221,7 @@ mod tests { namespace: "ns".to_string(), workflow_id: "wf".to_string(), first_run_id: "run-1".to_string(), + ..Default::default() }; let pattern = owner_pattern("p", "ns", "wf"); assert_eq!(pattern, "p:{ns:wf:*"); diff --git a/crates/streams/src/redis/mod.rs b/crates/streams/src/redis/mod.rs index ca6b6ffcc6..c710af5017 100644 --- a/crates/streams/src/redis/mod.rs +++ b/crates/streams/src/redis/mod.rs @@ -873,6 +873,7 @@ mod tests { namespace: "ns".to_string(), workflow_id: workflow_id.to_string(), first_run_id: "run-1".to_string(), + ..Default::default() } } diff --git a/crates/streams/src/service.rs b/crates/streams/src/service.rs index 18aa0de077..40a6210ef0 100644 --- a/crates/streams/src/service.rs +++ b/crates/streams/src/service.rs @@ -375,6 +375,7 @@ mod tests { workflow_id: "wf".to_string(), run_id: String::new(), topic: "out".to_string(), + ..Default::default() } } @@ -578,6 +579,7 @@ mod tests { namespace: "ns".to_string(), owner_kind: StreamOwnerKind::Unspecified as i32, workflow_id: "wf".to_string(), + ..Default::default() }; let failure = service .call("DeleteOwner", &other_owner.encode_to_vec()) @@ -588,6 +590,7 @@ mod tests { namespace: "ns".to_string(), owner_kind: StreamOwnerKind::Workflow as i32, workflow_id: String::new(), + ..Default::default() }; let failure = service .call("DeleteOwner", &no_id.encode_to_vec()) diff --git a/crates/streams/src/subscribe.rs b/crates/streams/src/subscribe.rs index 8326d9b001..da2745c28f 100644 --- a/crates/streams/src/subscribe.rs +++ b/crates/streams/src/subscribe.rs @@ -183,6 +183,7 @@ impl Subscriptions { namespace: stream.namespace.clone(), workflow_id: stream.workflow_id.clone(), first_run_id: attached.first_run_id, + ..Default::default() }, None => streams.chain_now(stream).await?, }; @@ -363,6 +364,7 @@ mod tests { workflow_id: "owner".to_string(), run_id: String::new(), topic: "tokens".to_string(), + ..Default::default() } } @@ -391,6 +393,7 @@ mod tests { namespace: "ns".to_string(), workflow_id: "owner".to_string(), first_run_id: first_run_id.to_string(), + ..Default::default() } } @@ -641,6 +644,7 @@ mod tests { namespace: "ns".to_string(), workflow_id: "owner".to_string(), first_run_id, + ..Default::default() }), topic: "tokens".to_string(), producer_id: "p".to_string(), diff --git a/crates/streams/tests/conformance.rs b/crates/streams/tests/conformance.rs index 0bcd9789c3..54618b4f75 100644 --- a/crates/streams/tests/conformance.rs +++ b/crates/streams/tests/conformance.rs @@ -58,6 +58,7 @@ impl Case for Memory { namespace: "default".to_string(), workflow_id: unique("conformance"), first_run_id: unique("run"), + ..Default::default() } } @@ -383,6 +384,7 @@ mod on_redis { namespace: "default".to_string(), workflow_id: unique("conformance"), first_run_id: unique("run"), + ..Default::default() } } @@ -1445,6 +1447,7 @@ mod cases { namespace: chain.namespace.clone(), owner_kind: StreamOwnerKind::Workflow as i32, workflow_id: chain.workflow_id.clone(), + ..Default::default() }; assert!(store.delete_owner(request.clone()).await.unwrap().deleted > 0); assert_eq!(store.delete_owner(request).await.unwrap().deleted, 0); diff --git a/crates/streams/tests/cross_impl.rs b/crates/streams/tests/cross_impl.rs index 70585ff96f..eef5e36e84 100644 --- a/crates/streams/tests/cross_impl.rs +++ b/crates/streams/tests/cross_impl.rs @@ -104,6 +104,7 @@ impl Both { namespace: NAMESPACE.to_string(), workflow_id: self.workflow_id.clone(), first_run_id: FIRST_RUN.to_string(), + ..Default::default() } } @@ -303,6 +304,7 @@ impl Both { workflow_id: self.workflow_id.clone(), run_id: String::new(), topic: "events".to_string(), + ..Default::default() }), producer: Some(Producer::Named(NamedProducer { producer_id: "p".to_string(), diff --git a/crates/streams/tests/driver.rs b/crates/streams/tests/driver.rs index c6d06d070c..46ae80cc67 100644 --- a/crates/streams/tests/driver.rs +++ b/crates/streams/tests/driver.rs @@ -131,6 +131,7 @@ async fn write( namespace: "ns".to_string(), workflow_id: workflow_id.clone(), first_run_id: "run-1".to_string(), + ..Default::default() }), topic: "out".to_string(), producer_id: "p".to_string(), @@ -422,6 +423,7 @@ async fn traffic( namespace: "ns".to_string(), workflow_id: workflow_id.clone(), first_run_id: "run-1".to_string(), + ..Default::default() }), topic: "out".to_string(), producer_id: "p".to_string(), diff --git a/crates/streams/tests/redis_store.rs b/crates/streams/tests/redis_store.rs index 7694cbfa04..cd28ce58f7 100644 --- a/crates/streams/tests/redis_store.rs +++ b/crates/streams/tests/redis_store.rs @@ -87,6 +87,7 @@ fn chain() -> ChainId { namespace: "ns".to_string(), workflow_id: unique("wf"), first_run_id: "run-1".to_string(), + ..Default::default() } } @@ -319,6 +320,7 @@ async fn ids_with_separators_and_braces_keep_their_own_keys() { namespace: "ns".to_string(), workflow_id: format!("{}:{{b}}", unique("a")), first_run_id: "c".to_string(), + ..Default::default() }; let (head, tail) = one.workflow_id.split_once(':').unwrap(); let other = ChainId { @@ -806,6 +808,7 @@ async fn a_new_producer_marks_an_ended_chain_closed_until_it_expires() { workflow_id: chain.workflow_id.clone(), run_id: String::new(), topic: "events".to_string(), + ..Default::default() }), producer: Some(Producer::Named(NamedProducer { producer_id: producer.to_string(), @@ -1274,6 +1277,7 @@ fn delete_request(workflow_id: &str) -> temporalio_streams::proto::DeleteOwnerRe namespace: "ns".to_string(), owner_kind: temporalio_streams::proto::StreamOwnerKind::Workflow as i32, workflow_id: workflow_id.to_string(), + ..Default::default() } } @@ -1425,6 +1429,7 @@ async fn a_delete_reaches_every_primary_of_a_cluster() { namespace: "ns".to_string(), workflow_id: workflow_id.clone(), first_run_id: format!("run-{run}"), + ..Default::default() }; store .append(append(&chain, "p", 1, 1, b"\x01")) @@ -1731,6 +1736,7 @@ async fn a_replay_read_on_redis_repeats_the_read_and_survives_retention_under_a_ workflow_id: chain.workflow_id.clone(), run_id: String::new(), topic: "events".to_string(), + ..Default::default() }; let append = |attempt: i64, sequence: i64, value: &str| AppendRequest { stream: Some(stream.clone()), @@ -1817,6 +1823,7 @@ async fn a_replay_read_from_a_trimmed_start_on_redis_is_rec_01() { workflow_id: chain.workflow_id.clone(), run_id: String::new(), topic: "events".to_string(), + ..Default::default() }; let append = |sequence: i64| AppendRequest { stream: Some(stream.clone()), @@ -1897,6 +1904,7 @@ async fn a_chains_notification_entry_lands_in_the_chains_own_slot() { namespace: unique("ns:with{braces}"), workflow_id: unique("wf"), first_run_id: unique("run"), + ..Default::default() }; let landed = store .append(append(&chain, "p", 1, 1, b"\x01")) From f4f497b2d91fd8c890b47c90431d5a8c5bc9178e Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Wed, 14 Oct 2026 23:22:41 -0700 Subject: [PATCH 3/5] Kept and checked streams that a standalone Activity owns. One Activity run is a chain of one run, so every chain rule holds. Core checks the run with DescribeActivityExecution, and the Redis tag marks an Activity with !a, which no Workflow id makes. --- crates/sdk-core/CHANGELOG.md | 9 + crates/sdk-core/src/streams.rs | 131 ++++- crates/streams/docs/redis.md | 12 +- crates/streams/src/handle.rs | 619 +++++++++++++++++++++--- crates/streams/src/lib.rs | 17 +- crates/streams/src/memory.rs | 31 +- crates/streams/src/owner.rs | 71 ++- crates/streams/src/record.rs | 16 +- crates/streams/src/redis/keys.rs | 127 +++-- crates/streams/src/redis/mod.rs | 26 +- crates/streams/src/service.rs | 16 +- crates/streams/src/subscribe.rs | 9 +- crates/streams/testdata/redis/keys.json | 44 ++ crates/streams/tests/conformance.rs | 157 +++++- crates/streams/tests/cross_impl.rs | 12 +- crates/streams/tests/redis_store.rs | 4 +- 16 files changed, 1087 insertions(+), 214 deletions(-) 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 9472b81590..7329f7a3c0 100644 --- a/crates/sdk-core/src/streams.rs +++ b/crates/sdk-core/src/streams.rs @@ -8,14 +8,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; @@ -252,25 +253,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()) + } + } } } @@ -300,12 +322,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::*; @@ -335,9 +381,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 91bdf0c53c..f080324f76 100644 --- a/crates/streams/docs/redis.md +++ b/crates/streams/docs/redis.md @@ -67,9 +67,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 9bd025d33b..146774a83d 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, @@ -181,9 +202,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 { @@ -560,17 +582,15 @@ impl Streams { let key = owner_key(stream); match self.describe(&key, &key.run_id).await { Err(OwnerError::NotFound(_)) => { - // The chain came from this Workflow, so History has since dropped it, and - // nothing more will be written for it. + // The chain came from this owner, so Temporal has since dropped it, and nothing + // more will be written for it. if key.run_id.is_empty() { self.store.close_chain(chain).await?; } Ok(true) } 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) } @@ -687,22 +707,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) @@ -730,11 +754,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.closed(&gone, "is gone from History").await); + return Err(self.closed(&gone, gone_text(owner)).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); @@ -749,10 +773,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 @@ -762,16 +786,12 @@ 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. - Err(OwnerError::NotFound(_)) => Err(self.closed(chain, "is gone from History").await), + // The chain came from this owner, so Temporal has since dropped it. + Err(OwnerError::NotFound(_)) => Err(self.closed(chain, gone_text(&key)).await), Err(error) => Err(describe_failed(&error)), Ok(latest) if latest.chain_ended() || latest.first_run_id != chain.first_run_id => { Err(self.closed(chain, "has closed").await) @@ -785,8 +805,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() )) } @@ -798,7 +818,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); @@ -809,13 +832,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); @@ -828,17 +854,14 @@ impl Streams { run_id: &str, ) -> Result { self.owner - .describe(&owner.namespace, &owner.workflow_id, run_id) + .describe(&owner.namespace, owner.owner(), run_id) .await } 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 { @@ -872,21 +895,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( @@ -928,44 +974,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), } @@ -991,10 +1076,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>>, @@ -1007,7 +1094,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() @@ -1019,8 +1106,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; } } @@ -1042,10 +1149,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 @@ -1455,16 +1566,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] @@ -2103,7 +2234,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(); @@ -2354,4 +2485,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 ea6570e7c1..525700109e 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::{CloseOutcome, MAX_CLOSE_BYTES, StreamStore, empty_close_result}; diff --git a/crates/streams/src/memory.rs b/crates/streams/src/memory.rs index 85d46b9867..5423e8db3e 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(), ) } @@ -591,16 +595,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 16328c9534..5dcf8c0dd6 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: @@ -24,7 +28,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 { @@ -47,12 +55,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) ), } @@ -96,20 +108,24 @@ pub(crate) fn doorbell(prefix: &str, doorbell: 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]. @@ -145,6 +161,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::*; @@ -154,22 +180,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")); @@ -182,12 +220,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"); @@ -201,6 +234,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] @@ -236,4 +271,42 @@ mod tests { // Each part is encoded as the keys encode it, so a glob character matches only itself. 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:*"); + } } diff --git a/crates/streams/src/redis/mod.rs b/crates/streams/src/redis/mod.rs index c710af5017..17283af939 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, Value, streams::StreamReadReply}; use std::{collections::HashMap, sync::Arc, time::Duration}; @@ -757,15 +757,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 cd28ce58f7..7f2dcbe2be 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()), }) } } From 9fc263ae8de6a302591932b9edc6392c7b60e326 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Wed, 14 Oct 2026 23:26:44 -0700 Subject: [PATCH 4/5] 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() { From 453b3298cb9f0d643facc45140e9f1c08c511aea Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Thu, 15 Oct 2026 02:29:12 -0700 Subject: [PATCH 5/5] Fit the Activity attempt key in 53 bits. The 16-bit form gave keys near 1.2e17, which the Nexus contract's models and JavaScript numbers refuse. Eleven bits for the attempt keep the key exact as a JSON number until 2109. --- .../temporal/sdk/core/streams/streams.proto | 2 +- crates/sdk-core/CHANGELOG.md | 4 +++ crates/streams/src/handle.rs | 17 +++++++--- crates/streams/src/record.rs | 17 +++++++--- .../testdata/streams/activity_attempts.json | 32 +++++++++++++++---- 5 files changed, 54 insertions(+), 18 deletions(-) diff --git a/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto b/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto index 2aacf8c76d..78f1d94b26 100644 --- a/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto +++ b/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto @@ -170,7 +170,7 @@ message ActivityProducer { string activity_id = 3; int32 attempt = 4; // When the attempt started, as the Activity task gives it. With it, Core writes the attempt as - // `started_unix_ms << 16 | min(attempt, 65535)`, so a later attempt and a reset attempt both + // `started_unix_ms << 11 | min(attempt, 2047)`, so a later attempt and a reset attempt both // supersede an earlier one. Unset writes `attempt` as it is. google.protobuf.Timestamp attempt_started_time = 5; } diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index 6f31cb35be..6de48a00ec 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/crates/sdk-core/CHANGELOG.md @@ -168,6 +168,10 @@ relevant information. attempt that writes on after a newer one started reads as stale. Without the field the attempt is the Temporal attempt as before. `testdata/streams/activity_attempts.json` in `temporalio-streams` pins the formula for other SDKs. +* Experimental: the Activity attempt key is `started_unix_ms << 11 | min(attempt, 2047)`, so it + fits 53 bits and a JSON number holds it exactly. The Nexus contract's models and JavaScript + refused the earlier 16-bit form. A start time from May 2109 on is refused. The vector file + holds the new keys. ### Breaking Changes * Replay detects changes to the service or operation of scheduled Nexus operations, reporting diff --git a/crates/streams/src/handle.rs b/crates/streams/src/handle.rs index 69d392756a..5a39cad93a 100644 --- a/crates/streams/src/handle.rs +++ b/crates/streams/src/handle.rs @@ -2845,7 +2845,7 @@ mod tests { } fn key(attempt: i64, after_ms: i64) -> i64 { - (DAY_START_MS + after_ms) << 16 | attempt + (DAY_START_MS + after_ms) << 11 | attempt } /// The read's records, as each stored record's value and attempt, or a supersession's two @@ -2895,11 +2895,18 @@ mod tests { setup.streams.append(late).await.unwrap(); let (records, _) = shape(&setup, "", &[]).await; assert_eq!(records[1].1, key(crate::MAX_KEYED_ACTIVITY_ATTEMPT, 0)); + assert_eq!(crate::MAX_KEYED_ACTIVITY_ATTEMPT, 2047); + // A JSON number holds the key exactly, so the Nexus contract's models take it. + assert!(records[1].1 < 1 << 53); let before_1970 = timed(1, -DAY_START_MS - 1000, 1, "c"); - assert_eq!( - kind(setup.streams.append(before_1970).await), - StreamFailureKind::InvalidArgument - ); + // 2^42 ms is in May 2109, and a key past it would need more than 53 bits. + let after_2109 = timed(1, (1 << 42) - DAY_START_MS, 1, "d"); + for refused in [before_1970, after_2109] { + assert_eq!( + kind(setup.streams.append(refused).await), + StreamFailureKind::InvalidArgument + ); + } } #[tokio::test] diff --git a/crates/streams/src/record.rs b/crates/streams/src/record.rs index a271f3eb30..ac6c15242f 100644 --- a/crates/streams/src/record.rs +++ b/crates/streams/src/record.rs @@ -36,12 +36,19 @@ pub fn activity_producer_id(activity: &ActivityProducer) -> String { format!("{}@{}", activity.activity_id, activity.run_id) } +/// The low bits of an Activity attempt key that hold the Temporal attempt. +const ACTIVITY_ATTEMPT_BITS: u32 = 11; + +/// The high bits of an Activity attempt key that hold the start time in Unix milliseconds. The +/// whole key fits 53 bits, so a JSON number holds it exactly. +const ACTIVITY_START_BITS: u32 = 53 - ACTIVITY_ATTEMPT_BITS; + /// The largest Temporal attempt an Activity attempt key holds. A higher attempt keeps this one. -pub const MAX_KEYED_ACTIVITY_ATTEMPT: i64 = 0xFFFF; +pub const MAX_KEYED_ACTIVITY_ATTEMPT: i64 = (1 << ACTIVITY_ATTEMPT_BITS) - 1; /// The attempt an Activity attempt writes as. /// -/// With `attempt_started_time`, it is `started_unix_ms << 16 | min(attempt, 65535)`. A retry and +/// With `attempt_started_time`, it is `started_unix_ms << 11 | min(attempt, 2047)`. A retry and /// a reset attempt start later, so they write a higher attempt and supersede the earlier one, even /// though a reset sets the Temporal attempt back to 1. An attempt that writes on after a newer one /// started reads as stale. With no start time, it is the Temporal attempt, as a lang that sends @@ -61,15 +68,15 @@ pub fn activity_attempt(activity: &ActivityProducer) -> StreamResult { .seconds .checked_mul(1000) .and_then(|ms| ms.checked_add(i64::from(started.nanos.div_euclid(1_000_000)))) - .filter(|ms| (1..1 << 47).contains(ms)) + .filter(|ms| (1..1 << ACTIVITY_START_BITS).contains(ms)) .ok_or_else(|| { StreamError::invalid_argument(format!( - "an Activity attempt's start time must be after 1970 and before the year 6429, \ + "an Activity attempt's start time must be after 1970 and before the year 2109, \ not {}s {}ns", started.seconds, started.nanos )) })?; - Ok(unix_ms << 16 | attempt.min(MAX_KEYED_ACTIVITY_ATTEMPT)) + Ok(unix_ms << ACTIVITY_ATTEMPT_BITS | attempt.min(MAX_KEYED_ACTIVITY_ATTEMPT)) } /// The producer id a Workflow run writes its own records as. diff --git a/crates/streams/testdata/streams/activity_attempts.json b/crates/streams/testdata/streams/activity_attempts.json index c16ae8c07d..bb1b9a4dc6 100644 --- a/crates/streams/testdata/streams/activity_attempts.json +++ b/crates/streams/testdata/streams/activity_attempts.json @@ -6,7 +6,7 @@ "seconds": 1791936000, "nanos": 0 }, - "key": "117436317696000001" + "key": "3669884928000001" }, { "name": "a retry starts later and keys higher", @@ -15,7 +15,7 @@ "seconds": 1791936002, "nanos": 500000000 }, - "key": "117436317859840002" + "key": "3669884933120002" }, { "name": "the start time keeps whole milliseconds only", @@ -24,7 +24,7 @@ "seconds": 1791936005, "nanos": 999999 }, - "key": "117436318023680003" + "key": "3669884938240003" }, { "name": "a reset attempt 1 starts later than attempt 3 and keys higher", @@ -33,16 +33,16 @@ "seconds": 1791936009, "nanos": 0 }, - "key": "117436318285824001" + "key": "3669884946432001" }, { - "name": "an attempt past 65535 keys as 65535", - "attempt": 70000, + "name": "an attempt past 2047 keys as 2047", + "attempt": 3000, "started_time": { "seconds": 1791936010, "nanos": 1000000 }, - "key": "117436318351491071" + "key": "3669884948484095" }, { "name": "no start time writes the Temporal attempt", @@ -76,5 +76,23 @@ "nanos": 0 }, "key": null + }, + { + "name": "a start time at 2^42 ms, in May 2109, is refused", + "attempt": 1, + "started_time": { + "seconds": 4398046511, + "nanos": 104000000 + }, + "key": null + }, + { + "name": "the last millisecond before 2^42 ms keys below 2^53", + "attempt": 2047, + "started_time": { + "seconds": 4398046511, + "nanos": 103000000 + }, + "key": "9007199254740991" } ]