diff --git a/.github/workflows/integration.yml b/.github/workflows/integration.yml index 7fd08685e..1d7a7dd39 100644 --- a/.github/workflows/integration.yml +++ b/.github/workflows/integration.yml @@ -419,6 +419,8 @@ jobs: # table_snapshot: the SnapshotEngine consistency check; it needs the # shared page-size hook, compiled only into the crate's own test # build, and scratch databases it creates and drops itself. + # catalog_check: the data relations the engine creates compared with + # the catalog, on a deployment with a separate data database. env: EXTENDDB_TEST_PG_CONNECTION_STRING: postgresql://postgres:devpass@127.0.0.1:5432 # table_snapshot fails rather than skips without the string above. diff --git a/crates/app/src/cmd_catalog_check.rs b/crates/app/src/cmd_catalog_check.rs index 91744c1de..287eb11f3 100755 --- a/crates/app/src/cmd_catalog_check.rs +++ b/crates/app/src/cmd_catalog_check.rs @@ -1,23 +1,24 @@ // Copyright 2026 ExtendDB contributors // SPDX-License-Identifier: Apache-2.0 -//! `extenddb catalog-check` — integrity check for catalog and data databases. +//! `extenddb catalog-check`: integrity check for the catalog and data databases. //! -//! Detects orphaned data/GSI tables, missing data tables, and catalog -//! inconsistencies. Refuses to run while the server is up (PID file check). +//! Detects data relations no catalog row names, catalog rows whose data +//! relations are gone, and catalog inconsistencies. Refuses to run while the +//! server is up (PID file check). //! -//! This command uses raw `sqlx` queries intentionally. It is a diagnostic -//! tool that runs outside the server process and needs direct database -//! access to cross-reference catalog metadata against physical `PostgreSQL` -//! tables. Routing through the storage abstraction would defeat the purpose -//! of an independent integrity check. - -use std::collections::HashSet; +//! The data relations are named by the storage backend from catalog ids, so +//! checks 1 and 2 run through the backend's `DiagnosticsStore`, which +//! computes the expected names with the same functions the engine uses to +//! create them. Checks 3 and 4 read the catalog with raw `sqlx` queries, as +//! the command always has; they need direct access to cross-reference catalog +//! rows against each other. use clap::Args; use sqlx::postgres::PgPoolOptions; use extenddb_config as config; +use extenddb_storage::diagnostics::{DiagError, OrphanDrop}; #[derive(Args)] pub struct CatalogCheckArgs { @@ -30,9 +31,9 @@ pub struct CatalogCheckArgs { fix: bool, } -// Linear sequence of 4 independent checks — splitting into helpers would scatter -// the logic without improving readability, and each helper would need the same -// pool/fix parameters threaded through. +// A linear sequence of 4 independent checks. Splitting it into helpers would +// scatter the logic without improving readability, and each helper would need +// the same pool and fix parameters threaded through. #[allow(clippy::too_many_lines)] pub async fn run(args: CatalogCheckArgs) -> anyhow::Result<()> { if !std::path::Path::new(&args.config).exists() { @@ -69,95 +70,74 @@ pub async fn run(args: CatalogCheckArgs) -> anyhow::Result<()> { .map_err(|e| anyhow::anyhow!("Cannot connect to catalog: {e}"))?; println!("Connected to catalog."); - // Connect to data database. - let data_conn: Option<(String,)> = - sqlx::query_as("SELECT value FROM settings WHERE key = 'data_database_connection_string'") - .fetch_optional(&catalog_pool) - .await?; - let Some((data_conn_str,)) = data_conn else { - anyhow::bail!("No data database connection string in settings. Run `extenddb init`."); - }; - let data_pool = PgPoolOptions::new() - .max_connections(2) - .connect(&data_conn_str) + // The backend's own view of its data relations. It connects to the data + // database the catalog names, and refuses a catalog that names none, as + // this command always has. + let diag_store = extenddb_storage::diagnostics_store::create_diagnostics_store( + app_config.storage.connection_config(), + ) + .await + .map_err(|e| anyhow::anyhow!("Cannot open the storage backend's diagnostics: {e}"))?; + let relations = diag_store + .check_storage_relations() .await - .map_err(|e| anyhow::anyhow!("Cannot connect to data database: {e}"))?; - println!("Connected to data database."); + .map_err(|e| match e { + DiagError::NotConfigured(msg) => anyhow::anyhow!("{msg}"), + e => anyhow::anyhow!("Cannot compare data relations with the catalog: {e}"), + })?; + println!("Compared data relations with the catalog."); println!(); - // Build expected set from catalog: base tables and GSI tables. - // Include CREATING tables — they have data tables even before transitioning - // to ACTIVE. Excluding them would flag their data tables as orphaned. - let catalog_tables: Vec<(String, String)> = sqlx::query_as( - "SELECT account_id, table_name FROM tables \ - WHERE table_status IN ('ACTIVE', 'CREATING')", - ) - .fetch_all(&catalog_pool) - .await?; - - let catalog_indexes: Vec<(String, String, String)> = sqlx::query_as( - "SELECT t.account_id, t.table_name, i.index_name \ - FROM indexes i JOIN tables t ON i.table_id = t.table_id \ - WHERE t.table_status IN ('ACTIVE', 'CREATING')", - ) - .fetch_all(&catalog_pool) - .await?; - - let mut expected: HashSet = HashSet::new(); - for (acct, name) in &catalog_tables { - expected.insert(format!("_ddb_{acct}_{name}")); - } - for (acct, name, idx) in &catalog_indexes { - expected.insert(format!("_ddb_{acct}_{name}__gsi__{idx}")); - } - - // Get actual tables in data database (only _ddb_ prefixed). - let actual_tables: Vec<(String,)> = sqlx::query_as( - "SELECT tablename FROM pg_tables \ - WHERE schemaname = 'public' AND tablename LIKE '_ddb_%'", - ) - .fetch_all(&data_pool) - .await?; - - let actual: HashSet = actual_tables.into_iter().map(|(n,)| n).collect(); - - // Check 1: Orphaned data tables (in data DB but not in catalog). + // Check 1: Orphaned data tables (in data DB but no catalog row names them). println!("--- Checking for orphaned data tables..."); - let orphaned: Vec<&String> = actual.difference(&expected).collect(); - if orphaned.is_empty() { + if relations.orphaned.is_empty() { println!(" OK: No orphaned tables."); } else { - println!(" FOUND: {} orphaned table(s):", orphaned.len()); - for name in &orphaned { + println!(" FOUND: {} orphaned table(s):", relations.orphaned.len()); + for name in &relations.orphaned { println!(" - {name}"); } - errors += u32::try_from(orphaned.len()).unwrap_or(u32::MAX); + errors += u32::try_from(relations.orphaned.len()).unwrap_or(u32::MAX); if args.fix { println!(" Cleaning up orphaned tables..."); - for name in &orphaned { - // Table names are from pg_tables (trusted), but quote anyway. - let quoted = format!("\"{}\"", name.replace('"', "\"\"")); - let ddl = format!("DROP TABLE IF EXISTS {quoted}"); - match sqlx::query(&ddl).execute(&data_pool).await { - Ok(_) => println!(" Dropped: {name}"), - Err(e) => println!(" FAILED to drop {name}: {e}"), + // The backend waits for the index changes in flight, checks each + // relation against the catalog again, and drops each one that no + // row names with a short lock wait of its own. + match diag_store + .drop_orphaned_relations(relations.orphaned.clone()) + .await + { + Ok(outcomes) => { + for (name, outcome) in &outcomes { + match outcome { + OrphanDrop::Dropped => println!(" Dropped: {name}"), + OrphanDrop::AlreadyGone => println!( + " Already gone: {name} (another session dropped it first)" + ), + OrphanDrop::Kept(reason) => println!(" Kept: {name} ({reason})"), + OrphanDrop::Failed(e) => println!(" FAILED to drop {name}: {e}"), + } + } } + Err(e) => println!(" FAILED to clean up orphaned tables: {e}"), } } } - // Check 2: Missing data tables (in catalog but not in data DB). + // Check 2: Missing data tables (a catalog row names them, data DB lacks them). println!("--- Checking for missing data tables..."); - let missing: Vec<&String> = expected.difference(&actual).collect(); - if missing.is_empty() { + if relations.missing.is_empty() { println!(" OK: All catalog tables have backing data tables."); } else { - println!(" FOUND: {} missing data table(s):", missing.len()); - for name in &missing { - println!(" - {name}"); + println!( + " FOUND: {} missing data table(s):", + relations.missing.len() + ); + for entry in &relations.missing { + println!(" - {} ({})", entry.relation, entry.owner); } - errors += u32::try_from(missing.len()).unwrap_or(u32::MAX); + errors += u32::try_from(relations.missing.len()).unwrap_or(u32::MAX); } // Check 3: Indexes without parent table in catalog. diff --git a/crates/storage-postgres/src/catalog_store.rs b/crates/storage-postgres/src/catalog_store.rs index 834c88a5a..36f2cf13f 100755 --- a/crates/storage-postgres/src/catalog_store.rs +++ b/crates/storage-postgres/src/catalog_store.rs @@ -11,6 +11,7 @@ use std::sync::Arc; +use extenddb_storage::diagnostics::{DiagError, DiagResult, OrphanDrop, StorageRelationCheck}; use extenddb_storage::management_store::{MetricsRow, OpError, OpResult}; use futures::future::BoxFuture; use sqlx::PgPool; @@ -193,6 +194,74 @@ impl extenddb_storage::diagnostics::DiagnosticsStore for PostgresCatalogStore { } }) } + + fn check_storage_relations(&self) -> BoxFuture<'_, DiagResult> { + Box::pin(async move { + let data = data_pool(&self.pool).await?; + let budget = crate::data::catalog_check::INDEX_LOCK_BUDGET; + crate::data::catalog_check::check(&self.pool, &data, budget) + .await + .map_err(|e| catalog_check_error(&e, budget, "the check did not finish")) + }) + } + + fn drop_orphaned_relations( + &self, + orphaned: Vec, + ) -> BoxFuture<'_, DiagResult>> { + Box::pin(async move { + let data = data_pool(&self.pool).await?; + let budget = crate::data::catalog_check::INDEX_LOCK_BUDGET; + crate::data::catalog_check::drop_orphaned(&self.pool, &data, orphaned, budget) + .await + .map_err(|e| catalog_check_error(&e, budget, "nothing was dropped")) + }) + } +} + +/// The error `catalog-check` reports when its catalog work fails. A lock wait +/// that ran out its budget says so, with `outcome`, the effect on this run. +fn catalog_check_error(e: &sqlx::Error, budget: std::time::Duration, outcome: &str) -> DiagError { + if crate::data::lock_wait_failed(e) { + DiagError::QueryFailed(format!( + "index changes in progress kept the index catalog locked for more than {} \ + seconds, so {outcome}; run the command again: {e}", + budget.as_secs() + )) + } else { + DiagError::QueryFailed(e.to_string()) + } +} + +/// The pool that holds the data relations: the data database the catalog +/// names, as `extenddb init` records it. +/// +/// A catalog without the setting, or with an empty one, is refused: the data +/// relations could then only be looked for in the catalog's own database, a +/// layout `init` does not create, where another deployment's relations can +/// sit unnamed by this catalog. +/// +/// Statement logging is off, so a drop that waits out its lock timeout does +/// not print a slow-statement line in the middle of the command's report. +async fn data_pool(catalog: &PgPool) -> DiagResult { + let conn: Option<(String,)> = + sqlx::query_as("SELECT value FROM settings WHERE key = 'data_database_connection_string'") + .fetch_optional(catalog) + .await + .map_err(|e| DiagError::QueryFailed(e.to_string()))?; + let Some((conn,)) = conn.filter(|(conn,)| !conn.is_empty()) else { + return Err(DiagError::NotConfigured( + "No data database connection string in settings. Run `extenddb init`.".to_owned(), + )); + }; + let options = conn + .parse::() + .map_err(|e| DiagError::ConnectionFailed(e.to_string()))?; + sqlx::postgres::PgPoolOptions::new() + .max_connections(2) + .connect_with(sqlx::ConnectOptions::disable_statement_logging(options)) + .await + .map_err(|e| DiagError::ConnectionFailed(e.to_string())) } // ── MetricsStore ─────────────────────────────────────────────────────── diff --git a/crates/storage-postgres/src/data/catalog_check.rs b/crates/storage-postgres/src/data/catalog_check.rs new file mode 100644 index 000000000..165842f36 --- /dev/null +++ b/crates/storage-postgres/src/data/catalog_check.rs @@ -0,0 +1,499 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Data relations checked against the catalog, for `extenddb catalog-check`. +//! +//! Every data relation this backend creates for a table is named from a +//! catalog id by [`table_storage_relations`], the list `DeleteTable` drops, +//! and every such name starts with [`STORAGE_RELATION_PREFIX`]. The check +//! lists the relations with that prefix in the data database and compares +//! them with the names the catalog's ids give. +//! +//! # Why a relation unnamed after the index writers finish stays unnamed +//! +//! Every id that names a relation is a new UUID generated just before its +//! catalog row is inserted (`create_table.rs` for the table, its GSIs, LSIs +//! and vector indexes, which restore uses too; `update_table.rs` for a GSI +//! or vector index added later). No path inserts a row for an id it did not +//! just generate, so a row inserted later cannot name a relation that exists +//! already. The recovery paths (`gsi_build.rs`, `vector_index.rs`) only +//! create a relation for a row that has committed. +//! +//! Every path creates the row first. All but one also commit it before they +//! create the relation: `CreateTable` and a vector index add. The exception +//! is a GSI added by `UpdateTable`, which inserts its row, creates the +//! relation and commits the relation, and only then commits the row. Its +//! transaction holds `ROW EXCLUSIVE` on `indexes` from the insert to its end, +//! so a `SHARE` lock on `indexes` is granted only once every such add whose +//! relation exists has committed or rolled back, and a catalog read taken +//! after the grant sees the row of each one that committed. A relation that +//! this read does not name is therefore named by no row, now or later. +//! +//! # The reads, and what they leave out while a server is running +//! +//! [`check`] lists the relations and reads the catalog. When a listed +//! relation is unnamed, it waits for the `indexes` writers in flight and +//! reads the catalog again ([`read_after_index_writers`]); a check that +//! finds every relation named takes no lock. Then it lists the relations +//! again and reads the catalog a last time. +//! +//! - Orphaned: in the first listing, unnamed by the read after the wait, and +//! still in the second listing. A relation whose catalog row a server has +//! just deleted and whose drop has not yet committed when the second +//! listing runs is reported: a synchronous `DeleteTable` between its +//! catalog commit and its data commit, an `UpdateTable` that deletes a GSI +//! or vector index before its drop runs, and a GSI whose build failed, for +//! the second (`GRACE_PUBLISH`) between the removal of its row and its +//! drop. +//! - Missing: named by a row that is expected in both catalog reads, and +//! absent from the second listing, which runs between them. A row is not +//! expected while it or its table is `CREATING` or `DELETING`, which +//! covers a catalog-first path between its row's commit and its relations, +//! and a drop that runs before the row goes. A synchronous `DeleteTable` +//! commits the row's removal before its drop, so the second read no longer +//! expects the relations a listing misses. What is still reported is a +//! table row that is `ACTIVE` before its relations commit: a `CreateTable` +//! when `control_plane_delay_seconds` is 0, which inserts the row +//! `ACTIVE`, and one whose relations take longer to commit than that +//! delay (0.25 s by default). +//! +//! # The drop +//! +//! [`drop_orphaned`] waits for the `indexes` writers the same way, reads the +//! catalog under the lock and releases it. Each reported relation that the +//! read does not name is then dropped in a transaction of its own with a +//! lock timeout of [`DROP_LOCK_TIMEOUT`], so a session holding the relation +//! costs one failed line, and nothing a server does waits on the command +//! while it drops. +//! +//! `migrate`'s advisory lock would not help: servers do not hold it, so it +//! does not exclude a running `CreateTable` or `UpdateTable`. + +use std::collections::{BTreeMap, BTreeSet}; +use std::time::{Duration, Instant}; + +use extenddb_storage::diagnostics::{CatalogRelation, OrphanDrop, StorageRelationCheck}; +use sqlx::{PgConnection, PgPool, Postgres, Transaction}; + +use super::{ + STORAGE_RELATION_PREFIX, data_table_name, index_table_name, lock_not_available, + table_storage_relations, vector_table_name, +}; + +/// How long `catalog-check` waits, in all, for the index changes in flight +/// to finish, before the check or the drop gives up. +pub(crate) const INDEX_LOCK_BUDGET: Duration = Duration::from_secs(30); + +/// The lock timeout of one attempt at the `SHARE` lock on `indexes`. +/// +/// While the request waits, every new write to `indexes` (a `CreateTable` +/// with indexes, a GSI add, a status change, the cascade of a +/// `DeleteTable`) queues behind it, so this, plus the one catalog read taken +/// once the lock is granted, is the longest a server write waits on the +/// command. A tenth of PostgreSQL's default `deadlock_timeout` (1 s), and +/// well under the default `control_plane_delay_seconds` (0.25 s) that a +/// `CreateTable` held this long eats into. The command holds no other lock +/// while it waits, so it cannot be part of a deadlock. +const INDEX_LOCK_ATTEMPT: Duration = Duration::from_millis(100); + +/// The pause between two attempts, in which the writers that queued behind +/// the last one run. +const INDEX_LOCK_PAUSE: Duration = Duration::from_millis(50); + +/// The lock timeout of each drop: the bound the server's own synchronous +/// drops use ([`crate::delete_table::SYNC_DROP_LOCK_TIMEOUT`], 1 s). +pub(crate) const DROP_LOCK_TIMEOUT: Duration = crate::delete_table::SYNC_DROP_LOCK_TIMEOUT; + +/// SQLSTATE `undefined_table`: the relation to drop no longer exists. +const SQLSTATE_UNDEFINED_TABLE: &str = "42P01"; + +/// Every catalog row that names data relations, with its table's row. One +/// statement, so the three tables are read from one snapshot. +const CATALOG_ROWS: &str = "\ + SELECT 'table' AS kind, t.table_id, NULL::text AS index_id, NULL::text AS index_name, \ + NULL::text AS index_status, t.account_id, t.table_name, t.table_status \ + FROM tables t \ + UNION ALL \ + SELECT 'index', i.table_id, i.index_id, i.index_name, i.index_status, \ + t.account_id, t.table_name, t.table_status \ + FROM indexes i LEFT JOIN tables t ON t.table_id = i.table_id \ + UNION ALL \ + SELECT 'vector', v.table_id, v.index_id, v.index_name, v.index_status, \ + t.account_id, t.table_name, t.table_status \ + FROM vector_indexes v LEFT JOIN tables t ON t.table_id = v.table_id"; + +type CatalogRow = ( + String, + String, + Option, + Option, + Option, + Option, + Option, + Option, +); + +/// A relation name the catalog gives, and the row that gives it. +struct Named { + owner: String, + /// Whether the relation should exist: the row and its table are present + /// and neither is being created or deleted. + expected: bool, +} + +/// One table's catalog rows, keyed by the relation each one names. +#[derive(Default)] +struct TableRows { + index_ids: Vec, + vector_index_ids: Vec, + owners: BTreeMap, +} + +/// A relation name as `pg_tables` lists it, without the quotes the naming +/// functions put around it. +fn bare(quoted: &str) -> String { + quoted.trim_matches('"').to_owned() +} + +/// Whether a row in `status` has its relations: it exists and is neither +/// being created (a catalog-first path commits the row before the relations) +/// nor deleted (a drop may already have run). +fn settled(status: Option<&str>) -> bool { + status.is_some_and(|s| s != "CREATING" && s != "DELETING") +} + +/// Every relation name the catalog gives, unquoted, for rows in any status. +async fn named_relations( + catalog: &mut PgConnection, +) -> Result, sqlx::Error> { + let rows: Vec = sqlx::query_as(CATALOG_ROWS).fetch_all(catalog).await?; + let mut tables: BTreeMap = BTreeMap::new(); + for (kind, table_id, index_id, index_name, index_status, account, name, status) in rows { + let table = match (account, name) { + (Some(account), Some(name)) => format!("table {account}/{name}"), + _ => format!("table id {table_id}"), + }; + let table_settled = settled(status.as_deref()); + let entry = tables.entry(table_id.clone()).or_default(); + match (kind.as_str(), index_id) { + ("index" | "vector", Some(index_id)) => { + let vector = kind == "vector"; + let relation = if vector { + entry.vector_index_ids.push(index_id.clone()); + vector_table_name(&index_id) + } else { + entry.index_ids.push(index_id.clone()); + index_table_name(&index_id) + }; + let index = index_name.unwrap_or(index_id); + let noun = if vector { "vector index" } else { "index" }; + entry.owners.insert( + bare(&relation), + Named { + owner: format!("{noun} {index} of {table}"), + expected: table_settled && settled(index_status.as_deref()), + }, + ); + } + ("table", _) => { + entry.owners.insert( + bare(&data_table_name(&table_id)), + Named { + owner: table, + expected: table_settled, + }, + ); + } + // `index_id` is NOT NULL in both index tables. + _ => {} + } + } + + // The names come from the list DeleteTable drops, so a relation family + // added there is covered here too. The owners above only label them. + let mut named = BTreeMap::new(); + for (table_id, mut entry) in tables { + for relation in + table_storage_relations(&table_id, &entry.index_ids, &entry.vector_index_ids) + { + let relation = bare(&relation); + let owner = entry.owners.remove(&relation).unwrap_or_else(|| Named { + owner: format!("table id {table_id}"), + expected: false, + }); + named.insert(relation, owner); + } + } + Ok(named) +} + +/// A catalog transaction at READ COMMITTED whose lock waits are bounded by +/// `lock_timeout`. +/// +/// Both settings are made before any statement that takes a snapshot, so +/// every later statement reads the data committed when it starts, whatever +/// `default_transaction_isolation` the database, the role or the connection +/// sets. Under REPEATABLE READ a snapshot taken before the lock wait would +/// hide the rows committed during it. +async fn begin_read_committed( + catalog: &PgPool, + lock_timeout: Duration, +) -> Result, sqlx::Error> { + let mut tx = catalog.begin().await?; + sqlx::query("SET TRANSACTION ISOLATION LEVEL READ COMMITTED") + .execute(&mut *tx) + .await?; + // A number of milliseconds, at least 1 (0 would disable the timeout). + let ms = lock_timeout.as_millis().max(1); + sqlx::query(&format!("SET LOCAL lock_timeout = '{ms}ms'")) + .execute(&mut *tx) + .await?; + Ok(tx) +} + +/// Wait until every transaction writing `indexes` when this is called has +/// ended, then read the catalog. See the module documentation for why a +/// relation this read does not name is named by no row later. +/// +/// The `SHARE` lock is requested in attempts of [`INDEX_LOCK_ATTEMPT`], +/// [`INDEX_LOCK_PAUSE`] apart, until `deadline`; the last attempt's error +/// (SQLSTATE 55P03) is returned once it has passed. The lock is released as +/// soon as the read returns. +async fn read_after_index_writers( + catalog: &PgPool, + deadline: Instant, +) -> Result, sqlx::Error> { + loop { + let left = deadline.saturating_duration_since(Instant::now()); + let mut hold = begin_read_committed(catalog, left.min(INDEX_LOCK_ATTEMPT)).await?; + let read = async { + sqlx::query("LOCK TABLE indexes IN SHARE MODE") + .execute(&mut *hold) + .await?; + named_relations(&mut hold).await + } + .await; + // The transaction only locked and read; ending it releases the lock. + hold.rollback().await?; + match read { + Err(e) if lock_not_available(&e) && Instant::now() < deadline => { + let left = deadline.saturating_duration_since(Instant::now()); + tokio::time::sleep(left.min(INDEX_LOCK_PAUSE)).await; + } + read => return read, + } + } +} + +/// Read the catalog with lock waits bounded by what is left until `deadline`. +async fn read_catalog( + catalog: &PgPool, + deadline: Instant, +) -> Result, sqlx::Error> { + let left = deadline.saturating_duration_since(Instant::now()); + let mut tx = begin_read_committed(catalog, left).await?; + let named = named_relations(&mut tx).await; + tx.rollback().await?; + named +} + +/// The data relations this backend created that the data database holds now. +async fn storage_relations(data: &PgPool) -> Result, sqlx::Error> { + let rows: Vec<(String,)> = sqlx::query_as( + "SELECT tablename::text FROM pg_tables \ + WHERE schemaname = current_schema() AND starts_with(tablename::text, $1)", + ) + .bind(STORAGE_RELATION_PREFIX) + .fetch_all(data) + .await?; + Ok(rows.into_iter().map(|(name,)| name).collect()) +} + +/// Compare the data database's relations with the catalog, waiting at most +/// `budget` for the index changes in flight. +pub(crate) async fn check( + catalog: &PgPool, + data: &PgPool, + budget: Duration, +) -> Result { + let deadline = Instant::now() + budget; + let listed = storage_relations(data).await?; + let mut named = read_catalog(catalog, deadline).await?; + // A listed relation no row names may be a GSI add's whose row has not + // committed. Only then is the lock taken, so a check that finds every + // relation named holds up no write to `indexes`. + if listed.iter().any(|relation| !named.contains_key(relation)) { + named = read_after_index_writers(catalog, deadline).await?; + } + #[cfg(feature = "test-hooks")] + test_pause::reached(test_pause::CatalogCheckPoint::AfterCatalogRead).await; + let listed_again = storage_relations(data).await?; + let named_again = read_catalog(catalog, deadline).await?; + + let orphaned = listed + .into_iter() + .filter(|relation| listed_again.contains(relation) && !named.contains_key(relation)) + .collect(); + let missing = named + .into_iter() + .filter(|(relation, row)| { + row.expected + && !listed_again.contains(relation) + && named_again + .get(relation) + .is_some_and(|again| again.expected) + }) + .map(|(relation, row)| CatalogRelation { + relation, + owner: row.owner, + }) + .collect(); + Ok(StorageRelationCheck { orphaned, missing }) +} + +/// Drop the relations in `orphaned` that no catalog row names once the index +/// changes in flight have finished, waiting at most `budget` for those. +/// +/// Fails only when the catalog cannot be locked or read; nothing is dropped +/// then. A relation that cannot be dropped gets [`OrphanDrop::Failed`] and +/// the rest still run. +pub(crate) async fn drop_orphaned( + catalog: &PgPool, + data: &PgPool, + orphaned: Vec, + budget: Duration, +) -> Result, sqlx::Error> { + if orphaned.is_empty() { + return Ok(Vec::new()); + } + let named = read_after_index_writers(catalog, Instant::now() + budget).await?; + + let mut outcomes = Vec::with_capacity(orphaned.len()); + for relation in orphaned { + let outcome = if !relation.starts_with(STORAGE_RELATION_PREFIX) { + OrphanDrop::Kept("not a data relation of this backend".to_owned()) + } else if let Some(row) = named.get(&relation) { + OrphanDrop::Kept(format!("the catalog names it: {}", row.owner)) + } else { + drop_relation(data, &relation).await + }; + outcomes.push((relation, outcome)); + } + Ok(outcomes) +} + +/// Drop one relation in a transaction of its own, waiting at most +/// [`DROP_LOCK_TIMEOUT`] for its lock. +/// +/// Without `IF EXISTS`, so a relation another session dropped first is told +/// apart from one dropped here, and PostgreSQL sends no notice to print. +async fn drop_relation(data: &PgPool, relation: &str) -> OrphanDrop { + let dropped = async { + let mut tx = data.begin().await?; + let ms = DROP_LOCK_TIMEOUT.as_millis(); + sqlx::query(&format!("SET LOCAL lock_timeout = '{ms}ms'")) + .execute(&mut *tx) + .await?; + let quoted = format!("\"{}\"", relation.replace('"', "\"\"")); + sqlx::query(&format!("DROP TABLE {quoted}")) + .execute(&mut *tx) + .await?; + tx.commit().await + } + .await; + match dropped { + Ok(()) => OrphanDrop::Dropped, + Err(e) if sqlstate(&e).as_deref() == Some(SQLSTATE_UNDEFINED_TABLE) => { + OrphanDrop::AlreadyGone + } + Err(e) if lock_not_available(&e) => OrphanDrop::Failed(format!( + "another session held a lock on it for more than {} ms, so it was kept; run the \ + command again: {e}", + DROP_LOCK_TIMEOUT.as_millis() + )), + Err(e) => OrphanDrop::Failed(e.to_string()), + } +} + +fn sqlstate(e: &sqlx::Error) -> Option { + e.as_database_error() + .and_then(sqlx::error::DatabaseError::code) + .map(std::borrow::Cow::into_owned) +} + +/// Test hooks: hold the next check at a chosen point, so a test can change +/// the catalog or the data database in the window the point opens. +/// Compiled only with `test-hooks`. +#[cfg(feature = "test-hooks")] +pub(crate) mod test_pause { + use extenddb_storage::test_pause::{Pause, PausePoint}; + + /// Where a check can be held. + #[doc(hidden)] + #[derive(Debug, Clone, Copy, PartialEq, Eq)] + pub enum CatalogCheckPoint { + /// After the catalog read taken once the index writers have finished, + /// before the second listing and the second catalog read. + AfterCatalogRead, + } + + static PAUSE: PausePoint = PausePoint::new(); + + /// A pause armed by [`arm_catalog_check_pause`]. Dropping it disarms the + /// pause and releases a check waiting at it. + #[doc(hidden)] + pub type CatalogCheckPause = Pause; + + /// Arm a pause for the next check in this process that reaches `point` + /// (one check, once). + /// + /// # Panics + /// + /// In debug builds, when a pause is already armed ([`PausePoint::arm`]). + #[doc(hidden)] + #[must_use] + pub fn arm_catalog_check_pause(point: CatalogCheckPoint) -> CatalogCheckPause { + PAUSE.arm(point) + } + + pub(crate) async fn reached(point: CatalogCheckPoint) { + PAUSE.reached(point).await; + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn every_storage_relation_name_carries_the_listing_prefix() { + let names = table_storage_relations( + "8f2b5c1e-0000-4000-8000-000000000001", + &["8f2b5c1e-0000-4000-8000-000000000002".to_owned()], + &["8f2b5c1e-0000-4000-8000-000000000003".to_owned()], + ); + assert_eq!(names.len(), 3); + for name in names { + assert!( + bare(&name).starts_with(STORAGE_RELATION_PREFIX), + "{name} would be invisible to catalog-check" + ); + } + } + + #[test] + fn bare_removes_the_identifier_quotes() { + assert_eq!(bare(&data_table_name("abc")), "_ddb_abc"); + assert_eq!(bare(&vector_table_name("abc")), "_ddb_vec_abc"); + } + + #[test] + fn rows_being_created_or_deleted_are_not_settled() { + for status in ["ACTIVE", "UPDATING"] { + assert!(settled(Some(status)), "{status}"); + } + for status in [None, Some("CREATING"), Some("DELETING")] { + assert!(!settled(status), "{status:?}"); + } + } +} diff --git a/crates/storage-postgres/src/data/mod.rs b/crates/storage-postgres/src/data/mod.rs index d8056ea14..7c79a8900 100755 --- a/crates/storage-postgres/src/data/mod.rs +++ b/crates/storage-postgres/src/data/mod.rs @@ -3,28 +3,33 @@ //! Per-DynamoDB-table DDL and item CRUD for the `PostgreSQL` backend. //! -//! Each Virtual `DynamoDB` table maps to a `PostgreSQL` table named `_ddb_`. +//! Each Virtual `DynamoDB` table maps to a `PostgreSQL` table named `_ddb_`. //! Partition keys are stored as TEXT. Sort keys use typed columns (`sk_s`, `sk_n`, `sk_b`) //! for correct ordering. The full item is stored as JSONB in `item_data`. use extenddb_core::types::{AttributeDefinition, Item, KeySchemaElement, ScalarAttributeType}; use extenddb_storage::error::StorageError; +/// Prefix of every data relation this backend creates for a table: the base +/// table, its secondary indexes and its vector indexes. It keeps them apart +/// from the catalog's tables and the data database's queue tables, and it is +/// how `extenddb catalog-check` finds them. +pub(crate) const STORAGE_RELATION_PREFIX: &str = "_ddb_"; + /// SQL table name for a Virtual `DynamoDB` table. /// -/// Uses `_ddb_` prefix to avoid collisions with catalog metadata tables. -/// Includes `account_id` for multi-account isolation (Phase 12a). -/// Table names are validated at the engine layer (alphanumeric + `_.-`), -/// so this is safe for identifier construction. +/// Uses the `_ddb_` prefix to avoid collisions with catalog metadata tables. +/// Named from the catalog's `table_id`, a server-generated UUID, so no client +/// input reaches this identifier. #[must_use] pub fn data_table_name(table_id: &str) -> String { - format!("\"_ddb_{table_id}\"") + format!("\"{STORAGE_RELATION_PREFIX}{table_id}\"") } /// SQL table name for a GSI/LSI data table. #[must_use] pub fn index_table_name(index_id: &str) -> String { - format!("\"_ddb_{index_id}\"") + format!("\"{STORAGE_RELATION_PREFIX}{index_id}\"") } /// SQL table name for a vector index's data table. @@ -42,7 +47,7 @@ pub fn index_table_name(index_id: &str) -> String { /// The id is a server-generated UUID, so no client input reaches this identifier. #[must_use] pub fn vector_table_name(index_id: &str) -> String { - format!("\"_ddb_vec_{index_id}\"") + format!("\"{STORAGE_RELATION_PREFIX}vec_{index_id}\"") } /// Look up all RANGE key attribute definitions from the key schema (preserving order). @@ -132,6 +137,7 @@ macro_rules! bind_sk_execute { } // Submodules declared after macros so they can use bind_sk_fetch_optional/bind_sk_execute. +pub(crate) mod catalog_check; mod data_engine; mod ddl; mod delete_item; diff --git a/crates/storage-postgres/src/lib.rs b/crates/storage-postgres/src/lib.rs index 81480973c..4569cdf1c 100755 --- a/crates/storage-postgres/src/lib.rs +++ b/crates/storage-postgres/src/lib.rs @@ -44,6 +44,13 @@ pub use catalog_store::PostgresCatalogStore; pub use config::PostgresStorageConfig; pub use config::parse_connection_string; pub use credential_store::DbCredentialStore; +/// Hold a `catalog-check` comparison between its first catalog read and its +/// second listing of the data relations. +#[cfg(feature = "test-hooks")] +#[doc(hidden)] +pub use data::catalog_check::test_pause::{ + CatalogCheckPause, CatalogCheckPoint, arm_catalog_check_pause, +}; /// Finish GSI builds left `CREATING`, once. The server runs it from a background /// worker; exported for its test. pub use data::gsi_build::reconcile_creating_gsis; diff --git a/crates/storage-postgres/tests/catalog_check/fix.rs b/crates/storage-postgres/tests/catalog_check/fix.rs new file mode 100644 index 000000000..162486092 --- /dev/null +++ b/crates/storage-postgres/tests/catalog_check/fix.rs @@ -0,0 +1,360 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! What the drop drops and keeps, and what it waits for. + +use std::time::{Duration, Instant}; + +use extenddb_storage::diagnostics::{DiagnosticsStore, OrphanDrop}; +use extenddb_storage_postgres::PostgresCatalogStore; +use sqlx::{Connection, PgPool}; + +use crate::support::{ACCOUNT, Deployment, TABLE, base_conn, hold_relation, skip}; + +/// Wait until a session of the catalog database waits for `SHARE` on +/// `indexes`, and return the catalog tables among `tables` and +/// `vector_indexes` that this session holds a lock on while it waits. One +/// read of `pg_locks` gives both, and `SHARE` is never a fast-path lock, so +/// the two come from the same state of the lock table. +async fn other_tables_locked_by_the_index_lock_waiter(catalog: &PgPool) -> Vec { + let deadline = Instant::now() + Duration::from_secs(10); + loop { + let waiters: Vec<(Option>,)> = sqlx::query_as( + "SELECT array_agg(c.relname::text ORDER BY c.relname) \ + FILTER (WHERE l.granted AND c.relname IN ('tables', 'vector_indexes')) \ + FROM pg_locks l JOIN pg_class c ON c.oid = l.relation \ + WHERE l.database = (SELECT oid FROM pg_database WHERE datname = current_database()) \ + GROUP BY l.pid \ + HAVING bool_or(c.relname = 'indexes' AND l.mode = 'ShareLock' AND NOT l.granted)", + ) + .fetch_all(catalog) + .await + .expect("read pg_locks"); + if let Some((held,)) = waiters.into_iter().next() { + return held.unwrap_or_default(); + } + assert!( + Instant::now() < deadline, + "no session waited for ShareLock on indexes within 10 s" + ); + tokio::time::sleep(Duration::from_millis(5)).await; + } +} + +/// Wait until a session of the catalog database started waiting for `SHARE` +/// on `indexes` less than 20 ms ago, so that a commit sent at once reaches +/// the server while that attempt still waits, not in the pause between two +/// attempts, when the next attempt would start after the commit. +async fn wait_for_a_fresh_index_lock_wait(catalog: &PgPool) { + let deadline = Instant::now() + Duration::from_secs(10); + loop { + let fresh: bool = sqlx::query_scalar( + "SELECT EXISTS (SELECT 1 FROM pg_locks l JOIN pg_class c ON c.oid = l.relation \ + WHERE l.database = (SELECT oid FROM pg_database WHERE datname = current_database()) \ + AND c.relname = 'indexes' AND l.mode = 'ShareLock' AND NOT l.granted \ + AND l.waitstart > clock_timestamp() - interval '20 milliseconds')", + ) + .fetch_one(catalog) + .await + .expect("read pg_locks"); + if fresh { + return; + } + assert!( + Instant::now() < deadline, + "no session started waiting for ShareLock on indexes within 10 s" + ); + tokio::time::sleep(Duration::from_millis(1)).await; + } +} + +/// `--fix` drops the relation no catalog row names and nothing else: not the +/// live table's relations, not a table created after the check, not a table +/// without the backend's prefix. The live data is intact afterwards. +#[tokio::test] +async fn fix_drops_only_the_relation_no_catalog_row_names() { + let test = "fix_drops_only_the_relation_no_catalog_row_names"; + if base_conn().is_none() { + return skip(test); + } + let d = Deployment::new().await; + let key_info = d.create_table_with_indexes(TABLE).await; + let live = d.storage_relations().await; + let orphan = d.plant_orphan().await; + sqlx::query("CREATE TABLE operator_notes (note TEXT)") + .execute(&d.data) + .await + .expect("an unrelated table"); + + let check = d + .diagnostics + .check_storage_relations() + .await + .expect("check"); + assert_eq!(check.orphaned, vec![orphan.clone()]); + assert!(check.missing.is_empty(), "missing: {:?}", check.missing); + + // A CreateTable that lands between the check and the drop: its relations + // were not in the listing, and a stale candidate list naming them must + // not get them dropped. + d.create_table_with_indexes("later").await; + let later: Vec = d + .storage_relations() + .await + .difference(&live) + .filter(|r| **r != orphan) + .cloned() + .collect(); + assert_eq!(later.len(), if d.pgvector().await { 4 } else { 3 }); + let mut candidates = vec![orphan.clone(), "operator_notes".to_owned()]; + candidates.extend(later.iter().cloned()); + + let outcomes = d + .diagnostics + .drop_orphaned_relations(candidates.clone()) + .await + .expect("drop"); + let names: Vec<&String> = outcomes.iter().map(|(n, _)| n).collect(); + assert_eq!(names, candidates.iter().collect::>()); + assert_eq!(outcomes[0].1, OrphanDrop::Dropped, "{outcomes:?}"); + assert_eq!( + outcomes[1].1, + OrphanDrop::Kept("not a data relation of this backend".to_owned()), + "{outcomes:?}" + ); + for (name, outcome) in &outcomes[2..] { + match outcome { + OrphanDrop::Kept(reason) => assert!( + reason.starts_with("the catalog names it: ") + && reason.ends_with(&format!("table {ACCOUNT}/later")), + "{name}: {reason}" + ), + other => panic!("{name}: {other:?}"), + } + } + + assert!(!d.relation_exists(&orphan).await, "the orphan was dropped"); + assert!(d.relation_exists("operator_notes").await); + let after = d.storage_relations().await; + assert!(live.is_subset(&after), "live relations survived: {after:?}"); + assert!(later.iter().all(|r| after.contains(r)), "{after:?}"); + d.assert_data_intact(&key_info).await; + + let check = d + .diagnostics + .check_storage_relations() + .await + .expect("check again"); + assert!(check.orphaned.is_empty(), "orphaned: {:?}", check.orphaned); + assert!(check.missing.is_empty(), "missing: {:?}", check.missing); +} + +/// An `UpdateTable` adding a GSI has inserted the index row and committed +/// the index's relation, and its catalog transaction is still open when the +/// drop is asked to drop that relation. The drop waits for the add, holding +/// no lock on `tables` or `vector_indexes`; the add's next write, to the +/// table row, does not wait on the drop; another write to `indexes` queues +/// behind the drop's request for one attempt at most; and once the add +/// commits, its relation is kept. Run at both the default isolation and +/// REPEATABLE READ. +async fn an_index_added_during_the_drop_keeps_its_relation( + store: &PostgresCatalogStore, + d: &Deployment, +) { + d.create_table_with_indexes(TABLE).await; + let mut add = d.begin_index_add("gsi9").await; + let relation = add.relation.clone(); + let five = Duration::from_secs(5); + let (outcomes, (held_while_waiting, table_write, other_write)) = tokio::join!( + store.drop_orphaned_relations(vec![relation.clone()]), + async move { + let held = other_tables_locked_by_the_index_lock_waiter(&d.catalog).await; + let table_write = tokio::time::timeout(five, add.write_table_row()).await; + let started = Instant::now(); + let other_write = tokio::time::timeout( + five, + sqlx::query( + "UPDATE indexes SET index_status = index_status WHERE index_name = 'gsi1'", + ) + .execute(&d.catalog), + ) + .await + .map(|r| r.map(|_| started.elapsed())); + // Open past several attempts, so the drop has to try again, and + // committed while an attempt waits: the read that attempt takes + // once it is granted must see the row that committed during the + // wait, whatever isolation the session defaults to. + tokio::time::sleep(Duration::from_millis(600)).await; + wait_for_a_fresh_index_lock_wait(&d.catalog).await; + add.commit().await; + (held, table_write, other_write) + } + ); + + assert!( + held_while_waiting.is_empty(), + "the drop held locks on {held_while_waiting:?} while it waited for indexes" + ); + let table_write = table_write + .expect("the add's table write did not wait out 5 s") + .expect("the add's table write succeeded"); + assert!( + table_write < Duration::from_millis(900), + "the add's table write waited {table_write:?}, as long as a deadlock check" + ); + let other_write = other_write + .expect("the other index write did not wait out 5 s") + .expect("the other index write succeeded"); + assert!( + other_write < Duration::from_secs(1), + "another index write waited {other_write:?} behind the drop" + ); + assert_eq!( + outcomes.expect("drop"), + vec![( + relation.clone(), + OrphanDrop::Kept(format!( + "the catalog names it: index gsi9 of table {ACCOUNT}/{TABLE}" + )) + )] + ); + assert!(d.relation_exists(&relation).await, "{relation} kept"); +} + +#[tokio::test] +async fn an_index_added_during_the_drop_keeps_its_relation_at_the_default_isolation() { + let test = "an_index_added_during_the_drop_keeps_its_relation_at_the_default_isolation"; + if base_conn().is_none() { + return skip(test); + } + let d = Deployment::new().await; + an_index_added_during_the_drop_keeps_its_relation(&d.diagnostics, &d).await; +} + +#[tokio::test] +async fn an_index_added_during_the_drop_keeps_its_relation_under_repeatable_read() { + let test = "an_index_added_during_the_drop_keeps_its_relation_under_repeatable_read"; + if base_conn().is_none() { + return skip(test); + } + let d = Deployment::new().await; + let store = d.diagnostics_at("repeatable read").await; + an_index_added_during_the_drop_keeps_its_relation(&store, &d).await; +} + +/// The same GSI add in flight while the check runs: its relation exists and +/// its row has not committed, so the check waits for the add before it reads +/// the catalog, and the relation is not reported orphaned. +#[tokio::test] +async fn an_index_added_during_the_check_is_not_reported_orphaned() { + let test = "an_index_added_during_the_check_is_not_reported_orphaned"; + if base_conn().is_none() { + return skip(test); + } + let d = Deployment::new().await; + d.create_table_with_indexes(TABLE).await; + let mut add = d.begin_index_add("gsi9").await; + let d = &d; + let (check, table_write) = tokio::join!(d.diagnostics.check_storage_relations(), async move { + Deployment::wait_for_lock_request(&d.catalog, "indexes", "ShareLock").await; + let table_write = add.write_table_row().await; + tokio::time::sleep(Duration::from_millis(300)).await; + add.commit().await; + table_write + }); + let table_write = table_write.expect("the add's table write succeeded"); + assert!(table_write < Duration::from_millis(900), "{table_write:?}"); + let check = check.expect("check"); + assert!(check.orphaned.is_empty(), "orphaned: {:?}", check.orphaned); + assert!(check.missing.is_empty(), "missing: {:?}", check.missing); +} + +/// A session holds the orphan the drop is asked to drop. The drop gives up +/// after its own lock timeout and reports the relation failed, and a +/// `CreateTable` sent while it waits completes without waiting on it. +#[tokio::test] +async fn a_held_orphan_fails_within_its_lock_timeout_and_catalog_writes_go_on() { + let test = "a_held_orphan_fails_within_its_lock_timeout_and_catalog_writes_go_on"; + if base_conn().is_none() { + return skip(test); + } + let d = Deployment::new().await; + d.create_table_with_indexes(TABLE).await; + let orphan = d.plant_orphan().await; + let holder = hold_relation(&d, &orphan).await; + + let d = &d; + let started = Instant::now(); + let ((outcomes, dropped_at), ()) = tokio::join!( + async { + // The holder is released only after the drop returns, so a drop + // without a lock timeout of its own would wait here for good. + let outcomes = tokio::time::timeout( + Duration::from_secs(10), + d.diagnostics.drop_orphaned_relations(vec![orphan.clone()]), + ) + .await + .expect("the drop gave up on the held orphan within 10 s"); + (outcomes, started.elapsed()) + }, + async { + Deployment::wait_for_lock_request(&d.data, &orphan, "AccessExclusiveLock").await; + // Before the change, the drop held SHARE on `tables` and `indexes` + // until it finished, so this waited for the whole lock budget. + tokio::time::timeout( + Duration::from_secs(5), + d.create_table_with_indexes("meanwhile"), + ) + .await + .expect("the CreateTable did not wait out 5 s"); + } + ); + assert!( + dropped_at < Duration::from_secs(5), + "the drop took {dropped_at:?}" + ); + let outcomes = outcomes.expect("drop"); + assert_eq!(outcomes.len(), 1, "{outcomes:?}"); + match &outcomes[0] { + (name, OrphanDrop::Failed(reason)) if *name == orphan => assert!( + reason.starts_with("another session held a lock on it for more than 1000 ms"), + "{reason}" + ), + other => panic!("{other:?}"), + } + assert!(d.relation_exists(&orphan).await, "the held orphan was kept"); + + holder.close().await.expect("release the orphan"); + let outcomes = d + .diagnostics + .drop_orphaned_relations(vec![orphan.clone()]) + .await + .expect("drop again"); + assert_eq!(outcomes, vec![(orphan, OrphanDrop::Dropped)]); +} + +/// A relation another session dropped after the check reported it, as a +/// server's `DeleteTable` drops one, is reported as already gone, not as +/// dropped by the command. +#[tokio::test] +async fn a_relation_dropped_first_by_another_session_is_already_gone() { + let test = "a_relation_dropped_first_by_another_session_is_already_gone"; + if base_conn().is_none() { + return skip(test); + } + let d = Deployment::new().await; + let orphan = d.plant_orphan().await; + let check = d + .diagnostics + .check_storage_relations() + .await + .expect("check"); + assert_eq!(check.orphaned, vec![orphan.clone()]); + d.remove_relation(&orphan).await; + let outcomes = d + .diagnostics + .drop_orphaned_relations(check.orphaned) + .await + .expect("drop"); + assert_eq!(outcomes, vec![(orphan, OrphanDrop::AlreadyGone)]); +} diff --git a/crates/storage-postgres/tests/catalog_check/main.rs b/crates/storage-postgres/tests/catalog_check/main.rs new file mode 100644 index 000000000..4eb49d9ab --- /dev/null +++ b/crates/storage-postgres/tests/catalog_check/main.rs @@ -0,0 +1,27 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! `extenddb catalog-check` against the data relations the engine creates. +//! +//! The check compares the relations in the data database with the names the +//! catalog's ids give through the backend's own naming functions. These tests +//! build a deployment the way `init` leaves it (catalog and data in separate +//! databases), create a table with a GSI, an LSI and a vector index through +//! the engine, and run the check and the drop through the backend's +//! `DiagnosticsStore`, which is what the command calls. +//! +//! - `report`: what the check reports, for rows in every status, for a +//! catalog without the data database setting, and for changes that land +//! while it runs. +//! - `fix`: what the drop drops and keeps, against an index added while it +//! waits (at both isolation levels), a relation another session holds or +//! dropped first. +//! +//! Requires `EXTENDDB_TEST_PG_CONNECTION_STRING` (a base URL +//! `postgres://user:pass@host:port`, no database); without it every test +//! reports a skip and passes. Each test creates its own databases and drops +//! them when its deployment is dropped, on a panic too. + +mod fix; +mod report; +mod support; diff --git a/crates/storage-postgres/tests/catalog_check/report.rs b/crates/storage-postgres/tests/catalog_check/report.rs new file mode 100644 index 000000000..46ba28ac4 --- /dev/null +++ b/crates/storage-postgres/tests/catalog_check/report.rs @@ -0,0 +1,352 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! What the check reports. + +use std::collections::BTreeSet; + +use extenddb_core::types::DeleteTableInput; +use extenddb_storage::TableEngine; +use extenddb_storage::diagnostics::{CatalogRelation, DiagError, DiagnosticsStore}; +use extenddb_storage_postgres::{CatalogCheckPoint, arm_catalog_check_pause}; +use sqlx::Connection; + +use crate::support::{ACCOUNT, Deployment, TABLE, base_conn, skip}; + +/// A table with a GSI, an LSI and a vector index has every relation the +/// engine created named by the catalog: nothing orphaned, nothing missing. +#[tokio::test] +async fn live_relations_are_neither_orphaned_nor_missing() { + let test = "live_relations_are_neither_orphaned_nor_missing"; + if base_conn().is_none() { + return skip(test); + } + let d = Deployment::new().await; + d.create_table_with_indexes(TABLE).await; + let relations = d.storage_relations().await; + let expected = if d.pgvector().await { 4 } else { 3 }; + assert_eq!(relations.len(), expected, "{relations:?}"); + + let check = d + .diagnostics + .check_storage_relations() + .await + .expect("check"); + assert!(check.orphaned.is_empty(), "orphaned: {:?}", check.orphaned); + assert!(check.missing.is_empty(), "missing: {:?}", check.missing); +} + +/// A relation the catalog names and the data database lacks is reported +/// missing, with the index and table that own it; the relation that took +/// its place is reported orphaned. +#[tokio::test] +async fn a_relation_the_catalog_names_is_reported_missing() { + let test = "a_relation_the_catalog_names_is_reported_missing"; + if base_conn().is_none() { + return skip(test); + } + let d = Deployment::new().await; + d.create_table_with_indexes(TABLE).await; + let gsi = d.index_relation(TABLE, "gsi1").await; + assert!(d.relation_exists(&gsi).await, "{gsi}"); + let elsewhere = format!("_ddb_{}", uuid::Uuid::new_v4()); + sqlx::query(&format!("ALTER TABLE \"{gsi}\" RENAME TO \"{elsewhere}\"")) + .execute(&d.data) + .await + .expect("rename the GSI relation"); + + let check = d + .diagnostics + .check_storage_relations() + .await + .expect("check"); + assert_eq!(check.orphaned, vec![elsewhere.clone()]); + assert_eq!( + check.missing, + vec![CatalogRelation { + relation: gsi.clone(), + owner: format!("index gsi1 of table {ACCOUNT}/{TABLE}"), + }] + ); + + sqlx::query(&format!("ALTER TABLE \"{elsewhere}\" RENAME TO \"{gsi}\"")) + .execute(&d.data) + .await + .expect("rename it back"); + let check = d + .diagnostics + .check_storage_relations() + .await + .expect("check again"); + assert!(check.orphaned.is_empty(), "orphaned: {:?}", check.orphaned); + assert!(check.missing.is_empty(), "missing: {:?}", check.missing); +} + +/// Rows in every status name their relations: a table in CREATING, UPDATING +/// or DELETING and an index or vector index in CREATING or DELETING keep +/// their relations out of the orphan list. +#[tokio::test] +async fn rows_in_every_status_count_as_live() { + let test = "rows_in_every_status_count_as_live"; + if base_conn().is_none() { + return skip(test); + } + let d = Deployment::new().await; + d.create_table_with_indexes(TABLE).await; + let table_status = "UPDATE tables SET table_status = $1"; + let index_status = "UPDATE indexes SET index_status = $1"; + let vector_status = "UPDATE vector_indexes SET index_status = $1"; + + for status in ["CREATING", "UPDATING", "DELETING"] { + d.set_status(table_status, status).await; + let check = d + .diagnostics + .check_storage_relations() + .await + .expect("check"); + assert!(check.orphaned.is_empty(), "{status}: {:?}", check.orphaned); + assert!(check.missing.is_empty(), "{status}: {:?}", check.missing); + } + d.set_status(table_status, "ACTIVE").await; + + for status in ["CREATING", "DELETING"] { + d.set_status(index_status, status).await; + d.set_status(vector_status, status).await; + let check = d + .diagnostics + .check_storage_relations() + .await + .expect("check"); + assert!(check.orphaned.is_empty(), "{status}: {:?}", check.orphaned); + assert!(check.missing.is_empty(), "{status}: {:?}", check.missing); + } +} + +/// A row being created or deleted is not expected to have its relations: +/// a catalog-first path commits the row before it creates them, and a drop +/// may run before the row goes. With the relations gone, a table or index in +/// CREATING or DELETING is not reported missing; back in ACTIVE, the same +/// rows are. +#[tokio::test] +async fn rows_being_created_or_deleted_are_not_reported_missing() { + let test = "rows_being_created_or_deleted_are_not_reported_missing"; + if base_conn().is_none() { + return skip(test); + } + let d = Deployment::new().await; + d.create_table_with_indexes(TABLE).await; + let table_status = "UPDATE tables SET table_status = $1"; + let index_status = "UPDATE indexes SET index_status = $1"; + let vector_status = "UPDATE vector_indexes SET index_status = $1"; + let mut index_relations = vec![ + d.index_relation(TABLE, "gsi1").await, + d.index_relation(TABLE, "lsi1").await, + ]; + if d.pgvector().await { + index_relations.push(d.vector_relation(TABLE, "vec1").await); + } + let all: BTreeSet = d.storage_relations().await; + + // The index relations go while their rows are DELETING, the way a drop + // runs before its row is removed. + d.set_status(index_status, "DELETING").await; + d.set_status(vector_status, "DELETING").await; + for relation in &index_relations { + d.remove_relation(relation).await; + } + for status in ["DELETING", "CREATING"] { + d.set_status(index_status, status).await; + d.set_status(vector_status, status).await; + let check = d + .diagnostics + .check_storage_relations() + .await + .expect("check"); + assert!( + check.missing.is_empty(), + "index {status}: {:?}", + check.missing + ); + assert!( + check.orphaned.is_empty(), + "index {status}: {:?}", + check.orphaned + ); + } + + // Then the base relation, with the table DELETING. + let base: Vec = all + .iter() + .filter(|r| !index_relations.contains(r)) + .cloned() + .collect(); + assert_eq!(base.len(), 1, "{all:?}"); + d.set_status(table_status, "DELETING").await; + d.remove_relation(&base[0]).await; + for status in ["DELETING", "CREATING"] { + d.set_status(table_status, status).await; + let check = d + .diagnostics + .check_storage_relations() + .await + .expect("check"); + assert!( + check.missing.is_empty(), + "table {status}: {:?}", + check.missing + ); + } + + // Settled again, every relation is missing. + d.set_status(table_status, "ACTIVE").await; + d.set_status(index_status, "ACTIVE").await; + d.set_status(vector_status, "ACTIVE").await; + let check = d + .diagnostics + .check_storage_relations() + .await + .expect("check"); + let missing: BTreeSet = check.missing.into_iter().map(|m| m.relation).collect(); + assert_eq!(missing, all); +} + +/// A check that finds every listed relation named takes no lock: a write to +/// `indexes` left open (a slow GSI add, a status change) does not hold it up. +#[tokio::test] +async fn a_check_that_finds_every_relation_named_does_not_wait_for_index_writers() { + let test = "a_check_that_finds_every_relation_named_does_not_wait_for_index_writers"; + if base_conn().is_none() { + return skip(test); + } + let d = Deployment::new().await; + d.create_table_with_indexes(TABLE).await; + let mut writer = sqlx::PgConnection::connect(&d.catalog_url()) + .await + .expect("a catalog session for the writer"); + sqlx::query("BEGIN") + .execute(&mut writer) + .await + .expect("begin"); + sqlx::query("UPDATE indexes SET index_status = index_status WHERE index_name = 'gsi1'") + .execute(&mut writer) + .await + .expect("an open write to indexes"); + + let started = std::time::Instant::now(); + let check = tokio::time::timeout( + std::time::Duration::from_secs(5), + d.diagnostics.check_storage_relations(), + ) + .await + .expect("the check did not wait out 5 s") + .expect("check"); + let took = started.elapsed(); + assert!(check.orphaned.is_empty(), "orphaned: {:?}", check.orphaned); + assert!(check.missing.is_empty(), "missing: {:?}", check.missing); + assert!( + took < std::time::Duration::from_secs(2), + "the check took {took:?} with an index write open" + ); + writer.close().await.expect("end the writer"); +} + +/// A catalog with no data database setting, or an empty one, is refused +/// with the message the command has always given, by the check and by the +/// drop: the relations could only be looked for in the catalog's own +/// database, a layout `init` does not create. +#[tokio::test] +async fn a_catalog_without_the_data_database_setting_is_refused() { + let test = "a_catalog_without_the_data_database_setting_is_refused"; + if base_conn().is_none() { + return skip(test); + } + let d = Deployment::new().await; + d.create_table_with_indexes(TABLE).await; + let orphan = d.plant_orphan().await; + let refusal = "No data database connection string in settings. Run `extenddb init`."; + + for change in [ + "UPDATE settings SET value = '' WHERE key = 'data_database_connection_string'", + "DELETE FROM settings WHERE key = 'data_database_connection_string'", + ] { + sqlx::query(change) + .execute(&d.catalog) + .await + .expect("change the setting"); + match d.diagnostics.check_storage_relations().await { + Err(DiagError::NotConfigured(msg)) => assert_eq!(msg, refusal, "{change}"), + other => panic!("{change}: check gave {other:?}"), + } + match d + .diagnostics + .drop_orphaned_relations(vec![orphan.clone()]) + .await + { + Err(DiagError::NotConfigured(msg)) => assert_eq!(msg, refusal, "{change}"), + other => panic!("{change}: drop gave {other:?}"), + } + assert!( + d.relation_exists(&orphan).await, + "{change}: nothing dropped" + ); + } +} + +/// Changes that land between the check's first catalog read and its second +/// listing are not reported: a table a `DeleteTable` removes (row first, +/// then relations) is not missing, a relation another session drops is not +/// orphaned, and a relation created after the first listing waits for the +/// next check. +#[tokio::test] +async fn changes_during_the_check_are_not_reported() { + let test = "changes_during_the_check_are_not_reported"; + if base_conn().is_none() { + return skip(test); + } + let d = Deployment::new().await; + d.create_table_with_indexes(TABLE).await; + let kept = d.storage_relations().await; + d.create_table_with_indexes("doomed").await; + let doomed: Vec = d + .storage_relations() + .await + .difference(&kept) + .cloned() + .collect(); + let gone = d.plant_orphan().await; + + let mut pause = arm_catalog_check_pause(CatalogCheckPoint::AfterCatalogRead); + let reached = pause.take_reached(); + let mut check = d.diagnostics.check_storage_relations(); + tokio::select! { + r = &mut check => panic!("the check finished while it should be paused: {r:?}"), + r = reached => r.expect("the check reached the pause"), + } + d.engine + .delete_table( + ACCOUNT, + DeleteTableInput { + table_name: "doomed".to_owned(), + }, + ) + .await + .expect("delete the table"); + for relation in &doomed { + assert!(!d.relation_exists(relation).await, "{relation} dropped"); + } + d.remove_relation(&gone).await; + let late = d.plant_orphan().await; + pause.resume(); + let found = check.await.expect("check"); + drop(pause); + assert!(found.missing.is_empty(), "missing: {:?}", found.missing); + assert!(found.orphaned.is_empty(), "orphaned: {:?}", found.orphaned); + + let next = d + .diagnostics + .check_storage_relations() + .await + .expect("check again"); + assert_eq!(next.orphaned, vec![late]); + assert!(next.missing.is_empty(), "missing: {:?}", next.missing); +} diff --git a/crates/storage-postgres/tests/catalog_check/support.rs b/crates/storage-postgres/tests/catalog_check/support.rs new file mode 100644 index 000000000..ca362676a --- /dev/null +++ b/crates/storage-postgres/tests/catalog_check/support.rs @@ -0,0 +1,591 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! A deployment for the `catalog_check` tests, and the catalog and data +//! database operations they act with. + +use std::collections::{BTreeMap, BTreeSet}; +use std::time::{Duration, Instant}; + +use extenddb_core::expression::ExpressionMaps; +use extenddb_core::types::{ + AttributeDefinition, AttributeValue, BillingMode, CreateTableInput, DistanceFunction, GsiInput, + Item, KeySchemaElement, KeyType, LsiInput, Projection, ProjectionType, ScalarAttributeType, + TableKeyInfo, VectorAttribute, VectorIndexSpecification, +}; +use extenddb_storage::bootstrapper::{BootstrapConfig, Bootstrapper}; +use extenddb_storage::{DataEngine, TableEngine}; +use extenddb_storage_postgres::{ + PostgresBootstrapper, PostgresCatalogStore, PostgresConfig, PostgresEngine, index_table_name, + vector_table_name, +}; +use sqlx::postgres::{PgConnectOptions, PgPoolOptions}; +use sqlx::{Connection, PgConnection, PgPool}; + +pub(crate) const ACCOUNT: &str = "123456789012"; +const REGION: &str = "us-east-1"; +pub(crate) const TABLE: &str = "orders"; + +pub(crate) fn base_conn() -> Option { + let conn = std::env::var("EXTENDDB_TEST_PG_CONNECTION_STRING").ok()?; + (!conn.trim().is_empty()).then(|| conn.trim_end_matches('/').to_owned()) +} + +pub(crate) fn skip(test: &str) { + eprintln!("SKIP {test}: EXTENDDB_TEST_PG_CONNECTION_STRING is not set."); +} + +struct Parts { + host: String, + port: u16, + user: String, + pass: String, +} + +fn parts(base: &str) -> Parts { + let rest = base + .strip_prefix("postgres://") + .or_else(|| base.strip_prefix("postgresql://")) + .expect("a postgres:// URL"); + let (creds, addr) = rest.split_once('@').expect("user:pass@host:port"); + let (user, pass) = creds.split_once(':').expect("user:pass"); + let (host, port) = addr.split_once(':').unwrap_or((addr, "5432")); + Parts { + host: host.to_owned(), + port: port.parse().expect("port"), + user: user.to_owned(), + pass: pass.to_owned(), + } +} + +/// One deployment at a time across the binary: each holds an engine pool +/// with a floor of ten connections on a server other tests share. It also +/// keeps the check's pause point to the one test that arms it. +async fn deployment_permit() -> tokio::sync::OwnedSemaphorePermit { + static DEPLOYMENTS: std::sync::OnceLock> = + std::sync::OnceLock::new(); + std::sync::Arc::clone( + DEPLOYMENTS.get_or_init(|| std::sync::Arc::new(tokio::sync::Semaphore::new(1))), + ) + .acquire_owned() + .await + .expect("the deployment semaphore is never closed") +} + +/// A deployment initialised the way `init` does it: a catalog database and a +/// separate data database, with the data connection recorded in the catalog. +pub(crate) struct Deployment { + base: String, + catalog_db: String, + data_db: String, + pub(crate) catalog: PgPool, + pub(crate) data: PgPool, + pub(crate) engine: PostgresEngine, + pub(crate) diagnostics: PostgresCatalogStore, + _permit: tokio::sync::OwnedSemaphorePermit, +} + +impl Drop for Deployment { + fn drop(&mut self) { + // Runs on a panic too, so it cannot await on the test's runtime: a + // thread of its own with its own runtime and connection. + let base = self.base.clone(); + let names = [self.catalog_db.clone(), self.data_db.clone()]; + let dropped = std::thread::spawn(move || { + let rt = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("cleanup runtime"); + rt.block_on(async { + let Ok(admin) = PgPoolOptions::new() + .max_connections(1) + .connect(&format!("{base}/postgres")) + .await + else { + return; + }; + for n in names { + let _ = sqlx::query(&format!("DROP DATABASE IF EXISTS \"{n}\" WITH (FORCE)")) + .execute(&admin) + .await; + } + admin.close().await; + }); + }) + .join(); + if dropped.is_err() { + eprintln!("cleanup of the test databases failed"); + } + } +} + +impl Deployment { + pub(crate) async fn new() -> Self { + let base = base_conn().expect("caller checks base_conn() first"); + let permit = deployment_permit().await; + let stem = format!("eddb_cck_{}", uuid::Uuid::new_v4().simple())[..20].to_owned(); + let catalog_db = format!("{stem}_catalog"); + let data_db = format!("{stem}_data"); + let p = parts(&base); + let init = PostgresBootstrapper::connect(BootstrapConfig { + host: p.host.clone(), + port: p.port, + admin_user: p.user.clone(), + admin_password: Some(p.pass.clone()), + app_user: p.user.clone(), + app_password: p.pass.clone(), + catalog_db: catalog_db.clone(), + data_db: data_db.clone(), + }) + .await + .expect("connect"); + init.create_catalog_db().await.expect("catalog db"); + init.create_data_db().await.expect("data db"); + let catalog: PgConnectOptions = + format!("{base}/{catalog_db}").parse().expect("catalog URL"); + extenddb_storage_postgres::migrations::migrate( + &extenddb_storage_postgres::migrations::PostgresTarget::new(catalog, data_db.clone()), + "test", + Duration::ZERO, + ) + .await + .expect("catalog migrations"); + init.record_data_connection().await.expect("record"); + + let catalog_url = format!("{base}/{catalog_db}"); + let catalog = PgPoolOptions::new() + .max_connections(2) + .connect(&catalog_url) + .await + .expect("catalog"); + let data = PgPoolOptions::new() + .max_connections(2) + .connect(&format!("{base}/{data_db}")) + .await + .expect("data"); + // No background workers run here, so transitions and index writes + // must complete inline. + for key in ["control_plane_delay_seconds", "index_propagation_delay_ms"] { + sqlx::query( + "INSERT INTO settings (key, value) VALUES ($1, '0') \ + ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value", + ) + .bind(key) + .execute(&catalog) + .await + .expect("pin a delay to zero"); + } + sqlx::query("INSERT INTO accounts (account_id, account_name) VALUES ($1, $2)") + .bind(ACCOUNT) + .bind(format!("acct-{stem}")) + .execute(&catalog) + .await + .expect("seed the account row"); + let engine = PostgresEngine::new( + &PostgresConfig { + connection_string: catalog_url.clone(), + pool_size: 10, + max_item_size_bytes: 400_000, + }, + REGION, + ) + .await + .expect("open a PostgresEngine"); + let diagnostics = PostgresCatalogStore::new( + PgPoolOptions::new() + .max_connections(2) + .connect(&catalog_url) + .await + .expect("diagnostics pool"), + ); + Self { + base, + catalog_db, + data_db, + catalog, + data, + engine, + diagnostics, + _permit: permit, + } + } + + pub(crate) fn catalog_url(&self) -> String { + format!("{}/{}", self.base, self.catalog_db) + } + + fn data_url(&self) -> String { + format!("{}/{}", self.base, self.data_db) + } + + /// A second diagnostics store whose catalog sessions start every + /// transaction at `isolation` unless it sets its own, as a database or + /// role default would make them. + pub(crate) async fn diagnostics_at(&self, isolation: &str) -> PostgresCatalogStore { + // The server splits the startup options at spaces; a backslash keeps + // one inside the value. + let value = isolation.replace(' ', "\\ "); + let options = self + .catalog_url() + .parse::() + .expect("the catalog URL") + .options([("default_transaction_isolation", value)]); + let pool = PgPoolOptions::new() + .max_connections(2) + .connect_with(options) + .await + .expect("a pool at the isolation level"); + let level: String = sqlx::query_scalar("SHOW default_transaction_isolation") + .fetch_one(&pool) + .await + .expect("read the default isolation"); + assert_eq!(level, isolation, "the sessions start at {isolation}"); + PostgresCatalogStore::new(pool) + } + + /// Whether the data database can hold vector data. + pub(crate) async fn pgvector(&self) -> bool { + sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM pg_extension WHERE extname = 'vector')") + .fetch_one(&self.data) + .await + .expect("list the installed extensions") + } + + /// Create `name` with GSI `gsi1`, LSI `lsi1` and, when pgvector is + /// installed, vector index `vec1`; write three items. + pub(crate) async fn create_table_with_indexes(&self, name: &str) -> TableKeyInfo { + let string = |attr: &str| AttributeDefinition { + attribute_name: attr.to_owned(), + attribute_type: ScalarAttributeType::S, + }; + let key = |attr: &str, key_type: KeyType| KeySchemaElement { + attribute_name: attr.to_owned(), + key_type, + }; + let all = || Projection { + projection_type: ProjectionType::All, + non_key_attributes: None, + }; + let vector_indexes = self.pgvector().await.then(|| { + vec![VectorIndexSpecification { + index_name: "vec1".to_owned(), + dimensions: 4, + distance_function: DistanceFunction::Cosine, + vector_attribute: VectorAttribute { + attribute_name: "emb".to_owned(), + }, + search_schema: None, + projection: Some(all()), + }] + }); + self.engine + .create_table( + ACCOUNT, + CreateTableInput { + table_name: name.to_owned(), + key_schema: vec![key("pk", KeyType::Hash), key("sk", KeyType::Range)], + attribute_definitions: vec![ + string("pk"), + string("sk"), + string("gpk"), + string("lsk"), + ], + billing_mode: Some(BillingMode::PayPerRequest), + global_secondary_indexes: Some(vec![GsiInput { + index_name: "gsi1".to_owned(), + key_schema: vec![key("gpk", KeyType::Hash)], + projection: all(), + provisioned_throughput: None, + }]), + local_secondary_indexes: Some(vec![LsiInput { + index_name: "lsi1".to_owned(), + key_schema: vec![key("pk", KeyType::Hash), key("lsk", KeyType::Range)], + projection: all(), + }]), + vector_indexes, + ..Default::default() + }, + ) + .await + .expect("create the table"); + let key_info = self + .engine + .table_key_info(ACCOUNT, name) + .await + .expect("read the key info"); + for n in 0..3 { + self.engine + .put_item( + &key_info, + item(n), + false, + None, + &ExpressionMaps::default(), + None, + ) + .await + .expect("put an item"); + } + key_info + } + + /// The relations in the data database that carry the backend's prefix. + pub(crate) async fn storage_relations(&self) -> BTreeSet { + let rows: Vec<(String,)> = sqlx::query_as( + "SELECT tablename::text FROM pg_tables \ + WHERE schemaname = current_schema() AND tablename LIKE '\\_ddb\\_%'", + ) + .fetch_all(&self.data) + .await + .expect("list the data relations"); + rows.into_iter().map(|(n,)| n).collect() + } + + /// Create a relation that no catalog row names, with the prefix the + /// engine uses, and return its name. + pub(crate) async fn plant_orphan(&self) -> String { + let name = format!("_ddb_{}", uuid::Uuid::new_v4()); + sqlx::query(&format!( + "CREATE TABLE \"{name}\" (pk TEXT PRIMARY KEY, item_data JSONB)" + )) + .execute(&self.data) + .await + .expect("plant the orphan"); + sqlx::query(&format!("INSERT INTO \"{name}\" VALUES ('x', '{{}}')")) + .execute(&self.data) + .await + .expect("fill the orphan"); + name + } + + /// Drop `relation` from the data database, as a server's drop would. + pub(crate) async fn remove_relation(&self, relation: &str) { + sqlx::query(&format!("DROP TABLE \"{relation}\"")) + .execute(&self.data) + .await + .expect("drop a data relation"); + } + + pub(crate) async fn relation_exists(&self, name: &str) -> bool { + sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM pg_tables WHERE tablename = $1)") + .bind(name) + .fetch_one(&self.data) + .await + .expect("look up a relation") + } + + /// The unquoted name of `table`'s secondary index `index` relation, from + /// its catalog id. + pub(crate) async fn index_relation(&self, table: &str, index: &str) -> String { + let (index_id,): (String,) = sqlx::query_as( + "SELECT i.index_id FROM indexes i JOIN tables t ON t.table_id = i.table_id \ + WHERE t.account_id = $1 AND t.table_name = $2 AND i.index_name = $3", + ) + .bind(ACCOUNT) + .bind(table) + .bind(index) + .fetch_one(&self.catalog) + .await + .expect("read the index id"); + index_table_name(&index_id).trim_matches('"').to_owned() + } + + /// The unquoted name of `table`'s vector index `index` relation. + pub(crate) async fn vector_relation(&self, table: &str, index: &str) -> String { + let (index_id,): (String,) = sqlx::query_as( + "SELECT v.index_id FROM vector_indexes v JOIN tables t ON t.table_id = v.table_id \ + WHERE t.account_id = $1 AND t.table_name = $2 AND v.index_name = $3", + ) + .bind(ACCOUNT) + .bind(table) + .bind(index) + .fetch_one(&self.catalog) + .await + .expect("read the vector index id"); + vector_table_name(&index_id).trim_matches('"').to_owned() + } + + async fn row_count(&self, relation: &str) -> i64 { + sqlx::query_scalar(&format!("SELECT COUNT(*) FROM \"{relation}\"")) + .fetch_one(&self.data) + .await + .expect("count the rows") + } + + pub(crate) async fn set_status(&self, sql: &str, status: &str) { + sqlx::query(sql) + .bind(status) + .execute(&self.catalog) + .await + .expect("set a status"); + } + + /// GetItem on every item written by `create_table_with_indexes`, and the + /// row count of each index relation. + pub(crate) async fn assert_data_intact(&self, key_info: &TableKeyInfo) { + for n in 0..3 { + let got = self + .engine + .get_item(key_info, &key(n)) + .await + .expect("get an item"); + assert_eq!(got, Some(item(n)), "item {n} after the fix"); + } + for index in ["gsi1", "lsi1"] { + let relation = self.index_relation(TABLE, index).await; + assert_eq!(self.row_count(&relation).await, 3, "{index} rows"); + } + if self.pgvector().await { + let relation = self.vector_relation(TABLE, "vec1").await; + assert_eq!(self.row_count(&relation).await, 3, "vec1 rows"); + } + } + + /// Wait until a session waits for a `mode` lock on `relation` in the + /// database `pool` connects to (`pg_locks` shows the request ungranted). + pub(crate) async fn wait_for_lock_request(pool: &PgPool, relation: &str, mode: &str) { + let deadline = Instant::now() + Duration::from_secs(10); + loop { + let waiting: bool = sqlx::query_scalar( + "SELECT EXISTS (SELECT 1 FROM pg_locks l JOIN pg_class c ON c.oid = l.relation \ + WHERE l.database = (SELECT oid FROM pg_database WHERE datname = current_database()) \ + AND c.relname = $1 AND l.mode = $2 AND NOT l.granted)", + ) + .bind(relation) + .bind(mode) + .fetch_one(pool) + .await + .expect("read pg_locks"); + if waiting { + return; + } + assert!( + Instant::now() < deadline, + "no session waited for {mode} on {relation} within 10 s" + ); + tokio::time::sleep(Duration::from_millis(5)).await; + } + } + + /// Start adding GSI `index` to `TABLE` the way `UpdateTable` does it: + /// lock the table row, insert the index row, and create and commit the + /// index's relation while the catalog transaction stays open. + pub(crate) async fn begin_index_add(&self, index: &str) -> IndexAdd { + let mut conn = PgConnection::connect(&self.catalog_url()) + .await + .expect("a catalog session for the index add"); + sqlx::query("BEGIN") + .execute(&mut conn) + .await + .expect("begin"); + let (table_id,): (String,) = sqlx::query_as( + "SELECT table_id FROM tables WHERE account_id = $1 AND table_name = $2 FOR UPDATE", + ) + .bind(ACCOUNT) + .bind(TABLE) + .fetch_one(&mut conn) + .await + .expect("lock the table row"); + let index_id = uuid::Uuid::new_v4().to_string(); + sqlx::query( + "INSERT INTO indexes (table_id, index_name, index_id, index_type, key_schema, \ + projection, index_status) VALUES ($1, $2, $3, 'GSI', \ + '[{\"AttributeName\": \"gpk\", \"KeyType\": \"HASH\"}]', \ + '{\"ProjectionType\": \"ALL\"}', 'CREATING')", + ) + .bind(&table_id) + .bind(index) + .bind(&index_id) + .execute(&mut conn) + .await + .expect("insert the index row"); + let relation = index_table_name(&index_id).trim_matches('"').to_owned(); + let mut data = PgConnection::connect(&self.data_url()) + .await + .expect("a data session for the index add"); + sqlx::query(&format!( + "CREATE TABLE \"{relation}\" (pk TEXT PRIMARY KEY)" + )) + .execute(&mut data) + .await + .expect("create and commit the index relation"); + data.close().await.expect("close the data session"); + IndexAdd { + conn, + table_id, + relation, + } + } +} + +/// A GSI add in flight, from [`Deployment::begin_index_add`]. +pub(crate) struct IndexAdd { + conn: PgConnection, + table_id: String, + pub(crate) relation: String, +} + +impl IndexAdd { + /// The add's next step: write the table row, which needs `ROW EXCLUSIVE` + /// on `tables` (an `UpdateTable` whose new index keys on a new attribute + /// updates `attribute_definitions` here). Returns how long it took. + pub(crate) async fn write_table_row(&mut self) -> Result { + let started = Instant::now(); + sqlx::query( + "UPDATE tables SET attribute_definitions = attribute_definitions \ + WHERE table_id = $1", + ) + .bind(&self.table_id) + .execute(&mut self.conn) + .await?; + Ok(started.elapsed()) + } + + pub(crate) async fn commit(mut self) { + sqlx::query("COMMIT") + .execute(&mut self.conn) + .await + .expect("commit the index add"); + self.conn.close().await.expect("close the catalog session"); + } +} + +/// A session of the data database holding `ACCESS SHARE` on `relation`, as +/// a reader such as `pg_dump` holds it, until it is dropped. +pub(crate) async fn hold_relation(d: &Deployment, relation: &str) -> PgConnection { + let mut conn = PgConnection::connect(&d.data_url()) + .await + .expect("a data session for the holder"); + sqlx::query("BEGIN") + .execute(&mut conn) + .await + .expect("begin"); + sqlx::query(&format!("LOCK TABLE \"{relation}\" IN ACCESS SHARE MODE")) + .execute(&mut conn) + .await + .expect("hold the relation"); + conn +} + +fn key(n: u32) -> Item { + BTreeMap::from([ + ("pk".to_owned(), AttributeValue::S(format!("p{n}"))), + ("sk".to_owned(), AttributeValue::S("s".to_owned())), + ]) +} + +fn item(n: u32) -> Item { + let mut it = key(n); + it.insert("gpk".to_owned(), AttributeValue::S("g".to_owned())); + it.insert("lsk".to_owned(), AttributeValue::S(format!("l{n}"))); + it.insert( + "emb".to_owned(), + AttributeValue::L( + ["0.1", "0.2", "0.3", &format!("{n}.4")] + .iter() + .map(|v| AttributeValue::N((*v).to_owned())) + .collect(), + ), + ); + it +} diff --git a/crates/storage/src/diagnostics.rs b/crates/storage/src/diagnostics.rs index 8f4943b57..2a77a5018 100644 --- a/crates/storage/src/diagnostics.rs +++ b/crates/storage/src/diagnostics.rs @@ -13,6 +13,11 @@ pub type DiagResult = Result; pub enum DiagError { ConnectionFailed(String), QueryFailed(String), + /// The installed backend does not implement the operation. + Unsupported(String), + /// The deployment lacks configuration the operation needs; the message + /// says what to run. + NotConfigured(String), } impl std::fmt::Display for DiagError { @@ -20,12 +25,54 @@ impl std::fmt::Display for DiagError { match self { Self::ConnectionFailed(msg) => write!(f, "Connection failed: {msg}"), Self::QueryFailed(msg) => write!(f, "Query failed: {msg}"), + Self::Unsupported(msg) => write!(f, "Not supported: {msg}"), + Self::NotConfigured(msg) => write!(f, "{msg}"), } } } impl std::error::Error for DiagError {} +/// A data relation the catalog names, with the catalog object it belongs to. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct CatalogRelation { + /// The relation's name in the data database. + pub relation: String, + /// The catalog object that owns the relation, written for an operator, for + /// example `table 123456789012/orders`. + pub owner: String, +} + +/// What [`DiagnosticsStore::check_storage_relations`] found. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct StorageRelationCheck { + /// Data relations that no catalog row names, sorted by name. + pub orphaned: Vec, + /// Data relations that a catalog row names and the data database does not + /// have, sorted by relation name. + pub missing: Vec, +} + +/// What [`DiagnosticsStore::drop_orphaned_relations`] did with one relation. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum OrphanDrop { + /// The relation was dropped here. + Dropped, + /// The relation no longer existed when the drop ran: another session, for + /// example a server's `DeleteTable`, had dropped it first. + AlreadyGone, + /// The relation was kept, for the reason given. + Kept(String), + /// The drop failed with the error given, and the relation is still there. + Failed(String), +} + +fn unsupported(operation: &str) -> DiagError { + DiagError::Unsupported(format!( + "{operation} is not implemented for this storage backend" + )) +} + /// Diagnostic and verification operations for deployment health checks. /// /// Used by `ddbo verify` to check catalog integrity and enumerate resources. @@ -40,4 +87,34 @@ pub trait DiagnosticsStore: Send + Sync { /// /// Returns the database name on success, or an error if connection fails. fn test_data_database_connection(&self) -> BoxFuture<'_, DiagResult>; + + /// Compare the data relations the backend created with the catalog, for + /// `extenddb catalog-check`. + /// + /// A relation is orphaned when no catalog row names it, whatever the row's + /// status. A relation is missing when a catalog row names it, neither the + /// row nor its table is being created or deleted, and the data database + /// does not have the relation. + /// + /// The default reports that the backend does not implement the check. + fn check_storage_relations(&self) -> BoxFuture<'_, DiagResult> { + Box::pin(async { Err(unsupported("checking data relations against the catalog")) }) + } + + /// Drop relations that [`Self::check_storage_relations`] reported as + /// orphaned, for `extenddb catalog-check --fix`. + /// + /// The backend waits for the catalog changes in flight that could name a + /// relation, checks each relation against the catalog again, and keeps + /// any relation a catalog row names by then. Each drop waits for its own + /// lock for a short time only and is reported failed when it cannot get + /// it. Returns one outcome per relation, in the order given. + /// + /// The default reports that the backend does not implement the drop. + fn drop_orphaned_relations( + &self, + _orphaned: Vec, + ) -> BoxFuture<'_, DiagResult>> { + Box::pin(async { Err(unsupported("dropping orphaned data relations")) }) + } } diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index f22a2c4d1..3c5e3cf91 100755 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -391,7 +391,13 @@ SELECT l.pid, l.mode, a.query_start, l.relation::regclass AS relation, a.query **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.) -To find such relations, compare the `_ddb_%` tables on the data database with the table ids the catalog still knows. On the catalog database: +To find such relations, stop the server and run `extenddb catalog-check` with the deployment's config file. The check lists the `_ddb_` relations in the data database's current schema and compares them with the relations the catalog's table, index and vector index rows name, through the same naming the server uses, for rows in every status. A relation no row names is reported under "orphaned data tables". A relation a row names that the data database lacks is reported under "missing data tables" with the table or index that owns it; a table or index in CREATING or DELETING is not reported missing, since its relations are being created or removed. A catalog without a data database connection string is refused, as on earlier releases: `extenddb init` always records one. + +The command refuses to run while it finds the server's PID file, but it does not see a server on another host, one started in the foreground without `--write-pid-file`, or one using another `run_dir` or port. When a server is running and the check finds a relation that no row names, it waits for the GSI additions in flight and reads the catalog again, so the relation of a GSI being added is not reported orphaned; a check that finds every relation named does not wait. It then lists the relations and reads the catalog a second time, and reports a relation missing only when both catalog reads expect it, so a table deleted while it runs is not reported missing. Reports made while a server change is in progress can still be wrong, and a later run does not repeat them: the relations of a table, GSI or vector index whose catalog row a server has just removed, and whose drop has not committed yet, are reported orphaned; and the relations of a table whose row is ACTIVE before its relations commit are reported missing (a CreateTable when `control_plane_delay_seconds` is 0, or one whose relations take longer to commit than that delay). + +`extenddb catalog-check --fix` drops the orphaned relations. Before it drops anything it waits for the GSI additions in flight to commit, reads the ids again, and keeps every relation a row names by then, so a CreateTable or UpdateTable that lands between the report and the drop loses nothing. The wait is a `SHARE` lock on the catalog's `indexes` table, requested in attempts of 100 ms for up to 30 seconds and released after the read, so a server's write to `indexes` waits behind it for at most 100 ms plus the one catalog read. Each relation is then dropped in a transaction of its own that waits at most one second for the relation's lock, and no server request waits on the command while it drops. The command prints what it did with each relation: `Dropped`; `Already gone` when another session, such as a server's DeleteTable, dropped it first; `Kept` with the reason; or `FAILED` with the error, which is what a relation another session holds gets (run the command again once that session ends). If the GSI additions in flight do not finish within 30 seconds it reports that, drops nothing, and can be run again. Relations without the `_ddb_` prefix are never touched. `--fix` treats every `_ddb_` relation in the data database's current schema as this deployment's, so run it only when no other deployment keeps data relations in that database and schema. + +To do the comparison by hand, compare the `_ddb_%` tables on the data database with the table ids the catalog still knows. On the catalog database: ```sql SELECT table_id FROM tables @@ -407,7 +413,8 @@ SELECT d.tablename LEFT JOIN (VALUES (''), (''), ('')) AS live(id) ON d.tablename = '_ddb_' || live.id OR d.tablename = '_ddb_vec_' || live.id - WHERE d.tablename LIKE '\_ddb\_%' AND live.id IS NULL; + WHERE d.tablename LIKE '\_ddb\_%' AND d.schemaname = current_schema() + AND live.id IS NULL; ``` Each row is a data relation with no catalog row. After confirming the table is gone from the catalog (`DescribeTable` returns `ResourceNotFoundException`), drop the listed relations on the data database: @@ -418,8 +425,6 @@ DROP TABLE IF EXISTS "_ddb_", "_ddb_", "_ddb_vec_