fix(streams): one clock read per stream record on all four backends

.github/workflows/integration.yml

@@ -299,7 +299,8 @@ jobs:
       # this backend declares no vector search capability, so its tests drive the
       # storage layer directly against this job's PostgreSQL. The PutItem create
       # race tests hold a create open in an outside transaction, the
-      # write-transaction conflict tests force a deadlock, and the backup restore
+      # write-transaction conflict tests force a deadlock, the stream clock test
+      # holds a row lock open while a write waits on it, and the backup restore
       # cases need catalog rows the current binary would not write; all need
       # direct access. The online GSI build tests reorder queue rows by hand and
       # pause writers with a test hook. They build their own throwaway databases;
@@ -316,7 +317,7 @@ jobs:
           cargo test --release -p extenddb-storage-postgres
           --test vector_control_plane --test put_create_race --test twi_conflict
           --test stream_shard_retention --test stream_shard_ownership --test stream_records_page_order --test backup_restore
-          --test gsi_online_build --test gsi_stale_writer --test table_snapshot
+          --test gsi_online_build --test gsi_stale_writer --test table_snapshot --test stream_record_clock
 
       # The daemonized server logs to syslog; dump it so server-side failures
       # are diagnosable from the job log.

crates/storage-cassandra/src/stream_util.rs

@@ -58,8 +58,30 @@ impl HybridClock {
     /// Returns a 23-digit zero-padded string. Lexicographic ordering matches
     /// temporal ordering.
     pub fn generate(&mut self) -> String {
+        self.generate_with_created_at().0
+    }
+
+    /// Generate the next sequence number and the wall-clock reading taken by
+    /// this call, as `(sequence_number, created_at_ms)`.
+    ///
+    /// The sequence prefix is that reading or a later clamped value, so a
+    /// record's stored creation time never exceeds the timestamp prefix of
+    /// its own sequence number (`created_at_ms <= hlc_ms`). A reading at or
+    /// below the last prefix keeps the last prefix; when the counter
+    /// overflows, the prefix moves to the later of a fresh reading and the
+    /// last prefix plus one.
+    pub fn generate_with_created_at(&mut self) -> (String, i64) {
         let now_ms = chrono::Utc::now().timestamp_millis();
+        self.generate_from(now_ms, || {
+            std::thread::sleep(std::time::Duration::from_millis(1));
+            chrono::Utc::now().timestamp_millis()
+        })
+    }
 
+    /// One clock step for the reading `now_ms`. `reread` is called only when
+    /// the counter overflows and returns the reading taken after the wait;
+    /// tests pass both readings in.
+    fn generate_from(&mut self, now_ms: i64, reread: impl FnOnce() -> i64) -> (String, i64) {
         if now_ms > self.last_timestamp_ms {
             self.last_timestamp_ms = now_ms;
             self.logical_counter = 0;
@@ -80,22 +102,21 @@ impl HybridClock {
             // component stays close to real time; ordering is carried by the
             // floor below, not by the sleep.
             if self.logical_counter > 999_999 {
-                std::thread::sleep(std::time::Duration::from_millis(1));
-                // The re-read can still sit behind `last_timestamp_ms` when
-                // the clock was clamped forward by an earlier backwards step.
-                // The physical component must move past the millisecond that
-                // already issued counter 999_999, so it never goes below
-                // `last_timestamp_ms + 1`.
-                let reread_ms = chrono::Utc::now().timestamp_millis();
-                self.last_timestamp_ms = reread_ms.max(self.last_timestamp_ms + 1);
+                // The re-read can sit below the last prefix, after a backwards
+                // step during the wait or under a forward clamp, so the prefix
+                // moves at least one past the last one. It stays above every
+                // earlier reading, and the sequence number sorts after every
+                // earlier one.
+                self.last_timestamp_ms = reread().max(self.last_timestamp_ms + 1);
                 self.logical_counter = 0;
             }
         }
 
-        format!(
+        let seq = format!(
             "{:013}{:06}{:04}",
             self.last_timestamp_ms, self.logical_counter, self.node_id
-        )
+        );
+        (seq, now_ms)
     }
 }
 
@@ -161,13 +182,14 @@ pub fn stream_record_statement(
     hlc: &Arc<Mutex<HybridClock>>,
     retention_seconds: u32,
 ) -> Option<String> {
+    let (sequence_number, created_at_ms) = hlc
+        .lock()
+        .unwrap_or_else(std::sync::PoisonError::into_inner)
+        .generate_with_created_at();
     let identity = StreamRecordIdentity {
         event_id: uuid::Uuid::new_v4().to_string(),
-        sequence_number: hlc
-            .lock()
-            .unwrap_or_else(std::sync::PoisonError::into_inner)
-            .generate(),
-        created_at_ms: chrono::Utc::now().timestamp_millis(),
+        sequence_number,
+        created_at_ms,
     };
     stream_record_statement_with_identity(
         account_keyspace,
@@ -287,6 +309,146 @@ mod tests {
         assert!(b < c, "sequence numbers must be ordered: {b} < {c}");
     }
 
+    /// The stored creation time never exceeds the millisecond prefix of the
+    /// record's own sequence number, including when the clock is clamped to
+    /// an earlier read (same-millisecond bursts and backwards steps both take
+    /// the clamp branch). On the clamp branch the returned value is the
+    /// call's own wall-clock reading, not the clamped prefix.
+    #[test]
+    fn created_at_never_exceeds_sequence_prefix() {
+        let mut clock = HybridClock::new(7);
+        // Force the clamp branch: pretend the last read was far in the future.
+        clock.last_timestamp_ms = chrono::Utc::now().timestamp_millis() + 5_000;
+        for _ in 0..1_000 {
+            let before = chrono::Utc::now().timestamp_millis();
+            let (seq, created_at_ms) = clock.generate_with_created_at();
+            let after = chrono::Utc::now().timestamp_millis();
+            let hlc_ms: i64 = seq[..13].parse().unwrap();
+            assert!(
+                created_at_ms <= hlc_ms,
+                "created_at_ms {created_at_ms} exceeds sequence prefix {hlc_ms} in {seq}"
+            );
+            assert!(
+                (before..=after).contains(&created_at_ms),
+                "created_at_ms {created_at_ms} must be the reading between {before} and \
+                 {after}, below the clamped prefix {hlc_ms}"
+            );
+        }
+        // On the normal branch the prefix is the reading itself.
+        let mut fresh = HybridClock::new(7);
+        let before = chrono::Utc::now().timestamp_millis();
+        let (seq, created_at_ms) = fresh.generate_with_created_at();
+        let after = chrono::Utc::now().timestamp_millis();
+        let hlc_ms: i64 = seq[..13].parse().unwrap();
+        assert_eq!(created_at_ms, hlc_ms);
+        assert!(
+            (before..=after).contains(&created_at_ms),
+            "created_at_ms {created_at_ms} is the wall clock read between {before} and {after}"
+        );
+    }
+
+    /// The counter-overflow branch re-reads the clock for the sequence
+    /// prefix. The readings are injected, so the branch runs on every run.
+    /// The returned `created_at_ms` is still the call's own reading, at or
+    /// below the new prefix.
+    #[test]
+    fn created_at_holds_across_counter_overflow() {
+        let now_ms: i64 = 1_800_000_000_000;
+        let mut clock = HybridClock::new(42);
+        let (first, _) =
+            clock.generate_from(now_ms, || unreachable!("no overflow on a fresh clock"));
+        clock.logical_counter = 999_999;
+        let mut reread = false;
+        let (seq, created_at_ms) = clock.generate_from(now_ms, || {
+            reread = true;
+            now_ms + 1
+        });
+        assert!(reread, "the call takes the overflow branch");
+        let hlc_ms: i64 = seq[..13].parse().unwrap();
+        assert_eq!(
+            created_at_ms, now_ms,
+            "created_at_ms is the call's own reading"
+        );
+        assert_eq!(hlc_ms, now_ms + 1, "the prefix moves to the re-read");
+        assert!(
+            created_at_ms <= hlc_ms,
+            "overflow branch: created_at_ms {created_at_ms} exceeds prefix {hlc_ms} in {seq}"
+        );
+        assert_eq!(
+            clock.logical_counter, 0,
+            "the counter resets after overflow"
+        );
+        assert_eq!(
+            &seq[13..19],
+            "000000",
+            "the counter field resets in the sequence"
+        );
+        assert!(
+            seq > first,
+            "the overflow sequence follows the previous one: {first} < {seq}"
+        );
+    }
+
+    /// When the wall clock steps back during the overflow wait, the re-read
+    /// lands below the call's own reading. The prefix still stays at or
+    /// above that reading, and the sequence still follows the previous one.
+    #[test]
+    fn overflow_prefix_stays_above_the_reading_when_the_reread_steps_back() {
+        let now_ms: i64 = 1_800_000_000_000;
+        let mut clock = HybridClock::new(3);
+        clock.last_timestamp_ms = now_ms;
+        clock.logical_counter = 999_999;
+        let previous = format!("{now_ms:013}{:06}{:04}", 999_999, 3);
+        let (seq, created_at_ms) = clock.generate_from(now_ms, || now_ms - 10);
+        let hlc_ms: i64 = seq[..13].parse().unwrap();
+        assert!(
+            created_at_ms <= hlc_ms,
+            "created_at_ms {created_at_ms} exceeds sequence prefix {hlc_ms} in {seq}"
+        );
+        assert!(
+            seq > previous,
+            "the overflow sequence follows the previous one: {previous} < {seq}"
+        );
+    }
+
+    /// Overflow under a forward clamp: the last prefix sits 5 s ahead of the
+    /// call's reading and the re-read lands a further 5 s back. The prefix
+    /// moves to the last prefix plus one, not to the reading plus one, so the
+    /// sequence still follows the previous one and `created_at_ms` stays at
+    /// or below the prefix.
+    #[test]
+    fn overflow_under_a_forward_clamp_moves_the_prefix_past_the_last_one() {
+        let now_ms: i64 = 1_800_000_000_000;
+        let mut clock = HybridClock::new(5);
+        clock.last_timestamp_ms = now_ms + 5_000;
+        clock.logical_counter = 999_999;
+        let previous = format!("{:013}{:06}{:04}", now_ms + 5_000, 999_999, 5);
+        let mut reread_called = false;
+        let (seq, created_at_ms) = clock.generate_from(now_ms, || {
+            reread_called = true;
+            now_ms - 5_000
+        });
+        let hlc_ms: i64 = seq[..13].parse().unwrap();
+        assert!(reread_called, "the call takes the overflow branch");
+        assert_eq!(
+            created_at_ms, now_ms,
+            "created_at_ms is the call's own reading"
+        );
+        assert!(
+            created_at_ms <= hlc_ms,
+            "created_at_ms {created_at_ms} exceeds sequence prefix {hlc_ms} in {seq}"
+        );
+        assert!(
+            seq > previous,
+            "the overflow sequence follows the previous one: {previous} < {seq}"
+        );
+        assert_eq!(
+            hlc_ms,
+            now_ms + 5_001,
+            "the prefix moves to the last prefix plus one: {seq}"
+        );
+    }
+
     #[test]
     fn test_sequence_length() {
         let mut clock = HybridClock::new(1);

crates/storage-cassandra/src/ttl_worker.rs

@@ -871,19 +871,20 @@ async fn process_ttl_work_row(
             .as_ref()
             .and_then(|specification| {
                 if specification.stream_enabled {
-                    specification
-                        .stream_view_type
-                        .map(|view_type| TtlStreamPlan {
+                    specification.stream_view_type.map(|view_type| {
+                        let (sequence_number, created_at_ms) = storage
+                            .hlc
+                            .lock()
+                            .unwrap_or_else(std::sync::PoisonError::into_inner)
+                            .generate_with_created_at();
+                        TtlStreamPlan {
                             event_id: uuid::Uuid::new_v4().to_string(),
-                            sequence_number: storage
-                                .hlc
-                                .lock()
-                                .unwrap_or_else(std::sync::PoisonError::into_inner)
-                                .generate(),
-                            created_at_ms: chrono::Utc::now().timestamp_millis(),
+                            sequence_number,
+                            created_at_ms,
                             region: storage.region.clone(),
                             view_type,
-                        })
+                        }
+                    })
                 } else {
                     None
                 }

crates/storage-cassandra/tests/direct/streams.rs

@@ -653,3 +653,92 @@ async fn test_list_streams_includes_stream_enabled_table() {
 
     assert!(streams.iter().any(|s| s.stream_arn == expected_arn));
 }
+
+/// `created_at` and `ApproximateCreationDateTime` derive from one clock
+/// read, and the sequence prefix is that reading or a later clamped value.
+/// For every stored record `created_at_ms <= hlc_ms`, and
+/// `created_at_ms / 1000` equals the record's timestamp. Covers plain
+/// writes and a service-identity deletion through `delete_item`.
+#[tokio::test]
+async fn test_stream_record_created_at_matches_the_record_and_sequence() {
+    if crate::helpers::skip_without_cassandra() {
+        return;
+    }
+    let engine = setup_engine().await;
+    let table = TestTable::new(&engine, "StreamClock", false).await;
+    let view = StreamViewType::NewAndOldImages;
+
+    for v in ["one", "two"] {
+        let mut item = BTreeMap::new();
+        item.insert("id".to_string(), AttributeValue::S("pk-clock".to_string()));
+        item.insert("val".to_string(), AttributeValue::S(v.to_string()));
+        engine
+            .put_item(
+                &table.key_info,
+                item,
+                false,
+                None,
+                &Default::default(),
+                Some(&capture(view)),
+            )
+            .await
+            .unwrap();
+    }
+    let mut key = BTreeMap::new();
+    key.insert("id".to_string(), AttributeValue::S("pk-clock".to_string()));
+    engine
+        .delete_item(
+            &table.key_info,
+            &key,
+            false,
+            None,
+            &Default::default(),
+            Some(&StreamCapture {
+                view_type: view,
+                user_identity: Some(extenddb_core::types::UserIdentity {
+                    identity_type: "Service".to_string(),
+                    principal_id: "dynamodb.amazonaws.com".to_string(),
+                }),
+                region: Arc::from("us-east-1"),
+            }),
+        )
+        .await
+        .unwrap();
+
+    let shard_id = shard_for("pk-clock", &table.key_info.table_id);
+    let keyspace = format!("extenddb_ttl_test_account_{}", table.key_info.account_id);
+    let query = format!(
+        "SELECT sequence_number, created_at, record_data FROM {keyspace}.stream_records \
+         WHERE shard_id = ?"
+    );
+    let result = engine
+        .session_arc()
+        .query_with_values(&query, cdrs_tokio::query_values!(shard_id.as_str()))
+        .await
+        .unwrap();
+    let rows = result
+        .response_body()
+        .unwrap()
+        .into_rows()
+        .unwrap_or_default();
+    assert_eq!(rows.len(), 3, "two puts and one delete");
+    for row in rows {
+        let seq: String = row.get_r_by_name("sequence_number").unwrap();
+        let created_at_ms: i64 = row.get_r_by_name("created_at").unwrap();
+        let data: String = row.get_r_by_name("record_data").unwrap();
+        let record: serde_json::Value = serde_json::from_str(&data).unwrap();
+        let hlc_ms: i64 = seq[..13].parse().unwrap();
+        let approx = record["dynamodb"]["ApproximateCreationDateTime"]
+            .as_i64()
+            .unwrap();
+        assert!(
+            created_at_ms <= hlc_ms,
+            "created_at {created_at_ms} exceeds the sequence prefix {hlc_ms} in {seq}"
+        );
+        assert_eq!(
+            created_at_ms.div_euclid(1_000),
+            approx,
+            "created_at {created_at_ms} and ApproximateCreationDateTime {approx} disagree"
+        );
+    }
+}