[Jetty] Ordered Activity producer attempts by their start time.

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;
 }

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

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)]);
+    }
 }

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::{

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<i64> {
+    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

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"
+  }
+]