From 61bd04e95052e4c9108d1fd29226aa1c9f795b36 Mon Sep 17 00:00:00 2001
From: Lee Hannigan <29015211+LeeroyHannigan@users.noreply.github.com>
Date: Thu, 29 Oct 2026 21:00:37 +0000
Subject: [PATCH 01/10] fix(streams): one clock read per stream record on all
 four backends

A stream record stored two creation times from two clock reads. The record
body's ApproximateCreationDateTime was SystemTime::now() read when the record
was built. The created_at column, which the 24-hour retention sweep filters
on, was the column default: NOW() on PostgreSQL, the transaction start time,
and strftime('now') at insert time on SQLite. MongoDB read DateTime::now() a
second time in the insert. Cassandra read Utc::now() once for the HLC
sequence number and again for created_at_ms.

On PostgreSQL a write that waits inside its transaction (a row lock, a long
TransactWriteItems) stored a created_at earlier than its own
ApproximateCreationDateTime by the length of the wait, so the retention sweep
deleted it before records with the same ApproximateCreationDateTime written
in a short transaction.

Each backend now takes one wall-clock read per record and derives both
values from it. SQLite and PostgreSQL bind created_at explicitly in the
transaction path and in the write_stream_record trait method; the column
defaults stay for any other writer. SQLite writes the millisecond text shape
the default produces (24 characters), so rows from before and after this
change sort together as text. MongoDB uses one DateTime for the record and
the document. Cassandra's HybridClock gains generate_with_created_at, which
returns the sequence number and the reading it was derived from, so
created_at_ms <= hlc_ms holds for every record (the clock clamps its physical
component upward on a backwards step, never below the reading). Records
written before the upgrade keep their default-clock created_at; the
difference is at most the transaction length and ages out after 24 hours.

Tests. crates/storage-postgres/tests/stream_record_clock.rs holds a row lock
from an outside transaction while a PutItem with stream capture waits 2.1 s,
then asserts created_at's second equals ApproximateCreationDateTime on every
row, including a service-identity delete. On main the waited record stores
created_at two seconds before its own timestamp (measured: 1792765145 against
1792765147); with this change both are 1792765147. The same read-back runs on
SQLite (in-memory engine, in data/tx_helpers.rs), on Cassandra
(tests/direct/streams.rs, which also asserts created_at_ms <= hlc_ms on every
stored row), and on MongoDB (an ignored live test in backup_live_tests.rs,
run with EXTENDDB_TEST_MONGODB_URI). HybridClock unit tests pin
created_at_ms <= hlc_ms across 1,000 clamped reads and across the
counter-overflow branch, and created_at_ms equal to the wall-clock read on
the normal branch.

Closes #459.
---
 crates/storage-cassandra/src/stream_util.rs   |  90 ++++-
 crates/storage-cassandra/src/ttl_worker.rs    |  21 +-
 .../storage-cassandra/tests/direct/streams.rs |  89 +++++
 .../storage-mongodb/src/backup_live_tests.rs  | 101 ++++++
 crates/storage-mongodb/src/data_engine.rs     |  13 +-
 .../storage-postgres/src/data/tx_helpers.rs   |  19 +-
 crates/storage-postgres/src/stream_engine.rs  |  11 +-
 .../tests/stream_record_clock.rs              | 333 ++++++++++++++++++
 crates/storage-sqlite/src/data/tx_helpers.rs  | 125 ++++++-
 crates/storage-sqlite/src/sqlite_util.rs      |  40 +++
 crates/storage-sqlite/src/stream.rs           |  13 +-
 11 files changed, 813 insertions(+), 42 deletions(-)
 create mode 100644 crates/storage-postgres/tests/stream_record_clock.rs

diff --git a/crates/storage-cassandra/src/stream_util.rs b/crates/storage-cassandra/src/stream_util.rs
index 1ad6d8fbc..49e00f4cb 100644
--- a/crates/storage-cassandra/src/stream_util.rs
+++ b/crates/storage-cassandra/src/stream_util.rs
@@ -58,6 +58,18 @@ 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 it was
+    /// derived from, as `(sequence_number, created_at_ms)`.
+    ///
+    /// Stream records store both values. Taking them from one read keeps
+    /// `created_at_ms <= hlc_ms` for every record: the clock clamps its
+    /// physical component upward on a backwards step but never below the
+    /// reading, so a record's stored creation time never exceeds the
+    /// timestamp prefix of its own sequence number.
+    pub fn generate_with_created_at(&mut self) -> (String, i64) {
         let now_ms = chrono::Utc::now().timestamp_millis();
 
         if now_ms > self.last_timestamp_ms {
@@ -92,10 +104,11 @@ impl HybridClock {
             }
         }
 
-        format!(
+        let seq = format!(
             "{:013}{:06}{:04}",
             self.last_timestamp_ms, self.logical_counter, self.node_id
-        )
+        );
+        (seq, now_ms)
     }
 }
 
@@ -161,13 +174,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 +301,68 @@ 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).
+    #[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 (seq, created_at_ms) = clock.generate_with_created_at();
+            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}"
+            );
+        }
+        // And on the normal branch the two agree exactly: created_at_ms is the
+        // wall-clock read itself, not the clamped value.
+        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 sleeps a millisecond and re-reads the
+    /// clock for the sequence prefix. The returned `created_at_ms` is still
+    /// the original read, so it stays at or below the prefix.
+    #[test]
+    fn created_at_holds_across_counter_overflow() {
+        let mut clock = HybridClock::new(42);
+        // Land on the same-millisecond path, then drive the counter to the
+        // limit so the next call takes the overflow branch.
+        let (first, _) = clock.generate_with_created_at();
+        clock.logical_counter = 999_999;
+        let (seq, created_at_ms) = clock.generate_with_created_at();
+        let hlc_ms: i64 = seq[..13].parse().unwrap();
+        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}"
+        );
+    }
+
     #[test]
     fn test_sequence_length() {
         let mut clock = HybridClock::new(1);
diff --git a/crates/storage-cassandra/src/ttl_worker.rs b/crates/storage-cassandra/src/ttl_worker.rs
index f6a170ddd..b85b9900b 100644
--- a/crates/storage-cassandra/src/ttl_worker.rs
+++ b/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
                 }
diff --git a/crates/storage-cassandra/tests/direct/streams.rs b/crates/storage-cassandra/tests/direct/streams.rs
index 16c86540e..b7be9820f 100644
--- a/crates/storage-cassandra/tests/direct/streams.rs
+++ b/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`, the sequence number's millisecond prefix, and the record's
+/// `ApproximateCreationDateTime` all derive from one clock read: for every
+/// stored record `created_at_ms <= hlc_ms` and `created_at_ms / 1000` equals
+/// the record's timestamp. Covers a plain write and a service-identity
+/// deletion (the TTL worker's shape).
+#[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"
+        );
+    }
+}
diff --git a/crates/storage-mongodb/src/backup_live_tests.rs b/crates/storage-mongodb/src/backup_live_tests.rs
index 75ed23761..fb42f09da 100644
--- a/crates/storage-mongodb/src/backup_live_tests.rs
+++ b/crates/storage-mongodb/src/backup_live_tests.rs
@@ -1773,3 +1773,104 @@ async fn live_update_table_on_a_creating_restore_target_is_refused() {
     assert!(desc.deletion_protection_enabled);
     fx.finish().await;
 }
+
+/// A stream record's `created_at` field and its own `ApproximateCreationDateTime`
+/// come from one clock read on the live write path, for a plain write and for
+/// a service-identity deletion (the TTL worker's shape).
+#[tokio::test]
+#[ignore = "needs EXTENDDB_TEST_MONGODB_URI"]
+async fn live_stream_record_created_at_matches_the_record_timestamp() {
+    use extenddb_core::expression::ExpressionMaps;
+    use extenddb_core::types::{StreamSpecification, StreamViewType, UserIdentity};
+    use extenddb_storage::{DataEngine, StreamCapture};
+
+    let fx = Fixture::new().await;
+    fx.engine
+        .create_table(
+            ACCOUNT,
+            CreateTableInput {
+                table_name: "clock".to_owned(),
+                key_schema: vec![key("pk", KeyType::Hash)],
+                attribute_definitions: vec![attr("pk", ScalarAttributeType::S)],
+                billing_mode: Some(BillingMode::PayPerRequest),
+                stream_specification: Some(StreamSpecification {
+                    stream_enabled: true,
+                    stream_view_type: Some(StreamViewType::NewAndOldImages),
+                }),
+                ..Default::default()
+            },
+        )
+        .await
+        .expect("create the table");
+    let key_info = fx
+        .engine
+        .table_key_info(ACCOUNT, "clock")
+        .await
+        .expect("key info");
+    let maps = ExpressionMaps::default();
+    let capture = StreamCapture {
+        view_type: StreamViewType::NewAndOldImages,
+        user_identity: None,
+        region: Arc::from("us-east-1"),
+    };
+    for v in ["one", "two"] {
+        let item: Item = [
+            ("pk".to_owned(), AttributeValue::S("k".to_owned())),
+            ("v".to_owned(), AttributeValue::S(v.to_owned())),
+        ]
+        .into_iter()
+        .collect();
+        fx.engine
+            .put_item(&key_info, item, false, None, &maps, Some(&capture))
+            .await
+            .expect("put");
+    }
+    let key: Item = [("pk".to_owned(), AttributeValue::S("k".to_owned()))]
+        .into_iter()
+        .collect();
+    fx.engine
+        .delete_item(
+            &key_info,
+            &key,
+            false,
+            None,
+            &maps,
+            Some(&StreamCapture {
+                view_type: StreamViewType::NewAndOldImages,
+                user_identity: Some(UserIdentity {
+                    identity_type: "Service".to_owned(),
+                    principal_id: "dynamodb.amazonaws.com".to_owned(),
+                }),
+                region: Arc::from("us-east-1"),
+            }),
+        )
+        .await
+        .expect("delete");
+
+    let mut cursor = fx
+        .engine
+        .data_db
+        .collection::<Document>("stream_records")
+        .find(doc! {})
+        .await
+        .expect("read back");
+    let mut seen = 0;
+    while cursor.advance().await.expect("cursor") {
+        let d = cursor.deserialize_current().expect("document");
+        let created_at = d.get_datetime("created_at").expect("created_at");
+        let approx = d
+            .get_document("record_data")
+            .and_then(|r| r.get_document("dynamodb"))
+            .expect("record_data.dynamodb")
+            .get_i64("ApproximateCreationDateTime")
+            .expect("ApproximateCreationDateTime");
+        assert_eq!(
+            created_at.timestamp_millis().div_euclid(1_000),
+            approx,
+            "created_at {created_at} and ApproximateCreationDateTime {approx} disagree"
+        );
+        seen += 1;
+    }
+    assert_eq!(seen, 3, "two puts and one delete");
+    fx.finish().await;
+}
diff --git a/crates/storage-mongodb/src/data_engine.rs b/crates/storage-mongodb/src/data_engine.rs
index dd8f4cc9f..1582c7c5b 100644
--- a/crates/storage-mongodb/src/data_engine.rs
+++ b/crates/storage-mongodb/src/data_engine.rs
@@ -2038,6 +2038,9 @@ impl MongoEngine {
             .next_sequence_number_in_session(&shard_id, &mut *session)
             .await?;
 
+        // One wall-clock read feeds both the record's ApproximateCreationDateTime
+        // and the created_at field, so the two never disagree.
+        let created_at = mongodb::bson::DateTime::now();
         let record = StreamRecord {
             event_id: uuid::Uuid::new_v4().to_string(),
             event_name: event,
@@ -2045,13 +2048,7 @@ impl MongoEngine {
             event_source: "aws:dynamodb".to_owned(),
             aws_region: capture.region.to_string(),
             dynamodb: StreamRecordData {
-                approximate_creation_date_time: i64::try_from(
-                    std::time::SystemTime::now()
-                        .duration_since(std::time::UNIX_EPOCH)
-                        .unwrap_or_default()
-                        .as_secs(),
-                )
-                .unwrap_or(i64::MAX),
+                approximate_creation_date_time: created_at.timestamp_millis().div_euclid(1_000),
                 keys,
                 new_image,
                 old_image,
@@ -2081,7 +2078,7 @@ impl MongoEngine {
                 "table_id": table_id,
                 "event_name": crate::stream_engine::event_name_ddb_str(record.event_name),
                 "record_data": record_bson,
-                "created_at": mongodb::bson::DateTime::now(),
+                "created_at": created_at,
             })
             .session(&mut *session)
             .await
diff --git a/crates/storage-postgres/src/data/tx_helpers.rs b/crates/storage-postgres/src/data/tx_helpers.rs
index 6468f159b..d8df053b2 100755
--- a/crates/storage-postgres/src/data/tx_helpers.rs
+++ b/crates/storage-postgres/src/data/tx_helpers.rs
@@ -344,6 +344,12 @@ pub(super) async fn write_stream_record_in_tx(
         .map_err(db_error)?;
     let seq = format!("{seq_val:021}");
 
+    // One wall-clock read feeds both the record's ApproximateCreationDateTime
+    // and the created_at column. The column default (NOW()) is the transaction
+    // start time, which in a long transaction sits seconds before the record.
+    let created_at = time::OffsetDateTime::now_utc();
+    let approximate_creation_date_time = created_at.unix_timestamp();
+
     let record = StreamRecord {
         event_id: uuid::Uuid::new_v4().to_string(),
         event_name: event,
@@ -351,13 +357,7 @@ pub(super) async fn write_stream_record_in_tx(
         event_source: "aws:dynamodb".to_owned(),
         aws_region: capture.region.to_string(),
         dynamodb: StreamRecordData {
-            approximate_creation_date_time: i64::try_from(
-                std::time::SystemTime::now()
-                    .duration_since(std::time::UNIX_EPOCH)
-                    .unwrap_or_default()
-                    .as_secs(),
-            )
-            .unwrap_or(i64::MAX),
+            approximate_creation_date_time,
             keys,
             new_image,
             old_image,
@@ -372,14 +372,15 @@ pub(super) async fn write_stream_record_in_tx(
         serde_json::to_value(&record).map_err(|e| StorageError::Internal(e.to_string()))?;
 
     sqlx::query(
-        "INSERT INTO stream_records (sequence_number, shard_id, table_id, event_name, record_data) \
-         VALUES ($1, $2, $3, $4, $5)",
+        "INSERT INTO stream_records (sequence_number, shard_id, table_id, event_name, record_data, created_at) \
+         VALUES ($1, $2, $3, $4, $5, $6)",
     )
     .bind(&record.dynamodb.sequence_number)
     .bind(shard_id)
     .bind(&key_info.table_id)
     .bind(format!("{:?}", record.event_name))
     .bind(&record_json)
+    .bind(created_at)
     .execute(&mut **tx)
     .await
     .map_err(db_error)?;
diff --git a/crates/storage-postgres/src/stream_engine.rs b/crates/storage-postgres/src/stream_engine.rs
index 970a81755..8a6b609e4 100755
--- a/crates/storage-postgres/src/stream_engine.rs
+++ b/crates/storage-postgres/src/stream_engine.rs
@@ -177,15 +177,22 @@ impl StreamEngine for PostgresEngine {
             .await
             .map_err(|e| StorageError::Internal(e.to_string()))?;
 
+            // created_at follows the record's own timestamp, never a second
+            // clock read (NOW() would be the transaction start time).
+            let created_at = time::OffsetDateTime::from_unix_timestamp(
+                record.dynamodb.approximate_creation_date_time,
+            )
+            .unwrap_or_else(|_| time::OffsetDateTime::now_utc());
             sqlx::query(
-                "INSERT INTO stream_records (sequence_number, shard_id, table_id, event_name, record_data) \
-                 VALUES ($1, $2, $3, $4, $5)",
+                "INSERT INTO stream_records (sequence_number, shard_id, table_id, event_name, record_data, created_at) \
+                 VALUES ($1, $2, $3, $4, $5, $6)",
             )
             .bind(&record.dynamodb.sequence_number)
             .bind(&shard_id)
             .bind(&table_id)
             .bind(format!("{:?}", record.event_name))
             .bind(&record_json)
+            .bind(created_at)
             .execute(&self.data_pool)
             .await
             .map_err(|e| StorageError::Internal(e.to_string()))?;
diff --git a/crates/storage-postgres/tests/stream_record_clock.rs b/crates/storage-postgres/tests/stream_record_clock.rs
new file mode 100644
index 000000000..4aa87c835
--- /dev/null
+++ b/crates/storage-postgres/tests/stream_record_clock.rs
@@ -0,0 +1,333 @@
+// Copyright 2026 ExtendDB contributors
+// SPDX-License-Identifier: Apache-2.0
+
+//! A stream record's `created_at` column and its `ApproximateCreationDateTime`
+//! come from one clock read.
+//!
+//! The column default (`NOW()`) is the transaction start time. A write that
+//! waits inside its transaction, on a row lock here, stores a `created_at`
+//! that is earlier than the record's own timestamp by the length of the wait.
+//! The retention sweep filters on `created_at`, so the two values must agree.
+//!
+//! Needs `EXTENDDB_TEST_PG_CONNECTION_STRING` (host-only, e.g.
+//! `postgres://user:pass@127.0.0.1:5432`); skips otherwise. Each test builds
+//! its own scratch database and drops it afterwards.
+
+use std::collections::BTreeMap;
+use std::sync::Arc;
+use std::time::Duration;
+
+use extenddb_core::expression::ExpressionMaps;
+use extenddb_core::types::{
+    AttributeDefinition, AttributeValue, BillingMode, CreateTableInput, Item, KeySchemaElement,
+    KeyType, ScalarAttributeType, StreamSpecification, StreamViewType, TableKeyInfo, UserIdentity,
+};
+use extenddb_storage::{DataEngine, StreamCapture, TableEngine};
+use extenddb_storage_postgres::{PostgresConfig, PostgresEngine};
+use sqlx::PgPool;
+use sqlx::postgres::PgPoolOptions;
+
+const ACCOUNT: &str = "123456789012";
+const REGION: &str = "us-east-1";
+const TABLE: &str = "t_stream_clock";
+
+struct Scratch {
+    engine: PostgresEngine,
+    db: PgPool,
+    admin: PgPool,
+    db_name: String,
+}
+
+impl Scratch {
+    async fn cleanup(self) {
+        let Scratch {
+            engine,
+            db,
+            admin,
+            db_name,
+        } = self;
+        drop(engine);
+        db.close().await;
+        sqlx::query(&format!(
+            "DROP DATABASE IF EXISTS \"{db_name}\" WITH (FORCE)"
+        ))
+        .execute(&admin)
+        .await
+        .expect("drop the scratch database");
+        admin.close().await;
+    }
+}
+
+fn base_conn() -> Option<String> {
+    let conn = std::env::var("EXTENDDB_TEST_PG_CONNECTION_STRING").ok()?;
+    (!conn.trim().is_empty()).then(|| conn.trim_end_matches('/').to_owned())
+}
+
+fn skip(test: &str) {
+    eprintln!(
+        "SKIP {test}: EXTENDDB_TEST_PG_CONNECTION_STRING is not set, so there is no PostgreSQL \
+         to build a scratch catalog in."
+    );
+}
+
+fn shipped_migrations() -> Vec<String> {
+    let root = std::path::Path::new(env!("CARGO_MANIFEST_DIR"));
+    let mut out = Vec::new();
+    for dir in ["migrations", "data_migrations"] {
+        let mut files: Vec<_> = std::fs::read_dir(root.join(dir))
+            .expect("read the migrations directory")
+            .map(|e| e.expect("directory entry").path())
+            .filter(|p| p.extension().is_some_and(|e| e == "sql"))
+            .collect();
+        files.sort();
+        for f in files {
+            out.push(std::fs::read_to_string(&f).expect("read a migration file"));
+        }
+    }
+    out
+}
+
+async fn scratch() -> Scratch {
+    let base = base_conn().expect("caller checks base_conn() first");
+    let db_name = format!("eddb_sclk_{}", uuid::Uuid::new_v4().simple())[..24].to_owned();
+    let admin = PgPoolOptions::new()
+        .max_connections(1)
+        .connect(&format!("{base}/postgres"))
+        .await
+        .expect("connect to the postgres maintenance database");
+    sqlx::query(&format!("CREATE DATABASE \"{db_name}\""))
+        .execute(&admin)
+        .await
+        .expect("create the scratch database");
+    let url = format!("{base}/{db_name}");
+    let db = PgPoolOptions::new()
+        .max_connections(3)
+        .connect(&url)
+        .await
+        .expect("connect to the scratch database");
+    for sql in shipped_migrations() {
+        sqlx::raw_sql(&sql)
+            .execute(&db)
+            .await
+            .expect("apply a shipped migration");
+    }
+    sqlx::query("ALTER TABLE stream_shards DROP CONSTRAINT IF EXISTS stream_shards_table_id_fkey")
+        .execute(&db)
+        .await
+        .expect("match the two-database layout");
+    sqlx::query("UPDATE settings SET value = '0' WHERE key = 'control_plane_delay_seconds'")
+        .execute(&db)
+        .await
+        .expect("pin the control-plane delay to zero");
+    sqlx::query("INSERT INTO accounts (account_id, account_name) VALUES ($1, $2)")
+        .bind(ACCOUNT)
+        .bind(format!("acct-{db_name}"))
+        .execute(&db)
+        .await
+        .expect("seed the account row");
+    let engine = PostgresEngine::new(
+        &PostgresConfig {
+            connection_string: url,
+            pool_size: 10,
+            max_item_size_bytes: 400_000,
+        },
+        REGION,
+    )
+    .await
+    .expect("open a PostgresEngine on the scratch database");
+    Scratch {
+        engine,
+        db,
+        admin,
+        db_name,
+    }
+}
+
+fn item(pk: &str, v: &str) -> Item {
+    BTreeMap::from([
+        ("pk".to_owned(), AttributeValue::S(pk.to_owned())),
+        ("v".to_owned(), AttributeValue::S(v.to_owned())),
+    ])
+}
+
+async fn stream_table(s: &Scratch) -> TableKeyInfo {
+    s.engine
+        .create_table(
+            ACCOUNT,
+            CreateTableInput {
+                table_name: TABLE.to_owned(),
+                key_schema: vec![KeySchemaElement {
+                    attribute_name: "pk".to_owned(),
+                    key_type: KeyType::Hash,
+                }],
+                attribute_definitions: vec![AttributeDefinition {
+                    attribute_name: "pk".to_owned(),
+                    attribute_type: ScalarAttributeType::S,
+                }],
+                billing_mode: Some(BillingMode::PayPerRequest),
+                stream_specification: Some(StreamSpecification {
+                    stream_enabled: true,
+                    stream_view_type: Some(StreamViewType::NewImage),
+                }),
+                ..Default::default()
+            },
+        )
+        .await
+        .expect("create the table");
+    s.engine
+        .table_key_info(ACCOUNT, TABLE)
+        .await
+        .expect("read the key info")
+}
+
+async fn data_table(db: &PgPool) -> String {
+    let id: String =
+        sqlx::query_scalar("SELECT table_id FROM tables WHERE account_id = $1 AND table_name = $2")
+            .bind(ACCOUNT)
+            .bind(TABLE)
+            .fetch_one(db)
+            .await
+            .expect("look up the table id");
+    format!("\"_ddb_{id}\"")
+}
+
+async fn wait_for_lock_waiters(db: &PgPool, own_pid: i32, n: i64) {
+    for _ in 0..500 {
+        let waiting: i64 = sqlx::query_scalar(
+            "SELECT count(*) FROM pg_stat_activity \
+             WHERE datname = current_database() AND wait_event_type = 'Lock' AND pid <> $1",
+        )
+        .bind(own_pid)
+        .fetch_one(db)
+        .await
+        .expect("read pg_stat_activity");
+        if waiting >= n {
+            return;
+        }
+        tokio::time::sleep(Duration::from_millis(10)).await;
+    }
+    panic!("{n} backends never waited on a lock");
+}
+
+/// Stored `(created_at as epoch seconds, ApproximateCreationDateTime)` for
+/// every record in the scratch database.
+async fn stored_times(db: &PgPool) -> Vec<(i64, i64)> {
+    let rows: Vec<(f64, serde_json::Value)> = sqlx::query_as(
+        "SELECT EXTRACT(EPOCH FROM created_at)::float8, record_data FROM stream_records \
+         ORDER BY sequence_number",
+    )
+    .fetch_all(db)
+    .await
+    .expect("read the stream records");
+    rows.into_iter()
+        .map(|(created, data)| {
+            let approx = data["dynamodb"]["ApproximateCreationDateTime"]
+                .as_i64()
+                .expect("ApproximateCreationDateTime is an integer");
+            #[allow(clippy::cast_possible_truncation)]
+            let created_secs = created.floor() as i64;
+            (created_secs, approx)
+        })
+        .collect()
+}
+
+#[tokio::test]
+async fn created_at_matches_the_record_timestamp_when_the_write_waited() {
+    let test = "created_at_matches_the_record_timestamp_when_the_write_waited";
+    if base_conn().is_none() {
+        return skip(test);
+    }
+    let s = scratch().await;
+    let maps = ExpressionMaps::default();
+    let key_info = stream_table(&s).await;
+    let capture = StreamCapture {
+        view_type: StreamViewType::NewImage,
+        user_identity: None,
+        region: Arc::from(REGION),
+    };
+    // Seed so the write below takes the locked update path.
+    s.engine
+        .put_item(
+            &key_info,
+            item("k", "seed"),
+            false,
+            None,
+            &maps,
+            Some(&capture),
+        )
+        .await
+        .expect("seed the item");
+    let table = data_table(&s.db).await;
+
+    // An outside transaction holds the row. The write begins its own
+    // transaction (fixing NOW()), waits on the lock for longer than a second,
+    // then builds its record once the lock is released.
+    let mut holder = s.db.begin().await.expect("begin the outside transaction");
+    let holder_pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()")
+        .fetch_one(&mut *holder)
+        .await
+        .expect("read the backend pid");
+    sqlx::query(&format!("SELECT 1 FROM {table} WHERE pk = 'k' FOR UPDATE"))
+        .execute(&mut *holder)
+        .await
+        .expect("lock the row");
+
+    let write = s.engine.put_item(
+        &key_info,
+        item("k", "late"),
+        false,
+        None,
+        &maps,
+        Some(&capture),
+    );
+    let release = async {
+        wait_for_lock_waiters(&s.db, holder_pid, 1).await;
+        // Past the next second boundary and then some, so NOW() and the
+        // record's own clock read fall in different seconds.
+        tokio::time::sleep(Duration::from_millis(2_100)).await;
+        holder.commit().await.expect("release the row");
+    };
+    let (result, ()) = tokio::time::timeout(Duration::from_secs(30), async {
+        tokio::join!(write, release)
+    })
+    .await
+    .expect("the write finished");
+    result.expect("the write succeeded");
+
+    // A service-identity deletion, the TTL worker's shape, through the same
+    // helper.
+    s.engine
+        .delete_item(
+            &key_info,
+            &BTreeMap::from([("pk".to_owned(), AttributeValue::S("k".to_owned()))]),
+            false,
+            None,
+            &maps,
+            Some(&StreamCapture {
+                view_type: StreamViewType::NewImage,
+                user_identity: Some(UserIdentity {
+                    identity_type: "Service".to_owned(),
+                    principal_id: "dynamodb.amazonaws.com".to_owned(),
+                }),
+                region: Arc::from(REGION),
+            }),
+        )
+        .await
+        .expect("the TTL-shaped delete succeeded");
+
+    let times = stored_times(&s.db).await;
+    assert_eq!(times.len(), 3, "seed, late, and remove records: {times:?}");
+    for (created_secs, approx) in &times {
+        assert_eq!(
+            created_secs, approx,
+            "created_at second must equal ApproximateCreationDateTime: {times:?}"
+        );
+    }
+    // And the waited record is at least two seconds after the seed.
+    assert!(
+        times[1].1 - times[0].1 >= 2,
+        "the late record carries the post-wait time: {times:?}"
+    );
+
+    s.cleanup().await;
+}
diff --git a/crates/storage-sqlite/src/data/tx_helpers.rs b/crates/storage-sqlite/src/data/tx_helpers.rs
index 1fd1bfca8..0ff033523 100644
--- a/crates/storage-sqlite/src/data/tx_helpers.rs
+++ b/crates/storage-sqlite/src/data/tx_helpers.rs
@@ -230,13 +230,16 @@ pub(super) async fn write_stream_record_in_tx(
     let shard_id = shards[idx].0.clone();
 
     let seq = format!("{:021}", next_stream_seq(tx).await?);
-    let approximate_creation_date_time = i64::try_from(
+    // One wall-clock read feeds both the record's ApproximateCreationDateTime
+    // and the created_at column, so the two never disagree.
+    let created_at_ms = i64::try_from(
         std::time::SystemTime::now()
             .duration_since(std::time::UNIX_EPOCH)
             .unwrap_or_default()
-            .as_secs(),
+            .as_millis(),
     )
     .unwrap_or(i64::MAX);
+    let approximate_creation_date_time = created_at_ms.div_euclid(1_000);
 
     let record = StreamRecord {
         event_id: uuid::Uuid::new_v4().to_string(),
@@ -259,14 +262,15 @@ pub(super) async fn write_stream_record_in_tx(
         serde_json::to_string(&record).map_err(|e| StorageError::Internal(e.to_string()))?;
 
     sqlx::query(
-        "INSERT INTO stream_records (shard_id, sequence_number, table_id, event_name, record_data) \
-         VALUES (?, ?, ?, ?, ?)",
+        "INSERT INTO stream_records (shard_id, sequence_number, table_id, event_name, record_data, created_at) \
+         VALUES (?, ?, ?, ?, ?, ?)",
     )
     .bind(&shard_id)
     .bind(&seq)
     .bind(&key_info.table_id)
     .bind(format!("{:?}", record.event_name))
     .bind(&record_json)
+    .bind(crate::sqlite_util::format_epoch_millis(created_at_ms))
     .execute(&mut **tx)
     .await
     .map_err(|e| StorageError::Internal(e.to_string()))?;
@@ -378,3 +382,116 @@ mod idempotency_tests {
         check(&pool, "acctB", "tok", "fpX").await.unwrap();
     }
 }
+
+#[cfg(test)]
+mod stream_clock_tests {
+    //! The stored `created_at` and the record's own `ApproximateCreationDateTime`
+    //! come from one clock read on the engine write path.
+
+    use std::collections::BTreeMap;
+    use std::sync::Arc;
+
+    use extenddb_core::expression::ExpressionMaps;
+    use extenddb_core::types::{AttributeValue, Item, StreamViewType};
+    use extenddb_storage::{DataEngine, StreamCapture, TableEngine};
+    use serde_json::json;
+
+    use crate::SqliteEngine;
+
+    const ACCOUNT: &str = "123456789012";
+
+    async fn engine() -> SqliteEngine {
+        let engine = SqliteEngine::new(":memory:", 1, "us-east-1", 409_600)
+            .await
+            .expect("engine");
+        crate::schema::apply(&engine.pool).await.expect("schema");
+        sqlx::query("UPDATE settings SET value = '0' WHERE key = 'control_plane_delay_seconds'")
+            .execute(&engine.pool)
+            .await
+            .expect("delay");
+        sqlx::query("INSERT INTO accounts (account_id, account_name) VALUES (?, 'default')")
+            .bind(ACCOUNT)
+            .execute(&engine.pool)
+            .await
+            .expect("account");
+        engine
+    }
+
+    #[tokio::test]
+    async fn stored_created_at_matches_the_record_timestamp() {
+        let engine = engine().await;
+        let input: extenddb_core::types::CreateTableInput = serde_json::from_value(json!({
+            "TableName": "t_clock",
+            "KeySchema": [{"AttributeName": "pk", "KeyType": "HASH"}],
+            "AttributeDefinitions": [{"AttributeName": "pk", "AttributeType": "S"}],
+            "BillingMode": "PAY_PER_REQUEST",
+            "StreamSpecification": {"StreamEnabled": true, "StreamViewType": "NEW_AND_OLD_IMAGES"}
+        }))
+        .expect("input");
+        engine.create_table(ACCOUNT, input).await.expect("create");
+        let key_info = engine
+            .table_key_info(ACCOUNT, "t_clock")
+            .await
+            .expect("key info");
+        let capture = StreamCapture {
+            view_type: StreamViewType::NewAndOldImages,
+            user_identity: None,
+            region: Arc::from("us-east-1"),
+        };
+        let maps = ExpressionMaps::default();
+        for v in ["one", "two", "three"] {
+            let item: Item = BTreeMap::from([
+                ("pk".to_owned(), AttributeValue::S("k".to_owned())),
+                ("v".to_owned(), AttributeValue::S(v.to_owned())),
+            ]);
+            engine
+                .put_item(&key_info, item, false, None, &maps, Some(&capture))
+                .await
+                .expect("put");
+        }
+        // Expiry of the item deletes it with a service identity, the TTL
+        // worker's shape, through the same helper.
+        engine
+            .delete_item(
+                &key_info,
+                &BTreeMap::from([("pk".to_owned(), AttributeValue::S("k".to_owned()))]),
+                false,
+                None,
+                &maps,
+                Some(&StreamCapture {
+                    view_type: StreamViewType::NewAndOldImages,
+                    user_identity: Some(extenddb_core::types::UserIdentity {
+                        identity_type: "Service".to_owned(),
+                        principal_id: "dynamodb.amazonaws.com".to_owned(),
+                    }),
+                    region: Arc::from("us-east-1"),
+                }),
+            )
+            .await
+            .expect("delete");
+
+        let rows: Vec<(String, String)> = sqlx::query_as(
+            "SELECT created_at, record_data FROM stream_records ORDER BY sequence_number",
+        )
+        .fetch_all(&engine.pool)
+        .await
+        .expect("read back");
+        assert_eq!(rows.len(), 4, "{rows:?}");
+        for (created_at, data) in &rows {
+            // Same shape as the column default: 24 characters, millisecond
+            // fraction, trailing Z.
+            assert_eq!(created_at.len(), 24, "{created_at}");
+            let record: serde_json::Value = serde_json::from_str(data).expect("record json");
+            let approx = record["dynamodb"]["ApproximateCreationDateTime"]
+                .as_i64()
+                .expect("ApproximateCreationDateTime");
+            let stored_secs = crate::sqlite_util::timestamp_epoch_seconds(created_at).floor();
+            #[allow(clippy::cast_possible_truncation)]
+            let stored_secs = stored_secs as i64;
+            assert_eq!(
+                stored_secs, approx,
+                "created_at {created_at} second must equal ApproximateCreationDateTime {approx}"
+            );
+        }
+    }
+}
diff --git a/crates/storage-sqlite/src/sqlite_util.rs b/crates/storage-sqlite/src/sqlite_util.rs
index c37ce480f..917f1daaf 100644
--- a/crates/storage-sqlite/src/sqlite_util.rs
+++ b/crates/storage-sqlite/src/sqlite_util.rs
@@ -53,6 +53,26 @@ pub(crate) fn format_epoch_seconds(secs: f64) -> String {
     )
 }
 
+/// Format epoch milliseconds in the shape the `stream_records.created_at`
+/// column default produces (`strftime('%Y-%m-%dT%H:%M:%fZ')`, millisecond
+/// precision, 24 characters): `2026-10-15T12:00:00.438Z`. Same width as the
+/// default, so rows written before and after this formatter was introduced
+/// sort together as text in time order.
+pub(crate) fn format_epoch_millis(ms: i64) -> String {
+    let dt = OffsetDateTime::from_unix_timestamp_nanos(i128::from(ms) * 1_000_000)
+        .unwrap_or(OffsetDateTime::UNIX_EPOCH);
+    format!(
+        "{:04}-{:02}-{:02}T{:02}:{:02}:{:02}.{:03}Z",
+        dt.year(),
+        u8::from(dt.month()),
+        dt.day(),
+        dt.hour(),
+        dt.minute(),
+        dt.second(),
+        dt.millisecond()
+    )
+}
+
 /// Epoch seconds, sub-second precision, of a catalog timestamp; 0 if it does
 /// not parse.
 #[allow(clippy::cast_precision_loss)] // epoch seconds fit f64's mantissa
@@ -93,6 +113,26 @@ pub(crate) fn is_fk_violation(e: &sqlx::Error) -> bool {
 mod tests {
     use super::*;
 
+    /// Same shape and width as the `stream_records.created_at` column default
+    /// (`strftime('%Y-%m-%dT%H:%M:%fZ')`), so rows from before and after the
+    /// explicit bind sort together as text.
+    #[test]
+    fn epoch_millis_matches_the_column_default_shape() {
+        let s = format_epoch_millis(1_792_765_147_123);
+        assert_eq!(s, "2026-10-23T14:19:07.123Z");
+        assert_eq!(s.len(), 24);
+        // Text order equals time order against a default-shaped neighbour.
+        assert!(s.as_str() < "2026-10-23T14:19:07.124Z");
+        assert!(s.as_str() > "2026-10-23T14:19:06.999Z");
+        // Against the retention cutoff (RFC 3339 from the `time` crate, no
+        // fraction on a whole second) the comparison is decided by the second
+        // digits in every case but the cutoff's own second, where `Z` sorts
+        // after `.`; that edge predates this formatter and is the default's
+        // behaviour too.
+        let cutoff = format_timestamp(OffsetDateTime::from_unix_timestamp(1_792_765_146).unwrap());
+        assert!(s.as_str() > cutoff.as_str(), "{s} vs {cutoff}");
+    }
+
     #[test]
     fn sqlite_url_variants() {
         assert_eq!(sqlite_url(":memory:"), "sqlite::memory:");
diff --git a/crates/storage-sqlite/src/stream.rs b/crates/storage-sqlite/src/stream.rs
index 7cd3d4729..5b00eabfb 100644
--- a/crates/storage-sqlite/src/stream.rs
+++ b/crates/storage-sqlite/src/stream.rs
@@ -93,15 +93,24 @@ impl StreamEngine for SqliteEngine {
             .map_err(|e| StorageError::Internal(e.to_string()))?;
 
             let _writer = self.write_lock.lock().await;
+            // created_at follows the record's own timestamp (whole seconds here,
+            // since that is all the record carries), never a second clock read.
+            let created_at = crate::sqlite_util::format_epoch_millis(
+                record
+                    .dynamodb
+                    .approximate_creation_date_time
+                    .saturating_mul(1_000),
+            );
             sqlx::query(
-                "INSERT INTO stream_records (sequence_number, shard_id, table_id, event_name, record_data) \
-                 VALUES (?, ?, ?, ?, ?)",
+                "INSERT INTO stream_records (sequence_number, shard_id, table_id, event_name, record_data, created_at) \
+                 VALUES (?, ?, ?, ?, ?, ?)",
             )
             .bind(&record.dynamodb.sequence_number)
             .bind(&shard_id)
             .bind(&table_id)
             .bind(format!("{:?}", record.event_name))
             .bind(&record_json)
+            .bind(created_at)
             .execute(&self.pool)
             .await
             .map_err(|e| StorageError::Internal(e.to_string()))?;

From cf02ee2a850b1fe12f4abda310407542701b959d Mon Sep 17 00:00:00 2001
From: Lee Hannigan <29015211+LeeroyHannigan@users.noreply.github.com>
Date: Thu, 29 Oct 2026 21:00:37 +0000
Subject: [PATCH 02/10] ci: run the PostgreSQL stream clock test in the
 integration job

stream_record_clock skips when EXTENDDB_TEST_PG_CONNECTION_STRING is
unset, and the storage-level step that sets the variable names its test
targets one by one without this test, so in CI it finished in 0.00 s on
the skip path. The step now names it.

The test holds a row lock from an outside transaction while a PutItem
with stream capture waits 2.1 s on it, then checks that every stored
created_at second equals the record's ApproximateCreationDateTime. With
the created_at bind removed from the stream insert, the column falls back
to NOW(), the transaction start, and the test fails on that check.
---
 .github/workflows/integration.yml | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/.github/workflows/integration.yml b/.github/workflows/integration.yml
index da6c251d3..a02028404 100644
--- a/.github/workflows/integration.yml
+++ b/.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.

From 987ab48dc5bd06cc12e0636f81d3eec7d6869e42 Mon Sep 17 00:00:00 2001
From: Lee Hannigan <29015211+LeeroyHannigan@users.noreply.github.com>
Date: Thu, 29 Oct 2026 21:00:37 +0000
Subject: [PATCH 03/10] fix(streams): read the PostgreSQL record time from the
 database clock

The stream helper took created_at and ApproximateCreationDateTime from
the ExtendDB host clock. The retention sweep deletes rows where
created_at < NOW() - 24 hours, on the database clock. With the host
behind the database by S seconds, every record was deleted S seconds
early, and a host more than 24 hours behind lost each record on the next
sweep.

The helper now reads clock_timestamp() in the statement that already
draws nextval('stream_seq'). Both values derive from that reading, so
each record keeps one clock read and the write adds no round trip. The
retention window is measured on the database clock again, as it was
when created_at was the column default, without the default's
transaction start offset.

The PostgreSQL clock test also reads clock_timestamp() before the outside
transaction releases the row and after the write returns. It asserts
that the waited record's created_at lies between the two reads.
---
 .../storage-postgres/src/data/tx_helpers.rs   | 23 +++++-----
 .../tests/stream_record_clock.rs              | 42 +++++++++++++------
 2 files changed, 43 insertions(+), 22 deletions(-)

diff --git a/crates/storage-postgres/src/data/tx_helpers.rs b/crates/storage-postgres/src/data/tx_helpers.rs
index d8df053b2..d17d0c437 100755
--- a/crates/storage-postgres/src/data/tx_helpers.rs
+++ b/crates/storage-postgres/src/data/tx_helpers.rs
@@ -337,17 +337,20 @@ pub(super) async fn write_stream_record_in_tx(
     let idx = (hash as usize) % shards.len();
     let shard_id = &shards[idx].0;
 
-    // Generate monotonic sequence number within the transaction (CB-21).
-    let (seq_val,): (i64,) = sqlx::query_as("SELECT nextval('stream_seq')")
-        .fetch_one(&mut **tx)
-        .await
-        .map_err(db_error)?;
+    // Generate monotonic sequence number within the transaction (CB-21). The
+    // same statement takes the record's one clock read: clock_timestamp() is
+    // the database clock when this statement runs, after the write's locking
+    // reads, while the column default (NOW()) is the transaction start, which
+    // in a long transaction sits seconds before the record. created_at and
+    // ApproximateCreationDateTime both derive from this reading, and the
+    // retention sweep compares created_at with the database's NOW(), so the
+    // retention window is measured on the database clock alone.
+    let (seq_val, created_at): (i64, time::OffsetDateTime) =
+        sqlx::query_as("SELECT nextval('stream_seq'), clock_timestamp()")
+            .fetch_one(&mut **tx)
+            .await
+            .map_err(db_error)?;
     let seq = format!("{seq_val:021}");
-
-    // One wall-clock read feeds both the record's ApproximateCreationDateTime
-    // and the created_at column. The column default (NOW()) is the transaction
-    // start time, which in a long transaction sits seconds before the record.
-    let created_at = time::OffsetDateTime::now_utc();
     let approximate_creation_date_time = created_at.unix_timestamp();
 
     let record = StreamRecord {
diff --git a/crates/storage-postgres/tests/stream_record_clock.rs b/crates/storage-postgres/tests/stream_record_clock.rs
index 4aa87c835..f3913f9ed 100644
--- a/crates/storage-postgres/tests/stream_record_clock.rs
+++ b/crates/storage-postgres/tests/stream_record_clock.rs
@@ -8,6 +8,10 @@
 //! waits inside its transaction, on a row lock here, stores a `created_at`
 //! that is earlier than the record's own timestamp by the length of the wait.
 //! The retention sweep filters on `created_at`, so the two values must agree.
+//! The one reading is the database's `clock_timestamp()`, the clock the
+//! sweep's `NOW()` cutoff is taken on, so the test also bounds it with
+//! database clock reads taken before the lock is released and after the
+//! write returns.
 //!
 //! Needs `EXTENDDB_TEST_PG_CONNECTION_STRING` (host-only, e.g.
 //! `postgres://user:pass@127.0.0.1:5432`); skips otherwise. Each test builds
@@ -209,12 +213,11 @@ async fn wait_for_lock_waiters(db: &PgPool, own_pid: i32, n: i64) {
     panic!("{n} backends never waited on a lock");
 }
 
-/// Stored `(created_at as epoch seconds, ApproximateCreationDateTime)` for
-/// every record in the scratch database.
-async fn stored_times(db: &PgPool) -> Vec<(i64, i64)> {
-    let rows: Vec<(f64, serde_json::Value)> = sqlx::query_as(
-        "SELECT EXTRACT(EPOCH FROM created_at)::float8, record_data FROM stream_records \
-         ORDER BY sequence_number",
+/// Stored `(created_at, ApproximateCreationDateTime)` for every record in the
+/// scratch database, in sequence order.
+async fn stored_times(db: &PgPool) -> Vec<(time::OffsetDateTime, i64)> {
+    let rows: Vec<(time::OffsetDateTime, serde_json::Value)> = sqlx::query_as(
+        "SELECT created_at, record_data FROM stream_records ORDER BY sequence_number",
     )
     .fetch_all(db)
     .await
@@ -224,9 +227,7 @@ async fn stored_times(db: &PgPool) -> Vec<(i64, i64)> {
             let approx = data["dynamodb"]["ApproximateCreationDateTime"]
                 .as_i64()
                 .expect("ApproximateCreationDateTime is an integer");
-            #[allow(clippy::cast_possible_truncation)]
-            let created_secs = created.floor() as i64;
-            (created_secs, approx)
+            (created, approx)
         })
         .collect()
 }
@@ -285,14 +286,23 @@ async fn created_at_matches_the_record_timestamp_when_the_write_waited() {
         // Past the next second boundary and then some, so NOW() and the
         // record's own clock read fall in different seconds.
         tokio::time::sleep(Duration::from_millis(2_100)).await;
+        let released_at: time::OffsetDateTime = sqlx::query_scalar("SELECT clock_timestamp()")
+            .fetch_one(&mut *holder)
+            .await
+            .expect("read the database clock before the release");
         holder.commit().await.expect("release the row");
+        released_at
     };
-    let (result, ()) = tokio::time::timeout(Duration::from_secs(30), async {
+    let (result, released_at) = tokio::time::timeout(Duration::from_secs(30), async {
         tokio::join!(write, release)
     })
     .await
     .expect("the write finished");
     result.expect("the write succeeded");
+    let written_by: time::OffsetDateTime = sqlx::query_scalar("SELECT clock_timestamp()")
+        .fetch_one(&s.db)
+        .await
+        .expect("read the database clock after the write");
 
     // A service-identity deletion, the TTL worker's shape, through the same
     // helper.
@@ -317,9 +327,10 @@ async fn created_at_matches_the_record_timestamp_when_the_write_waited() {
 
     let times = stored_times(&s.db).await;
     assert_eq!(times.len(), 3, "seed, late, and remove records: {times:?}");
-    for (created_secs, approx) in &times {
+    for (created_at, approx) in &times {
         assert_eq!(
-            created_secs, approx,
+            created_at.unix_timestamp(),
+            *approx,
             "created_at second must equal ApproximateCreationDateTime: {times:?}"
         );
     }
@@ -328,6 +339,13 @@ async fn created_at_matches_the_record_timestamp_when_the_write_waited() {
         times[1].1 - times[0].1 >= 2,
         "the late record carries the post-wait time: {times:?}"
     );
+    // Its created_at is a database clock read taken after the lock was
+    // released and before the write returned.
+    let late = times[1].0;
+    assert!(
+        released_at <= late && late <= written_by,
+        "created_at {late} must be a database clock read between {released_at} and {written_by}"
+    );
 
     s.cleanup().await;
 }

From 9430ee96cf7b3a024191f0561d1da36f4f69664b Mon Sep 17 00:00:00 2001
From: Lee Hannigan <29015211+LeeroyHannigan@users.noreply.github.com>
Date: Thu, 29 Oct 2026 21:00:37 +0000
Subject: [PATCH 04/10] fix(streams): keep the HLC overflow prefix above
 earlier readings

When the logical counter passes 999,999, HybridClock sleeps a
millisecond and re-reads the wall clock for the sequence prefix. The
branch assigned the raw re-read. If the clock stepped back during the
sleep, the re-read sat below the call's own reading and the returned
created_at_ms exceeded the prefix of its own sequence number. Under a
forward clamp the re-read sat below the previous prefix and the sequence
number went backwards by the clamp width, the regression reported in
issue #465; this change fixes that too.

The overflow branch now sets the prefix to the later of the re-read and
the previous prefix plus one. The prefix stays above every earlier
reading, and the sequence number sorts after every earlier one. The doc
comment states what the method returns: the wall-clock reading taken by
this call, with a sequence prefix equal to that reading or a later
clamped value.

The clock step moved into generate_from, which takes the reading and a
re-read function as arguments. A unit test drives the overflow branch
with a re-read ten milliseconds below the reading and asserts
created_at_ms <= hlc_ms and that the sequence follows the previous one.
With the raw re-read assigned again, the test fails on the first
assertion.
---
 crates/storage-cassandra/src/stream_util.rs | 59 +++++++++++++++------
 1 file changed, 44 insertions(+), 15 deletions(-)

diff --git a/crates/storage-cassandra/src/stream_util.rs b/crates/storage-cassandra/src/stream_util.rs
index 49e00f4cb..31fd15304 100644
--- a/crates/storage-cassandra/src/stream_util.rs
+++ b/crates/storage-cassandra/src/stream_util.rs
@@ -61,17 +61,27 @@ impl HybridClock {
         self.generate_with_created_at().0
     }
 
-    /// Generate the next sequence number and the wall-clock reading it was
-    /// derived from, as `(sequence_number, created_at_ms)`.
+    /// Generate the next sequence number and the wall-clock reading taken by
+    /// this call, as `(sequence_number, created_at_ms)`.
     ///
-    /// Stream records store both values. Taking them from one read keeps
-    /// `created_at_ms <= hlc_ms` for every record: the clock clamps its
-    /// physical component upward on a backwards step but never below the
-    /// reading, so a record's stored creation time never exceeds the
-    /// timestamp prefix of its own sequence number.
+    /// 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;
@@ -92,14 +102,11 @@ 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 above every earlier sequence number.
+                self.last_timestamp_ms = reread().max(self.last_timestamp_ms + 1);
                 self.logical_counter = 0;
             }
         }
@@ -363,6 +370,28 @@ mod tests {
         );
     }
 
+    /// 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}"
+        );
+    }
+
     #[test]
     fn test_sequence_length() {
         let mut clock = HybridClock::new(1);

From e437ddb8fab20921260c882906c71b35fbfdd661 Mon Sep 17 00:00:00 2001
From: Lee Hannigan <29015211+LeeroyHannigan@users.noreply.github.com>
Date: Thu, 29 Oct 2026 21:00:38 +0000
Subject: [PATCH 05/10] test(streams): pin the HLC reading on the clamp and
 overflow branches

created_at_never_exceeds_sequence_prefix checked only created_at_ms <=
hlc_ms on the clamp branch, so a clock that returned the clamped prefix
in place of its reading passed. Each clamped call now has a wall-clock
read on either side of it, and the test asserts that the returned value
lies between the two; with the clamped prefix returned, it fails on that
assertion.

created_at_holds_across_counter_overflow reached the overflow branch
only when its two calls fell in the same millisecond. Otherwise the
second call took the first branch and every assertion still held. The
test now drives the step with injected readings and asserts that the
re-read ran, that created_at_ms is the call's own reading, that the
prefix is the re-read, and that the counter resets. With the overflow
branch skipped, it fails on the re-read assertion.
---
 crates/storage-cassandra/src/stream_util.rs | 38 +++++++++++++++------
 1 file changed, 28 insertions(+), 10 deletions(-)

diff --git a/crates/storage-cassandra/src/stream_util.rs b/crates/storage-cassandra/src/stream_util.rs
index 31fd15304..b898ba9ff 100644
--- a/crates/storage-cassandra/src/stream_util.rs
+++ b/crates/storage-cassandra/src/stream_util.rs
@@ -311,22 +311,29 @@ mod tests {
     /// 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).
+    /// 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}"
+            );
         }
-        // And on the normal branch the two agree exactly: created_at_ms is the
-        // wall-clock read itself, not the clamped value.
+        // 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();
@@ -339,18 +346,29 @@ mod tests {
         );
     }
 
-    /// The counter-overflow branch sleeps a millisecond and re-reads the
-    /// clock for the sequence prefix. The returned `created_at_ms` is still
-    /// the original read, so it stays at or below the prefix.
+    /// 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);
-        // Land on the same-millisecond path, then drive the counter to the
-        // limit so the next call takes the overflow branch.
-        let (first, _) = clock.generate_with_created_at();
+        let (first, _) =
+            clock.generate_from(now_ms, || unreachable!("no overflow on a fresh clock"));
         clock.logical_counter = 999_999;
-        let (seq, created_at_ms) = clock.generate_with_created_at();
+        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}"

From f208bfb5e11fcc7fa66d9b2e9950709a2ef1bfa7 Mon Sep 17 00:00:00 2001
From: Lee Hannigan <29015211+LeeroyHannigan@users.noreply.github.com>
Date: Thu, 29 Oct 2026 21:00:38 +0000
Subject: [PATCH 06/10] test(streams): pin the Cassandra TTL record's clock
 pair

The direct streams test's comment said its service-identity delete had
the TTL worker's shape. That delete goes through delete_item and
stream_record_statement, while the TTL worker writes through
apply_ttl_delete_effects with the TtlStreamPlan it persisted. The comment
now says the delete runs through delete_item, and that the sequence
prefix is the reading or a later clamped value.

The TTL sweep test now reads sequence_number and created_at beside
record_data on the REMOVE record and asserts created_at_ms <= hlc_ms and
that ApproximateCreationDateTime is created_at_ms / 1000, the properties
the direct test checks for the other writes. A plan that stores a
created_at_ms one second past its reading fails the first assertion. A
plan that took a second clock read microseconds after the sequence
number would fail it only when a millisecond boundary fell between the
two reads.
---
 .../storage-cassandra/tests/direct/streams.rs | 10 +++----
 .../tests/ttl_integration.rs                  | 27 ++++++++++++++++---
 2 files changed, 28 insertions(+), 9 deletions(-)

diff --git a/crates/storage-cassandra/tests/direct/streams.rs b/crates/storage-cassandra/tests/direct/streams.rs
index b7be9820f..bc0efa206 100644
--- a/crates/storage-cassandra/tests/direct/streams.rs
+++ b/crates/storage-cassandra/tests/direct/streams.rs
@@ -654,11 +654,11 @@ async fn test_list_streams_includes_stream_enabled_table() {
     assert!(streams.iter().any(|s| s.stream_arn == expected_arn));
 }
 
-/// `created_at`, the sequence number's millisecond prefix, and the record's
-/// `ApproximateCreationDateTime` all derive from one clock read: for every
-/// stored record `created_at_ms <= hlc_ms` and `created_at_ms / 1000` equals
-/// the record's timestamp. Covers a plain write and a service-identity
-/// deletion (the TTL worker's shape).
+/// `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() {
diff --git a/crates/storage-cassandra/tests/ttl_integration.rs b/crates/storage-cassandra/tests/ttl_integration.rs
index 279d79492..4f712918a 100644
--- a/crates/storage-cassandra/tests/ttl_integration.rs
+++ b/crates/storage-cassandra/tests/ttl_integration.rs
@@ -593,7 +593,8 @@ async fn test_ttl_sweep_emits_service_remove_stream_record() {
         &key_info.table_id,
     );
     let query = format!(
-        "SELECT event_name, record_data FROM {account_keyspace}.stream_records WHERE shard_id = ?"
+        "SELECT event_name, sequence_number, created_at, record_data \
+         FROM {account_keyspace}.stream_records WHERE shard_id = ?"
     );
     let rows = engine
         .session_arc()
@@ -604,20 +605,38 @@ async fn test_ttl_sweep_emits_service_remove_stream_record() {
         .expect("parse TTL stream response")
         .into_rows()
         .unwrap_or_default();
-    let remove_records: Vec<String> = rows
+    let remove_records: Vec<(String, i64, String)> = rows
         .into_iter()
         .filter_map(|row| {
             let event: String = row.get_r_by_name("event_name").ok()?;
             if event == "Remove" {
-                row.get_r_by_name("record_data").ok()
+                Some((
+                    row.get_r_by_name("sequence_number")
+                        .expect("sequence_number"),
+                    row.get_r_by_name("created_at").expect("created_at"),
+                    row.get_r_by_name("record_data").expect("record_data"),
+                ))
             } else {
                 None
             }
         })
         .collect();
     assert_eq!(remove_records.len(), 1);
+    let (seq, created_at_ms, data) = &remove_records[0];
     let record: serde_json::Value =
-        serde_json::from_str(&remove_records[0]).expect("parse TTL stream record JSON");
+        serde_json::from_str(data).expect("parse TTL stream record JSON");
+    // The TTL plan takes its sequence number and created_at from one clock
+    // call, so the pair the worker persisted keeps the stream invariants.
+    let hlc_ms: i64 = seq[..13].parse().expect("sequence number prefix");
+    assert!(
+        *created_at_ms <= hlc_ms,
+        "TTL record created_at_ms {created_at_ms} exceeds sequence prefix {hlc_ms} in {seq}"
+    );
+    assert_eq!(
+        record["dynamodb"]["ApproximateCreationDateTime"].as_i64(),
+        Some(created_at_ms / 1_000),
+        "TTL record timestamp must be created_at_ms / 1000"
+    );
     assert_eq!(record["userIdentity"]["Type"], "Service");
     assert_eq!(
         record["userIdentity"]["PrincipalId"],

From 6bccbedbdf8c0fed10078ed011557affdd0faa44 Mon Sep 17 00:00:00 2001
From: Lee Hannigan <29015211+LeeroyHannigan@users.noreply.github.com>
Date: Thu, 29 Oct 2026 21:00:38 +0000
Subject: [PATCH 07/10] test(cassandra): pin HLC overflow under a forward clamp

Adds the unit test #465 asks for: the last prefix 5 s ahead of the
reading and the re-read a further 5 s back. The prefix moves to the
last prefix plus one, the sequence follows the previous one and
created_at_ms stays at or below the prefix. Flooring the overflow
prefix at the reading plus one fails this test and passes the other
ten. Also reword the branch comment, which compared a millisecond
prefix with sequence numbers.
---
 crates/storage-cassandra/src/stream_util.rs | 41 ++++++++++++++++++++-
 1 file changed, 40 insertions(+), 1 deletion(-)

diff --git a/crates/storage-cassandra/src/stream_util.rs b/crates/storage-cassandra/src/stream_util.rs
index b898ba9ff..09746e9d2 100644
--- a/crates/storage-cassandra/src/stream_util.rs
+++ b/crates/storage-cassandra/src/stream_util.rs
@@ -105,7 +105,8 @@ impl HybridClock {
                 // 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 above every earlier sequence number.
+                // 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;
             }
@@ -410,6 +411,44 @@ mod tests {
         );
     }
 
+    /// 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);

From 4394c0fffcb92b9591bb4bd3ede553533b61d8a7 Mon Sep 17 00:00:00 2001
From: Lee Hannigan <29015211+LeeroyHannigan@users.noreply.github.com>
Date: Thu, 29 Oct 2026 21:00:38 +0000
Subject: [PATCH 08/10] fix(sqlite): stamp an out-of-range record time with the
 current time

format_epoch_millis fell back to 1970-01-01 for a millisecond value the
time crate cannot represent, so such a row would have been swept at the
next retention pass. The PostgreSQL engine stamps the same case with the
current time; the SQLite engine now does the same, and the two engines
agree on the fallback.

A SystemTime reading never reaches this branch: the range ends in the
year 9999. A unit test passes i64::MAX and checks that the result is the
current time in the column's shape, not 1970.
---
 crates/storage-sqlite/src/sqlite_util.rs | 25 +++++++++++++++++++++++-
 1 file changed, 24 insertions(+), 1 deletion(-)

diff --git a/crates/storage-sqlite/src/sqlite_util.rs b/crates/storage-sqlite/src/sqlite_util.rs
index 917f1daaf..903f19019 100644
--- a/crates/storage-sqlite/src/sqlite_util.rs
+++ b/crates/storage-sqlite/src/sqlite_util.rs
@@ -58,9 +58,14 @@ pub(crate) fn format_epoch_seconds(secs: f64) -> String {
 /// precision, 24 characters): `2026-10-15T12:00:00.438Z`. Same width as the
 /// default, so rows written before and after this formatter was introduced
 /// sort together as text in time order.
+///
+/// A value outside the range `time` can represent (beyond the year 9999)
+/// cannot come from a `SystemTime` reading; should one arrive, the row is
+/// stamped with the current time, as the PostgreSQL engine does, so the
+/// retention sweep treats it as a fresh record rather than one from 1970.
 pub(crate) fn format_epoch_millis(ms: i64) -> String {
     let dt = OffsetDateTime::from_unix_timestamp_nanos(i128::from(ms) * 1_000_000)
-        .unwrap_or(OffsetDateTime::UNIX_EPOCH);
+        .unwrap_or_else(|_| OffsetDateTime::now_utc());
     format!(
         "{:04}-{:02}-{:02}T{:02}:{:02}:{:02}.{:03}Z",
         dt.year(),
@@ -133,6 +138,24 @@ mod tests {
         assert!(s.as_str() > cutoff.as_str(), "{s} vs {cutoff}");
     }
 
+    /// A millisecond value `time` cannot represent is stamped with the current
+    /// time in the same shape, as the PostgreSQL engine stamps it, so the row
+    /// is not dated 1970 and swept at the next retention pass.
+    #[test]
+    fn epoch_millis_out_of_range_is_stamped_now() {
+        let before = OffsetDateTime::now_utc();
+        let s = format_epoch_millis(i64::MAX);
+        let after = OffsetDateTime::now_utc();
+        assert_eq!(s.len(), 24);
+        assert!(!s.starts_with("1970-"), "{s}");
+        let lo = format_epoch_millis((before.unix_timestamp_nanos() / 1_000_000) as i64);
+        let hi = format_epoch_millis((after.unix_timestamp_nanos() / 1_000_000) as i64);
+        assert!(
+            lo.as_str() <= s.as_str() && s.as_str() <= hi.as_str(),
+            "{lo} <= {s} <= {hi}"
+        );
+    }
+
     #[test]
     fn sqlite_url_variants() {
         assert_eq!(sqlite_url(":memory:"), "sqlite::memory:");

From a5f4390ca93301d033c64c9d2d672edbe7815cd0 Mon Sep 17 00:00:00 2001
From: Lee Hannigan <29015211+LeeroyHannigan@users.noreply.github.com>
Date: Thu, 29 Oct 2026 21:00:38 +0000
Subject: [PATCH 09/10] docs(postgres): state the record-time fallback in the
 stream engine

The comment said created_at is never a second clock read, while the
fallback for a timestamp outside the time crate's range reads the current
time. The comment now names that case and that the SQLite engine stamps
it the same way.
---
 crates/storage-postgres/src/stream_engine.rs | 7 +++++--
 1 file changed, 5 insertions(+), 2 deletions(-)

diff --git a/crates/storage-postgres/src/stream_engine.rs b/crates/storage-postgres/src/stream_engine.rs
index 8a6b609e4..da7fd122d 100755
--- a/crates/storage-postgres/src/stream_engine.rs
+++ b/crates/storage-postgres/src/stream_engine.rs
@@ -177,8 +177,11 @@ impl StreamEngine for PostgresEngine {
             .await
             .map_err(|e| StorageError::Internal(e.to_string()))?;
 
-            // created_at follows the record's own timestamp, never a second
-            // clock read (NOW() would be the transaction start time).
+            // created_at follows the record's own timestamp rather than a
+            // second clock read (NOW() would be the transaction start time).
+            // A timestamp outside the range `time` can represent, which a
+            // SystemTime reading never is, is stamped with the current time,
+            // as the SQLite engine stamps it.
             let created_at = time::OffsetDateTime::from_unix_timestamp(
                 record.dynamodb.approximate_creation_date_time,
             )

From abca35eb533108c8ace8257c6667a1d6d3e362e6 Mon Sep 17 00:00:00 2001
From: Lee Hannigan <29015211+LeeroyHannigan@users.noreply.github.com>
Date: Tue, 3 Nov 2026 10:06:02 +0000
Subject: [PATCH 10/10] test(storage): update the stream clock test setup for
 the migration runner

#666 removed crate::schema::apply in favour of the migration runner, and the
migrations on main seed no settings rows. After the merge of main:

- The SQLite stream clock test in data/tx_helpers.rs called the removed
  function, so the crate's tests did not compile. It now calls
  crate::migrations::apply_for_tests, as the other SQLite unit tests do.
- The SQLite and PostgreSQL stream clock tests pinned the control-plane
  delay to zero with an UPDATE that matched no row, so CreateTable left the
  SQLite table CREATING and the test failed with TableNotFound. Both now
  write the row, as the other tests on main do.
---
 .../storage-postgres/tests/stream_record_clock.rs   | 11 +++++++----
 crates/storage-sqlite/src/data/tx_helpers.rs        | 13 ++++++++-----
 2 files changed, 15 insertions(+), 9 deletions(-)

diff --git a/crates/storage-postgres/tests/stream_record_clock.rs b/crates/storage-postgres/tests/stream_record_clock.rs
index f3913f9ed..1ea136154 100644
--- a/crates/storage-postgres/tests/stream_record_clock.rs
+++ b/crates/storage-postgres/tests/stream_record_clock.rs
@@ -119,10 +119,13 @@ async fn scratch() -> Scratch {
         .execute(&db)
         .await
         .expect("match the two-database layout");
-    sqlx::query("UPDATE settings SET value = '0' WHERE key = 'control_plane_delay_seconds'")
-        .execute(&db)
-        .await
-        .expect("pin the control-plane delay to zero");
+    sqlx::query(
+        "INSERT INTO settings (key, value) VALUES ('control_plane_delay_seconds', '0') \
+         ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value",
+    )
+    .execute(&db)
+    .await
+    .expect("pin the control-plane delay to zero");
     sqlx::query("INSERT INTO accounts (account_id, account_name) VALUES ($1, $2)")
         .bind(ACCOUNT)
         .bind(format!("acct-{db_name}"))
diff --git a/crates/storage-sqlite/src/data/tx_helpers.rs b/crates/storage-sqlite/src/data/tx_helpers.rs
index 0ff033523..0ee6ba208 100644
--- a/crates/storage-sqlite/src/data/tx_helpers.rs
+++ b/crates/storage-sqlite/src/data/tx_helpers.rs
@@ -404,11 +404,14 @@ mod stream_clock_tests {
         let engine = SqliteEngine::new(":memory:", 1, "us-east-1", 409_600)
             .await
             .expect("engine");
-        crate::schema::apply(&engine.pool).await.expect("schema");
-        sqlx::query("UPDATE settings SET value = '0' WHERE key = 'control_plane_delay_seconds'")
-            .execute(&engine.pool)
-            .await
-            .expect("delay");
+        crate::migrations::apply_for_tests(&engine.pool).await;
+        // The migrations seed no settings, so the row is written, not updated.
+        sqlx::query(
+            "INSERT OR REPLACE INTO settings (key, value) VALUES ('control_plane_delay_seconds', '0')",
+        )
+        .execute(&engine.pool)
+        .await
+        .expect("delay");
         sqlx::query("INSERT INTO accounts (account_id, account_name) VALUES (?, 'default')")
             .bind(ACCOUNT)
             .execute(&engine.pool)