diff --git a/crates/core/src/types/stream.rs b/crates/core/src/types/stream.rs index 870169b9..ad0edad0 100755 --- a/crates/core/src/types/stream.rs +++ b/crates/core/src/types/stream.rs @@ -119,6 +119,14 @@ pub struct StreamDescription { pub stream_arn: String, #[serde(rename = "StreamLabel")] pub stream_label: String, + /// When the stream was requested, epoch seconds. The service returns it + /// on every stream, enabled or disabled. Backends leave it `None`; the + /// engine derives it from `stream_label` at the response boundary. + #[serde( + rename = "CreationRequestDateTime", + skip_serializing_if = "Option::is_none" + )] + pub creation_request_date_time: Option, #[serde(rename = "StreamStatus")] pub stream_status: StreamStatus, #[serde(rename = "StreamViewType")] diff --git a/crates/engine/src/streams.rs b/crates/engine/src/streams.rs index 83b29725..9ea574d9 100755 --- a/crates/engine/src/streams.rs +++ b/crates/engine/src/streams.rs @@ -7,9 +7,10 @@ use extenddb_core::error::DynamoDbError; use extenddb_core::types::{ DescribeStreamInput, DescribeStreamOutput, GetRecordsInput, GetRecordsOutput, GetShardIteratorInput, GetShardIteratorOutput, ListStreamsInput, ListStreamsOutput, - ShardIteratorType, + ShardIteratorType, StreamDescription, }; use extenddb_storage::error::StorageError; +use extenddb_storage::util::stream_label_creation_date_time; use serde_json::Value; use crate::OperationContext; @@ -35,11 +36,19 @@ pub async fn handle_describe_stream( .map_err(storage_to_dynamo)?; let output = DescribeStreamOutput { - stream_description: desc, + stream_description: with_creation_request_date_time(desc), }; serialize_output(&output) } +/// The service returns `CreationRequestDateTime` on every stream, enabled or +/// disabled. The label is the creation instant to the millisecond, so it is +/// derived here for every backend rather than stored by each. +fn with_creation_request_date_time(mut desc: StreamDescription) -> StreamDescription { + desc.creation_request_date_time = stream_label_creation_date_time(&desc.stream_label); + desc +} + /// Handle `ListStreams`. /// /// # Errors @@ -362,6 +371,49 @@ fn storage_to_dynamo(e: StorageError) -> DynamoDbError { #[cfg(test)] mod tests { + /// `DescribeStream` renders `CreationRequestDateTime` as the label's + /// instant in epoch seconds, next to the members it already had, and + /// omits the member only when the label is not a timestamp. + #[test] + fn describe_stream_output_carries_creation_request_date_time() { + use extenddb_core::types::{ + DescribeStreamOutput, StreamDescription, StreamStatus, StreamViewType, + }; + let desc = StreamDescription { + creation_request_date_time: None, + stream_arn: "arn:aws:dynamodb:us-east-1:1:table/T/stream/2026-10-24T11:31:45.748" + .to_owned(), + stream_label: "2026-10-24T11:31:45.748".to_owned(), + stream_status: StreamStatus::Disabled, + stream_view_type: StreamViewType::KeysOnly, + table_name: "T".to_owned(), + key_schema: Vec::new(), + shards: Vec::new(), + last_evaluated_shard_id: None, + }; + let v = crate::serialize_output(&DescribeStreamOutput { + stream_description: super::with_creation_request_date_time(desc.clone()), + }) + .unwrap(); + let sd = &v["StreamDescription"]; + assert_eq!(sd["CreationRequestDateTime"], 1_792_841_505.748); + assert_eq!(sd["StreamLabel"], "2026-10-24T11:31:45.748"); + assert_eq!(sd["StreamStatus"], "DISABLED"); + + let v = crate::serialize_output(&DescribeStreamOutput { + stream_description: super::with_creation_request_date_time(StreamDescription { + stream_label: "not-a-label".to_owned(), + ..desc + }), + }) + .unwrap(); + assert!( + v["StreamDescription"] + .get("CreationRequestDateTime") + .is_none() + ); + } + /// 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) diff --git a/crates/engine/src/update_table.rs b/crates/engine/src/update_table.rs index 6db763f8..14970ac8 100755 --- a/crates/engine/src/update_table.rs +++ b/crates/engine/src/update_table.rs @@ -103,22 +103,12 @@ pub async fn handle_update_table( )); } - // Validate: enabling streams requires a view type. - if let Some(spec) = &input.stream_specification - && spec.stream_enabled - && spec.stream_view_type.is_none() - { - return Err(DynamoDbError::ValidationException( - "StreamViewType must be specified when StreamEnabled is true".to_owned(), - )); - } - // Switching to PROVISIONED requires explicit throughput values. if matches!(input.billing_mode, Some(BillingMode::Provisioned)) && input.provisioned_throughput.is_none() { return Err(DynamoDbError::ValidationException( - "One or more parameter values were invalid: ProvisionedThroughput must be specified when changing BillingMode to PROVISIONED".to_owned(), + "One or more parameter values were invalid: ProvisionedThroughput must be specified when BillingMode is PROVISIONED".to_owned(), )); } @@ -141,6 +131,7 @@ pub async fn handle_update_table( } validate_gsi_updates(&input)?; + validate_stream_specification(&input)?; // A created GSI's projected attributes count toward the table's limit // together with the indexes it already has: the total CreateTable checks, @@ -203,6 +194,25 @@ pub async fn handle_update_table( serialize_output(&output) } +/// Enabling a stream requires a view type. The sentence is the service's +/// (measured 2026-10-24 on a never-streamed table and on a table whose stream +/// is enabled). The service reports a defective index Create and a missing +/// ProvisionedThroughput before this (measured 2026-10-25), and this before +/// the table lookup, so the call sits after the billing and index rules and +/// before the backend call, where it also precedes the already-enabled +/// refusal. The `CreateTable` form has its own sentence. +fn validate_stream_specification(input: &UpdateTableInput) -> Result<(), DynamoDbError> { + if let Some(spec) = &input.stream_specification + && spec.stream_enabled + && spec.stream_view_type.is_none() + { + return Err(DynamoDbError::ValidationException( + "One or more parameter values were invalid: If stream is being enabled then UpdateViewType is required".to_owned(), + )); + } + Ok(()) +} + /// The per-entry rules for `GlobalSecondaryIndexUpdates`, checked before any /// storage read: each entry has exactly one of Create, Update, or Delete, /// and a Create has a valid name, a key schema of defined attributes, and @@ -239,24 +249,47 @@ fn validate_gsi_updates(input: &UpdateTableInput) -> Result<(), DynamoDbError> { "One or more parameter values were invalid: KeySchema must not be empty for GSI creation".to_owned(), )); } - // Validate that all key attributes are defined in AttributeDefinitions - let attr_defs = input.attribute_definitions.as_deref().unwrap_or(&[]); - for ks in &create.key_schema { - if !attr_defs - .iter() - .any(|ad| ad.attribute_name == ks.attribute_name) - { + // Every key attribute must be in AttributeDefinitions. The + // service's sentence depends on the member's shape (measured + // 2026-10-25): absent names the index; present but lacking + // the key lists the keys and the definitions (the CreateTable + // sentence, whose CreateTable copy lives in core's validation + // and is not shared with this path); an empty list has its + // own sentence. + let missing: Vec<&str> = match &input.attribute_definitions { + None => { return Err(DynamoDbError::ValidationException(format!( - "One or more parameter values were invalid: Some index key attributes are not defined in AttributeDefinitions. \ - Keys: [{}], AttributeDefinitions: [{}]", - ks.attribute_name, - attr_defs - .iter() - .map(|ad| ad.attribute_name.as_str()) - .collect::>() - .join(", ") + "One or more parameter values were invalid: AttributeDefinitions is not specified for index: {}", + create.index_name ))); } + Some(defs) if defs.is_empty() => { + return Err(DynamoDbError::ValidationException( + "Invalid KeySchema: Some index key attribute have no definition" + .to_owned(), + )); + } + Some(defs) => create + .key_schema + .iter() + .filter(|ks| !defs.iter().any(|ad| ad.attribute_name == ks.attribute_name)) + .map(|ks| ks.attribute_name.as_str()) + .collect(), + }; + if let Some(first_missing) = missing.first() { + let defined: Vec<&str> = input + .attribute_definitions + .as_deref() + .unwrap_or(&[]) + .iter() + .map(|ad| ad.attribute_name.as_str()) + .collect(); + return Err(DynamoDbError::ValidationException(format!( + "One or more parameter values were invalid: Some index key attributes are not defined in AttributeDefinitions. \ + Keys: [{}], AttributeDefinitions: [{}]", + first_missing, + defined.join(", ") + ))); } } if let Some(delete) = &update.delete { @@ -347,157 +380,5 @@ fn update_table_err_to_dynamo( } #[cfg(test)] -mod tests { - use super::*; - use extenddb_storage::error::StorageError; - - #[test] - fn a_created_gsi_throughput_below_one_is_refused() { - let input = |pt: Value| -> UpdateTableInput { - serde_json::from_value(serde_json::json!({ - "TableName": "t", - "AttributeDefinitions": [{ "AttributeName": "g", "AttributeType": "S" }], - "GlobalSecondaryIndexUpdates": [{ "Create": { - "IndexName": "gix", - "KeySchema": [{ "AttributeName": "g", "KeyType": "HASH" }], - "Projection": { "ProjectionType": "ALL" }, - "ProvisionedThroughput": pt, - }}], - })) - .unwrap() - }; - for (read, write) in [(0, 0), (0, 1), (1, 0), (-1, 5)] { - let pt = serde_json::json!({ "ReadCapacityUnits": read, "WriteCapacityUnits": write }); - match validate_gsi_updates(&input(pt)).unwrap_err() { - DynamoDbError::ValidationException(m) => { - assert!(m.ends_with("for index: gix"), "{m}"); - } - other => panic!("({read}, {write}): {other:?}"), - } - } - let ok = serde_json::json!({ "ReadCapacityUnits": 1, "WriteCapacityUnits": 1 }); - validate_gsi_updates(&input(ok)).unwrap(); - } - - #[test] - fn projected_total_counts_kept_indexes_and_created_ones() { - use extenddb_core::types::{ - GlobalSecondaryIndexUpdate, GsiDescription, LsiDescription, Projection, ProjectionType, - TableDescription, - }; - let include = |n: usize| Projection { - projection_type: ProjectionType::Include, - non_key_attributes: Some((0..n).map(|i| format!("a{i}")).collect()), - }; - let existing = TableDescription { - global_secondary_indexes: Some(vec![ - GsiDescription { - index_name: "kept".to_owned(), - key_schema: Vec::new(), - projection: include(30), - index_status: "ACTIVE".to_owned(), - provisioned_throughput: None, - index_size_bytes: 0, - item_count: 0, - index_arn: String::new(), - }, - GsiDescription { - index_name: "dropped".to_owned(), - key_schema: Vec::new(), - projection: include(50), - index_status: "ACTIVE".to_owned(), - provisioned_throughput: None, - index_size_bytes: 0, - item_count: 0, - index_arn: String::new(), - }, - ]), - local_secondary_indexes: Some(vec![LsiDescription { - index_name: "lsi".to_owned(), - key_schema: Vec::new(), - projection: include(20), - index_size_bytes: 0, - item_count: 0, - index_arn: String::new(), - }]), - ..Default::default() - }; - let create = |n: usize| -> GlobalSecondaryIndexUpdate { - serde_json::from_value(serde_json::json!({ "Create": { - "IndexName": "new", "KeySchema": [], - "Projection": serde_json::to_value(include(n)).unwrap(), - }})) - .unwrap() - }; - let delete: GlobalSecondaryIndexUpdate = - serde_json::from_value(serde_json::json!({ "Delete": { "IndexName": "dropped" } })) - .unwrap(); - - assert_eq!(projected_after_gsi_updates(&existing, &[create(1)]), 101); - assert_eq!( - projected_after_gsi_updates(&existing, &[create(1), delete]), - 51 - ); - assert!( - extenddb_core::validation::validate_projected_attribute_total( - projected_after_gsi_updates(&existing, &[create(1)]) - ) - .is_err() - ); - } - - /// A backend that cannot serve the request reports a 400, not a 500. - /// - /// This arm was missing while CreateTable had it, so the same refusal - /// answered differently depending on which operation carried it. It matters - /// for a backend whose vector capability is decided at runtime rather than at - /// compile time: its refusal arrives through this path. - #[test] - fn an_unsupported_feature_is_a_validation_exception() { - let err = update_table_err_to_dynamo( - StorageError::Unsupported("Vector indexes are not supported here".to_owned()), - "t", - ); - match err { - DynamoDbError::ValidationException(msg) => { - assert_eq!(msg, "Vector indexes are not supported here"); - } - other => panic!("expected ValidationException, got {other:?}"), - } - } - - /// The backend's whole message survives, unwrapped and unprefixed, because - /// the service's own wording for this case names both the table and the index - /// and no layer above the backend knows either. - #[test] - fn a_resource_in_use_refusal_keeps_the_backend_message() { - let measured = extenddb_core::types::vector_index_delete_in_allocation_phase("t", "vidx"); - let err = update_table_err_to_dynamo(StorageError::IndexesInUse(measured.clone()), "t"); - match err { - DynamoDbError::ResourceInUseException(msg) => assert_eq!(msg, measured), - other => panic!("expected ResourceInUseException, got {other:?}"), - } - } - - /// The measured whole string, byte for byte, from probe P2 on 2026-08-19. - #[test] - fn the_allocation_phase_refusal_is_the_measured_whole_string() { - assert_eq!( - extenddb_core::types::vector_index_delete_in_allocation_phase( - "eddbprobe-backfill", - "vidx2" - ), - "Attempt to change a resource which is still in use: Index creation is in resource \ - allocation phase. Retry deletion during backfilling phase or when the index is \ - active. Table: eddbprobe-backfill Index: vidx2" - ); - } - - /// A genuine fault still reports a 500 and is still logged as one. - #[test] - fn an_internal_failure_is_still_an_internal_server_error() { - let err = - update_table_err_to_dynamo(StorageError::Internal("disk on fire".to_owned()), "t"); - assert!(matches!(err, DynamoDbError::InternalServerError(_))); - } -} +#[path = "update_table_tests.rs"] +mod tests; diff --git a/crates/engine/src/update_table_tests.rs b/crates/engine/src/update_table_tests.rs new file mode 100644 index 00000000..aa4cfda3 --- /dev/null +++ b/crates/engine/src/update_table_tests.rs @@ -0,0 +1,238 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Tests for `update_table`. + +use super::*; +use extenddb_storage::error::StorageError; + +/// Enabling a stream without a view type is refused with the service's +/// sentence; a view type, or a disable, passes this check. +#[test] +fn enabling_a_stream_without_a_view_type_is_refused() { + let input = |spec: Value| -> UpdateTableInput { + serde_json::from_value(serde_json::json!({ + "TableName": "t", + "StreamSpecification": spec, + })) + .unwrap() + }; + match validate_stream_specification(&input(serde_json::json!({ "StreamEnabled": true }))) + .unwrap_err() + { + DynamoDbError::ValidationException(m) => assert_eq!( + m, + "One or more parameter values were invalid: If stream is being enabled then UpdateViewType is required" + ), + other => panic!("{other:?}"), + } + validate_stream_specification(&input(serde_json::json!({ + "StreamEnabled": true, "StreamViewType": "NEW_IMAGE" + }))) + .unwrap(); + validate_stream_specification(&input(serde_json::json!({ "StreamEnabled": false }))).unwrap(); +} + +/// A created index whose key attribute is not in AttributeDefinitions is +/// refused with the sentence the service picks by the member's shape +/// (measured 2026-10-25): absent names the index, present but lacking the +/// key lists the key and the definitions, empty has its own sentence. +#[test] +fn a_created_gsi_on_an_undefined_key_attribute_is_refused_by_member_shape() { + let input = |defs: Option| -> UpdateTableInput { + let mut v = serde_json::json!({ + "TableName": "t", + "GlobalSecondaryIndexUpdates": [{ "Create": { + "IndexName": "gix", + "KeySchema": [ + { "AttributeName": "g", "KeyType": "HASH" }, + { "AttributeName": "r", "KeyType": "RANGE" }, + ], + "Projection": { "ProjectionType": "ALL" }, + }}], + }); + if let Some(d) = defs { + v["AttributeDefinitions"] = d; + } + serde_json::from_value(v).unwrap() + }; + let message = |defs: Option| match validate_gsi_updates(&input(defs)).unwrap_err() { + DynamoDbError::ValidationException(m) => m, + other => panic!("{other:?}"), + }; + assert_eq!( + message(None), + "One or more parameter values were invalid: AttributeDefinitions is not specified for index: gix" + ); + assert_eq!( + message(Some(serde_json::json!([]))), + "Invalid KeySchema: Some index key attribute have no definition" + ); + assert_eq!( + message(Some( + serde_json::json!([{ "AttributeName": "g", "AttributeType": "S" }]) + )), + "One or more parameter values were invalid: Some index key attributes are not defined in AttributeDefinitions. Keys: [r], AttributeDefinitions: [g]" + ); + assert_eq!( + message(Some( + serde_json::json!([{ "AttributeName": "other", "AttributeType": "S" }]) + )), + "One or more parameter values were invalid: Some index key attributes are not defined in AttributeDefinitions. Keys: [g], AttributeDefinitions: [other]" + ); + validate_gsi_updates(&input(Some(serde_json::json!([ + { "AttributeName": "g", "AttributeType": "S" }, + { "AttributeName": "r", "AttributeType": "S" }, + ])))) + .unwrap(); +} + +#[test] +fn a_created_gsi_throughput_below_one_is_refused() { + let input = |pt: Value| -> UpdateTableInput { + serde_json::from_value(serde_json::json!({ + "TableName": "t", + "AttributeDefinitions": [{ "AttributeName": "g", "AttributeType": "S" }], + "GlobalSecondaryIndexUpdates": [{ "Create": { + "IndexName": "gix", + "KeySchema": [{ "AttributeName": "g", "KeyType": "HASH" }], + "Projection": { "ProjectionType": "ALL" }, + "ProvisionedThroughput": pt, + }}], + })) + .unwrap() + }; + for (read, write) in [(0, 0), (0, 1), (1, 0), (-1, 5)] { + let pt = serde_json::json!({ "ReadCapacityUnits": read, "WriteCapacityUnits": write }); + match validate_gsi_updates(&input(pt)).unwrap_err() { + DynamoDbError::ValidationException(m) => { + assert!(m.ends_with("for index: gix"), "{m}"); + } + other => panic!("({read}, {write}): {other:?}"), + } + } + let ok = serde_json::json!({ "ReadCapacityUnits": 1, "WriteCapacityUnits": 1 }); + validate_gsi_updates(&input(ok)).unwrap(); +} + +#[test] +fn projected_total_counts_kept_indexes_and_created_ones() { + use extenddb_core::types::{ + GlobalSecondaryIndexUpdate, GsiDescription, LsiDescription, Projection, ProjectionType, + TableDescription, + }; + let include = |n: usize| Projection { + projection_type: ProjectionType::Include, + non_key_attributes: Some((0..n).map(|i| format!("a{i}")).collect()), + }; + let existing = TableDescription { + global_secondary_indexes: Some(vec![ + GsiDescription { + index_name: "kept".to_owned(), + key_schema: Vec::new(), + projection: include(30), + index_status: "ACTIVE".to_owned(), + provisioned_throughput: None, + index_size_bytes: 0, + item_count: 0, + index_arn: String::new(), + }, + GsiDescription { + index_name: "dropped".to_owned(), + key_schema: Vec::new(), + projection: include(50), + index_status: "ACTIVE".to_owned(), + provisioned_throughput: None, + index_size_bytes: 0, + item_count: 0, + index_arn: String::new(), + }, + ]), + local_secondary_indexes: Some(vec![LsiDescription { + index_name: "lsi".to_owned(), + key_schema: Vec::new(), + projection: include(20), + index_size_bytes: 0, + item_count: 0, + index_arn: String::new(), + }]), + ..Default::default() + }; + let create = |n: usize| -> GlobalSecondaryIndexUpdate { + serde_json::from_value(serde_json::json!({ "Create": { + "IndexName": "new", "KeySchema": [], + "Projection": serde_json::to_value(include(n)).unwrap(), + }})) + .unwrap() + }; + let delete: GlobalSecondaryIndexUpdate = + serde_json::from_value(serde_json::json!({ "Delete": { "IndexName": "dropped" } })) + .unwrap(); + + assert_eq!(projected_after_gsi_updates(&existing, &[create(1)]), 101); + assert_eq!( + projected_after_gsi_updates(&existing, &[create(1), delete]), + 51 + ); + assert!( + extenddb_core::validation::validate_projected_attribute_total(projected_after_gsi_updates( + &existing, + &[create(1)] + )) + .is_err() + ); +} + +/// A backend that cannot serve the request reports a 400, not a 500. +/// +/// This arm was missing while CreateTable had it, so the same refusal +/// answered differently depending on which operation carried it. It matters +/// for a backend whose vector capability is decided at runtime rather than at +/// compile time: its refusal arrives through this path. +#[test] +fn an_unsupported_feature_is_a_validation_exception() { + let err = update_table_err_to_dynamo( + StorageError::Unsupported("Vector indexes are not supported here".to_owned()), + "t", + ); + match err { + DynamoDbError::ValidationException(msg) => { + assert_eq!(msg, "Vector indexes are not supported here"); + } + other => panic!("expected ValidationException, got {other:?}"), + } +} + +/// The backend's whole message survives, unwrapped and unprefixed, because +/// the service's own wording for this case names both the table and the index +/// and no layer above the backend knows either. +#[test] +fn a_resource_in_use_refusal_keeps_the_backend_message() { + let measured = extenddb_core::types::vector_index_delete_in_allocation_phase("t", "vidx"); + let err = update_table_err_to_dynamo(StorageError::IndexesInUse(measured.clone()), "t"); + match err { + DynamoDbError::ResourceInUseException(msg) => assert_eq!(msg, measured), + other => panic!("expected ResourceInUseException, got {other:?}"), + } +} + +/// The measured whole string, byte for byte, from probe P2 on 2026-08-19. +#[test] +fn the_allocation_phase_refusal_is_the_measured_whole_string() { + assert_eq!( + extenddb_core::types::vector_index_delete_in_allocation_phase( + "eddbprobe-backfill", + "vidx2" + ), + "Attempt to change a resource which is still in use: Index creation is in resource \ + allocation phase. Retry deletion during backfilling phase or when the index is \ + active. Table: eddbprobe-backfill Index: vidx2" + ); +} + +/// A genuine fault still reports a 500 and is still logged as one. +#[test] +fn an_internal_failure_is_still_an_internal_server_error() { + let err = update_table_err_to_dynamo(StorageError::Internal("disk on fire".to_owned()), "t"); + assert!(matches!(err, DynamoDbError::InternalServerError(_))); +} diff --git a/crates/storage-cassandra/src/stream_engine.rs b/crates/storage-cassandra/src/stream_engine.rs index e3c6088f..60601bab 100755 --- a/crates/storage-cassandra/src/stream_engine.rs +++ b/crates/storage-cassandra/src/stream_engine.rs @@ -345,6 +345,7 @@ impl StreamEngine for CassandraEngine { }; Ok(StreamDescription { + creation_request_date_time: None, stream_arn: stream_arn_str, stream_label, stream_status, diff --git a/crates/storage-mongodb/src/stream_engine.rs b/crates/storage-mongodb/src/stream_engine.rs index e976a667..3075910b 100644 --- a/crates/storage-mongodb/src/stream_engine.rs +++ b/crates/storage-mongodb/src/stream_engine.rs @@ -494,6 +494,7 @@ impl StreamEngine for MongoEngine { }; Ok(StreamDescription { + creation_request_date_time: None, stream_arn: stream_arn_val, stream_label, stream_status, diff --git a/crates/storage-postgres/src/stream_engine.rs b/crates/storage-postgres/src/stream_engine.rs index 6eee0965..574b2b33 100755 --- a/crates/storage-postgres/src/stream_engine.rs +++ b/crates/storage-postgres/src/stream_engine.rs @@ -376,6 +376,7 @@ impl StreamEngine for PostgresEngine { }; Ok(StreamDescription { + creation_request_date_time: None, stream_arn, stream_label, stream_status, diff --git a/crates/storage-sqlite/src/stream.rs b/crates/storage-sqlite/src/stream.rs index 62943f67..49dbc054 100644 --- a/crates/storage-sqlite/src/stream.rs +++ b/crates/storage-sqlite/src/stream.rs @@ -263,6 +263,7 @@ impl StreamEngine for SqliteEngine { }; Ok(StreamDescription { + creation_request_date_time: None, stream_arn, stream_label, stream_status, diff --git a/crates/storage/src/util/arn.rs b/crates/storage/src/util/arn.rs index 083e62c3..459bbee2 100755 --- a/crates/storage/src/util/arn.rs +++ b/crates/storage/src/util/arn.rs @@ -54,6 +54,57 @@ pub fn new_stream_label() -> String { format_stream_label(time::OffsetDateTime::now_utc()) } +/// The instant a [`format_stream_label`] label names, as the epoch-seconds +/// number `DescribeStream` renders in `CreationRequestDateTime` (measured on +/// the service 2026-10-24: present on every stream, enabled or disabled). The +/// label is the stream's creation instant to the millisecond, so no stored +/// column is needed. +/// +/// Two shapes are accepted: the 23-byte `YYYY-MM-DDThh:mm:ss.sss` this module +/// writes, and the 19-byte `YYYY-MM-DDThh:mm:ss` that SQLite, PostgreSQL and +/// MongoDB wrote on v0.1.3 to v0.1.13 and Cassandra up to v0.1.14, which a +/// stream enabled on those versions and never re-enabled still carries. The +/// legacy shape names the instant with a `.000` fraction. `None` for any +/// other shape. +#[must_use] +#[allow(clippy::cast_precision_loss)] // epoch ms fit f64's 53-bit mantissa +pub fn stream_label_creation_date_time(label: &str) -> Option { + let b = label.as_bytes(); + let has_fraction = match b.len() { + 19 => false, + 23 if b[19] == b'.' => true, + _ => return None, + }; + if b[4] != b'-' || b[7] != b'-' || b[10] != b'T' || b[13] != b':' || b[16] != b':' { + return None; + } + let num = |range: std::ops::Range| -> Option { + let s = &label[range]; + if s.bytes().all(|c| c.is_ascii_digit()) { + s.parse().ok() + } else { + None + } + }; + let year = i32::try_from(num(0..4)?).ok()?; + let month = time::Month::try_from(u8::try_from(num(5..7)?).ok()?).ok()?; + let date = time::Date::from_calendar_date(year, month, u8::try_from(num(8..10)?).ok()?).ok()?; + let millis = if has_fraction { + u16::try_from(num(20..23)?).ok()? + } else { + 0 + }; + let clock = time::Time::from_hms_milli( + u8::try_from(num(11..13)?).ok()?, + u8::try_from(num(14..16)?).ok()?, + u8::try_from(num(17..19)?).ok()?, + millis, + ) + .ok()?; + let secs = date.with_time(clock).assume_utc().unix_timestamp(); + Some((secs * 1000 + i64::from(millis)) as f64 / 1000.0) +} + /// Returns an ARN for the specified `DynamoDB` stream. #[must_use] pub fn stream_arn(region: &str, account_id: &str, table_name: &str, stream_label: &str) -> String { @@ -130,4 +181,56 @@ mod stream_label_tests { "2026-01-01T22:04:05.500" ); } + + /// `CreationRequestDateTime` is the label's instant, to the millisecond, + /// and round-trips through the formatter. + #[test] + fn creation_date_time_is_the_label_instant() { + use super::stream_label_creation_date_time; + assert_eq!( + stream_label_creation_date_time("1970-01-01T00:00:00.000"), + Some(0.0) + ); + assert_eq!( + stream_label_creation_date_time("1970-01-01T00:00:01.748"), + Some(1.748) + ); + // 2026-10-24T11:31:45.748Z, the label in the issue. + assert_eq!( + stream_label_creation_date_time("2026-10-24T11:31:45.748"), + Some(1_792_841_505.748) + ); + // The 19-byte legacy label (v0.1.3 to v0.1.13 writers) is the instant + // with a .000 fraction. + assert_eq!( + stream_label_creation_date_time("2026-07-04T10:20:30"), + Some(1_783_160_430.0) + ); + // 2026-01-02T03:04:05.007 UTC, through the formatter. + let label = format_stream_label(at((2026, 1, 2), (3, 4, 5, 7_000), 0)); + assert_eq!( + stream_label_creation_date_time(&label), + Some(1_767_323_045.007) + ); + } + + #[test] + fn creation_date_time_refuses_other_shapes() { + use super::stream_label_creation_date_time; + for bad in [ + "", + "2026-10-24T11:31:4", + "2026-10-24T11:31:4S", + "2026-10-24T11:31:45.7480", + "2026-10-24T11:31:45.748Z", + "2026-10-24 11:31:45.748", + "2026-10-24 11:31:45", + "2026-13-24T11:31:45.748", + "2026-10-24T25:31:45.748", + "2026-1O-24T11:31:45.748", + "2026-10-24T11:31:45:748", + ] { + assert_eq!(stream_label_creation_date_time(bad), None, "{bad:?}"); + } + } } diff --git a/crates/storage/src/util/mod.rs b/crates/storage/src/util/mod.rs index b9c68a0f..ce3c5a49 100755 --- a/crates/storage/src/util/mod.rs +++ b/crates/storage/src/util/mod.rs @@ -13,7 +13,7 @@ mod key; pub use arn::{ ARN_PARTITION, format_stream_label, index_arn, new_stream_label, parse_stream_arn, stream_arn, - table_arn, + stream_label_creation_date_time, table_arn, }; pub use backup_id::{ BackupArnError, BackupArnParts, MAX_BACKUP_ARN_LEN, backup_arn, backup_creation_date_time, diff --git a/tests/python/test_streams.py b/tests/python/test_streams.py index cc3b3d2e..d6fdcea3 100755 --- a/tests/python/test_streams.py +++ b/tests/python/test_streams.py @@ -16,6 +16,7 @@ import os import time import uuid +from datetime import datetime, timezone import boto3 import botocore.config @@ -167,6 +168,48 @@ def test_describe_stream_invalid_arn(self, streams_client): "ValidationException", ) + def test_creation_request_date_time_is_the_label_instant( + self, dynamodb_client, streams_client + ): + """CreationRequestDateTime is present on an enabled and on a disabled + stream and equals the instant the StreamLabel names (UTC, to the + millisecond).""" + name, stream_arn = _create_stream_table(dynamodb_client) + try: + desc = streams_client.describe_stream(StreamArn=stream_arn)["StreamDescription"] + assert desc["StreamStatus"] == "ENABLED" + self._assert_creation_matches_label(desc) + + dynamodb_client.update_table( + TableName=name, StreamSpecification={"StreamEnabled": False} + ) + wait_for_active(dynamodb_client, name) + table = dynamodb_client.describe_table(TableName=name)["Table"] + # The service drops StreamSpecification from DescribeTable once the + # stream is disabled; ExtendDB answers {StreamEnabled: false} until + # every backend does the same. + assert "StreamSpecification" not in table or table["StreamSpecification"] == { + "StreamEnabled": False + } + assert table["LatestStreamArn"] == stream_arn + desc = streams_client.describe_stream(StreamArn=stream_arn)["StreamDescription"] + self._assert_creation_matches_label(desc) + finally: + dynamodb_client.delete_table(TableName=name) + wait_for_deleted(dynamodb_client, name) + + @staticmethod + def _assert_creation_matches_label(desc: dict) -> None: + label = desc["StreamLabel"] + expected = ( + datetime.strptime(label, "%Y-%m-%dT%H:%M:%S.%f") + .replace(tzinfo=timezone.utc) + .timestamp() + ) + created = desc["CreationRequestDateTime"] + assert isinstance(created, datetime), created + assert abs(created.timestamp() - expected) < 0.0005, (label, created) + class TestStreamRecords: """GetShardIterator and GetRecords tests. diff --git a/tests/python/test_update_table_sentences.py b/tests/python/test_update_table_sentences.py new file mode 100644 index 00000000..09d35d90 --- /dev/null +++ b/tests/python/test_update_table_sentences.py @@ -0,0 +1,170 @@ +# Copyright 2026 ExtendDB contributors +# SPDX-License-Identifier: Apache-2.0 + +"""UpdateTable refusals whose sentence the service fixes. + +Each row asserts the code and the whole message as DynamoDB returns them +(measured 2026-10-24 and 2026-10-25). +""" + +from __future__ import annotations + +import pytest +from botocore.exceptions import ClientError + +from helpers import wait_for_active + +STREAM_VIEW_TYPE_REQUIRED = ( + "One or more parameter values were invalid: " + "If stream is being enabled then UpdateViewType is required" +) + + +def _validation_message(exc_info) -> str: + err = exc_info.value.response["Error"] + assert err["Code"] == "ValidationException" + assert exc_info.value.response["ResponseMetadata"]["HTTPStatusCode"] == 400 + return err["Message"] + + +class TestStreamSpecification: + """StreamEnabled without StreamViewType.""" + + def test_never_streamed_table(self, table_factory, dynamodb_client): + name = table_factory() + with pytest.raises(ClientError) as exc_info: + dynamodb_client.update_table( + TableName=name, StreamSpecification={"StreamEnabled": True} + ) + assert _validation_message(exc_info) == STREAM_VIEW_TYPE_REQUIRED + + def test_table_with_an_enabled_stream(self, table_factory, dynamodb_client): + """The view type check comes before the already-enabled check.""" + name = table_factory( + StreamSpecification={ + "StreamEnabled": True, + "StreamViewType": "NEW_AND_OLD_IMAGES", + } + ) + with pytest.raises(ClientError) as exc_info: + dynamodb_client.update_table( + TableName=name, StreamSpecification={"StreamEnabled": True} + ) + assert _validation_message(exc_info) == STREAM_VIEW_TYPE_REQUIRED + + +class TestValidationOrder: + """Which refusal a request with two defects gets (measured 2026-10-25).""" + + def test_index_without_definitions_precedes_the_view_type(self, table_factory, dynamodb_client): + name = table_factory() + with pytest.raises(ClientError) as exc_info: + dynamodb_client.update_table( + TableName=name, + StreamSpecification={"StreamEnabled": True}, + GlobalSecondaryIndexUpdates=[ + { + "Create": { + "IndexName": "by-g", + "KeySchema": [{"AttributeName": "g", "KeyType": "HASH"}], + "Projection": {"ProjectionType": "ALL"}, + } + } + ], + ) + assert _validation_message(exc_info) == ( + "One or more parameter values were invalid: " + "AttributeDefinitions is not specified for index: by-g" + ) + + def test_missing_throughput_precedes_the_view_type(self, table_factory, dynamodb_client): + name = table_factory() + with pytest.raises(ClientError) as exc_info: + dynamodb_client.update_table( + TableName=name, + StreamSpecification={"StreamEnabled": True}, + BillingMode="PROVISIONED", + ) + assert _validation_message(exc_info) == ( + "One or more parameter values were invalid: " + "ProvisionedThroughput must be specified when BillingMode is PROVISIONED" + ) + + +class TestIndexKeyAttributeUndefined: + """A secondary index key attribute that AttributeDefinitions does not define.""" + + @staticmethod + def _create_by_g(): + return [ + { + "Create": { + "IndexName": "by-g", + "KeySchema": [{"AttributeName": "g", "KeyType": "HASH"}], + "Projection": {"ProjectionType": "ALL"}, + } + } + ] + + def test_update_table_without_the_member_names_the_index(self, table_factory, dynamodb_client): + name = table_factory() + with pytest.raises(ClientError) as exc_info: + dynamodb_client.update_table( + TableName=name, + GlobalSecondaryIndexUpdates=self._create_by_g(), + ) + assert _validation_message(exc_info) == ( + "One or more parameter values were invalid: " + "AttributeDefinitions is not specified for index: by-g" + ) + # The refusal left the table as it was. + wait_for_active(dynamodb_client, name) + desc = dynamodb_client.describe_table(TableName=name)["Table"] + assert "GlobalSecondaryIndexes" not in desc + + def test_update_table_with_the_member_lacking_the_key_lists_keys_and_definitions( + self, table_factory, dynamodb_client + ): + name = table_factory() + with pytest.raises(ClientError) as exc_info: + dynamodb_client.update_table( + TableName=name, + AttributeDefinitions=[{"AttributeName": "pk", "AttributeType": "S"}], + GlobalSecondaryIndexUpdates=self._create_by_g(), + ) + assert _validation_message(exc_info) == ( + "One or more parameter values were invalid: " + "Some index key attributes are not defined in AttributeDefinitions. " + "Keys: [g], AttributeDefinitions: [pk]" + ) + + def test_update_table_with_an_empty_member_has_its_own_sentence(self, table_factory, dynamodb_client): + name = table_factory() + with pytest.raises(ClientError) as exc_info: + dynamodb_client.update_table( + TableName=name, + AttributeDefinitions=[], + GlobalSecondaryIndexUpdates=self._create_by_g(), + ) + assert _validation_message(exc_info) == "Invalid KeySchema: Some index key attribute have no definition" + + def test_create_table_lists_keys_and_definitions(self, dynamodb_client): + with pytest.raises(ClientError) as exc_info: + dynamodb_client.create_table( + TableName="never-created-by-g", + AttributeDefinitions=[{"AttributeName": "pk", "AttributeType": "S"}], + KeySchema=[{"AttributeName": "pk", "KeyType": "HASH"}], + BillingMode="PAY_PER_REQUEST", + GlobalSecondaryIndexes=[ + { + "IndexName": "by-g", + "KeySchema": [{"AttributeName": "g", "KeyType": "HASH"}], + "Projection": {"ProjectionType": "ALL"}, + } + ], + ) + assert _validation_message(exc_info) == ( + "One or more parameter values were invalid: " + "Some index key attributes are not defined in AttributeDefinitions. " + "Keys: [g], AttributeDefinitions: [pk]" + )