From db7b3dff4e28d11efc8be2250cf3eb420cca310c Mon Sep 17 00:00:00 2001 From: Lee Hannigan Date: Sun, 25 Oct 2026 12:26:26 +0000 Subject: [PATCH 1/2] fix: answer requests during a table's DELETING window as the service does A second DeleteTable and an UpdateTable during a table's DELETING window answer 400 ResourceInUseException with "Attempt to change a resource which is still in use: Table is being deleted: " on all four backends. PutItem, GetItem, Query and Scan are served during the window on PostgreSQL and SQLite and answer ResourceNotFoundException once the table is gone; a write made in the window goes away with the table. Two concurrent DeleteTable calls on one ACTIVE table get one description and one ResourceInUseException on PostgreSQL, SQLite and MongoDB. All measured against the service on 2026-10-24 and 2026-10-25. DeleteTable on PostgreSQL and SQLite looked the name up with a status filter that excluded DELETING rows and answered ResourceNotFoundException. On MongoDB and Cassandra it had no status filter and ran the delete again. UpdateTable on every backend answered "Table is not in ACTIVE state", the sentence for the other non-ACTIVE statuses. The data plane on PostgreSQL and SQLite answered ResourceNotFoundException for the whole window, because the key-info read that authorization and the handlers share mapped a DELETING row to not found. On SQLite the DeleteTable status lookup ran outside the write transaction and on MongoDB the DELETING mark was unconditioned, so two concurrent calls both proceeded. New StorageError::TableBeingDeleted, worded once in the engine's message table and mapped to ResourceInUseException by the DeleteTable, UpdateTable and TTL error mappers. The key-info read on PostgreSQL and SQLite reads a DELETING row as found; a statement or transaction that reaches a data table the control plane has dropped is answered as not found instead of as an internal error. SQLite's DeleteTable looks the row up inside its write transaction and fences the DELETING mark on the table id; MongoDB's mark is an update_one conditioned on the table id and status, and its final delete_one is keyed on both. On PostgreSQL and SQLite the DELETING row exists only on the asynchronous DeleteTable path (control_plane_delay_seconds of 1 or more; the default is 0.25), so on a default deployment the new answers are reachable after #504, which marks the row on both paths. On Cassandra DeleteTable never marks the row, so the check is reachable only for an abandoned restore target, and the data plane there is unchanged. MongoDB drops the collections inside the DeleteTable call a few milliseconds after the mark, so its data plane keeps answering not found during that window. The differences document carries the per-backend shape. Tests: engine unit tests for the three mappers; storage-level tests on each backend for both control-plane requests and for a CREATING table keeping its answer; on PostgreSQL and SQLite, for the four data-plane requests during the window, for a put, a get and the two transaction calls after the data table is dropped under a DELETING row, and for ten rounds of two concurrent DeleteTable calls (also on MongoDB); a dual-target wire test for the PostgreSQL case that holds the data table locked from a second connection, with the data-plane rows, which skips on a server whose delay is under 1 s until #504 lands. The PostgreSQL storage-level CI step runs the new binary. Closes #505 --- .github/workflows/integration.yml | 11 +- crates/core/src/error/messages.rs | 10 + crates/engine/src/create_table.rs | 3 + crates/engine/src/delete_table.rs | 25 + crates/engine/src/ttl.rs | 28 + crates/engine/src/update_table.rs | 30 + crates/server/src/key_info_cache.rs | 9 +- crates/storage-cassandra/src/delete_table.rs | 14 +- crates/storage-cassandra/src/update_table.rs | 5 + .../tests/direct/deleting_window.rs | 145 +++++ .../tests/direct_integration.rs | 2 + .../src/deleting_window_tests.rs | 296 +++++++++ crates/storage-mongodb/src/table_engine.rs | 58 +- crates/storage-postgres/src/data/ddl.rs | 25 +- .../storage-postgres/src/data/delete_item.rs | 32 +- crates/storage-postgres/src/data/put_item.rs | 32 +- crates/storage-postgres/src/data/query.rs | 12 +- .../storage-postgres/src/data/tx_helpers.rs | 2 +- .../storage-postgres/src/data/update_item.rs | 18 +- crates/storage-postgres/src/delete_table.rs | 23 +- crates/storage-postgres/src/update_table.rs | 5 + .../storage-postgres/tests/deleting_window.rs | 598 ++++++++++++++++++ crates/storage-sqlite/src/data/data_engine.rs | 61 +- crates/storage-sqlite/src/data/ddl.rs | 25 +- crates/storage-sqlite/src/delete_table.rs | 180 +++++- .../src/deleting_window_data_tests.rs | 308 +++++++++ crates/storage-sqlite/src/lib.rs | 2 + crates/storage-sqlite/src/update_table.rs | 5 + crates/storage/src/error.rs | 6 + docs/differences-from-dynamodb.md | 1 + .../test_deleting_window_control_plane.py | 189 ++++++ 31 files changed, 2030 insertions(+), 130 deletions(-) create mode 100644 crates/storage-cassandra/tests/direct/deleting_window.rs create mode 100644 crates/storage-mongodb/src/deleting_window_tests.rs create mode 100644 crates/storage-postgres/tests/deleting_window.rs create mode 100644 crates/storage-sqlite/src/deleting_window_data_tests.rs create mode 100644 tests/python/test_deleting_window_control_plane.py diff --git a/.github/workflows/integration.yml b/.github/workflows/integration.yml index 8a4818f75..e5efba1c6 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 18d3e521d..e5aedc057 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 424e27b55..e3bad2a30 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 4c48e4239..022c38793 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 fac027d4b..ed1e509bc 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 6db763f82..6cba5c610 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 ba527f659..ddd42040a 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 b1a5b426e..479bb86c6 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 4a20b8618..334f34eb0 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 000000000..4b289870d --- /dev/null +++ b/crates/storage-cassandra/tests/direct/deleting_window.rs @@ -0,0 +1,145 @@ +// 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. + +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"; + +async fn set_status(engine: &CassandraEngine, account_id: &str, status: &str) { + let query = format!( + "UPDATE {}.tables SET table_status = ? 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"); +} + +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") + ); +} + +/// 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:?}" + ); +} + +/// 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:?}" + ); +} diff --git a/crates/storage-cassandra/tests/direct_integration.rs b/crates/storage-cassandra/tests/direct_integration.rs index 43eb28b81..5826be230 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 000000000..c8aeb44a2 --- /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 aa8c5272c..27d37ef16 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 f608c89e8..d12c6a890 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 9a6dfda93..163cc3dae 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 2fbc3102b..34fdf3010 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 6a84dc1b7..5e20ce3a1 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 6468f159b..60c09220e 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 80ab9a515..8d1541b81 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 7475eed6a..b760715db 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 dec99a93b..814e736ff 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 000000000..a53c43865 --- /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 964ea5f96..151587896 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 739e1310e..bb859ae11 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 72533abea..a62a4a4e1 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 000000000..afb850a68 --- /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 3953fc350..3d3e0d906 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 8adff6173..d3663565d 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 34a93c340..9acf333b7 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 2d6469b7e..9c8493e55 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 000000000..b6837856f --- /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" From b0ca08567a4bec3186cc033eee4da9f834181682 Mon Sep 17 00:00:00 2001 From: Lee Hannigan Date: Sun, 25 Oct 2026 16:55:10 +0000 Subject: [PATCH 2/2] test(cassandra): keep the DELETING-window rows out of the control-plane sweep The three deleting_window tests set a fresh table's catalog row to DELETING to produce the window. A fresh table carries its activation deadline in status_transition_at, and process_control_plane_transitions removes any DELETING row whose deadline has passed: it clears the table's TTL entries in the account keyspace, drops the indexes and the data table, and deletes the row. The tests left their rows in place, so the next test in the binary to run the sweep after their account keyspaces were dropped failed on the missing keyspace. On CI that was index::test_ttl_on_a_table_with_an_async_gsi_is_refused_as_a_validation_error in the cassandra-storage job. The sequence reproduces locally with --test-threads=1 on deleting_window, groups, then that test. set_status now clears status_transition_at together with the status, the column activation clears, so the sweep's deadline filter excludes the row for the rest of the test. Each test deletes its row before it returns. The same sequence passes with the change, the full direct_integration binary passes in parallel (152 of 152), and the catalog holds no DeletingWindowTable row afterwards. --- .../tests/direct/deleting_window.rs | 33 ++++++++++++++++++- 1 file changed, 32 insertions(+), 1 deletion(-) diff --git a/crates/storage-cassandra/tests/direct/deleting_window.rs b/crates/storage-cassandra/tests/direct/deleting_window.rs index 4b289870d..cc43f7da4 100644 --- a/crates/storage-cassandra/tests/direct/deleting_window.rs +++ b/crates/storage-cassandra/tests/direct/deleting_window.rs @@ -11,6 +11,17 @@ //! 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; @@ -21,9 +32,12 @@ 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 = ? WHERE account_id = ? AND table_name = ? IF EXISTS", + "UPDATE {}.tables SET table_status = ?, status_transition_at = null \ + WHERE account_id = ? AND table_name = ? IF EXISTS", engine.catalog_keyspace() ); engine @@ -33,6 +47,20 @@ async fn set_status(engine: &CassandraEngine, account_id: &str, status: &str) { .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!( @@ -103,6 +131,7 @@ async fn a_second_delete_table_during_the_window_is_table_being_deleted() { 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 @@ -122,6 +151,7 @@ async fn update_table_during_the_window_is_table_being_deleted() { 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 @@ -142,4 +172,5 @@ async fn update_table_on_a_creating_table_keeps_table_not_active() { matches!(&err, StorageError::TableNotActive(name) if name == TABLE), "expected TableNotActive({TABLE:?}), got {err:?}" ); + remove_row(&engine, &account.account_id).await; }