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 e7f3d29ce3..6de48a00ec 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/crates/sdk-core/CHANGELOG.md @@ -162,6 +162,16 @@ 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. +* 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 e38fa6a726..644df6cf32 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, @@ -999,7 +1000,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 { @@ -3017,4 +3018,168 @@ 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) << 11 | 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)); + 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"); + // 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] + 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 e98add792b..d589d611d4 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::{ diff --git a/crates/streams/src/record.rs b/crates/streams/src/record.rs index 6ca174e1b3..ac6c15242f 100644 --- a/crates/streams/src/record.rs +++ b/crates/streams/src/record.rs @@ -36,6 +36,49 @@ 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 = (1 << ACTIVITY_ATTEMPT_BITS) - 1; + +/// The attempt an Activity attempt writes as. +/// +/// 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 +/// 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 << 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 2109, \ + not {}s {}ns", + started.seconds, started.nanos + )) + })?; + Ok(unix_ms << ACTIVITY_ATTEMPT_BITS | 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..bb1b9a4dc6 --- /dev/null +++ b/crates/streams/testdata/streams/activity_attempts.json @@ -0,0 +1,98 @@ +[ + { + "name": "first attempt", + "attempt": 1, + "started_time": { + "seconds": 1791936000, + "nanos": 0 + }, + "key": "3669884928000001" + }, + { + "name": "a retry starts later and keys higher", + "attempt": 2, + "started_time": { + "seconds": 1791936002, + "nanos": 500000000 + }, + "key": "3669884933120002" + }, + { + "name": "the start time keeps whole milliseconds only", + "attempt": 3, + "started_time": { + "seconds": 1791936005, + "nanos": 999999 + }, + "key": "3669884938240003" + }, + { + "name": "a reset attempt 1 starts later than attempt 3 and keys higher", + "attempt": 1, + "started_time": { + "seconds": 1791936009, + "nanos": 0 + }, + "key": "3669884946432001" + }, + { + "name": "an attempt past 2047 keys as 2047", + "attempt": 3000, + "started_time": { + "seconds": 1791936010, + "nanos": 1000000 + }, + "key": "3669884948484095" + }, + { + "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 + }, + { + "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" + } +] 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}"), + } + } +}