From d7c94536d67a4c72257dfdc21978a056926a6ca0 Mon Sep 17 00:00:00 2001 From: Lee Hannigan Date: Sun, 25 Oct 2026 12:01:21 +0000 Subject: [PATCH 1/2] DescribeStream returns CreationRequestDateTime The service returns CreationRequestDateTime in StreamDescription on every stream, enabled or disabled (measured 2026-10-24). ExtendDB omitted it on every backend. The value is derived from the stream label rather than stored. stream_label_creation_date_time parses the 23-byte label format_stream_label writes and the 19-byte label SQLite, PostgreSQL and MongoDB wrote on v0.1.3 to v0.1.13 and Cassandra up to v0.1.14, so a stream enabled on those versions reports the member too. The engine sets the member in handle_describe_stream, so all four backends return it. On the service the member is a stored request time that preceded the label's instant by 0 to 13 ms in three measurements (2026-10-25); on ExtendDB the two name the same millisecond. Fixes #517 --- crates/core/src/types/stream.rs | 8 ++ crates/engine/src/streams.rs | 56 +++++++++- crates/storage-cassandra/src/stream_engine.rs | 1 + crates/storage-mongodb/src/stream_engine.rs | 1 + crates/storage-postgres/src/stream_engine.rs | 1 + crates/storage-sqlite/src/stream.rs | 1 + crates/storage/src/util/arn.rs | 103 ++++++++++++++++++ crates/storage/src/util/mod.rs | 2 +- tests/python/test_streams.py | 43 ++++++++ 9 files changed, 213 insertions(+), 3 deletions(-) diff --git a/crates/core/src/types/stream.rs b/crates/core/src/types/stream.rs index 870169b9a..ad0edad0c 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 83b297253..9ea574d95 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/storage-cassandra/src/stream_engine.rs b/crates/storage-cassandra/src/stream_engine.rs index e3c6088fa..60601bab7 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 e976a667a..3075910b1 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 6eee09658..574b2b332 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 62943f678..49dbc0541 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 083e62c35..459bbee21 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 b9c68a0f3..ce3c5a495 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 cc3b3d2e5..d6fdcea3a 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. From 8e3d8f08c8c46232f339bfddc0ae0f69bb480375 Mon Sep 17 00:00:00 2001 From: Lee Hannigan Date: Sun, 25 Oct 2026 12:01:21 +0000 Subject: [PATCH 2/2] UpdateTable: service sentences for a stream without a view type and a GSI without definitions UpdateTable StreamSpecification {StreamEnabled: true} with no StreamViewType is refused by the service with "One or more parameter values were invalid: If stream is being enabled then UpdateViewType is required" (measured 2026-10-24 on a never-streamed table and on a table whose stream is enabled). ExtendDB said "StreamViewType must be specified when StreamEnabled is true". The service reports a Create whose key attribute has no definition, and a PROVISIONED billing mode with no ProvisionedThroughput, before the missing view type, and the view type before the table lookup (measured 2026-10-25); the check now runs after the billing and index rules and before the backend call. The throughput sentence is the service's: "ProvisionedThroughput must be specified when BillingMode is PROVISIONED". An UpdateTable GlobalSecondaryIndexUpdates Create whose key attribute is not defined is refused by the service with one of three sentences by the shape of AttributeDefinitions (measured 2026-10-25): member absent, "AttributeDefinitions is not specified for index: ", naming the first Create in request order; member present and lacking the key, "Some index key attributes are not defined in AttributeDefinitions. Keys: [..], AttributeDefinitions: [..]"; member present and empty, "Invalid KeySchema: Some index key attribute have no definition". ExtendDB rendered the second sentence for every shape. The UpdateTable check has its own rendering in the engine and does not call core's CreateTable validation, so the CreateTable sentence is unchanged. The update_table tests move to update_table_tests.rs unchanged. Fixes #519 Fixes #520 --- crates/engine/src/update_table.rs | 241 +++++--------------- crates/engine/src/update_table_tests.rs | 238 +++++++++++++++++++ tests/python/test_update_table_sentences.py | 170 ++++++++++++++ 3 files changed, 469 insertions(+), 180 deletions(-) create mode 100644 crates/engine/src/update_table_tests.rs create mode 100644 tests/python/test_update_table_sentences.py diff --git a/crates/engine/src/update_table.rs b/crates/engine/src/update_table.rs index 6db763f82..14970ac8a 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 000000000..aa4cfda3a --- /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/tests/python/test_update_table_sentences.py b/tests/python/test_update_table_sentences.py new file mode 100644 index 000000000..09d35d909 --- /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]" + )