fix(cassandra): fence concurrent DeleteTable with a lease

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::<StorageError>(
@@ -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<Option<i64>, _> = row.get_by_name("deadline_ms");
+                if !matches!(deadline, Ok(None)) {
+                    continue;
+                }
                 let heartbeat: Option<i64> = row.get_by_name("restore_heartbeat").ok().flatten();
                 if heartbeat.is_some_and(|at| at >= cutoff) {
                     continue;

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)
 }