fix(streams): disabling a stream closes its shards; enabling makes a new one

.github/workflows/integration.yml

@@ -323,7 +323,7 @@ jobs:
         run: >-
           cargo test --release -p extenddb-storage-postgres
           --test vector_control_plane --test put_create_race --test twi_conflict
-          --test stream_shard_retention --test stream_shard_ownership --test stream_records_page_order --test backup_restore
+          --test stream_shard_retention --test stream_shard_ownership --test stream_records_page_order --test stream_disable --test stream_disable_legacy --test backup_restore
           --test gsi_online_build --test gsi_stale_writer --test table_snapshot
 
       # The daemonized server logs to syslog; dump it so server-side failures

crates/core/src/types/table.rs

@@ -815,8 +815,20 @@ impl TableDescription {
     ///   table is ACTIVE (measured on DynamoDB, and still absent 30 s later).
     ///   Backends report the summary whenever the table has a registered
     ///   restore.
+    /// - A disabled `StreamSpecification` is omitted: the service carries the
+    ///   member only while a stream is enabled (measured 2026-10-24 on the
+    ///   disable response and on `DescribeTable` afterwards). Backends keep
+    ///   the disabled specification stored, with the closed stream's view
+    ///   type, so `DescribeStream` of that stream can still report it.
     pub fn prepare_for_response(&mut self) {
         self.populate_table_throughput_mode_summary();
+        if self
+            .stream_specification
+            .as_ref()
+            .is_some_and(|s| !s.stream_enabled)
+        {
+            self.stream_specification = None;
+        }
         if self.table_status != TableStatus::Creating {
             self.restore_summary = None;
         } else if let Some(summary) = self.restore_summary.as_mut() {

crates/core/src/validation/mod.rs

@@ -659,15 +659,21 @@ fn validate_index_projections(input: &CreateTableInput) -> Result<(), DynamoDbEr
 /// # Errors
 ///
 /// Returns `DynamoDbError::ValidationException` when `StreamEnabled` is false
-/// but a `StreamViewType` is present.
+/// but a `StreamViewType` is present, or `StreamEnabled` is true without one
+/// (measured 2026-10-24; its order against the other CreateTable checks is not
+/// measured, so it sits with the other stream check).
 fn validate_stream_specification(input: &CreateTableInput) -> Result<(), DynamoDbError> {
-    if let Some(stream) = &input.stream_specification
-        && !stream.stream_enabled
-        && stream.stream_view_type.is_some()
-    {
-        return Err(DynamoDbError::ValidationException(
-            "One or more parameter values were invalid: Table is being created with a stream disabled, UpdateViewType should not be specified".to_owned(),
-        ));
+    if let Some(stream) = &input.stream_specification {
+        if !stream.stream_enabled && stream.stream_view_type.is_some() {
+            return Err(DynamoDbError::ValidationException(
+                "One or more parameter values were invalid: Table is being created with a stream disabled, UpdateViewType should not be specified".to_owned(),
+            ));
+        }
+        if stream.stream_enabled && stream.stream_view_type.is_none() {
+            return Err(DynamoDbError::ValidationException(
+                "One or more parameter values were invalid: Table is being created with a stream enabled, UpdateViewType is required".to_owned(),
+            ));
+        }
     }
     Ok(())
 }
@@ -3722,6 +3728,17 @@ mod tests {
             stream_view_type: Some(StreamViewType::NewImage),
         });
         assert!(validate_create_table(&input, &LimitsConfig::default()).is_ok());
+
+        // Enabled without a view type is refused (measured 2026-10-24).
+        input.stream_specification = Some(StreamSpecification {
+            stream_enabled: true,
+            stream_view_type: None,
+        });
+        let err = validate_create_table(&input, &LimitsConfig::default()).unwrap_err();
+        assert_eq!(
+            err.to_string(),
+            "One or more parameter values were invalid: Table is being created with a stream enabled, UpdateViewType is required"
+        );
     }
 
     #[test]

crates/engine/src/backup.rs

@@ -21,7 +21,7 @@ use extenddb_core::limits::LimitsConfig;
 use extenddb_core::types::{
     BackupDescription, BackupSummary, BillingMode, CreateTableInput, GlobalSecondaryIndexInfo,
     LocalSecondaryIndexInfo, ProvisionedThroughput, RestoreSummary, SourceTableFeatureDetails,
-    SseDescription, TableDescription, TableStatus,
+    SseDescription, StreamSpecification, TableDescription, TableStatus,
 };
 use extenddb_storage::BackupEngine;
 use extenddb_storage::error::StorageError;
@@ -489,7 +489,23 @@ fn apply_source_to_description(
                         })
                         .collect()
                 }),
-                stream_description: d.stream_specification.clone(),
+                // A backup of a table whose stream was disabled carries the
+                // stored `{StreamEnabled: false}`, and DescribeBackup shows it
+                // as `StreamDescription: {StreamEnabled: false}`, as it did
+                // before the view type was kept in the stored specification;
+                // the service's answer for that backup is not measured. The
+                // view type the backends now keep for the closed stream is
+                // dropped here so the answer stays the one before.
+                stream_description: d.stream_specification.clone().map(|s| {
+                    if s.stream_enabled {
+                        s
+                    } else {
+                        StreamSpecification {
+                            stream_enabled: false,
+                            stream_view_type: None,
+                        }
+                    }
+                }),
                 time_to_live_description: d.time_to_live.clone(),
                 sse_description: SseDescription::from_specification(
                     d.sse_specification.as_ref(),

crates/engine/src/create_table.rs

@@ -29,6 +29,7 @@ pub async fn handle_create_table(
         ],
     )?;
 
+    crate::require_stream_enabled_member(&body)?;
     let mut input: CreateTableInput = serde_json::from_value(body).map_err(|e| {
         let msg = e.to_string();
         if msg.contains("validation error detected")

crates/engine/src/lib.rs

@@ -151,6 +151,26 @@ pub(crate) fn sanitize_storage_error(e: extenddb_storage::error::StorageError) -
     DynamoDbError::InternalServerError("Internal server error".to_owned())
 }
 
+/// The null-member sentence for a `StreamSpecification` object without
+/// `StreamEnabled`, on `CreateTable` and `UpdateTable` alike (measured
+/// 2026-10-24). Checked before the body is deserialised, where the missing
+/// field would otherwise surface as a serialization error; its order against
+/// the other request checks is not measured. A `StreamSpecification` that is
+/// not an object is left to deserialisation, which refuses it as a
+/// `SerializationException`.
+pub(crate) fn require_stream_enabled_member(body: &serde_json::Value) -> Result<(), DynamoDbError> {
+    if let Some(spec) = body.get("StreamSpecification").filter(|v| v.is_object())
+        && spec
+            .get("StreamEnabled")
+            .is_none_or(serde_json::Value::is_null)
+    {
+        return Err(DynamoDbError::ValidationException(
+            "1 validation error detected: Value null at 'streamSpecification.streamEnabled' failed to satisfy constraint: Member must not be null".to_owned(),
+        ));
+    }
+    Ok(())
+}
+
 /// Map a serde deserialization error to the appropriate `DynamoDB` error type.
 ///
 /// Enum validation errors (produced by custom Deserialize impls) contain