diff --git a/crates/storage-cassandra/src/backup_engine.rs b/crates/storage-cassandra/src/backup_engine.rs index 7c070934..f31bc5ce 100755 --- a/crates/storage-cassandra/src/backup_engine.rs +++ b/crates/storage-cassandra/src/backup_engine.rs @@ -1103,7 +1103,8 @@ impl CassandraEngine { /// Remove restore targets whose copy is gone: a CREATING or DELETING /// target carrying a restore registration whose heartbeat is older than - /// `stale_after` (or absent). Returns the names removed. + /// `stale_after` (or absent), and no `status_transition_at`. Returns the + /// names removed. /// /// Each removal first claims the target with a lightweight transaction /// conditioned on the heartbeat it observed, setting DELETING, a fresh @@ -1130,7 +1131,8 @@ impl CassandraEngine { // CREATING tables; restore targets are few and short-lived. let query = format!( "SELECT account_id, table_name, table_id, table_status, \ - restore_source_backup_arn, restore_heartbeat FROM {catalog}.tables \ + restore_source_backup_arn, restore_heartbeat, \ + toUnixTimestamp(status_transition_at) AS deadline_ms FROM {catalog}.tables \ WHERE table_status = ? ALLOW FILTERING" ); let rows = crate::cassandra_util::query_rows::( @@ -1146,6 +1148,15 @@ impl CassandraEngine { let Some(backup_arn) = backup_arn else { continue; }; + // A restore target never carries a deadline, and a DeleteTable + // claim always writes one, so a row with a deadline belongs to + // that call or to the control-plane sweep that took it over. A + // restored table keeps its registration columns after + // activation, which is why the status alone cannot tell. + let deadline: Result, _> = row.get_by_name("deadline_ms"); + if !matches!(deadline, Ok(None)) { + continue; + } let heartbeat: Option = row.get_by_name("restore_heartbeat").ok().flatten(); if heartbeat.is_some_and(|at| at >= cutoff) { continue; diff --git a/crates/storage-cassandra/src/data/index.rs b/crates/storage-cassandra/src/data/index.rs index 4ebd72da..23ad7f58 100644 --- a/crates/storage-cassandra/src/data/index.rs +++ b/crates/storage-cassandra/src/data/index.rs @@ -572,11 +572,11 @@ pub(crate) fn sk_to_value(sk: &SortKeyValue) -> cdrs_tokio::types::value::Value } } -/// Delete index metadata and drop index data tables for a given table. +/// Drop the index data tables of a table, then delete their index metadata. /// -/// Called by: -/// - `process_control_plane_transitions` when table is DELETING → deleted -/// - `update_table` when GSI is deleted via `UpdateTable` API +/// Called by `remove_claimed_table`, the removal both `DeleteTable` and the +/// control-plane sweep run. The tables go first, so a drop that fails leaves +/// the metadata naming the index ids for the next attempt to find. /// /// Returns list of index IDs that were deleted (for caller logging/tracking). pub(crate) async fn delete_indexes_for_table( @@ -603,6 +603,13 @@ pub(crate) async fn delete_indexes_for_table( index_ids.push(index_id); } + // Drop index data tables + for index_id in &index_ids { + engine + .drop_index_data_table(account_keyspace, index_id) + .await?; + } + // Delete index metadata from catalog let delete_query = format!("DELETE FROM {catalog_keyspace}.indexes WHERE table_id = ?"); @@ -614,13 +621,6 @@ pub(crate) async fn delete_indexes_for_table( ) .await?; - // Drop index data tables - for index_id in &index_ids { - engine - .drop_index_data_table(account_keyspace, index_id) - .await?; - } - Ok(index_ids) } diff --git a/crates/storage-cassandra/src/delete_table.rs b/crates/storage-cassandra/src/delete_table.rs index 479bb86c..26375a15 100644 --- a/crates/storage-cassandra/src/delete_table.rs +++ b/crates/storage-cassandra/src/delete_table.rs @@ -4,11 +4,37 @@ //! `delete_table` implementation for `CassandraEngine`. use cdrs_tokio::types::IntoRustByName; -use extenddb_core::types::{DeleteTableInput, TableDescription}; +use extenddb_core::types::{DeleteTableInput, TableDescription, TableStatus}; use extenddb_storage::error::StorageError; use crate::CassandraEngine; +/// How far ahead of a `DeleteTable` claim, or of a renewal of it, the row's +/// `status_transition_at` is set. +/// +/// The control-plane sweep (`process_control_plane_transitions`) selects a +/// `DELETING` row only once that deadline has passed, and takes it with +/// [`CassandraEngine::renew_delete_lease`] from the deadline it read. Until +/// then the row is left to the call that claimed it. A call that stops after +/// the claim, because its process dies or a statement after the claim fails, +/// leaves the row to the sweep, which removes the table after the deadline; +/// with no deadline the sweep would never select the row, and the name would +/// stay taken. A removal that stops part way, the call's or the sweep's, +/// keeps the catalog row, since the row delete is its last statement, and a +/// later pass takes the row over once the deadline written by the last +/// claim or renewal has passed. +/// +/// The length decides how long a stopped call keeps the name refused, and +/// how slow a call can be before the sweep takes the row over and runs the +/// remaining drops a second time. The renewal before the name-keyed deletes +/// is what keeps two actors apart: an actor whose renewal does not apply +/// leaves those deletes and the row delete to the actor that holds the row. +const DELETE_CLAIM_LEASE_SECS: i64 = 300; + +fn lease_deadline_ms() -> i64 { + (chrono::Utc::now() + chrono::Duration::seconds(DELETE_CLAIM_LEASE_SECS)).timestamp_millis() +} + /// Who is deleting a table, which decides what the delete refuses. #[derive(Clone, Copy)] pub(crate) enum DeleteTableCaller<'a> { @@ -35,8 +61,103 @@ impl CassandraEngine { account_id: &str, input: DeleteTableInput, ) -> Result { - self.delete_table_as(account_id, input, DeleteTableCaller::Client) - .await + let description = self + .delete_table_as(account_id, input, DeleteTableCaller::Client) + .await?; + // The row this call claimed is DELETING while the drops below it run, + // which is the status the service reports in the DeleteTable response + // and the status the other three backends report. + Ok(TableDescription { + table_status: TableStatus::Deleting, + ..description + }) + } + + /// Claims the catalog row for `DeleteTable`: sets `table_status` to + /// `DELETING` and `status_transition_at` to [`DELETE_CLAIM_LEASE_SECS`] + /// from now, if the row still holds `table_id` and is not `DELETING` + /// already. Returns whether the claim applied; it does not when another + /// `DeleteTable` holds the row or the row is gone. + /// + /// The table id in the condition keeps a stalled call from claiming a + /// table created into the same name after the one it read was gone. The + /// condition is the one the MongoDB backend's conditioned mark uses. + /// + /// # Errors + /// + /// Returns [`StorageError::Internal`] when the lightweight transaction + /// fails or its answer cannot be read. + #[doc(hidden)] + pub async fn claim_table_for_delete( + &self, + account_id: &str, + table_name: &str, + table_id: &str, + ) -> Result { + Ok(self + .claim_delete_lease(account_id, table_name, table_id) + .await? + .is_some()) + } + + /// [`Self::claim_table_for_delete`], returning the deadline it wrote, or + /// `None` when the claim did not apply. + pub(crate) async fn claim_delete_lease( + &self, + account_id: &str, + table_name: &str, + table_id: &str, + ) -> Result, StorageError> { + let lease_until = lease_deadline_ms(); + let claim = format!( + "UPDATE {}.tables \ + SET table_status = 'DELETING', status_transition_at = ? \ + WHERE account_id = ? AND table_name = ? \ + IF table_id = ? AND table_status != 'DELETING'", + self.catalog_keyspace() + ); + let result = crate::cassandra_util::query_lwt( + &self.session, + &claim, + cdrs_tokio::query_values!(lease_until, account_id, table_name, table_id), + ) + .await?; + Ok(crate::cassandra_util::lwt_applied(&result)?.then_some(lease_until)) + } + + /// Moves a claimed row's deadline from `held_ms` to + /// [`DELETE_CLAIM_LEASE_SECS`] from now, if the row still holds + /// `table_id`, still reads `DELETING` and still carries `held_ms`. + /// Returns the new deadline, or `None` when the swap did not apply. + /// + /// The control-plane sweep takes a row with it, from the deadline its + /// select returned; an actor renews its own claim with it before the + /// name-keyed deletes. Whoever's swap applies holds the row until the new + /// deadline. A sweeper whose select is stale, a second sweeper, and a call + /// whose row was taken over all find the deadline changed (or the row + /// gone, or holding another table id) and stop. + #[doc(hidden)] + pub async fn renew_delete_lease( + &self, + account_id: &str, + table_name: &str, + table_id: &str, + held_ms: i64, + ) -> Result, StorageError> { + let lease_until = lease_deadline_ms(); + let swap = format!( + "UPDATE {}.tables SET status_transition_at = ? \ + WHERE account_id = ? AND table_name = ? \ + IF table_id = ? AND table_status = 'DELETING' AND status_transition_at = ?", + self.catalog_keyspace() + ); + let result = crate::cassandra_util::query_lwt( + &self.session, + &swap, + cdrs_tokio::query_values!(lease_until, account_id, table_name, table_id, held_ms), + ) + .await?; + Ok(crate::cassandra_util::lwt_applied(&result)?.then_some(lease_until)) } pub(crate) async fn delete_table_as( @@ -80,8 +201,8 @@ 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 + // (a concurrent DeleteTable, or the abandoned-restore sweep). + // 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 = @@ -120,7 +241,9 @@ impl CassandraEngine { } } - // Fetch indexes for response + // The response is built before the claim: these two steps only read, + // so a failure here leaves the table as it was rather than claimed + // for a removal the client was told failed. let index_query = format!("SELECT * FROM {catalog_keyspace}.indexes WHERE table_id = ?"); let index_result = self @@ -135,45 +258,92 @@ impl CassandraEngine { let index_rows = index_body.into_rows().unwrap_or_default(); - // Build description for response (before deletion) let description = - self.build_table_description_from_row(account_id, table_row, index_rows.clone())?; + self.build_table_description_from_row(account_id, table_row, index_rows)?; - // Delete physical tables from the owning account keyspace. Catalog - // metadata and user data intentionally live in different keyspaces. - // - // The base table is dropped *before* the TTL queue is cleared. Clearing - // first would leave a still-live, still-TTL-enabled table whose items - // have no expiration entries and nothing to rebuild them if the drop - // below fails. + // Claim the row before the first drop. Two concurrent calls both read + // the same row; the claim applies for one of them and the other + // answers TableBeingDeleted, as the service answers the second of two + // concurrent calls (measured 2026-10-25, five rounds of five). + let lease = match caller { + DeleteTableCaller::Client => Some( + self.claim_delete_lease(account_id, &input.table_name, &table_id) + .await? + .ok_or_else(|| StorageError::TableBeingDeleted(input.table_name.clone()))?, + ), + DeleteTableCaller::RestoreCleanup { .. } => None, + }; + + // When the renewal inside does not apply, this call outlived its lease + // and the control-plane sweep took the row over; the sweep finishes + // the removal and the call answers with the description all the same. + self.remove_claimed_table( + account_id, + &input.table_name, + &table_id, + &description.table_arn, + lease, + ) + .await?; + Ok(description) + } + + /// Removes the table `table_id` whose catalog row this actor holds: + /// `lease` is the deadline its claim (a `DeleteTable` call) or its take + /// (the control-plane sweep) wrote, and `None` for a restore target, which + /// carries no deadline. Returns `false`, having changed nothing keyed on + /// the name, when the renewal of `lease` does not apply. + /// + /// Every statement up to the renewal is keyed on the table id and can run + /// twice: the base data table and the index data tables are dropped `IF + /// EXISTS`, the TTL entries are cleared, and the index metadata goes after + /// the index tables, so a removal that stops part way leaves the ids the + /// next attempt needs. The `continuous_backups` row is keyed on the table + /// name and the tags on the table ARN; neither carries a table id. They + /// are deleted last, after the renewal, while the catalog row still holds + /// the name, so a `CreateTable` of the name is refused until the row + /// delete releases it, and a tag or PITR write that landed during the + /// drops is removed with them. The renewal is the fence: an actor whose + /// lease was taken over, or whose row was removed or created again under + /// the name, stops before those deletes. It holds while the three + /// statements after the renewal finish within the lease it wrote. + pub(crate) async fn remove_claimed_table( + &self, + account_id: &str, + table_name: &str, + table_id: &str, + table_arn: &str, + lease: Option, + ) -> Result { + let catalog_keyspace = self.catalog_keyspace(); + // Catalog metadata and user data intentionally live in different + // keyspaces. let account_keyspace = self.account_keyspace(account_id); - self.drop_data_table(&account_keyspace, &table_id) - .await - .map_err(|e| StorageError::Internal(e.to_string()))?; - self.clear_ttl_entries_for_table_id(account_id, &table_id) + + // The base table is dropped *before* the TTL queue is cleared. Clearing + // first would leave a table that still holds items with no expiration + // entries and nothing to rebuild them if the drop below fails. + self.drop_data_table(&account_keyspace, table_id).await?; + self.clear_ttl_entries_for_table_id(account_id, table_id) .await?; + crate::data::index::delete_indexes_for_table( + &self.session_arc(), + &catalog_keyspace, + &account_keyspace, + table_id, + self, + ) + .await?; - // Delete index data tables - for idx_row in &index_rows { - let index_id: String = idx_row - .get_r_by_name("index_id") - .map_err(|e| StorageError::Internal(format!("Parse index_id: {e}")))?; - self.drop_index_data_table(&account_keyspace, &index_id) - .await - .map_err(|e| StorageError::Internal(format!("Drop index table: {e}")))?; + if let Some(held) = lease + && self + .renew_delete_lease(account_id, table_name, table_id, held) + .await? + .is_none() + { + return Ok(false); } - // Delete catalog entries (indexes first due to FK) - let delete_indexes_query = - format!("DELETE FROM {catalog_keyspace}.indexes WHERE table_id = ?"); - self.session - .query_with_values( - &delete_indexes_query, - cdrs_tokio::query_values!(table_id.as_str()), - ) - .await - .map_err(|e| StorageError::Internal(format!("Delete indexes: {e}")))?; - // Delete table-scoped continuous-backup configuration so recreating the // same table name does not inherit stale PITR state. let delete_continuous_backup_query = format!( @@ -182,7 +352,7 @@ impl CassandraEngine { self.session .query_with_values( &delete_continuous_backup_query, - cdrs_tokio::query_values!(account_id, input.table_name.as_str()), + cdrs_tokio::query_values!(account_id, table_name), ) .await .map_err(|e| StorageError::Internal(format!("Delete continuous backup state: {e}")))?; @@ -191,26 +361,24 @@ impl CassandraEngine { let delete_tags_query = format!("DELETE FROM {catalog_keyspace}.tags WHERE resource_arn = ?"); self.session - .query_with_values( - &delete_tags_query, - cdrs_tokio::query_values!(description.table_arn.as_str()), - ) + .query_with_values(&delete_tags_query, cdrs_tokio::query_values!(table_arn)) .await .map_err(|e| StorageError::Internal(format!("Delete tags: {e}")))?; - // Delete table catalog entry — LWT to avoid mixing plain/LWT writes on - // a row created with IF NOT EXISTS. See ADR-0021. + // Delete table catalog entry, only if it is still this incarnation. A + // lightweight transaction, to avoid mixing plain and LWT writes on a + // row created with IF NOT EXISTS. See ADR-0021. let delete_table_query = format!( "DELETE FROM {catalog_keyspace}.tables WHERE account_id = ? AND table_name = ? IF table_id = ?" ); crate::cassandra_util::query_lwt( &self.session, &delete_table_query, - cdrs_tokio::query_values!(account_id, input.table_name.as_str(), table_id.as_str()), + cdrs_tokio::query_values!(account_id, table_name, table_id), ) .await .map_err(|e| StorageError::Internal(format!("Delete table: {e}")))?; - Ok(description) + Ok(true) } } diff --git a/crates/storage-cassandra/src/worker_store.rs b/crates/storage-cassandra/src/worker_store.rs index f89422c9..21ded866 100755 --- a/crates/storage-cassandra/src/worker_store.rs +++ b/crates/storage-cassandra/src/worker_store.rs @@ -26,8 +26,10 @@ impl CassandraEngine { /// Process pending control plane transitions (H-5). /// /// Tables in CREATING state whose `status_transition_at` has passed are - /// moved to ACTIVE. Tables in DELETING state whose transition time has - /// passed are removed (along with their indexes and tags). + /// moved to ACTIVE. A DELETING table whose deadline has passed is taken + /// with a compare-and-set on that deadline and removed by the same steps + /// `DeleteTable` runs (see `remove_claimed_table`); a row the swap does + /// not take is left to the actor that holds it. /// /// Called by the background poller in `cmd_serve`. Also called at startup /// to recover in-flight operations from a previous server instance. @@ -37,7 +39,9 @@ impl CassandraEngine { /// /// # Errors /// - /// Returns [`StorageError`] if the database is unreachable or a query fails. + /// Returns [`StorageError`] if the catalog cannot be read or a row cannot + /// be parsed; a failure to remove one `DELETING` table is logged and the + /// table is left for a later pass. pub async fn process_control_plane_transitions( &self, ) -> Result, StorageError> { @@ -88,9 +92,14 @@ impl CassandraEngine { transitions.push((table_name, "CREATING → active")); } - // DELETING → remove row (with tags and data table cleanup) + // DELETING → removed. Only a DeleteTable claim and the renewals of it + // give a DELETING row a deadline, so a row here is a claim whose call + // stopped, a call still running past its lease, or a removal by an + // earlier pass that stopped part way. let query = format!( - "SELECT account_id, table_name, table_arn, table_id FROM {catalog_keyspace}.tables \ + "SELECT account_id, table_name, table_arn, table_id, \ + toUnixTimestamp(status_transition_at) AS deadline_ms \ + FROM {catalog_keyspace}.tables \ WHERE table_status = 'DELETING' AND status_transition_at <= toTimestamp(now()) \ ALLOW FILTERING" ); @@ -119,57 +128,44 @@ impl CassandraEngine { let table_id: String = row .get_r_by_name("table_id") .map_err(|e| StorageError::Internal(format!("Failed to parse table_id: {e}")))?; - - // Delete tags - let tag_delete = format!("DELETE FROM {catalog_keyspace}.tags WHERE resource_arn = ?"); - self.session - .query_with_values(&tag_delete, cdrs_tokio::query_values!(table_arn.as_str())) - .await - .map_err(|e| StorageError::Internal(format!("Failed to delete tags: {e}")))?; - - // Delete indexes (catalog + data tables) - let account_keyspace = self.account_keyspace(&account_id); - self.clear_ttl_entries_for_table_id(&account_id, &table_id) - .await?; - crate::data::index::delete_indexes_for_table( - &self.session_arc(), - &catalog_keyspace, - &account_keyspace, - &table_id, - self, - ) - .await?; - - let continuous_backup_delete = format!( - "DELETE FROM {catalog_keyspace}.continuous_backups WHERE account_id = ? AND table_name = ?" - ); - self.session - .query_with_values( - &continuous_backup_delete, - cdrs_tokio::query_values!(account_id.as_str(), table_name.as_str()), + let deadline_ms: i64 = row + .get_r_by_name("deadline_ms") + .map_err(|e| StorageError::Internal(format!("Failed to parse deadline: {e}")))?; + + // Take the row from the deadline this pass read. The swap applies + // for one actor; a row that another sweeper took, that its call + // renewed, or that was removed or created again under the name + // since the select is left alone. + let removal = async { + let Some(lease) = self + .renew_delete_lease(&account_id, &table_name, &table_id, deadline_ms) + .await? + else { + return Ok(false); + }; + // The call's removal, under this pass's lease. It returns false, + // having deleted nothing keyed on the name, when the lease is + // taken from this pass before the name-keyed deletes. + self.remove_claimed_table( + &account_id, + &table_name, + &table_id, + &table_arn, + Some(lease), ) .await - .map_err(|e| { - StorageError::Internal(format!("Failed to delete continuous backup state: {e}")) - })?; - - // Delete table row — LWT to avoid mixing plain/LWT writes on a row - // created with IF NOT EXISTS. See ADR-0021. - let table_delete = format!( - "DELETE FROM {catalog_keyspace}.tables WHERE account_id = ? AND table_name = ? IF EXISTS" - ); - crate::cassandra_util::query_lwt( - &self.session, - &table_delete, - cdrs_tokio::query_values!(account_id.as_str(), table_name.as_str()), - ) - .await - .map_err(|e| StorageError::Internal(format!("Failed to delete table: {e}")))?; - - // Drop base table in account keyspace - self.drop_data_table(&account_keyspace, &table_id).await?; - - transitions.push((table_name, "DELETING → deleted")); + }; + match removal.await { + Ok(true) => transitions.push((table_name, "DELETING → deleted")), + Ok(false) => {} + // Logged and left, as the abandoned-restore sweep does, so one + // row that cannot be removed does not stop the rows after it. A + // later pass takes it again once this pass's lease has passed. + Err(error) => tracing::warn!( + "Control-plane sweep: removing table {table_name} (account {account_id}): \ + {error}; retrying on a later pass" + ), + } } Ok(transitions) diff --git a/crates/storage-cassandra/tests/direct/concurrent_delete_table.rs b/crates/storage-cassandra/tests/direct/concurrent_delete_table.rs new file mode 100644 index 00000000..e43e7b91 --- /dev/null +++ b/crates/storage-cassandra/tests/direct/concurrent_delete_table.rs @@ -0,0 +1,534 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Direct tests for two `DeleteTable` calls on one table. +//! +//! Amazon DynamoDB answers the second of two concurrent `DeleteTable` calls on +//! an `ACTIVE` table with `ResourceInUseException` and the sentence `Attempt to +//! change a resource which is still in use: Table is being deleted: ` +//! (measured 2026-10-25, five rounds of five). On this backend `DeleteTable` +//! claims the row as `DELETING` with a lightweight transaction and then drops +//! the data and the catalog row inside the same call; these tests run against +//! the window that call opens. +//! +//! `create_table` in this harness leaves a fresh table `CREATING` with a 250 ms +//! activation deadline, so the tests activate their tables first (`activate`) +//! and the claim they reach is the one on an `ACTIVE` row; one round of the +//! first test deletes a `CREATING` table, which the claim also admits. +//! +//! The claim sets the row's `status_transition_at` five minutes ahead, which +//! the control-plane sweep (`process_control_plane_transitions`, run by other +//! tests in this binary against the shared catalog keyspace) waits for before +//! it removes a `DELETING` row. The stopped-claim test moves that deadline into +//! the past by hand and runs the sweep itself. Each test runs inside +//! `removing_rows_after`, so a test that fails leaves no row behind for a later +//! sweep to reach after its account keyspace is gone. + +use std::sync::Arc; +use std::time::Duration; + +use extenddb_core::types::{TableStatus, Tag}; +use extenddb_storage::error::StorageError; +use extenddb_storage::{BackupEngine, MetadataEngine, TableEngine}; + +use crate::delete_table_support::{ + activate, catalog_rows, create_input, create_one, data_table_exists, delete_input, + move_deadline_into_the_past, pitr_row, removing_rows_after, row_deadline_ms, row_status, + row_table_id, wait_for_row_gone, +}; +use crate::helpers::{TestAccount, setup_engine, test_config}; + +/// The round of `two_concurrent_delete_tables_have_one_winner` that deletes a +/// table still `CREATING`. The claim admits a `CREATING` row on purpose: this +/// backend has always deleted one, and refusing it is a separate change. +const CREATING_ROUND: usize = 9; + +/// Two concurrent `DeleteTable` calls on one `ACTIVE` table: the claim applies +/// for one of them, the other answers `TableBeingDeleted`, which is the +/// service's answer for the second of two concurrent calls (measured +/// 2026-10-25, five rounds of five). Ten rounds, a fresh table each; +/// afterwards neither the catalog row nor the data table remains. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn two_concurrent_delete_tables_have_one_winner() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let account = TestAccount::new(&engine, &test_config().keyspace_prefix).await; + let id = account.account_id.clone(); + let test = { + let (engine, account_id) = (engine.clone(), id.clone()); + async move { + let shared = Arc::new(engine.clone()); + for round in 0..10 { + let name = format!("ConcurrentDelete{round}"); + let created = create_one(&engine, &account_id, &name).await; + if round != CREATING_ROUND { + activate(&engine, &account_id, &name).await; + } + + let tasks: Vec<_> = (0..2) + .map(|_| { + let engine = Arc::clone(&shared); + let account_id = account_id.clone(); + let name = name.clone(); + tokio::spawn(async move { + engine.delete_table(&account_id, delete_input(&name)).await + }) + }) + .collect(); + + let mut ok = 0; + let mut being_deleted = 0; + for task in tasks { + match task.await.expect("task") { + Ok(description) => { + assert_eq!( + description.table_status, + TableStatus::Deleting, + "round {round}: the winner reports DELETING" + ); + ok += 1; + } + Err(StorageError::TableBeingDeleted(reported)) if reported == name => { + being_deleted += 1; + } + Err(other) => panic!("round {round}: unexpected {other:?}"), + } + } + assert_eq!((ok, being_deleted), (1, 1), "round {round}"); + assert_eq!( + row_status(&engine, &account_id, &name).await, + None, + "round {round}: the catalog row is gone" + ); + assert!( + !data_table_exists(&engine, &account_id, &created.table_id).await, + "round {round}: the data table is gone" + ); + } + } + }; + removing_rows_after(&engine, &id, test).await; +} + +/// The second half of #553. The deleted table's continuous-backup row is keyed +/// on the table name and its tags on the table ARN, and a table created into +/// the same name holds both under the same keys. Here a `CreateTable` takes the +/// name back as soon as the winner releases it and tags the new table at once, +/// while a second `DeleteTable` is still in flight; the new table keeps its +/// tags and its PITR row, because the second call is refused at the claim and +/// runs none of the deletes. +/// +/// The PITR flag is read from the stored row as well as through +/// `describe_continuous_backups`, whose `continuous_backups_status` is the +/// constant `ENABLED`. The new table enables PITR on even rounds, which a +/// missing row fails, and leaves it off on odd rounds, which a row inherited +/// from the first table fails; its tags must not include the first table's +/// `first-only` key. +/// +/// The second call is started a little behind the first so that it meets a +/// delete already under way; under the create loop's load either of the two +/// can reach the claim first, so the test counts the outcomes rather than +/// naming a winner. The call that loses answers `TableBeingDeleted` when it +/// meets the claim and `TableNotFound` when the other one had already +/// finished, and both mean it ran no deletes of its own. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_table_created_during_a_delete_keeps_its_tags_and_pitr_row() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let account = TestAccount::new(&engine, &test_config().keyspace_prefix).await; + let id = account.account_id.clone(); + let test = { + let (engine, account_id) = (engine.clone(), id.clone()); + async move { + let shared = Arc::new(engine.clone()); + for round in 0..5 { + recreate_during_a_delete(&shared, &account_id, round).await; + } + } + }; + removing_rows_after(&engine, &id, test).await; +} + +/// One round of `a_table_created_during_a_delete_keeps_its_tags_and_pitr_row`. +async fn recreate_during_a_delete( + shared: &Arc, + account_id: &str, + round: usize, +) { + let engine = shared.as_ref(); + let name = format!("RecreatedDuringDelete{round}"); + let first = create_one(engine, account_id, &name).await; + activate(engine, account_id, &name).await; + engine + .tag_resource( + &first.table_arn, + &[ + Tag { + key: "owner".to_owned(), + value: "first".to_owned(), + }, + Tag { + key: "first-only".to_owned(), + value: "yes".to_owned(), + }, + ], + ) + .await + .expect("tag the first table"); + engine + .update_continuous_backups(account_id, &name, true) + .await + .expect("enable PITR on the first table"); + + let deletes: Vec<_> = (0..2) + .map(|index| { + let engine = Arc::clone(shared); + let account_id = account_id.to_owned(); + let name = name.clone(); + tokio::spawn(async move { + // The second call starts a little behind so it meets a + // delete already under way rather than an untouched table. + if index == 1 { + tokio::time::sleep(Duration::from_millis(10)).await; + } + engine.delete_table(&account_id, delete_input(&name)).await + }) + }) + .collect(); + // Takes the name back the moment the winner releases it, and writes the + // new table's tags and PITR row while the second call may still be + // running. A delete that is not fenced takes both away from here. + let creator = { + let engine = Arc::clone(shared); + let account_id = account_id.to_owned(); + let name = name.clone(); + tokio::spawn(async move { + for _ in 0..3_000 { + match engine.create_table(&account_id, create_input(&name)).await { + Ok(description) => { + engine + .tag_resource( + &description.table_arn, + &[Tag { + key: "owner".to_owned(), + value: "second".to_owned(), + }], + ) + .await + .expect("tag the new table"); + if round.is_multiple_of(2) { + engine + .update_continuous_backups(&account_id, &name, true) + .await + .expect("enable PITR on the new table"); + } + return Some(description); + } + Err(StorageError::TableAlreadyExists(_)) => { + tokio::time::sleep(Duration::from_millis(1)).await; + } + Err(other) => panic!("recreating {name}: {other:?}"), + } + } + None + }) + }; + + let mut ok = 0; + let mut refused = 0; + for task in deletes { + match task.await.expect("task") { + Ok(_) => ok += 1, + Err(StorageError::TableBeingDeleted(reported)) if reported == name => refused += 1, + Err(StorageError::TableNotFound(_)) => refused += 1, + Err(other) => panic!("round {round}: unexpected {other:?}"), + } + } + assert_eq!( + (ok, refused), + (1, 1), + "round {round}: one call deletes the table and the other runs no deletes" + ); + let recreated = creator + .await + .expect("task") + .expect("the name became free again"); + assert_ne!( + recreated.table_id, first.table_id, + "round {round}: a new incarnation" + ); + + let tags = engine + .list_tags(&recreated.table_arn) + .await + .expect("read the new table's tags"); + assert_eq!( + tags.iter() + .map(|tag| (tag.key.as_str(), tag.value.as_str())) + .collect::>(), + vec![("owner", "second")], + "round {round}: the new table keeps its tags" + ); + let backups = engine + .describe_continuous_backups(account_id, &name) + .await + .expect("read the new table's continuous backups"); + let pitr_status = backups + .point_in_time_recovery_description + .map(|pitr| pitr.point_in_time_recovery_status); + let stored = pitr_row(engine, account_id, &name).await; + if round.is_multiple_of(2) { + assert_eq!( + (pitr_status.as_deref(), stored), + (Some("ENABLED"), Some(true)), + "round {round}: the new table keeps its PITR row" + ); + } else { + assert_eq!( + (pitr_status.as_deref(), stored), + (Some("DISABLED"), None), + "round {round}: the new table inherits no PITR row" + ); + } + assert_eq!( + row_table_id(engine, account_id, &name).await.as_deref(), + Some(recreated.table_id.as_str()), + "round {round}: the catalog row is the new table's" + ); + + engine + .delete_table(account_id, delete_input(&name)) + .await + .expect("remove the new table"); +} + +/// The claim is conditioned on the table id as well as on the status, so a +/// call holding the id of a table that has since been deleted and created +/// again under the name does not claim the newer table. +#[tokio::test] +async fn a_claim_carrying_another_table_id_does_not_apply() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let account = TestAccount::new(&engine, &test_config().keyspace_prefix).await; + let id = account.account_id.clone(); + let test = { + let (engine, account_id) = (engine.clone(), id.clone()); + async move { + let name = "ClaimedWithAnotherId"; + create_one(&engine, &account_id, name).await; + activate(&engine, &account_id, name).await; + let other_table_id = uuid::Uuid::new_v4().to_string(); + assert!( + !engine + .claim_table_for_delete(&account_id, name, &other_table_id) + .await + .expect("claim"), + "a claim carrying another table id does not apply" + ); + assert_eq!( + row_status(&engine, &account_id, name).await.as_deref(), + Some("ACTIVE") + ); + engine + .delete_table(&account_id, delete_input(name)) + .await + .expect("remove the table"); + } + }; + removing_rows_after(&engine, &id, test).await; +} + +/// The lease is renewed with a compare-and-set on the deadline the holder +/// last wrote. A renewal that carries an older deadline belongs to an actor +/// whose row was taken over since, and does not apply. +#[tokio::test] +async fn a_renewal_against_a_stale_deadline_does_not_apply() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let account = TestAccount::new(&engine, &test_config().keyspace_prefix).await; + let id = account.account_id.clone(); + let test = { + let (engine, account_id) = (engine.clone(), id.clone()); + async move { + let name = "RenewedWithStaleDeadline"; + let created = create_one(&engine, &account_id, name).await; + activate(&engine, &account_id, name).await; + assert!( + engine + .claim_table_for_delete(&account_id, name, &created.table_id) + .await + .expect("claim"), + "the claim applies to an ACTIVE row" + ); + let first = row_deadline_ms(&engine, &account_id, name) + .await + .expect("the claim sets a deadline"); + let second = engine + .renew_delete_lease(&account_id, name, &created.table_id, first) + .await + .expect("renew") + .expect("a renewal carrying the current deadline applies"); + assert!(second >= first, "a renewal moves the deadline forward"); + assert_eq!( + engine + .renew_delete_lease(&account_id, name, &created.table_id, first) + .await + .expect("renew again"), + None, + "a renewal carrying the deadline that was already replaced does not apply" + ); + assert_eq!( + row_deadline_ms(&engine, &account_id, name).await, + Some(second), + "the row keeps the deadline of the renewal that applied" + ); + move_deadline_into_the_past(&engine, &account_id, name).await; + engine + .process_control_plane_transitions() + .await + .expect("sweep the claimed row"); + wait_for_row_gone(&engine, &account_id, name).await; + } + }; + removing_rows_after(&engine, &id, test).await; +} + +// ── A claim whose call stopped ────────────────────────────────────────────── + +/// A `DeleteTable` that stops after its claim, because its process dies or a +/// statement after the claim fails, leaves the row `DELETING` with the claim's +/// deadline ahead of it. Until that deadline the control-plane sweep leaves the +/// row alone, so it does not run the removal beside a call that is still +/// dropping; once it has passed, the sweep removes the row, the data table, the +/// tags and the continuous-backup row, and the name can be created again. The +/// test claims the row and stops there, moves the deadline into the past rather +/// than waiting five minutes, and runs the sweep itself. +#[tokio::test] +async fn a_claim_left_by_a_stopped_call_is_swept_after_its_deadline() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let account = TestAccount::new(&engine, &test_config().keyspace_prefix).await; + let id = account.account_id.clone(); + let test = { + let (engine, account_id) = (engine.clone(), id.clone()); + async move { claim_then_sweep(&engine, &account_id).await } + }; + removing_rows_after(&engine, &id, test).await; +} + +/// The body of `a_claim_left_by_a_stopped_call_is_swept_after_its_deadline`. +async fn claim_then_sweep(engine: &extenddb_storage_cassandra::CassandraEngine, account_id: &str) { + let name = "ClaimedThenStopped"; + let created = create_one(engine, account_id, name).await; + activate(engine, account_id, name).await; + engine + .tag_resource( + &created.table_arn, + &[Tag { + key: "owner".to_owned(), + value: "first".to_owned(), + }], + ) + .await + .expect("tag the table"); + engine + .update_continuous_backups(account_id, name, true) + .await + .expect("enable PITR"); + + let claimed_at_ms = chrono::Utc::now().timestamp_millis(); + assert!( + engine + .claim_table_for_delete(account_id, name, &created.table_id) + .await + .expect("claim"), + "the claim applies to an ACTIVE row" + ); + assert!( + !engine + .claim_table_for_delete(account_id, name, &created.table_id) + .await + .expect("second claim"), + "a second claim does not apply" + ); + assert_eq!( + row_status(engine, account_id, name).await.as_deref(), + Some("DELETING") + ); + let lease_ms = row_deadline_ms(engine, account_id, name) + .await + .expect("the claim sets a deadline") + - claimed_at_ms; + assert!( + (299_000..=310_000).contains(&lease_ms), + "the deadline is five minutes after the claim, got {lease_ms} ms" + ); + + engine + .process_control_plane_transitions() + .await + .expect("sweep before the deadline"); + assert_eq!( + row_status(engine, account_id, name).await.as_deref(), + Some("DELETING"), + "before the deadline the sweep leaves the row to the call" + ); + assert!( + data_table_exists(engine, account_id, &created.table_id).await, + "the data table is still there" + ); + + move_deadline_into_the_past(engine, account_id, name).await; + engine + .process_control_plane_transitions() + .await + .expect("sweep after the deadline"); + // Another test's sweep may have taken the row first; this sweep then + // leaves it to that one, so wait for the removal rather than assume it. + wait_for_row_gone(engine, account_id, name).await; + assert!( + !data_table_exists(engine, account_id, &created.table_id).await, + "the data table is gone" + ); + let catalog = engine.catalog_keyspace(); + assert_eq!( + catalog_rows( + engine, + &format!("SELECT tag_key FROM {catalog}.tags WHERE resource_arn = ?"), + cdrs_tokio::query_values!(created.table_arn.as_str()), + ) + .await, + 0, + "the tags are gone" + ); + assert_eq!( + catalog_rows( + engine, + &format!( + "SELECT table_name FROM {catalog}.continuous_backups \ + WHERE account_id = ? AND table_name = ?" + ), + cdrs_tokio::query_values!(account_id, name), + ) + .await, + 0, + "the continuous-backup row is gone" + ); + + let again = create_one(engine, account_id, name).await; + assert_ne!(again.table_id, created.table_id, "a new incarnation"); + engine + .delete_table(account_id, delete_input(name)) + .await + .expect("remove the new table"); +} diff --git a/crates/storage-cassandra/tests/direct/delete_table_lease.rs b/crates/storage-cassandra/tests/direct/delete_table_lease.rs new file mode 100644 index 00000000..12334799 --- /dev/null +++ b/crates/storage-cassandra/tests/direct/delete_table_lease.rs @@ -0,0 +1,525 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Direct tests for who may finish a `DeleteTable` on this backend. +//! +//! A `DeleteTable` claims the catalog row as `DELETING` with a deadline five +//! minutes ahead. The control-plane sweep takes a row whose deadline has +//! passed with a compare-and-set on that deadline, and the actor that holds +//! the row renews it with the same swap before the two deletes keyed on the +//! name (the `continuous_backups` row and the tags), which run last, then +//! deletes the row on its table id. These tests put a second actor beside the +//! one that holds the row, and a table created into the name in between, and +//! check that the newer table keeps its catalog row, its PITR row and its tags. +//! +//! Rows with a deadline in the past are visible to every sweep in this binary, +//! so the tests wait on the state they expect rather than on the sweep they +//! started, and each runs inside `removing_rows_after`, which removes the +//! account's catalog rows whether the test passes or fails. + +use std::time::Duration; + +use extenddb_core::types::{AttributeValue, Tag}; +use extenddb_storage::{ + BackupEngine, DataEngine, MetadataEngine, RestoreRequest, RestoreTableOverrides, TableEngine, +}; +use extenddb_storage_cassandra::CassandraEngine; + +use crate::delete_table_support::{ + activate, catalog_rows, create_input, create_one, data_table_exists, delete_input, gsi_input, + move_deadline_into_the_past, pitr_row, removing_rows_after, row_deadline_ms, row_status, + row_table_id, tags_of, wait_for_data_table_gone, wait_for_row_gone, wait_for_status, +}; +use crate::helpers::{TestAccount, setup_engine, test_config}; + +/// A claim whose call stopped, its deadline passed, and two servers' sweeps +/// started 300 ms apart. The first sweep takes the row by moving its deadline, +/// so the second one's select no longer returns it, and the first holds the +/// row until it has deleted it. A table created into the name once the row is +/// gone keeps its catalog row, its PITR row and its data table. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn two_sweeps_over_a_stopped_claim_leave_a_newer_table_alone() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let other_server = setup_engine().await; + let account = TestAccount::new(&engine, &test_config().keyspace_prefix).await; + let id = account.account_id.clone(); + let test = { + let (engine, id) = (engine.clone(), id.clone()); + async move { + let name = "SweptByTwoServers"; + let first = engine + .create_table(&id, gsi_input(name, 6)) + .await + .expect("create the first table"); + activate(&engine, &id, name).await; + engine + .update_continuous_backups(&id, name, true) + .await + .expect("enable PITR"); + assert!( + engine + .claim_table_for_delete(&id, name, &first.table_id) + .await + .expect("claim"), + "the claim applies" + ); + move_deadline_into_the_past(&engine, &id, name).await; + + let sweep_a = { + let engine = engine.clone(); + tokio::spawn(async move { engine.process_control_plane_transitions().await }) + }; + tokio::time::sleep(Duration::from_millis(300)).await; + let sweep_b = + tokio::spawn(async move { other_server.process_control_plane_transitions().await }); + + wait_for_row_gone(&engine, &id, name).await; + let second = create_one(&engine, &id, name).await; + engine + .update_continuous_backups(&id, name, true) + .await + .expect("enable PITR on the new table"); + sweep_a.await.expect("task").expect("first sweep"); + sweep_b.await.expect("task").expect("second sweep"); + + assert_eq!( + row_table_id(&engine, &id, name).await.as_deref(), + Some(second.table_id.as_str()), + "the new table keeps its catalog row" + ); + assert_eq!( + pitr_row(&engine, &id, name).await, + Some(true), + "the new table keeps its PITR row" + ); + assert!(data_table_exists(&engine, &id, &second.table_id).await); + assert!(!data_table_exists(&engine, &id, &first.table_id).await); + engine + .delete_table(&id, delete_input(name)) + .await + .expect("remove the new table"); + } + }; + removing_rows_after(&engine, &id, test).await; +} + +/// A sweep holds a row from its select, and while it drops the index tables +/// the row is removed and the name created again (here the row delete is run +/// by hand, the last statement of whichever actor finished first). The sweep's +/// renewal then finds the row holding another table id, so it stops before the +/// deletes keyed on the name and the row delete, and does not report the +/// table as removed. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_sweep_whose_row_was_replaced_leaves_the_newer_table_alone() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let account = TestAccount::new(&engine, &test_config().keyspace_prefix).await; + let id = account.account_id.clone(); + let test = { + let (engine, id) = (engine.clone(), id.clone()); + async move { + let name = "ReplacedUnderTheSweep"; + let first = engine + .create_table(&id, gsi_input(name, 6)) + .await + .expect("create the first table"); + activate(&engine, &id, name).await; + engine + .update_continuous_backups(&id, name, true) + .await + .expect("enable PITR"); + assert!( + engine + .claim_table_for_delete(&id, name, &first.table_id) + .await + .expect("claim") + ); + move_deadline_into_the_past(&engine, &id, name).await; + + let sweep = { + let engine = engine.clone(); + tokio::spawn(async move { engine.process_control_plane_transitions().await }) + }; + // The sweep is past its select once its take has moved the + // deadline, or once the table's index metadata is gone. + let catalog = engine.catalog_keyspace(); + let started = std::time::Instant::now(); + loop { + let taken = row_deadline_ahead(&engine, &id, name).await; + let metadata_gone = catalog_rows( + &engine, + &format!("SELECT index_id FROM {catalog}.indexes WHERE table_id = ?"), + cdrs_tokio::query_values!(first.table_id.as_str()), + ) + .await + == 0; + if taken || metadata_gone { + break; + } + assert!( + started.elapsed() < Duration::from_secs(60), + "the sweep never started" + ); + tokio::time::sleep(Duration::from_millis(5)).await; + } + let row_delete = format!( + "DELETE FROM {catalog}.tables WHERE account_id = ? AND table_name = ? \ + IF table_id = ?" + ); + extenddb_storage_cassandra::cassandra_util::query_lwt( + engine.session(), + &row_delete, + cdrs_tokio::query_values!(id.as_str(), name, first.table_id.as_str()), + ) + .await + .expect("remove the row"); + let second = create_one(&engine, &id, name).await; + engine + .update_continuous_backups(&id, name, true) + .await + .expect("enable PITR on the new table"); + + let swept = sweep.await.expect("task").expect("the sweep"); + assert_eq!( + row_table_id(&engine, &id, name).await.as_deref(), + Some(second.table_id.as_str()), + "the new table keeps its catalog row" + ); + assert_eq!( + pitr_row(&engine, &id, name).await, + Some(true), + "the new table keeps its PITR row" + ); + assert!(data_table_exists(&engine, &id, &second.table_id).await); + assert!( + !swept.iter().any(|(table, _)| table == name), + "the sweep stopped at its renewal: {swept:?}" + ); + engine + .delete_table(&id, delete_input(name)) + .await + .expect("remove the new table"); + } + }; + removing_rows_after(&engine, &id, test).await; +} + +/// True when the row's deadline is more than four minutes away, which only a +/// fresh claim or a take by a sweep writes. +async fn row_deadline_ahead(engine: &CassandraEngine, id: &str, name: &str) -> bool { + row_deadline_ms(engine, id, name) + .await + .is_some_and(|deadline| deadline > chrono::Utc::now().timestamp_millis() + 240_000) +} + +/// A restored table keeps its restore registration columns after activation, +/// and its heartbeat stops there. A `DeleteTable` claim on it makes a +/// `DELETING` row with a registration and a stale heartbeat, the row the +/// abandoned-restore sweep removes. The sweep skips it because the claim +/// wrote a deadline, so the claim stays with its call. The heartbeat is moved +/// two minutes back, the state the row reaches two minutes after the restore, +/// and the sweep runs with the production threshold of 60 s, so the test +/// reaches no other test's restore. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn the_restore_sweep_leaves_a_claimed_restored_table_to_its_delete() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let account = TestAccount::new(&engine, &test_config().keyspace_prefix).await; + let id = account.account_id.clone(); + let test = { + let (engine, id) = (engine.clone(), id.clone()); + async move { + create_one(&engine, &id, "RestoreSource").await; + activate(&engine, &id, "RestoreSource").await; + let key_info = engine + .table_key_info(&id, "RestoreSource") + .await + .expect("key info"); + let item = [("pk".to_owned(), AttributeValue::S("k0".to_owned()))].into(); + engine + .put_item(&key_info, item, false, None, &Default::default(), None) + .await + .expect("put an item"); + let backup = engine + .create_backup(&id, "RestoreSource", "lease") + .await + .expect("create a backup"); + let source = engine + .backup_restore_source(&id, &backup.backup_arn) + .await + .expect("restore source"); + let input = source.restore_input("Restored", &RestoreTableOverrides::default()); + let restored = engine + .restore_table_from_backup( + &id, + RestoreRequest { + backup_arn: backup.backup_arn.clone(), + source_table_arn: source.source_table_arn.clone(), + restore_date_time: source.backup_creation_date_time, + input, + }, + ) + .await + .expect("restore"); + wait_for_status(&engine, &id, "Restored", "ACTIVE").await; + + let stale = extenddb_storage_cassandra::cassandra_util::now_millis() - 120_000; + let age_heartbeat = format!( + "UPDATE {}.tables SET restore_heartbeat = ? WHERE account_id = ? \ + AND table_name = ? IF table_id = ? AND table_status = 'ACTIVE'", + engine.catalog_keyspace() + ); + let result = extenddb_storage_cassandra::cassandra_util::query_lwt( + engine.session(), + &age_heartbeat, + cdrs_tokio::query_values!( + stale, + id.as_str(), + "Restored", + restored.table_id.as_str() + ), + ) + .await + .expect("age the heartbeat"); + assert!( + extenddb_storage_cassandra::cassandra_util::lwt_applied(&result) + .expect("read the answer"), + "the restored table is ACTIVE" + ); + assert!( + engine + .claim_table_for_delete(&id, "Restored", &restored.table_id) + .await + .expect("claim"), + "the claim applies" + ); + + let removed = engine + .sweep_abandoned_restores(Duration::from_secs(60)) + .await + .expect("restore sweep"); + assert!( + !removed.iter().any(|table| table == "Restored"), + "the restore sweep left the claimed table alone: {removed:?}" + ); + assert_eq!( + row_status(&engine, &id, "Restored").await.as_deref(), + Some("DELETING") + ); + assert!(data_table_exists(&engine, &id, &restored.table_id).await); + } + }; + removing_rows_after(&engine, &id, test).await; +} + +/// A `DeleteTable` of a table with eight indexes, and, once its base data +/// table is gone and while it drops the index tables, an +/// `UpdateContinuousBackups` (the storage call the engine makes; nothing +/// refuses it during `DELETING`, as the service serves it) and a tag written +/// at the storage layer, below the engine's validation. The deletes keyed on +/// the name run after the drops, so both rows go with the table, and a table +/// created into the name afterwards has no PITR row and no tags. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn pitr_and_tags_written_during_the_drops_go_with_the_table() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let account = TestAccount::new(&engine, &test_config().keyspace_prefix).await; + let id = account.account_id.clone(); + let test = { + let (engine, id) = (engine.clone(), id.clone()); + async move { + let name = "WrittenDuringTheDrops"; + let first = engine + .create_table(&id, gsi_input(name, 8)) + .await + .expect("create the table"); + activate(&engine, &id, name).await; + let call = { + let (engine, id) = (engine.clone(), id.clone()); + tokio::spawn(async move { engine.delete_table(&id, delete_input(name)).await }) + }; + wait_for_status(&engine, &id, name, "DELETING").await; + wait_for_data_table_gone(&engine, &id, &first.table_id).await; + engine + .update_continuous_backups(&id, name, true) + .await + .expect("UpdateContinuousBackups is served during DELETING"); + engine + .tag_resource( + &first.table_arn, + &[Tag { + key: "left-by".to_owned(), + value: "the deleted table".to_owned(), + }], + ) + .await + .expect("write a tag row"); + call.await.expect("task").expect("DeleteTable"); + + let second = engine + .create_table(&id, create_input(name)) + .await + .expect("create the name again"); + assert_eq!( + pitr_row(&engine, &id, name).await, + None, + "the new table has no PITR row" + ); + assert_eq!( + tags_of(&engine, &second.table_arn).await, + Vec::<(String, String)>::new(), + "the new table has no tags" + ); + engine + .delete_table(&id, delete_input(name)) + .await + .expect("remove the new table"); + } + }; + removing_rows_after(&engine, &id, test).await; +} + +/// Two claims whose calls stopped, in two accounts, both past their deadline; +/// one account's keyspace is gone, so removing its table fails at the TTL +/// clear. The sweep logs that row and goes on: it returns `Ok`, and the other +/// account's row is removed. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_row_the_sweep_cannot_remove_does_not_stop_the_others() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let broken = TestAccount::new(&engine, &test_config().keyspace_prefix).await; + let healthy = TestAccount::new(&engine, &test_config().keyspace_prefix).await; + let (broken_id, healthy_id) = (broken.account_id.clone(), healthy.account_id.clone()); + let test = { + let (engine, broken_id, healthy_id) = + (engine.clone(), broken_id.clone(), healthy_id.clone()); + let inner_engine = engine.clone(); + let healthy_guard = healthy_id.clone(); + async move { + let inner = async move { + for (id, name) in [(&broken_id, "Unremovable"), (&healthy_id, "Removable")] { + let table = create_one(&engine, id, name).await; + activate(&engine, id, name).await; + assert!( + engine + .claim_table_for_delete(id, name, &table.table_id) + .await + .expect("claim") + ); + } + // Dropped before either deadline passes, so no sweep can remove + // the first table while its keyspace is still there. + let keyspace = engine.account_keyspace(&broken_id); + engine + .session() + .query(format!("DROP KEYSPACE IF EXISTS {keyspace}")) + .await + .expect("drop the account keyspace"); + move_deadline_into_the_past(&engine, &broken_id, "Unremovable").await; + move_deadline_into_the_past(&engine, &healthy_id, "Removable").await; + + engine + .process_control_plane_transitions() + .await + .expect("the sweep logs the row it cannot remove and goes on"); + wait_for_row_gone(&engine, &healthy_id, "Removable").await; + assert_eq!( + row_status(&engine, &broken_id, "Unremovable") + .await + .as_deref(), + Some("DELETING"), + "the row whose removal failed is left for a later pass" + ); + }; + removing_rows_after(&inner_engine, &healthy_guard, inner).await; + } + }; + removing_rows_after(&engine, &broken.account_id, test).await; +} + +/// The order of a `DeleteTable`: the deletes keyed on the name run after the +/// drops. Eight rounds, each a tagged, PITR-enabled table with one index; +/// while the call runs, the moment its base data table is gone the table's +/// tag rows and its continuous-backup row are still there (the index table is +/// dropped between the two), and once the call returns both are gone. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn the_name_keyed_rows_outlive_the_drops_and_go_with_the_row() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let account = TestAccount::new(&engine, &test_config().keyspace_prefix).await; + let id = account.account_id.clone(); + let test = { + let (engine, id) = (engine.clone(), id.clone()); + async move { + for round in 0..8 { + let name = format!("NameKeyedLast{round}"); + let table = engine + .create_table(&id, gsi_input(&name, 1)) + .await + .expect("create the table"); + activate(&engine, &id, &name).await; + engine + .tag_resource( + &table.table_arn, + &[Tag { + key: "owner".to_owned(), + value: "deleted".to_owned(), + }], + ) + .await + .expect("tag the table"); + engine + .update_continuous_backups(&id, &name, true) + .await + .expect("enable PITR"); + let call = { + let (engine, id, name) = (engine.clone(), id.clone(), name.clone()); + tokio::spawn(async move { engine.delete_table(&id, delete_input(&name)).await }) + }; + wait_for_data_table_gone(&engine, &id, &table.table_id).await; + let during = ( + tag_rows(&engine, &table.table_arn).await, + pitr_row(&engine, &id, &name).await, + ); + call.await.expect("task").expect("DeleteTable"); + let after = ( + tag_rows(&engine, &table.table_arn).await, + pitr_row(&engine, &id, &name).await, + ); + assert_eq!( + during, + (1, Some(true)), + "round {round}: the tag and PITR rows are there once the data table is gone" + ); + assert_eq!(after, (0, None), "round {round}: both go with the table"); + } + } + }; + removing_rows_after(&engine, &id, test).await; +} + +/// How many tag rows the catalog holds for `arn`. +async fn tag_rows(engine: &CassandraEngine, arn: &str) -> usize { + catalog_rows( + engine, + &format!( + "SELECT tag_key FROM {}.tags WHERE resource_arn = ?", + engine.catalog_keyspace() + ), + cdrs_tokio::query_values!(arn), + ) + .await +} diff --git a/crates/storage-cassandra/tests/direct/delete_table_support.rs b/crates/storage-cassandra/tests/direct/delete_table_support.rs new file mode 100644 index 00000000..57e75697 --- /dev/null +++ b/crates/storage-cassandra/tests/direct/delete_table_support.rs @@ -0,0 +1,394 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Helpers shared by the direct `DeleteTable` tests (`concurrent_delete_table` +//! and `delete_table_lease`): a one-key table, and reads of the catalog row, +//! the data table and the name-keyed rows that a delete removes. + +use extenddb_core::types::{ + AttributeDefinition, BillingMode, CreateTableInput, DeleteTableInput, GsiInput, + KeySchemaElement, KeyType, Projection, ProjectionType, ScalarAttributeType, TableDescription, +}; +use extenddb_storage::{MetadataEngine, TableEngine}; +use extenddb_storage_cassandra::CassandraEngine; + +use crate::helpers::test_config; + +/// How long the waits below poll before they fail the test. +const WAIT_LIMIT: std::time::Duration = std::time::Duration::from_secs(60); + +/// Runs `body` as its own task, then removes every catalog row the account +/// holds, whether `body` returned or panicked, and re-raises its panic. +/// +/// The catalog keyspace is shared by every test in this binary. A test that +/// fails between a claim and its own clean-up would otherwise leave a +/// `DELETING` row there; once the test's account keyspace is dropped, the +/// control-plane sweep that next reaches the row fails on the missing +/// keyspace, inside whichever test ran it. Each row goes by its table id, +/// with the condition `DeleteTable` uses, together with its index rows, its +/// tags and the account's continuous-backup rows. +pub(crate) async fn removing_rows_after(engine: &CassandraEngine, account_id: &str, body: F) +where + F: std::future::Future + Send + 'static, +{ + let outcome = tokio::spawn(body).await; + remove_account_rows(engine, account_id).await; + if let Err(error) = outcome { + std::panic::resume_unwind(error.into_panic()); + } +} + +/// Best effort: a failure here must not hide the panic being re-raised. +async fn remove_account_rows(engine: &CassandraEngine, account_id: &str) { + use cdrs_tokio::types::IntoRustByName; + let catalog = engine.catalog_keyspace(); + let select = format!( + "SELECT table_name, table_id, table_arn FROM {catalog}.tables WHERE account_id = ?" + ); + let rows = match engine + .session() + .query_with_values(&select, cdrs_tokio::query_values!(account_id)) + .await + .ok() + .and_then(|frame| frame.response_body().ok()) + .and_then(cdrs_tokio::frame::message_response::ResponseBody::into_rows) + { + Some(rows) => rows, + None => return, + }; + for row in rows { + let name: Result = row.get_r_by_name("table_name"); + let table_id: Result = row.get_r_by_name("table_id"); + let arn: Result = row.get_r_by_name("table_arn"); + let (Ok(name), Ok(table_id), Ok(arn)) = (name, table_id, arn) else { + continue; + }; + let _ = engine + .session() + .query_with_values( + &format!("DELETE FROM {catalog}.tags WHERE resource_arn = ?"), + cdrs_tokio::query_values!(arn.as_str()), + ) + .await; + let _ = engine + .session() + .query_with_values( + &format!("DELETE FROM {catalog}.indexes WHERE table_id = ?"), + cdrs_tokio::query_values!(table_id.as_str()), + ) + .await; + let _ = extenddb_storage_cassandra::cassandra_util::query_lwt( + engine.session(), + &format!( + "DELETE FROM {catalog}.tables WHERE account_id = ? AND table_name = ? \ + IF table_id = ?" + ), + cdrs_tokio::query_values!(account_id, name.as_str(), table_id.as_str()), + ) + .await; + } + let _ = engine + .session() + .query_with_values( + &format!("DELETE FROM {catalog}.continuous_backups WHERE account_id = ?"), + cdrs_tokio::query_values!(account_id), + ) + .await; +} + +/// Activates a `CREATING` row as the activation sweep does, scoped to this +/// row: `ACTIVE`, deadline cleared, `IF table_status = 'CREATING'`. In this +/// harness `create_table` writes `CREATING` with a 250 ms activation deadline +/// and nothing else activates it unless some test happens to run the sweep. +pub(crate) async fn activate(engine: &CassandraEngine, account_id: &str, name: &str) { + let query = format!( + "UPDATE {}.tables SET table_status = 'ACTIVE', status_transition_at = null \ + WHERE account_id = ? AND table_name = ? IF table_status = 'CREATING'", + engine.catalog_keyspace() + ); + extenddb_storage_cassandra::cassandra_util::query_lwt( + engine.session(), + &query, + cdrs_tokio::query_values!(account_id, name), + ) + .await + .expect("activate the table"); + assert_eq!( + row_status(engine, account_id, name).await.as_deref(), + Some("ACTIVE"), + "{name} is ACTIVE" + ); +} + +/// Moves a claimed row's deadline one second into the past, the state a claim +/// reaches five minutes after its call stopped. +pub(crate) async fn move_deadline_into_the_past( + engine: &CassandraEngine, + account_id: &str, + name: &str, +) { + let past_ms = chrono::Utc::now().timestamp_millis() - 1_000; + let query = format!( + "UPDATE {}.tables SET status_transition_at = ? \ + WHERE account_id = ? AND table_name = ? IF EXISTS", + engine.catalog_keyspace() + ); + engine + .session() + .query_with_values(&query, cdrs_tokio::query_values!(past_ms, account_id, name)) + .await + .expect("move the deadline into the past"); +} + +/// Polls until the name has no catalog row. +pub(crate) async fn wait_for_row_gone(engine: &CassandraEngine, account_id: &str, name: &str) { + let started = std::time::Instant::now(); + while row_status(engine, account_id, name).await.is_some() { + assert!( + started.elapsed() < WAIT_LIMIT, + "{name}: the row is still there" + ); + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + } +} + +/// Polls until the row reads `status`. +pub(crate) async fn wait_for_status( + engine: &CassandraEngine, + account_id: &str, + name: &str, + status: &str, +) { + let started = std::time::Instant::now(); + while row_status(engine, account_id, name).await.as_deref() != Some(status) { + assert!( + started.elapsed() < WAIT_LIMIT, + "{name}: never read {status}" + ); + tokio::time::sleep(std::time::Duration::from_millis(2)).await; + } +} + +/// Polls until the table's base data table has been dropped. +pub(crate) async fn wait_for_data_table_gone( + engine: &CassandraEngine, + account_id: &str, + table_id: &str, +) { + let started = std::time::Instant::now(); + while data_table_exists(engine, account_id, table_id).await { + assert!( + started.elapsed() < WAIT_LIMIT, + "data table of {table_id} still there" + ); + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + } +} + +/// The tags stored for `arn`, as `(key, value)` pairs in key order. +pub(crate) async fn tags_of(engine: &CassandraEngine, arn: &str) -> Vec<(String, String)> { + engine + .list_tags(arn) + .await + .expect("read the tags") + .into_iter() + .map(|tag| (tag.key, tag.value)) + .collect() +} + +/// [`create_input`] with `gsis` global secondary indexes, one CQL table each +/// for the delete to drop. +pub(crate) fn gsi_input(name: &str, gsis: usize) -> CreateTableInput { + let mut input = create_input(name); + let mut indexes = Vec::new(); + for i in 0..gsis { + let attribute = format!("g{i}"); + input.attribute_definitions.push(AttributeDefinition { + attribute_name: attribute.clone(), + attribute_type: ScalarAttributeType::S, + }); + indexes.push(GsiInput { + index_name: format!("Idx{i}"), + key_schema: vec![KeySchemaElement { + attribute_name: attribute, + key_type: KeyType::Hash, + }], + projection: Projection { + projection_type: ProjectionType::All, + non_key_attributes: None, + }, + provisioned_throughput: None, + }); + } + input.global_secondary_indexes = Some(indexes); + input +} + +/// A table with one string partition key, created through the engine so the +/// test has the ARN and the table id the catalog holds. +pub(crate) async fn create_one( + engine: &CassandraEngine, + account_id: &str, + name: &str, +) -> TableDescription { + engine + .create_table(account_id, create_input(name)) + .await + .expect("create the table") +} + +pub(crate) fn create_input(name: &str) -> CreateTableInput { + CreateTableInput { + table_name: name.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() + } +} + +pub(crate) fn delete_input(name: &str) -> DeleteTableInput { + DeleteTableInput { + table_name: name.to_owned(), + } +} + +/// True when the table's data table is still in the account keyspace. +pub(crate) async fn data_table_exists( + engine: &CassandraEngine, + account_id: &str, + table_id: &str, +) -> bool { + let keyspace = format!("{}_account_{account_id}", test_config().keyspace_prefix); + let data_table = format!("items_{}", table_id.replace('-', "_")); + let rows = engine + .session() + .query_with_values( + "SELECT table_name FROM system_schema.tables \ + WHERE keyspace_name = ? AND table_name = ?", + cdrs_tokio::query_values!(keyspace.as_str(), data_table.as_str()), + ) + .await + .expect("read system_schema.tables") + .response_body() + .expect("response body") + .into_rows() + .unwrap_or_default(); + !rows.is_empty() +} + +pub(crate) async fn row_status( + engine: &CassandraEngine, + account_id: &str, + name: &str, +) -> Option { + row_column(engine, account_id, name, "table_status").await +} + +/// The table id the catalog row holds, which says which incarnation of the +/// name the row belongs to. +pub(crate) async fn row_table_id( + engine: &CassandraEngine, + account_id: &str, + name: &str, +) -> Option { + row_column(engine, account_id, name, "table_id").await +} + +pub(crate) async fn row_column( + engine: &CassandraEngine, + account_id: &str, + name: &str, + column: &str, +) -> Option { + use cdrs_tokio::types::IntoRustByName; + let query = format!( + "SELECT {column} 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, name)) + .await + .expect("read the row") + .response_body() + .expect("response body") + .into_rows() + .unwrap_or_default(); + rows.first() + .map(|row| row.get_r_by_name(column).expect("column")) +} + +/// The PITR flag the `continuous_backups` row stores for the name, or `None` +/// when there is no row. +pub(crate) async fn pitr_row( + engine: &CassandraEngine, + account_id: &str, + name: &str, +) -> Option { + use cdrs_tokio::types::IntoRustByName; + let query = format!( + "SELECT pitr_enabled FROM {}.continuous_backups WHERE account_id = ? AND table_name = ?", + engine.catalog_keyspace() + ); + let rows = engine + .session() + .query_with_values(&query, cdrs_tokio::query_values!(account_id, name)) + .await + .expect("read continuous_backups") + .response_body() + .expect("response body") + .into_rows() + .unwrap_or_default(); + rows.first() + .map(|row| row.get_r_by_name("pitr_enabled").expect("pitr_enabled")) +} + +/// How many rows a catalog query returns. +pub(crate) async fn catalog_rows( + engine: &CassandraEngine, + query: &str, + values: cdrs_tokio::query::QueryValues, +) -> usize { + engine + .session() + .query_with_values(query, values) + .await + .expect("query the catalog") + .response_body() + .expect("response body") + .into_rows() + .unwrap_or_default() + .len() +} + +/// The row's `status_transition_at` in milliseconds since the epoch. +pub(crate) async fn row_deadline_ms( + engine: &CassandraEngine, + account_id: &str, + name: &str, +) -> Option { + use cdrs_tokio::types::IntoRustByName; + let query = format!( + "SELECT toUnixTimestamp(status_transition_at) AS deadline_ms 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, name)) + .await + .expect("read the row") + .response_body() + .expect("response body") + .into_rows() + .unwrap_or_default(); + rows.first() + .and_then(|row| row.get_by_name("deadline_ms").expect("deadline_ms")) +} diff --git a/crates/storage-cassandra/tests/direct/deleting_window.rs b/crates/storage-cassandra/tests/direct/deleting_window.rs index cc43f7da..9931152c 100644 --- a/crates/storage-cassandra/tests/direct/deleting_window.rs +++ b/crates/storage-cassandra/tests/direct/deleting_window.rs @@ -7,10 +7,12 @@ //! 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. +//! 2026-10-24). On this backend `DeleteTable` claims the row as `DELETING` +//! with a lightweight transaction and then drops the data and the catalog row +//! inside the same call, so the window lasts as long as the drops. The tests +//! here need it held open for longer than that, and set the row's status +//! directly; `concurrent_delete_table.rs` runs against the window the request +//! itself opens. //! //! The catalog keyspace is shared by every test in this binary, and the //! control-plane sweep (`process_control_plane_transitions`) removes any diff --git a/crates/storage-cassandra/tests/direct_integration.rs b/crates/storage-cassandra/tests/direct_integration.rs index 5826be23..fa111e25 100644 --- a/crates/storage-cassandra/tests/direct_integration.rs +++ b/crates/storage-cassandra/tests/direct_integration.rs @@ -25,8 +25,14 @@ mod authorization_store; mod backup_engine; #[path = "direct/cassandra_engine.rs"] mod cassandra_engine; +#[path = "direct/concurrent_delete_table.rs"] +mod concurrent_delete_table; #[path = "direct/delete_item.rs"] mod delete_item; +#[path = "direct/delete_table_lease.rs"] +mod delete_table_lease; +#[path = "direct/delete_table_support.rs"] +mod delete_table_support; #[path = "direct/deleting_window.rs"] mod deleting_window; #[path = "direct/groups.rs"] diff --git a/tests/python/test_deleting_window_control_plane.py b/tests/python/test_deleting_window_control_plane.py index b6837856..c7a180da 100644 --- a/tests/python/test_deleting_window_control_plane.py +++ b/tests/python/test_deleting_window_control_plane.py @@ -187,3 +187,68 @@ def test_data_plane_is_served_until_the_table_is_gone(self, dynamodb_client, is_ with pytest.raises(ClientError) as exc: dynamodb_client.get_item(TableName=name, Key={"pk": {"S": "w1"}}, ConsistentRead=True) assert exc.value.response["Error"]["Code"] == "ResourceNotFoundException" + + +class TestConcurrentDeleteTable: + """Two DeleteTable calls sent at once, which needs no held window. + + The service answers one of them 200 and the other + ``ResourceInUseException`` with the sentence above (measured 2026-10-25, + five rounds of five). Cassandra and MongoDB answer the same. PostgreSQL + and SQLite with ``control_plane_delay_seconds`` under 1 (the CI setting) + remove the catalog row inside the request, so a second call that arrives + after the first has finished finds no row and answers + ``ResourceNotFoundException``; that answer is accepted there only. The + PostgreSQL and SQLite CI legs run ``devtools/run-tests`` without + ``--backend``, so an empty ``EXTENDDB_TEST_BACKEND`` counts as one of them. + Two 200s mean both calls ran the deletion, which is what #553 reports on + Cassandra and #550 fixed on SQLite and MongoDB. + """ + + ROUNDS = 3 + ROW_REMOVED_IN_REQUEST = ("", "postgres", "sqlite") + + def test_only_one_of_two_concurrent_delete_tables_deletes( + self, dynamodb_client, endpoint_url, is_extenddb + ): + from concurrent.futures import ThreadPoolExecutor + + from helpers import make_dynamodb_client + + loser_codes = ("ResourceInUseException",) + if is_extenddb and BACKEND in self.ROW_REMOVED_IN_REQUEST: + loser_codes += ("ResourceNotFoundException",) + + for round_index in range(self.ROUNDS): + name = unique_name("del-pair") + 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) + + def _delete(_worker: int, table: str = name): + client = make_dynamodb_client(endpoint_url) + try: + return ("ok", client.delete_table(TableName=table)) + except ClientError as exc: + return ("err", exc) + + with ThreadPoolExecutor(max_workers=2) as pool: + outcomes = list(pool.map(_delete, range(2))) + + accepted = [o for kind, o in outcomes if kind == "ok"] + refused = [o for kind, o in outcomes if kind == "err"] + assert len(accepted) == 1, ( + f"round {round_index}: {len(accepted)} of the two DeleteTable calls were " + f"accepted; one call deletes the table and the other is refused" + ) + assert accepted[0]["TableDescription"]["TableStatus"] == "DELETING" + err = refused[0].response["Error"] + assert err["Code"] in loser_codes, (round_index, BACKEND, err) + if err["Code"] == "ResourceInUseException": + assert err["Message"] == _being_deleted(name) + + wait_for_deleted(dynamodb_client, name, timeout=120.0)