From 32bd4510d5f59705f97527e638f9cb8b7143bf9c Mon Sep 17 00:00:00 2001 From: Lee Hannigan Date: Mon, 26 Oct 2026 10:08:27 +0000 Subject: [PATCH] fix(cassandra): fence concurrent DeleteTable with a lease (#553) Two concurrent DeleteTable calls on one table each read the catalog row, dropped the data tables and answered 200. Amazon DynamoDB answers one of them 200 and the other ResourceInUseException with "Attempt to change a resource which is still in use: Table is being deleted: ". DeleteTable now claims the row with a lightweight transaction before the first drop, conditioned on the table id and on a status other than DELETING, and sets status_transition_at five minutes ahead. The call whose claim applies removes the table and reports DELETING; the other answers the being-deleted sentence and deletes nothing. The control-plane sweep takes a DELETING row whose deadline has passed with a compare-and-set on that deadline, and skips the row when the swap does not apply. The call and the sweep run the same removal: the drops keyed on the table id first, then a renewal of the lease with the same swap, then the continuous-backup and tag deletes, which are keyed on the name and the ARN, then the row delete conditioned on the table id. An actor whose renewal does not apply stops before the name-keyed deletes, so a call that outlives its lease leaves them to the sweep that took the row over. The abandoned-restore sweep skips any row that carries a deadline, since a restore target never carries one and a claimed row always does. One case still depends on the lease length. An actor that has renewed its lease and then stalls for more than 300 s before the continuous-backup delete, the tag delete and the row delete can find that a sweep took the row over, finished, and freed the name; its two deletes then reach a table created under that name in the meantime. Those two keys carry no table id, so closing the case needs a schema change. A DeleteTable that stops after its claim leaves the row DELETING. A retry is refused with the being-deleted sentence until the lease has passed and the sweep's next pass has removed the table, 300 s plus up to 60 s for the pass; before this change a retry finished the removal at once. With the server's key-info cache warm, data-plane requests during the delete answer as before: 200 until the base data table is dropped, then 500. With it cold they answer 404 for the whole delete, where they answered 200 and then 500 before; the 404 comes from the read that maps DELETING to not found on this backend. The service answers 200 throughout. The 200 of DeleteTable reports DELETING for a table that is already gone when the answer arrives, so a DescribeTable sent after it answers 404. The sweep now logs a row it cannot remove and goes on to the next row; it used to return at the first failure. Direct storage tests cover two concurrent calls (ten rounds, one of them on a CREATING table), a table created into the name during a delete, a claim whose call stopped, a claim carrying another table id, a renewal against a stale deadline, two sweeps over one stopped claim, a sweep whose row is replaced under it, the abandoned-restore sweep over a claimed restored table, PITR and tag writes during the drops, the order of the name-keyed deletes, and a sweep pass with a row it cannot remove. A dual-target wire row sends two concurrent DeleteTable calls and accepts a not-found answer for the loser only on PostgreSQL and SQLite. --- crates/storage-cassandra/src/backup_engine.rs | 15 +- crates/storage-cassandra/src/data/index.rs | 22 +- crates/storage-cassandra/src/delete_table.rs | 262 +++++++-- crates/storage-cassandra/src/worker_store.rs | 104 ++-- .../tests/direct/concurrent_delete_table.rs | 534 ++++++++++++++++++ .../tests/direct/delete_table_lease.rs | 525 +++++++++++++++++ .../tests/direct/delete_table_support.rs | 394 +++++++++++++ .../tests/direct/deleting_window.rs | 10 +- .../tests/direct_integration.rs | 6 + .../test_deleting_window_control_plane.py | 65 +++ 10 files changed, 1819 insertions(+), 118 deletions(-) create mode 100644 crates/storage-cassandra/tests/direct/concurrent_delete_table.rs create mode 100644 crates/storage-cassandra/tests/direct/delete_table_lease.rs create mode 100644 crates/storage-cassandra/tests/direct/delete_table_support.rs diff --git a/crates/storage-cassandra/src/backup_engine.rs b/crates/storage-cassandra/src/backup_engine.rs index 7c070934e..f31bc5ce7 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 4ebd72dac..23ad7f583 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 479bb86c6..26375a153 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 f89422c97..21ded866d 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 000000000..e43e7b914 --- /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 000000000..123347997 --- /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 000000000..57e756971 --- /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 cc43f7da4..9931152cd 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 5826be230..fa111e259 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 b6837856f..c7a180da5 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)