From 5a9eeef1200cd4961c817cbb0c1812e767b84636 Mon Sep 17 00:00:00 2001 From: Lee Hannigan <29015211+LeeroyHannigan@users.noreply.github.com> Date: Fri, 30 Oct 2026 22:35:41 +0000 Subject: [PATCH 1/2] fix(storage-postgres): return UpdateTable before the GSI backfill finishes The service answers UpdateTable at once with a new GSI CREATING and backfills in the background. On PostgreSQL, UpdateTable with a GSI Create spawned the index build and awaited it inside the request, so the response arrived only after every existing item was indexed, and a 20,000-item table exceeded boto3's default 60 s read timeout. The build task is now detached. UpdateTable returns once the CREATING index row is committed, and the task waits 1 s, takes a write barrier, backfills, and flips the row to ACTIVE. Writes already maintain a CREATING index, and the recovery worker already finishes a build that a restart left behind. A build that fails for a transient reason leaves the index CREATING for the recovery worker; one that would fail on every attempt is logged and its index removed. One server process runs at most GSI_BUILD_SLOTS (4) GSI builds at once, and the recovery worker's builds count toward them. UpdateTable never waits for a slot. A build that finds none waits in its task with no database session open, takes its ownership lock when a slot comes, and then builds only if the index is still CREATING, since the recovery worker of this or another server may have built it in the meantime. Without the limit, in a storage-level test, 30 UpdateTable calls in a row on 30 tables, each build held after its first page, raised the database's sessions from 11 to 42. A CREATING index answers as the service does in its backfilling phase. Query and Scan get ValidationException "Cannot read from backfilling global secondary index: " where they got "The table does not have the specified index". DeleteTable gets ResourceInUseException "Attempt to change a resource which is still in use: Cannot delete table while indexes are being created, updated, or deleted.", also on a table with deletion protection on; before, it deleted an unprotected table under the build. On the service, measured on 30 October 2026, Backfilling turned true 16 to 39 s after UpdateTable, and the index went ACTIVE about 16 minutes after the call on 2,000-item tables. The new tests gsi_async_build, gsi_build_task and gsi_build_slots hold a table's backfill through a test-hooks hook and cover the response during the build, writes made during it, a restart partway through, both failure outcomes, the ownership lock that keeps the recovery sweep off a running build, DeleteTable on a protected table, the build limit, and a waiting build whose index another server's recovery sweep published. They are added to the storage-level PostgreSQL step of the integration workflow. Storage tests and two wire tests that used an index right after UpdateTable now wait for it to be ACTIVE. The design doc, the gsi_build module doc and differences-from-dynamodb.md describe the detached build and the differences from the service that remain. --- .github/workflows/integration.yml | 4 +- crates/storage-postgres/src/data/ddl.rs | 30 +- crates/storage-postgres/src/data/gsi_build.rs | 152 ++++- crates/storage-postgres/src/delete_table.rs | 19 + crates/storage-postgres/src/lib.rs | 26 + crates/storage-postgres/src/update_table.rs | 125 +++- .../storage-postgres/tests/gsi_async_build.rs | 603 +++++++++++++++++ .../storage-postgres/tests/gsi_build_slots.rs | 630 ++++++++++++++++++ .../storage-postgres/tests/gsi_build_task.rs | 620 +++++++++++++++++ .../tests/gsi_online_build.rs | 50 +- .../tests/gsi_stale_writer.rs | 15 +- docs/design/04-component-storage.md | 17 +- docs/differences-from-dynamodb.md | 4 + tests/rust/src/gsi_on_hash_only_table.rs | 12 +- tests/rust/src/test_base.rs | 25 +- .../src/update_table_attribute_definitions.rs | 2 +- tests/test_gsi_online_backfill.py | 9 +- 17 files changed, 2252 insertions(+), 91 deletions(-) create mode 100644 crates/storage-postgres/tests/gsi_async_build.rs create mode 100644 crates/storage-postgres/tests/gsi_build_slots.rs create mode 100644 crates/storage-postgres/tests/gsi_build_task.rs diff --git a/.github/workflows/integration.yml b/.github/workflows/integration.yml index b1de36c0b..0a9cd31ae 100644 --- a/.github/workflows/integration.yml +++ b/.github/workflows/integration.yml @@ -323,8 +323,8 @@ jobs: run: >- cargo test --release -p extenddb-storage-postgres --test vector_control_plane --test put_create_race --test twi_conflict - --test stream_shard_retention --test stream_shard_ownership --test stream_records_page_order --test backup_restore - --test gsi_online_build --test gsi_stale_writer --test table_snapshot + --test stream_shard_retention --test stream_shard_ownership --test stream_records_page_order --test backup_restore --test gsi_async_build + --test gsi_build_task --test gsi_build_slots --test gsi_online_build --test gsi_stale_writer --test table_snapshot # The daemonized server logs to syslog; dump it so server-side failures # are diagnosable from the job log. diff --git a/crates/storage-postgres/src/data/ddl.rs b/crates/storage-postgres/src/data/ddl.rs index 16f1a3172..40d5500a7 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 f03d2c7f0..cb7a11d6a 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 9603d68ff..108c1144c 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; @@ -294,6 +303,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 @@ -427,6 +440,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(), @@ -475,6 +491,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..0d2ff6016 --- /dev/null +++ b/crates/storage-postgres/tests/gsi_async_build.rs @@ -0,0 +1,603 @@ +// 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("UPDATE settings SET value = '0' WHERE key = 'control_plane_delay_seconds'") + .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..171b11f6c --- /dev/null +++ b/crates/storage-postgres/tests/gsi_build_slots.rs @@ -0,0 +1,630 @@ +// 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("UPDATE settings SET value = '0' WHERE key = 'control_plane_delay_seconds'") + .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..901a7d57c --- /dev/null +++ b/crates/storage-postgres/tests/gsi_build_task.rs @@ -0,0 +1,620 @@ +// 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("UPDATE settings SET value = '0' WHERE key = 'control_plane_delay_seconds'") + .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 7fdb52b81..0c6535416 100644 --- a/crates/storage-postgres/tests/gsi_online_build.rs +++ b/crates/storage-postgres/tests/gsi_online_build.rs @@ -185,7 +185,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"})]; @@ -207,6 +215,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 { @@ -268,13 +281,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; } @@ -706,9 +736,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) @@ -737,6 +768,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 c79fe8cab..cd5e9b093 100644 --- a/crates/storage-postgres/tests/gsi_stale_writer.rs +++ b/crates/storage-postgres/tests/gsi_stale_writer.rs @@ -186,7 +186,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"})]; @@ -208,6 +214,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 dcfda3a3d..52a4203de 100755 --- a/docs/design/04-component-storage.md +++ b/docs/design/04-component-storage.md @@ -690,17 +690,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 6bd1a1607..fcf557b47 100755 --- a/docs/differences-from-dynamodb.md +++ b/docs/differences-from-dynamodb.md @@ -95,6 +95,10 @@ adaptation when switching between ExtendDB and the real service. | GSI throughput on a `PROVISIONED` table | Every GSI needs `ProvisionedThroughput` of at least 1 read and 1 write unit | A GSI `ProvisionedThroughput` below 1 (including 0/0) is `ValidationException` on `CreateTable` of a `PROVISIONED` table and on `UpdateTable`'s GSI `Create` ("ReadCapacityUnits and WriteCapacityUnits must both be greater than or equal to 1 for index: "; the service's text was not captured). A GSI with no `ProvisionedThroughput` is still accepted on a `PROVISIONED` table, and `UpdateTable` to `PROVISIONED` leaves an on-demand table's GSIs without one; see the `BillingModeOverride` row under Backup and Restore for how a backup of such a table restores | | 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) | | `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, From c05494bce1ce268b4b5d96218dee261372047e24 Mon Sep 17 00:00:00 2001 From: Lee <29015211+LeeroyHannigan@users.noreply.github.com> Date: Thu, 5 Nov 2026 02:09:46 +0000 Subject: [PATCH 2/2] test(storage-postgres): upsert the control-plane delay setting Main's migrations no longer seed the control_plane_delay_seconds row in settings, so the UPDATE that set it to zero in these tests changed no row. The engine kept its default delay, CreateTable left the table CREATING, and table_key_info then returned TableNotFound. The tests now insert the row or update it, as main's own tests do. --- crates/storage-postgres/tests/gsi_async_build.rs | 11 +++++++---- crates/storage-postgres/tests/gsi_build_slots.rs | 11 +++++++---- crates/storage-postgres/tests/gsi_build_task.rs | 11 +++++++---- 3 files changed, 21 insertions(+), 12 deletions(-) diff --git a/crates/storage-postgres/tests/gsi_async_build.rs b/crates/storage-postgres/tests/gsi_async_build.rs index 0d2ff6016..4dd3db0b1 100644 --- a/crates/storage-postgres/tests/gsi_async_build.rs +++ b/crates/storage-postgres/tests/gsi_async_build.rs @@ -130,10 +130,13 @@ async fn scratch() -> Scratch { .await .expect("apply a shipped migration"); } - sqlx::query("UPDATE settings SET value = '0' WHERE key = 'control_plane_delay_seconds'") - .execute(&db) - .await - .expect("pin the control-plane delay to zero"); + 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}")) diff --git a/crates/storage-postgres/tests/gsi_build_slots.rs b/crates/storage-postgres/tests/gsi_build_slots.rs index 171b11f6c..7ad47fafb 100644 --- a/crates/storage-postgres/tests/gsi_build_slots.rs +++ b/crates/storage-postgres/tests/gsi_build_slots.rs @@ -134,10 +134,13 @@ async fn scratch_with_slots(slots: Option) -> Scratch { .await .expect("apply a shipped migration"); } - sqlx::query("UPDATE settings SET value = '0' WHERE key = 'control_plane_delay_seconds'") - .execute(&db) - .await - .expect("pin the control-plane delay to zero"); + 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}")) diff --git a/crates/storage-postgres/tests/gsi_build_task.rs b/crates/storage-postgres/tests/gsi_build_task.rs index 901a7d57c..f3b25e970 100644 --- a/crates/storage-postgres/tests/gsi_build_task.rs +++ b/crates/storage-postgres/tests/gsi_build_task.rs @@ -134,10 +134,13 @@ async fn scratch() -> Scratch { .await .expect("apply a shipped migration"); } - sqlx::query("UPDATE settings SET value = '0' WHERE key = 'control_plane_delay_seconds'") - .execute(&db) - .await - .expect("pin the control-plane delay to zero"); + 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}"))