diff --git a/crates/storage-postgres/src/delete_table.rs b/crates/storage-postgres/src/delete_table.rs index f03d2c7f0..1fd1e2e76 100755 --- a/crates/storage-postgres/src/delete_table.rs +++ b/crates/storage-postgres/src/delete_table.rs @@ -4,14 +4,16 @@ //! `delete_table` implementation for `PostgresEngine`. //! //! The synchronous path (`control_plane_delay_seconds` below 1 and no reader -//! on the table's relations) removes the catalog row and drops the data -//! relations in the request. On one shared database both go in one commit. -//! With a separate data database the drops are run first and held -//! uncommitted, the catalog removal commits, and the data transaction -//! commits after it ([`drop_on_data_database`]); a data commit that fails -//! then has the drop retried on fresh connections for -//! [`DATA_DROP_RETRY_BOUND`] before the request gives up. What is left -//! behind when it does, and how to remove it, is stated at +//! on the table's relations) removes the table's catalog row and drops its +//! data relations in the request. On one shared database both go in one +//! commit. With a separate data database the drops are run first and held +//! uncommitted ([`drop_on_data_database`]); the row is committed `DELETING`, +//! the data transaction commits, and only then is the row removed +//! ([`remove_deleting_row`]). A data commit that fails has the drop retried +//! on fresh connections for [`DATA_DROP_RETRY_BOUND`]; a process that dies, +//! or a data database unreachable for that whole window, leaves the row +//! `DELETING` for the control-plane poller to finish, not data relations +//! without a row. What that costs, and the recovery, is stated at //! [`DATA_DROP_RETRY_BOUND`]. use std::time::{Duration, Instant}; @@ -233,19 +235,25 @@ impl PostgresEngine { .await? { // Separate data database: no transaction covers both. The - // drops hold their locks, uncommitted, in `data_tx`; the - // catalog removal is made here and the two commits follow, - // catalog first, below. - 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()))?; + // drops hold their locks, uncommitted, in `data_tx`. The row + // is marked DELETING here, not removed, and the catalog + // commit and the data commit follow, catalog first, below; + // the row is removed only once the data commit has landed + // (`remove_deleting_row`). A process that dies, or a data + // database unreachable past `DATA_DROP_RETRY_BOUND`, between + // the two leaves a DELETING row the control-plane poller + // finishes, not data tables that no catalog row names. The + // transition is due at once (`NOW()`), so the poller takes it + // the moment this request's own task does not. + sqlx::query( + r"UPDATE tables SET table_status = 'DELETING', + status_transition_at = NOW() + WHERE table_id = $1", + ) + .bind(&row.table_id) + .execute(&mut *tx) + .await + .map_err(|e| StorageError::Internal(e.to_string()))?; data_commit = Some(data_tx); true } else { @@ -257,42 +265,75 @@ impl PostgresEngine { if removed { if let Some(data_tx) = data_commit { - // Both commits, and the retries a failed data commit needs, - // run in a task of their own, which the request awaits: a - // caller cancelled from here on cannot stop between the - // commits and leave the row gone with the data tables kept, - // and cannot stop the retries either (dropping the request - // detaches the task; it does not cancel it). The drops - // already hold their locks, so the data commit does not - // wait. Once the catalog commit is in, the row is gone - // whatever happens next: a data commit that fails (the - // connection terminated, a failover, a network fault) has - // the drop redone on a fresh connection by - // `drop_after_catalog_commit`, and only a process that dies, - // or a data database unreachable for longer than - // `DATA_DROP_RETRY_BOUND`, between the catalog commit and - // the last retry leaves the data tables without a row. - let data_pool = self.data_pool.clone(); + // The catalog commit, the data commit, the retries a failed + // data commit needs, and the row removal that a landed data + // commit allows run in a task of their own, which the request + // awaits. The row is DELETING across the whole window, so a + // caller cancelled from here on (dropping the request only + // detaches the task; it does not cancel it) and a process + // that dies both leave the row DELETING, never gone with the + // data tables kept. The drops already hold their locks, so + // the data commit does not wait. A data commit that fails + // (the connection terminated, a failover, a network fault) + // has the drop redone on a fresh connection by + // `drop_after_catalog_commit`; the row is removed only once + // the data relations are gone. A process that dies, or a data + // database unreachable for longer than + // `DATA_DROP_RETRY_BOUND`, between the catalog commit and the + // last retry leaves the row DELETING for the control-plane + // poller to finish, so the drop is retried rather than the + // data tables leaked. + let engine = self.clone(); + let table_arn = row.table_arn.clone(); let table_id = row.table_id.clone(); + let task_index_ids = index_ids.clone(); + let task_vector_index_ids = vector_index_ids.clone(); tokio::spawn(async move { tx.commit() .await .map_err(|e| StorageError::Internal(e.to_string()))?; #[cfg(feature = "test-hooks")] test_pause::reached(test_pause::DeleteTablePoint::BetweenCommits).await; - match data_tx.commit().await { + // Commit the data drop; a failed commit (a lost + // connection, a failover, a network fault) has the drop + // redone on a fresh connection, bounded by + // `DATA_DROP_RETRY_BOUND`. + let data_dropped = match data_tx.commit().await { Ok(()) => Ok(()), Err(e) => { drop_after_catalog_commit( - &data_pool, + &engine.data_pool, &table_id, - &index_ids, - &vector_index_ids, + &task_index_ids, + &task_vector_index_ids, &e, ) .await } + }; + if let Err(drop_err) = data_dropped { + // The data relations could not be dropped within the + // retry bound, and the row stays DELETING. The wake + // starts a control-plane pass now. When the bound ran + // out on lock waits, that pass defers the drop while + // the holder remains, and the poller retries it every + // second until the holder ends. When the data database + // is unreachable, that pass fails and the poller goes + // idle; its idle pass, every 60 s, finishes the drop + // once the data database answers. + engine.control_plane_notify.notify_one(); + return Err(drop_err); + } + #[cfg(feature = "test-hooks")] + test_pause::reached(test_pause::DeleteTablePoint::AfterDataCommit).await; + // The data relations are gone; remove the catalog row. + // When another session holds it, a pass removes it: the + // pass that holds it, or the next one, which the wake + // starts, once the holder is done. + if !remove_deleting_row(&engine.pool, &table_arn, &table_id).await? { + engine.control_plane_notify.notify_one(); } + Ok(()) }) .await .map_err(|e| StorageError::Internal(e.to_string()))??; @@ -374,17 +415,19 @@ pub(crate) const SYNC_DROP_LOCK_TIMEOUT: Duration = Duration::from_secs(1); /// names, and the `DROP TABLE` to run) is what they need to finish the /// delete by hand. /// -/// What is left behind when the retries are exhausted, or when the process -/// dies between the catalog commit and the last retry: the table's data -/// relations (`_ddb_`, one `_ddb_` per GSI or LSI, one -/// `_ddb_vec_` per vector index) on the data database with no -/// catalog row naming them, and possibly a `vector_index_holds` row for the -/// table id, which the hold sweep clears by age. Nothing in the server drops -/// such relations later: `DescribeTable` reports the table gone, a -/// `CreateTable` of the same name gets a new table id, and the control -/// plane enumerates relations from catalog rows only. They stay until an -/// operator drops them (`docs/troubleshooting.md`, "Data tables without a -/// catalog row"). On one shared database the row and the drops are one +/// When the retries are exhausted, or when the process dies between the +/// catalog commit and the last retry, the table's catalog row stays +/// `DELETING` with its transition due, and the control-plane poller (the +/// startup pass, or a later pass) drops the data relations and removes the +/// row once the data database is reachable again (`worker_store.rs`). The +/// data relations (`_ddb_`, one `_ddb_` per GSI or LSI, +/// one `_ddb_vec_` per vector index) are therefore not left on the +/// data database without a catalog row naming them: the `DELETING` row names +/// them until the poller drops them. The error the exhausted request returns +/// and the error log it writes (the table id, the relation names, the `DROP +/// TABLE` to run) are for the operator watching the outage, who can finish +/// the drop by hand; the poller finishes it without them once the data +/// database answers. On one shared database the row and the drops are one /// commit, and none of this applies. pub(crate) const DATA_DROP_RETRY_BOUND: Duration = Duration::from_secs(60); @@ -398,8 +441,8 @@ fn data_drop_retry_bound() -> Duration { DATA_DROP_RETRY_BOUND } -/// Separate data database: the catalog commit is in and the data commit -/// failed with `commit_error`, so the table's row is gone while its data +/// Separate data database: the catalog commit is in (the row is `DELETING`) +/// and the data commit failed with `commit_error`, so the table's data /// relations may still exist (a commit whose reply was lost may also have /// gone through; nothing here can tell). Redo the drop on fresh connections /// from `data` until it succeeds or [`DATA_DROP_RETRY_BOUND`] has passed. @@ -413,11 +456,11 @@ fn data_drop_retry_bound() -> Duration { /// backoff from 100 ms doubling to 5 s: a lock wait that fails /// (`lock_wait_failed`) because a session took a lock after the first drop's /// locks were released with its connection, a connection the pool cannot -/// open, a statement the server did not answer. On success the delete is -/// complete and the caller returns the description as usual. When the bound -/// is exhausted the relation names and the table id are logged at error with -/// the `DROP TABLE` that removes them, and the request fails with -/// `Internal`; the row is gone either way. +/// open, a statement the server did not answer. On success the data +/// relations are gone and the caller removes the `DELETING` row. When the +/// bound is exhausted the relation names and the table id are logged at +/// error with the `DROP TABLE` that removes them, and `Err` is returned; the +/// row stays `DELETING` and the control-plane poller finishes the drop. async fn drop_after_catalog_commit( data: &sqlx::PgPool, table_id: &str, @@ -431,7 +474,7 @@ async fn drop_after_catalog_commit( tracing::warn!( table_id, ?relations, - "DeleteTable: the catalog row is removed but the data commit failed ({commit_error}); \ + "DeleteTable: the catalog row is DELETING but the data commit failed ({commit_error}); \ retrying the drop of the data relations for up to {bound:?}" ); let mut backoff = Duration::from_millis(100); @@ -470,13 +513,13 @@ async fn drop_after_catalog_commit( tracing::error!( table_id, ?relations, - "DeleteTable: the catalog row of table {table_id} is removed but its data relations \ - could not be dropped in {attempts} attempt(s) over {:?} ({last_error}); they stay on \ - the data database until dropped by hand: {drop_statement}", + "DeleteTable: table {table_id} is DELETING but its data relations could not be \ + dropped in {attempts} attempt(s) over {:?} ({last_error}); the control plane finishes \ + the drop once the data database is reachable, or run it by hand: {drop_statement}", started.elapsed() ); Err(StorageError::Internal(format!( - "DeleteTable: table {table_id} is removed from the catalog but its data relations \ + "DeleteTable: table {table_id} is DELETING but its data relations \ {relations:?} could not be dropped ({last_error})" ))) } @@ -489,11 +532,11 @@ async fn drop_after_catalog_commit( /// is returned, and the data transaction rolls back when it is dropped. /// /// Nothing commits here, so a failure up to and including the catalog commit -/// leaves the table as it was. After the catalog commit the row is gone, and -/// a data commit that fails for any reason (not only a process death) is -/// followed by [`drop_after_catalog_commit`]'s retries; what a process death -/// or an outage longer than [`DATA_DROP_RETRY_BOUND`] leaves behind is -/// stated there. +/// leaves the table as it was. After the catalog commit the row is +/// `DELETING`, and a data commit that fails for any reason (not only a +/// process death) is followed by [`drop_after_catalog_commit`]'s retries; +/// what a process death or an outage longer than [`DATA_DROP_RETRY_BOUND`] +/// leaves behind is stated there. /// /// The order is `remove_table_and_storage`'s: the pending queue rows first, /// with no timeout (a claim in flight is short and the drop waits for it, as @@ -550,6 +593,52 @@ async fn drop_on_data_database( } } +/// Separate data database: the data relations are gone, so remove the +/// table's tags and catalog row, in a catalog transaction of its own after +/// the row was committed `DELETING` and the data drop has landed. +/// +/// The row is locked first, without waiting (`FOR UPDATE SKIP LOCKED`), and +/// the tags are deleted after it. That is a control-plane pass's order: it +/// locks the row, deletes the tags, then deletes the row. Deleting the tags +/// first and then waiting for a row a pass holds deadlocks with that pass +/// when the table has tags. `Ok(false)`: the row is already removed, or +/// another session holds it (a control-plane pass, or a request that returns +/// at once on a `DELETING` row). Nothing is changed then, and the caller +/// wakes the poller: the pass that holds the row removes it, or the next +/// pass does once the holder is done. +async fn remove_deleting_row( + catalog: &sqlx::PgPool, + table_arn: &str, + table_id: &str, +) -> Result { + let internal = |e: sqlx::Error| StorageError::Internal(e.to_string()); + let mut tx = catalog.begin().await.map_err(internal)?; + let locked: Option = sqlx::query_scalar( + "SELECT 1 FROM tables WHERE table_id = $1 AND table_status = 'DELETING' \ + FOR UPDATE SKIP LOCKED", + ) + .bind(table_id) + .fetch_optional(&mut *tx) + .await + .map_err(internal)?; + if locked.is_none() { + tx.rollback().await.map_err(internal)?; + return Ok(false); + } + sqlx::query("DELETE FROM tags WHERE resource_arn = $1") + .bind(table_arn) + .execute(&mut *tx) + .await + .map_err(internal)?; + sqlx::query("DELETE FROM tables WHERE table_id = $1 AND table_status = 'DELETING'") + .bind(table_id) + .execute(&mut *tx) + .await + .map_err(internal)?; + tx.commit().await.map_err(internal)?; + Ok(true) +} + /// One shared database: drop the table's storage and remove its tags and /// catalog row in `tx`, behind a savepoint. `Ok(true)`: done, for the caller /// to commit. `Ok(false)`: a lock wait failed (`data::lock_wait_failed`: @@ -643,9 +732,15 @@ pub(crate) mod test_pause { /// on the table in the window a reader the probe missed has. BeforeDrops, /// Separate data database only: between the catalog commit and the - /// data commit of the synchronous path, so a test can make the data - /// commit fail with the row already gone. + /// data commit of the synchronous path, with the row `DELETING` and + /// the drops held uncommitted, so a test can make the data commit + /// fail or kill the request there. BetweenCommits, + /// Separate data database only: after the data drop has committed + /// (or been redone after a failed commit) and before the `DELETING` + /// row is removed, so a test can stop the request with the data + /// relations gone and the row still there. + AfterDataCommit, } static PAUSE: PausePoint = PausePoint::new(); diff --git a/crates/storage-postgres/src/lib.rs b/crates/storage-postgres/src/lib.rs index 9603d68ff..13f916f80 100755 --- a/crates/storage-postgres/src/lib.rs +++ b/crates/storage-postgres/src/lib.rs @@ -83,8 +83,9 @@ pub use data::vector_index::reconcile_incomplete_vector_indexes; #[doc(hidden)] pub use data::{data_table_name, index_table_name, vector_table_name}; /// Hold a `DeleteTable` between its in-use probe and its drops, or (separate -/// data database) between its catalog commit and its data commit; and shorten -/// the bound on the drop retried after a failed data commit. +/// data database) between its catalog commit and its data commit, or after +/// its data commit; and shorten the bound on the drop retried after a failed +/// data commit. #[cfg(feature = "test-hooks")] #[doc(hidden)] pub use delete_table::test_pause::{ diff --git a/crates/storage-postgres/tests/table_snapshot.rs b/crates/storage-postgres/tests/table_snapshot.rs index adfad63b2..56b730f07 100644 --- a/crates/storage-postgres/tests/table_snapshot.rs +++ b/crates/storage-postgres/tests/table_snapshot.rs @@ -2571,12 +2571,10 @@ async fn make_data_unreachable(s: &Scratch) { } /// Separate data database: when the data commit fails after the catalog -/// commit, the spawned task retries the drop on a fresh connection, so the -/// delete still succeeds and no data table is left without its row. This is -/// the window the PR used to describe as "a process that dies between the -/// two commits"; any failure of the data commit reaches it, with the process -/// up, and the retry closes it. Mutant: return the commit error instead of -/// retrying, and the relations stay. +/// commit, the row is DELETING from the catalog commit on, and the spawned +/// task redoes the drop on a fresh connection, so `DeleteTable` still +/// returns `Ok` and no data table is left without its row. Mutant: return +/// the commit error instead of redoing the drop, and the relations stay. #[tokio::test] async fn a_failed_data_commit_after_the_catalog_commit_is_retried() { if base_conn().is_none() { @@ -2602,8 +2600,8 @@ async fn a_failed_data_commit_after_the_catalog_commit_is_retried() { } assert_eq!( catalog_status(&s, "recommit").await.as_deref(), - None, - "the catalog commit is in by the pause" + Some("DELETING"), + "the row is DELETING by the pause, not removed" ); terminate_held_data_backend(&s).await; pause.resume(); @@ -2630,15 +2628,18 @@ async fn a_failed_data_commit_after_the_catalog_commit_is_retried() { } /// Separate data database: when the data database stays unreachable past the -/// retry bound, the delete gives up, names the leaked relations in its error -/// (and logs them with the `DROP TABLE` to run), and the catalog row is gone -/// regardless. A test-shortened bound keeps it quick. +/// retry bound, the delete gives up and names the relations it could not drop +/// in its error (and logs them with the `DROP TABLE` to run), but the catalog +/// row stays DELETING, not removed. Once the data database is reachable again +/// the control-plane poller finishes the drop and removes the row, so the +/// relations are never left without one. A test-shortened bound keeps it +/// quick. #[tokio::test] -async fn an_unreachable_data_database_past_the_bound_reports_the_leaked_relations() { +async fn an_unreachable_data_database_past_the_bound_leaves_the_row_deleting_for_the_poller() { if base_conn().is_none() { eprintln!( - "SKIP an_unreachable_data_database_past_the_bound_reports_the_leaked_relations: no \ - PostgreSQL" + "SKIP an_unreachable_data_database_past_the_bound_leaves_the_row_deleting_for_the_poller: \ + no PostgreSQL" ); return; } @@ -2676,12 +2677,15 @@ async fn an_unreachable_data_database_past_the_bound_reports_the_leaked_relation } assert_eq!( catalog_status(&s, "stranded").await.as_deref(), - None, - "the catalog row is removed even though the data drop could not be redone" + Some("DELETING"), + "the row stays DELETING when the data drop could not be redone, not removed" ); - // Let the engine reach the data database again so the Drop cleanup can - // force-drop it. + + // The data database is reachable again; the control-plane poller finishes + // the drop and removes the row, leaving nothing without a catalog row. set_data_allow_connections(&s, true).await; + assert_no_orphan(&s, "stranded", &relations, "after the bound was exhausted").await; + pump_until_removed(&s, "stranded", &relations).await; } /// Separate data database: a caller that gives up after the catalog commit @@ -2729,6 +2733,410 @@ async fn a_caller_cancelled_between_the_commits_does_not_strand_the_data() { panic!("the detached task did not finish the drop after the caller was cancelled"); } +/// Run a `DeleteTable` of `table` the way another server process would, on +/// its own runtime with its own engine, until it signals `reached` (a pause +/// the caller armed), then drop that runtime, the way a kill does: the task +/// running the request is dropped where it waits, with whatever transactions +/// it holds. Returns once no session holds a lock on `relations`, so the +/// server has rolled back what the dead process left open. +async fn kill_delete_table_at_the_pause( + s: &Scratch, + table: &str, + relations: &[String], + reached: tokio::sync::oneshot::Receiver<()>, +) { + let catalog_url = format!("{}/{}", s.base, s.names[0]); + let table_name = table.to_owned(); + let (kill_tx, kill_rx) = std::sync::mpsc::channel::<()>(); + let dying = std::thread::spawn(move || { + let rt = tokio::runtime::Builder::new_multi_thread() + .worker_threads(2) + .enable_all() + .build() + .expect("build the dying process's runtime"); + let engine = rt.block_on(async { + PostgresEngine::new( + &PostgresConfig { + connection_string: catalog_url, + pool_size: 8, + max_item_size_bytes: 400_000, + }, + REGION, + ) + .await + .expect("open the dying process's engine") + }); + rt.spawn(async move { + let _ = engine + .delete_table(ACCOUNT, delete_input(&table_name)) + .await; + }); + // Block the thread until the test has seen the pause reached, then + // "die": dropping the runtime drops the task mid-request. + let _ = kill_rx.recv(); + drop(rt); + }); + + tokio::time::timeout(std::time::Duration::from_secs(30), reached) + .await + .expect("the dying DeleteTable reached the pause in time") + .expect("the pause was reached"); + kill_tx + .send(()) + .expect("signal the dying process to drop its runtime"); + dying.join().expect("the dying process's thread ends"); + + // Wait for the server to notice the dropped connections and roll back + // what they held, releasing the locks. + let data = s.data_pool().await; + for _ in 0..200 { + let held: bool = sqlx::query_scalar( + "SELECT EXISTS (SELECT 1 FROM pg_locks WHERE locktype = 'relation' AND granted \ + AND pid <> pg_backend_pid() \ + AND database = (SELECT oid FROM pg_database WHERE datname = current_database()) \ + AND relation = ANY (SELECT to_regclass(n)::oid FROM unnest($1::text[]) AS n))", + ) + .bind(relations) + .fetch_one(&data) + .await + .expect("probe the data relations' locks"); + if !held { + data.close().await; + return; + } + tokio::time::sleep(std::time::Duration::from_millis(25)).await; + } + panic!("a session still holds the data relations 5 s after the kill"); +} + +/// Separate data database: a process that dies between the catalog commit and +/// the data commit leaves the row DELETING, so the next process's startup +/// pass finishes the drop. The request runs on its own runtime with its own +/// engine; dropping that runtime drops the task holding the uncommitted data +/// drops, the way a kill does, so the data drops roll back and both data +/// tables remain. The DELETING row still names them, and the startup pass +/// drops the tables and removes the row. +#[tokio::test] +async fn process_death_between_the_commits_is_finished_by_the_startup_pass() { + if base_conn().is_none() { + eprintln!( + "SKIP process_death_between_the_commits_is_finished_by_the_startup_pass: no PostgreSQL" + ); + return; + } + let s = scratch_with(false).await; + create_active(&s, snapshot_conformance::table_input("killed")).await; + snapshot_conformance::populate(&s.engine, ACCOUNT, "killed", 5).await; + let relations = base_and_gsi_relations(&s, "killed").await; + + // The dying process reaches the between-commits pause after the catalog + // commit, with the data drops held uncommitted, and is killed there. + let mut pause = extenddb_storage_postgres::arm_delete_table_pause_at( + extenddb_storage_postgres::DeleteTablePoint::BetweenCommits, + ); + let reached = pause.take_reached(); + kill_delete_table_at_the_pause(&s, "killed", &relations, reached).await; + + // The row is DELETING and names the data tables the rolled-back drops + // left behind. + assert_eq!( + catalog_status(&s, "killed").await.as_deref(), + Some("DELETING"), + "the row is left DELETING, not removed, after the process died between the commits" + ); + assert_eq!( + present_relations(&s, &relations).await, + relations, + "the interrupted data drop rolled back, so both data tables remain" + ); + assert_no_orphan(&s, "killed", &relations, "after the process died").await; + + // The next process's startup pass finishes the delete: drops the data + // tables, then removes the row. + s.engine.recover_control_plane_at_startup_for_tests().await; + assert_eq!( + catalog_status(&s, "killed").await, + None, + "the startup pass removed the row" + ); + assert!( + present_relations(&s, &relations).await.is_empty(), + "the startup pass dropped every data relation" + ); + drop(pause); +} + +/// Separate data database: a process that dies after the data commit and +/// before the row removal leaves the row DELETING with no data relations, +/// and the next process's startup pass removes the row. +#[tokio::test] +async fn process_death_after_the_data_commit_is_finished_by_the_startup_pass() { + if base_conn().is_none() { + eprintln!( + "SKIP process_death_after_the_data_commit_is_finished_by_the_startup_pass: \ + no PostgreSQL" + ); + return; + } + let s = scratch_with(false).await; + create_active(&s, snapshot_conformance::table_input("dropped")).await; + snapshot_conformance::populate(&s.engine, ACCOUNT, "dropped", 5).await; + let relations = base_and_gsi_relations(&s, "dropped").await; + + let mut pause = extenddb_storage_postgres::arm_delete_table_pause_at( + extenddb_storage_postgres::DeleteTablePoint::AfterDataCommit, + ); + let reached = pause.take_reached(); + kill_delete_table_at_the_pause(&s, "dropped", &relations, reached).await; + + assert_eq!( + catalog_status(&s, "dropped").await.as_deref(), + Some("DELETING"), + "the row is DELETING after the process died past the data commit" + ); + assert!( + present_relations(&s, &relations).await.is_empty(), + "the data commit dropped every data relation before the process died" + ); + + s.engine.recover_control_plane_at_startup_for_tests().await; + assert_eq!( + catalog_status(&s, "dropped").await, + None, + "the startup pass removed the row" + ); + drop(pause); +} + +/// Separate data database: past the data commit the request removes the row +/// itself; no control-plane pass runs here. Mutant: skip the call to +/// `remove_deleting_row`, and the row is still DELETING after `DeleteTable` +/// returns. +#[tokio::test] +async fn delete_table_removes_the_row_after_the_data_commit() { + if base_conn().is_none() { + eprintln!("SKIP delete_table_removes_the_row_after_the_data_commit: no PostgreSQL"); + return; + } + let s = scratch_with(false).await; + create_active(&s, snapshot_conformance::table_input("removed")).await; + snapshot_conformance::populate(&s.engine, ACCOUNT, "removed", 5).await; + let relations = base_and_gsi_relations(&s, "removed").await; + + let mut pause = extenddb_storage_postgres::arm_delete_table_pause_at( + extenddb_storage_postgres::DeleteTablePoint::AfterDataCommit, + ); + let reached = pause.take_reached(); + let mut delete = Box::pin(tokio::time::timeout( + std::time::Duration::from_secs(30), + s.engine.delete_table(ACCOUNT, delete_input("removed")), + )); + tokio::select! { + r = &mut delete => panic!("DeleteTable finished before the after-data-commit pause: {r:?}"), + r = reached => r.expect("DeleteTable reached the after-data-commit pause"), + } + assert_eq!( + catalog_status(&s, "removed").await.as_deref(), + Some("DELETING"), + "the row is DELETING at the pause" + ); + assert!( + present_relations(&s, &relations).await.is_empty(), + "the data commit dropped every data relation by the pause" + ); + + pause.resume(); + let deleted = delete + .await + .expect("DeleteTable returns within 30 s") + .expect("DeleteTable returns Ok"); + assert_eq!( + deleted.table_status, + extenddb_core::types::TableStatus::Deleting + ); + assert_eq!( + catalog_status(&s, "removed").await, + None, + "the request removed the row" + ); +} + +/// Whether a session on the catalog database waits for a lock it has not +/// been granted. +async fn catalog_lock_waiter(s: &Scratch) -> bool { + sqlx::query_scalar( + "SELECT EXISTS (SELECT 1 FROM pg_locks l JOIN pg_stat_activity a ON a.pid = l.pid \ + WHERE NOT l.granted AND a.datname = $1)", + ) + .bind(&s.names[0]) + .fetch_one(&s.catalog) + .await + .expect("catalog lock waiters") +} + +/// Separate data database: a control-plane pass holds the DELETING row when +/// the request comes to remove it. A pass locks the row (`FOR UPDATE SKIP +/// LOCKED`), then deletes the table's tags, then the row, and commits at the +/// end of the pass. An outside catalog transaction runs those statements in +/// that order here, since no hook holds a pass between its select and its +/// tag delete. The request takes the row first too, without waiting for it, +/// so it leaves the tags and the row to the pass and returns `Ok`. Mutant: +/// delete the tags, then the row, with no lock taken on the row first, and +/// on a tagged table the request waits for the row while the pass waits for +/// the tags, until the server's deadlock check fails one of them. Mutant: +/// take the row with a `FOR UPDATE` that waits, and the request waits for +/// the pass. +async fn a_pass_holding_the_deleting_row_does_not_fail_delete_table(tagged: bool) { + let s = scratch_with(false).await; + let table = if tagged { "tagged" } else { "untagged" }; + create_active(&s, snapshot_conformance::table_input(table)).await; + snapshot_conformance::populate(&s.engine, ACCOUNT, table, 5).await; + let relations = base_and_gsi_relations(&s, table).await; + let (arn, table_id): (String, String) = + sqlx::query_as("SELECT table_arn, table_id FROM tables WHERE table_name = $1") + .bind(table) + .fetch_one(&s.catalog) + .await + .expect("the table's arn and id"); + if tagged { + sqlx::query("INSERT INTO tags (resource_arn, tag_key, tag_value) VALUES ($1, 'k', 'v')") + .bind(&arn) + .execute(&s.catalog) + .await + .expect("tag the table"); + } + + let mut pause = extenddb_storage_postgres::arm_delete_table_pause_at( + extenddb_storage_postgres::DeleteTablePoint::BetweenCommits, + ); + let reached = pause.take_reached(); + let mut delete = Box::pin(tokio::time::timeout( + std::time::Duration::from_secs(30), + s.engine.delete_table(ACCOUNT, delete_input(table)), + )); + tokio::select! { + r = &mut delete => panic!("DeleteTable finished before the between-commits pause: {r:?}"), + r = reached => r.expect("DeleteTable reached the between-commits pause"), + } + + // The pass takes the DELETING row, which is due at once. + let mut pass_conn = PgConnection::connect(&format!("{}/{}", s.base, s.names[0])) + .await + .expect("the pass's connection"); + let mut pass = pass_conn.begin().await.expect("begin the pass"); + let candidates: Vec = sqlx::query_scalar( + "SELECT table_id FROM tables \ + WHERE table_status = 'DELETING' AND status_transition_at <= NOW() \ + FOR UPDATE SKIP LOCKED", + ) + .fetch_all(&mut *pass) + .await + .expect("the pass's select"); + assert_eq!(candidates, vec![table_id.clone()], "the pass holds the row"); + + // The request commits its data drop and comes to remove the row. Wait + // until it has returned or waits on a lock. + pause.resume(); + let mut returned = None; + let mut waiting = false; + for _ in 0..200 { + tokio::select! { + r = &mut delete => { + returned = Some(r); + break; + } + () = tokio::time::sleep(std::time::Duration::from_millis(25)) => {} + } + if catalog_lock_waiter(&s).await { + waiting = true; + break; + } + } + assert!( + returned.is_some() || waiting, + "DeleteTable neither returned nor waited on a lock within 5 s" + ); + + // The pass's next statements for this row: its tags, then the row. + let tags_deleted = tokio::time::timeout( + std::time::Duration::from_secs(10), + sqlx::query("DELETE FROM tags WHERE resource_arn = $1") + .bind(&arn) + .execute(&mut *pass), + ) + .await + .expect("the pass's tag delete returns within 10 s"); + let pass_committed = match tags_deleted { + Ok(_) => { + sqlx::query("DELETE FROM tables WHERE table_id = $1") + .bind(&table_id) + .execute(&mut *pass) + .await + .expect("the pass deletes the row"); + pass.commit().await.map_err(|e| e.to_string()) + } + Err(e) => { + drop(pass); + Err(e.to_string()) + } + }; + let deleted = match returned { + Some(r) => r, + None => delete.await, + } + .expect("DeleteTable returns within 30 s") + .unwrap_or_else(|e| panic!("DeleteTable failed while a pass held the row: {e}")); + assert!( + pass_committed.is_ok(), + "the pass committed: {pass_committed:?}" + ); + assert!(!waiting, "DeleteTable waited for the row the pass holds"); + assert_eq!( + deleted.table_status, + extenddb_core::types::TableStatus::Deleting + ); + assert_eq!( + catalog_status(&s, table).await, + None, + "the pass removed the row" + ); + let tags: i64 = sqlx::query_scalar("SELECT count(*) FROM tags WHERE resource_arn = $1") + .bind(&arn) + .fetch_one(&s.catalog) + .await + .expect("count the tags"); + assert_eq!(tags, 0, "the pass removed the tags"); + assert!( + present_relations(&s, &relations).await.is_empty(), + "the request dropped every data relation" + ); + pass_conn + .close() + .await + .expect("close the pass's connection"); + drop(pause); +} + +mod a_pass_holding_the_deleting_row_does_not_fail_delete_table { + #[tokio::test] + async fn tagged() { + if super::base_conn().is_none() { + eprintln!("SKIP {}: no PostgreSQL", module_path!()); + return; + } + super::a_pass_holding_the_deleting_row_does_not_fail_delete_table(true).await; + } + + #[tokio::test] + async fn untagged() { + if super::base_conn().is_none() { + eprintln!("SKIP {}: no PostgreSQL", module_path!()); + return; + } + super::a_pass_holding_the_deleting_row_does_not_fail_delete_table(false).await; + } +} + /// A hash-only table with three GSIs, for the drop-deadline test. fn three_gsi_input(name: &str) -> extenddb_core::types::CreateTableInput { let gsi = |i: usize| { diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index 11a35c1b5..7587a0c1c 100755 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -389,7 +389,7 @@ SELECT l.pid, l.mode, a.query_start, l.relation::regclass AS relation, a.query ### Data tables without a catalog row (PostgreSQL, separate data database) -**Cause:** With a separate data database, a synchronous DeleteTable commits the catalog row removal first, then commits the drop of the table's data tables on the data database. If that data commit fails (the connection is terminated, a failover, a network fault), the server redoes the drop on a fresh connection for up to 60 seconds. Only a process that dies, or a data database unreachable for that whole window, between the catalog commit and the last retry leaves the table's data tables on the data database with no catalog row naming them. Nothing reclaims them afterwards: DescribeTable reports the table gone, a same-name CreateTable gets a new table id, and the control plane only enumerates relations that still have a catalog row. When the retries are exhausted the server logs, at error, the table id, the relation names, and the exact `DROP TABLE` to run. (On one shared database the row and the drops commit together, so this cannot happen there.) +**Cause:** A table's data relations (`_ddb_%`) are on the data database with no `tables` row naming them. Current releases do not create this state from an interrupted DeleteTable: with a separate data database a synchronous DeleteTable marks the catalog row `DELETING`, commits it, drops the table's data tables on the data database, and removes the row only after that drop has committed. If the data commit fails (the connection is terminated, a failover, a network fault), the server redoes the drop on a fresh connection for up to 60 seconds. A process that dies, or a data database unreachable for that whole window, between the catalog commit and the last retry leaves the row `DELETING` with its transition due; the control-plane poller (the next startup pass, or a later one) drops the data tables and removes the row once the data database is reachable again, so the relations are not left without a row. When the retries are exhausted the server logs, at error, the table id, the relation names, and the exact `DROP TABLE` to run, for an operator who would rather finish the drop by hand than wait for the poller. (On one shared database the row and the drops commit together, so none of this applies.) Relations a release from before this behaviour shipped left without a row are not swept automatically; find and drop them as below. To find such relations, compare the `_ddb_%` tables on the data database with the table ids the catalog still knows. On the catalog database: