From 98bca8ecb71e678e12c47a0bcecc45f0b2bbc0b3 Mon Sep 17 00:00:00 2001
From: Lee Hannigan <lhnng@amazon.com>
Date: Mon, 26 Oct 2026 10:11:23 +0000
Subject: [PATCH] fix: answer TTL and tag calls during a table's DELETING
 window as the service does

DescribeTimeToLive on a DELETING table answers 400 ValidationException
"Cannot describe time to live while table is in DELETING state: Current table
state is DELETING", and on a CREATING table the same sentence naming
CREATING; it is served on ACTIVE and UPDATING tables. UpdateTimeToLive on a
DELETING table runs the same rules as on an ACTIVE one. Amazon DynamoDB
answers these calls this way (measured 2026-10-25 and 2026-10-26). On
ExtendDB the UpdateTimeToLive write goes to the catalog row, which the
delete removes.

ExtendDB answered UpdateTimeToLive on a DELETING table with 400
ResourceInUseException "Table <name> is not in ACTIVE state", because every
backend's update_ttl admitted ACTIVE rows only, and DescribeTimeToLive with
200 on CREATING and DELETING tables, because nothing read the status.

The engine reads the status with describe_table, since TableKeyInfo carries
no status and MongoDB and Cassandra report a DELETING row as not found from
the key-info read. The read is four catalog queries on PostgreSQL and SQLite
and two on MongoDB and Cassandra, once per call. update_ttl admits DELETING
beside ACTIVE in the conditional write on PostgreSQL, SQLite and MongoDB, and
in the status check and both lightweight transactions on Cassandra. CREATING
and UPDATING keep TableNotActive, and a missing table keeps TableNotFound.

UpdateTimeToLive sends no index DDL for a table it read as DELETING, since the
delete removes the data table or collection with its index. On PostgreSQL a
disable's DROP INDEX waited with no lock timeout behind whatever held the data
table and queued the table's data-plane requests behind it. On MongoDB an
enable's create_index could create the collection again after the delete
dropped it, and the collection outlived the table. MongoDB create_ttl_index
also reads an ACTIVE catalog document only. A delete that commits between the
status read and the write meets the index work, which the base refused at the
write; on MongoDB the ACTIVE filter covers that order.

TagResource and UntagResource on the ARN of a DELETING table answer 400
ResourceInUseException "Attempt to change a resource which is still in use:
Table is being deleted: <name>", and ListTagsOfResource is served with the
tags. On the table's stream ARN both writes and the read are served. Amazon
DynamoDB answers this way (measured 2026-10-25 and 2026-10-26).

ExtendDB answered ResourceNotFoundException for all three on MongoDB and
Cassandra and accepted both writes on PostgreSQL and SQLite. An accepted
TagResource on the table ARN writes an unconditioned tag row; on PostgreSQL
the delete removes the ARN's tag rows before the catalog row, so a tag written
between the two statements could outlive the table and be inherited by the
next table of that name.

The tag path reads the table's status before the key-info read, so all four
backends reach the window's answers, and refuses the two writes when the ARN
names the table. The key-info read still follows for every status but
DELETING, so a tag call on an ACTIVE table sends seven catalog queries before
its tag statements on PostgreSQL and SQLite, where it sent three, and four on
MongoDB and Cassandra, where it sent two. The refusal removes the writes
admitted during the window. A write whose status read came before the delete
can still be written after the delete removed the tag rows, as before.

ListTagsOfResource returns the tags for the whole window on PostgreSQL and
SQLite. On MongoDB it returns no tags once the delete call has removed them;
the service returns them for the whole window.

A MongoDB storage test drives the tag and TTL handlers on a DELETING
document, and devtools/run-mongodb-tests runs the crate's ignored
deleting_window tests in CI.

Closes #552
Closes #551
---
 Cargo.lock                                    |   2 +
 crates/core/src/error/messages.rs             |  18 ++
 crates/engine/src/lib.rs                      |   1 +
 crates/engine/src/table_lifecycle.rs          |  64 +++++
 crates/engine/src/tagging.rs                  | 164 ++++++++++-
 crates/engine/src/ttl.rs                      |  86 +++++-
 .../storage-cassandra/src/metadata_engine.rs  |  10 +-
 .../tests/direct/deleting_window.rs           |  49 ++++
 crates/storage-mongodb/Cargo.toml             |   3 +
 .../src/deleting_window_tests.rs              | 258 ++++++++++++++++++
 crates/storage-mongodb/src/metadata_engine.rs |  17 +-
 crates/storage-postgres/Cargo.toml            |   3 +
 .../storage-postgres/src/metadata_engine.rs   |   7 +-
 .../storage-postgres/tests/deleting_window.rs |   3 +
 .../tests/deleting_window/ttl.rs              | 180 ++++++++++++
 crates/storage-sqlite/src/delete_table.rs     |  80 +++++-
 crates/storage-sqlite/src/metadata.rs         |   7 +-
 devtools/run-mongodb-tests                    |  10 +
 .../test_deleting_window_control_plane.py     | 174 +++++++++++-
 19 files changed, 1105 insertions(+), 31 deletions(-)
 create mode 100644 crates/engine/src/table_lifecycle.rs
 create mode 100644 crates/storage-postgres/tests/deleting_window/ttl.rs

diff --git a/Cargo.lock b/Cargo.lock
index 003569679..e0ba802e9 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1448,6 +1448,7 @@ dependencies = [
  "dashmap",
  "extenddb-auth",
  "extenddb-core",
+ "extenddb-engine",
  "extenddb-storage",
  "futures",
  "mongodb",
@@ -1476,6 +1477,7 @@ dependencies = [
  "crc32fast",
  "extenddb-auth",
  "extenddb-core",
+ "extenddb-engine",
  "extenddb-storage",
  "futures",
  "libc",
diff --git a/crates/core/src/error/messages.rs b/crates/core/src/error/messages.rs
index e5aedc057..8db9ee01b 100755
--- a/crates/core/src/error/messages.rs
+++ b/crates/core/src/error/messages.rs
@@ -44,6 +44,14 @@ pub enum ErrorMessageKey {
     /// Measured on the service 2026-10-24 for a second `DeleteTable` and for
     /// `UpdateTable` during the window.
     TableBeingDeleted,
+    /// `DescribeTimeToLive` refused because the table is `DELETING`. Measured
+    /// on the service 2026-10-25 on three tables, every call inside the window.
+    /// `UpdateTimeToLive` has no such refusal: it runs its ordinary rules.
+    DescribeTtlWhileDeleting,
+    /// `DescribeTimeToLive` refused because the table is `CREATING`: the
+    /// `DELETING` sentence with the state substituted. Measured on the service
+    /// 2026-10-26, five calls, each while `DescribeTable` read `CREATING`.
+    DescribeTtlWhileCreating,
     MissingKeySchema,
     KeySchemaTooMany,
     KeySchemaFirstNotHash,
@@ -92,6 +100,16 @@ pub fn error_message(key: ErrorMessageKey, args: &[&str]) -> String {
                 args[0]
             )
         }
+        ErrorMessageKey::DescribeTtlWhileDeleting => {
+            "Cannot describe time to live while table is in DELETING state: \
+             Current table state is DELETING"
+                .to_owned()
+        }
+        ErrorMessageKey::DescribeTtlWhileCreating => {
+            "Cannot describe time to live while table is in CREATING state: \
+             Current table state is CREATING"
+                .to_owned()
+        }
         ErrorMessageKey::MissingKeySchema => {
             "1 validation error detected: Value null at 'keySchema' failed to satisfy constraint: \
              Member must not be null"
diff --git a/crates/engine/src/lib.rs b/crates/engine/src/lib.rs
index 74083fbfa..c960fa4c7 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 table_lifecycle;
 mod tagging;
 mod transact_get_items;
 mod transact_write_helpers;
diff --git a/crates/engine/src/table_lifecycle.rs b/crates/engine/src/table_lifecycle.rs
new file mode 100644
index 000000000..839df5117
--- /dev/null
+++ b/crates/engine/src/table_lifecycle.rs
@@ -0,0 +1,64 @@
+// Copyright 2026 ExtendDB contributors
+// SPDX-License-Identifier: Apache-2.0
+
+//! The catalog's lifecycle status for a table, for handlers that decide on it.
+//!
+//! `TableKeyInfo` carries no status, and the key-info read reports a `DELETING`
+//! row as not found on MongoDB and Cassandra, so a handler that has to tell a
+//! `DELETING` table from a missing one cannot get the answer from it. The
+//! description carries the status on every backend. The callers (the two TTL
+//! handlers and the tag path) are control-plane calls, so the read is not on a
+//! data-plane path.
+//!
+//! The read is `describe_table`: four catalog queries on PostgreSQL and SQLite
+//! (`tables`, `indexes`, `vector_indexes`, `table_restores`) and two on MongoDB
+//! and Cassandra (`tables`, `indexes`). `DescribeTimeToLive`,
+//! `UpdateTimeToLive`, `TagResource`, `UntagResource` and `ListTagsOfResource`
+//! each make it once per call. The tag path keeps the key-info read after it
+//! for every status but `DELETING`, so a tag call on an `ACTIVE` table sends
+//! seven queries before its tag statements on PostgreSQL and SQLite, where it
+//! sent three, and four on MongoDB and Cassandra, where it sent two.
+//! `DescribeTimeToLive` on an `ACTIVE` table sends five and three, where it
+//! sent one. `UpdateTimeToLive` sends the four or two more, and on a
+//! `DELETING` table it skips the TTL read after the write and the index
+//! statements.
+
+use extenddb_core::error::DynamoDbError;
+use extenddb_core::types::{DescribeTableInput, TableStatus};
+use extenddb_storage::error::StorageError;
+
+use crate::OperationContext;
+
+/// Read `table_name`'s current status from the catalog.
+///
+/// # Errors
+///
+/// Returns [`StorageError::TableNotFound`] when no catalog row exists, and
+/// whatever the backend raised otherwise. The caller maps it, so each operation
+/// keeps the message it already answers for a missing table.
+pub(crate) async fn read_table_status(
+    table_name: &str,
+    ctx: &OperationContext,
+) -> Result<TableStatus, StorageError> {
+    let desc = ctx
+        .storage
+        .describe_table(
+            &ctx.account_id,
+            DescribeTableInput {
+                table_name: table_name.to_owned(),
+            },
+        )
+        .await?;
+    Ok(desc.table_status)
+}
+
+/// The service's refusal of a change to a table that is being deleted, worded
+/// once in [`extenddb_core::error::ErrorMessageKey::TableBeingDeleted`] and
+/// classified the way the control-plane mappers classify
+/// [`StorageError::TableBeingDeleted`].
+pub(crate) fn being_deleted(table_name: &str) -> DynamoDbError {
+    DynamoDbError::ResourceInUseException(extenddb_core::error::error_message(
+        extenddb_core::error::ErrorMessageKey::TableBeingDeleted,
+        &[table_name],
+    ))
+}
diff --git a/crates/engine/src/tagging.rs b/crates/engine/src/tagging.rs
index 432d75dd4..c0520f8b4 100755
--- a/crates/engine/src/tagging.rs
+++ b/crates/engine/src/tagging.rs
@@ -5,7 +5,8 @@
 
 use extenddb_core::error::DynamoDbError;
 use extenddb_core::types::{
-    ListTagsOfResourceInput, ListTagsOfResourceOutput, TagResourceInput, UntagResourceInput,
+    ListTagsOfResourceInput, ListTagsOfResourceOutput, TableStatus, TagResourceInput,
+    UntagResourceInput,
 };
 use serde_json::Value;
 
@@ -34,6 +35,15 @@ fn tag_subresource_not_table_or_stream(resource: &str) -> bool {
     !matches!(segs.next(), None | Some("") | Some("stream"))
 }
 
+/// True for a stream ARN, `table/<name>/stream/<label>`. A stream's tags belong
+/// to the stream, which stays describable after its table is deleted, so the
+/// table's `DELETING` refusal does not apply to them.
+fn arn_names_stream(arn: &str) -> bool {
+    extenddb_core::arn::fields(arn)
+        .and_then(|f| f.resource.strip_prefix("table/"))
+        .is_some_and(|tail| tail.split('/').nth(1) == Some("stream"))
+}
+
 /// Extract the table name from a `DynamoDB` table ARN, using the one shared
 /// resource parser ([`extenddb_core::arn::fields`]) so this layer splits an ARN
 /// the same way authorization and the engine stream resolver do.
@@ -46,7 +56,8 @@ fn extract_table_name_from_arn(arn: &str) -> Option<&str> {
     Some(table_name.split('/').next().unwrap_or(table_name))
 }
 
-/// Validate that the ARN refers to an existing table owned by the caller.
+/// Validate that the ARN refers to an existing table owned by the caller, and
+/// report that table's catalog status so the caller can decide on it.
 ///
 /// The error classes and messages below match `DynamoDB`, verified against the
 /// service:
@@ -67,7 +78,10 @@ fn extract_table_name_from_arn(arn: &str) -> Option<&str> {
 /// fields are valid. For a table ARN, a no-permissions principal is denied by
 /// authorization before this runs; the partition, service, region, and
 /// existence errors therefore surface only for an authorized principal.
-async fn validate_resource_arn(arn: &str, ctx: &OperationContext) -> Result<(), DynamoDbError> {
+async fn validate_resource_arn<'a>(
+    arn: &'a str,
+    ctx: &OperationContext,
+) -> Result<(TableStatus, &'a str), DynamoDbError> {
     if !arn.starts_with("arn:") {
         return Err(DynamoDbError::ValidationException(format!(
             "One or more parameter values were invalid: ARNs must start with 'arn:': {arn}"
@@ -140,19 +154,62 @@ async fn validate_resource_arn(arn: &str, ctx: &OperationContext) -> Result<(),
         ))
     })?;
 
+    // The table's status, which TableKeyInfo does not carry and the key-info
+    // read cannot supply for a DELETING row on MongoDB and Cassandra (it
+    // reports one as not found). The status read comes first so every backend
+    // reaches the same answer for the window, for a table ARN and for a stream
+    // ARN alike; every other status is left to the key-info read below, which
+    // already classifies it.
+    let status = crate::table_lifecycle::read_table_status(table_name, ctx)
+        .await
+        .map_err(|e| arn_storage_error(e, arn))?;
+    if status == TableStatus::Deleting {
+        return Ok((status, table_name));
+    }
+
     // Verify the table exists via table_key_info (lightweight check).
     ctx.storage
         .table_key_info(&ctx.account_id, table_name)
         .await
-        .map_err(|e| match e {
-            extenddb_storage::error::StorageError::TableNotFound(_) => {
-                DynamoDbError::ResourceNotFoundException(format!(
-                    "Requested resource not found: ResourceArn: {arn} not found"
-                ))
-            }
-            other => sanitize_storage_error(other),
-        })?;
+        .map_err(|e| arn_storage_error(e, arn))?;
+
+    Ok((status, table_name))
+}
 
+/// Report a storage error from the table lookup behind a resource ARN: a
+/// missing table names the ARN, as the service does, and anything else is a
+/// fault.
+fn arn_storage_error(e: extenddb_storage::error::StorageError, arn: &str) -> DynamoDbError {
+    match e {
+        extenddb_storage::error::StorageError::TableNotFound(_) => {
+            DynamoDbError::ResourceNotFoundException(format!(
+                "Requested resource not found: ResourceArn: {arn} not found"
+            ))
+        }
+        other => sanitize_storage_error(other),
+    }
+}
+
+/// Refuse a tag write on a table ARN while the table is being deleted.
+///
+/// `TagResource` and `UntagResource` on a table ARN during the table's
+/// `DELETING` window answer 400 `ResourceInUseException` with the `Table is
+/// being deleted` sentence, and `ListTagsOfResource` is served with the tags
+/// (measured 2026-10-25). On the table's stream ARN both writes are served in
+/// the same window (measured 2026-10-26), so `arn` decides. A write admitted
+/// on the table ARN in the window would land an unconditioned tag row that can
+/// outlive the table, since the delete removes the ARN's tag rows before the
+/// catalog row. The refusal removes those admissions; a write whose status
+/// read came before the delete can still land after the delete removed the
+/// tag rows, as it could before.
+fn refuse_tag_write_while_deleting(
+    status: TableStatus,
+    table_name: &str,
+    arn: &str,
+) -> Result<(), DynamoDbError> {
+    if status == TableStatus::Deleting && !arn_names_stream(arn) {
+        return Err(crate::table_lifecycle::being_deleted(table_name));
+    }
     Ok(())
 }
 
@@ -161,6 +218,8 @@ async fn validate_resource_arn(arn: &str, ctx: &OperationContext) -> Result<(),
 /// # Errors
 ///
 /// Returns `ResourceNotFoundException` if the resource does not exist.
+/// Returns `ResourceInUseException` if the ARN names a table that is being
+/// deleted.
 /// Returns `ValidationException` if the resource ARN is empty.
 /// Returns `InternalServerError` on storage failures.
 pub async fn handle_tag_resource(
@@ -175,7 +234,8 @@ pub async fn handle_tag_resource(
         ));
     }
 
-    validate_resource_arn(&input.resource_arn, ctx).await?;
+    let (status, table_name) = validate_resource_arn(&input.resource_arn, ctx).await?;
+    refuse_tag_write_while_deleting(status, table_name, &input.resource_arn)?;
 
     ctx.storage
         .tag_resource(&input.resource_arn, &input.tags)
@@ -197,6 +257,8 @@ pub async fn handle_tag_resource(
 /// # Errors
 ///
 /// Returns `ResourceNotFoundException` if the resource does not exist.
+/// Returns `ResourceInUseException` if the ARN names a table that is being
+/// deleted.
 /// Returns `ValidationException` if the resource ARN is empty.
 /// Returns `InternalServerError` on storage failures.
 pub async fn handle_untag_resource(
@@ -212,7 +274,8 @@ pub async fn handle_untag_resource(
         ));
     }
 
-    validate_resource_arn(&input.resource_arn, ctx).await?;
+    let (status, table_name) = validate_resource_arn(&input.resource_arn, ctx).await?;
+    refuse_tag_write_while_deleting(status, table_name, &input.resource_arn)?;
 
     ctx.storage
         .untag_resource(&input.resource_arn, &input.tag_keys)
@@ -247,7 +310,9 @@ pub async fn handle_list_tags_of_resource(
         ));
     }
 
-    validate_resource_arn(&input.resource_arn, ctx).await?;
+    // Served while the table is DELETING, with the tags, as the service serves
+    // it (measured 2026-10-25): the read is not a change to the resource.
+    let (_status, _table_name) = validate_resource_arn(&input.resource_arn, ctx).await?;
 
     let tags = ctx
         .storage
@@ -264,7 +329,76 @@ pub async fn handle_list_tags_of_resource(
 
 #[cfg(test)]
 mod tests {
-    use super::{tag_subresource_not_dynamodb_resource, tag_subresource_not_table_or_stream};
+    use super::{
+        arn_storage_error, refuse_tag_write_while_deleting, tag_subresource_not_dynamodb_resource,
+        tag_subresource_not_table_or_stream,
+    };
+    use extenddb_core::error::DynamoDbError;
+    use extenddb_core::types::TableStatus;
+    use extenddb_storage::error::StorageError;
+
+    const TABLE_ARN: &str = "arn:aws:dynamodb:us-east-1:123456789012:table/t";
+    const STREAM_ARN: &str =
+        "arn:aws:dynamodb:us-east-1:123456789012:table/t/stream/2026-10-26T00:13:40.188";
+
+    /// A tag write on the table ARN is refused while the table is DELETING,
+    /// with the service's sentence, and admitted in every other status.
+    #[test]
+    fn a_tag_write_is_refused_only_while_deleting() {
+        match refuse_tag_write_while_deleting(TableStatus::Deleting, "t", TABLE_ARN).unwrap_err() {
+            DynamoDbError::ResourceInUseException(msg) => assert_eq!(
+                msg,
+                "Attempt to change a resource which is still in use: Table is being deleted: t"
+            ),
+            other => panic!("expected ResourceInUseException, got {other:?}"),
+        }
+        for status in [
+            TableStatus::Active,
+            TableStatus::Creating,
+            TableStatus::Updating,
+        ] {
+            assert!(
+                refuse_tag_write_while_deleting(status, "t", TABLE_ARN).is_ok(),
+                "{status:?} must keep its existing answer"
+            );
+        }
+    }
+
+    /// A tag write on the table's stream ARN is admitted in every status,
+    /// DELETING included: the service serves both writes there during the
+    /// window (measured 2026-10-26).
+    #[test]
+    fn a_stream_arn_tag_write_is_admitted_while_deleting() {
+        for status in [
+            TableStatus::Deleting,
+            TableStatus::Active,
+            TableStatus::Creating,
+            TableStatus::Updating,
+        ] {
+            assert!(
+                refuse_tag_write_while_deleting(status, "t", STREAM_ARN).is_ok(),
+                "{status:?} on the stream ARN must be admitted"
+            );
+        }
+    }
+
+    /// A missing table behind a resource ARN names the ARN; anything else is a
+    /// fault, not a client error.
+    #[test]
+    fn a_missing_table_behind_an_arn_names_the_arn() {
+        let arn = "arn:aws:dynamodb:us-east-1:123456789012:table/t";
+        match arn_storage_error(StorageError::TableNotFound("t".to_owned()), arn) {
+            DynamoDbError::ResourceNotFoundException(msg) => assert_eq!(
+                msg,
+                format!("Requested resource not found: ResourceArn: {arn} not found")
+            ),
+            other => panic!("expected ResourceNotFoundException, got {other:?}"),
+        }
+        assert!(matches!(
+            arn_storage_error(StorageError::Internal("boom".to_owned()), arn),
+            DynamoDbError::InternalServerError(_)
+        ));
+    }
 
     #[test]
     fn tag_subresource_classification() {
diff --git a/crates/engine/src/ttl.rs b/crates/engine/src/ttl.rs
index ed1e509bc..0d6642bac 100755
--- a/crates/engine/src/ttl.rs
+++ b/crates/engine/src/ttl.rs
@@ -5,7 +5,7 @@
 
 use extenddb_core::error::DynamoDbError;
 use extenddb_core::types::{
-    DescribeTimeToLiveInput, DescribeTimeToLiveOutput, TimeToLiveSpecificationOutput,
+    DescribeTimeToLiveInput, DescribeTimeToLiveOutput, TableStatus, TimeToLiveSpecificationOutput,
     TimeToLiveStatus, UpdateTimeToLiveInput, UpdateTimeToLiveOutput,
 };
 use extenddb_core::validation::validate_table_name;
@@ -15,10 +15,31 @@ use serde_json::Value;
 use crate::OperationContext;
 use crate::serialize_output;
 
+/// Gate a `DescribeTimeToLive` request on the table's catalog status.
+///
+/// The service refuses the call while the table is `CREATING` or `DELETING`,
+/// with one sentence that names the state, and serves it while the table is
+/// `ACTIVE` or `UPDATING`. Measured: `DELETING` on 2026-10-25 on three tables,
+/// the refusal coming back before and after an `UpdateTimeToLive` inside the
+/// same window; `CREATING` and `UPDATING` on 2026-10-26, five calls each.
+/// `UpdateTimeToLive` has no equivalent refusal, so the gate is on this path
+/// only.
+fn check_describe_status(status: TableStatus) -> Result<(), DynamoDbError> {
+    let key = match status {
+        TableStatus::Creating => extenddb_core::error::ErrorMessageKey::DescribeTtlWhileCreating,
+        TableStatus::Deleting => extenddb_core::error::ErrorMessageKey::DescribeTtlWhileDeleting,
+        TableStatus::Active | TableStatus::Updating => return Ok(()),
+    };
+    Err(DynamoDbError::ValidationException(
+        extenddb_core::error::error_message(key, &[]),
+    ))
+}
+
 /// Handle `DescribeTimeToLive` — return TTL configuration for a table.
 ///
 /// # Errors
 ///
+/// Returns `ValidationException` if the table is `CREATING` or `DELETING`.
 /// Returns `ResourceNotFoundException` if the table does not exist.
 /// Returns `InternalServerError` on storage failures.
 pub async fn handle_describe_time_to_live(
@@ -30,6 +51,14 @@ pub async fn handle_describe_time_to_live(
 
     validate_table_name(&input.table_name, &ctx.limits)?;
 
+    // The TTL columns live on the catalog row, which outlives the status
+    // change, so the read below answers for a DELETING table too: the status
+    // is what decides, and the key-info read cannot supply it.
+    let status = crate::table_lifecycle::read_table_status(&input.table_name, ctx)
+        .await
+        .map_err(storage_to_dynamo)?;
+    check_describe_status(status)?;
+
     let desc = ctx
         .storage
         .describe_ttl(&ctx.account_id, &input.table_name)
@@ -116,6 +145,13 @@ pub async fn handle_update_time_to_live(
         None
     };
 
+    // The status decides whether the index work after the write runs at all.
+    // It is read before the write: a delete that starts between the two meets
+    // that work, as before.
+    let status = crate::table_lifecycle::read_table_status(&input.table_name, ctx)
+        .await
+        .map_err(storage_to_dynamo)?;
+
     ctx.storage
         .update_ttl(
             &ctx.account_id,
@@ -126,7 +162,15 @@ pub async fn handle_update_time_to_live(
         .await
         .map_err(storage_to_dynamo)?;
 
-    if input.time_to_live_specification.enabled {
+    if status == TableStatus::Deleting {
+        // No index DDL for a table being deleted: the delete removes its data
+        // table or collection with the index, so the catalog write above is
+        // the whole change. If it ran, a PostgreSQL DROP INDEX would wait, with
+        // no lock timeout, behind whatever holds the data table, and the
+        // table's data-plane requests would queue behind it; a MongoDB
+        // create_index on the dropped collection would create the collection
+        // again, and no catalog document would be left to name it.
+    } else if input.time_to_live_specification.enabled {
         // Backends that report ENABLING (Cassandra, PostgreSQL) get the
         // backfill kicked off in the background: it is a full table scan, and
         // awaiting it made the API call take as long as the table is large.
@@ -226,9 +270,11 @@ fn storage_to_dynamo(e: StorageError) -> DynamoDbError {
         StorageError::TableNotActive(name) => {
             DynamoDbError::ResourceInUseException(format!("Table {name} is not in ACTIVE state"))
         }
-        // No TTL path raises this today (the TTL metadata paths answer
-        // TableNotActive for every non-ACTIVE status), but the variant exists
-        // and a backend that raises it here must not become a 500.
+        // No TTL path raises this today: UpdateTimeToLive is served on a
+        // DELETING row, DescribeTimeToLive refuses it in the handler with the
+        // service's ValidationException, and the other non-ACTIVE statuses
+        // answer TableNotActive. The variant exists and a backend that raises
+        // it here must not become a 500.
         StorageError::TableBeingDeleted(name) => {
             DynamoDbError::ResourceInUseException(extenddb_core::error::error_message(
                 extenddb_core::error::ErrorMessageKey::TableBeingDeleted,
@@ -291,6 +337,36 @@ mod tests {
         assert!(validate_ttl_attribute_name(&max).is_ok());
     }
 
+    /// DescribeTimeToLive is refused while the table is CREATING or DELETING,
+    /// with the service's sentence for each byte for byte, and served while it
+    /// is ACTIVE or UPDATING.
+    #[test]
+    fn describe_is_refused_while_creating_or_deleting() {
+        for (status, sentence) in [
+            (
+                TableStatus::Deleting,
+                "Cannot describe time to live while table is in DELETING state: \
+                 Current table state is DELETING",
+            ),
+            (
+                TableStatus::Creating,
+                "Cannot describe time to live while table is in CREATING state: \
+                 Current table state is CREATING",
+            ),
+        ] {
+            match check_describe_status(status).unwrap_err() {
+                DynamoDbError::ValidationException(msg) => assert_eq!(msg, sentence),
+                other => panic!("expected ValidationException for {status:?}, got {other:?}"),
+            }
+        }
+        for status in [TableStatus::Active, TableStatus::Updating] {
+            assert!(
+                check_describe_status(status).is_ok(),
+                "{status:?} must be served"
+            );
+        }
+    }
+
     #[test]
     fn transition_gate() {
         use TimeToLiveStatus::{Disabled, Disabling, Enabled, Enabling};
diff --git a/crates/storage-cassandra/src/metadata_engine.rs b/crates/storage-cassandra/src/metadata_engine.rs
index 94c9bfd58..001d7c796 100755
--- a/crates/storage-cassandra/src/metadata_engine.rs
+++ b/crates/storage-cassandra/src/metadata_engine.rs
@@ -962,7 +962,11 @@ impl MetadataEngine for CassandraEngine {
                 crate::cassandra_util::get_column(&row, "table_id", "update_ttl_status")?;
             let previous_generation: Option<uuid::Uuid> =
                 row.get_by_name("ttl_generation").ok().flatten();
-            if status != "ACTIVE" {
+            // DELETING is admitted beside ACTIVE: the service runs
+            // UpdateTimeToLive's ordinary rules during the window and the write
+            // lands on the catalog row the delete then removes (measured
+            // 2026-10-25). The other non-ACTIVE statuses keep their refusal.
+            if status != "ACTIVE" && status != "DELETING" {
                 return Err(StorageError::TableNotActive(table_name));
             }
             if enabled {
@@ -1000,7 +1004,7 @@ impl MetadataEngine for CassandraEngine {
                     "UPDATE {}.tables SET ttl_attribute = ?, ttl_generation = ?, \
                      ttl_index_ready = false, ttl_backfill_cursor = null \
                      WHERE account_id = ? AND table_name = ? \
-                     IF table_status = 'ACTIVE' AND ttl_sweep_owner = null \
+                     IF table_status IN ('ACTIVE', 'DELETING') AND ttl_sweep_owner = null \
                      AND ttl_cleanup_generation = null \
                      AND ttl_attribute = null AND ttl_generation = null",
                     self.catalog_keyspace()
@@ -1011,7 +1015,7 @@ impl MetadataEngine for CassandraEngine {
                      ttl_cleanup_generation = ?, ttl_index_ready = false, \
                      ttl_backfill_cursor = null \
                      WHERE account_id = ? AND table_name = ? \
-                     IF table_status = 'ACTIVE' AND ttl_sweep_owner = null \
+                     IF table_status IN ('ACTIVE', 'DELETING') AND ttl_sweep_owner = null \
                      AND ttl_cleanup_generation = null \
                      AND ttl_attribute = ? AND ttl_generation = ?",
                     self.catalog_keyspace()
diff --git a/crates/storage-cassandra/tests/direct/deleting_window.rs b/crates/storage-cassandra/tests/direct/deleting_window.rs
index cc43f7da4..12150ada4 100644
--- a/crates/storage-cassandra/tests/direct/deleting_window.rs
+++ b/crates/storage-cassandra/tests/direct/deleting_window.rs
@@ -174,3 +174,52 @@ async fn update_table_on_a_creating_table_keeps_table_not_active() {
     );
     remove_row(&engine, &account.account_id).await;
 }
+
+/// UpdateTimeToLive during the window writes the catalog row, the way the
+/// service runs its ordinary TTL rules there (measured 2026-10-25), and leaves
+/// the status alone; the other non-ACTIVE statuses keep their refusal. The
+/// enable is undone before the row is removed so the account keyspace's TTL
+/// entries do not outlive the test.
+#[tokio::test]
+async fn update_ttl_during_the_window_is_served() {
+    use extenddb_core::types::TimeToLiveStatus;
+    use extenddb_storage::MetadataEngine;
+
+    if crate::helpers::skip_without_cassandra() {
+        return;
+    }
+    let (engine, account, _table) = open_window().await;
+
+    engine
+        .update_ttl(&account.account_id, TABLE, "ttl", true)
+        .await
+        .expect("UpdateTimeToLive on a DELETING table");
+    let desc = engine
+        .describe_ttl(&account.account_id, TABLE)
+        .await
+        .expect("the TTL description");
+    assert_eq!(desc.attribute_name.as_deref(), Some("ttl"));
+    assert_ne!(desc.time_to_live_status, TimeToLiveStatus::Disabled);
+    assert_eq!(
+        status_of(&engine, &account.account_id).await.as_deref(),
+        Some("DELETING"),
+        "the write must not change the status"
+    );
+
+    engine
+        .update_ttl(&account.account_id, TABLE, "ttl", false)
+        .await
+        .expect("the disable direction during the window");
+
+    set_status(&engine, &account.account_id, "CREATING").await;
+    let err = engine
+        .update_ttl(&account.account_id, TABLE, "ttl", true)
+        .await
+        .expect_err("UpdateTimeToLive on a CREATING table");
+    assert!(
+        matches!(&err, StorageError::TableNotActive(name) if name == TABLE),
+        "expected TableNotActive({TABLE:?}), got {err:?}"
+    );
+
+    remove_row(&engine, &account.account_id).await;
+}
diff --git a/crates/storage-mongodb/Cargo.toml b/crates/storage-mongodb/Cargo.toml
index a8d5380a9..df4934381 100644
--- a/crates/storage-mongodb/Cargo.toml
+++ b/crates/storage-mongodb/Cargo.toml
@@ -61,3 +61,6 @@ dashmap.workspace = true
 # Property-based testing for the filter-pushdown parity harness
 # (crates/storage-mongodb/tests/pushdown_parity.rs).
 proptest = "1"
+# The DELETING-window rows call the engine's tag and TTL handlers over a
+# MongoEngine (crates/storage-mongodb/src/deleting_window_tests.rs).
+extenddb-engine.workspace = true
diff --git a/crates/storage-mongodb/src/deleting_window_tests.rs b/crates/storage-mongodb/src/deleting_window_tests.rs
index c8aeb44a2..e8e6edd67 100644
--- a/crates/storage-mongodb/src/deleting_window_tests.rs
+++ b/crates/storage-mongodb/src/deleting_window_tests.rs
@@ -112,6 +112,36 @@ impl Fixture {
             .map(|d| d.get_str("table_status").expect("table_status").to_owned())
     }
 
+    /// Drop `TABLE`'s data collection, as a delete call does after it marks
+    /// the document `DELETING` and before it removes the document, and return
+    /// the collection's name.
+    async fn drop_data_collection(&self) -> String {
+        let doc = self
+            .engine
+            .catalog_db
+            .collection::<Document>("tables")
+            .find_one(doc! { "_id": { "account_id": ACCOUNT, "table_name": TABLE } })
+            .await
+            .expect("read the table document")
+            .expect("the table document");
+        let name = crate::data::data_collection_name(doc.get_str("table_id").expect("table_id"));
+        self.engine
+            .data_db
+            .collection::<Document>(&name)
+            .drop()
+            .await
+            .expect("drop the data collection");
+        name
+    }
+
+    async fn data_collections(&self) -> Vec<String> {
+        self.engine
+            .data_db
+            .list_collection_names()
+            .await
+            .expect("list collections")
+    }
+
     async fn finish(self) {
         self.engine.catalog_db.drop().await.expect("drop catalog");
         self.engine.data_db.drop().await.expect("drop data");
@@ -294,3 +324,231 @@ async fn two_concurrent_delete_tables_have_one_winner() {
     drop(engine);
     f.finish().await;
 }
+
+/// UpdateTimeToLive during the window writes the catalog document, the way the
+/// service runs its ordinary TTL rules there (measured 2026-10-25), and the
+/// other non-ACTIVE statuses keep their refusal.
+#[tokio::test]
+#[ignore = "needs a MongoDB replica set in EXTENDDB_TEST_MONGODB_URI"]
+async fn update_ttl_during_the_window_is_served() {
+    use extenddb_core::types::TimeToLiveStatus;
+    use extenddb_storage::MetadataEngine;
+
+    let f = Fixture::new().await;
+    f.open_window().await;
+
+    f.engine
+        .update_ttl(ACCOUNT, TABLE, "ttl", true)
+        .await
+        .expect("UpdateTimeToLive on a DELETING table");
+    let desc = f
+        .engine
+        .describe_ttl(ACCOUNT, TABLE)
+        .await
+        .expect("the TTL description");
+    assert_eq!(desc.time_to_live_status, TimeToLiveStatus::Enabled);
+    assert_eq!(desc.attribute_name.as_deref(), Some("ttl"));
+    assert_eq!(
+        f.status().await.as_deref(),
+        Some("DELETING"),
+        "the write must not change the status"
+    );
+
+    f.engine
+        .update_ttl(ACCOUNT, TABLE, "ttl", false)
+        .await
+        .expect("the disable direction during the window");
+    let desc = f
+        .engine
+        .describe_ttl(ACCOUNT, TABLE)
+        .await
+        .expect("the TTL description");
+    assert_eq!(desc.time_to_live_status, TimeToLiveStatus::Disabled);
+
+    f.set_status("CREATING").await;
+    let err = f
+        .engine
+        .update_ttl(ACCOUNT, TABLE, "ttl", true)
+        .await
+        .expect_err("UpdateTimeToLive on a CREATING table");
+    assert!(
+        matches!(&err, StorageError::TableNotActive(name) if name == TABLE),
+        "expected TableNotActive({TABLE:?}), got {err:?}"
+    );
+
+    f.set_status("DELETING").await;
+    f.finish().await;
+}
+
+/// A TTL index build that reaches a `DELETING` document after the delete call
+/// dropped the data collection creates no collection: the build reads an
+/// `ACTIVE` document only, so it answers `TableNotFound`. MongoDB would
+/// otherwise create the collection again to hold the index, and nothing would
+/// remove it once the catalog document is gone.
+#[tokio::test]
+#[ignore = "needs a MongoDB replica set in EXTENDDB_TEST_MONGODB_URI"]
+async fn a_ttl_index_build_during_the_window_creates_no_collection() {
+    use extenddb_storage::MetadataEngine;
+
+    let f = Fixture::new().await;
+    f.open_window().await;
+    let data_coll = f.drop_data_collection().await;
+
+    f.engine
+        .update_ttl(ACCOUNT, TABLE, "ttl", true)
+        .await
+        .expect("UpdateTimeToLive on a DELETING table");
+    let built = f.engine.create_ttl_index(ACCOUNT, TABLE, "ttl").await;
+    let collections = f.data_collections().await;
+    assert!(
+        !collections.contains(&data_coll),
+        "{data_coll} was created again: {collections:?}"
+    );
+    assert!(
+        matches!(&built, Err(StorageError::TableNotFound(name)) if name == TABLE),
+        "expected TableNotFound({TABLE:?}), got {built:?}"
+    );
+
+    f.finish().await;
+}
+
+/// The engine's tag and TTL handlers against a `DELETING` document, through a
+/// `MongoEngine`. The key-info read reports such a document as not found on
+/// this backend, so the answers below depend on the handlers reading the
+/// status first. The tag rows match the service (measured 2026-10-25 for the
+/// table ARN, 2026-10-26 for the stream ARN), as do the `DescribeTimeToLive`
+/// rows for `DELETING` and `CREATING`. The enable runs after the data
+/// collection is dropped, as the delete call drops it before it removes the
+/// document, and leaves no collection behind.
+#[tokio::test]
+#[ignore = "needs a MongoDB replica set in EXTENDDB_TEST_MONGODB_URI"]
+async fn engine_tag_and_ttl_handlers_during_the_window() {
+    use extenddb_core::error::DynamoDbError;
+    use extenddb_core::types::Tag;
+    use extenddb_storage::MetadataEngine;
+    use serde_json::json;
+
+    let f = Fixture::new().await;
+    f.open_window().await;
+    let ctx = extenddb_engine::OperationContext {
+        storage: Arc::new(f.engine.clone()),
+        limits: Arc::new(extenddb_core::limits::LimitsConfig::default()),
+        region: Arc::from("us-east-1"),
+        account_id: Arc::from(ACCOUNT),
+        import_paths: Arc::from(Vec::new()),
+        export_paths: Arc::from(Vec::new()),
+        authorized_key_info: extenddb_engine::AuthorizedKeyInfo::default(),
+        auth_cache: extenddb_auth::AuthCacheRegistry::empty(),
+        table_key_info_lookup: None,
+    };
+    let arn = format!("arn:aws:dynamodb:us-east-1:{ACCOUNT}:table/{TABLE}");
+    // The tag path does not look the stream up, so any label of the
+    // service's shape names the table's stream.
+    let stream_arn = format!("{arn}/stream/2026-10-26T00:13:40.188");
+    let in_use = format!(
+        "Attempt to change a resource which is still in use: Table is being deleted: {TABLE}"
+    );
+    // A tag written before the delete.
+    f.engine
+        .tag_resource(
+            &arn,
+            &[Tag {
+                key: "env".to_owned(),
+                value: "test".to_owned(),
+            }],
+        )
+        .await
+        .expect("seed a tag");
+
+    let tag = extenddb_engine::handle_tag_resource(
+        json!({"ResourceArn": arn, "Tags": [{"Key": "added", "Value": "in-window"}]}),
+        &ctx,
+    )
+    .await;
+    assert!(
+        matches!(&tag, Err(DynamoDbError::ResourceInUseException(m)) if *m == in_use),
+        "TagResource: {tag:?}"
+    );
+    let untag = extenddb_engine::handle_untag_resource(
+        json!({"ResourceArn": arn, "TagKeys": ["env"]}),
+        &ctx,
+    )
+    .await;
+    assert!(
+        matches!(&untag, Err(DynamoDbError::ResourceInUseException(m)) if *m == in_use),
+        "UntagResource: {untag:?}"
+    );
+    let listed = extenddb_engine::handle_list_tags_of_resource(json!({"ResourceArn": arn}), &ctx)
+        .await
+        .expect("ListTagsOfResource is served");
+    assert_eq!(listed["Tags"], json!([{"Key": "env", "Value": "test"}]));
+
+    extenddb_engine::handle_tag_resource(
+        json!({"ResourceArn": stream_arn, "Tags": [{"Key": "sk", "Value": "sv"}]}),
+        &ctx,
+    )
+    .await
+    .expect("TagResource on the stream ARN is served");
+    let listed =
+        extenddb_engine::handle_list_tags_of_resource(json!({"ResourceArn": stream_arn}), &ctx)
+            .await
+            .expect("ListTagsOfResource on the stream ARN is served");
+    assert_eq!(listed["Tags"], json!([{"Key": "sk", "Value": "sv"}]));
+    extenddb_engine::handle_untag_resource(
+        json!({"ResourceArn": stream_arn, "TagKeys": ["sk"]}),
+        &ctx,
+    )
+    .await
+    .expect("UntagResource on the stream ARN is served");
+
+    let described =
+        extenddb_engine::handle_describe_time_to_live(json!({"TableName": TABLE}), &ctx).await;
+    assert!(
+        matches!(&described, Err(DynamoDbError::ValidationException(m))
+            if m == "Cannot describe time to live while table is in DELETING state: \
+                     Current table state is DELETING"),
+        "DescribeTimeToLive: {described:?}"
+    );
+
+    let data_coll = f.drop_data_collection().await;
+    let updated = extenddb_engine::handle_update_time_to_live(
+        json!({"TableName": TABLE,
+               "TimeToLiveSpecification": {"Enabled": true, "AttributeName": "ttl"}}),
+        &ctx,
+    )
+    .await
+    .expect("UpdateTimeToLive is served");
+    assert_eq!(
+        updated["TimeToLiveSpecification"],
+        json!({"Enabled": true, "AttributeName": "ttl"})
+    );
+    let collections = f.data_collections().await;
+    assert!(
+        !collections.contains(&data_coll),
+        "{data_coll} was created again: {collections:?}"
+    );
+    let ttl = f
+        .engine
+        .describe_ttl(ACCOUNT, TABLE)
+        .await
+        .expect("the TTL description");
+    assert_eq!(
+        ttl.attribute_name.as_deref(),
+        Some("ttl"),
+        "the catalog write landed"
+    );
+    assert_eq!(f.status().await.as_deref(), Some("DELETING"));
+
+    f.set_status("CREATING").await;
+    let described =
+        extenddb_engine::handle_describe_time_to_live(json!({"TableName": TABLE}), &ctx).await;
+    assert!(
+        matches!(&described, Err(DynamoDbError::ValidationException(m))
+            if m == "Cannot describe time to live while table is in CREATING state: \
+                     Current table state is CREATING"),
+        "DescribeTimeToLive on CREATING: {described:?}"
+    );
+
+    drop(ctx);
+    f.finish().await;
+}
diff --git a/crates/storage-mongodb/src/metadata_engine.rs b/crates/storage-mongodb/src/metadata_engine.rs
index 7fe21f9c3..95b2e3cbe 100644
--- a/crates/storage-mongodb/src/metadata_engine.rs
+++ b/crates/storage-mongodb/src/metadata_engine.rs
@@ -152,9 +152,14 @@ impl MetadataEngine for MongoEngine {
 
             let result = coll
                 .update_one(
+                    // DELETING is admitted beside ACTIVE: the service runs
+                    // UpdateTimeToLive's ordinary rules during the window and
+                    // the write lands on the catalog document the delete then
+                    // removes (measured 2026-10-25). The other non-ACTIVE
+                    // statuses keep their refusal.
                     doc! {
                         "_id": { "account_id": &account_id, "table_name": &table_name },
-                        "table_status": "ACTIVE",
+                        "table_status": { "$in": ["ACTIVE", "DELETING"] },
                     },
                     doc! {
                         "$set": {
@@ -347,8 +352,16 @@ impl MetadataEngine for MongoEngine {
             let tables_coll = self.catalog_db.collection::<Document>("tables");
             let id_filter =
                 doc! { "_id": { "account_id": &account_id, "table_name": &table_name } };
+            // ACTIVE only: a delete call drops the data collection after it
+            // marks the document DELETING, and create_index on a dropped
+            // collection creates it again. A build that reads the document
+            // after the mark gets TableNotFound, which the engine logs as
+            // deferred.
             let table_doc = tables_coll
-                .find_one(id_filter.clone())
+                .find_one(doc! {
+                    "_id": { "account_id": &account_id, "table_name": &table_name },
+                    "table_status": "ACTIVE",
+                })
                 .await
                 .map_err(|e| StorageError::Internal(e.to_string()))?
                 .ok_or_else(|| StorageError::TableNotFound(table_name.clone()))?;
diff --git a/crates/storage-postgres/Cargo.toml b/crates/storage-postgres/Cargo.toml
index 604a31176..44c5096c6 100755
--- a/crates/storage-postgres/Cargo.toml
+++ b/crates/storage-postgres/Cargo.toml
@@ -11,6 +11,9 @@ license.workspace = true
 # The storage-level vector control-plane tests drive the engine directly against
 # a scratch database, so they need a runtime and the test macro.
 tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }
+# The DELETING-window TTL rows call the engine's UpdateTimeToLive handler over a
+# PostgresEngine, to see which index statements it sends for such a table.
+extenddb-engine = { workspace = true }
 
 [dependencies]
 anyhow = { workspace = true }
diff --git a/crates/storage-postgres/src/metadata_engine.rs b/crates/storage-postgres/src/metadata_engine.rs
index b2f958b3b..312480d27 100755
--- a/crates/storage-postgres/src/metadata_engine.rs
+++ b/crates/storage-postgres/src/metadata_engine.rs
@@ -77,9 +77,14 @@ impl MetadataEngine for PostgresEngine {
             let ttl_val: Option<&str> = if enabled { Some(&attribute_name) } else { None };
             let index_ready = false;
 
+            // DELETING is admitted beside ACTIVE: the service runs
+            // UpdateTimeToLive's ordinary rules during the window and the write
+            // lands on the catalog row the delete then removes (measured
+            // 2026-10-25). The other non-ACTIVE statuses keep their refusal.
             let result = sqlx::query(
                 "UPDATE tables SET ttl_attribute = $1, ttl_index_ready = $4 \
-                 WHERE account_id = $2 AND table_name = $3 AND table_status = 'ACTIVE'",
+                 WHERE account_id = $2 AND table_name = $3 \
+                 AND table_status IN ('ACTIVE', 'DELETING')",
             )
             .bind(ttl_val)
             .bind(&account_id)
diff --git a/crates/storage-postgres/tests/deleting_window.rs b/crates/storage-postgres/tests/deleting_window.rs
index a53c43865..5f1ae02aa 100644
--- a/crates/storage-postgres/tests/deleting_window.rs
+++ b/crates/storage-postgres/tests/deleting_window.rs
@@ -41,6 +41,9 @@ use extenddb_storage_postgres::{PostgresConfig, PostgresEngine};
 use sqlx::PgPool;
 use sqlx::postgres::PgPoolOptions;
 
+#[path = "deleting_window/ttl.rs"]
+mod ttl;
+
 const ACCOUNT: &str = "123456789012";
 const REGION: &str = "us-east-1";
 const TABLE: &str = "t_deleting_window";
diff --git a/crates/storage-postgres/tests/deleting_window/ttl.rs b/crates/storage-postgres/tests/deleting_window/ttl.rs
new file mode 100644
index 000000000..47d369a09
--- /dev/null
+++ b/crates/storage-postgres/tests/deleting_window/ttl.rs
@@ -0,0 +1,180 @@
+// Copyright 2026 ExtendDB contributors
+// SPDX-License-Identifier: Apache-2.0
+
+//! `UpdateTimeToLive` while a table is `DELETING`, on the window the parent
+//! module opens. Amazon DynamoDB runs the ordinary TTL rules on a `DELETING`
+//! table (measured 2026-10-25), so the catalog write is admitted beside
+//! `ACTIVE` and refused for the other statuses. The delete removes the data
+//! table with its index, so the engine sends no index DDL for such a table.
+
+use std::sync::Arc;
+
+use extenddb_storage::MetadataEngine;
+use extenddb_storage::error::StorageError;
+use serde_json::json;
+
+use super::{
+    ACCOUNT, Scratch, TABLE, base_conn, create_table_named, open_window, scratch, skip,
+    start_deleting, status_of,
+};
+
+/// UpdateTimeToLive during the window writes the catalog row, the way the
+/// service runs its ordinary TTL rules there (measured 2026-10-25), and leaves
+/// the status alone. The disable direction is served too.
+#[tokio::test]
+async fn update_ttl_during_the_window_is_served() {
+    if base_conn().is_none() {
+        return skip("update_ttl_during_the_window_is_served");
+    }
+    let s = scratch().await;
+    open_window(&s).await;
+
+    s.engine
+        .update_ttl(ACCOUNT, TABLE, "ttl", true)
+        .await
+        .expect("UpdateTimeToLive on a DELETING table");
+    let attr: Option<String> = sqlx::query_scalar(
+        "SELECT ttl_attribute FROM tables WHERE account_id = $1 AND table_name = $2",
+    )
+    .bind(ACCOUNT)
+    .bind(TABLE)
+    .fetch_one(&s.catalog)
+    .await
+    .expect("read the TTL attribute");
+    assert_eq!(attr.as_deref(), Some("ttl"));
+    assert_eq!(
+        status_of(&s.catalog, TABLE).await.as_deref(),
+        Some("DELETING"),
+        "the write must not change the status"
+    );
+
+    s.engine
+        .update_ttl(ACCOUNT, TABLE, "ttl", false)
+        .await
+        .expect("the disable direction during the window");
+    let attr: Option<String> = sqlx::query_scalar(
+        "SELECT ttl_attribute FROM tables WHERE account_id = $1 AND table_name = $2",
+    )
+    .bind(ACCOUNT)
+    .bind(TABLE)
+    .fetch_one(&s.catalog)
+    .await
+    .expect("read the TTL attribute");
+    assert_eq!(attr, None);
+
+    s.cleanup().await;
+}
+
+/// A CREATING table keeps the TTL refusal: the window admits DELETING only,
+/// and a table with no row at all is still reported as absent.
+#[tokio::test]
+async fn update_ttl_outside_active_and_deleting_keeps_its_refusal() {
+    if base_conn().is_none() {
+        return skip("update_ttl_outside_active_and_deleting_keeps_its_refusal");
+    }
+    let s = scratch().await;
+    open_window(&s).await;
+    sqlx::query(
+        "UPDATE tables SET table_status = 'CREATING' WHERE account_id = $1 AND table_name = $2",
+    )
+    .bind(ACCOUNT)
+    .bind(TABLE)
+    .execute(&s.catalog)
+    .await
+    .expect("mark CREATING");
+
+    let err = s
+        .engine
+        .update_ttl(ACCOUNT, TABLE, "ttl", true)
+        .await
+        .expect_err("UpdateTimeToLive on a CREATING table");
+    assert!(
+        matches!(&err, StorageError::TableNotActive(name) if name == TABLE),
+        "expected TableNotActive({TABLE:?}), got {err:?}"
+    );
+
+    let err = s
+        .engine
+        .update_ttl(ACCOUNT, "t_absent", "ttl", true)
+        .await
+        .expect_err("UpdateTimeToLive on a missing table");
+    assert!(
+        matches!(&err, StorageError::TableNotFound(name) if name == "t_absent"),
+        "expected TableNotFound(\"t_absent\"), got {err:?}"
+    );
+
+    s.cleanup().await;
+}
+
+/// The TTL index names in the scratch data database, which holds one table.
+async fn ttl_indexes(s: &Scratch) -> Vec<String> {
+    sqlx::query_scalar("SELECT indexname FROM pg_indexes WHERE indexname LIKE 'idx\\_ttl\\_%'")
+        .fetch_all(s.engine.data_pool())
+        .await
+        .expect("list the TTL indexes")
+}
+
+/// UpdateTimeToLive disabling TTL during the window, through the engine
+/// handler, sends no DROP INDEX: the index built while the table was ACTIVE is
+/// still there when the call returns, the catalog write landed, and the row is
+/// still DELETING.
+#[tokio::test]
+async fn update_ttl_disable_during_the_window_sends_no_drop_index() {
+    if base_conn().is_none() {
+        return skip("update_ttl_disable_during_the_window_sends_no_drop_index");
+    }
+    let s = scratch().await;
+    create_table_named(&s, TABLE).await;
+    s.engine
+        .update_ttl(ACCOUNT, TABLE, "ttl", true)
+        .await
+        .expect("enable TTL while ACTIVE");
+    s.engine
+        .create_ttl_index(ACCOUNT, TABLE, "ttl")
+        .await
+        .expect("build the TTL index while ACTIVE");
+    let built = ttl_indexes(&s).await;
+    assert_eq!(built.len(), 1, "one TTL index before the delete: {built:?}");
+    start_deleting(&s).await;
+
+    let ctx = extenddb_engine::OperationContext {
+        storage: Arc::new(s.engine.clone()),
+        limits: Arc::new(extenddb_core::limits::LimitsConfig::default()),
+        region: Arc::from("us-east-1"),
+        account_id: Arc::from(ACCOUNT),
+        import_paths: Arc::from(Vec::new()),
+        export_paths: Arc::from(Vec::new()),
+        authorized_key_info: extenddb_engine::AuthorizedKeyInfo::default(),
+        auth_cache: extenddb_auth::AuthCacheRegistry::empty(),
+        table_key_info_lookup: None,
+    };
+    let out = extenddb_engine::handle_update_time_to_live(
+        json!({"TableName": TABLE,
+               "TimeToLiveSpecification": {"Enabled": false, "AttributeName": "ttl"}}),
+        &ctx,
+    )
+    .await
+    .expect("UpdateTimeToLive disable during the window");
+    assert_eq!(
+        out["TimeToLiveSpecification"],
+        json!({"Enabled": false, "AttributeName": "ttl"})
+    );
+    drop(ctx);
+
+    assert_eq!(ttl_indexes(&s).await, built, "no DROP INDEX was sent");
+    let attr: Option<String> = sqlx::query_scalar(
+        "SELECT ttl_attribute FROM tables WHERE account_id = $1 AND table_name = $2",
+    )
+    .bind(ACCOUNT)
+    .bind(TABLE)
+    .fetch_one(&s.catalog)
+    .await
+    .expect("read the TTL attribute");
+    assert_eq!(attr, None, "the catalog write landed");
+    assert_eq!(
+        status_of(&s.catalog, TABLE).await.as_deref(),
+        Some("DELETING")
+    );
+
+    s.cleanup().await;
+}
diff --git a/crates/storage-sqlite/src/delete_table.rs b/crates/storage-sqlite/src/delete_table.rs
index a62a4a4e1..02520b3f0 100644
--- a/crates/storage-sqlite/src/delete_table.rs
+++ b/crates/storage-sqlite/src/delete_table.rs
@@ -222,9 +222,9 @@ impl SqliteEngine {
 #[cfg(test)]
 mod deleting_window_tests {
     use crate::SqliteEngine;
-    use extenddb_core::types::{DeleteTableInput, UpdateTableInput};
-    use extenddb_storage::TableEngine;
+    use extenddb_core::types::{DeleteTableInput, TimeToLiveStatus, UpdateTableInput};
     use extenddb_storage::error::StorageError;
+    use extenddb_storage::{MetadataEngine, TableEngine};
     use serde_json::json;
 
     const ACCOUNT: &str = "000000000000";
@@ -332,4 +332,80 @@ mod deleting_window_tests {
             "expected TableNotActive(\"t\"), got {err:?}"
         );
     }
+
+    /// UpdateTimeToLive during the window writes the catalog row, the way the
+    /// service runs its ordinary TTL rules there (measured 2026-10-25).
+    #[tokio::test]
+    async fn update_ttl_during_the_window_is_served() {
+        let engine = engine_with_deleting_table().await;
+        engine
+            .update_ttl(ACCOUNT, "t", "ttl", true)
+            .await
+            .expect("UpdateTimeToLive on a DELETING table");
+        let desc = engine
+            .describe_ttl(ACCOUNT, "t")
+            .await
+            .expect("the TTL description");
+        assert_eq!(desc.time_to_live_status, TimeToLiveStatus::Enabled);
+        assert_eq!(desc.attribute_name.as_deref(), Some("ttl"));
+        let status: String =
+            sqlx::query_scalar("SELECT table_status FROM tables WHERE table_name = 't'")
+                .fetch_one(&engine.pool)
+                .await
+                .expect("row still present");
+        assert_eq!(status, "DELETING", "the write must not change the status");
+    }
+
+    /// The disable direction is served in the window too, on the same admitted
+    /// status.
+    #[tokio::test]
+    async fn update_ttl_disable_during_the_window_is_served() {
+        let engine = engine_with_deleting_table().await;
+        engine
+            .update_ttl(ACCOUNT, "t", "ttl", true)
+            .await
+            .expect("the enable");
+        engine
+            .update_ttl(ACCOUNT, "t", "ttl", false)
+            .await
+            .expect("the disable");
+        let desc = engine
+            .describe_ttl(ACCOUNT, "t")
+            .await
+            .expect("the TTL description");
+        assert_eq!(desc.time_to_live_status, TimeToLiveStatus::Disabled);
+        assert_eq!(desc.attribute_name, None);
+    }
+
+    /// A CREATING table keeps the TTL refusal: the window admits DELETING only.
+    #[tokio::test]
+    async fn update_ttl_on_a_creating_table_keeps_table_not_active() {
+        let engine = engine_with_deleting_table().await;
+        sqlx::query("UPDATE tables SET table_status = 'CREATING' WHERE table_name = 't'")
+            .execute(&engine.pool)
+            .await
+            .expect("mark CREATING");
+        let err = engine
+            .update_ttl(ACCOUNT, "t", "ttl", true)
+            .await
+            .expect_err("UpdateTimeToLive on a CREATING table");
+        assert!(
+            matches!(&err, StorageError::TableNotActive(name) if name == "t"),
+            "expected TableNotActive(\"t\"), got {err:?}"
+        );
+    }
+
+    /// A table with no catalog row at all is still reported as absent.
+    #[tokio::test]
+    async fn update_ttl_on_a_missing_table_is_table_not_found() {
+        let engine = engine_with_deleting_table().await;
+        let err = engine
+            .update_ttl(ACCOUNT, "absent", "ttl", true)
+            .await
+            .expect_err("UpdateTimeToLive on a missing table");
+        assert!(
+            matches!(&err, StorageError::TableNotFound(name) if name == "absent"),
+            "expected TableNotFound(\"absent\"), got {err:?}"
+        );
+    }
 }
diff --git a/crates/storage-sqlite/src/metadata.rs b/crates/storage-sqlite/src/metadata.rs
index 93c319873..77ec72eca 100644
--- a/crates/storage-sqlite/src/metadata.rs
+++ b/crates/storage-sqlite/src/metadata.rs
@@ -93,9 +93,14 @@ impl MetadataEngine for SqliteEngine {
             // D1: every writer holds the engine write lock.
             let _writer = self.write_lock.lock().await;
             let ttl_val: Option<&str> = if enabled { Some(&attribute_name) } else { None };
+            // DELETING is admitted beside ACTIVE: the service runs
+            // UpdateTimeToLive's ordinary rules during the window and the write
+            // lands on the catalog row the delete then removes (measured
+            // 2026-10-25). The other non-ACTIVE statuses keep their refusal.
             let result = sqlx::query(
                 "UPDATE tables SET ttl_attribute = ?, ttl_index_ready = 0 \
-                 WHERE account_id = ? AND table_name = ? AND table_status = 'ACTIVE'",
+                 WHERE account_id = ? AND table_name = ? \
+                 AND table_status IN ('ACTIVE', 'DELETING')",
             )
             .bind(ttl_val)
             .bind(&account_id)
diff --git a/devtools/run-mongodb-tests b/devtools/run-mongodb-tests
index 14ecb9867..06a595c54 100755
--- a/devtools/run-mongodb-tests
+++ b/devtools/run-mongodb-tests
@@ -249,6 +249,16 @@ fi
 # ---------------------------------------------------------------------------
 
 echo ""
+if [[ " ${RUN_TESTS_ARGS[*]} " == *" --rust"* ]]; then
+    # The storage-level MongoDB rows for a table's DELETING window need a live
+    # replica set, so they are #[ignore]d in the crate and read their server
+    # from this variable; this container is the one place in CI that has one.
+    echo "=== Running MongoDB storage-level live tests (deleting_window) ==="
+    echo ""
+    EXTENDDB_TEST_MONGODB_URI="mongodb://localhost:$MONGO_PORT/?replicaSet=rs0&directConnection=true" \
+        cargo test -p extenddb-storage-mongodb --lib -- --ignored deleting_window
+    echo ""
+fi
 echo "=== Running test suite via devtools/run-tests --backend mongodb ==="
 echo "  passthrough args: ${RUN_TESTS_ARGS[*]}"
 echo ""
diff --git a/tests/python/test_deleting_window_control_plane.py b/tests/python/test_deleting_window_control_plane.py
index b6837856f..37395fe96 100644
--- a/tests/python/test_deleting_window_control_plane.py
+++ b/tests/python/test_deleting_window_control_plane.py
@@ -11,6 +11,14 @@
 ``Scan`` are served; once the table is gone, ``GetItem`` of an item written in
 the window is ``ResourceNotFoundException`` (measured 2026-10-25).
 
+``TagResource`` and ``UntagResource`` on the table ARN answer that same
+sentence in the window and ``ListTagsOfResource`` is served with the tags; on
+the table's stream ARN all three are served (measured 2026-10-26).
+``DescribeTimeToLive`` answers ``ValidationException`` ``Cannot describe time
+to live while table is in DELETING state: Current table state is DELETING``,
+and ``UpdateTimeToLive`` is served and answers with the specification it wrote
+(both measured 2026-10-25).
+
 Against the service these tests run on its natural window. Against ExtendDB on
 PostgreSQL the window is held open the way a slow reader would hold it: an
 outside transaction keeps ``ACCESS SHARE`` on the table's data table, so the
@@ -32,6 +40,7 @@
 from __future__ import annotations
 
 import os
+import time
 
 import pytest
 from botocore.exceptions import ClientError
@@ -96,13 +105,22 @@ def _skip_unless_window_can_be_held(dynamodb_client, name: str) -> None:
         pytest.skip(reason)
 
 
-def _open_window(dynamodb_client, is_extenddb: bool, name: str) -> _HeldLock | None:
-    """Create the table and, on ExtendDB, hold the lock that keeps it DELETING."""
+def _open_window(
+    dynamodb_client, is_extenddb: bool, name: str, stream: bool = False
+) -> _HeldLock | None:
+    """Create the table, with a stream when ``stream`` is set, and on ExtendDB
+    hold the lock that keeps it DELETING."""
+    extra = (
+        {"StreamSpecification": {"StreamEnabled": True, "StreamViewType": "KEYS_ONLY"}}
+        if stream
+        else {}
+    )
     dynamodb_client.create_table(
         TableName=name,
         AttributeDefinitions=[{"AttributeName": "pk", "AttributeType": "S"}],
         KeySchema=[{"AttributeName": "pk", "KeyType": "HASH"}],
         BillingMode="PAY_PER_REQUEST",
+        **extra,
     )
     wait_for_active(dynamodb_client, name)
     if not is_extenddb:
@@ -118,6 +136,29 @@ def _assert_in_use(exc: ClientError, name: str) -> None:
     assert err["Message"] == _being_deleted(name)
 
 
+def _settle_tags(dynamodb_client, is_extenddb: bool, arn: str) -> None:
+    """On the service, wait out the tag update a tag write starts.
+
+    For several seconds after ``TagResource`` or ``UntagResource`` the service
+    answers another tag write on the resource with ``LimitExceededException``
+    and ``DeleteTable`` on the table with ``ResourceInUseException``, both
+    ``Table tags are being updated: <name>`` (measured 2026-10-26). An
+    ``UntagResource`` of a key the resource does not carry is retried until it
+    is answered. ExtendDB has no such state.
+    """
+    if is_extenddb:
+        return
+    for _ in range(30):
+        try:
+            dynamodb_client.untag_resource(ResourceArn=arn, TagKeys=["settle-probe"])
+            return
+        except ClientError as exc:
+            if "tags are being updated" not in exc.response["Error"]["Message"]:
+                raise
+            time.sleep(1)
+    raise TimeoutError(f"the tags of {arn} were still being updated after 30 attempts")
+
+
 class TestDeletingWindowControlPlane:
     def test_second_delete_table_is_being_deleted(self, dynamodb_client, is_extenddb):
         """A second DeleteTable while the first is still running is ResourceInUseException."""
@@ -187,3 +228,132 @@ def test_data_plane_is_served_until_the_table_is_gone(self, dynamodb_client, is_
         with pytest.raises(ClientError) as exc:
             dynamodb_client.get_item(TableName=name, Key={"pk": {"S": "w1"}}, ConsistentRead=True)
         assert exc.value.response["Error"]["Code"] == "ResourceNotFoundException"
+
+
+class TestDeletingWindowTimeToLive:
+    """``DescribeTimeToLive`` is refused during the window and
+    ``UpdateTimeToLive`` runs its ordinary rules (measured 2026-10-25 on three
+    tables, every call issued while ``DescribeTable`` read ``DELETING``).
+    ``DescribeTimeToLive`` on a ``CREATING`` table is refused with the same
+    sentence naming ``CREATING`` (measured 2026-10-26); the fixture cannot hold
+    a table ``CREATING``, so that row is pinned in the engine's unit tests."""
+
+    def test_describe_time_to_live_is_refused(self, dynamodb_client, is_extenddb):
+        name = unique_name("del-window")
+        lock = _open_window(dynamodb_client, is_extenddb, name)
+        try:
+            dynamodb_client.delete_table(TableName=name)
+            with pytest.raises(ClientError) as exc:
+                dynamodb_client.describe_time_to_live(TableName=name)
+            err = exc.value.response["Error"]
+            assert err["Code"] == "ValidationException"
+            assert err["Message"] == (
+                "Cannot describe time to live while table is in DELETING state: "
+                "Current table state is DELETING"
+            )
+        finally:
+            if lock is not None:
+                lock.release()
+        wait_for_deleted(dynamodb_client, name, timeout=120.0)
+
+    def test_update_time_to_live_is_served(self, dynamodb_client, is_extenddb):
+        """The enable is accepted and answers with the specification it wrote;
+        DescribeTimeToLive stays refused after it."""
+        name = unique_name("del-window")
+        lock = _open_window(dynamodb_client, is_extenddb, name)
+        try:
+            dynamodb_client.delete_table(TableName=name)
+            resp = dynamodb_client.update_time_to_live(
+                TableName=name,
+                TimeToLiveSpecification={"Enabled": True, "AttributeName": "ttl"},
+            )
+            assert resp["TimeToLiveSpecification"] == {
+                "Enabled": True,
+                "AttributeName": "ttl",
+            }
+            with pytest.raises(ClientError) as exc:
+                dynamodb_client.describe_time_to_live(TableName=name)
+            assert exc.value.response["Error"]["Code"] == "ValidationException"
+        finally:
+            if lock is not None:
+                lock.release()
+        wait_for_deleted(dynamodb_client, name, timeout=120.0)
+
+
+class TestDeletingWindowTags:
+    """``TagResource`` and ``UntagResource`` on the table ARN are refused during
+    the window and ``ListTagsOfResource`` is served with the tags (measured
+    2026-10-25). The table's stream ARN is a resource of its own: both writes
+    and the read are served on it during the window (measured 2026-10-26)."""
+
+    def test_tag_calls_during_the_window(self, dynamodb_client, is_extenddb):
+        name = unique_name("del-window")
+        lock = _open_window(dynamodb_client, is_extenddb, name)
+        try:
+            arn = dynamodb_client.describe_table(TableName=name)["Table"]["TableArn"]
+            dynamodb_client.tag_resource(
+                ResourceArn=arn, Tags=[{"Key": "env", "Value": "test"}]
+            )
+            _settle_tags(dynamodb_client, is_extenddb, arn)
+            dynamodb_client.delete_table(TableName=name)
+
+            with pytest.raises(ClientError) as exc:
+                dynamodb_client.tag_resource(
+                    ResourceArn=arn, Tags=[{"Key": "added", "Value": "in-window"}]
+                )
+            _assert_in_use(exc.value, name)
+
+            with pytest.raises(ClientError) as exc:
+                dynamodb_client.untag_resource(ResourceArn=arn, TagKeys=["env"])
+            _assert_in_use(exc.value, name)
+
+            listed = dynamodb_client.list_tags_of_resource(ResourceArn=arn)
+            assert listed["Tags"] == [{"Key": "env", "Value": "test"}]
+        finally:
+            if lock is not None:
+                lock.release()
+        wait_for_deleted(dynamodb_client, name, timeout=120.0)
+
+    def test_stream_arn_tag_resource_is_served(self, dynamodb_client, is_extenddb):
+        """TagResource on the stream ARN, the first tag write after the delete,
+        answers 200 with an empty body."""
+        name = unique_name("del-window")
+        lock = _open_window(dynamodb_client, is_extenddb, name, stream=True)
+        try:
+            stream_arn = dynamodb_client.describe_table(TableName=name)["Table"]["LatestStreamArn"]
+            dynamodb_client.delete_table(TableName=name)
+
+            resp = dynamodb_client.tag_resource(
+                ResourceArn=stream_arn, Tags=[{"Key": "sk", "Value": "sv"}]
+            )
+            resp.pop("ResponseMetadata", None)
+            assert resp == {}
+        finally:
+            if lock is not None:
+                lock.release()
+        wait_for_deleted(dynamodb_client, name, timeout=120.0)
+
+    def test_stream_arn_untag_resource_is_served(self, dynamodb_client, is_extenddb):
+        """ListTagsOfResource on the stream ARN answers with the tags written
+        before the delete, and UntagResource on it, the first tag write after
+        the delete, answers 200 with an empty body."""
+        name = unique_name("del-window")
+        lock = _open_window(dynamodb_client, is_extenddb, name, stream=True)
+        try:
+            stream_arn = dynamodb_client.describe_table(TableName=name)["Table"]["LatestStreamArn"]
+            dynamodb_client.tag_resource(
+                ResourceArn=stream_arn, Tags=[{"Key": "sk", "Value": "sv"}]
+            )
+            _settle_tags(dynamodb_client, is_extenddb, stream_arn)
+            dynamodb_client.delete_table(TableName=name)
+
+            listed = dynamodb_client.list_tags_of_resource(ResourceArn=stream_arn)
+            assert listed["Tags"] == [{"Key": "sk", "Value": "sv"}]
+
+            resp = dynamodb_client.untag_resource(ResourceArn=stream_arn, TagKeys=["sk"])
+            resp.pop("ResponseMetadata", None)
+            assert resp == {}
+        finally:
+            if lock is not None:
+                lock.release()
+        wait_for_deleted(dynamodb_client, name, timeout=120.0)