diff --git a/.github/workflows/integration.yml b/.github/workflows/integration.yml index 8a4818f7..e5efba1c 100644 --- a/.github/workflows/integration.yml +++ b/.github/workflows/integration.yml @@ -300,17 +300,18 @@ jobs: # storage layer directly against this job's PostgreSQL. The PutItem create # race tests hold a create open in an outside transaction, the # write-transaction conflict tests force a deadlock, and the backup restore - # cases need catalog rows the current binary would not write; all need - # direct access. The online GSI build tests reorder queue rows by hand and - # pause writers with a test hook. They build their own throwaway databases; - # the connection string is the server, not a database. + # cases need catalog rows the current binary would not write, and the + # DELETING window tests hold a table DELETING with no poller to finish it; + # all need direct access. The online GSI build tests reorder queue rows by + # hand and pause writers with a test hook. They build their own throwaway + # databases; the connection string is the server, not a database. - name: Run PostgreSQL storage-level tests env: EXTENDDB_TEST_PG_CONNECTION_STRING: postgresql://postgres:devpass@127.0.0.1:5432 run: >- cargo test --release -p extenddb-storage-postgres --test vector_control_plane --test put_create_race --test twi_conflict - --test stream_shard_retention --test backup_restore + --test stream_shard_retention --test backup_restore --test deleting_window --test gsi_online_build --test gsi_stale_writer # The daemonized server logs to syslog; dump it so server-side failures diff --git a/crates/core/src/error/messages.rs b/crates/core/src/error/messages.rs index 18d3e521..e5aedc05 100755 --- a/crates/core/src/error/messages.rs +++ b/crates/core/src/error/messages.rs @@ -40,6 +40,10 @@ pub enum ErrorMessageKey { TableNotFound, TableAlreadyExists, TableInUse, + /// A control-plane change refused because the table is `DELETING`. + /// Measured on the service 2026-10-24 for a second `DeleteTable` and for + /// `UpdateTable` during the window. + TableBeingDeleted, MissingKeySchema, KeySchemaTooMany, KeySchemaFirstNotHash, @@ -82,6 +86,12 @@ pub fn error_message(key: ErrorMessageKey, args: &[&str]) -> String { ErrorMessageKey::TableInUse => { format!("Table is being created, updated, or deleted: {}", args[0]) } + ErrorMessageKey::TableBeingDeleted => { + format!( + "Attempt to change a resource which is still in use: Table is being deleted: {}", + args[0] + ) + } 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/create_table.rs b/crates/engine/src/create_table.rs index 424e27b5..e3bad2a3 100755 --- a/crates/engine/src/create_table.rs +++ b/crates/engine/src/create_table.rs @@ -132,6 +132,9 @@ pub fn storage_err_to_dynamo(e: extenddb_storage::error::StorageError) -> Dynamo ErrorMessageKey::TableInUse, &[&name], )), + StorageError::TableBeingDeleted(name) => DynamoDbError::ResourceInUseException( + error_message(ErrorMessageKey::TableBeingDeleted, &[&name]), + ), // Only RestoreTableFromBackup raises this; its own mapping makes it // TableInUseException. Elsewhere (a plain CreateTable into a name // that is a CREATING restore target) the service's text is the one diff --git a/crates/engine/src/delete_table.rs b/crates/engine/src/delete_table.rs index 4c48e423..022c3879 100755 --- a/crates/engine/src/delete_table.rs +++ b/crates/engine/src/delete_table.rs @@ -45,3 +45,28 @@ pub async fn handle_delete_table( }; serialize_output(&output) } + +#[cfg(test)] +mod tests { + use super::*; + use extenddb_storage::error::StorageError; + + /// A second DeleteTable during the DELETING window: the measured sentence + /// (2026-10-24), byte for byte. The other non-ACTIVE statuses keep theirs. + #[test] + fn a_deleting_table_has_its_own_sentence() { + match storage_err_to_dynamo(StorageError::TableBeingDeleted("t".to_owned())) { + 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:?}"), + } + match storage_err_to_dynamo(StorageError::TableNotActive("t".to_owned())) { + DynamoDbError::ResourceInUseException(msg) => { + assert_eq!(msg, "Table is being created, updated, or deleted: t"); + } + other => panic!("expected ResourceInUseException, got {other:?}"), + } + } +} diff --git a/crates/engine/src/ttl.rs b/crates/engine/src/ttl.rs index fac027d4..ed1e509b 100755 --- a/crates/engine/src/ttl.rs +++ b/crates/engine/src/ttl.rs @@ -226,6 +226,15 @@ 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. + StorageError::TableBeingDeleted(name) => { + DynamoDbError::ResourceInUseException(extenddb_core::error::error_message( + extenddb_core::error::ErrorMessageKey::TableBeingDeleted, + &[&name], + )) + } other => { tracing::error!(internal_error = %other, "storage internal error"); DynamoDbError::InternalServerError("Internal server error".to_owned()) @@ -237,6 +246,25 @@ fn storage_to_dynamo(e: StorageError) -> DynamoDbError { mod tests { use super::*; + /// The TTL mapper renders a DELETING table with the service's sentence and + /// keeps the other non-ACTIVE statuses on theirs. + #[test] + fn a_deleting_table_has_its_own_sentence() { + match storage_to_dynamo(StorageError::TableBeingDeleted("t".to_owned())) { + 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:?}"), + } + match storage_to_dynamo(StorageError::TableNotActive("t".to_owned())) { + DynamoDbError::ResourceInUseException(msg) => { + assert_eq!(msg, "Table t is not in ACTIVE state"); + } + other => panic!("expected ResourceInUseException, got {other:?}"), + } + } + #[test] fn valid_ttl_attribute_names() { assert!(validate_ttl_attribute_name("ttl").is_ok()); diff --git a/crates/engine/src/update_table.rs b/crates/engine/src/update_table.rs index 6db763f8..6cba5c61 100755 --- a/crates/engine/src/update_table.rs +++ b/crates/engine/src/update_table.rs @@ -321,6 +321,15 @@ fn update_table_err_to_dynamo( StorageError::TableNotActive(name) => { DynamoDbError::ResourceInUseException(format!("Table {name} is not in ACTIVE state")) } + // The service words a DELETING table differently from the other + // non-ACTIVE statuses (measured 2026-10-24); the sentence lives with + // the engine's other message templates. + StorageError::TableBeingDeleted(name) => { + DynamoDbError::ResourceInUseException(extenddb_core::error::error_message( + extenddb_core::error::ErrorMessageKey::TableBeingDeleted, + &[&name], + )) + } StorageError::IndexAlreadyExists(name) => DynamoDbError::ValidationException(format!( "One or more parameter values were invalid: Index already exists: {name}" )), @@ -500,4 +509,25 @@ mod tests { update_table_err_to_dynamo(StorageError::Internal("disk on fire".to_owned()), "t"); assert!(matches!(err, DynamoDbError::InternalServerError(_))); } + + /// UpdateTable on a DELETING table: the measured sentence (2026-10-24), + /// byte for byte, and the other non-ACTIVE statuses keep theirs. + #[test] + fn a_deleting_table_has_its_own_sentence() { + let err = update_table_err_to_dynamo(StorageError::TableBeingDeleted("t".to_owned()), "t"); + match 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:?}"), + } + let err = update_table_err_to_dynamo(StorageError::TableNotActive("t".to_owned()), "t"); + match err { + DynamoDbError::ResourceInUseException(msg) => { + assert_eq!(msg, "Table t is not in ACTIVE state"); + } + other => panic!("expected ResourceInUseException, got {other:?}"), + } + } } diff --git a/crates/server/src/key_info_cache.rs b/crates/server/src/key_info_cache.rs index ba527f65..ddd42040 100644 --- a/crates/server/src/key_info_cache.rs +++ b/crates/server/src/key_info_cache.rs @@ -160,10 +160,11 @@ impl CachedTableKeyInfoStore { /// cached and is what the caller sees. /// /// Every backend's `table_key_info` answers `TableNotFound` both for a name -/// the catalog does not have and for one it has in a state other than -/// `ACTIVE` (a `CreateTable` or `RestoreTableFromBackup` target that is still -/// `CREATING`; the data plane reports both as `ResourceNotFoundException`, as -/// the service does). Caching the second as a negative entry made reads on a +/// the catalog does not have and for one it has as `CREATING` (a `CreateTable` +/// or `RestoreTableFromBackup` target; the data plane reports both as +/// `ResourceNotFoundException`, as the service does). PostgreSQL and SQLite +/// answer the key info for a `DELETING` row, since the service serves the data +/// plane until the table is gone. Caching the second as a negative entry made reads on a /// restored table fail for up to `negative_ttl` after `DescribeTable` reported /// it `ACTIVE`: a probe during the copy stored the entry, and nothing /// invalidated it at the transition (the service's first diff --git a/crates/storage-cassandra/src/delete_table.rs b/crates/storage-cassandra/src/delete_table.rs index b1a5b426..479bb86c 100644 --- a/crates/storage-cassandra/src/delete_table.rs +++ b/crates/storage-cassandra/src/delete_table.rs @@ -79,6 +79,17 @@ impl CassandraEngine { match caller { DeleteTableCaller::Client => { + // A DELETING row belongs to the deletion already under way + // (the abandoned-restore sweep sets it on this backend). The + // service refuses a second DeleteTable in that window as a + // table still in use (measured 2026-10-24); checked before + // deletion protection, as the SQL backends order it. + let status: String = + crate::cassandra_util::get_column(table_row, "table_status", "delete_table")?; + if status == "DELETING" { + return Err(StorageError::TableBeingDeleted(input.table_name.clone())); + } + // Check deletion protection let deletion_protection: bool = table_row .get_r_by_name("deletion_protection_enabled") @@ -94,12 +105,11 @@ impl CassandraEngine { // A restore target is owned by its copy until the copy // activates it (or the sweep removes it). Deleting it mid-copy // would race the copy's writes into dropped tables. - let status: Option = table_row.get_by_name("table_status").ok().flatten(); let restoring_from: Option = table_row .get_by_name("restore_source_backup_arn") .ok() .flatten(); - if status.as_deref() == Some("CREATING") && restoring_from.is_some() { + if status == "CREATING" && restoring_from.is_some() { return Err(table_being_restored(&input.table_name)); } } diff --git a/crates/storage-cassandra/src/update_table.rs b/crates/storage-cassandra/src/update_table.rs index 4a20b861..334f34eb 100644 --- a/crates/storage-cassandra/src/update_table.rs +++ b/crates/storage-cassandra/src/update_table.rs @@ -41,6 +41,11 @@ impl CassandraEngine { let status: String = crate::cassandra_util::get_column(&row, "table_status", "update_table")?; + // A DELETING table has its own sentence on the service (measured + // 2026-10-24); every other non-ACTIVE status keeps the existing one. + if status == "DELETING" { + return Err(StorageError::TableBeingDeleted(input.table_name.clone())); + } if status != "ACTIVE" { return Err(StorageError::TableNotActive(input.table_name.clone())); } diff --git a/crates/storage-cassandra/tests/direct/deleting_window.rs b/crates/storage-cassandra/tests/direct/deleting_window.rs new file mode 100644 index 00000000..cc43f7da --- /dev/null +++ b/crates/storage-cassandra/tests/direct/deleting_window.rs @@ -0,0 +1,176 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Direct tests for what the control plane answers while a table is +//! `DELETING`. +//! +//! Amazon DynamoDB answers a second `DeleteTable` and an `UpdateTable` in that +//! window with `ResourceInUseException` and the sentence `Attempt to change a +//! resource which is still in use: Table is being deleted: ` (measured +//! 2026-10-24). On this backend `DeleteTable` drops the data and the catalog +//! row in one call and never sets `DELETING` itself; the abandoned-restore +//! sweep does. The window is produced here by setting the row's status +//! directly. +//! +//! The catalog keyspace is shared by every test in this binary, and the +//! control-plane sweep (`process_control_plane_transitions`) removes any +//! `DELETING` row whose `status_transition_at` has passed, then clears the +//! table's TTL entries in the account keyspace. A fresh table carries its +//! activation deadline in that column, so a row left as `DELETING` with it +//! would be swept by whichever test runs the sweep next; after this test's +//! account keyspace is dropped, that sweep fails on the missing keyspace and +//! takes an unrelated test down with it. `set_status` therefore clears the +//! column, as activation does, which keeps the row out of the sweep, and each +//! test removes its row before it returns. + +use extenddb_core::types::{DeleteTableInput, UpdateTableInput}; +use extenddb_storage::TableEngine; +use extenddb_storage::error::StorageError; +use extenddb_storage_cassandra::CassandraEngine; + +use crate::helpers::{TestAccount, TestTable, setup_engine, test_config}; + +const TABLE: &str = "DeletingWindowTable"; + +/// Sets the row's status and clears its transition deadline, so the +/// control-plane sweep leaves the row alone for the rest of the test. +async fn set_status(engine: &CassandraEngine, account_id: &str, status: &str) { + let query = format!( + "UPDATE {}.tables SET table_status = ?, status_transition_at = null \ + WHERE account_id = ? AND table_name = ? IF EXISTS", + engine.catalog_keyspace() + ); + engine + .session() + .query_with_values(&query, cdrs_tokio::query_values!(status, account_id, TABLE)) + .await + .expect("set the table status"); +} + +/// Removes the catalog row the test set to a non-ACTIVE status; the account +/// keyspace goes with the `TestAccount`. +async fn remove_row(engine: &CassandraEngine, account_id: &str) { + let query = format!( + "DELETE FROM {}.tables WHERE account_id = ? AND table_name = ? IF EXISTS", + engine.catalog_keyspace() + ); + engine + .session() + .query_with_values(&query, cdrs_tokio::query_values!(account_id, TABLE)) + .await + .expect("remove the table row"); +} + +async fn status_of(engine: &CassandraEngine, account_id: &str) -> Option { + use cdrs_tokio::types::IntoRustByName; + let query = format!( + "SELECT table_status FROM {}.tables WHERE account_id = ? AND table_name = ?", + engine.catalog_keyspace() + ); + let rows = engine + .session() + .query_with_values(&query, cdrs_tokio::query_values!(account_id, TABLE)) + .await + .expect("read the row") + .response_body() + .expect("response body") + .into_rows() + .unwrap_or_default(); + rows.first() + .map(|row| row.get_r_by_name("table_status").expect("table_status")) +} + +fn update_input() -> UpdateTableInput { + UpdateTableInput { + table_name: TABLE.to_owned(), + billing_mode: None, + provisioned_throughput: None, + deletion_protection_enabled: Some(true), + global_secondary_index_updates: None, + attribute_definitions: None, + stream_specification: None, + table_class: None, + on_demand_throughput: None, + table_throughput_mode: None, + vector_index_updates: None, + } +} + +/// An engine, an account, an ACTIVE `TABLE`, and its row set to `DELETING`. +async fn open_window() -> (CassandraEngine, TestAccount, TestTable) { + let engine = setup_engine().await; + let account = TestAccount::new(&engine, &test_config().keyspace_prefix).await; + let table = TestTable::with_account(&engine, &account.account_id, TABLE, false).await; + set_status(&engine, &account.account_id, "DELETING").await; + (engine, account, table) +} + +/// A second DeleteTable during the window reports the table as being deleted, +/// not as absent, and does not run the drop: the row stays. +#[tokio::test] +async fn a_second_delete_table_during_the_window_is_table_being_deleted() { + if crate::helpers::skip_without_cassandra() { + return; + } + let (engine, account, _table) = open_window().await; + + let err = engine + .delete_table( + &account.account_id, + DeleteTableInput { + table_name: TABLE.to_owned(), + }, + ) + .await + .expect_err("DeleteTable on a DELETING table"); + assert!( + matches!(&err, StorageError::TableBeingDeleted(name) if name == TABLE), + "expected TableBeingDeleted({TABLE:?}), got {err:?}" + ); + assert_eq!( + status_of(&engine, &account.account_id).await.as_deref(), + Some("DELETING") + ); + remove_row(&engine, &account.account_id).await; +} + +/// UpdateTable during the window reports the table as being deleted, not with +/// the sentence for the other non-ACTIVE statuses. +#[tokio::test] +async fn update_table_during_the_window_is_table_being_deleted() { + if crate::helpers::skip_without_cassandra() { + return; + } + let (engine, account, _table) = open_window().await; + + let err = engine + .update_table(&account.account_id, update_input()) + .await + .expect_err("UpdateTable on a DELETING table"); + assert!( + matches!(&err, StorageError::TableBeingDeleted(name) if name == TABLE), + "expected TableBeingDeleted({TABLE:?}), got {err:?}" + ); + remove_row(&engine, &account.account_id).await; +} + +/// A CREATING table keeps the existing answer: the new sentence is for +/// DELETING only. +#[tokio::test] +async fn update_table_on_a_creating_table_keeps_table_not_active() { + if crate::helpers::skip_without_cassandra() { + return; + } + let (engine, account, _table) = open_window().await; + set_status(&engine, &account.account_id, "CREATING").await; + + let err = engine + .update_table(&account.account_id, update_input()) + .await + .expect_err("UpdateTable 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-cassandra/tests/direct_integration.rs b/crates/storage-cassandra/tests/direct_integration.rs index 43eb28b8..5826be23 100644 --- a/crates/storage-cassandra/tests/direct_integration.rs +++ b/crates/storage-cassandra/tests/direct_integration.rs @@ -27,6 +27,8 @@ mod backup_engine; mod cassandra_engine; #[path = "direct/delete_item.rs"] mod delete_item; +#[path = "direct/deleting_window.rs"] +mod deleting_window; #[path = "direct/groups.rs"] mod groups; #[path = "direct/index.rs"] diff --git a/crates/storage-mongodb/src/deleting_window_tests.rs b/crates/storage-mongodb/src/deleting_window_tests.rs new file mode 100644 index 00000000..c8aeb44a --- /dev/null +++ b/crates/storage-mongodb/src/deleting_window_tests.rs @@ -0,0 +1,296 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Storage-level tests for what the control plane answers while a table is +//! `DELETING`, against a live MongoDB replica set from +//! `EXTENDDB_TEST_MONGODB_URI`, for example +//! `mongodb://127.0.0.1:27105/?directConnection=true`. Each test uses its own +//! catalog and data databases and drops them when it passes. +//! +//! Amazon DynamoDB answers a second `DeleteTable` and an `UpdateTable` in that +//! window with `ResourceInUseException` and the sentence `Attempt to change a +//! resource which is still in use: Table is being deleted: ` (measured +//! 2026-10-24). On this backend `DeleteTable` marks the document `DELETING` +//! and drops the collections in the same call, so the window is produced by +//! setting the document's status directly. + +use std::sync::Arc; + +use extenddb_core::types::{ + AttributeDefinition, BillingMode, CreateTableInput, DeleteTableInput, KeySchemaElement, + KeyType, ScalarAttributeType, UpdateTableInput, +}; +use extenddb_storage::TableEngine; +use extenddb_storage::error::StorageError; +use mongodb::bson::{Document, doc}; + +use crate::MongoEngine; + +const ACCOUNT: &str = "123456789012"; +const TABLE: &str = "t_deleting_window"; + +fn uri() -> String { + std::env::var("EXTENDDB_TEST_MONGODB_URI").expect( + "set EXTENDDB_TEST_MONGODB_URI to a MongoDB replica set, e.g. \ + mongodb://127.0.0.1:27105/?directConnection=true", + ) +} + +struct Fixture { + engine: MongoEngine, +} + +impl Fixture { + async fn new() -> Self { + let tag = uuid::Uuid::new_v4().simple().to_string(); + let client = mongodb::Client::with_uri_str(uri()).await.expect("client"); + let slots = crate::backup_engine::restore_server_slot_count(10); + let engine = MongoEngine { + client: client.clone(), + catalog_db: client.database(&format!("t{}_cat", &tag[..12])), + data_db: client.database(&format!("t{}_data", &tag[..12])), + region: "us-east-1".to_owned(), + gsi_cache: Arc::default(), + tx_read_concern: mongodb::options::ReadConcern::snapshot(), + backup_liveness: crate::backup_engine::BackupLiveness::default(), + restore_server_slot_count: slots, + restore_server_slots: Arc::new(tokio::sync::Semaphore::new(slots)), + restore_account_slots: crate::backup_engine::RestoreAccountSlots::default(), + }; + engine + .catalog_db + .collection::("settings") + .insert_one(doc! { "_id": "control_plane_delay_seconds", "value": "0" }) + .await + .expect("settings"); + Self { engine } + } + + /// Create `TABLE` and set its document to `DELETING`: the window. + async fn open_window(&self) { + self.engine + .create_table( + ACCOUNT, + CreateTableInput { + table_name: TABLE.to_owned(), + key_schema: vec![KeySchemaElement { + attribute_name: "pk".to_owned(), + key_type: KeyType::Hash, + }], + attribute_definitions: vec![AttributeDefinition { + attribute_name: "pk".to_owned(), + attribute_type: ScalarAttributeType::S, + }], + billing_mode: Some(BillingMode::PayPerRequest), + ..Default::default() + }, + ) + .await + .expect("create the table"); + self.set_status("DELETING").await; + } + + async fn set_status(&self, status: &str) { + self.engine + .catalog_db + .collection::("tables") + .update_one( + doc! { "_id": { "account_id": ACCOUNT, "table_name": TABLE } }, + doc! { "$set": { "table_status": status } }, + ) + .await + .expect("set the table status"); + } + + async fn status(&self) -> Option { + self.engine + .catalog_db + .collection::("tables") + .find_one(doc! { "_id": { "account_id": ACCOUNT, "table_name": TABLE } }) + .await + .expect("read the table document") + .map(|d| d.get_str("table_status").expect("table_status").to_owned()) + } + + async fn finish(self) { + self.engine.catalog_db.drop().await.expect("drop catalog"); + self.engine.data_db.drop().await.expect("drop data"); + } +} + +fn update_input() -> UpdateTableInput { + UpdateTableInput { + table_name: TABLE.to_owned(), + billing_mode: None, + provisioned_throughput: None, + deletion_protection_enabled: Some(true), + global_secondary_index_updates: None, + attribute_definitions: None, + stream_specification: None, + table_class: None, + on_demand_throughput: None, + table_throughput_mode: None, + vector_index_updates: None, + } +} + +/// A second DeleteTable during the window reports the table as being deleted, +/// not as absent, and does not run the drop again: the document stays. +#[tokio::test] +#[ignore = "needs a MongoDB replica set in EXTENDDB_TEST_MONGODB_URI"] +async fn a_second_delete_table_during_the_window_is_table_being_deleted() { + let f = Fixture::new().await; + f.open_window().await; + + let err = f + .engine + .delete_table( + ACCOUNT, + DeleteTableInput { + table_name: TABLE.to_owned(), + }, + ) + .await + .expect_err("DeleteTable on a DELETING table"); + assert!( + matches!(&err, StorageError::TableBeingDeleted(name) if name == TABLE), + "expected TableBeingDeleted({TABLE:?}), got {err:?}" + ); + assert_eq!(f.status().await.as_deref(), Some("DELETING")); + + f.finish().await; +} + +/// UpdateTable during the window reports the table as being deleted, not with +/// the sentence for the other non-ACTIVE statuses, and changes nothing. +#[tokio::test] +#[ignore = "needs a MongoDB replica set in EXTENDDB_TEST_MONGODB_URI"] +async fn update_table_during_the_window_is_table_being_deleted() { + let f = Fixture::new().await; + f.open_window().await; + + let err = f + .engine + .update_table(ACCOUNT, update_input()) + .await + .expect_err("UpdateTable on a DELETING table"); + assert!( + matches!(&err, StorageError::TableBeingDeleted(name) if name == TABLE), + "expected TableBeingDeleted({TABLE:?}), got {err:?}" + ); + let doc = f + .engine + .catalog_db + .collection::("tables") + .find_one(doc! { "_id": { "account_id": ACCOUNT, "table_name": TABLE } }) + .await + .expect("read the table document") + .expect("the document is still there"); + assert_ne!( + doc.get_bool("deletion_protection_enabled").ok(), + Some(true), + "a refused UpdateTable must not have written" + ); + + f.finish().await; +} + +/// A CREATING table keeps the existing answer: the new sentence is for +/// DELETING only. +#[tokio::test] +#[ignore = "needs a MongoDB replica set in EXTENDDB_TEST_MONGODB_URI"] +async fn update_table_on_a_creating_table_keeps_table_not_active() { + let f = Fixture::new().await; + f.open_window().await; + f.set_status("CREATING").await; + + let err = f + .engine + .update_table(ACCOUNT, update_input()) + .await + .expect_err("UpdateTable on a CREATING table"); + assert!( + matches!(&err, StorageError::TableNotActive(name) if name == TABLE), + "expected TableNotActive({TABLE:?}), got {err:?}" + ); + + f.finish().await; +} + +/// Two concurrent DeleteTable calls on one ACTIVE table: the conditioned mark +/// matches for one of them, the other answers TableBeingDeleted (the service's +/// answer for the second of two concurrent calls, measured 2026-10-25). Ten +/// rounds, a fresh table each; afterwards no catalog document and no data +/// collection for the table remain. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +#[ignore = "needs a MongoDB replica set in EXTENDDB_TEST_MONGODB_URI"] +async fn two_concurrent_delete_tables_have_one_winner() { + let f = Fixture::new().await; + let engine = Arc::new(f.engine.clone()); + + for round in 0..10 { + let table = format!("t_pair_{round}"); + let desc = engine + .create_table( + ACCOUNT, + CreateTableInput { + table_name: table.clone(), + key_schema: vec![KeySchemaElement { + attribute_name: "pk".to_owned(), + key_type: KeyType::Hash, + }], + attribute_definitions: vec![AttributeDefinition { + attribute_name: "pk".to_owned(), + attribute_type: ScalarAttributeType::S, + }], + billing_mode: Some(BillingMode::PayPerRequest), + ..Default::default() + }, + ) + .await + .expect("create the table"); + + let tasks: Vec<_> = (0..2) + .map(|_| { + let engine = Arc::clone(&engine); + let table = table.clone(); + tokio::spawn(async move { + engine + .delete_table(ACCOUNT, DeleteTableInput { table_name: table }) + .await + }) + }) + .collect(); + let mut ok = 0; + let mut being_deleted = 0; + for task in tasks { + match task.await.expect("task") { + Ok(_) => ok += 1, + Err(StorageError::TableBeingDeleted(name)) if name == table => being_deleted += 1, + Err(other) => panic!("round {round}: unexpected {other:?}"), + } + } + assert_eq!((ok, being_deleted), (1, 1), "round {round}"); + + let doc = engine + .catalog_db + .collection::("tables") + .find_one(doc! { "_id": { "account_id": ACCOUNT, "table_name": &table } }) + .await + .expect("read the table document"); + assert!(doc.is_none(), "round {round}: the catalog document is gone"); + let collections = engine + .data_db + .list_collection_names() + .await + .expect("list collections"); + let data_coll = crate::data::data_collection_name(&desc.table_id); + assert!( + !collections.contains(&data_coll), + "round {round}: the data collection is gone" + ); + } + + drop(engine); + f.finish().await; +} diff --git a/crates/storage-mongodb/src/table_engine.rs b/crates/storage-mongodb/src/table_engine.rs index aa8c5272..27d37ef1 100644 --- a/crates/storage-mongodb/src/table_engine.rs +++ b/crates/storage-mongodb/src/table_engine.rs @@ -576,6 +576,14 @@ impl MongoEngine { .describe_table_impl(account_id, &input.table_name) .await?; + // The document stays, marked DELETING, while the collections drop + // below. A second DeleteTable in that window is refused the way the + // service refuses it (measured 2026-10-24), before the deletion + // protection check, as the SQL backends order it. + if desc.table_status == TableStatus::Deleting { + return Err(StorageError::TableBeingDeleted(input.table_name.clone())); + } + // Check deletion protection if desc.deletion_protection_enabled { return Err(StorageError::DeletionProtected(input.table_name.clone())); @@ -592,15 +600,32 @@ impl MongoEngine { ))); } - // Mark as DELETING + // Mark as DELETING, conditioned on the document still being the one + // described above and not yet DELETING. Two concurrent DeleteTable + // calls both read the ACTIVE document; only one update matches. The + // table id in the filter keeps a stalled call from marking a table + // created into the same name after this one was gone. let tables_coll = self.catalog_db.collection::("tables"); - tables_coll + let marked = tables_coll .update_one( - doc! { "_id": { "account_id": account_id, "table_name": &input.table_name } }, + doc! { + "_id": { "account_id": account_id, "table_name": &input.table_name }, + "table_id": &desc.table_id, + "table_status": { "$ne": "DELETING" }, + }, doc! { "$set": { "table_status": "DELETING" } }, ) .await .map_err(|e| StorageError::Internal(e.to_string()))?; + if marked.matched_count == 0 { + // The document this call read is now DELETING or already removed, + // and only a concurrent DeleteTable does either: this call lost the + // race. The service answers the second of two concurrent calls on + // an ACTIVE table with the DELETING sentence (measured 2026-10-25, + // five rounds of five), whether or not the first has finished by + // the time the second looks again. + return Err(StorageError::TableBeingDeleted(input.table_name.clone())); + } // Drop the data collection let coll_name = data_collection_name(&desc.table_id); @@ -633,11 +658,14 @@ impl MongoEngine { // table's stream history. RFC-0003 §8.2 (table-name reuse). self.cleanup_stream_state_for_table(&desc.table_id).await?; - // Delete the table metadata + // Delete the table metadata: only the document this call marked, so a + // table created into the same name meanwhile is left alone. tables_coll - .delete_one( - doc! { "_id": { "account_id": account_id, "table_name": &input.table_name } }, - ) + .delete_one(doc! { + "_id": { "account_id": account_id, "table_name": &input.table_name }, + "table_id": &desc.table_id, + "table_status": "DELETING", + }) .await .map_err(|e| StorageError::Internal(e.to_string()))?; @@ -1132,9 +1160,10 @@ impl MongoEngine { } /// `TableNotFound` when no catalog document matches `filter`, - /// `TableNotActive` when one does but its `table_status` is not + /// `TableBeingDeleted` when one does and its `table_status` is + /// `DELETING`, `TableNotActive` when it is any other status but /// `ACTIVE`. UpdateTable's guard; also used after a conditional write - /// matched nothing to report which of the two happened. + /// matched nothing to report which of the three happened. async fn require_active_table( &self, tables_coll: &mongodb::Collection, @@ -1146,10 +1175,11 @@ impl MongoEngine { .await .map_err(|e| StorageError::Internal(e.to_string()))? .ok_or_else(|| StorageError::TableNotFound(table_name.to_owned()))?; - if table_doc.get_str("table_status").unwrap_or("ACTIVE") != "ACTIVE" { - return Err(StorageError::TableNotActive(table_name.to_owned())); + match table_doc.get_str("table_status").unwrap_or("ACTIVE") { + "ACTIVE" => Ok(()), + "DELETING" => Err(StorageError::TableBeingDeleted(table_name.to_owned())), + _ => Err(StorageError::TableNotActive(table_name.to_owned())), } - Ok(()) } pub(crate) async fn table_key_info_impl( @@ -1827,6 +1857,10 @@ fn active_table_filter(filter: &Document) -> Document { active } +#[cfg(test)] +#[path = "deleting_window_tests.rs"] +mod deleting_window_tests; + #[cfg(test)] mod tests { use super::active_table_filter; diff --git a/crates/storage-postgres/src/data/ddl.rs b/crates/storage-postgres/src/data/ddl.rs index f608c89e..d12c6a89 100755 --- a/crates/storage-postgres/src/data/ddl.rs +++ b/crates/storage-postgres/src/data/ddl.rs @@ -421,11 +421,17 @@ impl PostgresEngine { row.ok_or_else(|| StorageError::TableNotFound(table_name.to_owned()))?; match status.as_str() { - "ACTIVE" => {} - // A table that is still being created, or is being deleted, is not - // usable for data-plane operations; DynamoDB reports it as not found - // (not as in-use). UPDATING keeps the not-active classification. - "CREATING" | "DELETING" => { + // DynamoDB serves reads and writes on a DELETING table until the + // table is gone (measured 2026-10-25); a write made in that window + // goes away with the table. The data tables stay until the control + // plane drops them, so the data plane reads a DELETING row as found; + // a statement that reaches a dropped data table is answered as not + // found by the `DataEngine` implementation. + "ACTIVE" | "DELETING" => {} + // A table that is still being created is not usable for data-plane + // operations; DynamoDB reports it as not found (not as in-use). + // UPDATING keeps the not-active classification. + "CREATING" => { return Err(StorageError::TableNotFound(table_name.to_owned())); } _ => return Err(StorageError::TableNotActive(table_name.to_owned())), @@ -576,11 +582,10 @@ impl PostgresEngine { row.ok_or_else(|| StorageError::TableNotFound(table_name.to_owned()))?; match status.as_str() { - "ACTIVE" => {} - // A table that is still being created, or is being deleted, is not - // usable for data-plane operations; DynamoDB reports it as not found - // (not as in-use). UPDATING keeps the not-active classification. - "CREATING" | "DELETING" => { + // Same classification as `fetch_table_key_info`: a DELETING table + // is served until it is gone, a CREATING one is not found. + "ACTIVE" | "DELETING" => {} + "CREATING" => { return Err(StorageError::TableNotFound(table_name.to_owned())); } _ => return Err(StorageError::TableNotActive(table_name.to_owned())), diff --git a/crates/storage-postgres/src/data/delete_item.rs b/crates/storage-postgres/src/data/delete_item.rs index 9a6dfda9..163cc3da 100755 --- a/crates/storage-postgres/src/data/delete_item.rs +++ b/crates/storage-postgres/src/data/delete_item.rs @@ -10,7 +10,7 @@ use extenddb_storage::error::StorageError; use extenddb_storage::util::{SortKeyValue, parse_sk, pk_to_text, sk_column, sk_info}; use super::gsi_build::fix_up_stale_write; -use super::index::{enqueue_async_indexes, fetch_write_path_indexes, sync_indexes}; +use super::index::{db_error, enqueue_async_indexes, fetch_write_path_indexes, sync_indexes}; use super::query::check_condition; use super::tx_helpers::write_stream_record_in_tx; use super::{data_table_name, json_to_item}; @@ -91,11 +91,7 @@ impl PostgresEngine { ); let delete_sql = format!("DELETE FROM {ddb_table} WHERE pk = $1 AND {sk_col} = $2"); - let mut tx = self - .data_pool - .begin() - .await - .map_err(|e| StorageError::Internal(e.to_string()))?; + let mut tx = self.data_pool.begin().await.map_err(db_error)?; let old: Option<(serde_json::Value,)> = bind_sk_fetch_optional!(&select_sql, pk_text.as_ref(), &sk, &mut *tx)?; @@ -147,7 +143,7 @@ impl PostgresEngine { .await } } - .map_err(|e| StorageError::Internal(e.to_string()))?; + .map_err(db_error)?; // Sync GSI/LSI update within transaction (D-4). let old_item_for_idx = if indexes.is_empty() { @@ -220,9 +216,7 @@ impl PostgresEngine { async_enqueued += fix_up_stale_write(&mut tx, &self.pool, stamp, key_info, &indexes, key).await?; - tx.commit() - .await - .map_err(|e| StorageError::Internal(e.to_string()))?; + tx.commit().await.map_err(db_error)?; if async_enqueued > 0 && let Some(ref q) = self.gsi_queue @@ -261,7 +255,7 @@ impl PostgresEngine { .await } } - .map_err(|e| StorageError::Internal(e.to_string()))?; + .map_err(db_error)?; self.fix_up_autocommit(stamp, key_info, key).await; Ok(None) } @@ -272,17 +266,13 @@ impl PostgresEngine { format!("SELECT item_data FROM {ddb_table} WHERE pk = $1 FOR UPDATE"); let delete_sql = format!("DELETE FROM {ddb_table} WHERE pk = $1"); - let mut tx = self - .data_pool - .begin() - .await - .map_err(|e| StorageError::Internal(e.to_string()))?; + let mut tx = self.data_pool.begin().await.map_err(db_error)?; let old: Option<(serde_json::Value,)> = sqlx::query_as(&select_sql) .bind(pk_text.as_ref()) .fetch_optional(&mut *tx) .await - .map_err(|e| StorageError::Internal(e.to_string()))?; + .map_err(db_error)?; if let Some((ref old_json,)) = old { let old_item: Item = json_to_item(old_json.clone())?; @@ -309,7 +299,7 @@ impl PostgresEngine { .bind(pk_text.as_ref()) .execute(&mut *tx) .await - .map_err(|e| StorageError::Internal(e.to_string()))?; + .map_err(db_error)?; // Sync GSI/LSI update within transaction (D-4). let old_item_for_idx = if indexes.is_empty() { @@ -382,9 +372,7 @@ impl PostgresEngine { async_enqueued += fix_up_stale_write(&mut tx, &self.pool, stamp, key_info, &indexes, key).await?; - tx.commit() - .await - .map_err(|e| StorageError::Internal(e.to_string()))?; + tx.commit().await.map_err(db_error)?; if async_enqueued > 0 && let Some(ref q) = self.gsi_queue @@ -404,7 +392,7 @@ impl PostgresEngine { .bind(pk_text.as_ref()) .execute(&self.data_pool) .await - .map_err(|e| StorageError::Internal(e.to_string()))?; + .map_err(db_error)?; self.fix_up_autocommit(stamp, key_info, key).await; Ok(None) } diff --git a/crates/storage-postgres/src/data/put_item.rs b/crates/storage-postgres/src/data/put_item.rs index 2fbc3102..34fdf301 100755 --- a/crates/storage-postgres/src/data/put_item.rs +++ b/crates/storage-postgres/src/data/put_item.rs @@ -10,7 +10,7 @@ use extenddb_storage::error::StorageError; use extenddb_storage::util::{composite_pk_to_text, parse_sk, pk_to_text, sk_column, sk_info}; use super::gsi_build::fix_up_stale_write; -use super::index::{enqueue_async_indexes, fetch_write_path_indexes, sync_indexes}; +use super::index::{db_error, enqueue_async_indexes, fetch_write_path_indexes, sync_indexes}; use super::query::check_condition; use super::tx_helpers::write_stream_record_in_tx; use super::{data_table_name, json_to_item}; @@ -111,11 +111,7 @@ impl PostgresEngine { "SELECT item_data FROM {ddb_table} WHERE pk = $1 AND {sk_col} = $2 FOR UPDATE" ); - let mut tx = self - .data_pool - .begin() - .await - .map_err(|e| StorageError::Internal(e.to_string()))?; + let mut tx = self.data_pool.begin().await.map_err(db_error)?; let mut old: Option<(serde_json::Value,)> = bind_sk_fetch_optional!(&select_sql, pk_text.as_str(), &sk, &mut *tx)?; @@ -243,9 +239,7 @@ impl PostgresEngine { fix_up_stale_write(&mut tx, &self.pool, stamp, key_info, &indexes, &item) .await?; - tx.commit() - .await - .map_err(|e| StorageError::Internal(e.to_string()))?; + tx.commit().await.map_err(db_error)?; if async_enqueued > 0 && let Some(ref q) = self.gsi_queue @@ -280,18 +274,14 @@ impl PostgresEngine { let select_sql = format!("SELECT item_data FROM {ddb_table} WHERE pk = $1 FOR UPDATE"); - let mut tx = self - .data_pool - .begin() - .await - .map_err(|e| StorageError::Internal(e.to_string()))?; + let mut tx = self.data_pool.begin().await.map_err(db_error)?; let fetch_old = async |tx: &mut sqlx::PgConnection| { sqlx::query_as::<_, (serde_json::Value,)>(&select_sql) .bind(pk_text.as_str()) .fetch_optional(tx) .await - .map_err(|e| StorageError::Internal(e.to_string())) + .map_err(db_error) }; let mut old: Option<(serde_json::Value,)> = fetch_old(&mut tx).await?; @@ -314,7 +304,7 @@ impl PostgresEngine { .bind(&item_json) .execute(&mut *tx) .await - .map_err(|e| StorageError::Internal(e.to_string()))?; + .map_err(db_error)?; break; } let empty = std::collections::BTreeMap::new(); @@ -334,7 +324,7 @@ impl PostgresEngine { .bind(&item_json) .execute(&mut *tx) .await - .map_err(|e| StorageError::Internal(e.to_string()))?; + .map_err(db_error)?; if result.rows_affected() == 1 { break; } @@ -421,9 +411,7 @@ impl PostgresEngine { fix_up_stale_write(&mut tx, &self.pool, stamp, key_info, &indexes, &item) .await?; - tx.commit() - .await - .map_err(|e| StorageError::Internal(e.to_string()))?; + tx.commit().await.map_err(db_error)?; if async_enqueued > 0 && let Some(ref q) = self.gsi_queue @@ -447,7 +435,7 @@ impl PostgresEngine { .bind(&item_json) .execute(&self.data_pool) .await - .map_err(|e| StorageError::Internal(e.to_string()))?; + .map_err(db_error)?; self.fix_up_autocommit(stamp, key_info, &item).await; Ok(None) } @@ -486,7 +474,7 @@ impl PostgresEngine { .bind(pk_text.as_ref()) .fetch_optional(&self.data_pool) .await - .map_err(|e| StorageError::Internal(e.to_string()))?; + .map_err(db_error)?; row.map(|(v,)| v) }; diff --git a/crates/storage-postgres/src/data/query.rs b/crates/storage-postgres/src/data/query.rs index 6a84dc1b..5e20ce3a 100755 --- a/crates/storage-postgres/src/data/query.rs +++ b/crates/storage-postgres/src/data/query.rs @@ -14,6 +14,8 @@ use extenddb_storage::error::StorageError; use extenddb_storage::util::SortKeyValue; use extenddb_storage::util::{parse_sk, pk_to_text, sk_info}; +use super::index::db_error; + /// Extra pagination bind values for index queries. /// /// Built in `query_scan.rs` in the same branch that generates the SQL placeholders, @@ -203,10 +205,7 @@ pub(crate) async fn execute_query_sql( } } - let rows: Vec<(serde_json::Value,)> = query - .fetch_all(pool) - .await - .map_err(|e| StorageError::Internal(e.to_string()))?; + let rows: Vec<(serde_json::Value,)> = query.fetch_all(pool).await.map_err(db_error)?; Ok(rows.into_iter().map(|(v,)| v).collect()) } @@ -343,10 +342,7 @@ pub(crate) async fn execute_scan_sql( } } - let rows: Vec<(serde_json::Value,)> = query - .fetch_all(pool) - .await - .map_err(|e| StorageError::Internal(e.to_string()))?; + let rows: Vec<(serde_json::Value,)> = query.fetch_all(pool).await.map_err(db_error)?; Ok(rows.into_iter().map(|(v,)| v).collect()) } diff --git a/crates/storage-postgres/src/data/tx_helpers.rs b/crates/storage-postgres/src/data/tx_helpers.rs index 6468f159..60c09220 100755 --- a/crates/storage-postgres/src/data/tx_helpers.rs +++ b/crates/storage-postgres/src/data/tx_helpers.rs @@ -48,7 +48,7 @@ pub(super) async fn fetch_item_in_tx( .bind(pk_text.as_ref()) .fetch_optional(&mut **tx) .await - .map_err(|e| StorageError::Internal(e.to_string()))?; + .map_err(db_error)?; row.map(|(v,)| v) }; diff --git a/crates/storage-postgres/src/data/update_item.rs b/crates/storage-postgres/src/data/update_item.rs index 80ab9a51..8d1541b8 100755 --- a/crates/storage-postgres/src/data/update_item.rs +++ b/crates/storage-postgres/src/data/update_item.rs @@ -11,7 +11,7 @@ use extenddb_storage::error::StorageError; use extenddb_storage::util::{parse_sk, pk_to_text, sk_column, sk_info}; use super::gsi_build::fix_up_stale_write; -use super::index::{enqueue_async_indexes, fetch_write_path_indexes, sync_indexes}; +use super::index::{db_error, enqueue_async_indexes, fetch_write_path_indexes, sync_indexes}; use super::query::check_condition; use super::tx_helpers::write_stream_record_in_tx; use super::{data_table_name, json_to_item}; @@ -40,11 +40,7 @@ impl PostgresEngine { let pk_text = pk_to_text(pk_value)?; // UpdateItem always needs a transaction (read-modify-write) - let mut tx = self - .data_pool - .begin() - .await - .map_err(|e| StorageError::Internal(e.to_string()))?; + let mut tx = self.data_pool.begin().await.map_err(db_error)?; // Both index families in one catalog visit (D-4: sync + async split for the // secondary indexes). @@ -133,7 +129,7 @@ impl PostgresEngine { .bind(pk_text.as_ref()) .fetch_optional(&mut *tx) .await - .map_err(|e| StorageError::Internal(e.to_string()))?; + .map_err(db_error)?; row.map(|(v,)| v) } }; @@ -233,7 +229,7 @@ impl PostgresEngine { .bind(&item_json) .execute(&mut *tx) .await - .map_err(|e| StorageError::Internal(e.to_string()))?; + .map_err(db_error)?; } } break (old_item, new_item, item, pre_mutation_item); @@ -260,7 +256,7 @@ impl PostgresEngine { .bind(&item_json) .execute(&mut *tx) .await - .map_err(|e| StorageError::Internal(e.to_string()))? + .map_err(db_error)? .rows_affected() == 1 } @@ -333,9 +329,7 @@ impl PostgresEngine { async_enqueued += fix_up_stale_write(&mut tx, &self.pool, stamp, key_info, &indexes, key).await?; - tx.commit() - .await - .map_err(|e| StorageError::Internal(e.to_string()))?; + tx.commit().await.map_err(db_error)?; if async_enqueued > 0 && let Some(ref q) = self.gsi_queue diff --git a/crates/storage-postgres/src/delete_table.rs b/crates/storage-postgres/src/delete_table.rs index 7475eed6..b760715d 100755 --- a/crates/storage-postgres/src/delete_table.rs +++ b/crates/storage-postgres/src/delete_table.rs @@ -40,7 +40,28 @@ impl PostgresEngine { .await .map_err(|e| StorageError::Internal(e.to_string()))?; - let row = row.ok_or_else(|| StorageError::TableNotFound(input.table_name.clone()))?; + // The filtered select reads a DELETING row as absent. The service + // answers a second DeleteTable in that window as a table still in use + // (measured 2026-10-24), so look the name up again without the status + // filter before reporting it as not found. + let row = match row { + Some(row) => row, + None => { + let status: Option = sqlx::query_scalar( + "SELECT table_status FROM tables WHERE account_id = $1 AND table_name = $2", + ) + .bind(account_id) + .bind(&input.table_name) + .fetch_optional(&mut *tx) + .await + .map_err(|e| StorageError::Internal(e.to_string()))?; + return Err(if status.as_deref() == Some("DELETING") { + StorageError::TableBeingDeleted(input.table_name.clone()) + } else { + StorageError::TableNotFound(input.table_name.clone()) + }); + } + }; // REQ: DeletionProtectionEnabled check — real DynamoDB returns ValidationException if row.deletion_protection_enabled { diff --git a/crates/storage-postgres/src/update_table.rs b/crates/storage-postgres/src/update_table.rs index dec99a93..814e736f 100755 --- a/crates/storage-postgres/src/update_table.rs +++ b/crates/storage-postgres/src/update_table.rs @@ -131,6 +131,11 @@ impl PostgresEngine { let (status, table_id, ks_json, ad_json, stored_billing_mode) = row.ok_or_else(|| StorageError::TableNotFound(input.table_name.clone()))?; + // A DELETING table has its own sentence on the service (measured + // 2026-10-24); every other non-ACTIVE status keeps the existing one. + if status == "DELETING" { + return Err(StorageError::TableBeingDeleted(input.table_name.clone())); + } if status != "ACTIVE" { return Err(StorageError::TableNotActive(input.table_name.clone())); } diff --git a/crates/storage-postgres/tests/deleting_window.rs b/crates/storage-postgres/tests/deleting_window.rs new file mode 100644 index 00000000..a53c4386 --- /dev/null +++ b/crates/storage-postgres/tests/deleting_window.rs @@ -0,0 +1,598 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Storage-level tests for what the control plane answers while a table is +//! `DELETING`. +//! +//! Amazon DynamoDB answers a second `DeleteTable` and an `UpdateTable` in that +//! window with `ResourceInUseException` and the sentence `Attempt to change a +//! resource which is still in use: Table is being deleted: ` (measured +//! 2026-10-24). The window is produced here by `delete_table` itself: with a +//! control-plane delay of one second or more the row is marked `DELETING` with +//! a scheduled transition, and these tests run no poller to finish it. +//! +//! Each test builds its own throwaway database, applies the shipped migrations +//! to it, and drops it when it passes. A failing test leaves its database behind +//! on purpose, named `eddb_delw_*`, so the state that failed can be inspected. +//! +//! Requires `EXTENDDB_TEST_PG_CONNECTION_STRING`, a base URL with no database +//! component (for example `postgresql://postgres@127.0.0.1:5432`), pointing at a +//! server whose role may create and drop databases. Without it every test here +//! reports a skip and passes, the same convention the wire suites use. +//! +//! The data-plane tests below pin the other half of the window: DynamoDB serves +//! `PutItem`, `GetItem`, `Query` and `Scan` on a `DELETING` table until the +//! table is gone, then answers `ResourceNotFoundException` (measured +//! 2026-10-25). Two `DeleteTable` calls on one `ACTIVE` table get one `200` and +//! one `ResourceInUseException` with the `Table is being deleted` sentence. + +use std::collections::{BTreeMap, HashMap}; +use std::sync::Arc; + +use extenddb_core::expression::{Expr, ExpressionMaps, KeyCondition, PathElement}; +use extenddb_core::types::{ + AttributeDefinition, AttributeValue, BillingMode, CreateTableInput, DeleteTableInput, Item, + KeySchemaElement, KeyType, ReturnValuesOnConditionCheckFailure, ScalarAttributeType, + TableKeyInfo, TableStatus, UpdateTableInput, +}; +use extenddb_storage::error::StorageError; +use extenddb_storage::{DataEngine, TableEngine, TransactGetOp, TransactWriteOp}; +use extenddb_storage_postgres::{PostgresConfig, PostgresEngine}; +use sqlx::PgPool; +use sqlx::postgres::PgPoolOptions; + +const ACCOUNT: &str = "123456789012"; +const REGION: &str = "us-east-1"; +const TABLE: &str = "t_deleting_window"; + +struct Scratch { + engine: PostgresEngine, + catalog: PgPool, + admin: PgPool, + db_name: String, +} + +impl Scratch { + /// Drop the scratch database. Called only on the success path, so a failure + /// leaves the database for inspection. + async fn cleanup(self) { + let Scratch { + engine, + catalog, + admin, + db_name, + } = self; + drop(engine); + catalog.close().await; + sqlx::query(&format!( + "DROP DATABASE IF EXISTS \"{db_name}\" WITH (FORCE)" + )) + .execute(&admin) + .await + .expect("drop the scratch database"); + admin.close().await; + } +} + +fn base_conn() -> Option { + let conn = std::env::var("EXTENDDB_TEST_PG_CONNECTION_STRING").ok()?; + (!conn.trim().is_empty()).then(|| conn.trim_end_matches('/').to_owned()) +} + +/// Every `.sql` file under `migrations/` and `data_migrations/`, each +/// directory in filename order. +fn shipped_migrations() -> Vec { + let root = std::path::Path::new(env!("CARGO_MANIFEST_DIR")); + let mut out = Vec::new(); + for dir in ["migrations", "data_migrations"] { + let mut files: Vec<_> = std::fs::read_dir(root.join(dir)) + .expect("read the migrations directory") + .map(|e| e.expect("directory entry").path()) + .filter(|p| p.extension().is_some_and(|e| e == "sql")) + .collect(); + files.sort(); + for f in files { + out.push(std::fs::read_to_string(&f).expect("read a migration file")); + } + } + out +} + +fn skip(test: &str) { + eprintln!( + "SKIP {test}: EXTENDDB_TEST_PG_CONNECTION_STRING is not set, so there is no PostgreSQL \ + to build a scratch catalog in." + ); +} + +/// A scratch database with the shipped schema, a zero control-plane delay so +/// `create_table` completes inline, and an engine on it. +async fn scratch() -> Scratch { + let base = base_conn().expect("caller checks base_conn() first"); + let db_name = format!("eddb_delw_{}", uuid::Uuid::new_v4().simple())[..24].to_owned(); + + let admin = PgPoolOptions::new() + .max_connections(1) + .connect(&format!("{base}/postgres")) + .await + .expect("connect to the postgres maintenance database"); + sqlx::query(&format!("CREATE DATABASE \"{db_name}\"")) + .execute(&admin) + .await + .expect("create the scratch database"); + + let url = format!("{base}/{db_name}"); + let catalog = PgPoolOptions::new() + .max_connections(2) + .connect(&url) + .await + .expect("connect to the scratch database"); + for sql in shipped_migrations() { + sqlx::raw_sql(&sql) + .execute(&catalog) + .await + .expect("apply a shipped migration"); + } + set_delay(&catalog, "0").await; + sqlx::query("UPDATE settings SET value = '0' WHERE key = 'index_propagation_delay_ms'") + .execute(&catalog) + .await + .expect("pin the propagation delay to zero"); + sqlx::query("INSERT INTO accounts (account_id, account_name) VALUES ($1, $2)") + .bind(ACCOUNT) + .bind(format!("acct-{db_name}")) + .execute(&catalog) + .await + .expect("seed the account row"); + + let engine = PostgresEngine::new( + &PostgresConfig { + connection_string: url, + pool_size: 10, + max_item_size_bytes: 400_000, + }, + REGION, + ) + .await + .expect("open a PostgresEngine on the scratch database"); + + Scratch { + engine, + catalog, + admin, + db_name, + } +} + +async fn set_delay(catalog: &PgPool, seconds: &str) { + sqlx::query("UPDATE settings SET value = $1 WHERE key = 'control_plane_delay_seconds'") + .bind(seconds) + .execute(catalog) + .await + .expect("set the control-plane delay"); +} + +async fn status_of(catalog: &PgPool, table: &str) -> Option { + sqlx::query_scalar("SELECT table_status FROM tables WHERE account_id = $1 AND table_name = $2") + .bind(ACCOUNT) + .bind(table) + .fetch_optional(catalog) + .await + .expect("read the table status") +} + +fn update_input(table: &str) -> UpdateTableInput { + UpdateTableInput { + table_name: table.to_owned(), + billing_mode: None, + provisioned_throughput: None, + deletion_protection_enabled: Some(true), + global_secondary_index_updates: None, + attribute_definitions: None, + stream_specification: None, + table_class: None, + on_demand_throughput: None, + table_throughput_mode: None, + vector_index_updates: None, + } +} + +/// Create `table` with a single string hash key, completing inline (delay 0). +async fn create_table_named(s: &Scratch, table: &str) { + s.engine + .create_table( + ACCOUNT, + CreateTableInput { + table_name: table.to_owned(), + key_schema: vec![KeySchemaElement { + attribute_name: "pk".to_owned(), + key_type: KeyType::Hash, + }], + attribute_definitions: vec![AttributeDefinition { + attribute_name: "pk".to_owned(), + attribute_type: ScalarAttributeType::S, + }], + billing_mode: Some(BillingMode::PayPerRequest), + ..Default::default() + }, + ) + .await + .expect("create the table"); + assert_eq!( + status_of(&s.catalog, table).await.as_deref(), + Some("ACTIVE") + ); +} + +/// Delete `TABLE` with a delay long enough that the row stays `DELETING` for +/// the rest of the test: the window held open. +async fn start_deleting(s: &Scratch) { + set_delay(&s.catalog, "600").await; + let desc = s + .engine + .delete_table( + ACCOUNT, + DeleteTableInput { + table_name: TABLE.to_owned(), + }, + ) + .await + .expect("the first DeleteTable"); + assert_eq!(desc.table_status, TableStatus::Deleting); + assert_eq!( + status_of(&s.catalog, TABLE).await.as_deref(), + Some("DELETING"), + "the row must be DELETING for the rest of the test" + ); +} + +/// Create `TABLE`, then delete it with the window held open. +async fn open_window(s: &Scratch) { + create_table_named(s, TABLE).await; + start_deleting(s).await; +} + +fn item(pk: &str) -> Item { + BTreeMap::from([("pk".to_owned(), AttributeValue::S(pk.to_owned()))]) +} + +fn query_pk(pk: &str) -> (KeyCondition, ExpressionMaps) { + let condition = KeyCondition { + pk_path: vec![PathElement::Attribute("pk".to_owned())], + pk_value: Expr::Placeholder(":pk".to_owned()), + extra_pk_conditions: Vec::new(), + sk_condition: None, + extra_sk_conditions: Vec::new(), + }; + let maps = ExpressionMaps::new( + HashMap::new(), + HashMap::from([(":pk".to_owned(), AttributeValue::S(pk.to_owned()))]), + ); + (condition, maps) +} + +/// Create `TABLE`, write the item `pre`, and open the window. Returns the key +/// info read while the row was still ACTIVE, the snapshot a cache would hold. +async fn open_window_with_item(s: &Scratch) -> TableKeyInfo { + create_table_named(s, TABLE).await; + let key_info = s + .engine + .table_key_info(ACCOUNT, TABLE) + .await + .expect("key info of the ACTIVE table"); + s.engine + .put_item( + &key_info, + item("pre"), + false, + None, + &ExpressionMaps::default(), + None, + ) + .await + .expect("seed the pre-existing item"); + start_deleting(s).await; + key_info +} + +/// A second DeleteTable during the window reports the table as being deleted, +/// not as absent, and leaves the row in place for the deletion under way. +#[tokio::test] +async fn a_second_delete_table_during_the_window_is_table_being_deleted() { + if base_conn().is_none() { + return skip("a_second_delete_table_during_the_window_is_table_being_deleted"); + } + let s = scratch().await; + open_window(&s).await; + + let err = s + .engine + .delete_table( + ACCOUNT, + DeleteTableInput { + table_name: TABLE.to_owned(), + }, + ) + .await + .expect_err("DeleteTable on a DELETING table"); + assert!( + matches!(&err, StorageError::TableBeingDeleted(name) if name == TABLE), + "expected TableBeingDeleted({TABLE:?}), got {err:?}" + ); + assert_eq!( + status_of(&s.catalog, TABLE).await.as_deref(), + Some("DELETING") + ); + + s.cleanup().await; +} + +/// UpdateTable during the window reports the table as being deleted, not with +/// the sentence for the other non-ACTIVE statuses, and changes nothing. +#[tokio::test] +async fn update_table_during_the_window_is_table_being_deleted() { + if base_conn().is_none() { + return skip("update_table_during_the_window_is_table_being_deleted"); + } + let s = scratch().await; + open_window(&s).await; + + let err = s + .engine + .update_table(ACCOUNT, update_input(TABLE)) + .await + .expect_err("UpdateTable on a DELETING table"); + assert!( + matches!(&err, StorageError::TableBeingDeleted(name) if name == TABLE), + "expected TableBeingDeleted({TABLE:?}), got {err:?}" + ); + let protected: bool = sqlx::query_scalar( + "SELECT deletion_protection_enabled FROM tables WHERE account_id = $1 AND table_name = $2", + ) + .bind(ACCOUNT) + .bind(TABLE) + .fetch_one(&s.catalog) + .await + .expect("read the flag"); + assert!(!protected, "a refused UpdateTable must not have written"); + + s.cleanup().await; +} + +/// A CREATING table keeps the existing answer: the new sentence is for +/// DELETING only. +#[tokio::test] +async fn update_table_on_a_creating_table_keeps_table_not_active() { + if base_conn().is_none() { + return skip("update_table_on_a_creating_table_keeps_table_not_active"); + } + 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_table(ACCOUNT, update_input(TABLE)) + .await + .expect_err("UpdateTable on a CREATING table"); + assert!( + matches!(&err, StorageError::TableNotActive(name) if name == TABLE), + "expected TableNotActive({TABLE:?}), got {err:?}" + ); + + s.cleanup().await; +} + +/// During the window the key-info read finds the table, and PutItem, GetItem +/// (the pre-existing item and the one just written), Query and Scan are +/// served, as the service serves them. +#[tokio::test] +async fn data_plane_is_served_during_the_window() { + if base_conn().is_none() { + return skip("data_plane_is_served_during_the_window"); + } + let s = scratch().await; + open_window_with_item(&s).await; + let key_info = s + .engine + .table_key_info(ACCOUNT, TABLE) + .await + .expect("a DELETING table is found by the data plane"); + + s.engine + .put_item( + &key_info, + item("w1"), + false, + None, + &ExpressionMaps::default(), + None, + ) + .await + .expect("PutItem during the window"); + let got = s + .engine + .get_item(&key_info, &item("pre")) + .await + .expect("GetItem of the pre-existing item"); + assert_eq!(got, Some(item("pre"))); + let got = s + .engine + .get_item(&key_info, &item("w1")) + .await + .expect("GetItem of the window write"); + assert_eq!(got, Some(item("w1"))); + let (condition, maps) = query_pk("pre"); + let (rows, _) = s + .engine + .query(&key_info, &condition, &maps, true, None, None, None) + .await + .expect("Query during the window"); + assert_eq!(rows, vec![item("pre")]); + let (rows, _) = s + .engine + .scan(&key_info, Some(1), None, None, None, None) + .await + .expect("Scan during the window"); + assert_eq!(rows.len(), 1); + assert_eq!( + status_of(&s.catalog, TABLE).await.as_deref(), + Some("DELETING") + ); + + s.cleanup().await; +} + +/// Once the data table is dropped while the row still reads DELETING, a +/// request that reaches storage, here with the key info read before the +/// delete (the snapshot a cache would hold), answers the table as not found +/// rather than as an internal error. +#[tokio::test] +async fn a_request_after_the_data_drop_is_table_not_found() { + if base_conn().is_none() { + return skip("a_request_after_the_data_drop_is_table_not_found"); + } + let s = scratch().await; + let key_info = open_window_with_item(&s).await; + sqlx::query(&format!("DROP TABLE \"_ddb_{}\"", key_info.table_id)) + .execute(&s.catalog) + .await + .expect("drop the data table under the DELETING row"); + + let err = s + .engine + .put_item( + &key_info, + item("w1"), + false, + None, + &ExpressionMaps::default(), + None, + ) + .await + .expect_err("PutItem after the data drop"); + assert!( + matches!(&err, StorageError::TableNotFound(name) if name == TABLE), + "expected TableNotFound({TABLE:?}), got {err:?}" + ); + let err = s + .engine + .get_item(&key_info, &item("pre")) + .await + .expect_err("GetItem after the data drop"); + assert!( + matches!(&err, StorageError::TableNotFound(name) if name == TABLE), + "expected TableNotFound({TABLE:?}), got {err:?}" + ); + // The transaction paths run outside the single-request statements, so + // they are funnelled at their own entry points. + let pre = item("pre"); + let err = s + .engine + .transact_get_items(&[TransactGetOp { + key_info: &key_info, + key: &pre, + }]) + .await + .expect_err("TransactGetItems after the data drop"); + assert!( + matches!(&err, StorageError::TableNotFound(name) if name == TABLE), + "expected TableNotFound({TABLE:?}), got {err:?}" + ); + let maps = ExpressionMaps::default(); + let w2 = item("w2"); + let err = s + .engine + .transact_write_items( + &[TransactWriteOp::Put { + key_info: &key_info, + item: &w2, + condition: None, + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: None, + }], + None, + ) + .await + .expect_err("TransactWriteItems after the data drop"); + assert!( + matches!(&err, StorageError::TableNotFound(name) if name == TABLE), + "expected TableNotFound({TABLE:?}), got {err:?}" + ); + + s.cleanup().await; +} + +/// Two concurrent DeleteTable calls on one ACTIVE table: the first locks the +/// row `FOR UPDATE` and marks it DELETING; the second, blocked on that lock, +/// re-reads the row after the commit, finds it excluded by the status filter, +/// and answers TableBeingDeleted from the unfiltered lookup. Ten rounds, a +/// fresh table each; one row per table, DELETING, and its data table present. +#[tokio::test] +async fn two_concurrent_delete_tables_have_one_winner() { + if base_conn().is_none() { + return skip("two_concurrent_delete_tables_have_one_winner"); + } + let s = scratch().await; + let engine = Arc::new(s.engine.clone()); + + for round in 0..10 { + let table = format!("t_pair_{round}"); + set_delay(&s.catalog, "0").await; + create_table_named(&s, &table).await; + set_delay(&s.catalog, "600").await; + + let tasks: Vec<_> = (0..2) + .map(|_| { + let engine = Arc::clone(&engine); + let table = table.clone(); + tokio::spawn(async move { + engine + .delete_table(ACCOUNT, DeleteTableInput { table_name: table }) + .await + }) + }) + .collect(); + let mut ok = 0; + let mut being_deleted = 0; + for task in tasks { + match task.await.expect("task") { + Ok(_) => ok += 1, + Err(StorageError::TableBeingDeleted(name)) if name == table => being_deleted += 1, + Err(other) => panic!("round {round}: unexpected {other:?}"), + } + } + assert_eq!((ok, being_deleted), (1, 1), "round {round}"); + + let rows: Vec<(String, String)> = sqlx::query_as( + "SELECT table_status, table_id FROM tables WHERE account_id = $1 AND table_name = $2", + ) + .bind(ACCOUNT) + .bind(&table) + .fetch_all(&s.catalog) + .await + .expect("rows"); + assert_eq!(rows.len(), 1, "round {round}: one catalog row"); + assert_eq!(rows[0].0, "DELETING", "round {round}"); + let present: Option = sqlx::query_scalar("SELECT to_regclass($1)::text") + .bind(format!("\"_ddb_{}\"", rows[0].1)) + .fetch_one(&s.catalog) + .await + .expect("to_regclass"); + assert!( + present.is_some(), + "round {round}: the data table is still there" + ); + } + + drop(engine); + s.cleanup().await; +} diff --git a/crates/storage-sqlite/src/data/data_engine.rs b/crates/storage-sqlite/src/data/data_engine.rs index 964ea5f9..15158789 100644 --- a/crates/storage-sqlite/src/data/data_engine.rs +++ b/crates/storage-sqlite/src/data/data_engine.rs @@ -16,6 +16,49 @@ use futures::future::BoxFuture; use crate::store::SqliteEngine; +/// Answer a statement that found the table's data table gone as a missing table. +/// +/// The key info was read while the catalog row was `ACTIVE` or `DELETING`, and +/// `DeleteTable` dropped the data table before the statement ran. SQLite +/// reports the dropped table as `no such table`. DynamoDB answers a request on +/// a table that is gone with `ResourceNotFoundException`, so the error is +/// `TableNotFound` here rather than an internal error. A data-plane statement +/// also names the shared `stream_shards`, `stream_records`, `gsi_pending` and +/// `vector_index_holds` tables, and an index table a concurrent GSI delete has +/// just dropped; a missing one of those is reported the same way, so an +/// unapplied migration shows as `ResourceNotFoundException` on the wire and as +/// the named table in the server log. +fn dropped_table_as_not_found(key_info: &TableKeyInfo, e: StorageError) -> StorageError { + if super::index::is_no_such_table(&e) { + StorageError::TableNotFound(key_info.table_name.clone()) + } else { + e + } +} + +/// The transaction form of [`dropped_table_as_not_found`]: a transaction's +/// statements name every table in it, so the first operation's table stands +/// for the request. DynamoDB answers a transaction on a table that is gone +/// with `ResourceNotFoundException` without naming the table. +fn transaction_dropped_table_as_not_found( + first: Option<&TableKeyInfo>, + e: StorageError, +) -> StorageError { + match first { + Some(key_info) => dropped_table_as_not_found(key_info, e), + None => e, + } +} + +fn transact_write_op_key_info<'a>(op: &'a TransactWriteOp<'_>) -> &'a TableKeyInfo { + match op { + TransactWriteOp::Put { key_info, .. } + | TransactWriteOp::Delete { key_info, .. } + | TransactWriteOp::Update { key_info, .. } + | TransactWriteOp::ConditionCheck { key_info, .. } => key_info, + } +} + impl DataEngine for SqliteEngine { /// Declares vector support by handing over the implementation. `Some(self)` /// only compiles because `SqliteEngine` implements `VectorSearchEngine`, so @@ -47,6 +90,7 @@ impl DataEngine for SqliteEngine { stream.as_ref(), ) .await + .map_err(|e| dropped_table_as_not_found(&key_info, e)) }) } @@ -57,7 +101,11 @@ impl DataEngine for SqliteEngine { ) -> BoxFuture<'_, Result, StorageError>> { let key_info = key_info.clone(); let key = key.clone(); - Box::pin(async move { self.get_item_impl(&key_info, &key).await }) + Box::pin(async move { + self.get_item_impl(&key_info, &key) + .await + .map_err(|e| dropped_table_as_not_found(&key_info, e)) + }) } fn delete_item( @@ -84,6 +132,7 @@ impl DataEngine for SqliteEngine { stream.as_ref(), ) .await + .map_err(|e| dropped_table_as_not_found(&key_info, e)) }) } @@ -116,6 +165,7 @@ impl DataEngine for SqliteEngine { stream.as_ref(), ) .await + .map_err(|e| dropped_table_as_not_found(&key_info, e)) }) } @@ -145,6 +195,7 @@ impl DataEngine for SqliteEngine { index_name.as_deref(), ) .await + .map_err(|e| dropped_table_as_not_found(&key_info, e)) }) } @@ -170,6 +221,7 @@ impl DataEngine for SqliteEngine { index_name.as_deref(), ) .await + .map_err(|e| dropped_table_as_not_found(&key_info, e)) }) } @@ -209,7 +261,10 @@ impl DataEngine for SqliteEngine { .iter() .map(|(key_info, key)| TransactGetOp { key_info, key }) .collect(); - self.transact_get_items_impl(&borrowed).await + let first = owned.first().map(|(key_info, _)| key_info); + self.transact_get_items_impl(&borrowed) + .await + .map_err(|e| transaction_dropped_table_as_not_found(first, e)) }) } @@ -228,6 +283,7 @@ impl DataEngine for SqliteEngine { }); Box::pin(async move { let borrowed: Vec = owned.iter().map(OwnedWriteOp::as_op).collect(); + let first = borrowed.first().map(transact_write_op_key_info); self.transact_write_items_impl( &borrowed, idempotency.as_ref().map(|(a, t, f)| IdempotencyKey { @@ -237,6 +293,7 @@ impl DataEngine for SqliteEngine { }), ) .await + .map_err(|e| transaction_dropped_table_as_not_found(first, e)) }) } diff --git a/crates/storage-sqlite/src/data/ddl.rs b/crates/storage-sqlite/src/data/ddl.rs index 739e1310..bb859ae1 100644 --- a/crates/storage-sqlite/src/data/ddl.rs +++ b/crates/storage-sqlite/src/data/ddl.rs @@ -358,11 +358,17 @@ impl SqliteEngine { row.ok_or_else(|| StorageError::TableNotFound(table_name.to_owned()))?; match status.as_str() { - "ACTIVE" => {} - // A table still being created, or being deleted, is not usable for - // data-plane operations; DynamoDB reports it as not found (not - // in-use). UPDATING keeps the not-active classification. - "CREATING" | "DELETING" => { + // DynamoDB serves reads and writes on a DELETING table until the + // table is gone (measured 2026-10-25); a write made in that window + // goes away with the table. The data tables stay until the control + // plane drops them, so the data plane reads a DELETING row as found; + // a statement that reaches a dropped data table is answered as not + // found by the `DataEngine` implementation. + "ACTIVE" | "DELETING" => {} + // A table still being created is not usable for data-plane + // operations; DynamoDB reports it as not found (not in-use). + // UPDATING keeps the not-active classification. + "CREATING" => { return Err(StorageError::TableNotFound(table_name.to_owned())); } _ => return Err(StorageError::TableNotActive(table_name.to_owned())), @@ -505,11 +511,10 @@ impl SqliteEngine { let (table_id, status) = row.ok_or_else(|| StorageError::TableNotFound(table_name.to_owned()))?; match status.as_str() { - "ACTIVE" => {} - // A table still being created, or being deleted, is not usable for - // data-plane operations; DynamoDB reports it as not found (not - // in-use). UPDATING keeps the not-active classification. - "CREATING" | "DELETING" => { + // Same classification as `fetch_table_key_info`: a DELETING table + // is served until it is gone, a CREATING one is not found. + "ACTIVE" | "DELETING" => {} + "CREATING" => { return Err(StorageError::TableNotFound(table_name.to_owned())); } _ => return Err(StorageError::TableNotActive(table_name.to_owned())), diff --git a/crates/storage-sqlite/src/delete_table.rs b/crates/storage-sqlite/src/delete_table.rs index 72533abe..a62a4a4e 100644 --- a/crates/storage-sqlite/src/delete_table.rs +++ b/crates/storage-sqlite/src/delete_table.rs @@ -22,17 +22,49 @@ impl SqliteEngine { ) -> Result { Self::validate_account_id(account_id)?; + // The status lookup runs inside the write transaction, so two + // DeleteTable calls on one table are serialized: the second reads the + // row the first committed (DELETING, or gone) rather than the ACTIVE + // row both would otherwise read before either wrote. + let _writer = self.write_lock.lock().await; + let mut tx = self + .pool + .begin_with("BEGIN IMMEDIATE") + .await + .map_err(|e| StorageError::Internal(e.to_string()))?; + let row: Option = sqlx::query_as(&format!( "SELECT {TABLE_COLUMNS} FROM tables \ WHERE account_id = ? AND table_name = ? AND table_status IN ('ACTIVE', 'CREATING')" )) .bind(account_id) .bind(&input.table_name) - .fetch_optional(&self.pool) + .fetch_optional(&mut *tx) .await .map_err(|e| StorageError::Internal(e.to_string()))?; - let row = row.ok_or_else(|| StorageError::TableNotFound(input.table_name.clone()))?; + // The filtered select reads a DELETING row as absent. The service + // answers a second DeleteTable in that window as a table still in use + // (measured 2026-10-24), so look the name up again without the status + // filter before reporting it as not found. + let row = match row { + Some(row) => row, + None => { + let status: Option = sqlx::query_scalar( + "SELECT table_status FROM tables WHERE account_id = ? AND table_name = ?", + ) + .bind(account_id) + .bind(&input.table_name) + .fetch_optional(&mut *tx) + .await + .map_err(|e| StorageError::Internal(e.to_string()))?; + return Err(if status.as_deref() == Some("DELETING") { + StorageError::TableBeingDeleted(input.table_name.clone()) + } else { + StorageError::TableNotFound(input.table_name.clone()) + }); + } + }; if row.deletion_protection_enabled { return Err(StorageError::DeletionProtected(row.table_arn.clone())); } @@ -48,7 +80,7 @@ impl SqliteEngine { AND table_status = 'CREATING' AND status_transition_at IS NULL)", ) .bind(&row.table_id) - .fetch_one(&self.pool) + .fetch_one(&mut *tx) .await .map_err(|e| StorageError::Internal(e.to_string()))?; if restoring { @@ -64,14 +96,14 @@ impl SqliteEngine { "SELECT {INDEX_COLUMNS} FROM indexes WHERE table_id = ?" )) .bind(&row.table_id) - .fetch_all(&self.pool) + .fetch_all(&mut *tx) .await .map_err(|e| StorageError::Internal(e.to_string()))?; let delay_secs: f64 = sqlx::query_scalar::<_, String>( "SELECT value FROM settings WHERE key = 'control_plane_delay_seconds'", ) - .fetch_optional(&self.pool) + .fetch_optional(&mut *tx) .await .map_err(|e| StorageError::Internal(e.to_string()))? .and_then(|v| v.parse::().ok()) @@ -82,12 +114,6 @@ impl SqliteEngine { let table_arn = row.table_arn.clone(); if delay_secs < 1.0 { - let _writer = self.write_lock.lock().await; - let mut tx = self - .pool - .begin_with("BEGIN IMMEDIATE") - .await - .map_err(|e| StorageError::Internal(e.to_string()))?; // A vector index with `backfilling` set is an UpdateTable build in // flight (the with-table path never sets the column), and the // service refuses to delete the table underneath one; the refusal @@ -127,27 +153,28 @@ impl SqliteEngine { let transition = format_timestamp( time::OffsetDateTime::now_utc() + time::Duration::seconds_f64(delay_secs), ); - let _writer = self.write_lock.lock().await; - let mut tx = self - .pool - .begin_with("BEGIN IMMEDIATE") - .await - .map_err(|e| StorageError::Internal(e.to_string()))?; // Same in-flight-build refusal as the immediate branch, and inside // the transaction that flips the table to DELETING for the same // reason: without it, a build committing between an earlier check // and this flip would have its table deleted underneath it. Self::refuse_if_index_build_in_flight(&mut tx, &row.table_id).await?; - sqlx::query( + // Fenced on the table id read above, so the mark can only land on + // the table this call described, never on a table created into the + // same name afterwards. + let marked = sqlx::query( "UPDATE tables SET table_status = 'DELETING', status_transition_at = ? \ - WHERE account_id = ? AND table_name = ?", + WHERE account_id = ? AND table_name = ? AND table_id = ?", ) .bind(&transition) .bind(account_id) .bind(&input.table_name) + .bind(&table_id) .execute(&mut *tx) .await .map_err(|e| StorageError::Internal(e.to_string()))?; + if marked.rows_affected() == 0 { + return Err(StorageError::TableNotFound(input.table_name.clone())); + } tx.commit() .await .map_err(|e| StorageError::Internal(e.to_string()))?; @@ -191,3 +218,118 @@ impl SqliteEngine { Ok(()) } } + +#[cfg(test)] +mod deleting_window_tests { + use crate::SqliteEngine; + use extenddb_core::types::{DeleteTableInput, UpdateTableInput}; + use extenddb_storage::TableEngine; + use extenddb_storage::error::StorageError; + use serde_json::json; + + const ACCOUNT: &str = "000000000000"; + + /// An engine with one ACTIVE table `t`, whose catalog row is then set to + /// DELETING directly: the window a slow data drop would hold open. + async fn engine_with_deleting_table() -> SqliteEngine { + let engine = SqliteEngine::new(":memory:", 1, "us-east-1", 409_600) + .await + .expect("engine"); + crate::schema::apply(&engine.pool).await.expect("schema"); + sqlx::query("UPDATE settings SET value = '0' WHERE key = 'control_plane_delay_seconds'") + .execute(&engine.pool) + .await + .expect("delay"); + sqlx::query("INSERT INTO accounts (account_id, account_name) VALUES (?, 'default')") + .bind(ACCOUNT) + .execute(&engine.pool) + .await + .expect("account"); + let input: extenddb_core::types::CreateTableInput = serde_json::from_value(json!({ + "TableName": "t", + "KeySchema": [{"AttributeName": "pk", "KeyType": "HASH"}], + "AttributeDefinitions": [{"AttributeName": "pk", "AttributeType": "S"}], + "BillingMode": "PAY_PER_REQUEST" + })) + .expect("input"); + engine + .create_table_impl(ACCOUNT, input, false) + .await + .expect("create table"); + sqlx::query("UPDATE tables SET table_status = 'DELETING' WHERE table_name = 't'") + .execute(&engine.pool) + .await + .expect("mark DELETING"); + engine + } + + /// A second DeleteTable during the window reports the table as being + /// deleted, not as absent, and leaves the row in place. + #[tokio::test] + async fn a_second_delete_table_during_the_window_is_table_being_deleted() { + let engine = engine_with_deleting_table().await; + let err = engine + .delete_table( + ACCOUNT, + DeleteTableInput { + table_name: "t".to_owned(), + }, + ) + .await + .expect_err("DeleteTable on a DELETING table"); + assert!( + matches!(&err, StorageError::TableBeingDeleted(name) if name == "t"), + "expected TableBeingDeleted(\"t\"), got {err:?}" + ); + 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"); + } + + /// UpdateTable during the window reports the table as being deleted, + /// not with the sentence for the other non-ACTIVE statuses. + #[tokio::test] + async fn update_table_during_the_window_is_table_being_deleted() { + let engine = engine_with_deleting_table().await; + let update: UpdateTableInput = serde_json::from_value(json!({ + "TableName": "t", + "DeletionProtectionEnabled": true + })) + .expect("update"); + let err = engine + .update_table(ACCOUNT, update) + .await + .expect_err("UpdateTable on a DELETING table"); + assert!( + matches!(&err, StorageError::TableBeingDeleted(name) if name == "t"), + "expected TableBeingDeleted(\"t\"), got {err:?}" + ); + } + + /// A CREATING table keeps the existing answer: the new sentence is for + /// DELETING only. + #[tokio::test] + async fn update_table_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 update: UpdateTableInput = serde_json::from_value(json!({ + "TableName": "t", + "DeletionProtectionEnabled": true + })) + .expect("update"); + let err = engine + .update_table(ACCOUNT, update) + .await + .expect_err("UpdateTable on a CREATING table"); + assert!( + matches!(&err, StorageError::TableNotActive(name) if name == "t"), + "expected TableNotActive(\"t\"), got {err:?}" + ); + } +} diff --git a/crates/storage-sqlite/src/deleting_window_data_tests.rs b/crates/storage-sqlite/src/deleting_window_data_tests.rs new file mode 100644 index 00000000..afb850a6 --- /dev/null +++ b/crates/storage-sqlite/src/deleting_window_data_tests.rs @@ -0,0 +1,308 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Storage-level tests for the data plane while a table is `DELETING`, and +//! for two `DeleteTable` calls racing on one table. +//! +//! Amazon DynamoDB serves `PutItem`, `GetItem`, `Query` and `Scan` on a +//! `DELETING` table until the table is gone, and answers them +//! `ResourceNotFoundException` once it is (measured 2026-10-25). Two +//! `DeleteTable` calls on one `ACTIVE` table get one `200` and one +//! `ResourceInUseException` with the `Table is being deleted` sentence. +//! +//! The window is produced by setting the catalog row to `DELETING` directly, +//! as the control-plane tests in `delete_table.rs` do; the pair test lets +//! `delete_table` itself mark the row, with a delay long enough that no worker +//! finishes the deletion during the test. + +use std::collections::{BTreeMap, HashMap}; +use std::sync::Arc; + +use extenddb_core::expression::{Expr, ExpressionMaps, KeyCondition, PathElement}; +use extenddb_core::types::{ + AttributeValue, DeleteTableInput, Item, ReturnValuesOnConditionCheckFailure, TableKeyInfo, +}; +use extenddb_storage::error::StorageError; +use extenddb_storage::{DataEngine, TableEngine, TransactGetOp, TransactWriteOp}; +use serde_json::json; + +use crate::SqliteEngine; + +const ACCOUNT: &str = "000000000000"; + +fn item(pk: &str) -> Item { + BTreeMap::from([("pk".to_owned(), AttributeValue::S(pk.to_owned()))]) +} + +fn create_input(table: &str) -> extenddb_core::types::CreateTableInput { + serde_json::from_value(json!({ + "TableName": table, + "KeySchema": [{"AttributeName": "pk", "KeyType": "HASH"}], + "AttributeDefinitions": [{"AttributeName": "pk", "AttributeType": "S"}], + "BillingMode": "PAY_PER_REQUEST" + })) + .expect("input") +} + +async fn seed(engine: &SqliteEngine) { + crate::schema::apply(&engine.pool).await.expect("schema"); + sqlx::query("UPDATE settings SET value = '0' WHERE key = 'control_plane_delay_seconds'") + .execute(&engine.pool) + .await + .expect("delay"); + sqlx::query("INSERT INTO accounts (account_id, account_name) VALUES (?, 'default')") + .bind(ACCOUNT) + .execute(&engine.pool) + .await + .expect("account"); +} + +/// An in-memory engine with one table `t` holding the item `pre`, whose +/// catalog row is then set to `DELETING`: the window a slow data drop holds +/// open. Returns the engine and the key info read while the row was ACTIVE. +async fn engine_with_deleting_table() -> (SqliteEngine, TableKeyInfo) { + let engine = SqliteEngine::new(":memory:", 1, "us-east-1", 409_600) + .await + .expect("engine"); + seed(&engine).await; + engine + .create_table_impl(ACCOUNT, create_input("t"), false) + .await + .expect("create table"); + let key_info = engine.table_key_info(ACCOUNT, "t").await.expect("key info"); + engine + .put_item( + &key_info, + item("pre"), + false, + None, + &ExpressionMaps::default(), + None, + ) + .await + .expect("seed the pre-existing item"); + sqlx::query("UPDATE tables SET table_status = 'DELETING' WHERE table_name = 't'") + .execute(&engine.pool) + .await + .expect("mark DELETING"); + (engine, key_info) +} + +fn query_pk(pk: &str) -> (KeyCondition, ExpressionMaps) { + let condition = KeyCondition { + pk_path: vec![PathElement::Attribute("pk".to_owned())], + pk_value: Expr::Placeholder(":pk".to_owned()), + extra_pk_conditions: Vec::new(), + sk_condition: None, + extra_sk_conditions: Vec::new(), + }; + let maps = ExpressionMaps::new( + HashMap::new(), + HashMap::from([(":pk".to_owned(), AttributeValue::S(pk.to_owned()))]), + ); + (condition, maps) +} + +/// During the window the key-info read finds the table, and PutItem, GetItem +/// (the pre-existing item and the one just written), Query and Scan are +/// served, as the service serves them. +#[tokio::test] +async fn data_plane_is_served_during_the_window() { + let (engine, _) = engine_with_deleting_table().await; + let key_info = engine + .table_key_info(ACCOUNT, "t") + .await + .expect("a DELETING table is found by the data plane"); + + engine + .put_item( + &key_info, + item("w1"), + false, + None, + &ExpressionMaps::default(), + None, + ) + .await + .expect("PutItem during the window"); + let got = engine + .get_item(&key_info, &item("pre")) + .await + .expect("GetItem of the pre-existing item"); + assert_eq!(got, Some(item("pre"))); + let got = engine + .get_item(&key_info, &item("w1")) + .await + .expect("GetItem of the window write"); + assert_eq!(got, Some(item("w1"))); + let (condition, maps) = query_pk("pre"); + let (rows, _) = engine + .query(&key_info, &condition, &maps, true, None, None, None) + .await + .expect("Query during the window"); + assert_eq!(rows, vec![item("pre")]); + let (rows, _) = engine + .scan(&key_info, Some(1), None, None, None, None) + .await + .expect("Scan during the window"); + assert_eq!(rows.len(), 1); +} + +/// Once the data table is dropped while the row still reads DELETING, a +/// request that reaches storage answers the table as not found, not as an +/// internal error. +#[tokio::test] +async fn a_request_after_the_data_drop_is_table_not_found() { + let (engine, key_info) = engine_with_deleting_table().await; + sqlx::query(&format!( + "DROP TABLE {}", + crate::data::data_table_name(&key_info.table_id) + )) + .execute(&engine.pool) + .await + .expect("drop the data table under the DELETING row"); + + let err = engine + .put_item( + &key_info, + item("w1"), + false, + None, + &ExpressionMaps::default(), + None, + ) + .await + .expect_err("PutItem after the data drop"); + assert!( + matches!(&err, StorageError::TableNotFound(name) if name == "t"), + "expected TableNotFound(\"t\"), got {err:?}" + ); + let err = engine + .get_item(&key_info, &item("pre")) + .await + .expect_err("GetItem after the data drop"); + assert!( + matches!(&err, StorageError::TableNotFound(name) if name == "t"), + "expected TableNotFound(\"t\"), got {err:?}" + ); + // The transaction paths run outside the single-request statements, so + // they are funnelled at their own entry points. + let pre = item("pre"); + let err = engine + .transact_get_items(&[TransactGetOp { + key_info: &key_info, + key: &pre, + }]) + .await + .expect_err("TransactGetItems after the data drop"); + assert!( + matches!(&err, StorageError::TableNotFound(name) if name == "t"), + "expected TableNotFound(\"t\"), got {err:?}" + ); + let maps = ExpressionMaps::default(); + let w2 = item("w2"); + let err = engine + .transact_write_items( + &[TransactWriteOp::Put { + key_info: &key_info, + item: &w2, + condition: None, + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: None, + }], + None, + ) + .await + .expect_err("TransactWriteItems after the data drop"); + assert!( + matches!(&err, StorageError::TableNotFound(name) if name == "t"), + "expected TableNotFound(\"t\"), got {err:?}" + ); +} + +/// Two concurrent DeleteTable calls on one ACTIVE table: one marks the row +/// DELETING and answers the description, the other reads the marked row and +/// answers TableBeingDeleted. Ten rounds, a fresh table each; the row is left +/// DELETING exactly once and its data table is still present (no worker runs +/// here to finish the deletion). +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn two_concurrent_delete_tables_have_one_winner() { + let target = std::env::var_os("CARGO_TARGET_DIR") + .map(std::path::PathBuf::from) + .unwrap_or_else(|| { + std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../../target") + }); + assert!(target.is_dir(), "cargo target directory must exist"); + let path = target.join(format!("deleting-pair-{}.sqlite", uuid::Uuid::new_v4())); + let engine = Arc::new( + SqliteEngine::new(path.to_str().expect("UTF-8 path"), 4, "us-east-1", 409_600) + .await + .expect("file engine"), + ); + seed(&engine).await; + + for round in 0..10 { + let table = format!("t_pair_{round}"); + sqlx::query("UPDATE settings SET value = '0' WHERE key = 'control_plane_delay_seconds'") + .execute(&engine.pool) + .await + .expect("delay 0 so the create completes inline"); + engine + .create_table_impl(ACCOUNT, create_input(&table), false) + .await + .expect("create table"); + sqlx::query("UPDATE settings SET value = '600' WHERE key = 'control_plane_delay_seconds'") + .execute(&engine.pool) + .await + .expect("delay 600 so the delete marks the row"); + + let tasks: Vec<_> = (0..2) + .map(|_| { + let engine = Arc::clone(&engine); + let table = table.clone(); + tokio::spawn(async move { + engine + .delete_table(ACCOUNT, DeleteTableInput { table_name: table }) + .await + }) + }) + .collect(); + let mut ok = 0; + let mut being_deleted = 0; + for task in tasks { + match task.await.expect("task") { + Ok(_) => ok += 1, + Err(StorageError::TableBeingDeleted(name)) if name == table => being_deleted += 1, + Err(other) => panic!("round {round}: unexpected {other:?}"), + } + } + assert_eq!((ok, being_deleted), (1, 1), "round {round}"); + + let rows: Vec<(String, String)> = + sqlx::query_as("SELECT table_status, table_id FROM tables WHERE table_name = ?") + .bind(&table) + .fetch_all(&engine.pool) + .await + .expect("rows"); + assert_eq!(rows.len(), 1, "round {round}: one catalog row"); + assert_eq!(rows[0].0, "DELETING", "round {round}"); + let data_table = format!("_ddb_{}", rows[0].1); + let present: Option = + sqlx::query_scalar("SELECT name FROM sqlite_master WHERE type = 'table' AND name = ?") + .bind(&data_table) + .fetch_optional(&engine.pool) + .await + .expect("sqlite_master"); + assert_eq!( + present.as_deref(), + Some(data_table.as_str()), + "round {round}" + ); + } + + engine.pool.close().await; + for suffix in ["", "-wal", "-shm"] { + let _ = std::fs::remove_file(format!("{}{suffix}", path.display())); + } +} diff --git a/crates/storage-sqlite/src/lib.rs b/crates/storage-sqlite/src/lib.rs index 3953fc35..3d3e0d90 100644 --- a/crates/storage-sqlite/src/lib.rs +++ b/crates/storage-sqlite/src/lib.rs @@ -30,6 +30,8 @@ mod create_table; mod credential_store; mod data; mod delete_table; +#[cfg(test)] +mod deleting_window_data_tests; mod hooks; mod management_store; mod metadata; diff --git a/crates/storage-sqlite/src/update_table.rs b/crates/storage-sqlite/src/update_table.rs index 8adff617..d3663565 100644 --- a/crates/storage-sqlite/src/update_table.rs +++ b/crates/storage-sqlite/src/update_table.rs @@ -136,6 +136,11 @@ impl SqliteEngine { // disagree. A NULL stored value means PROVISIONED. let (status, table_id, ks_json, ad_json, stored_billing_mode) = row.ok_or_else(|| StorageError::TableNotFound(input.table_name.clone()))?; + // A DELETING table has its own sentence on the service (measured + // 2026-10-24); every other non-ACTIVE status keeps the existing one. + if status == "DELETING" { + return Err(StorageError::TableBeingDeleted(input.table_name.clone())); + } if status != "ACTIVE" { return Err(StorageError::TableNotActive(input.table_name.clone())); } diff --git a/crates/storage/src/error.rs b/crates/storage/src/error.rs index 34a93c34..9acf333b 100755 --- a/crates/storage/src/error.rs +++ b/crates/storage/src/error.rs @@ -10,6 +10,12 @@ pub enum StorageError { TableAlreadyExists(String), #[error("Table is not in ACTIVE state: {0}")] TableNotActive(String), + /// A control-plane change (a second `DeleteTable`, an `UpdateTable`) + /// refused because the table is `DELETING`. Distinct from + /// [`Self::TableNotActive`] because the service words this status + /// differently from the others; the engine owns the sentence. + #[error("Table is being deleted: {0}")] + TableBeingDeleted(String), /// A table name taken by a `CREATING` restore target. Raised by /// `RestoreTableFromBackup` into that name (the engine answers /// `TableInUseException`), distinct from [`Self::TableAlreadyExists`]. diff --git a/docs/differences-from-dynamodb.md b/docs/differences-from-dynamodb.md index 2d6469b7..9c8493e5 100755 --- a/docs/differences-from-dynamodb.md +++ b/docs/differences-from-dynamodb.md @@ -48,6 +48,7 @@ adaptation when switching between ExtendDB and the real service. |------|----------|------| | Table creation delay | Returns `CREATING` immediately; transitions to `ACTIVE` typically within seconds. Same behavior for on-demand and provisioned | Configurable via `control_plane_delay_seconds` runtime setting (default: 5s) | | DeletionProtectionEnabled | Enforced | Enforced (accepted and stored, DeleteTable rejects when enabled) | +| Requests during a table's `DELETING` window | A second `DeleteTable` and an `UpdateTable` answer `ResourceInUseException` `Attempt to change a resource which is still in use: Table is being deleted: `; `PutItem`, `GetItem`, `Query` and `Scan` are served until the table is gone, and a write made in the window goes away with the table (measured 2026-10-25) | The same on PostgreSQL and SQLite, where `DeleteTable` marks the row `DELETING` on the asynchronous path (`control_plane_delay_seconds` of 1 or more) and the data tables stay until the control plane drops them; a request that reaches a data table after the drop answers `ResourceNotFoundException`. On MongoDB the collections are dropped inside the `DeleteTable` call a few milliseconds after the document is marked, so the data plane reads a `DELETING` document as not found (`ResourceNotFoundException`); serving a write there would recreate the collection after its drop. On Cassandra `DeleteTable` is synchronous and never marks the row `DELETING`; the only `DELETING` row on that backend is an abandoned restore target under the restore sweep, whose data is partial, and the data plane reads it as not found | ## Time to Live (TTL) diff --git a/tests/python/test_deleting_window_control_plane.py b/tests/python/test_deleting_window_control_plane.py new file mode 100644 index 00000000..b6837856 --- /dev/null +++ b/tests/python/test_deleting_window_control_plane.py @@ -0,0 +1,189 @@ +# Copyright 2026 ExtendDB contributors +# SPDX-License-Identifier: Apache-2.0 + +"""Requests during a table's DELETING window. + +Amazon DynamoDB keeps a table describable as ``TableStatus DELETING`` for a +few seconds after ``DeleteTable`` returns. In that window a second +``DeleteTable`` and an ``UpdateTable`` answer ``ResourceInUseException`` with +``Attempt to change a resource which is still in use: Table is being deleted: +`` (measured 2026-10-24), and ``PutItem``, ``GetItem``, ``Query`` and +``Scan`` are served; once the table is gone, ``GetItem`` of an item written in +the window is ``ResourceNotFoundException`` (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 +drop waits and the catalog row stays ``DELETING``. That needs +``EXTENDDB_TEST_PG_ADMIN_CONNECTION_STRING`` (a base URL with no database +component) and the catalog database name (``EXTENDDB_TEST_PG_CATALOG_DB``, +default ``extenddb_catalog``), and a server whose ``DeleteTable`` marks the row +``DELETING`` before the data drop. Before #504 that is the asynchronous path +only, taken when ``control_plane_delay_seconds`` is 1 or more; on the +immediate path the catalog row is removed first and the drop waits on the held +lock inside the request, so the tests read the server's delay from the +``settings`` table and skip when it is under 1. #504 marks the row on both +paths and removes that condition. Without the connection string, or on a +backend other than PostgreSQL (``EXTENDDB_TEST_BACKEND``, exported by +``devtools/run-tests --backend``), these tests skip on ExtendDB; the +storage-level tests in each backend crate carry the rows there. +""" + +from __future__ import annotations + +import os + +import pytest +from botocore.exceptions import ClientError + +from helpers import unique_name, wait_for_active, wait_for_deleted + +PG_ADMIN = os.environ.get("EXTENDDB_TEST_PG_ADMIN_CONNECTION_STRING", "").rstrip("/") +CATALOG_DB = os.environ.get("EXTENDDB_TEST_PG_CATALOG_DB", "extenddb_catalog") +BACKEND = os.environ.get("EXTENDDB_TEST_BACKEND", "").strip() + + +def _being_deleted(name: str) -> str: + return f"Attempt to change a resource which is still in use: Table is being deleted: {name}" + + +def _catalog_setting(key: str) -> str | None: + import psycopg2 + + with psycopg2.connect(f"{PG_ADMIN}/{CATALOG_DB}") as catalog, catalog.cursor() as cur: + cur.execute("SELECT value FROM settings WHERE key = %s", (key,)) + row = cur.fetchone() + return row[0] if row else None + + +class _HeldLock: + """``ACCESS SHARE`` on ``_ddb_``, held until ``release``.""" + + def __init__(self, table_id: str): + import psycopg2 + + data_db = _catalog_setting("data_database_name") or CATALOG_DB + self.conn = psycopg2.connect(f"{PG_ADMIN}/{data_db}") + cur = self.conn.cursor() + cur.execute(f'LOCK TABLE "_ddb_{table_id}" IN ACCESS SHARE MODE') + + def release(self) -> None: + self.conn.rollback() + self.conn.close() + + +def _skip_unless_window_can_be_held(dynamodb_client, name: str) -> None: + """Skip, after removing the table, when this deployment cannot hold the window.""" + reason = None + if BACKEND and BACKEND != "postgres": + reason = f"the window is held through PostgreSQL only (EXTENDDB_TEST_BACKEND={BACKEND})" + elif not PG_ADMIN: + reason = "EXTENDDB_TEST_PG_ADMIN_CONNECTION_STRING is not set; cannot hold the table DELETING" + else: + delay = _catalog_setting("control_plane_delay_seconds") + try: + delay_secs = float(delay) if delay is not None else 0.25 + except ValueError: + delay_secs = 0.25 + if delay_secs < 1.0: + reason = ( + f"control_plane_delay_seconds is {delay}: DeleteTable removes the catalog row " + "before the drop, so a held lock blocks the request rather than holding the row " + "DELETING (#504 removes this condition)" + ) + if reason is not None: + dynamodb_client.delete_table(TableName=name) + 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.""" + dynamodb_client.create_table( + TableName=name, + AttributeDefinitions=[{"AttributeName": "pk", "AttributeType": "S"}], + KeySchema=[{"AttributeName": "pk", "KeyType": "HASH"}], + BillingMode="PAY_PER_REQUEST", + ) + wait_for_active(dynamodb_client, name) + if not is_extenddb: + return None + _skip_unless_window_can_be_held(dynamodb_client, name) + table_id = dynamodb_client.describe_table(TableName=name)["Table"]["TableId"] + return _HeldLock(table_id) + + +def _assert_in_use(exc: ClientError, name: str) -> None: + err = exc.response["Error"] + assert err["Code"] == "ResourceInUseException" + assert err["Message"] == _being_deleted(name) + + +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.""" + name = unique_name("del-window") + lock = _open_window(dynamodb_client, is_extenddb, name) + try: + resp = dynamodb_client.delete_table(TableName=name) + assert resp["TableDescription"]["TableStatus"] == "DELETING" + with pytest.raises(ClientError) as exc: + dynamodb_client.delete_table(TableName=name) + _assert_in_use(exc.value, name) + finally: + if lock is not None: + lock.release() + wait_for_deleted(dynamodb_client, name, timeout=120.0) + + def test_update_table_is_being_deleted(self, dynamodb_client, is_extenddb): + """UpdateTable while the table is DELETING is ResourceInUseException with the DELETING sentence.""" + 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.update_table(TableName=name, DeletionProtectionEnabled=True) + _assert_in_use(exc.value, name) + finally: + if lock is not None: + lock.release() + wait_for_deleted(dynamodb_client, name, timeout=120.0) + + +class TestDeletingWindowDataPlane: + def test_data_plane_is_served_until_the_table_is_gone(self, dynamodb_client, is_extenddb): + """PutItem, GetItem, Query and Scan are served while the table is DELETING; + GetItem of the item written in the window is ResourceNotFoundException once + the table is gone.""" + name = unique_name("del-window") + lock = _open_window(dynamodb_client, is_extenddb, name) + try: + dynamodb_client.put_item(TableName=name, Item={"pk": {"S": "pre"}}) + resp = dynamodb_client.delete_table(TableName=name) + assert resp["TableDescription"]["TableStatus"] == "DELETING" + + dynamodb_client.put_item(TableName=name, Item={"pk": {"S": "w1"}}) + got = dynamodb_client.get_item( + TableName=name, Key={"pk": {"S": "pre"}}, ConsistentRead=True + ) + assert got["Item"] == {"pk": {"S": "pre"}} + got = dynamodb_client.get_item( + TableName=name, Key={"pk": {"S": "w1"}}, ConsistentRead=True + ) + assert got["Item"] == {"pk": {"S": "w1"}} + queried = dynamodb_client.query( + TableName=name, + KeyConditionExpression="pk = :pk", + ExpressionAttributeValues={":pk": {"S": "pre"}}, + ConsistentRead=True, + ) + assert queried["Items"] == [{"pk": {"S": "pre"}}] + scanned = dynamodb_client.scan(TableName=name, Limit=1, ConsistentRead=True) + assert len(scanned["Items"]) == 1 + finally: + if lock is not None: + lock.release() + wait_for_deleted(dynamodb_client, name, timeout=120.0) + + 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"