From e7933918a3f89292e064ab13d0efa703ff456403 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Wed, 14 Oct 2026 23:30:22 -0700 Subject: [PATCH 1/2] 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 453b3298cb9f0d643facc45140e9f1c08c511aea Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Thu, 15 Oct 2026 02:29:12 -0700 Subject: [PATCH 2/2] 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" } ]