diff --git a/crates/storage-postgres/src/data/ddl.rs b/crates/storage-postgres/src/data/ddl.rs index 789447fd4..7f1d62ca1 100755 --- a/crates/storage-postgres/src/data/ddl.rs +++ b/crates/storage-postgres/src/data/ddl.rs @@ -5,7 +5,7 @@ use extenddb_core::types::{ AttributeDefinition, IndexInfo, IndexType, KeySchemaElement, Projection, StreamSpecification, - TableKeyInfo, + TableKeyInfo, VECTOR_INDEX_BACKFILLING_SCAN_PREFIX, }; use extenddb_storage::error::StorageError; use extenddb_storage::util::{sk_column, sk_column_n}; @@ -566,13 +566,22 @@ impl PostgresEngine { index_name: &str, ) -> Result { // A GSI still being built is not readable: its table is partial, or not - // there yet. The service answers a Query or Scan naming one with "The - // table does not have the specified index", the same as for no index at - // all, which is what `IndexNotFound` becomes on the wire. - let idx_row: Option<(String, String, serde_json::Value, serde_json::Value)> = + // there yet. In its backfilling phase the service refuses a Query or + // Scan naming one with "Cannot read from backfilling global secondary + // index: ". Measured 2026-10-30 on six index builds on four + // tables: DescribeTable reported `Backfilling` true 16 to 39 s after + // UpdateTable. On two tables the first Query and Scan sent after that, + // about two minutes after UpdateTable, got the backfilling text. On a + // third, Queries 0 and 2 s after `Backfilling` turned true still got + // "The table does not have the specified index", the answer before it + // turned true, and Queries about 45 s after it got the backfilling + // text. This backend has no allocation phase, so every CREATING index + // gets the backfilling refusal. `Validation` reaches the wire verbatim + // as a ValidationException. + let idx_row: Option<(String, String, String, serde_json::Value, serde_json::Value)> = sqlx::query_as( - "SELECT index_type, index_id, key_schema, projection FROM indexes \ - WHERE table_id = $1 AND index_name = $2 AND index_status <> 'CREATING'", + "SELECT index_type, index_id, index_status, key_schema, projection \ + FROM indexes WHERE table_id = $1 AND index_name = $2", ) .bind(table_id) .bind(index_name) @@ -580,8 +589,13 @@ impl PostgresEngine { .await .map_err(|e| StorageError::Internal(e.to_string()))?; - let (idx_type_str, idx_id, ks_json, proj_json) = + let (idx_type_str, idx_id, idx_status, ks_json, proj_json) = idx_row.ok_or_else(|| StorageError::IndexNotFound(index_name.to_owned()))?; + if idx_status == "CREATING" { + return Err(StorageError::Validation(format!( + "{VECTOR_INDEX_BACKFILLING_SCAN_PREFIX}{index_name}" + ))); + } let index_type = match idx_type_str.as_str() { "GSI" => IndexType::Gsi, diff --git a/crates/storage-postgres/src/data/gsi_build.rs b/crates/storage-postgres/src/data/gsi_build.rs index 22ee5bf6f..d8773ba9b 100644 --- a/crates/storage-postgres/src/data/gsi_build.rs +++ b/crates/storage-postgres/src/data/gsi_build.rs @@ -6,13 +6,19 @@ //! Three pieces, and the order they run in is the design: //! //! 1. `UpdateTable` creates the index data table, then commits the catalog row as -//! `CREATING`. From that commit on, every write that reads the catalog maintains -//! the index the way it maintains an `ACTIVE` one, inline or through the queue. -//! 2. It waits [`GRACE_PUBLISH`], takes a [`write_barrier`], and runs -//! [`backfill`], which writes the current state of every base item into the -//! index under that item's row lock. -//! 3. It flips the row to `ACTIVE`. A build that fails or is cut short leaves the -//! index `CREATING`, and [`recovery_worker`] finishes it. +//! `CREATING`, starts the build in a task of its own and returns. From that +//! commit on, every write that reads the catalog maintains the index the way it +//! maintains an `ACTIVE` one, inline or through the queue. +//! 2. The build task, once it holds one of the engine's [`GSI_BUILD_SLOTS`], waits +//! [`GRACE_PUBLISH`], takes a [`write_barrier`], and runs [`backfill`], which +//! writes the current state of every base item into the index under that +//! item's row lock. +//! 3. The task flips the row to `ACTIVE`. A build that fails for a reason +//! [`is_transient`] accepts (a conflict, a timeout, a lost or refused +//! connection, a shutdown, a server-side I/O or internal error, an exhausted +//! connection pool), or that a restart cuts short, leaves the index `CREATING`, +//! and [`recovery_worker`] finishes it. Any other failure would recur on every +//! attempt: it is logged and the index removed ([`remove_failed_build`]). //! //! Every write to the index, from the backfill, the queue or a write's own inline //! maintenance, leaves the rows for one base key equal to that item's state at the @@ -27,11 +33,12 @@ //! //! - A write whose base table statements had all returned within [`GRACE_WRITE`] //! of its catalog read holds the table's `ROW EXCLUSIVE` lock from the first of -//! them until it commits. The barrier comes [`GRACE_PUBLISH`] (longer) after the -//! catalog commit, so it finds that lock held, waits for the commit, and the -//! backfill, which starts after it, sees the write. (A `SELECT ... FOR UPDATE` -//! alone takes only `ROW SHARE`, which the barrier does not wait for; the check -//! runs after the write's `INSERT`, `UPDATE` or `DELETE`, which is what counts.) +//! them until it commits. The barrier comes at least [`GRACE_PUBLISH`] (longer) +//! after the catalog commit, so it either finds that lock held and waits for the +//! commit or comes after the commit, and the backfill, which starts after it, +//! sees the write. (A `SELECT ... FOR UPDATE` alone takes only `ROW SHARE`, +//! which the barrier does not wait for; the check runs after the write's +//! `INSERT`, `UPDATE` or `DELETE`, which is what counts.) //! - A slower write re-reads the catalog before it commits and queues a repair for //! any index it did not know about. If the re-read does not show the index, the //! catalog commit came after it, so that write's lock was again taken before the @@ -39,8 +46,8 @@ //! //! Measurement error only ever makes a write look slower or the wait longer, which //! is the safe direction: the clock is one that keeps counting while the host is -//! suspended, the writer measures to a point after its write, and `UpdateTable` -//! measures from after its commit returned. +//! suspended, the writer measures to a point after its write, and the build task, +//! which `UpdateTable` starts after its commit returned, measures from its start. //! //! One gap is left. A write that needs no transaction (a plain put or delete on //! a table it believed had no index) commits with its one statement, so if it @@ -72,9 +79,11 @@ use crate::gsi_queue::{GsiApplyContext, GsiIndexDef, PendingApplyContext, enqueu /// its commit, and still rely on the backfill to have seen it. pub(crate) const GRACE_WRITE: Duration = Duration::from_millis(500); -/// How long `UpdateTable` waits after committing a `CREATING` index before the -/// barrier and backfill. Must exceed [`GRACE_WRITE`]; the margin covers the two -/// clocks (a writer's and this call's) disagreeing on how long half a second is, +/// How long a GSI build waits before its barrier and backfill. The build task +/// starts the wait after `UpdateTable` has committed the `CREATING` index row and +/// the task holds its build slot, so the barrier comes at least this long after +/// the commit. Must exceed [`GRACE_WRITE`]; the margin covers the two clocks (a +/// writer's and the build task's) disagreeing on how long half a second is, /// which is the one error the monotonic clock does not remove. pub(crate) const GRACE_PUBLISH: Duration = Duration::from_millis(1_000); @@ -85,6 +94,18 @@ pub(crate) const GRACE_PUBLISH: Duration = Duration::from_millis(1_000); /// point here. const BACKFILL_BATCH: i64 = 256; +/// GSI builds one engine (one server process) runs at once: the builds +/// `UpdateTable` starts and the recovery sweep's together. +/// +/// A running build holds a PostgreSQL session of its own for its ownership lock +/// (`vector_index::build_ownership`), and one data pool connection at a time for +/// its barrier and page transactions. `UpdateTable` returns before its build +/// starts, so without a bound a client that adds an index to one table after +/// another starts all the builds at once, each with its own session. A build +/// that finds every slot taken waits for one in its task with no connection +/// open, still `CREATING`; slots go to builds in the order they asked. +pub const GSI_BUILD_SLOTS: usize = 4; + /// A point in time on a clock that keeps counting while the host is suspended. /// /// `std::time::Instant` is `CLOCK_MONOTONIC` on Linux, which stops during suspend, @@ -187,6 +208,54 @@ pub(crate) async fn pause_before_autocommit_statement() { pause(&PAUSE_BEFORE_AUTOCOMMIT_MS).await; } +/// Test hook: the tables whose GSI builds are held, each with the number of +/// backfill pages a build writes before it stops (0: before the scan starts). +#[cfg(feature = "test-hooks")] +static BACKFILL_HOLDS: std::sync::Mutex> = std::sync::Mutex::new(Vec::new()); + +/// Hold the GSI builds of table `table_id` once they have written `after_pages` +/// backfill pages (`Some(0)`: before the scan starts), or release them (`None`). +/// +/// Lets a test see `UpdateTable` return while the backfill is stuck, write to the +/// table during the build, or stop a build partway and restart the server under +/// it. Keyed by table so the tests of one binary, which run in parallel, do not +/// hold each other's builds. A hold lasts at most a minute, so a test that fails +/// before releasing it cannot wedge a build for good. +#[cfg(feature = "test-hooks")] +#[doc(hidden)] +pub fn hold_gsi_backfill_for_tests(table_id: &str, after_pages: Option) { + let mut holds = BACKFILL_HOLDS + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + holds.retain(|(t, _)| t != table_id); + if let Some(pages) = after_pages { + holds.push((table_id.to_owned(), pages)); + } +} + +/// Where a build waits while [`hold_gsi_backfill_for_tests`] holds it: at the +/// build's entry with `pages` 0, and after each backfill page with the count of +/// pages written so far. +#[cfg(feature = "test-hooks")] +async fn backfill_hold_point(table_id: &str, pages: usize) { + let held = || { + BACKFILL_HOLDS + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .iter() + .any(|(t, n)| t == table_id && *n == pages) + }; + let step = Duration::from_millis(20); + let mut waited = Duration::ZERO; + while held() && waited < Duration::from_secs(60) { + tokio::time::sleep(step).await; + waited += step; + } +} + +#[cfg(not(feature = "test-hooks"))] +async fn backfill_hold_point(_table_id: &str, _pages: usize) {} + /// One typed key value, as stored in a base or index table's key column. #[derive(Debug, Clone)] enum KeyVal { @@ -573,6 +642,7 @@ pub(crate) async fn backfill( .transpose()?; let mut cursor: Option = None; + let mut pages = 0usize; while let Some(end) = &last { // 1. The page's keys. let mut conds = Vec::new(); @@ -642,6 +712,8 @@ pub(crate) async fn backfill( .await?; } + pages += 1; + backfill_hold_point(table_id, pages).await; if same_key(&upper, end) { break; } @@ -926,6 +998,8 @@ pub(crate) struct GsiBuild { pub(crate) index_name: String, pub(crate) def: GsiIndexDef, /// Held for the whole build so a peer's sweep leaves it alone; released on drop. + /// An `UpdateTable` build that had to wait for its build slot takes it in its + /// task when the slot comes. pub(crate) _owner: Option, } @@ -1009,6 +1083,26 @@ async fn index_row_exists( Ok(found.is_some()) } +/// Whether this index is still a `CREATING` GSI on an `ACTIVE` table. Read under +/// the index's ownership by a build that has waited for a build slot (an +/// `UpdateTable` build or the recovery sweep): while it waited, another build +/// may have published the index, or the index or its table may have been +/// deleted. +pub(crate) async fn index_still_building( + catalog: &sqlx::PgPool, + index_id: &str, +) -> Result { + let found: Option<(i32,)> = sqlx::query_as( + "SELECT 1 FROM indexes i JOIN tables t ON t.table_id = i.table_id \ + WHERE i.index_id = $1 AND i.index_status = 'CREATING' AND t.table_status = 'ACTIVE'", + ) + .bind(index_id) + .fetch_optional(catalog) + .await + .map_err(db_error)?; + Ok(found.is_some()) +} + /// Wait, barrier, backfill and publish each build, in order. /// /// Returns how many this call published, or the position of the first build @@ -1026,6 +1120,7 @@ pub(crate) async fn build_gsis( if builds.is_empty() { return Ok(0); } + backfill_hold_point(table_id, 0).await; let mut published_here = 0usize; tokio::time::sleep(GRACE_PUBLISH).await; if let Err(e) = write_barrier(data, table_id).await { @@ -1218,18 +1313,21 @@ async fn reconcile_pass( else { continue; }; + // A build slot ([`GSI_BUILD_SLOTS`]), the same ones UpdateTable's builds + // take. After ownership, so the sweep waits only for an index it is going + // to build, holding that one ownership session while it waits; before the + // re-read, so the re-read sees the index as it is when the wait ends. + let Ok(_slot) = std::sync::Arc::clone(&engine.gsi_build_slots) + .acquire_owned() + .await + else { + continue; + }; // Re-read under ownership: the list above is a snapshot, and the build // that held this index a moment ago may have published or abandoned it. - let still_building: Result, _> = sqlx::query_as( - "SELECT 1 FROM indexes i JOIN tables t ON t.table_id = i.table_id \ - WHERE i.index_id = $1 AND i.index_status = 'CREATING' AND t.table_status = 'ACTIVE'", - ) - .bind(&index_id) - .fetch_optional(&engine.pool) - .await; - match still_building { - Ok(Some(_)) => {} - Ok(None) => continue, + match index_still_building(&engine.pool, &index_id).await { + Ok(true) => {} + Ok(false) => continue, Err(e) => { tracing::warn!(index_id = %index_id, "could not re-check a CREATING GSI: {e}"); continue; diff --git a/crates/storage-postgres/src/delete_table.rs b/crates/storage-postgres/src/delete_table.rs index 81b75856b..83b492117 100755 --- a/crates/storage-postgres/src/delete_table.rs +++ b/crates/storage-postgres/src/delete_table.rs @@ -85,6 +85,25 @@ impl PostgresEngine { }); } + // A GSI whose backfill is still running: the service refuses to delete + // the table until the index is ACTIVE (measured 2026-10-30, twice on a + // table whose index stayed CREATING for 15 minutes, both answers + // ResourceInUseException with this text). The row lock above orders + // this check against UpdateTable's insert of a CREATING row, which + // locks the same table row. + // + // It comes before the deletion-protection check: on a protected table + // with an index being created, the service answers this text, not the + // protection ValidationException (measured 2026-10-30, twice before + // and twice after `Backfilling` turned true). + if index_rows.iter().any(|idx| idx.index_status == "CREATING") { + return Err(StorageError::IndexesInUse( + "Attempt to change a resource which is still in use: Cannot delete table \ + while indexes are being created, updated, or deleted." + .to_owned(), + )); + } + // REQ: DeletionProtectionEnabled check — real DynamoDB returns ValidationException if row.deletion_protection_enabled { return Err(StorageError::DeletionProtected(row.table_arn.clone())); diff --git a/crates/storage-postgres/src/lib.rs b/crates/storage-postgres/src/lib.rs index 81480973c..9b549be58 100755 --- a/crates/storage-postgres/src/lib.rs +++ b/crates/storage-postgres/src/lib.rs @@ -44,6 +44,15 @@ pub use catalog_store::PostgresCatalogStore; pub use config::PostgresStorageConfig; pub use config::parse_connection_string; pub use credential_store::DbCredentialStore; +/// GSI builds one engine runs at once, so a test can start one more than that. +#[cfg(feature = "test-hooks")] +#[doc(hidden)] +pub use data::gsi_build::GSI_BUILD_SLOTS; +/// Hold a table's GSI builds at their entry or after a number of backfill pages, +/// so a test can act while a build is under way. +#[cfg(feature = "test-hooks")] +#[doc(hidden)] +pub use data::gsi_build::hold_gsi_backfill_for_tests; /// 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; @@ -298,6 +307,10 @@ pub struct PostgresEngine { /// wait; the open needs only a data connection by then, so they delay it /// no further. pub(crate) snapshot_open_slots: Arc, + /// One permit per GSI build that may run at once (`GSI_BUILD_SLOTS` in + /// `data::gsi_build`), taken by `UpdateTable`'s build task and by the + /// recovery sweep. Shared by clones. + pub(crate) gsi_build_slots: Arc, /// Whether `data_pool` is the catalog pool (no separate data database). /// The snapshot open uses one transaction on that layout, and /// `DeleteTable` runs its in-use probe and its synchronous drop on the @@ -432,6 +445,9 @@ impl PostgresEngine { snapshot_open_slots: Arc::new(tokio::sync::Semaphore::new(snapshot_slot_count( pool_size, ))), + gsi_build_slots: Arc::new(tokio::sync::Semaphore::new( + data::gsi_build::GSI_BUILD_SLOTS, + )), shared_data_pool, #[cfg(feature = "test-hooks")] drop_attempts: Arc::default(), @@ -480,6 +496,16 @@ impl PostgresEngine { self } + /// Give this engine `slots` GSI build slots instead of `GSI_BUILD_SLOTS`, so + /// a test can make builds wait for one. Call it before any build starts. + #[cfg(feature = "test-hooks")] + #[doc(hidden)] + #[must_use] + pub fn with_gsi_build_slots_for_tests(mut self, slots: usize) -> Self { + self.gsi_build_slots = Arc::new(tokio::sync::Semaphore::new(slots)); + self + } + /// Returns a handle to the control plane notify, for use by the /// background poller task (F-3). #[must_use] diff --git a/crates/storage-postgres/src/update_table.rs b/crates/storage-postgres/src/update_table.rs index dec99a93b..d9d7a9fc7 100755 --- a/crates/storage-postgres/src/update_table.rs +++ b/crates/storage-postgres/src/update_table.rs @@ -11,7 +11,7 @@ use extenddb_storage::util::{building_refuses, effective_attribute_definitions}; use crate::PostgresEngine; use crate::data::gsi_build::{ - GsiBuild, build_gsis, drop_gsi_tables, is_transient, remove_failed_build, + GsiBuild, build_gsis, drop_gsi_tables, index_still_building, is_transient, remove_failed_build, }; /// Give back the build holds a failed `UpdateTable` took. @@ -792,6 +792,13 @@ impl PostgresEngine { // recovery sweep never sees a CREATING row with nobody holding it. It is an // efficiency measure only: a second, concurrent build is correct, because a // full backfill pass is enough on its own to publish the index. + // + // That is for a build that gets one of the engine's build slots + // (`GSI_BUILD_SLOTS`) here; the request takes one only if it is free, and + // never waits. A build that gets none waits for a slot in its task and + // takes ownership then, so that while it waits it holds no PostgreSQL + // session. Until then its CREATING row has no owner, and a recovery sweep + // may take the index and build it instead. let base_key_schema: Vec = serde_json::from_value(ks_json.clone()) .map_err(|e| StorageError::Internal(e.to_string()))?; let base_attr_defs: Vec = serde_json::from_value(ad_json.clone()) @@ -803,12 +810,23 @@ impl PostgresEngine { .clone() .unwrap_or_else(|| base_attr_defs.clone()); let mut gsi_builds: Vec = Vec::new(); + let build_slot = if created_index_ids.is_empty() { + None + } else { + std::sync::Arc::clone(&self.gsi_build_slots) + .try_acquire_owned() + .ok() + }; if let Some(updates) = &input.global_secondary_index_updates { let creates = updates.iter().filter_map(|u| u.create.as_ref()); for (create, index_id) in creates.zip(&created_index_ids) { - let owner = - crate::data::vector_index::build_ownership(&self.data_pool, index_id).await; - if owner.is_none() { + let owner = match &build_slot { + Some(_) => { + crate::data::vector_index::build_ownership(&self.data_pool, index_id).await + } + None => None, + }; + if owner.is_none() && build_slot.is_some() { // The build goes ahead without it: a second build of the same index // by the recovery worker is harmless to the index's contents, and // removal of a failed build is conditional on the row still being @@ -1023,25 +1041,84 @@ impl PostgresEngine { // why the wait and the barrier make the backfill see every write that did // not see the index. // - // A build that fails for a passing reason (a conflict, a timeout, a lost - // connection) leaves its index CREATING, with its data table, for the - // recovery worker to finish, and the request still succeeds, reporting the - // index as CREATING: the index exists, writes are maintaining it, and what - // is missing is a scan that can be re-run. One that would fail the same way - // every time is removed and reported, as before. + // The build runs in a task of its own and the request returns without + // waiting for it, which is what the service does: UpdateTable answers at + // once with the index `CREATING`, the backfill runs in the background, and + // DescribeTable reports `ACTIVE` when it has finished. Measured on + // 2026-10-30 with 2,000 items: the response came back in under 100 ms with + // `IndexStatus: CREATING`, and the index turned `ACTIVE` minutes later. + // Awaiting the build here made the response wait for every existing item + // to be indexed, which exceeded the SDKs' default read timeout on a + // 20,000-item table (#654). // - // In a task of its own, awaited here, so a client that disconnects cannot - // cancel the build halfway: that would leave the barrier's lock request or a - // page's row locks to be released whenever the pooled connection is next + // Detached rather than awaited also means a client that disconnects cannot + // cancel the build halfway: that would leave the barrier's lock request or + // a page's row locks to be released whenever the pooled connection is next // used. + // + // What the task leaves behind on failure. A build that fails for a passing + // reason (a conflict, a timeout, a lost connection) leaves its index + // CREATING, with its data table, and ends; the ownership lock goes with the + // task, and the recovery worker (`gsi_build::recovery_worker`) rebuilds the + // index on its next pass, as it does after a server restart. One that would + // fail the same way every time is removed, as the recovery worker removes + // one, so that an index never sits CREATING forever refusing every other + // index change on the table; the failure is logged at error level, and the + // caller sees the index gone from DescribeTable and can issue the Create + // again. The service never reports a failed backfill, so there is no + // measured state to copy here. if !gsi_builds.is_empty() { let catalog = self.pool.clone(); let data = self.data_pool.clone(); + let slots = std::sync::Arc::clone(&self.gsi_build_slots); let build_table_id = table_id.clone(); let table_name = input.table_name.clone(); let base_ks = base_key_schema.clone(); let attr_defs = effective_attr_defs.clone(); - let task = tokio::spawn(async move { + tokio::spawn(async move { + // Held until the task ends, whichever way it ends. + let _slot = match build_slot { + Some(slot) => slot, + None => { + // No slot was free when the request ran: wait for one, then + // take ownership. An index someone else owns by then (a + // recovery sweep, here or on a peer) is theirs to build, and + // so is one whose ownership session cannot be opened, which + // the recovery sweep retries. + let Ok(slot) = slots.acquire_owned().await else { + // Only a closed semaphore refuses, and this one is + // never closed. + return; + }; + for mut b in std::mem::take(&mut gsi_builds) { + b._owner = + crate::data::vector_index::build_ownership(&data, &b.def.index_id) + .await; + if b._owner.is_none() { + continue; + } + // Re-read under ownership, as the recovery sweep does + // after its own wait. The index had no owner while this + // task waited, so a sweep may have built, published and + // let go of it by now, or the index may have been + // deleted. A second build would scan the whole table + // again behind a write barrier for nothing. + match index_still_building(&catalog, &b.def.index_id).await { + Ok(true) => gsi_builds.push(b), + Ok(false) => {} + Err(e) => tracing::warn!( + index_id = %b.def.index_id, + "could not re-check a CREATING GSI; leaving it to the \ + recovery worker: {e}" + ), + } + } + if gsi_builds.is_empty() { + return; + } + slot + } + }; let built = build_gsis( &catalog, &data, @@ -1052,7 +1129,7 @@ impl PostgresEngine { ) .await; let Err((first_failed, e)) = built else { - return Ok(()); + return; }; let unfinished = &gsi_builds[first_failed..]; if is_transient(&e) { @@ -1063,12 +1140,12 @@ impl PostgresEngine { b.index_name, ); } - return Ok(()); + return; } - // A failure that would recur on every attempt: remove the index and - // report it, as before. The engine allows one index action per - // UpdateTable, so `unfinished` is the one build in practice; the - // loop is for the shape of the data, not for a second index. + // A failure that would recur on every attempt: remove the index. The + // engine allows one index action per UpdateTable, so `unfinished` is + // the one build in practice; the loop is for the shape of the data, + // not for a second index. for b in unfinished { tracing::error!( "Failed to build GSI '{}' on '{table_name}', removing it: {e}", @@ -1076,15 +1153,7 @@ impl PostgresEngine { ); remove_failed_build(&catalog, &data, &build_table_id, b).await; } - Err(e) }); - match task.await { - Ok(result) => result?, - Err(e) => tracing::error!( - "GSI build task on '{}' ended abnormally; recovery will finish it: {e}", - input.table_name - ), - } } self.build_table_description(account_id, &input.table_name) diff --git a/crates/storage-postgres/tests/gsi_async_build.rs b/crates/storage-postgres/tests/gsi_async_build.rs new file mode 100644 index 000000000..4dd3db0b1 --- /dev/null +++ b/crates/storage-postgres/tests/gsi_async_build.rs @@ -0,0 +1,606 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 +//! Storage-level tests for a GSI build that runs after `UpdateTable` returns. +//! +//! `UpdateTable` with a GSI `Create` answers once the index row is committed, +//! with the index `CREATING`, and a detached task backfills it (issue #654). +//! These tests hold that task with the crate's test hook and check what a +//! client sees meanwhile, what the backfill leaves behind once released, and +//! that a build cut off by a server restart still reaches `ACTIVE`. +//! +//! Each test builds its own throwaway database, applies the shipped migrations +//! to it, and drops it when it passes. A failing test leaves its database behind +//! on purpose, named `eddb_gsia_*`, so the state that failed can be inspected. +//! +//! Requires `EXTENDDB_TEST_PG_CONNECTION_STRING`, a base URL with no database +//! component (for example `postgresql://postgres@127.0.0.1:5432`), pointing at a +//! server whose role may create and drop databases. Without it every test here +//! reports a skip and passes, the same convention the wire suites use. + +use std::collections::{BTreeMap, BTreeSet}; +use std::time::Duration; + +use extenddb_core::expression::ExpressionMaps; +use extenddb_core::types::{ + AttributeDefinition, AttributeValue, BillingMode, CreateTableInput, DeleteTableInput, + DescribeTableInput, Item, KeySchemaElement, KeyType, ScalarAttributeType, TableDescription, + TableKeyInfo, TableStatus, UpdateTableInput, +}; +use extenddb_storage::error::StorageError; +use extenddb_storage::{DataEngine, TableEngine}; +use extenddb_storage_postgres::{PostgresConfig, PostgresEngine, hold_gsi_backfill_for_tests}; +use sqlx::PgPool; +use sqlx::postgres::PgPoolOptions; + +const ACCOUNT: &str = "123456789012"; +const REGION: &str = "us-east-1"; +const TABLE: &str = "t_gsi_async_build"; +/// Items the tests load before adding the index: more than two backfill pages +/// of 256 keys, so a build can be stopped between pages. +const ITEMS: u32 = 600; + +struct Scratch { + engine: PostgresEngine, + db: PgPool, + admin: PgPool, + db_name: String, + url: String, +} + +impl Scratch { + async fn cleanup(self) { + let Scratch { + engine, + db, + admin, + db_name, + .. + } = self; + drop(engine); + db.close().await; + sqlx::query(&format!( + "DROP DATABASE IF EXISTS \"{db_name}\" WITH (FORCE)" + )) + .execute(&admin) + .await + .expect("drop the scratch database"); + admin.close().await; + } +} + +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()) +} + +fn shipped_migrations() -> Vec { + let root = std::path::Path::new(env!("CARGO_MANIFEST_DIR")); + let mut out = Vec::new(); + for dir in ["migrations", "data_migrations"] { + let mut files: Vec<_> = std::fs::read_dir(root.join(dir)) + .expect("read the migrations directory") + .map(|e| e.expect("directory entry").path()) + .filter(|p| p.extension().is_some_and(|e| e == "sql")) + .collect(); + files.sort(); + for f in files { + out.push(std::fs::read_to_string(&f).expect("read a migration file")); + } + } + out +} + +fn skip(test: &str) { + eprintln!( + "SKIP {test}: EXTENDDB_TEST_PG_CONNECTION_STRING is not set, so there is no PostgreSQL \ + to build a scratch catalog in." + ); +} + +fn config(url: &str) -> PostgresConfig { + PostgresConfig { + connection_string: url.to_owned(), + pool_size: 10, + max_item_size_bytes: 400_000, + } +} + +async fn scratch() -> Scratch { + let base = base_conn().expect("caller checks base_conn() first"); + let db_name = format!("eddb_gsia_{}", uuid::Uuid::new_v4().simple())[..24].to_owned(); + let admin = PgPoolOptions::new() + .max_connections(1) + .connect(&format!("{base}/postgres")) + .await + .expect("connect to the postgres maintenance database"); + sqlx::query(&format!("CREATE DATABASE \"{db_name}\"")) + .execute(&admin) + .await + .expect("create the scratch database"); + let url = format!("{base}/{db_name}"); + let db = PgPoolOptions::new() + .max_connections(2) + .connect(&url) + .await + .expect("connect to the scratch database"); + // One scratch database stands in for both the catalog and the data database. + for sql in shipped_migrations() { + sqlx::raw_sql(&sql) + .execute(&db) + .await + .expect("apply a shipped migration"); + } + sqlx::query( + "INSERT INTO settings (key, value) VALUES ('control_plane_delay_seconds', '0') \ + ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value", + ) + .execute(&db) + .await + .expect("pin the control-plane delay to zero"); + sqlx::query("INSERT INTO accounts (account_id, account_name) VALUES ($1, $2)") + .bind(ACCOUNT) + .bind(format!("acct-{db_name}")) + .execute(&db) + .await + .expect("seed the account row"); + let engine = PostgresEngine::new(&config(&url), REGION) + .await + .expect("open a PostgresEngine on the scratch database") + .start_gsi_workers(); + Scratch { + engine, + db, + admin, + db_name, + url, + } +} + +async fn create_table(s: &Scratch) -> TableKeyInfo { + s.engine + .create_table( + ACCOUNT, + CreateTableInput { + table_name: TABLE.to_owned(), + key_schema: vec![ + KeySchemaElement { + attribute_name: "pk".to_owned(), + key_type: KeyType::Hash, + }, + KeySchemaElement { + attribute_name: "sk".to_owned(), + key_type: KeyType::Range, + }, + ], + attribute_definitions: vec![ + AttributeDefinition { + attribute_name: "pk".to_owned(), + attribute_type: ScalarAttributeType::S, + }, + AttributeDefinition { + attribute_name: "sk".to_owned(), + attribute_type: ScalarAttributeType::N, + }, + ], + billing_mode: Some(BillingMode::PayPerRequest), + ..Default::default() + }, + ) + .await + .expect("create the table"); + s.engine + .table_key_info(ACCOUNT, TABLE) + .await + .expect("read the key info") +} + +/// Load `ITEMS` items, every one carrying `g`, so the backfill has more than +/// two pages to write. +async fn load_items(s: &Scratch, key_info: &TableKeyInfo) { + for n in 0..ITEMS { + put(s, key_info, item(n, Some(&format!("g{}", n % 7)), Some(n))).await; + } +} + +/// The `UpdateTable` input that adds GSI `gidx` keyed on `g` (S) and `r` (N). +fn add_gidx() -> UpdateTableInput { + serde_json::from_value(serde_json::json!({ + "TableName": TABLE, + "AttributeDefinitions": [ + {"AttributeName": "g", "AttributeType": "S"}, + {"AttributeName": "r", "AttributeType": "N"}, + ], + "GlobalSecondaryIndexUpdates": [{"Create": { + "IndexName": "gidx", + "KeySchema": [ + {"AttributeName": "g", "KeyType": "HASH"}, + {"AttributeName": "r", "KeyType": "RANGE"}, + ], + "Projection": {"ProjectionType": "ALL"}, + }}], + })) + .expect("build the UpdateTable input") +} + +fn item(n: u32, g: Option<&str>, r: Option) -> Item { + let mut it = BTreeMap::from([ + ("pk".to_owned(), AttributeValue::S(format!("p{}", n % 5))), + ("sk".to_owned(), AttributeValue::N(n.to_string())), + ]); + if let Some(g) = g { + it.insert("g".to_owned(), AttributeValue::S(g.to_owned())); + } + if let Some(r) = r { + it.insert("r".to_owned(), AttributeValue::N(r.to_string())); + } + it +} + +fn key(n: u32) -> Item { + item(n, None, None) +} + +async fn put(s: &Scratch, key_info: &TableKeyInfo, it: Item) { + s.engine + .put_item(key_info, it, false, None, &ExpressionMaps::default(), None) + .await + .expect("put"); +} + +async fn delete(s: &Scratch, key_info: &TableKeyInfo, n: u32) { + s.engine + .delete_item( + key_info, + &key(n), + false, + None, + &ExpressionMaps::default(), + None, + ) + .await + .expect("delete"); +} + +async fn describe(s: &Scratch) -> TableDescription { + s.engine + .describe_table( + ACCOUNT, + DescribeTableInput { + table_name: TABLE.to_owned(), + }, + ) + .await + .expect("describe the table") +} + +/// `IndexStatus` of `gidx` in a description, as a client would read it. +fn gidx_status(desc: &TableDescription) -> String { + desc.global_secondary_indexes + .as_ref() + .and_then(|gsis| gsis.iter().find(|g| g.index_name == "gidx")) + .map(|g| g.index_status.clone()) + .expect("the description lists gidx") +} + +const IDS_SQL: &str = "SELECT t.table_id, i.index_id, i.index_status FROM tables t \ + JOIN indexes i ON i.table_id = t.table_id \ + WHERE t.table_name = $1 AND i.index_name = 'gidx'"; + +async fn ids(db: &PgPool) -> (String, String, String) { + sqlx::query_as(IDS_SQL) + .bind(TABLE) + .fetch_one(db) + .await + .expect("look up the ids") +} + +/// `ids` for an index another thread's `UpdateTable` is adding: wait for its +/// catalog row to commit. +async fn wait_ids(db: &PgPool) -> (String, String, String) { + for _ in 0..600 { + let row: Option<(String, String, String)> = sqlx::query_as(IDS_SQL) + .bind(TABLE) + .fetch_optional(db) + .await + .expect("look up the ids"); + if let Some(row) = row { + return row; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + panic!("the gidx row never appeared"); +} + +async fn data_table_id(db: &PgPool) -> String { + sqlx::query_scalar("SELECT table_id FROM tables WHERE table_name = $1") + .bind(TABLE) + .fetch_one(db) + .await + .expect("look up the table id") +} + +async fn index_row_count(db: &PgPool, index_id: &str) -> i64 { + sqlx::query_scalar(&format!("SELECT count(*) FROM \"_ddb_{index_id}\"")) + .fetch_one(db) + .await + .expect("count the index rows") +} + +/// Wait until the backfill has written at least `n` rows, which with a hold +/// after the first page means the first page is in and the build has stopped. +async fn wait_index_rows_at_least(db: &PgPool, index_id: &str, n: i64) -> i64 { + let mut rows = 0; + for _ in 0..600 { + rows = index_row_count(db, index_id).await; + if rows >= n { + break; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + rows +} + +/// `(base_pk, base_sk, g, item_data)` for every row of the index: the key +/// columns and the projected item, so a row with the right key but a stale +/// projection shows up as a difference. +async fn index_rows(db: &PgPool, index_id: &str) -> BTreeSet<(String, String, String, String)> { + let rows: Vec<(String, String, String, String)> = sqlx::query_as(&format!( + "SELECT base_pk, base_sk_n::text, pk, item_data::text FROM \"_ddb_{index_id}\"" + )) + .fetch_all(db) + .await + .expect("read the index table"); + rows.into_iter().collect() +} + +/// `(pk, sk, g, item_data)` for every base item carrying `g`; the GSI projects +/// ALL, so the index row's item is the base item. +async fn expected_rows(db: &PgPool, table_id: &str) -> BTreeSet<(String, String, String, String)> { + let rows: Vec<(String, String, String, String)> = sqlx::query_as(&format!( + "SELECT pk, sk_n::text, item_data->'g'->>'S', item_data::text FROM \"_ddb_{table_id}\" \ + WHERE item_data ? 'g'" + )) + .fetch_all(db) + .await + .expect("read the base table"); + rows.into_iter().collect() +} + +/// Wait for GSI `gidx` to become ACTIVE, which the detached build does when its +/// backfill finishes. A build that hit a transient failure stays CREATING for +/// the recovery worker, which does not run in these tests, so a wait that never +/// sees ACTIVE tells the two apart. +async fn wait_active(db: &PgPool) -> String { + let mut status = String::new(); + for _ in 0..600 { + status = ids(db).await.2; + if status == "ACTIVE" { + break; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + status +} + +async fn wait_queue_empty(db: &PgPool) { + for _ in 0..500 { + let n: i64 = sqlx::query_scalar("SELECT count(*) FROM gsi_pending") + .fetch_one(db) + .await + .expect("count pending rows"); + if n == 0 { + return; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + panic!("the propagation queue did not drain"); +} + +/// The index holds exactly the base items that carry `g`, each once. +async fn assert_index_matches_base(db: &PgPool, table_id: &str, index_id: &str) { + wait_queue_empty(db).await; + let expected = expected_rows(db, table_id).await; + let indexed = index_rows(db, index_id).await; + assert_eq!( + indexed, expected, + "the index rows differ from the base items" + ); + assert_eq!( + index_row_count(db, index_id).await, + expected.len() as i64, + "an item is indexed more than once" + ); +} + +/// Writes made while the backfill is stopped after its first page. The scan +/// runs in `(pk, sk)` order and the first page covers every `p0` and `p1` item, +/// so `p0` writes land behind the scan and `p3`/`p4` writes ahead of it. +async fn write_during_build(s: &Scratch, key_info: &TableKeyInfo) { + // Behind the scan: a changed key, a removed item, a new item, and an item + // that loses `g` and must leave the index. + put(s, key_info, item(0, Some("moved"), Some(0))).await; + delete(s, key_info, 5).await; + put(s, key_info, item(1000, Some("new"), Some(1000))).await; + put(s, key_info, item(10, None, Some(10))).await; + // Ahead of the scan: the same four shapes. + put(s, key_info, item(3, Some("moved"), Some(3))).await; + delete(s, key_info, 4).await; + put(s, key_info, item(1003, Some("new"), Some(1003))).await; + put(s, key_info, item(13, None, Some(13))).await; +} + +/// `UpdateTable` returns with the index `CREATING` while the backfill has not +/// started; a read of the index, and a DeleteTable, are refused as the service +/// refuses them; writes made during the build end up indexed once; and the index +/// reaches `ACTIVE` when the build finishes. +/// +/// With the build awaited inside `UpdateTable` again, the first assertion +/// fails: the request waits out the hold. +#[tokio::test] +async fn update_table_returns_while_the_backfill_runs() { + let test = "update_table_returns_while_the_backfill_runs"; + if base_conn().is_none() { + return skip(test); + } + let s = scratch().await; + let key_info = create_table(&s).await; + load_items(&s, &key_info).await; + let table_id = data_table_id(&s.db).await; + + hold_gsi_backfill_for_tests(&table_id, Some(0)); + let response = tokio::time::timeout( + Duration::from_secs(10), + s.engine.update_table(ACCOUNT, add_gidx()), + ) + .await + .expect("UpdateTable waited for the held backfill instead of returning") + .expect("add the GSI"); + assert_eq!(response.table_status, TableStatus::Active); + assert_eq!(gidx_status(&response), "CREATING"); + let described = describe(&s).await; + assert_eq!(described.table_status, TableStatus::Active); + assert_eq!(gidx_status(&described), "CREATING"); + let (_, index_id, status) = ids(&s.db).await; + assert_eq!(status, "CREATING"); + assert_eq!(index_row_count(&s.db, &index_id).await, 0); + + // Measured 2026-10-30: a Query or Scan naming the index while it backfills + // is a ValidationException with this text. + let read = s.engine.index_info_by_table_id(&table_id, "gidx").await; + assert!( + matches!( + &read, + Err(StorageError::Validation(msg)) + if msg == "Cannot read from backfilling global secondary index: gidx" + ), + "a read of the CREATING index answered {read:?}" + ); + + // Measured 2026-10-30: DeleteTable while the index is CREATING is a + // ResourceInUseException with this text, and the table stays. + let refused = s + .engine + .delete_table( + ACCOUNT, + DeleteTableInput { + table_name: TABLE.to_owned(), + }, + ) + .await; + assert!( + matches!( + &refused, + Err(StorageError::IndexesInUse(msg)) + if msg == "Attempt to change a resource which is still in use: Cannot delete \ + table while indexes are being created, updated, or deleted." + ), + "DeleteTable during the build answered {refused:?}" + ); + assert_eq!(describe(&s).await.table_status, TableStatus::Active); + + // Let the build write one page, then stop it again and write around it. + hold_gsi_backfill_for_tests(&table_id, Some(1)); + let rows = wait_index_rows_at_least(&s.db, &index_id, 256).await; + assert!( + rows >= 256, + "the first backfill page did not land: {rows} rows" + ); + assert_eq!(ids(&s.db).await.2, "CREATING"); + write_during_build(&s, &key_info).await; + + hold_gsi_backfill_for_tests(&table_id, None); + assert_eq!(wait_active(&s.db).await, "ACTIVE"); + assert_eq!(gidx_status(&describe(&s).await), "ACTIVE"); + assert_index_matches_base(&s.db, &table_id, &index_id).await; + s.engine + .index_info_by_table_id(&table_id, "gidx") + .await + .expect("the ACTIVE index resolves"); + + // Once the index is ACTIVE the table can be deleted. + s.engine + .delete_table( + ACCOUNT, + DeleteTableInput { + table_name: TABLE.to_owned(), + }, + ) + .await + .expect("DeleteTable after the build"); + s.cleanup().await; +} + +/// A server that dies partway through a backfill leaves the index `CREATING` +/// with a partial data table; the next server's recovery sweep finishes it, and +/// writes made in between are indexed once. +/// +/// The first server runs on a runtime of its own, on another thread, so that +/// dropping the runtime cancels its detached build the way a crash would, and +/// closes the session that holds the build's ownership lock. +#[tokio::test] +async fn a_restart_during_the_backfill_still_reaches_active() { + let test = "a_restart_during_the_backfill_still_reaches_active"; + if base_conn().is_none() { + return skip(test); + } + let s = scratch().await; + let key_info = create_table(&s).await; + load_items(&s, &key_info).await; + let table_id = data_table_id(&s.db).await; + + hold_gsi_backfill_for_tests(&table_id, Some(1)); + let (stop_tx, stop_rx) = std::sync::mpsc::channel::<()>(); + let url = s.url.clone(); + let first_server = std::thread::spawn(move || { + let rt = tokio::runtime::Builder::new_multi_thread() + .worker_threads(2) + .enable_all() + .build() + .expect("build the first server's runtime"); + rt.block_on(async { + let engine = PostgresEngine::new(&config(&url), REGION) + .await + .expect("open the first server") + .start_gsi_workers(); + let response = engine + .update_table(ACCOUNT, add_gidx()) + .await + .expect("add the GSI on the first server"); + assert_eq!(gidx_status(&response), "CREATING"); + }); + // The detached build keeps running on `rt` after `block_on` returns. + stop_rx.recv().expect("wait for the crash signal"); + rt.shutdown_timeout(Duration::from_secs(5)); + }); + + let (_, index_id, _) = wait_ids(&s.db).await; + let rows = wait_index_rows_at_least(&s.db, &index_id, 256).await; + assert!( + rows >= 256, + "the first backfill page did not land: {rows} rows" + ); + stop_tx.send(()).expect("signal the crash"); + first_server.join().expect("the first server's thread"); + assert_eq!(ids(&s.db).await.2, "CREATING"); + + hold_gsi_backfill_for_tests(&table_id, None); + write_during_build(&s, &key_info).await; + + // The dead server's ownership session is closed with its connection; the + // sweep skips the index until the server notices, so give it a few passes. + let mut finished = 0; + for _ in 0..100 { + finished = extenddb_storage_postgres::reconcile_creating_gsis(&s.engine) + .await + .expect("run the recovery sweep"); + if finished > 0 { + break; + } + tokio::time::sleep(Duration::from_millis(100)).await; + } + assert_eq!(finished, 1, "the recovery sweep did not finish the build"); + assert_eq!(ids(&s.db).await.2, "ACTIVE"); + assert_eq!(gidx_status(&describe(&s).await), "ACTIVE"); + assert_index_matches_base(&s.db, &table_id, &index_id).await; + s.cleanup().await; +} diff --git a/crates/storage-postgres/tests/gsi_build_slots.rs b/crates/storage-postgres/tests/gsi_build_slots.rs new file mode 100644 index 000000000..7ad47fafb --- /dev/null +++ b/crates/storage-postgres/tests/gsi_build_slots.rs @@ -0,0 +1,633 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 +//! Storage-level tests for the bound on GSI builds one engine runs at once. +//! +//! `UpdateTable` returns before its GSI build starts (issue #654), so a client +//! that adds an index to one table after another would otherwise start every +//! build at once. Each build takes one of the engine's build slots, the recovery +//! sweep's builds included. These tests give the engine one slot and check that +//! a second build waits for it without holding a connection, that the waiting +//! index answers as a `CREATING` index does, that both indexes get built, and +//! that the waiting build leaves the index alone when another server's recovery +//! sweep has built it in the meantime. The last test runs on an engine with its +//! own number of slots, `GSI_BUILD_SLOTS`, and starts one build more than that. +//! +//! Each test builds its own throwaway database, applies the shipped migrations +//! to it, and drops it when it passes. A failing test leaves its database behind +//! on purpose, named `eddb_gsis_*`, so the state that failed can be inspected. +//! +//! Requires `EXTENDDB_TEST_PG_CONNECTION_STRING`, a base URL with no database +//! component (for example `postgresql://postgres@127.0.0.1:5432`), pointing at a +//! server whose role may create and drop databases. Without it every test here +//! reports a skip and passes, the same convention the wire suites use. + +use std::collections::{BTreeMap, BTreeSet}; +use std::time::Duration; + +use extenddb_core::expression::ExpressionMaps; +use extenddb_core::types::{ + AttributeValue, BillingMode, CreateTableInput, Item, TableDescription, TableKeyInfo, + UpdateTableInput, +}; +use extenddb_storage::error::StorageError; +use extenddb_storage::{DataEngine, TableEngine}; +use extenddb_storage_postgres::{ + GSI_BUILD_SLOTS, PostgresConfig, PostgresEngine, hold_gsi_backfill_for_tests, + reconcile_creating_gsis, +}; +use sqlx::PgPool; +use sqlx::postgres::PgPoolOptions; + +const ACCOUNT: &str = "123456789012"; +const REGION: &str = "us-east-1"; +const FIRST: &str = "t_gsi_slots_first"; +const SECOND: &str = "t_gsi_slots_second"; +/// More than two backfill pages of 256 keys, so a build can be stopped between +/// pages. +const ITEMS: u32 = 600; + +struct Scratch { + engine: PostgresEngine, + db: PgPool, + admin: PgPool, + db_name: String, +} + +impl Scratch { + async fn cleanup(self) { + let Scratch { + engine, + db, + admin, + db_name, + } = self; + drop(engine); + db.close().await; + sqlx::query(&format!( + "DROP DATABASE IF EXISTS \"{db_name}\" WITH (FORCE)" + )) + .execute(&admin) + .await + .expect("drop the scratch database"); + admin.close().await; + } +} + +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()) +} + +fn shipped_migrations() -> Vec { + let root = std::path::Path::new(env!("CARGO_MANIFEST_DIR")); + let mut out = Vec::new(); + for dir in ["migrations", "data_migrations"] { + let mut files: Vec<_> = std::fs::read_dir(root.join(dir)) + .expect("read the migrations directory") + .map(|e| e.expect("directory entry").path()) + .filter(|p| p.extension().is_some_and(|e| e == "sql")) + .collect(); + files.sort(); + for f in files { + out.push(std::fs::read_to_string(&f).expect("read a migration file")); + } + } + out +} + +fn skip(test: &str) { + eprintln!( + "SKIP {test}: EXTENDDB_TEST_PG_CONNECTION_STRING is not set, so there is no PostgreSQL \ + to build a scratch catalog in." + ); +} + +/// A scratch database and an engine on it with one GSI build slot. +async fn scratch() -> Scratch { + scratch_with_slots(Some(1)).await +} + +/// A scratch database and an engine on it with `slots` GSI build slots, or with +/// the engine's own `GSI_BUILD_SLOTS` for `None`. +async fn scratch_with_slots(slots: Option) -> Scratch { + let base = base_conn().expect("caller checks base_conn() first"); + let db_name = format!("eddb_gsis_{}", uuid::Uuid::new_v4().simple())[..24].to_owned(); + let admin = PgPoolOptions::new() + .max_connections(1) + .connect(&format!("{base}/postgres")) + .await + .expect("connect to the postgres maintenance database"); + sqlx::query(&format!("CREATE DATABASE \"{db_name}\"")) + .execute(&admin) + .await + .expect("create the scratch database"); + let url = format!("{base}/{db_name}"); + let db = PgPoolOptions::new() + .max_connections(4) + .connect(&url) + .await + .expect("connect to the scratch database"); + // One scratch database stands in for both the catalog and the data database. + for sql in shipped_migrations() { + sqlx::raw_sql(&sql) + .execute(&db) + .await + .expect("apply a shipped migration"); + } + sqlx::query( + "INSERT INTO settings (key, value) VALUES ('control_plane_delay_seconds', '0') \ + ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value", + ) + .execute(&db) + .await + .expect("pin the control-plane delay to zero"); + sqlx::query("INSERT INTO accounts (account_id, account_name) VALUES ($1, $2)") + .bind(ACCOUNT) + .bind(format!("acct-{db_name}")) + .execute(&db) + .await + .expect("seed the account row"); + let config = PostgresConfig { + connection_string: url, + pool_size: 10, + max_item_size_bytes: 400_000, + }; + let engine = PostgresEngine::new(&config, REGION) + .await + .expect("open a PostgresEngine on the scratch database"); + let engine = match slots { + Some(n) => engine.with_gsi_build_slots_for_tests(n), + None => engine, + } + .start_gsi_workers(); + Scratch { + engine, + db, + admin, + db_name, + } +} + +/// A second engine on the scratch database, with the engine's own build slots, +/// standing in for another server process. It runs no background worker; a +/// test calls its recovery sweep directly. +async fn peer_engine(s: &Scratch) -> PostgresEngine { + let base = base_conn().expect("caller checks base_conn() first"); + let config = PostgresConfig { + connection_string: format!("{base}/{}", s.db_name), + pool_size: 10, + max_item_size_bytes: 400_000, + }; + PostgresEngine::new(&config, REGION) + .await + .expect("open a second PostgresEngine on the scratch database") +} + +/// A table of `ITEMS` items, every one carrying the index's key attributes. +async fn table_with_items(s: &Scratch, name: &str) -> String { + let mut input: CreateTableInput = serde_json::from_value(serde_json::json!({ + "TableName": name, + "KeySchema": [ + {"AttributeName": "pk", "KeyType": "HASH"}, + {"AttributeName": "sk", "KeyType": "RANGE"}, + ], + "AttributeDefinitions": [ + {"AttributeName": "pk", "AttributeType": "S"}, + {"AttributeName": "sk", "AttributeType": "N"}, + ], + })) + .expect("build the CreateTable input"); + input.billing_mode = Some(BillingMode::PayPerRequest); + s.engine + .create_table(ACCOUNT, input) + .await + .expect("create the table"); + let key_info: TableKeyInfo = s + .engine + .table_key_info(ACCOUNT, name) + .await + .expect("read the key info"); + for n in 0..ITEMS { + let it: Item = BTreeMap::from([ + ("pk".to_owned(), AttributeValue::S(format!("p{}", n % 5))), + ("sk".to_owned(), AttributeValue::N(n.to_string())), + ("g".to_owned(), AttributeValue::S(format!("g{}", n % 7))), + ("r".to_owned(), AttributeValue::N(n.to_string())), + ]); + s.engine + .put_item(&key_info, it, false, None, &ExpressionMaps::default(), None) + .await + .expect("put"); + } + sqlx::query_scalar("SELECT table_id FROM tables WHERE table_name = $1") + .bind(name) + .fetch_one(&s.db) + .await + .expect("look up the table id") +} + +/// The `UpdateTable` input that adds GSI `gidx` keyed on `g` (S) and `r` (N). +fn add_gidx(table: &str) -> UpdateTableInput { + serde_json::from_value(serde_json::json!({ + "TableName": table, + "AttributeDefinitions": [ + {"AttributeName": "g", "AttributeType": "S"}, + {"AttributeName": "r", "AttributeType": "N"}, + ], + "GlobalSecondaryIndexUpdates": [{"Create": { + "IndexName": "gidx", + "KeySchema": [ + {"AttributeName": "g", "KeyType": "HASH"}, + {"AttributeName": "r", "KeyType": "RANGE"}, + ], + "Projection": {"ProjectionType": "ALL"}, + }}], + })) + .expect("build the UpdateTable input") +} + +/// `IndexStatus` of `gidx` in a description, or `None` when it is not listed. +fn gidx_status(desc: &TableDescription) -> Option { + desc.global_secondary_indexes + .as_ref() + .and_then(|gsis| gsis.iter().find(|g| g.index_name == "gidx")) + .map(|g| g.index_status.clone()) +} + +/// `(index_id, index_status)` of `gidx` on `table` in the catalog. +async fn gidx_row(db: &PgPool, table: &str) -> (String, String) { + sqlx::query_as( + "SELECT i.index_id, i.index_status FROM indexes i JOIN tables t \ + ON t.table_id = i.table_id WHERE t.table_name = $1 AND i.index_name = 'gidx'", + ) + .bind(table) + .fetch_one(db) + .await + .expect("look up gidx") +} + +/// Poll the catalog status of `gidx` on `table` until it is ACTIVE, for at most +/// 30 s, and return the last status seen. +async fn wait_active(db: &PgPool, table: &str) -> String { + let mut status = String::new(); + for _ in 0..600 { + status = gidx_row(db, table).await.1; + if status == "ACTIVE" { + break; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + status +} + +async fn index_row_count(db: &PgPool, index_id: &str) -> i64 { + sqlx::query_scalar(&format!("SELECT count(*) FROM \"_ddb_{index_id}\"")) + .fetch_one(db) + .await + .expect("count the index rows") +} + +/// Wait until the backfill has written at least `n` rows, which with a hold +/// after the first page means the first page is in and the build has stopped. +async fn wait_index_rows_at_least(db: &PgPool, index_id: &str, n: i64) -> i64 { + let mut rows = 0; + for _ in 0..600 { + rows = index_row_count(db, index_id).await; + if rows >= n { + break; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + rows +} + +/// Wait up to 10 s for the scratch database to hold `n` granted advisory locks, +/// and return the last count. Build ownership is the only advisory lock these +/// tests take, one per index being built; a lock is released when its session +/// ends, which the server sees a moment after the owner drops it. +async fn wait_ownership_locks(s: &Scratch, n: i64) -> i64 { + let mut held = -1; + for _ in 0..200 { + held = sqlx::query_scalar( + "SELECT count(*) FROM pg_locks l JOIN pg_database d ON d.oid = l.database \ + WHERE l.locktype = 'advisory' AND l.granted AND d.datname = $1", + ) + .bind(&s.db_name) + .fetch_one(&s.db) + .await + .expect("count the advisory locks"); + if held == n { + break; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + held +} + +async fn wait_queue_empty(db: &PgPool) { + for _ in 0..500 { + let n: i64 = sqlx::query_scalar("SELECT count(*) FROM gsi_pending") + .fetch_one(db) + .await + .expect("count pending rows"); + if n == 0 { + return; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + panic!("the propagation queue did not drain"); +} + +/// The index holds exactly the base items, each once, with the projected item +/// equal to the base item (every item carries the key attributes, and the GSI +/// projects ALL). +async fn assert_index_matches_base(db: &PgPool, table_id: &str, index_id: &str) { + wait_queue_empty(db).await; + let expected: BTreeSet<(String, String, String, String)> = sqlx::query_as(&format!( + "SELECT pk, sk_n::text, item_data->'g'->>'S', item_data::text FROM \"_ddb_{table_id}\"" + )) + .fetch_all(db) + .await + .expect("read the base table") + .into_iter() + .collect(); + let indexed: BTreeSet<(String, String, String, String)> = sqlx::query_as(&format!( + "SELECT base_pk, base_sk_n::text, pk, item_data::text FROM \"_ddb_{index_id}\"" + )) + .fetch_all(db) + .await + .expect("read the index table") + .into_iter() + .collect(); + assert_eq!( + indexed, expected, + "the index rows differ from the base items" + ); + assert_eq!( + index_row_count(db, index_id).await, + expected.len() as i64, + "an item is indexed more than once" + ); +} + +/// Start the first table's build and stop it after its first page, so it holds +/// the engine's one build slot; then add the index to the second table, which +/// must answer at once. Returns the two tables' ids and index ids. +async fn first_build_holds_the_slot(s: &Scratch) -> (String, String, String, String) { + let first = table_with_items(s, FIRST).await; + let second = table_with_items(s, SECOND).await; + hold_gsi_backfill_for_tests(&first, Some(1)); + hold_gsi_backfill_for_tests(&second, Some(1)); + s.engine + .update_table(ACCOUNT, add_gidx(FIRST)) + .await + .expect("add the GSI to the first table"); + let first_index = gidx_row(&s.db, FIRST).await.0; + let rows = wait_index_rows_at_least(&s.db, &first_index, 256).await; + assert!( + rows >= 256, + "the first build's first page did not land: {rows} rows" + ); + + let response = tokio::time::timeout( + Duration::from_secs(10), + s.engine.update_table(ACCOUNT, add_gidx(SECOND)), + ) + .await + .expect("UpdateTable waited for a build slot") + .expect("add the GSI to the second table"); + assert_eq!(gidx_status(&response).as_deref(), Some("CREATING")); + let (second_index, status) = gidx_row(&s.db, SECOND).await; + assert_eq!(status, "CREATING"); + (first, first_index, second, second_index) +} + +/// With one build slot, the second table's build waits for the first to finish. +/// While it waits, its index is `CREATING` and refuses reads as a backfilling +/// index, it has written nothing, and it holds no ownership session; once the +/// first index is ACTIVE, its own task takes ownership and builds it. +/// +/// With the build task taking no slot, the second build writes its first page +/// while the first build holds the slot, and the row count check fails. +#[tokio::test] +async fn a_build_waits_for_a_free_slot_holding_no_session() { + let test = "a_build_waits_for_a_free_slot_holding_no_session"; + if base_conn().is_none() { + return skip(test); + } + let s = scratch().await; + let (first, first_index, second, second_index) = first_build_holds_the_slot(&s).await; + + let read = s.engine.index_info_by_table_id(&second, "gidx").await; + assert!( + matches!( + &read, + Err(StorageError::Validation(msg)) + if msg == "Cannot read from backfilling global secondary index: gidx" + ), + "a read of the waiting index answered {read:?}" + ); + // A build of the second index that had started would have written its first + // page in this time: the first build's took its 1 s wait and well under a + // second more. + tokio::time::sleep(Duration::from_secs(3)).await; + assert_eq!( + index_row_count(&s.db, &second_index).await, + 0, + "the second build ran while the first held the only slot" + ); + assert_eq!( + wait_ownership_locks(&s, 1).await, + 1, + "the waiting build holds an ownership session" + ); + + hold_gsi_backfill_for_tests(&first, None); + assert_eq!(wait_active(&s.db, FIRST).await, "ACTIVE"); + let rows = wait_index_rows_at_least(&s.db, &second_index, 256).await; + assert!( + rows >= 256, + "the second build did not start once the slot was free: {rows} rows" + ); + assert_eq!( + wait_ownership_locks(&s, 1).await, + 1, + "the second build runs without its ownership lock" + ); + hold_gsi_backfill_for_tests(&second, None); + assert_eq!(wait_active(&s.db, SECOND).await, "ACTIVE"); + assert_index_matches_base(&s.db, &first, &first_index).await; + assert_index_matches_base(&s.db, &second, &second_index).await; + s.cleanup().await; +} + +/// A recovery sweep that runs while the only slot is taken takes the waiting +/// index, which has no owner yet, and waits for the slot too. When the slot +/// frees, the index's own task finds the sweep owning it and leaves it, and the +/// sweep builds and publishes it. +/// +/// With the sweep taking no slot, it builds the index at once and the row count +/// check fails. With the waiting task building without ownership, it publishes +/// the index first and the sweep publishes nothing. +#[tokio::test] +async fn the_recovery_sweep_waits_for_a_free_slot() { + let test = "the_recovery_sweep_waits_for_a_free_slot"; + if base_conn().is_none() { + return skip(test); + } + let s = scratch().await; + let (first, first_index, second, second_index) = first_build_holds_the_slot(&s).await; + + let engine = s.engine.clone(); + let sweep = tokio::spawn(async move { reconcile_creating_gsis(&engine).await }); + assert_eq!( + wait_ownership_locks(&s, 2).await, + 2, + "the sweep did not take the waiting index" + ); + tokio::time::sleep(Duration::from_secs(3)).await; + assert_eq!( + index_row_count(&s.db, &second_index).await, + 0, + "the sweep built the index while the first build held the only slot" + ); + + hold_gsi_backfill_for_tests(&first, None); + assert_eq!(wait_active(&s.db, FIRST).await, "ACTIVE"); + let rows = wait_index_rows_at_least(&s.db, &second_index, 256).await; + assert!( + rows >= 256, + "the sweep's build did not start once the slot was free: {rows} rows" + ); + hold_gsi_backfill_for_tests(&second, None); + let swept = tokio::time::timeout(Duration::from_secs(60), sweep) + .await + .expect("the recovery sweep did not finish") + .expect("the sweep task") + .expect("the recovery sweep"); + assert_eq!(swept, 1, "the recovery sweep did not publish the index"); + assert_eq!(gidx_row(&s.db, SECOND).await.1, "ACTIVE"); + assert_index_matches_base(&s.db, &first, &first_index).await; + assert_index_matches_base(&s.db, &second, &second_index).await; + s.cleanup().await; +} + +/// A build that waited for its slot leaves its index alone when the slot comes +/// if another server's recovery sweep has built and published the index in the +/// meantime. While the build waits its index has no owner, so the other server's +/// sweep (a second engine on the same database) takes it; once that sweep has +/// let go of the ownership lock, the waiting task can take the lock, and it must +/// find the index ACTIVE rather than build it again. A hold at the index's build +/// entry makes a build that starts visible: it keeps its ownership lock while it +/// is held there. +/// +/// With the waiting task building without re-reading the index, it holds the +/// published index's ownership lock at that hold, and the lock count check fails. +#[tokio::test] +async fn a_waiting_build_leaves_an_index_a_peer_published() { + let test = "a_waiting_build_leaves_an_index_a_peer_published"; + if base_conn().is_none() { + return skip(test); + } + let s = scratch().await; + let peer = peer_engine(&s).await; + let (first, first_index, second, second_index) = first_build_holds_the_slot(&s).await; + assert_eq!( + wait_ownership_locks(&s, 1).await, + 1, + "the waiting build holds an ownership session" + ); + + hold_gsi_backfill_for_tests(&second, None); + let swept = reconcile_creating_gsis(&peer) + .await + .expect("the peer's recovery sweep"); + assert_eq!( + swept, 1, + "the peer's sweep did not publish the waiting index" + ); + assert_eq!(gidx_row(&s.db, SECOND).await.1, "ACTIVE"); + assert_eq!( + wait_ownership_locks(&s, 1).await, + 1, + "the peer's sweep kept the ownership lock of the index it published" + ); + + // A build of the second index that starts from here waits at its entry, + // holding the ownership lock. Releasing the first build frees the slot the + // second index's own task is waiting for. + hold_gsi_backfill_for_tests(&second, Some(0)); + hold_gsi_backfill_for_tests(&first, None); + assert_eq!(wait_active(&s.db, FIRST).await, "ACTIVE"); + tokio::time::sleep(Duration::from_secs(3)).await; + assert_eq!( + wait_ownership_locks(&s, 0).await, + 0, + "the waiting build took the published index's ownership lock to build it again" + ); + hold_gsi_backfill_for_tests(&second, None); + assert_index_matches_base(&s.db, &first, &first_index).await; + assert_index_matches_base(&s.db, &second, &second_index).await; + drop(peer); + s.cleanup().await; +} + +/// On an engine with its own number of build slots, the builds of +/// `GSI_BUILD_SLOTS` tables run, each holding its ownership lock, and the build +/// of one table more waits: its index has written nothing and has no owner. +/// +/// With the engine's slots unbounded, the last build runs too and writes its +/// first page, and the row count check fails. +#[tokio::test] +async fn an_engine_runs_at_most_gsi_build_slots_builds() { + let test = "an_engine_runs_at_most_gsi_build_slots_builds"; + if base_conn().is_none() { + return skip(test); + } + let s = scratch_with_slots(None).await; + let mut tables = Vec::new(); + for n in 0..=GSI_BUILD_SLOTS { + let name = format!("t_gsi_slots_{n}"); + let table_id = table_with_items(&s, &name).await; + hold_gsi_backfill_for_tests(&table_id, Some(1)); + tables.push((name, table_id)); + } + for (name, _) in &tables { + s.engine + .update_table(ACCOUNT, add_gidx(name)) + .await + .expect("add the GSI"); + } + for (name, _) in &tables[..GSI_BUILD_SLOTS] { + let index_id = gidx_row(&s.db, name).await.0; + let rows = wait_index_rows_at_least(&s.db, &index_id, 256).await; + assert!( + rows >= 256, + "the build on {name} did not start: {rows} rows" + ); + } + let last_index = gidx_row(&s.db, &tables[GSI_BUILD_SLOTS].0).await.0; + // A build of the last index that had started would have written its first + // page in this time, as the running builds did. + tokio::time::sleep(Duration::from_secs(3)).await; + assert_eq!( + index_row_count(&s.db, &last_index).await, + 0, + "a build ran beyond the engine's {GSI_BUILD_SLOTS} build slots" + ); + assert_eq!( + wait_ownership_locks(&s, GSI_BUILD_SLOTS as i64).await, + GSI_BUILD_SLOTS as i64, + "the ownership locks are not one per running build" + ); + + for (_, table_id) in &tables { + hold_gsi_backfill_for_tests(table_id, None); + } + for (name, table_id) in &tables { + assert_eq!(wait_active(&s.db, name).await, "ACTIVE"); + let index_id = gidx_row(&s.db, name).await.0; + assert_index_matches_base(&s.db, table_id, &index_id).await; + } + s.cleanup().await; +} diff --git a/crates/storage-postgres/tests/gsi_build_task.rs b/crates/storage-postgres/tests/gsi_build_task.rs new file mode 100644 index 000000000..f3b25e970 --- /dev/null +++ b/crates/storage-postgres/tests/gsi_build_task.rs @@ -0,0 +1,623 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 +//! Storage-level tests for the task that builds a GSI after `UpdateTable` has +//! returned (issue #654): what it leaves behind when the backfill fails, how it +//! shares an index with the recovery sweep, and what DeleteTable answers while +//! it runs. +//! +//! The client has had its 200 with the index `CREATING` before the task starts, +//! so a failed backfill no longer reaches it as an error. These tests check the +//! catalog and the data tables instead. +//! +//! Each test builds its own throwaway database, applies the shipped migrations +//! to it, and drops it when it passes. A failing test leaves its database behind +//! on purpose, named `eddb_gsit_*`, so the state that failed can be inspected. +//! +//! Requires `EXTENDDB_TEST_PG_CONNECTION_STRING`, a base URL with no database +//! component (for example `postgresql://postgres@127.0.0.1:5432`), pointing at a +//! server whose role may create and drop databases. Without it every test here +//! reports a skip and passes, the same convention the wire suites use. + +use std::collections::{BTreeMap, BTreeSet}; +use std::time::Duration; + +use extenddb_core::expression::ExpressionMaps; +use extenddb_core::types::{ + AttributeValue, BillingMode, CreateTableInput, DeleteTableInput, DescribeTableInput, Item, + TableDescription, TableKeyInfo, UpdateTableInput, +}; +use extenddb_storage::error::StorageError; +use extenddb_storage::{DataEngine, TableEngine}; +use extenddb_storage_postgres::{ + PostgresConfig, PostgresEngine, hold_gsi_backfill_for_tests, reconcile_creating_gsis, +}; +use sqlx::PgPool; +use sqlx::postgres::PgPoolOptions; + +const ACCOUNT: &str = "123456789012"; +const REGION: &str = "us-east-1"; +const TABLE: &str = "t_gsi_build_task"; +/// More than two backfill pages of 256 keys, so a build can be stopped between +/// pages. +const ITEMS: u32 = 600; +/// The service's DeleteTable refusal while an index is being created (measured +/// 2026-10-30). +const IN_USE: &str = "Attempt to change a resource which is still in use: Cannot delete table \ + while indexes are being created, updated, or deleted."; + +struct Scratch { + engine: PostgresEngine, + db: PgPool, + admin: PgPool, + db_name: String, +} + +impl Scratch { + async fn cleanup(self) { + let Scratch { + engine, + db, + admin, + db_name, + } = self; + drop(engine); + db.close().await; + sqlx::query(&format!( + "DROP DATABASE IF EXISTS \"{db_name}\" WITH (FORCE)" + )) + .execute(&admin) + .await + .expect("drop the scratch database"); + admin.close().await; + } +} + +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()) +} + +fn shipped_migrations() -> Vec { + let root = std::path::Path::new(env!("CARGO_MANIFEST_DIR")); + let mut out = Vec::new(); + for dir in ["migrations", "data_migrations"] { + let mut files: Vec<_> = std::fs::read_dir(root.join(dir)) + .expect("read the migrations directory") + .map(|e| e.expect("directory entry").path()) + .filter(|p| p.extension().is_some_and(|e| e == "sql")) + .collect(); + files.sort(); + for f in files { + out.push(std::fs::read_to_string(&f).expect("read a migration file")); + } + } + out +} + +fn skip(test: &str) { + eprintln!( + "SKIP {test}: EXTENDDB_TEST_PG_CONNECTION_STRING is not set, so there is no PostgreSQL \ + to build a scratch catalog in." + ); +} + +fn config(url: &str) -> PostgresConfig { + PostgresConfig { + connection_string: url.to_owned(), + pool_size: 10, + max_item_size_bytes: 400_000, + } +} + +async fn scratch() -> Scratch { + let base = base_conn().expect("caller checks base_conn() first"); + let db_name = format!("eddb_gsit_{}", uuid::Uuid::new_v4().simple())[..24].to_owned(); + let admin = PgPoolOptions::new() + .max_connections(1) + .connect(&format!("{base}/postgres")) + .await + .expect("connect to the postgres maintenance database"); + sqlx::query(&format!("CREATE DATABASE \"{db_name}\"")) + .execute(&admin) + .await + .expect("create the scratch database"); + let url = format!("{base}/{db_name}"); + let db = PgPoolOptions::new() + .max_connections(4) + .connect(&url) + .await + .expect("connect to the scratch database"); + // One scratch database stands in for both the catalog and the data database. + for sql in shipped_migrations() { + sqlx::raw_sql(&sql) + .execute(&db) + .await + .expect("apply a shipped migration"); + } + sqlx::query( + "INSERT INTO settings (key, value) VALUES ('control_plane_delay_seconds', '0') \ + ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value", + ) + .execute(&db) + .await + .expect("pin the control-plane delay to zero"); + sqlx::query("INSERT INTO accounts (account_id, account_name) VALUES ($1, $2)") + .bind(ACCOUNT) + .bind(format!("acct-{db_name}")) + .execute(&db) + .await + .expect("seed the account row"); + let engine = PostgresEngine::new(&config(&url), REGION) + .await + .expect("open a PostgresEngine on the scratch database") + .start_gsi_workers(); + Scratch { + engine, + db, + admin, + db_name, + } +} + +async fn create_table(s: &Scratch, name: &str, protected: bool) -> TableKeyInfo { + let mut input: CreateTableInput = serde_json::from_value(serde_json::json!({ + "TableName": name, + "KeySchema": [ + {"AttributeName": "pk", "KeyType": "HASH"}, + {"AttributeName": "sk", "KeyType": "RANGE"}, + ], + "AttributeDefinitions": [ + {"AttributeName": "pk", "AttributeType": "S"}, + {"AttributeName": "sk", "AttributeType": "N"}, + ], + "DeletionProtectionEnabled": protected, + })) + .expect("build the CreateTable input"); + input.billing_mode = Some(BillingMode::PayPerRequest); + s.engine + .create_table(ACCOUNT, input) + .await + .expect("create the table"); + s.engine + .table_key_info(ACCOUNT, name) + .await + .expect("read the key info") +} + +/// The `UpdateTable` input that adds GSI `gidx` keyed on `g` (S) and `r` (N). +fn add_gidx(table: &str) -> UpdateTableInput { + serde_json::from_value(serde_json::json!({ + "TableName": table, + "AttributeDefinitions": [ + {"AttributeName": "g", "AttributeType": "S"}, + {"AttributeName": "r", "AttributeType": "N"}, + ], + "GlobalSecondaryIndexUpdates": [{"Create": { + "IndexName": "gidx", + "KeySchema": [ + {"AttributeName": "g", "KeyType": "HASH"}, + {"AttributeName": "r", "KeyType": "RANGE"}, + ], + "Projection": {"ProjectionType": "ALL"}, + }}], + })) + .expect("build the UpdateTable input") +} + +fn item(n: u32, g: &str) -> Item { + BTreeMap::from([ + ("pk".to_owned(), AttributeValue::S(format!("p{}", n % 5))), + ("sk".to_owned(), AttributeValue::N(n.to_string())), + ("g".to_owned(), AttributeValue::S(g.to_owned())), + ("r".to_owned(), AttributeValue::N(n.to_string())), + ]) +} + +async fn load_items(s: &Scratch, key_info: &TableKeyInfo, count: u32) { + for n in 0..count { + s.engine + .put_item( + key_info, + item(n, &format!("g{}", n % 7)), + false, + None, + &ExpressionMaps::default(), + None, + ) + .await + .expect("put"); + } +} + +async fn describe(s: &Scratch, table: &str) -> TableDescription { + s.engine + .describe_table( + ACCOUNT, + DescribeTableInput { + table_name: table.to_owned(), + }, + ) + .await + .expect("describe the table") +} + +async fn delete_table(s: &Scratch, table: &str) -> Result { + s.engine + .delete_table( + ACCOUNT, + DeleteTableInput { + table_name: table.to_owned(), + }, + ) + .await +} + +/// `IndexStatus` of `gidx` in a description, or `None` when it is not listed. +fn gidx_status(desc: &TableDescription) -> Option { + desc.global_secondary_indexes + .as_ref() + .and_then(|gsis| gsis.iter().find(|g| g.index_name == "gidx")) + .map(|g| g.index_status.clone()) +} + +async fn table_id(db: &PgPool, table: &str) -> String { + sqlx::query_scalar("SELECT table_id FROM tables WHERE table_name = $1") + .bind(table) + .fetch_one(db) + .await + .expect("look up the table id") +} + +/// `(index_id, index_status)` of `gidx` on `table` in the catalog. +async fn gidx_row(db: &PgPool, table: &str) -> Option<(String, String)> { + sqlx::query_as( + "SELECT i.index_id, i.index_status FROM indexes i JOIN tables t \ + ON t.table_id = i.table_id WHERE t.table_name = $1 AND i.index_name = 'gidx'", + ) + .bind(table) + .fetch_optional(db) + .await + .expect("look up gidx") +} + +/// Poll the catalog status of `gidx` on `table` until it is `want` (`None`: the +/// row is gone), for at most 30 s, and return the last status seen. +async fn wait_status(db: &PgPool, table: &str, want: Option<&str>) -> Option { + let mut last = None; + for _ in 0..600 { + last = gidx_row(db, table).await.map(|r| r.1); + if last.as_deref() == want { + break; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + last +} + +async fn relation_exists(db: &PgPool, id: &str) -> bool { + sqlx::query_scalar("SELECT to_regclass($1) IS NOT NULL") + .bind(format!("\"_ddb_{id}\"")) + .fetch_one(db) + .await + .expect("look up the relation") +} + +async fn index_row_count(db: &PgPool, index_id: &str) -> i64 { + sqlx::query_scalar(&format!("SELECT count(*) FROM \"_ddb_{index_id}\"")) + .fetch_one(db) + .await + .expect("count the index rows") +} + +/// Wait until the backfill has written at least `n` rows, which with a hold +/// after the first page means the first page is in and the build has stopped. +async fn wait_index_rows_at_least(db: &PgPool, index_id: &str, n: i64) -> i64 { + let mut rows = 0; + for _ in 0..600 { + rows = index_row_count(db, index_id).await; + if rows >= n { + break; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + rows +} + +/// Granted advisory locks on the scratch database. Build ownership is the only +/// advisory lock these tests take, one per index being built. +async fn ownership_locks(s: &Scratch) -> i64 { + sqlx::query_scalar( + "SELECT count(*) FROM pg_locks l JOIN pg_database d ON d.oid = l.database \ + WHERE l.locktype = 'advisory' AND l.granted AND d.datname = $1", + ) + .bind(&s.db_name) + .fetch_one(&s.db) + .await + .expect("count the advisory locks") +} + +async fn wait_queue_empty(db: &PgPool) { + for _ in 0..500 { + let n: i64 = sqlx::query_scalar("SELECT count(*) FROM gsi_pending") + .fetch_one(db) + .await + .expect("count pending rows"); + if n == 0 { + return; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + panic!("the propagation queue did not drain"); +} + +/// The index holds exactly the base items that carry both key attributes, each +/// once, with the projected item equal to the base item (the GSI projects ALL). +async fn assert_index_matches_base(db: &PgPool, table_id: &str, index_id: &str) { + wait_queue_empty(db).await; + let expected: BTreeSet<(String, String, String, String)> = sqlx::query_as(&format!( + "SELECT pk, sk_n::text, item_data->'g'->>'S', item_data::text FROM \"_ddb_{table_id}\" \ + WHERE item_data ? 'g' AND item_data ? 'r'" + )) + .fetch_all(db) + .await + .expect("read the base table") + .into_iter() + .collect(); + let indexed: BTreeSet<(String, String, String, String)> = sqlx::query_as(&format!( + "SELECT base_pk, base_sk_n::text, pk, item_data::text FROM \"_ddb_{index_id}\"" + )) + .fetch_all(db) + .await + .expect("read the index table") + .into_iter() + .collect(); + assert_eq!( + indexed, expected, + "the index rows differ from the base items" + ); + assert_eq!( + index_row_count(db, index_id).await, + expected.len() as i64, + "an item is indexed more than once" + ); +} + +/// DeleteTable on a deletion-protected table answers the in-use refusal while +/// a GSI is being created, and the protection refusal before and after. +/// +/// Measured 2026-10-30 on a protected table: the protection ValidationException +/// with no index change running, and while the index was CREATING, before and +/// after `Backfilling` turned true, ResourceInUseException with [`IN_USE`]. With +/// the protection check ahead of the CREATING check, the second answer is the +/// protection refusal and this test fails there. +#[tokio::test] +async fn delete_table_of_a_protected_table_during_a_build_is_in_use() { + let test = "delete_table_of_a_protected_table_during_a_build_is_in_use"; + if base_conn().is_none() { + return skip(test); + } + let s = scratch().await; + let key_info = create_table(&s, TABLE, true).await; + load_items(&s, &key_info, 20).await; + let table_id = table_id(&s.db, TABLE).await; + + let before = delete_table(&s, TABLE).await; + assert!( + matches!(&before, Err(StorageError::DeletionProtected(_))), + "DeleteTable of the protected table before the build answered {before:?}" + ); + + hold_gsi_backfill_for_tests(&table_id, Some(0)); + let response = s + .engine + .update_table(ACCOUNT, add_gidx(TABLE)) + .await + .expect("add the GSI"); + assert_eq!(gidx_status(&response).as_deref(), Some("CREATING")); + let during = delete_table(&s, TABLE).await; + assert!( + matches!(&during, Err(StorageError::IndexesInUse(msg)) if msg == IN_USE), + "DeleteTable of the protected table during the build answered {during:?}" + ); + + hold_gsi_backfill_for_tests(&table_id, None); + assert_eq!( + wait_status(&s.db, TABLE, Some("ACTIVE")).await.as_deref(), + Some("ACTIVE") + ); + let after = delete_table(&s, TABLE).await; + assert!( + matches!(&after, Err(StorageError::DeletionProtected(_))), + "DeleteTable of the protected table after the build answered {after:?}" + ); + s.cleanup().await; +} + +/// A backfill that fails the same way on every attempt removes the index and +/// its data table after `UpdateTable` has answered with the index `CREATING`, +/// and once the cause is gone the same Create builds the index. +/// +/// With every failure treated as transient, the index stays `CREATING` and the +/// wait for its removal fails. +#[tokio::test] +async fn a_build_that_fails_every_time_removes_the_index() { + let test = "a_build_that_fails_every_time_removes_the_index"; + if base_conn().is_none() { + return skip(test); + } + let s = scratch().await; + let key_info = create_table(&s, TABLE, false).await; + load_items(&s, &key_info, ITEMS).await; + let table_id = table_id(&s.db, TABLE).await; + + // One stored item that no longer decodes, so the backfill fails on it every + // time. + let base = format!("\"_ddb_{table_id}\""); + let original: String = sqlx::query_scalar(&format!( + "SELECT item_data::text FROM {base} WHERE pk = 'p0' AND sk_n = 0" + )) + .fetch_one(&s.db) + .await + .expect("read the item to break"); + sqlx::query(&format!( + "UPDATE {base} SET item_data = $1::jsonb WHERE pk = 'p0' AND sk_n = 0" + )) + .bind(r#"{"pk":{"S":"p0"},"sk":{"N":"0"},"g":{"BOGUS":1}}"#) + .execute(&s.db) + .await + .expect("make one stored item undecodable"); + + let response = s + .engine + .update_table(ACCOUNT, add_gidx(TABLE)) + .await + .expect("UpdateTable answers before the backfill fails"); + assert_eq!(gidx_status(&response).as_deref(), Some("CREATING")); + let (index_id, _) = gidx_row(&s.db, TABLE).await.expect("the gidx row"); + + assert_eq!( + wait_status(&s.db, TABLE, None).await, + None, + "the index whose backfill fails every time was not removed" + ); + // The data table is dropped GRACE_PUBLISH (1 s) after the catalog row. + let mut exists = true; + for _ in 0..100 { + exists = relation_exists(&s.db, &index_id).await; + if !exists { + break; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + assert!(!exists, "the removed index's data table was left behind"); + assert_eq!(gidx_status(&describe(&s, TABLE).await), None); + + sqlx::query(&format!( + "UPDATE {base} SET item_data = $1::jsonb WHERE pk = 'p0' AND sk_n = 0" + )) + .bind(&original) + .execute(&s.db) + .await + .expect("restore the item"); + s.engine + .update_table(ACCOUNT, add_gidx(TABLE)) + .await + .expect("create the index again"); + assert_eq!( + wait_status(&s.db, TABLE, Some("ACTIVE")).await.as_deref(), + Some("ACTIVE") + ); + let (index_id, _) = gidx_row(&s.db, TABLE).await.expect("the new gidx row"); + assert_index_matches_base(&s.db, &table_id, &index_id).await; + s.cleanup().await; +} + +/// A backfill that fails for a passing reason leaves the index `CREATING` with +/// its data table, and its task ends, which frees the build's ownership lock; +/// the next recovery sweep publishes the index. +/// +/// With every failure treated as permanent, the index is removed instead and +/// the `CREATING` assertion fails. +#[tokio::test] +async fn a_build_that_fails_for_a_passing_reason_stays_creating() { + let test = "a_build_that_fails_for_a_passing_reason_stays_creating"; + if base_conn().is_none() { + return skip(test); + } + let s = scratch().await; + let key_info = create_table(&s, TABLE, false).await; + load_items(&s, &key_info, ITEMS).await; + let table_id = table_id(&s.db, TABLE).await; + + hold_gsi_backfill_for_tests(&table_id, Some(0)); + s.engine + .update_table(ACCOUNT, add_gidx(TABLE)) + .await + .expect("add the GSI"); + let (index_id, _) = gidx_row(&s.db, TABLE).await.expect("the gidx row"); + // An outside transaction holds the base table's ROW EXCLUSIVE lock, so every + // write barrier attempt times out with SQLSTATE 55P03, which is transient. + let mut blocker = s.db.begin().await.expect("begin the blocker"); + sqlx::query(&format!( + "LOCK TABLE \"_ddb_{table_id}\" IN ROW EXCLUSIVE MODE" + )) + .execute(&mut *blocker) + .await + .expect("take the blocking lock"); + hold_gsi_backfill_for_tests(&table_id, None); + + // The barrier's six attempts take about 20 s. + let mut held = ownership_locks(&s).await; + for _ in 0..80 { + if held == 0 { + break; + } + tokio::time::sleep(Duration::from_millis(500)).await; + held = ownership_locks(&s).await; + } + assert_eq!( + held, 0, + "the build kept its ownership lock after a transient failure" + ); + assert_eq!( + gidx_row(&s.db, TABLE).await.map(|r| r.1).as_deref(), + Some("CREATING") + ); + assert!( + relation_exists(&s.db, &index_id).await, + "the index's data table was dropped" + ); + + blocker.rollback().await.expect("release the blocker"); + let finished = reconcile_creating_gsis(&s.engine) + .await + .expect("run the recovery sweep"); + assert_eq!(finished, 1, "the recovery sweep did not finish the build"); + assert_index_matches_base(&s.db, &table_id, &index_id).await; + s.cleanup().await; +} + +/// The detached build holds its ownership lock from before the index row +/// commits until it ends, so a recovery sweep that runs during the build skips +/// the index rather than starting a second build of it. +/// +/// With the ownership dropped before the task starts, the sweep takes the index +/// and its own build stops at the same hold, so the sweep does not return in +/// time. +#[tokio::test] +async fn the_recovery_sweep_leaves_a_running_build_alone() { + let test = "the_recovery_sweep_leaves_a_running_build_alone"; + if base_conn().is_none() { + return skip(test); + } + let s = scratch().await; + let key_info = create_table(&s, TABLE, false).await; + load_items(&s, &key_info, ITEMS).await; + let table_id = table_id(&s.db, TABLE).await; + + hold_gsi_backfill_for_tests(&table_id, Some(1)); + s.engine + .update_table(ACCOUNT, add_gidx(TABLE)) + .await + .expect("add the GSI"); + let (index_id, _) = gidx_row(&s.db, TABLE).await.expect("the gidx row"); + let rows = wait_index_rows_at_least(&s.db, &index_id, 256).await; + assert!( + rows >= 256, + "the first backfill page did not land: {rows} rows" + ); + + let swept = + tokio::time::timeout(Duration::from_secs(5), reconcile_creating_gsis(&s.engine)).await; + assert!( + matches!(swept, Ok(Ok(0))), + "the recovery sweep did not skip the running build: {swept:?}" + ); + hold_gsi_backfill_for_tests(&table_id, None); + assert_eq!( + wait_status(&s.db, TABLE, Some("ACTIVE")).await.as_deref(), + Some("ACTIVE") + ); + assert_index_matches_base(&s.db, &table_id, &index_id).await; + s.cleanup().await; +} diff --git a/crates/storage-postgres/tests/gsi_online_build.rs b/crates/storage-postgres/tests/gsi_online_build.rs index 4a02d3147..4e98b24c9 100644 --- a/crates/storage-postgres/tests/gsi_online_build.rs +++ b/crates/storage-postgres/tests/gsi_online_build.rs @@ -201,7 +201,15 @@ async fn create_table(s: &Scratch) -> TableKeyInfo { .expect("read the key info") } -/// Add GSI `gidx` keyed on `g` (S) and, when `range`, `r` (N). +/// Add GSI `gidx` keyed on `g` (S) and, when `range`, `r` (N), and wait for the +/// backfill to finish. +/// +/// `UpdateTable` now returns while the index is still `CREATING` and the backfill +/// runs in a detached task (issue #654). The tests in this file check the state +/// the backfill leaves behind, so the helper waits for the index to reach +/// `ACTIVE` before returning, which is the point the old in-request build reached +/// before `UpdateTable` returned. A test that wants to observe the `CREATING` +/// phase drives it directly rather than through this helper. async fn add_gsi(s: &Scratch, range: bool) { let mut defs = vec![serde_json::json!({"AttributeName": "g", "AttributeType": "S"})]; let mut schema = vec![serde_json::json!({"AttributeName": "g", "KeyType": "HASH"})]; @@ -223,6 +231,11 @@ async fn add_gsi(s: &Scratch, range: bool) { .update_table(ACCOUNT, input) .await .expect("add the GSI"); + assert_eq!( + wait_named_active(&s.db, "gidx").await, + "ACTIVE", + "the detached GSI backfill did not reach ACTIVE" + ); } fn item(n: u32, g: Option<&str>, r: Option) -> Item { @@ -284,13 +297,30 @@ async fn expected_rows(db: &PgPool, table_id: &str) -> BTreeSet<(String, String, rows.into_iter().collect() } -/// Wait for the GSI to become ACTIVE. UpdateTable publishes it before returning -/// unless the build hit a transient failure, in which case the recovery worker -/// (not running in these tests) would; a short wait tells the two apart. +/// Wait for the GSI `gidx` to become ACTIVE. +/// +/// Since issue #654, `UpdateTable` returns while the index is still `CREATING` +/// and a detached task runs the backfill, which flips the index to `ACTIVE` when +/// it finishes. A build that hits a transient failure leaves the index +/// `CREATING` for the recovery worker, which does not run in these tests, so a +/// bounded wait that never reaches `ACTIVE` tells the two apart. async fn wait_active(db: &PgPool) -> String { + wait_named_active(db, "gidx").await +} + +/// [`wait_active`] for an index named other than `gidx`. +async fn wait_named_active(db: &PgPool, index_name: &str) -> String { let mut status = String::new(); - for _ in 0..100 { - status = ids(db).await.2; + for _ in 0..200 { + status = sqlx::query_scalar( + "SELECT index_status FROM indexes i JOIN tables t ON t.table_id = i.table_id \ + WHERE t.table_name = $1 AND i.index_name = $2", + ) + .bind(TABLE) + .bind(index_name) + .fetch_one(db) + .await + .expect("look up the index status"); if status == "ACTIVE" { break; } @@ -722,9 +752,10 @@ async fn a_creating_gsi_refuses_a_second_change_and_reads() { assert!( matches!( s.engine.index_info_by_table_id(&table_id, "gidx").await, - Err(extenddb_storage::error::StorageError::IndexNotFound(_)) + Err(extenddb_storage::error::StorageError::Validation(msg)) + if msg == "Cannot read from backfilling global secondary index: gidx" ), - "a read of a CREATING index resolves as no such index" + "a read of a CREATING index is refused as backfilling" ); sqlx::query("UPDATE indexes SET index_status = 'ACTIVE' WHERE index_id = $1") .bind(&index_id) @@ -753,6 +784,9 @@ async fn a_creating_gsi_refuses_a_second_change_and_reads() { .update_table(ACCOUNT, other) .await .expect("add a second GSI while none builds"); + // "other" also backfills in a detached task now; wait for it to finish so the + // only CREATING index below is the one this test marks by hand. + assert_eq!(wait_named_active(&s.db, "other").await, "ACTIVE"); sqlx::query("UPDATE indexes SET index_status = 'CREATING' WHERE index_id = $1") .bind(&index_id) .execute(&s.db) diff --git a/crates/storage-postgres/tests/gsi_stale_writer.rs b/crates/storage-postgres/tests/gsi_stale_writer.rs index 2791fbdfb..e7618d5c3 100644 --- a/crates/storage-postgres/tests/gsi_stale_writer.rs +++ b/crates/storage-postgres/tests/gsi_stale_writer.rs @@ -202,7 +202,13 @@ async fn create_table(s: &Scratch) -> TableKeyInfo { .expect("read the key info") } -/// Add GSI `gidx` keyed on `g` (S) and, when `range`, `r` (N). +/// Add GSI `gidx` keyed on `g` (S) and, when `range`, `r` (N), and wait for the +/// backfill to finish. +/// +/// Since issue #654 `UpdateTable` returns while the index is still `CREATING` +/// and the backfill runs in a detached task. This helper waits for `ACTIVE`, +/// which is where the old in-request build left the index before `UpdateTable` +/// returned, so the join points in these tests keep their original timing. async fn add_gsi(s: &Scratch, range: bool) { let mut defs = vec![serde_json::json!({"AttributeName": "g", "AttributeType": "S"})]; let mut schema = vec![serde_json::json!({"AttributeName": "g", "KeyType": "HASH"})]; @@ -224,6 +230,13 @@ async fn add_gsi(s: &Scratch, range: bool) { .update_table(ACCOUNT, input) .await .expect("add the GSI"); + for _ in 0..200 { + if ids(&s.db).await.2 == "ACTIVE" { + return; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + panic!("the detached GSI backfill did not reach ACTIVE"); } fn item(n: u32, g: Option<&str>, r: Option) -> Item { diff --git a/docs/design/04-component-storage.md b/docs/design/04-component-storage.md index f9ac92dab..5708419d1 100755 --- a/docs/design/04-component-storage.md +++ b/docs/design/04-component-storage.md @@ -696,17 +696,24 @@ When `UpdateTable` adds a new GSI to a table with existing data, the PostgreSQL separate databases, so no transaction can tell those writes the index exists. 3. Scans the base table in keyset pages (256 rows, `FOR SHARE SKIP LOCKED`, skipped rows taken one at a time afterwards), writing each item's current state into the index, then deletes index rows whose item is gone. -4. Sets the status to `ACTIVE`. A build that fails for a transient reason leaves the index `CREATING`, and a - background worker finishes it; one that would fail every time removes the index and fails the request. +4. Sets the status to `ACTIVE`. A build that fails for a transient reason (a conflict, a timeout, a lost + connection, an exhausted connection pool) leaves the index `CREATING`, and a background worker finishes it. + One that would fail every time is logged at error level and its index removed, so `DescribeTable` no longer + lists the index and the `Create` can be sent again. `UpdateTable` has answered before either happens. Queued GSI updates apply the item's state at apply time rather than the image the row carried, so a queue row applied late, twice, or after the backfill wrote a newer state cannot leave the index behind the table. The worker reads the base row under `FOR SHARE` for that, so a queued update and a concurrent write to the same item now wait for each other where before they did not meet. -`UpdateTable` returns after the build. While a GSI is `CREATING`, a Query or Scan naming it is answered as if the -table had no such index (the service's response), and a GSI Create, or a Delete of another GSI, on the table is -refused with LimitExceededException. A Delete of the building GSI and a throughput Update are not refused. DeleteTable stops the propagation queue for the table (a `vector_index_holds` row, which the +`UpdateTable` returns once step 1 has committed, with the index `CREATING`; steps 2 to 4 run in a task of their +own. The task first takes one of the server process's GSI build slots (`GSI_BUILD_SLOTS`, four, shared with the +background worker), waiting for one with no database connection open when all are taken, so a client that adds +indexes to many tables in a row does not start every build at once. While a GSI is `CREATING`, a Query or Scan +naming it is refused with ValidationException "Cannot read from backfilling global secondary index: ", the +service's answer in its backfilling phase; DeleteTable is refused with ResourceInUseException; and a GSI Create, +or a Delete of another GSI, on the table is refused with LimitExceededException. A Delete of the building GSI is not +refused. A DeleteTable that goes ahead stops the propagation queue for the table (a `vector_index_holds` row, which the claim query honours) and deletes its pending rows before it drops any data table, so a worker holding a claimed row and waiting on the base table cannot deadlock with the drop. diff --git a/docs/differences-from-dynamodb.md b/docs/differences-from-dynamodb.md index 0686d8445..49bc9b951 100755 --- a/docs/differences-from-dynamodb.md +++ b/docs/differences-from-dynamodb.md @@ -100,6 +100,10 @@ from a request sent to that service or backend. | Projected attributes across all indexes | At most 100 | Enforced on `CreateTable` (and so on `RestoreTableFromBackup`, which validates its target the same way) and on `UpdateTable`'s GSI `Create`: every `NonKeyAttributes` entry of every GSI and LSI is counted, so an attribute projected into two indexes counts twice. Over 100 is `ValidationException` ("Number of projected attributes in all indexes exceeds limit of 100, projected: "; the service's text was not captured) | | Index key validation on `UpdateItem` and transaction writes | A secondary-index key attribute must carry its declared scalar type and may not be empty. On `UpdateItem` a literal operand of the wrong type (`SET k = :v`, `ADD`, `DELETE`, `list_append` of literals) is refused on the statement, ahead of the `ConditionExpression`, naming the attribute and the index; a value computed from the item (an attribute reference, the fallback of `if_not_exists`) is refused after the condition passes with the message "The update expression attempted to update the secondary index key to unsupported type"; a key left empty is refused after the condition passes with the update-expression wording. In `TransactWriteItems` a wrong-typed key (table or index) on any member cancels the transaction with the first such member's `ValidationError` in position and `None` elsewhere; otherwise an empty index key on any member is a top-level `ValidationException`; otherwise members report their own conditions. On a `Put` member both checks come ahead of the member's condition. (Measured on the service.) | Matches on every backend. Two known shared differences: an empty index key supplied through the legacy `AttributeUpdates` parameter is reported with the update-expression wording where the service uses the `PutItem` wording naming the index; and a request whose own value exceeds 400 KB is refused for size by the service before any index-key check, where ExtendDB reports the index-key fault first. | | `Query` on a GSI with several RANGE attributes (MongoDB and Cassandra backends) | Orders results over every RANGE attribute in key-schema order and serves a key condition on each of them (equality on every one before the last, any operator on the last) | SQLite and PostgreSQL match the service. The MongoDB and Cassandra backends filter the first RANGE attribute of such an index only: a key condition on a later RANGE attribute is refused with `ValidationException` naming the attribute and the backend. MongoDB orders results by the first RANGE attribute then the base table key. Cassandra orders a page over every RANGE attribute then the base table key, and refuses a `Query` that carries an `ExclusiveStartKey` on such an index with `ValidationException` (the cursor compares the first RANGE attribute only). Every refusal the service gives for the condition's shape (one condition per key, condition count, missing HASH attribute, RANGE gap, non-key attribute, value type) is the same on every backend. | +| `TableStatus` while a GSI added by `UpdateTable` is built (PostgreSQL) | `UPDATING` in the `UpdateTable` response and until the index reports `Backfilling` true, then `ACTIVE` while the index stays `CREATING`. The change came 16 to 39 s after the call in six builds on four tables (measured 2026-10-30) | `ACTIVE` in the response and for the whole build | +| `Backfilling` member of a GSI added by `UpdateTable` (PostgreSQL) | `false` in the `UpdateTable` response and until 16 to 39 s after the call, then `true` until the index is `ACTIVE`, when the member goes away (measured 2026-10-30, DescribeTable every 1 or 2 s) | Never present | +| Query or Scan naming a GSI that `UpdateTable` is building (PostgreSQL) | `ValidationException` "The table does not have the specified index: " before the index reports `Backfilling` true, and on one table still 0 and 2 s after it did. "Cannot read from backfilling global secondary index: " about 45 s after `Backfilling` turned true on that table, and about two minutes after `UpdateTable` on two others (measured 2026-10-30) | `ValidationException` "Cannot read from backfilling global secondary index: " for as long as the index is `CREATING` | +| `UpdateTable` Delete of a GSI that is being built (PostgreSQL) | Before the index reports `Backfilling` true, `ResourceInUseException` "Attempt to change a resource which is still in use: Index creation is in resource allocation phase. Retry deletion during backfilling phase or when the index is active. Table: Index: ". After, accepted: the index is reported `DELETING` for about 5 s, and a Query naming it meanwhile gets the backfilling text (measured once, 2026-10-30) | Accepted at any point of the build. The index is removed at once: the `UpdateTable` response no longer lists it, and a Query naming it gets "The table does not have the specified index: " | ## Backup and Restore diff --git a/tests/rust/src/gsi_on_hash_only_table.rs b/tests/rust/src/gsi_on_hash_only_table.rs index ed4f34d46..17072d5b1 100644 --- a/tests/rust/src/gsi_on_hash_only_table.rs +++ b/tests/rust/src/gsi_on_hash_only_table.rs @@ -6,11 +6,11 @@ //! The backfill that copies existing rows into a new index reads the base table //! ordered by its key columns. Naming the sort-key columns unconditionally makes //! that query fail with `column "sk_s" does not exist` on a hash-only base -//! table, and because the backfill runs inside the index creation the whole -//! `UpdateTable` is rolled back and the caller sees `InternalServerError`. A -//! partition-key-only table is the simplest table DynamoDB allows, so this -//! covers the add end to end: the call succeeds, the index reaches ACTIVE, and a -//! row written before the index existed is visible through it afterwards. +//! table. The failure is not transient, so the build removes the index and it +//! never reaches ACTIVE. A partition-key-only table is the simplest table +//! DynamoDB allows, so this covers the add end to end: the call succeeds, the +//! index reaches ACTIVE, and a row written before the index existed is visible +//! through it afterwards. use crate::test_base::*; use aws_sdk_dynamodb::types::{ @@ -162,7 +162,7 @@ async fn gsi_can_be_added_to_a_table_with_no_sort_key() { err_msg(&e) ) }); - wait_for_active(c, &table).await; + wait_for_index_active(c, &table, "hashOnlyIdx").await; // The index must exist on the table description, not merely have been // accepted by UpdateTable. diff --git a/tests/rust/src/test_base.rs b/tests/rust/src/test_base.rs index cab50e07c..a7c521948 100755 --- a/tests/rust/src/test_base.rs +++ b/tests/rust/src/test_base.rs @@ -10,8 +10,8 @@ use aws_credential_types::provider::SharedCredentialsProvider; use aws_credential_types::Credentials; use aws_sdk_dynamodb::config::Region; use aws_sdk_dynamodb::types::{ - AttributeDefinition, AttributeValue, BillingMode, GlobalSecondaryIndex, KeySchemaElement, - KeyType, Projection, ProjectionType, ScalarAttributeType, TableStatus, + AttributeDefinition, AttributeValue, BillingMode, GlobalSecondaryIndex, IndexStatus, + KeySchemaElement, KeyType, Projection, ProjectionType, ScalarAttributeType, TableStatus, }; use aws_sdk_dynamodb::Client; use aws_smithy_http_client::tls; @@ -373,6 +373,27 @@ pub async fn wait_for_active(c: &Client, name: &str) { panic!("Table {name} did not become ACTIVE within 60s"); } +/// Poll DescribeTable until the table and its GSI `index` are both ACTIVE (up +/// to 60s). UpdateTable answers while a GSI it adds is still CREATING, and until +/// the backfill finishes the index refuses reads and the table refuses another +/// index change. +pub async fn wait_for_index_active(c: &Client, table: &str, index: &str) { + for _ in 0..300 { + if let Ok(resp) = c.describe_table().table_name(table).send().await { + if let Some(t) = resp.table() { + let index_active = t.global_secondary_indexes().iter().any(|g| { + g.index_name() == Some(index) && g.index_status() == Some(&IndexStatus::Active) + }); + if index_active && t.table_status() == Some(&TableStatus::Active) { + return; + } + } + } + tokio::time::sleep(Duration::from_millis(200)).await; + } + panic!("Index {index} on {table} did not become ACTIVE within 60s"); +} + /// Poll DescribeTable until the table no longer exists (up to 60s). pub async fn wait_for_deleted(c: &Client, name: &str) { for _ in 0..60 { diff --git a/tests/rust/src/update_table_attribute_definitions.rs b/tests/rust/src/update_table_attribute_definitions.rs index 677b87606..b861afa39 100644 --- a/tests/rust/src/update_table_attribute_definitions.rs +++ b/tests/rust/src/update_table_attribute_definitions.rs @@ -92,7 +92,7 @@ async fn add_gsi(table: &str, index: &str, attr: &str, extra_defs: Vec type map. diff --git a/tests/test_gsi_online_backfill.py b/tests/test_gsi_online_backfill.py index 34e17885e..9de18168d 100644 --- a/tests/test_gsi_online_backfill.py +++ b/tests/test_gsi_online_backfill.py @@ -206,9 +206,12 @@ def test_gsi_added_under_concurrent_writes_matches_table( # A sort key on an attribute the table had not defined before. defs.append({"AttributeName": "r", "AttributeType": "N"}) schema.append({"AttributeName": "r", "KeyType": "RANGE"}) - # UpdateTable returns when the build is done. The session client's default - # 60 s read timeout and retries would turn a slow build into a retried call - # that fails with ResourceInUse, so the control call uses a writer client. + # On SQLite, UpdateTable returns when the build is done; on PostgreSQL it + # returns with the index CREATING and the build runs after it. For SQLite, + # the session client's default 60 s read timeout and retries would turn a + # slow build into a retried call that fails with ResourceInUse, so the + # control call uses a writer client, and the test waits for the index to + # be ACTIVE on every backend. _client(endpoint_url).update_table( TableName=table, AttributeDefinitions=defs,