fix(postgres): keep a deleted table DELETING until its data tables drop

.github/workflows/integration.yml

@@ -310,8 +310,8 @@ jobs:
         run: >-
           cargo test --release -p extenddb-storage-postgres
           --test vector_control_plane --test put_create_race --test twi_conflict
-          --test stream_shard_retention --test backup_restore
-          --test gsi_online_build --test gsi_stale_writer
+          --test stream_shard_retention --test backup_restore --test delete_table_interrupt
+          --test delete_table_lock_wait --test gsi_online_build --test gsi_stale_writer
 
       # The daemonized server logs to syslog; dump it so server-side failures
       # are diagnosable from the job log.

crates/storage-postgres/src/data/ddl.rs

@@ -673,8 +673,8 @@ impl PostgresEngine {
     /// rows' claims, a CreateBackup reading the table, a client transaction) is
     /// bounded; a timeout fails the call with nothing dropped, and the hold
     /// stays until the next attempt or the recovery sweep. The control plane
-    /// worker uses it so one table cannot stall a whole pass; DeleteTable itself
-    /// waits.
+    /// pass uses it, DeleteTable's pass for its own table (`finish_table_delete`)
+    /// included, so one table cannot stall a whole pass or a DeleteTable call.
     pub(crate) async fn drop_table_storage(
         data: &sqlx::PgPool,
         table_id: &str,

crates/storage-postgres/src/delete_table.rs

@@ -72,7 +72,7 @@ impl PostgresEngine {
 
         // Fetch indexes for the response description.
         let index_rows: Vec<IndexRow> = sqlx::query_as(
-            r"SELECT index_name, index_id, index_type, key_schema, projection,
+            r"SELECT index_name, index_type, key_schema, projection,
                       index_status, provisioned_throughput
                FROM indexes WHERE table_id = $1",
         )
@@ -81,83 +81,83 @@ impl PostgresEngine {
         .await
         .map_err(|e| StorageError::Internal(e.to_string()))?;
 
-        // Vector index ids, read while the rows still exist. Deleting the table
-        // row cascades them away, and each id names a data table that has to be
-        // dropped afterwards: the same ordering the GSI path uses, and for the same
-        // reason.
-        let vector_index_ids: Vec<String> =
-            sqlx::query_scalar("SELECT index_id FROM vector_indexes WHERE table_id = $1")
-                .bind(&row.table_id)
-                .fetch_all(&mut *tx)
-                .await
-                .map_err(|e| StorageError::Internal(e.to_string()))?;
-
-        // H-5 (delete): synchronous control-plane shortcut when delay < 1s.
-        // When control_plane_delay_seconds is small, the poller may not run
-        // before the next request, causing stale DELETING rows. Synchronous
-        // delete avoids this. Mirrors create_table's H-5 fast path.
+        // H-5 (delete): the control-plane delay decides whether the table is
+        // removed before this call answers or by the poller later. Both paths
+        // mark the row DELETING here and commit; the data tables are dropped
+        // afterwards by the control-plane pass, which deletes the catalog row
+        // only once the data transaction has committed. A server that dies
+        // between the two leaves a DELETING row that the startup pass finishes,
+        // instead of data tables that no catalog row names.
         let delay_row: (f64,) = sqlx::query_as(
             "SELECT COALESCE((SELECT value::FLOAT8 FROM settings WHERE key = 'control_plane_delay_seconds'), 0.25)",
         )
         .fetch_one(&mut *tx)
         .await
         .map_err(|e| StorageError::Internal(e.to_string()))?;
         let delay_secs = delay_row.0;
+        let immediate = delay_secs < 1.0;
+        let scheduled_delay = if immediate { 0.0 } else { delay_secs };
 
-        let index_ids: Vec<String> = index_rows.iter().map(|r| r.index_id.clone()).collect();
-
-        if delay_secs < 1.0 {
-            // Synchronous delete: remove tags, catalog row, and data tables inline.
-            // Note: deleting the tables row cascades to indexes and stream rows via FK CASCADE.
-            // Vector index rows cascade the same way, through
-            // vector_indexes_table_id_fkey, so a table with vector indexes needs
-            // no extra catalog cleanup. Their data tables do need dropping, and
-            // the cascade takes away the rows that name them, which is why the
-            // index ids are collected BEFORE the catalog row goes.
-            sqlx::query("DELETE FROM tags WHERE resource_arn = $1")
-                .bind(&row.table_arn)
-                .execute(&mut *tx)
-                .await
-                .map_err(|e| StorageError::Internal(e.to_string()))?;
-
-            sqlx::query("DELETE FROM tables WHERE table_id = $1")
-                .bind(&row.table_id)
-                .execute(&mut *tx)
-                .await
-                .map_err(|e| StorageError::Internal(e.to_string()))?;
-
-            tx.commit()
-                .await
-                .map_err(|e| StorageError::Internal(e.to_string()))?;
+        sqlx::query(
+            r"UPDATE tables SET table_status = 'DELETING',
+                status_transition_at = NOW() + make_interval(secs => $3)
+               WHERE account_id = $1 AND table_name = $2",
+        )
+        .bind(account_id)
+        .bind(&input.table_name)
+        .bind(scheduled_delay)
+        .execute(&mut *tx)
+        .await
+        .map_err(|e| StorageError::Internal(e.to_string()))?;
 
-            // Drop data tables after catalog commit. See `drop_table_storage` for
-            // why the queue is stopped before anything is dropped.
-            Self::drop_table_storage(
-                &self.data_pool,
-                &row.table_id,
-                &index_ids,
-                &vector_index_ids,
-                None,
-            )
-            .await?;
-        } else {
-            // Async delete: set DELETING status with scheduled removal time.
-            sqlx::query(
-                r"UPDATE tables SET table_status = 'DELETING',
-                    status_transition_at = NOW() + make_interval(secs => $3)
-                   WHERE account_id = $1 AND table_name = $2",
-            )
-            .bind(account_id)
-            .bind(&input.table_name)
-            .bind(delay_secs)
-            .execute(&mut *tx)
+        tx.commit()
             .await
             .map_err(|e| StorageError::Internal(e.to_string()))?;
 
-            tx.commit()
-                .await
-                .map_err(|e| StorageError::Internal(e.to_string()))?;
-
+        if immediate {
+            // Finish the delete before answering. In the common case (no lock
+            // held on the data tables) the name is free when the response
+            // arrives, and a poller that may not run before the next request
+            // leaves no stale DELETING row (mirrors create_table's H-5 fast
+            // path). The response is DELETING in every case. The exceptions,
+            // where the catalog row is still DELETING when the pass returns
+            // and the poller finishes the drop: a poller pass held the catalog
+            // row past 3 s, a lock wait in the drop or in the later catalog
+            // statements passed the pass's 3 s lock timeout, no pool
+            // connection was free within the acquire timeout, or the drop was
+            // the victim of a deadlock with a client transaction. The work
+            // runs in a spawned task, as
+            // UpdateTable runs a GSI build: a client that times out drops this
+            // future, and a drop that was waiting on a lock must not be rolled
+            // back with it. The task, not this future, wakes the poller when the
+            // table is not removed, so a client that is gone by then does not
+            // leave the row to the poller's idle timeout. If this future is
+            // dropped during the commit above and the commit landed, no task is
+            // spawned and the row stays DELETING until the poller's idle sweep;
+            // one round trip wide, same as the delayed path, no data at risk.
+            let engine = self.clone();
+            let table_id = row.table_id.clone();
+            let table_name = input.table_name.clone();
+            let task = tokio::spawn(async move {
+                match engine.finish_table_delete(&table_id).await {
+                    Ok(true) => return,
+                    Ok(false) => {}
+                    Err(e) => tracing::warn!(
+                        "DeleteTable '{table_name}': could not finish the delete inline, the \
+                         table remains DELETING and the control plane will retry: {e}"
+                    ),
+                }
+                engine.control_plane_notify.notify_one();
+            });
+            if let Err(e) = task.await {
+                tracing::warn!(
+                    "DeleteTable '{}': the delete task ended early, the table remains DELETING \
+                     and the control plane will retry: {e}",
+                    input.table_name
+                );
+                self.control_plane_notify.notify_one();
+            }
+        } else {
             // F-3: Wake the control plane poller so it processes the DELETING →
             // removed transition without waiting for the idle timeout.
             self.control_plane_notify.notify_one();

crates/storage-postgres/src/table_helpers.rs

@@ -40,7 +40,6 @@ pub(crate) struct TableRow {
 #[derive(sqlx::FromRow)]
 pub(crate) struct IndexRow {
     pub index_name: String,
-    pub index_id: String,
     pub index_type: String,
     pub key_schema: serde_json::Value,
     pub projection: serde_json::Value,
@@ -124,7 +123,7 @@ impl PostgresEngine {
         let row = row.ok_or_else(|| StorageError::TableNotFound(table_name.to_owned()))?;
 
         let index_rows: Vec<IndexRow> = sqlx::query_as(
-            r"SELECT index_name, index_id, index_type, key_schema, projection,
+            r"SELECT index_name, index_type, key_schema, projection,
                       index_status, provisioned_throughput
                FROM indexes WHERE table_id = $1",
         )

crates/storage-postgres/src/worker_store.rs

@@ -10,6 +10,16 @@ use extenddb_storage::error::StorageError;
 
 use crate::PostgresEngine;
 
+/// The `lock_timeout` of a pass's data drop, and of the catalog transaction of
+/// the pass `DeleteTable` runs for its own table.
+const PASS_LOCK_TIMEOUT: &str = "3s";
+
+/// Whether a `sqlx` error is PostgreSQL's lock_not_available (55P03), the
+/// error `lock_timeout` raises.
+fn is_lock_timeout(e: &sqlx::Error) -> bool {
+    matches!(e, sqlx::Error::Database(d) if d.code().as_deref() == Some("55P03"))
+}
+
 impl WorkerStore for PostgresEngine {
     fn process_control_plane_transitions(
         &self,
@@ -39,20 +49,53 @@ impl PostgresEngine {
     /// Returns [`StorageError`] if the database is unreachable or a query fails.
     pub async fn process_control_plane_transitions(
         &self,
+    ) -> Result<Vec<(String, &'static str)>, StorageError> {
+        self.process_transitions(None).await
+    }
+
+    /// Finish the delete of one table whose catalog row is `DELETING` with a
+    /// transition time that has passed: drop its data tables, then remove the
+    /// row. `DeleteTable` runs this from a spawned task so a client that gives
+    /// up on the request cannot cancel the drop part way.
+    ///
+    /// Returns `true` when this call removed the table. `false` means the row
+    /// was not a candidate (already gone, including when a poller pass that
+    /// held the row removed it while this call waited for the lock), the lock
+    /// on the row was not granted within the pass's lock timeout, or the drop
+    /// failed and was logged. Returns `Err` when the pool acquire or a
+    /// statement after the candidate select fails, a lock wait past the
+    /// timeout included. In every non-`true` case the row, if it still
+    /// exists, stays `DELETING` for the poller to finish.
+    ///
+    /// Public for the integration tests.
+    #[doc(hidden)]
+    pub async fn finish_table_delete(&self, table_id: &str) -> Result<bool, StorageError> {
+        let transitions = self.process_transitions(Some(table_id)).await?;
+        Ok(!transitions.is_empty())
+    }
+
+    /// The pass behind [`Self::process_control_plane_transitions`]. With
+    /// `only_table` set, only that table's `DELETING` row is a candidate and the
+    /// `CREATING` activation and the abandoned-restore sweep are skipped.
+    async fn process_transitions(
+        &self,
+        only_table: Option<&str>,
     ) -> Result<Vec<(String, &'static str)>, StorageError> {
         let mut transitions = Vec::new();
 
         // CREATING → ACTIVE
-        let activated: Vec<(String,)> = sqlx::query_as(
-            r"UPDATE tables SET table_status = 'ACTIVE', status_transition_at = NULL
-               WHERE table_status = 'CREATING' AND status_transition_at <= NOW()
-               RETURNING table_name",
-        )
-        .fetch_all(&self.pool)
-        .await
-        .map_err(|e| StorageError::Internal(e.to_string()))?;
-        for (name,) in activated {
-            transitions.push((name, "CREATING → active"));
+        if only_table.is_none() {
+            let activated: Vec<(String,)> = sqlx::query_as(
+                r"UPDATE tables SET table_status = 'ACTIVE', status_transition_at = NULL
+                   WHERE table_status = 'CREATING' AND status_transition_at <= NOW()
+                   RETURNING table_name",
+            )
+            .fetch_all(&self.pool)
+            .await
+            .map_err(|e| StorageError::Internal(e.to_string()))?;
+            for (name,) in activated {
+                transitions.push((name, "CREATING → active"));
+            }
         }
 
         // DELETING → remove row (with tags and data table cleanup).
@@ -62,20 +105,60 @@ impl PostgresEngine {
         // fails, the transaction rolls back and the row remains DELETING for
         // the next pass. Once the data transaction commits, deleting the table
         // row cascades its index and stream rows without orphaning data tables.
+        //
+        // The poller pass skips rows another pass holds. The scoped pass run by
+        // DeleteTable waits for them, for at most the 3 s `lock_timeout` it
+        // sets on this transaction. A poller pass that holds the row is
+        // removing the table. If that pass commits within that time having
+        // deleted the row, the select returns no row. If its drop of this
+        // table failed, the select returns the row and this pass drops it.
+        // Past the timeout the select fails with lock_not_available, the
+        // request answers with the row still DELETING, and the poller finishes
+        // the delete. The timeout also bounds the later lock waits in this
+        // transaction; a timeout there is an error and leaves the row DELETING
+        // for the poller.
         let mut tx = self
             .pool
             .begin()
             .await
             .map_err(|e| StorageError::Internal(e.to_string()))?;
+        if only_table.is_some() {
+            sqlx::query("SELECT set_config('lock_timeout', $1, true)")
+                .bind(PASS_LOCK_TIMEOUT)
+                .execute(&mut *tx)
+                .await
+                .map_err(|e| StorageError::Internal(e.to_string()))?;
+        }
 
-        let candidates: Vec<(String, String, String)> = sqlx::query_as(
-            r"SELECT table_name, table_arn, table_id FROM tables
-               WHERE table_status = 'DELETING' AND status_transition_at <= NOW()
-               FOR UPDATE SKIP LOCKED",
-        )
-        .fetch_all(&mut *tx)
-        .await
-        .map_err(|e| StorageError::Internal(e.to_string()))?;
+        let lock_clause = if only_table.is_some() {
+            "FOR UPDATE"
+        } else {
+            "FOR UPDATE SKIP LOCKED"
+        };
+        let candidate_sql = format!(
+            "SELECT table_name, table_arn, table_id FROM tables \
+             WHERE table_status = 'DELETING' AND status_transition_at <= NOW() \
+               AND ($1::TEXT IS NULL OR table_id = $1) \
+             {lock_clause}"
+        );
+        let candidates: Vec<(String, String, String)> = match sqlx::query_as(&candidate_sql)
+            .bind(only_table)
+            .fetch_all(&mut *tx)
+            .await
+        {
+            Ok(rows) => rows,
+            Err(e) if only_table.is_some() && is_lock_timeout(&e) => {
+                let _ = tx.rollback().await;
+                let table_id = only_table.unwrap_or_default();
+                tracing::info!(
+                    "table {table_id}: the lock on the catalog row was not granted within \
+                     {PASS_LOCK_TIMEOUT} (a poller pass removing the table is the usual \
+                     holder); DeleteTable leaves the table DELETING for the poller to finish"
+                );
+                return Ok(Vec::new());
+            }
+            Err(e) => return Err(StorageError::Internal(e.to_string())),
+        };
 
         for (name, arn, table_id) in &candidates {
             let index_ids: Vec<String> =
@@ -101,7 +184,7 @@ impl PostgresEngine {
                 table_id,
                 &index_ids,
                 &vector_index_ids,
-                Some("3s"),
+                Some(PASS_LOCK_TIMEOUT),
             )
             .await;
 
@@ -144,13 +227,15 @@ impl PostgresEngine {
         // lock; a still-owned restore closes that probe and is checked again on
         // the next pass. Failures are logged and skipped, so they cannot hold
         // up the transitions above.
-        match self.sweep_abandoned_restores().await {
-            Ok(names) => {
-                for name in names {
-                    transitions.push((name, "abandoned restore → removed"));
+        if only_table.is_none() {
+            match self.sweep_abandoned_restores().await {
+                Ok(names) => {
+                    for name in names {
+                        transitions.push((name, "abandoned restore → removed"));
+                    }
                 }
+                Err(e) => tracing::warn!("abandoned restore sweep failed: {e}"),
             }
-            Err(e) => tracing::warn!("abandoned restore sweep failed: {e}"),
         }
 
         Ok(transitions)