From 4444bc608f5a75d5f0a8e2ebb8dbb2f6488b2489 Mon Sep 17 00:00:00 2001 From: Lee Hannigan <29015211+LeeroyHannigan@users.noreply.github.com> Date: Thu, 29 Oct 2026 23:04:56 +0000 Subject: [PATCH] feat(streams): AT_TIMESTAMP shard iterator on all four backends DynamoDB Streams added the AT_TIMESTAMP shard iterator type on 2026-10-15. GetShardIterator takes ShardIteratorType AT_TIMESTAMP with a Timestamp member (epoch seconds, fractional milliseconds allowed) and returns an iterator positioned at the first record created at or after that time. ExtendDB answered SerializationException for the type. Behaviour, measured on the service (us-east-1, 2026-10-23; raw requests and responses recorded in the issue): - The seek compares at millisecond precision. A Timestamp of 14:04:08.5 skips records written at 14:04:08.050 to .071 and starts at the next group, while the integer 14:04:08 includes them. - A timestamp older than the oldest retained record returns everything with no error; one past the newest record positions after it, so GetRecords returns records written from then on. - Validation order: the type enum, then Timestamp presence or absence for the type, then SequenceNumber presence or absence for the type, then the future check, then stream and shard existence. Every parameter message carries the prefix "One or more parameter values were invalid: ". - The future buffer is about one second: now + 1 s was accepted, now + 2 s refused. A Timestamp sent as a JSON string, boolean or object is a SerializationException; null, zero and negative are the "Valid timestamp must be provided" refusal. Implementation: - ShardIteratorType::AtTimestamp and GetShardIteratorInput.timestamp. - The handler refuses a missing or null type with the service's "Member must not be null" text, pre-validates the type enum (the service's member order, rendered without the offending value as the service does), refuses a Timestamp whose JSON type is not a number with the service's deserializer texts, then runs the parameter rules in streams_iterator_params.rs. The same module now owns the AT_SEQUENCE_NUMBER to AFTER conversion. - StreamEngine::first_sequence_number_at_or_after(shard_id, at_ms) on all four backends, each in its own stream_at_timestamp.rs. SQLite, PostgreSQL and MongoDB take MIN(sequence_number) over created_at >= T. Cassandra seeks the clustering key with the HLC millisecond prefix of T as the lower bound and filters created_at within that partition: created_at_ms <= hlc_ms holds for every record since the one-clock-read change, so created_at_ms >= T implies the sequence is at or after the bound and no qualifying record is skipped, and the filter drops a record written under a clamped clock whose prefix clears the bound while its created_at is before T. MongoDB and SQLite gain a (shard_id, created_at) index for the seek; on MongoDB the existing index walked 50,000 documents (84 ms) for a mid-range seek over a 100,000-record shard. - Resolution: a match positions AT that record (the existing n - 1 conversion); no match positions after the newest record. The tip is read before the seek so a write landing between the two reads is not skipped. - TRIM_HORIZON and LATEST with a SequenceNumber now refuse as the service does (#461), since the same check covers them. Tests: tests/test_streams_at_timestamp.py, eleven tests with every expected value from the measurement, including the at-or-after boundary at a record's own millisecond, passing on SQLite, PostgreSQL, MongoDB and Cassandra; nine of the original ten fail against a server built from the base commit. The existing tests/test_streams.py passes on all four. Unit tests cover the millisecond floor, the validation order and texts, the Timestamp JSON-type texts, the future buffer, the Cassandra bound, and the tip-then-seek order against a double that writes during the tip read (the test fails with the reads swapped). A Cassandra direct test inserts a clamped record and asserts the seek does not return it for a T after its created_at. Docs: a Streams section in differences-from-dynamodb.md, the usage guide example, the streams design doc, and EXTENDDB_STREAM_SINCE in samples/stream_consumer.py. Closes #460. Closes #461. --- crates/core/src/types/stream.rs | 7 + crates/engine/src/lib.rs | 1 + crates/engine/src/streams.rs | 133 ++--- crates/engine/src/streams_iterator_params.rs | 463 ++++++++++++++++++ crates/storage-cassandra/src/lib.rs | 1 + .../src/stream_at_timestamp.rs | 85 ++++ crates/storage-cassandra/src/stream_engine.rs | 17 +- .../storage-cassandra/tests/direct/streams.rs | 108 ++++ crates/storage-mongodb/src/bootstrapper.rs | 14 + crates/storage-mongodb/src/lib.rs | 1 + .../src/stream_at_timestamp.rs | 44 ++ crates/storage-mongodb/src/stream_engine.rs | 12 + crates/storage-postgres/src/lib.rs | 1 + .../src/stream_at_timestamp.rs | 31 ++ crates/storage-postgres/src/stream_engine.rs | 12 + crates/storage-sqlite/src/lib.rs | 1 + crates/storage-sqlite/src/schema.rs | 2 + crates/storage-sqlite/src/stream.rs | 12 + .../storage-sqlite/src/stream_at_timestamp.rs | 33 ++ crates/storage/src/lib.rs | 13 + docs/design/07-component-streams.md | 2 +- docs/differences-from-dynamodb.md | 7 + docs/manuals/03-usage-guide.md | 9 + samples/stream_consumer.py | 22 +- tests/test_streams_at_timestamp.py | 399 +++++++++++++++ 25 files changed, 1367 insertions(+), 63 deletions(-) create mode 100644 crates/engine/src/streams_iterator_params.rs create mode 100644 crates/storage-cassandra/src/stream_at_timestamp.rs create mode 100644 crates/storage-mongodb/src/stream_at_timestamp.rs create mode 100644 crates/storage-postgres/src/stream_at_timestamp.rs create mode 100644 crates/storage-sqlite/src/stream_at_timestamp.rs create mode 100644 tests/test_streams_at_timestamp.py diff --git a/crates/core/src/types/stream.rs b/crates/core/src/types/stream.rs index 870169b9a..41426e7cc 100755 --- a/crates/core/src/types/stream.rs +++ b/crates/core/src/types/stream.rs @@ -27,6 +27,9 @@ pub enum ShardIteratorType { Latest, AtSequenceNumber, AfterSequenceNumber, + /// Start at the first record whose creation time is at or after + /// `GetShardIteratorInput::timestamp`. + AtTimestamp, } /// Stream status. @@ -201,6 +204,10 @@ pub struct GetShardIteratorInput { pub shard_iterator_type: ShardIteratorType, #[serde(rename = "SequenceNumber", default)] pub sequence_number: Option, + /// Epoch seconds, fractional milliseconds allowed. Used with + /// `AT_TIMESTAMP`; the service refuses it with every other type. + #[serde(rename = "Timestamp", default)] + pub timestamp: Option, } /// `GetShardIterator` response. diff --git a/crates/engine/src/lib.rs b/crates/engine/src/lib.rs index 74083fbfa..46278b5d7 100755 --- a/crates/engine/src/lib.rs +++ b/crates/engine/src/lib.rs @@ -33,6 +33,7 @@ mod scan; mod search_vectors; pub mod stream_capture; mod streams; +mod streams_iterator_params; mod tagging; mod transact_get_items; mod transact_write_helpers; diff --git a/crates/engine/src/streams.rs b/crates/engine/src/streams.rs index c8ba419bc..5ba804d35 100755 --- a/crates/engine/src/streams.rs +++ b/crates/engine/src/streams.rs @@ -14,6 +14,7 @@ use serde_json::Value; use crate::OperationContext; use crate::serialize_output; +use crate::streams_iterator_params as iterator_params; /// Handle `DescribeStream`. /// @@ -75,6 +76,11 @@ pub async fn handle_list_streams( /// /// Encodes the shard ID and starting position into a base64 iterator token. /// +/// Validation runs in the service's order, measured 2026-10-23 (us-east-1): +/// the `ShardIteratorType` enum, then `Timestamp` presence or absence for the +/// type, then `SequenceNumber` presence or absence for the type, then the +/// future check, then stream and shard existence. +/// /// # Errors /// /// Returns [`DynamoDbError::ResourceNotFoundException`] if the stream/shard does not exist. @@ -83,8 +89,36 @@ pub async fn handle_get_shard_iterator( body: Value, ctx: &OperationContext, ) -> Result { - let input: GetShardIteratorInput = serde_json::from_value(body) - .map_err(|e| DynamoDbError::SerializationException(e.to_string()))?; + // The service renders both ShardIteratorType violations without the + // offending value (measured 2026-10-23): a missing member is "Member must + // not be null", a bad member lists the set in this order. + let type_value = body.get("ShardIteratorType"); + if type_value.is_none_or(serde_json::Value::is_null) { + return Err(DynamoDbError::ValidationException( + "1 validation error detected: Value at 'shardIteratorType' failed to satisfy \ + constraint: Member must not be null" + .to_owned(), + )); + } + crate::validate_enum_fields( + &body, + &[crate::EnumField { + json_name: "ShardIteratorType", + valid: &[ + "AFTER_SEQUENCE_NUMBER", + "LATEST", + "AT_TIMESTAMP", + "AT_SEQUENCE_NUMBER", + "TRIM_HORIZON", + ], + clause: crate::EnumClause::Bare(iterator_params::SHARD_ITERATOR_TYPE_ENUM_CLAUSE), + }], + )?; + iterator_params::check_timestamp_shape(&body)?; + let input: GetShardIteratorInput = + serde_json::from_value(body).map_err(crate::deserialize_error)?; + + let at_timestamp_ms = iterator_params::validate_iterator_parameters(&input)?; // Validate the stream and shard exist before issuing an iterator. ctx.storage @@ -92,6 +126,7 @@ pub async fn handle_get_shard_iterator( .await .map_err(storage_to_dynamo)?; + let width = ctx.storage.sequence_number_width(); let seq = match input.shard_iterator_type { ShardIteratorType::TrimHorizon => String::new(), ShardIteratorType::Latest => { @@ -104,33 +139,18 @@ pub async fn handle_get_shard_iterator( .unwrap_or_default() } ShardIteratorType::AtSequenceNumber => { - // Convert to AFTER_SEQUENCE_NUMBER by subtracting 1, so the - // exclusive "after" semantics produce inclusive "at" behavior. - let raw = input.sequence_number.clone().ok_or_else(|| { - DynamoDbError::ValidationException( - "SequenceNumber is required for AT_SEQUENCE_NUMBER iterator type".to_owned(), - ) - })?; - let n = raw.parse::().map_err(|_| { - DynamoDbError::ValidationException("Invalid SequenceNumber".to_owned()) - })?; - // n == 0: sequence 0 is the first possible record, so "at 0" - // means "read from the beginning" — same as TRIM_HORIZON. - if n > 0 { - // Pad to the backend's stored width so lexicographic order - // matches numeric order against stored sequence numbers. - let width = ctx.storage.sequence_number_width(); - format!("{:0>width$}", n - 1) - } else { - String::new() - } + let raw = input.sequence_number.as_deref().unwrap_or_default(); + iterator_params::at_sequence_to_after(raw, width)? } - ShardIteratorType::AfterSequenceNumber => { - input.sequence_number.clone().ok_or_else(|| { - DynamoDbError::ValidationException( - "SequenceNumber is required for AFTER_SEQUENCE_NUMBER iterator type".to_owned(), - ) - })? + ShardIteratorType::AfterSequenceNumber => input.sequence_number.clone().unwrap_or_default(), + ShardIteratorType::AtTimestamp => { + resolve_at_timestamp( + ctx.storage.as_ref(), + &input.shard_id, + at_timestamp_ms.unwrap_or_default(), + width, + ) + .await? } }; @@ -140,8 +160,8 @@ pub async fn handle_get_shard_iterator( let type_str = "AFTER_SEQUENCE_NUMBER"; // Encode creation timestamp (seconds since epoch) for 15-minute expiration. - // unwrap_or_default: returns epoch 0 if system clock is before 1970 — safe - // because the iterator would just expire immediately on the next GetRecords. + // unwrap_or_default: returns epoch 0 if the system clock is before 1970, + // which only makes the iterator expire on the next GetRecords. let created_at = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .unwrap_or_default() @@ -161,6 +181,33 @@ pub async fn handle_get_shard_iterator( serialize_output(&output) } +/// The AFTER position for an `AT_TIMESTAMP` request: at the first record +/// created at or after `at_ms`, or after the newest record when none +/// qualifies (the service's tip-plus-one resolution). +/// +/// The tip is read before the seek. If the seek then finds nothing, the +/// iterator starts after that tip, and a record landing between the two +/// reads is still ahead of it. The other order would skip such a record. +pub(crate) async fn resolve_at_timestamp( + storage: &dyn extenddb_storage::StreamEngine, + shard_id: &str, + at_ms: i64, + width: usize, +) -> Result { + let tip = storage + .latest_sequence_number(shard_id) + .await + .map_err(storage_to_dynamo)?; + let first = storage + .first_sequence_number_at_or_after(shard_id, at_ms) + .await + .map_err(storage_to_dynamo)?; + match first { + Some(first) => iterator_params::at_sequence_to_after(&first, width), + None => Ok(tip.unwrap_or_default()), + } +} + /// Shard iterator expiration: 15 minutes (900 seconds), matching real `DynamoDB`. const SHARD_ITERATOR_EXPIRY_SECS: u64 = 900; @@ -419,32 +466,6 @@ fn storage_to_dynamo(e: StorageError) -> DynamoDbError { #[cfg(test)] mod tests { - /// AT_SEQUENCE_NUMBER converts to AFTER by subtracting 1 and padding to - /// the backend's stored width. Verify that an unpadded client input is - /// normalised correctly for both the 21-digit (postgres/sqlite/mongodb) - /// and 23-digit (cassandra) cases, and that the edge case n=0 maps to - /// TRIM_HORIZON (empty string). - #[test] - fn at_sequence_number_padding() { - let cases: &[(&str, usize, &str)] = &[ - ("5", 21, "000000000000000000004"), - ("000000000000000000005", 21, "000000000000000000004"), - ("5", 23, "00000000000000000000004"), - ("00000000000000000000005", 23, "00000000000000000000004"), - ("0", 21, ""), - ("1", 21, "000000000000000000000"), - ]; - for (input, width, expected) in cases { - let n: u128 = input.parse().unwrap(); - let result = if n > 0 { - format!("{:0>width$}", n - 1) - } else { - String::new() - }; - assert_eq!(&result, expected, "input={input} width={width}"); - } - } - #[test] fn shard_iterator_round_trip_backward_compat_and_fail_closed() { let stream = "arn:aws:dynamodb:us-east-1:1:table/T/stream/2026-10-20T16:17:40.188"; diff --git a/crates/engine/src/streams_iterator_params.rs b/crates/engine/src/streams_iterator_params.rs new file mode 100644 index 000000000..623199740 --- /dev/null +++ b/crates/engine/src/streams_iterator_params.rs @@ -0,0 +1,463 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! `GetShardIterator` parameter rules and position conversions. +//! +//! Messages and the order of the checks were measured on the service +//! 2026-10-23 (us-east-1); see `tests/test_streams_at_timestamp.py` for the +//! wire-level expectations taken from the same run. + +use extenddb_core::error::DynamoDbError; +use extenddb_core::types::{GetShardIteratorInput, ShardIteratorType}; + +/// How the service renders a `ShardIteratorType` that is not a member: no +/// value, and the set in this order. Measured 2026-10-23 (us-east-1) for an +/// unknown name and for a lowercase one alike. +pub(crate) const SHARD_ITERATOR_TYPE_ENUM_CLAUSE: &str = "Value at 'shardIteratorType' failed to \ + satisfy constraint: Member must satisfy enum value set: [AFTER_SEQUENCE_NUMBER, LATEST, \ + AT_TIMESTAMP, AT_SEQUENCE_NUMBER, TRIM_HORIZON]"; + +/// Refuse a `Timestamp` whose JSON type is not a number with the text the +/// service's deserializer renders for it (measured 2026-10-23): a string is +/// `STRING_VALUE`, a boolean `TRUE_VALUE` or `FALSE_VALUE`, an array or object +/// the structure message. Numbers and null pass through to the parameter +/// rules, which treat null as absent. +pub(crate) fn check_timestamp_shape(body: &serde_json::Value) -> Result<(), DynamoDbError> { + use serde_json::Value; + let Some(v) = body.get("Timestamp") else { + return Ok(()); + }; + let msg = match v { + Value::Null | Value::Number(_) => return Ok(()), + Value::String(_) => "STRING_VALUE cannot be converted to Date", + Value::Bool(true) => "TRUE_VALUE cannot be converted to Date", + Value::Bool(false) => "FALSE_VALUE cannot be converted to Date", + Value::Array(_) | Value::Object(_) => "Start of structure or map found where not expected", + }; + Err(DynamoDbError::SerializationException(msg.to_owned())) +} + +/// The service's prefix on every parameter-value refusal below. +const INVALID_PARAMETER_PREFIX: &str = "One or more parameter values were invalid: "; + +/// A future `Timestamp` is accepted up to this far ahead of the server clock. +/// Measured 2026-10-23 (us-east-1): now + 1 s accepted, now + 2 s refused. +const AT_TIMESTAMP_FUTURE_BUFFER_MS: i64 = 1_000; + +/// Check the per-type parameter rules in the service's order and return the +/// `AT_TIMESTAMP` position in milliseconds when that is the type. +/// +/// Messages and order measured on the service 2026-10-23 (us-east-1). +pub(crate) fn validate_iterator_parameters( + input: &GetShardIteratorInput, +) -> Result, DynamoDbError> { + let refuse = |msg: &str| { + Err(DynamoDbError::ValidationException(format!( + "{INVALID_PARAMETER_PREFIX}{msg}" + ))) + }; + let is_at_timestamp = input.shard_iterator_type == ShardIteratorType::AtTimestamp; + let takes_sequence = matches!( + input.shard_iterator_type, + ShardIteratorType::AtSequenceNumber | ShardIteratorType::AfterSequenceNumber + ); + + // Timestamp presence or absence for the type comes first. + let at_ms = if is_at_timestamp { + match input.timestamp { + Some(t) if t > 0.0 && t.is_finite() => Some(timestamp_to_millis(t)), + _ => { + return refuse( + "Valid timestamp must be provided when the iterator type is AT_TIMESTAMP", + ); + } + } + } else { + if input.timestamp.is_some() { + return refuse("Timestamp should not be passed for this iterator type"); + } + None + }; + + // Then SequenceNumber presence or absence for the type. + if takes_sequence { + if input.sequence_number.is_none() { + return refuse("SequenceNumber cannot be null"); + } + } else if input.sequence_number.is_some() { + return refuse("SequenceNumber should not be passed for this iterator type"); + } + + // Then the future check. + if let Some(at_ms) = at_ms { + let now_ms = i64::try_from( + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_millis(), + ) + .unwrap_or(i64::MAX); + if at_ms > now_ms.saturating_add(AT_TIMESTAMP_FUTURE_BUFFER_MS) { + return refuse("Timestamp parameter cannot be greater than the current timestamp"); + } + } + Ok(at_ms) +} + +/// Epoch seconds with an optional fraction to whole milliseconds, floored. +/// Saturates at the `i64` edge; the caller has already refused values that +/// are not positive and finite. +pub(crate) fn timestamp_to_millis(secs: f64) -> i64 { + let ms = (secs * 1000.0).floor(); + if ms >= 9.2e18 { + i64::MAX + } else { + // Non-negative and in range after the guard above. + #[allow(clippy::cast_possible_truncation)] + let v = ms as i64; + v + } +} + +/// Convert an AT_SEQUENCE_NUMBER position to the AFTER position the token +/// carries: `n - 1`, zero-padded to the backend's stored width so lexicographic +/// order matches numeric order. `n == 0` is the first possible record, so +/// "at 0" reads from the beginning, the same as TRIM_HORIZON. +pub(crate) fn at_sequence_to_after(raw: &str, width: usize) -> Result { + let n = raw + .parse::() + .map_err(|_| DynamoDbError::ValidationException("Invalid SequenceNumber".to_owned()))?; + Ok(if n > 0 { + format!("{:0>width$}", n - 1) + } else { + String::new() + }) +} + +#[cfg(test)] +mod tests { + use extenddb_core::error::DynamoDbError; + use extenddb_core::types::{GetShardIteratorInput, ShardIteratorType}; + + /// AT_SEQUENCE_NUMBER converts to AFTER by subtracting 1 and padding to + /// the backend's stored width. Verify that an unpadded client input is + /// normalised correctly for both the 21-digit (postgres/sqlite/mongodb) + /// and 23-digit (cassandra) cases, and that the edge case n=0 maps to + /// TRIM_HORIZON (empty string). + #[test] + fn at_sequence_number_padding() { + let cases: &[(&str, usize, &str)] = &[ + ("5", 21, "000000000000000000004"), + ("000000000000000000005", 21, "000000000000000000004"), + ("5", 23, "00000000000000000000004"), + ("00000000000000000000005", 23, "00000000000000000000004"), + ("0", 21, ""), + ("1", 21, "000000000000000000000"), + ]; + for (input, width, expected) in cases { + let result = super::at_sequence_to_after(input, *width).unwrap(); + assert_eq!(&result, expected, "input={input} width={width}"); + } + assert!(super::at_sequence_to_after("abc", 21).is_err()); + } + + /// Epoch seconds with a fraction floor to whole milliseconds, as the + /// service seeks at millisecond precision (measured: a `Timestamp` of + /// `.5` skipped records written at `.050`). + #[test] + fn timestamp_to_millis_floors_to_the_millisecond() { + assert_eq!( + super::timestamp_to_millis(1_792_764_248.0), + 1_792_764_248_000 + ); + assert_eq!( + super::timestamp_to_millis(1_792_764_248.05), + 1_792_764_248_050 + ); + assert_eq!( + super::timestamp_to_millis(1_792_764_248.999), + 1_792_764_248_999 + ); + assert_eq!( + super::timestamp_to_millis(1_792_764_248.123_456), + 1_792_764_248_123 + ); + // Floors, never rounds: .9996 ms stays at 123. + assert_eq!( + super::timestamp_to_millis(1_792_764_248.123_999_6), + 1_792_764_248_123 + ); + assert_eq!(super::timestamp_to_millis(1e18), i64::MAX); + } + + fn input(t: ShardIteratorType, seq: Option<&str>, ts: Option) -> GetShardIteratorInput { + GetShardIteratorInput { + stream_arn: + "arn:aws:dynamodb:us-east-1:123456789012:table/T/stream/2026-10-23T14:03:56.871" + .to_owned(), + shard_id: "shardId-00000001792764241200-b719fef1".to_owned(), + shard_iterator_type: t, + sequence_number: seq.map(str::to_owned), + timestamp: ts, + } + } + + fn message(r: Result, DynamoDbError>) -> String { + match r { + Err(DynamoDbError::ValidationException(m)) => m, + other => panic!("expected ValidationException, got {other:?}"), + } + } + + /// Messages, their prefix, and the order of the checks, as measured on + /// the service 2026-10-23 (us-east-1). + #[test] + fn parameter_validation_matches_the_service_order_and_text() { + use ShardIteratorType::{ + AfterSequenceNumber, AtSequenceNumber, AtTimestamp, Latest, TrimHorizon, + }; + const P: &str = "One or more parameter values were invalid: "; + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_secs_f64(); + let past = 1_792_764_148.0; + let v = super::validate_iterator_parameters; + + // AT_TIMESTAMP needs a positive Timestamp; null, zero and negative all refuse. + for ts in [None, Some(0.0), Some(-1.0), Some(f64::NAN)] { + assert_eq!( + message(v(&input(AtTimestamp, None, ts))), + format!( + "{P}Valid timestamp must be provided when the iterator type is AT_TIMESTAMP" + ), + "ts={ts:?}" + ); + } + // The Timestamp check runs before the SequenceNumber check. + assert_eq!( + message(v(&input(AtTimestamp, Some("1"), None))), + format!("{P}Valid timestamp must be provided when the iterator type is AT_TIMESTAMP") + ); + // SequenceNumber with AT_TIMESTAMP refuses, and does so before the future check. + for ts in [past, now + 3_600.0] { + assert_eq!( + message(v(&input(AtTimestamp, Some("1"), Some(ts)))), + format!("{P}SequenceNumber should not be passed for this iterator type") + ); + } + // Timestamp with any other type refuses, before the SequenceNumber check. + for t in [TrimHorizon, Latest, AtSequenceNumber, AfterSequenceNumber] { + for seq in [None, Some("1")] { + assert_eq!( + message(v(&input(t, seq, Some(past)))), + format!("{P}Timestamp should not be passed for this iterator type"), + "type={t:?} seq={seq:?}" + ); + } + } + // SequenceNumber rules for the four other types. + for t in [TrimHorizon, Latest] { + assert_eq!( + message(v(&input(t, Some("1"), None))), + format!("{P}SequenceNumber should not be passed for this iterator type") + ); + assert_eq!(v(&input(t, None, None)).unwrap(), None); + } + for t in [AtSequenceNumber, AfterSequenceNumber] { + assert_eq!( + message(v(&input(t, None, None))), + format!("{P}SequenceNumber cannot be null") + ); + assert_eq!(v(&input(t, Some("1"), None)).unwrap(), None); + } + // A far-future Timestamp refuses; a past one resolves to milliseconds. + assert_eq!( + message(v(&input(AtTimestamp, None, Some(now + 3_600.0)))), + format!("{P}Timestamp parameter cannot be greater than the current timestamp") + ); + assert_eq!( + v(&input(AtTimestamp, None, Some(past))).unwrap(), + Some(1_792_764_148_000) + ); + } + + /// Malformed `Timestamp` JSON types render the service's deserializer + /// texts (measured); numbers and null pass to the parameter rules. + #[test] + fn timestamp_shape_texts_match_the_service() { + use serde_json::json; + let msg = + |v: serde_json::Value| match super::check_timestamp_shape(&json!({"Timestamp": v})) { + Err(DynamoDbError::SerializationException(m)) => m, + other => panic!("expected SerializationException, got {other:?}"), + }; + assert_eq!( + msg(json!("1792764248.5")), + "STRING_VALUE cannot be converted to Date" + ); + assert_eq!(msg(json!("")), "STRING_VALUE cannot be converted to Date"); + assert_eq!(msg(json!(true)), "TRUE_VALUE cannot be converted to Date"); + assert_eq!(msg(json!(false)), "FALSE_VALUE cannot be converted to Date"); + assert_eq!( + msg(json!({"S": "x"})), + "Start of structure or map found where not expected" + ); + assert_eq!( + msg(json!([1])), + "Start of structure or map found where not expected" + ); + assert!(super::check_timestamp_shape(&json!({"Timestamp": 1.5})).is_ok()); + assert!(super::check_timestamp_shape(&json!({"Timestamp": null})).is_ok()); + assert!(super::check_timestamp_shape(&json!({})).is_ok()); + } + + /// The future buffer is about one second: now + 0.5 s is accepted, now + + /// 2 s is refused (measured; the 1 s to 2 s band depends on clock skew). + #[test] + fn future_buffer_admits_half_a_second_and_refuses_two() { + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_secs_f64(); + let v = super::validate_iterator_parameters; + assert!( + v(&input( + ShardIteratorType::AtTimestamp, + None, + Some(now + 0.5) + )) + .is_ok() + ); + assert!( + v(&input( + ShardIteratorType::AtTimestamp, + None, + Some(now + 2.0) + )) + .is_err() + ); + } + + /// A write that lands while the tip is being read is still ahead of the + /// resolved position. The double performs that write inside + /// `latest_sequence_number` and reports the tip including it. Tip first: + /// the seek then finds the new record and the iterator positions AT it. + /// Seek first: the seek finds nothing, the tip read then includes the new + /// record, and the iterator starts after it, skipping it. + #[tokio::test] + async fn at_timestamp_positions_before_a_write_that_lands_between_the_two_reads() { + use std::sync::Mutex; + + use extenddb_storage::BoxedFuture as BoxFuture; + use extenddb_storage::error::StorageError; + use extenddb_storage::{StreamEngine, StreamListResult, StreamRecordsResult}; + + struct RacingShard { + /// Sequence numbers in the shard, with their creation ms. + records: Mutex>, + /// The record the racing writer adds during the tip read. + late: (String, i64), + } + impl StreamEngine for RacingShard { + fn write_stream_record( + &self, + _: &str, + _: &extenddb_core::types::StreamRecord, + _: &str, + _: &str, + ) -> BoxFuture<'_, Result<(), StorageError>> { + unreachable!() + } + fn get_stream_records( + &self, + _: &str, + _: &str, + _: &str, + _: Option<&str>, + _: i64, + ) -> BoxFuture<'_, StreamRecordsResult> { + unreachable!() + } + fn describe_stream( + &self, + _: &str, + _: &extenddb_core::types::DescribeStreamInput, + ) -> BoxFuture<'_, Result> + { + unreachable!() + } + fn list_streams( + &self, + _: &str, + _: Option<&str>, + _: i64, + _: Option<&str>, + ) -> BoxFuture<'_, StreamListResult> { + unreachable!() + } + fn cleanup_expired_stream_records( + &self, + _: i64, + ) -> BoxFuture<'_, Result> { + unreachable!() + } + fn assign_shard( + &self, + _: &str, + _: &str, + _: &str, + ) -> BoxFuture<'_, Result> { + unreachable!() + } + fn next_sequence_number(&self, _: &str) -> BoxFuture<'_, Result> { + unreachable!() + } + fn validate_shard( + &self, + _: &str, + _: &str, + _: &str, + ) -> BoxFuture<'_, Result<(), StorageError>> { + Box::pin(async { Ok(()) }) + } + fn latest_sequence_number( + &self, + _: &str, + ) -> BoxFuture<'_, Result, StorageError>> { + Box::pin(async move { + let mut recs = self.records.lock().unwrap(); + // The racing write lands while this read runs; the store + // reports the tip as it stands, which includes it. + recs.push(self.late.clone()); + Ok(recs.last().map(|(s, _)| s.clone())) + }) + } + fn first_sequence_number_at_or_after( + &self, + _: &str, + at_ms: i64, + ) -> BoxFuture<'_, Result, StorageError>> { + Box::pin(async move { + let recs = self.records.lock().unwrap(); + Ok(recs + .iter() + .filter(|(_, ms)| *ms >= at_ms) + .map(|(s, _)| s.clone()) + .min()) + }) + } + } + + let shard = RacingShard { + records: Mutex::new(vec![("000000000000000000005".to_owned(), 1_000)]), + late: ("000000000000000000006".to_owned(), 2_500), + }; + // T is after the only stored record and before the racing write. + let pos = crate::streams::resolve_at_timestamp(&shard, "shardId-x", 2_000, 21) + .await + .unwrap(); + // AT record 6 is encoded as AFTER record 5: GetRecords returns 6 first. + assert_eq!(pos, "000000000000000000005"); + } +} diff --git a/crates/storage-cassandra/src/lib.rs b/crates/storage-cassandra/src/lib.rs index 7dbfe98c4..c38891fa5 100644 --- a/crates/storage-cassandra/src/lib.rs +++ b/crates/storage-cassandra/src/lib.rs @@ -28,6 +28,7 @@ pub mod migrations; pub mod operations; pub(crate) mod propagation_hold; mod snapshot_engine; +pub mod stream_at_timestamp; mod stream_engine; pub mod stream_util; pub mod table_engine; diff --git a/crates/storage-cassandra/src/stream_at_timestamp.rs b/crates/storage-cassandra/src/stream_at_timestamp.rs new file mode 100644 index 000000000..5654744e3 --- /dev/null +++ b/crates/storage-cassandra/src/stream_at_timestamp.rs @@ -0,0 +1,85 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! `AT_TIMESTAMP` seek: the first stream record in a shard created at or +//! after a point in time. +//! +//! `created_at` is not a clustering column, so the seek uses the sequence +//! number's millisecond prefix as a clustering-key lower bound and then +//! filters `created_at` within that single partition. For every record +//! `created_at_ms <= hlc_ms`: the `HybridClock` clamps its physical +//! component upward and never below the wall-clock read that `created_at` +//! stores. So `created_at_ms >= at_ms` implies `hlc_ms >= at_ms`, and every +//! qualifying record sits at or after the bound; the seek never skips one. +//! The filter removes the other case: a record written under a clamped clock +//! (after a backwards step) whose prefix clears the bound while its +//! `created_at` is before `at_ms`. The filter walks at most that clamp +//! window, which is zero under a monotonic clock. + +use cdrs_tokio::types::IntoRustByName; +use extenddb_storage::error::StorageError; + +use crate::engine::CassandraEngine; + +/// The smallest sequence number whose millisecond prefix is `at_ms`. +#[must_use] +pub fn sequence_lower_bound(at_ms: i64) -> String { + format!("{:013}{:06}{:04}", at_ms.max(0), 0, 0) +} + +impl CassandraEngine { + pub(crate) async fn first_sequence_number_at_or_after_impl( + &self, + shard_id: &str, + at_ms: i64, + ) -> Result, StorageError> { + let keyspace = self.account_keyspace_for_shard(shard_id).await?; + let bound = sequence_lower_bound(at_ms); + // A single-partition clustering range with one non-key filter; the + // rows come back in clustering (sequence) order, so the first is the + // minimum qualifying sequence. + let query = format!( + "SELECT sequence_number FROM {keyspace}.stream_records \ + WHERE shard_id = ? AND sequence_number >= ? AND created_at >= ? \ + LIMIT 1 ALLOW FILTERING" + ); + let result = self + .session + .query_with_values( + &query, + cdrs_tokio::query_values!(shard_id, bound.as_str(), at_ms), + ) + .await + .map_err(|e| StorageError::Internal(e.to_string()))?; + let rows = result + .response_body() + .map_err(|e| StorageError::Internal(e.to_string()))? + .into_rows() + .unwrap_or_default(); + Ok(rows + .into_iter() + .next() + .and_then(|row| row.get_r_by_name("sequence_number").ok())) + } +} + +#[cfg(test)] +mod tests { + use super::sequence_lower_bound; + + #[test] + fn lower_bound_is_the_first_sequence_in_that_millisecond() { + let b = sequence_lower_bound(1_792_764_248_050); + assert_eq!(b, "17927642480500000000000"); + assert_eq!(b.len(), 23); + // Any real sequence in that millisecond sorts at or after the bound, + // and every sequence from the previous millisecond sorts before it. + assert!("17927642480500000010042" > b.as_str()); + assert!("17927642480499999990042" < b.as_str()); + } + + #[test] + fn negative_timestamps_clamp_to_the_epoch() { + assert_eq!(sequence_lower_bound(-5), "00000000000000000000000"); + } +} diff --git a/crates/storage-cassandra/src/stream_engine.rs b/crates/storage-cassandra/src/stream_engine.rs index 8c83b0927..62bbea71b 100755 --- a/crates/storage-cassandra/src/stream_engine.rs +++ b/crates/storage-cassandra/src/stream_engine.rs @@ -24,7 +24,10 @@ impl CassandraEngine { /// (embedded in the shard ID) via the secondary index on `tables.table_id`. /// /// Shard ID format: `shardId-{table_id}-{index:012}` - async fn account_keyspace_for_shard(&self, shard_id: &str) -> Result { + pub(crate) async fn account_keyspace_for_shard( + &self, + shard_id: &str, + ) -> Result { // Strip "shardId-" prefix and trailing "-{12 digits}" to get table_id. let without_prefix = shard_id.strip_prefix("shardId-").ok_or_else(|| { StorageError::Internal(format!("Invalid shard_id format: {shard_id}")) @@ -595,4 +598,16 @@ impl StreamEngine for CassandraEngine { .map(|row| row.get_r_by_name("sequence_number").unwrap_or_default())) }) } + + fn first_sequence_number_at_or_after( + &self, + shard_id: &str, + at_ms: i64, + ) -> BoxFuture<'_, Result, StorageError>> { + let shard_id = shard_id.to_owned(); + Box::pin(async move { + self.first_sequence_number_at_or_after_impl(&shard_id, at_ms) + .await + }) + } } diff --git a/crates/storage-cassandra/tests/direct/streams.rs b/crates/storage-cassandra/tests/direct/streams.rs index bc0efa206..075d8cf2e 100644 --- a/crates/storage-cassandra/tests/direct/streams.rs +++ b/crates/storage-cassandra/tests/direct/streams.rs @@ -742,3 +742,111 @@ async fn test_stream_record_created_at_matches_the_record_and_sequence() { ); } } + +/// A record written under a clamped clock (after a backwards step) carries a +/// sequence prefix later than its `created_at`. The `AT_TIMESTAMP` seek must +/// not return it for a `T` after its `created_at`, and must still return a +/// record created at or after `T`. +#[tokio::test] +async fn test_at_timestamp_seek_skips_a_clamped_record_older_than_t() { + use extenddb_storage::StreamEngine; + + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::new(&engine, "StreamClampSeek", false).await; + let view = StreamViewType::NewAndOldImages; + + let mut item = BTreeMap::new(); + item.insert("id".to_string(), AttributeValue::S("pk-clamp".to_string())); + item.insert("val".to_string(), AttributeValue::S("real".to_string())); + engine + .put_item( + &table.key_info, + item, + false, + None, + &Default::default(), + Some(&capture(view)), + ) + .await + .unwrap(); + + let shard_id = shard_for("pk-clamp", &table.key_info.table_id); + let keyspace = format!("extenddb_ttl_test_account_{}", table.key_info.account_id); + let rows = engine + .session_arc() + .query_with_values( + &format!( + "SELECT sequence_number, created_at, record_data FROM {keyspace}.stream_records \ + WHERE shard_id = ?" + ), + cdrs_tokio::query_values!(shard_id.as_str()), + ) + .await + .unwrap() + .response_body() + .unwrap() + .into_rows() + .unwrap_or_default(); + assert_eq!(rows.len(), 1); + let real_seq: String = rows[0].get_r_by_name("sequence_number").unwrap(); + let real_created: i64 = rows[0].get_r_by_name("created_at").unwrap(); + let data: String = rows[0].get_r_by_name("record_data").unwrap(); + + // Model the clamped record: sequence prefix 8 s after the real record, + // created_at 8 s before it. T sits between the two: after the clamped + // created_at, before the real record's. + let clamped_hlc = real_created + 8_000; + let clamped_created = real_created - 8_000; + let clamped_seq = format!("{clamped_hlc:013}{:06}{:04}", 0, 9999); + engine + .session_arc() + .query_with_values( + &format!( + "INSERT INTO {keyspace}.stream_records \ + (shard_id, sequence_number, table_id, event_name, record_data, created_at) \ + VALUES (?, ?, ?, 'Insert', ?, ?)" + ), + cdrs_tokio::query_values!( + shard_id.as_str(), + clamped_seq.as_str(), + table.key_info.table_id.as_str(), + data.as_str(), + clamped_created + ), + ) + .await + .unwrap(); + + // T after the clamped created_at and before the real record: the real + // record qualifies, the clamped one does not. + let t = real_created - 1_000; + let got = engine + .first_sequence_number_at_or_after(&shard_id, t) + .await + .unwrap(); + assert_eq!( + got.as_deref(), + Some(real_seq.as_str()), + "the real record is first at or after T" + ); + + // T after every created_at: nothing qualifies, even though the clamped + // sequence prefix clears the bound. + let t = real_created + 1_000; + let got = engine + .first_sequence_number_at_or_after(&shard_id, t) + .await + .unwrap(); + assert_eq!(got, None, "a clamped record older than T is not returned"); + + // T before everything: the clamped record is the first, as its + // created_at is the oldest. + let got = engine + .first_sequence_number_at_or_after(&shard_id, clamped_created - 1) + .await + .unwrap(); + assert!(got.is_some()); +} diff --git a/crates/storage-mongodb/src/bootstrapper.rs b/crates/storage-mongodb/src/bootstrapper.rs index c37c3f36f..4487651e6 100644 --- a/crates/storage-mongodb/src/bootstrapper.rs +++ b/crates/storage-mongodb/src/bootstrapper.rs @@ -183,6 +183,20 @@ impl Bootstrapper for MongoBootstrapper { .await .map_err(|e| OpError::Internal(format!("stream_records shard index: {e}")))?; + // (shard_id, created_at) serves the AT_TIMESTAMP seek, which filters + // one shard by creation time and takes the lowest sequence. Without + // it the (shard_id, sequence_number) index walks every document up to + // the first match: 84 ms over 50,000 documents measured on one shard + // of 100,000. + stream_records + .create_index( + IndexModel::builder() + .keys(doc! { "shard_id": 1, "created_at": 1 }) + .build(), + ) + .await + .map_err(|e| OpError::Internal(format!("stream_records created_at index: {e}")))?; + Ok(()) } diff --git a/crates/storage-mongodb/src/lib.rs b/crates/storage-mongodb/src/lib.rs index b27cc0baf..fef02e4c3 100644 --- a/crates/storage-mongodb/src/lib.rs +++ b/crates/storage-mongodb/src/lib.rs @@ -22,6 +22,7 @@ mod metadata_engine; mod operations; pub mod pushdown; mod snapshot_engine; +mod stream_at_timestamp; mod stream_engine; mod table_engine; mod ttl_worker; diff --git a/crates/storage-mongodb/src/stream_at_timestamp.rs b/crates/storage-mongodb/src/stream_at_timestamp.rs new file mode 100644 index 000000000..964fe0ab1 --- /dev/null +++ b/crates/storage-mongodb/src/stream_at_timestamp.rs @@ -0,0 +1,44 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! `AT_TIMESTAMP` seek: the first stream record in a shard created at or +//! after a point in time. + +use extenddb_storage::error::StorageError; +use futures::TryStreamExt; +use mongodb::bson::{DateTime as BsonDateTime, Document, doc}; +use mongodb::options::FindOptions; + +use crate::MongoEngine; + +impl MongoEngine { + pub(crate) async fn first_sequence_number_at_or_after_impl( + &self, + shard_id: &str, + at_ms: i64, + ) -> Result, StorageError> { + let records_coll = self.data_db.collection::("stream_records"); + let opts = FindOptions::builder() + .sort(doc! { "sequence_number": 1 }) + .limit(1) + .projection(doc! { "sequence_number": 1 }) + .build(); + let cursor = records_coll + .find(doc! { + "shard_id": shard_id, + "created_at": { "$gte": BsonDateTime::from_millis(at_ms) }, + }) + .with_options(opts) + .await + .map_err(|e| StorageError::Internal(e.to_string()))?; + let docs: Vec = cursor + .try_collect() + .await + .map_err(|e| StorageError::Internal(e.to_string()))?; + Ok(docs.first().and_then(|d| { + d.get_str("sequence_number") + .ok() + .map(std::borrow::ToOwned::to_owned) + })) + } +} diff --git a/crates/storage-mongodb/src/stream_engine.rs b/crates/storage-mongodb/src/stream_engine.rs index 097b3bb2a..85e5392e1 100644 --- a/crates/storage-mongodb/src/stream_engine.rs +++ b/crates/storage-mongodb/src/stream_engine.rs @@ -751,6 +751,18 @@ impl StreamEngine for MongoEngine { })) }) } + + fn first_sequence_number_at_or_after( + &self, + shard_id: &str, + at_ms: i64, + ) -> BoxFuture<'_, Result, StorageError>> { + let shard_id = shard_id.to_owned(); + Box::pin(async move { + self.first_sequence_number_at_or_after_impl(&shard_id, at_ms) + .await + }) + } } #[cfg(test)] diff --git a/crates/storage-postgres/src/lib.rs b/crates/storage-postgres/src/lib.rs index 9603d68ff..a90a5a9a3 100755 --- a/crates/storage-postgres/src/lib.rs +++ b/crates/storage-postgres/src/lib.rs @@ -27,6 +27,7 @@ mod pg_util; mod snapshot_engine; #[doc(hidden)] pub use snapshot_engine::map_fetch_error; +mod stream_at_timestamp; mod stream_engine; #[doc(hidden)] pub mod stream_records_sql; diff --git a/crates/storage-postgres/src/stream_at_timestamp.rs b/crates/storage-postgres/src/stream_at_timestamp.rs new file mode 100644 index 000000000..1f43a12b2 --- /dev/null +++ b/crates/storage-postgres/src/stream_at_timestamp.rs @@ -0,0 +1,31 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! `AT_TIMESTAMP` seek: the first stream record in a shard created at or +//! after a point in time. + +use extenddb_storage::error::StorageError; + +use crate::PostgresEngine; + +impl PostgresEngine { + pub(crate) async fn first_sequence_number_at_or_after_impl( + &self, + shard_id: &str, + at_ms: i64, + ) -> Result, StorageError> { + let bound = time::OffsetDateTime::from_unix_timestamp_nanos(i128::from(at_ms) * 1_000_000) + .map_err(|e| StorageError::Validation(format!("Invalid Timestamp: {e}")))?; + // MIN over no rows is one NULL row, so the column is read as Option. + let row: Option<(Option,)> = sqlx::query_as( + "SELECT MIN(sequence_number) FROM stream_records \ + WHERE shard_id = $1 AND created_at >= $2", + ) + .bind(shard_id) + .bind(bound) + .fetch_optional(&self.data_pool) + .await + .map_err(|e| StorageError::Internal(e.to_string()))?; + Ok(row.and_then(|(s,)| s)) + } +} diff --git a/crates/storage-postgres/src/stream_engine.rs b/crates/storage-postgres/src/stream_engine.rs index da7fd122d..0a546a717 100755 --- a/crates/storage-postgres/src/stream_engine.rs +++ b/crates/storage-postgres/src/stream_engine.rs @@ -616,4 +616,16 @@ impl StreamEngine for PostgresEngine { Ok(row.map(|(s,)| s)) }) } + + fn first_sequence_number_at_or_after( + &self, + shard_id: &str, + at_ms: i64, + ) -> BoxFuture<'_, Result, StorageError>> { + let shard_id = shard_id.to_owned(); + Box::pin(async move { + self.first_sequence_number_at_or_after_impl(&shard_id, at_ms) + .await + }) + } } diff --git a/crates/storage-sqlite/src/lib.rs b/crates/storage-sqlite/src/lib.rs index 7e8085ff9..2ce3b45fe 100644 --- a/crates/storage-sqlite/src/lib.rs +++ b/crates/storage-sqlite/src/lib.rs @@ -41,6 +41,7 @@ mod snapshot_engine; mod sqlite_util; mod store; mod stream; +mod stream_at_timestamp; #[cfg(test)] mod stream_tests; mod table_engine; diff --git a/crates/storage-sqlite/src/schema.rs b/crates/storage-sqlite/src/schema.rs index d04ee5b56..a6c5a974b 100644 --- a/crates/storage-sqlite/src/schema.rs +++ b/crates/storage-sqlite/src/schema.rs @@ -181,6 +181,8 @@ CREATE TABLE IF NOT EXISTS stream_records ( ); CREATE INDEX IF NOT EXISTS idx_stream_records_created ON stream_records (created_at); +-- Serves the AT_TIMESTAMP seek: one shard filtered by creation time. +CREATE INDEX IF NOT EXISTS idx_stream_records_shard_created ON stream_records (shard_id, created_at); -- Admin users. CREATE TABLE IF NOT EXISTS admin_users ( diff --git a/crates/storage-sqlite/src/stream.rs b/crates/storage-sqlite/src/stream.rs index 5b00eabfb..719b78914 100644 --- a/crates/storage-sqlite/src/stream.rs +++ b/crates/storage-sqlite/src/stream.rs @@ -517,4 +517,16 @@ impl StreamEngine for SqliteEngine { Ok(row.map(|(s,)| s)) }) } + + fn first_sequence_number_at_or_after( + &self, + shard_id: &str, + at_ms: i64, + ) -> BoxFuture<'_, Result, StorageError>> { + let shard_id = shard_id.to_owned(); + Box::pin(async move { + self.first_sequence_number_at_or_after_impl(&shard_id, at_ms) + .await + }) + } } diff --git a/crates/storage-sqlite/src/stream_at_timestamp.rs b/crates/storage-sqlite/src/stream_at_timestamp.rs new file mode 100644 index 000000000..04e903249 --- /dev/null +++ b/crates/storage-sqlite/src/stream_at_timestamp.rs @@ -0,0 +1,33 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! `AT_TIMESTAMP` seek: the first stream record in a shard created at or +//! after a point in time. + +use extenddb_storage::error::StorageError; + +use crate::SqliteEngine; +use crate::sqlite_util::format_epoch_millis; + +impl SqliteEngine { + /// `created_at` is stored as fixed-width millisecond text in the column + /// default's shape, so a bound in the same shape compares chronologically. + pub(crate) async fn first_sequence_number_at_or_after_impl( + &self, + shard_id: &str, + at_ms: i64, + ) -> Result, StorageError> { + let bound = format_epoch_millis(at_ms); + // MIN over no rows is one NULL row, so the column is read as Option. + let row: Option<(Option,)> = sqlx::query_as( + "SELECT MIN(sequence_number) FROM stream_records \ + WHERE shard_id = ? AND created_at >= ?", + ) + .bind(shard_id) + .bind(&bound) + .fetch_optional(&self.pool) + .await + .map_err(|e| StorageError::Internal(e.to_string()))?; + Ok(row.and_then(|(s,)| s)) + } +} diff --git a/crates/storage/src/lib.rs b/crates/storage/src/lib.rs index b7aec12ff..529a6da21 100755 --- a/crates/storage/src/lib.rs +++ b/crates/storage/src/lib.rs @@ -689,6 +689,19 @@ pub trait StreamEngine: Send + Sync { &self, shard_id: &str, ) -> BoxFuture<'_, Result, StorageError>>; + + /// Sequence number of the first record in the shard whose creation time + /// is at or after `at_ms` (milliseconds since the Unix epoch), or `None` + /// when no stored record qualifies. + /// + /// Used by `GetShardIterator` with `AT_TIMESTAMP`. The comparison is + /// against the stored `created_at`, which every backend takes from the + /// same clock read as the record's `ApproximateCreationDateTime`. + fn first_sequence_number_at_or_after( + &self, + shard_id: &str, + at_ms: i64, + ) -> BoxFuture<'_, Result, StorageError>>; } /// Background worker operations that require storage access. diff --git a/docs/design/07-component-streams.md b/docs/design/07-component-streams.md index dbff306b5..ac2468d4d 100755 --- a/docs/design/07-component-streams.md +++ b/docs/design/07-component-streams.md @@ -18,7 +18,7 @@ Four operations, served on the same HTTP endpoint as DynamoDB operations: |-----------|-------------| | `DescribeStream` | Return stream ARN, status, shard list with parent/child relationships | | `ListStreams` | List streams, optionally filtered by table name | -| `GetShardIterator` | Get an iterator for a shard (TRIM_HORIZON, LATEST, AT_SEQUENCE_NUMBER, AFTER_SEQUENCE_NUMBER) | +| `GetShardIterator` | Get an iterator for a shard (TRIM_HORIZON, LATEST, AT_SEQUENCE_NUMBER, AFTER_SEQUENCE_NUMBER, AT_TIMESTAMP) | | `GetRecords` | Read up to 1000 records (or 1 MB) from a shard using an iterator | **Note:** In real DynamoDB, Streams is a separate service with its own endpoint (`streams.dynamodb..amazonaws.com`) and its own SDK client (`boto3.client('dynamodbstreams')` in Python, `DynamoDbStreamsClient` in Java). extenddb serves both services on a single port. This works because SDK clients accept `endpoint_url` overrides — the user sets both `AWS_ENDPOINT_URL_DYNAMODB` and `AWS_ENDPOINT_URL_DYNAMODB_STREAMS` to the same extenddb address. The `X-Amz-Target` prefix distinguishes the two services: DynamoDB operations use `DynamoDB_20120810.`, Streams operations use `DynamoDBStreams_20120810.`. Both use `dynamodb` as the SigV4 signing name, so authentication is identical. See `01-requirements.md` §11 and `08-component-config.md` §10 for SDK configuration details. diff --git a/docs/differences-from-dynamodb.md b/docs/differences-from-dynamodb.md index 80c1f671b..68fd37ff7 100755 --- a/docs/differences-from-dynamodb.md +++ b/docs/differences-from-dynamodb.md @@ -49,6 +49,13 @@ adaptation when switching between ExtendDB and the real service. | Table creation delay | Returns `CREATING` immediately; transitions to `ACTIVE` typically within seconds. Same behavior for on-demand and provisioned | Configurable via `control_plane_delay_seconds` runtime setting (default: 5s) | | DeletionProtectionEnabled | Enforced | Enforced (accepted and stored, DeleteTable rejects when enabled) | +## Streams + +| Area | DynamoDB | ExtendDB | +|------|----------|------| +| Shards | Shards open and close over time, with parent and child shards | A fixed number of shards per table (default 4), assigned by partition key hash; shards never close | +| `AT_TIMESTAMP` iterator | The first record created at or after `Timestamp`, compared at millisecond precision; older than retention resolves to the oldest record; past the newest record resolves to after it; more than about a second in the future is a `ValidationException` | The same, on all four backends. The seek compares the record's stored creation time, taken from the same clock read as its `ApproximateCreationDateTime`. The future buffer is one second against the server clock | + ## Time to Live (TTL) | Area | DynamoDB | ExtendDB | diff --git a/docs/manuals/03-usage-guide.md b/docs/manuals/03-usage-guide.md index 861f6bba9..9d8c5e84e 100755 --- a/docs/manuals/03-usage-guide.md +++ b/docs/manuals/03-usage-guide.md @@ -391,6 +391,15 @@ aws dynamodbstreams get-shard-iterator \ --shard-id "shard-0" \ --shard-iterator-type TRIM_HORIZON +# Get a shard iterator positioned at the first record created at or after a +# point in time (epoch seconds, fractional milliseconds allowed) +aws dynamodbstreams get-shard-iterator \ + --endpoint-url https://127.0.0.1:18443 \ + --stream-arn "" \ + --shard-id "shard-0" \ + --shard-iterator-type AT_TIMESTAMP \ + --timestamp 1792764248.050 + # Read records aws dynamodbstreams get-records \ --endpoint-url https://127.0.0.1:18443 \ diff --git a/samples/stream_consumer.py b/samples/stream_consumer.py index b360f82b6..2b254c6ae 100755 --- a/samples/stream_consumer.py +++ b/samples/stream_consumer.py @@ -125,11 +125,23 @@ def refresh_shards(): for shard in resp["StreamDescription"]["Shards"]: sid = shard["ShardId"] if sid not in shard_iterators: - it = streams_client.get_shard_iterator( - StreamArn=stream_arn, - ShardId=sid, - ShardIteratorType="TRIM_HORIZON", - ) + # EXTENDDB_STREAM_SINCE (epoch seconds) starts each shard at + # the first record created at or after that time instead of + # the oldest retained record. + since = os.environ.get("EXTENDDB_STREAM_SINCE") + if since: + it = streams_client.get_shard_iterator( + StreamArn=stream_arn, + ShardId=sid, + ShardIteratorType="AT_TIMESTAMP", + Timestamp=float(since), + ) + else: + it = streams_client.get_shard_iterator( + StreamArn=stream_arn, + ShardId=sid, + ShardIteratorType="TRIM_HORIZON", + ) shard_iterators[sid] = it["ShardIterator"] log("POLLER", f"Tracking shard {sid}") diff --git a/tests/test_streams_at_timestamp.py b/tests/test_streams_at_timestamp.py new file mode 100644 index 000000000..10811b02f --- /dev/null +++ b/tests/test_streams_at_timestamp.py @@ -0,0 +1,399 @@ +# Copyright 2026 ExtendDB contributors +# SPDX-License-Identifier: Apache-2.0 + +"""DynamoDB Streams AT_TIMESTAMP shard iterator. + +Every expected value here was measured on the service (us-east-1, +2026-10-23): positioning at millisecond precision, the four validation +messages and their order, the future buffer, and the error classes for +malformed Timestamp values. + +These tests require a running extenddb server (EXTENDDB_TEST_ENDPOINT must be +set) because the service routes Streams to a separate endpoint; extenddb +serves both on the same endpoint. +""" + +from __future__ import annotations + +import json +import os +import time +import uuid +from datetime import datetime, timezone + +import boto3 +import pytest +from botocore.config import Config +from botocore.exceptions import ClientError +from conftest import wait_for_active, wait_for_deleted + +PREFIX = "One or more parameter values were invalid: " + + +@pytest.fixture(scope="module") +def endpoint_url() -> str: + return os.environ["EXTENDDB_TEST_ENDPOINT"].strip() + + +def _client(service: str, endpoint_url: str, validate: bool = True): + kwargs: dict = dict( + service_name=service, + region_name=os.environ.get("AWS_DEFAULT_REGION", "us-east-1"), + endpoint_url=endpoint_url, + config=Config(parameter_validation=validate, retries={"max_attempts": 1}), + ) + if endpoint_url.startswith("https://"): + kwargs["verify"] = False + return boto3.client(**kwargs) + + +@pytest.fixture(scope="module") +def dynamodb_client(endpoint_url: str): + return _client("dynamodb", endpoint_url) + + +@pytest.fixture(scope="module") +def streams_client(endpoint_url: str): + return _client("dynamodbstreams", endpoint_url) + + +@pytest.fixture(scope="module") +def raw_streams_client(endpoint_url: str): + """Parameter validation off, so malformed Timestamp shapes reach the server.""" + return _client("dynamodbstreams", endpoint_url, validate=False) + + +@pytest.fixture() +def stream_table(dynamodb_client): + table_name = f"extenddb-ts-{uuid.uuid4().hex[:8]}" + dynamodb_client.create_table( + TableName=table_name, + AttributeDefinitions=[{"AttributeName": "pk", "AttributeType": "S"}], + KeySchema=[{"AttributeName": "pk", "KeyType": "HASH"}], + BillingMode="PAY_PER_REQUEST", + StreamSpecification={"StreamEnabled": True, "StreamViewType": "NEW_AND_OLD_IMAGES"}, + ) + wait_for_active(dynamodb_client, table_name) + stream_arn = dynamodb_client.describe_table(TableName=table_name)["Table"]["LatestStreamArn"] + yield table_name, stream_arn + try: + dynamodb_client.delete_table(TableName=table_name) + wait_for_deleted(dynamodb_client, table_name) + except Exception: + pass + + +def _shard_with_records(streams_client, stream_arn: str) -> str: + """The one shard holding a single partition key's records.""" + for shard in streams_client.describe_stream(StreamArn=stream_arn)["StreamDescription"]["Shards"]: + it = streams_client.get_shard_iterator( + StreamArn=stream_arn, ShardId=shard["ShardId"], ShardIteratorType="TRIM_HORIZON" + )["ShardIterator"] + if streams_client.get_records(ShardIterator=it)["Records"]: + return shard["ShardId"] + raise AssertionError("no shard holds records") + + +def _drain(streams_client, iterator: str, rounds: int = 6) -> list[dict]: + out: list[dict] = [] + it = iterator + for _ in range(rounds): + resp = streams_client.get_records(ShardIterator=it) + out.extend(resp.get("Records", [])) + it = resp.get("NextShardIterator") + if not resp.get("Records") or not it: + break + return out + + +def _values(records: list[dict]) -> list[str]: + return [r["dynamodb"]["NewImage"]["v"]["S"] for r in records] + + +def _at(streams_client, stream_arn: str, shard_id: str, ts) -> list[dict]: + it = streams_client.get_shard_iterator( + StreamArn=stream_arn, ShardId=shard_id, ShardIteratorType="AT_TIMESTAMP", Timestamp=ts + )["ShardIterator"] + return _drain(streams_client, it) + + +def _write_group(dynamodb_client, table: str, tag: str, n: int) -> tuple[float, float]: + """Write n items just after a second boundary; return (start, end) wall times.""" + now = time.time() + time.sleep((1.0 - (now % 1.0)) + 0.05) + start = time.time() + for i in range(n): + dynamodb_client.put_item(TableName=table, Item={"pk": {"S": "k"}, "v": {"S": f"{tag}-{i}"}}) + return start, time.time() + + +def _error(fn, **kw) -> tuple[str, str]: + try: + fn(**kw) + except ClientError as e: + err = e.response["Error"] + return err["Code"], err["Message"] + raise AssertionError("expected a ClientError") + + +def _raw_error(endpoint_url: str, body: dict) -> tuple[str, str]: + """POST a GetShardIterator body exactly as given, SigV4-signed with the + test user's key from the environment, for shapes the SDK cannot + serialise (JSON null, an object). Returns (code, message).""" + import ssl + import urllib.request + + from botocore.auth import SigV4Auth + from botocore.awsrequest import AWSRequest + from botocore.credentials import Credentials + + data = json.dumps(body).encode() + req = AWSRequest(method="POST", url=endpoint_url, data=data, headers={ + "Content-Type": "application/x-amz-json-1.0", + "X-Amz-Target": "DynamoDBStreams_20120810.GetShardIterator", + }) + signer_creds = Credentials(os.environ["AWS_ACCESS_KEY_ID"], os.environ["AWS_SECRET_ACCESS_KEY"]) + SigV4Auth(signer_creds, "dynamodb", os.environ.get("AWS_DEFAULT_REGION", "us-east-1")).add_auth(req) + prepped = req.prepare() + ctx = ssl.create_default_context() + ctx.check_hostname = False + ctx.verify_mode = ssl.CERT_NONE + hr = urllib.request.Request(prepped.url, data=data, headers=dict(prepped.headers), method="POST") + try: + with urllib.request.urlopen(hr, timeout=30, context=ctx) as r: + raise AssertionError(f"expected an error, got {r.status}") + except urllib.error.HTTPError as e: + parsed = json.loads(e.read().decode()) + return parsed.get("__type", "").split("#")[-1], parsed.get("message") or parsed.get("Message") or "" + + +class TestPositioning: + """Measured on the service with three groups written in distinct seconds.""" + + def test_seek_positions_at_millisecond_precision(self, dynamodb_client, streams_client, stream_table): + table, arn = stream_table + g1 = _write_group(dynamodb_client, table, "g1", 3) + time.sleep(2.2) + g2 = _write_group(dynamodb_client, table, "g2", 3) + time.sleep(2.2) + g3 = _write_group(dynamodb_client, table, "g3", 2) + shard = _shard_with_records(streams_client, arn) + all_vals = ["g1-0", "g1-1", "g1-2", "g2-0", "g2-1", "g2-2", "g3-0", "g3-1"] + + def ts(secs: float): + return datetime.fromtimestamp(secs, tz=timezone.utc) + + # Before the first record and 25 hours ago both return everything, no error. + assert _values(_at(streams_client, arn, shard, ts(g1[0] - 10))) == all_vals + assert _values(_at(streams_client, arn, shard, ts(time.time() - 25 * 3600))) == all_vals + # Between groups. + assert _values(_at(streams_client, arn, shard, ts(g1[1] + 1.0))) == all_vals[3:] + assert _values(_at(streams_client, arn, shard, ts(g2[1] + 1.0))) == all_vals[6:] + # The group's own second, as an integer, includes the group. + assert _values(_at(streams_client, arn, shard, ts(int(g2[0])))) == all_vals[3:] + assert _values(_at(streams_client, arn, shard, ts(int(g3[0])))) == all_vals[6:] + # Fractional seconds inside the group's second skip it: the writes + # landed at +0.050 to about +0.100, so .5 and .999 start at group 3. + # Guard on the measured span so a slow write on a loaded host skips + # the probe instead of failing it. + if g2[1] < int(g2[0]) + 0.5: + assert _values(_at(streams_client, arn, shard, ts(int(g2[0]) + 0.5))) == all_vals[6:] + assert _values(_at(streams_client, arn, shard, ts(int(g2[0]) + 0.999))) == all_vals[6:] + else: + pytest.skip(f"group 2 spanned past +0.5 s ({g2[1] - int(g2[0]):.3f}); boundary probe not meaningful") + # One second after the last record: nothing. + assert _at(streams_client, arn, shard, ts(int(g3[1]) + 1)) == [] + # ApproximateCreationDateTime is whole seconds. + recs = _at(streams_client, arn, shard, ts(g1[0] - 10)) + for r in recs: + assert r["dynamodb"]["ApproximateCreationDateTime"].microsecond == 0 + + def test_boundary_is_at_or_after_the_record_millisecond(self, dynamodb_client, streams_client, stream_table): + """T equal to a record's own creation millisecond returns that record + (the service's at-or-after); one millisecond later does not.""" + table, arn = stream_table + dynamodb_client.put_item(TableName=table, Item={"pk": {"S": "k"}, "v": {"S": "only"}}) + shard = _shard_with_records(streams_client, arn) + it = streams_client.get_shard_iterator(StreamArn=arn, ShardId=shard, ShardIteratorType="TRIM_HORIZON")["ShardIterator"] + rec = _drain(streams_client, it)[0] + second = int(rec["dynamodb"]["ApproximateCreationDateTime"].timestamp()) + + def returns_it(ms: int) -> bool: + t = datetime.fromtimestamp(second + ms / 1000.0, tz=timezone.utc) + return _values(_at(streams_client, arn, shard, t)) == ["only"] + + # Bisect within the record's second for the last millisecond that + # still returns it: that is the record's own creation millisecond. + assert returns_it(0) + lo, hi = 0, 1000 + while hi - lo > 1: + mid = (lo + hi) // 2 + if returns_it(mid): + lo = mid + else: + hi = mid + assert returns_it(lo), "T at the record's millisecond returns it" + assert not returns_it(lo + 1), "T one millisecond later does not" + + def test_after_tip_returns_only_later_writes(self, dynamodb_client, streams_client, stream_table): + table, arn = stream_table + _write_group(dynamodb_client, table, "old", 2) + shard = _shard_with_records(streams_client, arn) + time.sleep(0.3) + it = streams_client.get_shard_iterator( + StreamArn=arn, + ShardId=shard, + ShardIteratorType="AT_TIMESTAMP", + Timestamp=datetime.fromtimestamp(time.time() - 0.2, tz=timezone.utc), + )["ShardIterator"] + dynamodb_client.put_item(TableName=table, Item={"pk": {"S": "k"}, "v": {"S": "post-tip"}}) + assert _values(_drain(streams_client, it)) == ["post-tip"] + + +class TestTimestampShapes: + def test_integer_and_fractional_numbers_are_accepted(self, dynamodb_client, streams_client, raw_streams_client, stream_table): + table, arn = stream_table + g = _write_group(dynamodb_client, table, "g", 2) + shard = _shard_with_records(streams_client, arn) + for ts in (int(g[0]), round(g[0], 3), g[0] + 0.123456): + resp = raw_streams_client.get_shard_iterator( + StreamArn=arn, ShardId=shard, ShardIteratorType="AT_TIMESTAMP", Timestamp=ts + ) + assert "ShardIterator" in resp, ts + + def test_zero_negative_and_null_refuse(self, dynamodb_client, streams_client, raw_streams_client, stream_table, endpoint_url): + table, arn = stream_table + _write_group(dynamodb_client, table, "g", 1) + shard = _shard_with_records(streams_client, arn) + expected = PREFIX + "Valid timestamp must be provided when the iterator type is AT_TIMESTAMP" + for ts in (0, -1): + code, msg = _error( + raw_streams_client.get_shard_iterator, + StreamArn=arn, ShardId=shard, ShardIteratorType="AT_TIMESTAMP", Timestamp=ts, + ) + assert (code, msg) == ("ValidationException", expected), ts + # JSON null cannot go through the SDK; send it as the service saw it. + code, msg = _raw_error(endpoint_url, { + "StreamArn": arn, "ShardId": shard, "ShardIteratorType": "AT_TIMESTAMP", "Timestamp": None, + }) + assert (code, msg) == ("ValidationException", expected), "null" + + def test_milliseconds_as_seconds_is_in_the_future(self, dynamodb_client, streams_client, stream_table, endpoint_url): + table, arn = stream_table + g = _write_group(dynamodb_client, table, "g", 1) + shard = _shard_with_records(streams_client, arn) + for ts in (int(g[0]) * 1000, 10**18): + code, msg = _raw_error(endpoint_url, { + "StreamArn": arn, "ShardId": shard, "ShardIteratorType": "AT_TIMESTAMP", "Timestamp": ts, + }) + assert code == "ValidationException", ts + assert msg == PREFIX + "Timestamp parameter cannot be greater than the current timestamp", ts + + def test_non_numeric_timestamp_is_a_serialization_error(self, dynamodb_client, streams_client, stream_table, endpoint_url): + table, arn = stream_table + g = _write_group(dynamodb_client, table, "g", 1) + shard = _shard_with_records(streams_client, arn) + cases = [ + (f"{int(g[0])}.5", "STRING_VALUE cannot be converted to Date"), + ("", "STRING_VALUE cannot be converted to Date"), + (True, "TRUE_VALUE cannot be converted to Date"), + ({"S": "x"}, "Start of structure or map found where not expected"), + ] + for ts, expected in cases: + code, msg = _raw_error(endpoint_url, { + "StreamArn": arn, "ShardId": shard, "ShardIteratorType": "AT_TIMESTAMP", "Timestamp": ts, + }) + assert (code, msg) == ("SerializationException", expected), ts + + +class TestValidationMatrix: + """Messages and check order measured on the service.""" + + def test_matrix(self, dynamodb_client, streams_client, raw_streams_client, stream_table): + table, arn = stream_table + g = _write_group(dynamodb_client, table, "g", 1) + shard = _shard_with_records(streams_client, arn) + it = streams_client.get_shard_iterator(StreamArn=arn, ShardId=shard, ShardIteratorType="TRIM_HORIZON")["ShardIterator"] + seq = _drain(streams_client, it)[0]["dynamodb"]["SequenceNumber"] + ts = int(g[0]) + future = int(time.time()) + 3600 + ts_required = "Valid timestamp must be provided when the iterator type is AT_TIMESTAMP" + seq_refused = "SequenceNumber should not be passed for this iterator type" + ts_refused = "Timestamp should not be passed for this iterator type" + future_refused = "Timestamp parameter cannot be greater than the current timestamp" + cases = [ + (dict(ShardIteratorType="AT_TIMESTAMP"), ts_required), + (dict(ShardIteratorType="AT_TIMESTAMP", SequenceNumber=seq), ts_required), + (dict(ShardIteratorType="AT_TIMESTAMP", SequenceNumber=seq, Timestamp=ts), seq_refused), + (dict(ShardIteratorType="AT_TIMESTAMP", SequenceNumber=seq, Timestamp=future), seq_refused), + (dict(ShardIteratorType="AT_TIMESTAMP", SequenceNumber=seq, Timestamp=0), ts_required), + (dict(ShardIteratorType="AT_TIMESTAMP", Timestamp=future), future_refused), + (dict(ShardIteratorType="TRIM_HORIZON", Timestamp=ts), ts_refused), + (dict(ShardIteratorType="LATEST", Timestamp=ts), ts_refused), + (dict(ShardIteratorType="AT_SEQUENCE_NUMBER", SequenceNumber=seq, Timestamp=ts), ts_refused), + (dict(ShardIteratorType="AFTER_SEQUENCE_NUMBER", SequenceNumber=seq, Timestamp=ts), ts_refused), + (dict(ShardIteratorType="TRIM_HORIZON", SequenceNumber=seq, Timestamp=ts), ts_refused), + (dict(ShardIteratorType="TRIM_HORIZON", SequenceNumber=seq), seq_refused), + (dict(ShardIteratorType="LATEST", SequenceNumber=seq), seq_refused), + (dict(ShardIteratorType="AT_SEQUENCE_NUMBER"), "SequenceNumber cannot be null"), + ] + for extra, expected in cases: + code, msg = _error(raw_streams_client.get_shard_iterator, StreamArn=arn, ShardId=shard, **extra) + assert code == "ValidationException", extra + assert msg == PREFIX + expected, extra + + def test_unknown_or_missing_type_renders_without_the_value(self, dynamodb_client, streams_client, raw_streams_client, stream_table, endpoint_url): + table, arn = stream_table + _write_group(dynamodb_client, table, "g", 1) + shard = _shard_with_records(streams_client, arn) + members = "[AFTER_SEQUENCE_NUMBER, LATEST, AT_TIMESTAMP, AT_SEQUENCE_NUMBER, TRIM_HORIZON]" + # The service omits the offending value from the enum violation. + for bad in ("BOGUS", "at_timestamp"): + code, msg = _error(raw_streams_client.get_shard_iterator, StreamArn=arn, ShardId=shard, ShardIteratorType=bad) + assert code == "ValidationException" + assert msg == ( + "1 validation error detected: Value at 'shardIteratorType' failed to satisfy " + f"constraint: Member must satisfy enum value set: {members}" + ) + # A missing or null type is a ValidationException, not a serialization error. + for body in ({"Timestamp": 1792764248.0}, {"ShardIteratorType": None}): + code, msg = _raw_error(endpoint_url, {"StreamArn": arn, "ShardId": shard, **body}) + assert (code, msg) == ( + "ValidationException", + "1 validation error detected: Value at 'shardIteratorType' failed to satisfy " + "constraint: Member must not be null", + ), body + + def test_future_check_runs_before_shard_existence(self, dynamodb_client, streams_client, raw_streams_client, stream_table): + table, arn = stream_table + _write_group(dynamodb_client, table, "g", 1) + shard = _shard_with_records(streams_client, arn) + code, msg = _error( + raw_streams_client.get_shard_iterator, + StreamArn=arn, ShardId="shardId-00000000000000000000-00000000", + ShardIteratorType="AT_TIMESTAMP", Timestamp=int(time.time()) + 3600, + ) + assert (code, msg) == ("ValidationException", PREFIX + "Timestamp parameter cannot be greater than the current timestamp") + code, _ = _error( + raw_streams_client.get_shard_iterator, + StreamArn=arn, ShardId="shardId-00000000000000000000-00000000", + ShardIteratorType="AT_TIMESTAMP", Timestamp=int(time.time()) - 60, + ) + assert code == "ResourceNotFoundException" + assert shard # the real shard still exists + + def test_near_future_is_accepted(self, dynamodb_client, streams_client, stream_table): + table, arn = stream_table + _write_group(dynamodb_client, table, "g", 1) + shard = _shard_with_records(streams_client, arn) + resp = streams_client.get_shard_iterator( + StreamArn=arn, ShardId=shard, ShardIteratorType="AT_TIMESTAMP", + Timestamp=datetime.fromtimestamp(time.time() + 0.5, tz=timezone.utc), + ) + assert "ShardIterator" in resp + code, _ = _error( + streams_client.get_shard_iterator, + StreamArn=arn, ShardId=shard, ShardIteratorType="AT_TIMESTAMP", + Timestamp=datetime.fromtimestamp(time.time() + 5, tz=timezone.utc), + ) + assert code == "ValidationException"