diff --git a/crates/app/src/cmd_settings.rs b/crates/app/src/cmd_settings.rs index 76c35070a..b1c0cece6 100755 --- a/crates/app/src/cmd_settings.rs +++ b/crates/app/src/cmd_settings.rs @@ -18,7 +18,9 @@ use extenddb_storage::management_store::SettingsStore; use extenddb_config::{self as config, MASK}; // Re-use validation constants from the ops layer. -use extenddb_server::management::ops_settings::{KNOWN_KEYS, READONLY_KEYS}; +use extenddb_server::management::ops_settings::{ + KNOWN_KEYS, READONLY_KEYS, writable_key_validator, +}; #[derive(Args)] pub struct SettingsArgs { @@ -118,13 +120,32 @@ async fn get(store: &dyn SettingsStore, key: &str, show_secret: bool) -> anyhow: } async fn set(store: &dyn SettingsStore, key: &str, value: &str) -> anyhow::Result<()> { + validate(key, value)?; + + // Write under the canonical name so the deprecated alias updates the row the read + // path consults, rather than adding a second one that is silently ignored. + let key = extenddb_core::settings_keys::canonical_key(key); + store + .set_setting(key, value) + .await + .map_err(|e| anyhow::anyhow!("Failed to set setting: {e:?}"))?; + + tracing::warn!( + target: "extenddb::audit::settings", + "settings-set: key={key}, value={value}", + ); + println!("{key} = {value}"); + Ok(()) +} + +/// Refuse a read-only or unknown key, or a value its validator rejects, with +/// the same lookup the management API uses. +fn validate(key: &str, value: &str) -> anyhow::Result<()> { if READONLY_KEYS.contains(&key) { anyhow::bail!("Setting '{key}' is read-only and cannot be changed via this command."); } - // Validate against known keys. - let known = KNOWN_KEYS.iter().find(|(k, _)| *k == key); - if let Some((_, validator)) = known { + if let Some(validator) = writable_key_validator(key) { validator(value) .map_err(|reason| anyhow::anyhow!("Invalid value for '{key}': {reason}"))?; } else { @@ -138,20 +159,6 @@ async fn set(store: &dyn SettingsStore, key: &str, value: &str) -> anyhow::Resul .join(", ") ); } - - // Write under the canonical name so the deprecated alias updates the row the read - // path consults, rather than adding a second one that is silently ignored. - let key = extenddb_core::settings_keys::canonical_key(key); - store - .set_setting(key, value) - .await - .map_err(|e| anyhow::anyhow!("Failed to set setting: {e:?}"))?; - - tracing::warn!( - target: "extenddb::audit::settings", - "settings-set: key={key}, value={value}", - ); - println!("{key} = {value}"); Ok(()) } @@ -236,4 +243,23 @@ mod tests { ); } } + + #[test] + fn a_table_gsi_phase_hold_is_writable_from_the_cli() { + // The admin guide lists `gsi_creating_phase_delay_ms:` among the + // settings managed with `extenddb settings set`. + validate("gsi_creating_phase_delay_ms:orders", "1000").expect("a valid hold"); + validate("gsi_creating_phase_delay_ms:orders", "0").expect("zero clears it"); + let err = + validate("gsi_creating_phase_delay_ms:orders", "60001").expect_err("above the bound"); + assert!( + err.to_string() + .starts_with("Invalid value for 'gsi_creating_phase_delay_ms:orders'"), + "{err}" + ); + let err = validate("gsi_creating_phase_delay_ms:ab", "1000") + .expect_err("a table name the service would refuse"); + assert!(err.to_string().starts_with("Unknown setting"), "{err}"); + assert!(!config::should_redact("gsi_creating_phase_delay_ms:orders")); + } } diff --git a/crates/core/src/settings_keys.rs b/crates/core/src/settings_keys.rs index 5380da38d..b58792e7c 100644 --- a/crates/core/src/settings_keys.rs +++ b/crates/core/src/settings_keys.rs @@ -101,6 +101,28 @@ pub const VECTOR_INDEX_MIN_CREATING_MS: &str = "vector_index_min_creating_ms"; /// Unset, nothing waits and no branch is taken. pub const VECTOR_ALLOCATION_PHASE_DELAY_MS: &str = "vector_allocation_phase_delay_ms"; +/// Test-only hold on the two `CREATING` phases of a GSI that `UpdateTable` adds. +/// +/// Keys take the form `gsi_creating_phase_delay_ms:` (see +/// [`gsi_creating_phase_delay_key`]) and the value is milliseconds. While it is +/// set for a table, a GSI added to that table waits that long after its catalog +/// row commits and before its backfill starts (`Backfilling: false`), then that +/// long again after the backfill has started (`Backfilling: true`). The phases +/// come from the build itself; the hold only lengthens them. On a test-sized +/// table both phases otherwise last milliseconds, so a client polling +/// DescribeTable could not observe them, and the measured Query and Scan refusals +/// for each phase could not be tested at the wire. +/// +/// Scoped to one table so a test that sets it does not slow the GSI builds of the +/// tests running beside it. Unset or zero, nothing waits. +pub const GSI_CREATING_PHASE_DELAY_MS: &str = "gsi_creating_phase_delay_ms"; + +/// The settings key holding [`GSI_CREATING_PHASE_DELAY_MS`] for one table. +#[must_use] +pub fn gsi_creating_phase_delay_key(table_name: &str) -> String { + format!("{GSI_CREATING_PHASE_DELAY_MS}:{table_name}") +} + /// Marker recording that the resource-tag condition-key migration has run on /// this catalog, or that the catalog was created after the migration existed /// and so has nothing to rewrite. diff --git a/crates/core/src/types/table.rs b/crates/core/src/types/table.rs index fdebcab30..935772757 100755 --- a/crates/core/src/types/table.rs +++ b/crates/core/src/types/table.rs @@ -292,6 +292,21 @@ pub struct GsiDescription { pub projection: Projection, #[serde(rename = "IndexStatus")] pub index_status: String, + /// Whether a GSI added by `UpdateTable` is still being populated. + /// + /// The same three states, in the same order, as + /// [`VectorIndexDescription::backfilling`], whose comment carries the + /// vector measurement. Measured for a GSI on 2026-10-26 (us-east-1): `false` + /// while the index is `CREATING` and its backfill has not started, `true` + /// while it runs, absent once the index is `ACTIVE`. ExtendDB reports the + /// member only for a GSI that `UpdateTable` added, and `None` serialises as + /// an absent member. + #[serde( + rename = "Backfilling", + skip_serializing_if = "Option::is_none", + default + )] + pub backfilling: Option, #[serde( rename = "ProvisionedThroughput", skip_serializing_if = "Option::is_none" @@ -737,7 +752,9 @@ pub const VECTOR_INDEX_SCAN_NOT_SUPPORTED: &str = "Scan operation not supported /// Prefix of the Scan refusal while a vector index is still backfilling; /// continues with the index name. The service reuses the GSI wording even -/// though the index is a vector index. Measured 2026-08-20. +/// though the index is a vector index. Measured 2026-08-20. A GSI that +/// `UpdateTable` is backfilling refuses both Query and Scan with this sentence +/// (measured 2026-10-26). pub const VECTOR_INDEX_BACKFILLING_SCAN_PREFIX: &str = "Cannot read from backfilling global secondary index: "; diff --git a/crates/engine/src/backup/tests.rs b/crates/engine/src/backup/tests.rs index f7bd86281..9ec54f9a3 100644 --- a/crates/engine/src/backup/tests.rs +++ b/crates/engine/src/backup/tests.rs @@ -205,6 +205,7 @@ fn described(request: &RestoreRequest) -> TableDescription { key_schema: g.key_schema.clone(), projection: g.projection.clone(), index_status: "ACTIVE".to_owned(), + backfilling: None, provisioned_throughput: Some(ProvisionedThroughputDescription { read_capacity_units: 9, write_capacity_units: 9, diff --git a/crates/engine/src/update_table.rs b/crates/engine/src/update_table.rs index 6db763f82..900d9e4e7 100755 --- a/crates/engine/src/update_table.rs +++ b/crates/engine/src/update_table.rs @@ -396,6 +396,7 @@ mod tests { key_schema: Vec::new(), projection: include(30), index_status: "ACTIVE".to_owned(), + backfilling: None, provisioned_throughput: None, index_size_bytes: 0, item_count: 0, @@ -406,6 +407,7 @@ mod tests { key_schema: Vec::new(), projection: include(50), index_status: "ACTIVE".to_owned(), + backfilling: None, provisioned_throughput: None, index_size_bytes: 0, item_count: 0, diff --git a/crates/server/src/management/ops_settings.rs b/crates/server/src/management/ops_settings.rs index 82cd8962b..d07806966 100755 --- a/crates/server/src/management/ops_settings.rs +++ b/crates/server/src/management/ops_settings.rs @@ -112,6 +112,19 @@ fn is_table_scoped_unknown_commit_gate_key(key: &str) -> bool { .is_some() } +/// `gsi_creating_phase_delay_ms:`, with a name the service would +/// accept for a table. +fn is_gsi_creating_phase_delay_key(key: &str) -> bool { + key.strip_prefix(extenddb_core::settings_keys::GSI_CREATING_PHASE_DELAY_MS) + .and_then(|rest| rest.strip_prefix(':')) + .is_some_and(|table| { + (3..=255).contains(&table.len()) + && table + .bytes() + .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'.' | b'_' | b'-')) + }) +} + fn validate_bool(value: &str) -> Result<(), &'static str> { match value { "true" | "false" => Ok(()), @@ -135,6 +148,26 @@ fn validate_index_propagation_delay_ms(value: &str) -> Result<(), &'static str> } } +/// The validator for a writable key: one of [`KNOWN_KEYS`], or a table's +/// `gsi_creating_phase_delay_ms:` hold. The management API and +/// `extenddb settings set` both look keys up here, so they accept the same +/// keys. +#[must_use] +pub fn writable_key_validator(key: &str) -> Option { + KNOWN_KEYS + .iter() + .find(|(k, _)| *k == key) + .map(|(_, validator)| *validator) + // A test lever for the same reason as the vector ones above: the two + // CREATING phases of a GSI build last milliseconds on a test-sized table, + // so the measured refusal for each cannot be observed from a client unless + // a test can lengthen them. Scoped to one table; same bound and validator. + .or_else(|| { + is_gsi_creating_phase_delay_key(key) + .then_some(validate_backfill_batch_delay_ms as Validator) + }) +} + /// Set a runtime setting with validation. /// /// Validates the key and value, then delegates the write to the @@ -154,10 +187,7 @@ pub async fn set_setting( return Err(OpError::Validation(format!("Setting '{key}' is read-only"))); } - let known = KNOWN_KEYS - .iter() - .find(|(k, _)| *k == key) - .map(|(_, validator)| *validator); + let known = writable_key_validator(key); #[cfg(feature = "mongodb-test-hooks")] let known = known .or_else(|| { @@ -195,3 +225,30 @@ pub async fn set_setting( ); Ok(()) } + +#[cfg(test)] +mod tests { + use super::is_gsi_creating_phase_delay_key; + + #[test] + fn a_gsi_phase_hold_is_writable_for_one_valid_table_name() { + assert!(is_gsi_creating_phase_delay_key( + "gsi_creating_phase_delay_ms:orders-2026.v1_x" + )); + assert!(!is_gsi_creating_phase_delay_key( + "gsi_creating_phase_delay_ms" + )); + assert!(!is_gsi_creating_phase_delay_key( + "gsi_creating_phase_delay_ms:" + )); + assert!(!is_gsi_creating_phase_delay_key( + "gsi_creating_phase_delay_ms:ab" + )); + assert!(!is_gsi_creating_phase_delay_key( + "gsi_creating_phase_delay_ms:has space" + )); + assert!(!is_gsi_creating_phase_delay_key( + "gsi_creating_phase_delay_msx:orders" + )); + } +} diff --git a/crates/storage-cassandra/src/create_table.rs b/crates/storage-cassandra/src/create_table.rs index 5628926b4..ad3c92757 100644 --- a/crates/storage-cassandra/src/create_table.rs +++ b/crates/storage-cassandra/src/create_table.rs @@ -518,6 +518,7 @@ impl CassandraEngine { key_schema: g.key_schema.clone(), projection: g.projection.clone(), index_status: "ACTIVE".to_owned(), + backfilling: None, provisioned_throughput: Some(ProvisionedThroughputDescription { read_capacity_units: g .provisioned_throughput diff --git a/crates/storage-cassandra/src/data/index.rs b/crates/storage-cassandra/src/data/index.rs index 4ebd72dac..1f853daa6 100644 --- a/crates/storage-cassandra/src/data/index.rs +++ b/crates/storage-cassandra/src/data/index.rs @@ -125,6 +125,12 @@ async fn fetch_indexes_for_table_at( /// Fetch a single index by `table_id` and `index_name` (hot path for query routing). /// /// Uses the `PRIMARY KEY ((table_id), index_name)` for efficient single-row lookup. +/// +/// A GSI still being built is not readable: its table is partial. The service +/// refuses a Query or Scan naming one in two ways, by phase (see +/// `extenddb_storage::gsi_backfill`). The recovery worker writes +/// `backfill_heartbeat` when it takes the index up for its backfill (see +/// `crate::gsi_backfill_phase`), so a heartbeat means the backfill has started. pub async fn fetch_index_by_name( table_id: &str, index_name: &str, @@ -132,7 +138,8 @@ pub async fn fetch_index_by_name( catalog_keyspace: &str, ) -> Result { let query = format!( - "SELECT index_id, index_type, key_schema, projection, propagation_delay_ms \ + "SELECT index_id, index_type, key_schema, projection, propagation_delay_ms, \ + index_status, backfill_heartbeat \ FROM {catalog_keyspace}.indexes WHERE table_id = ? AND index_name = ?" ); @@ -149,6 +156,15 @@ pub async fn fetch_index_by_name( .next() .ok_or_else(|| StorageError::IndexNotFound(format!("Index {index_name} not found")))?; + let index_status: Option = row.get_by_name("index_status").ok().flatten(); + if index_status.as_deref() == Some("CREATING") { + let heartbeat: Option = row.get_by_name("backfill_heartbeat").ok().flatten(); + return Err(extenddb_storage::gsi_backfill::creating_read_refusal( + index_name, + heartbeat.is_some(), + )); + } + let index_id: String = get_column(&row, "index_id", "fetch_index_by_name")?; let index_type: String = get_column(&row, "index_type", "fetch_index_by_name")?; let key_schema_text: String = get_column(&row, "key_schema", "fetch_index_by_name")?; diff --git a/crates/storage-cassandra/src/gsi_backfill_phase.rs b/crates/storage-cassandra/src/gsi_backfill_phase.rs new file mode 100644 index 000000000..d0132cefa --- /dev/null +++ b/crates/storage-cassandra/src/gsi_backfill_phase.rs @@ -0,0 +1,94 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! The start of a GSI backfill, as the catalog records it. +//! +//! DescribeTable reports `Backfilling` for a `CREATING` GSI as whether its row +//! carries a `backfill_heartbeat`, and a Query or Scan naming it is refused by +//! the same test (see `extenddb_storage::gsi_backfill`). `UpdateTable` inserts +//! the row with no heartbeat. The recovery worker writes one when it takes the +//! index up, before it rebuilds the data table and starts the backfill, and the +//! backfill refreshes it after every page. Written before the backfill task +//! starts, it also tells a later scan of the recovery worker that the index is +//! taken, so the scan does not rebuild it a second time. + +use std::time::Duration; + +use cdrs_tokio::types::IntoRustByName as _; +use extenddb_storage::error::StorageError; + +use crate::engine::CassandraEngine; + +/// Record that the backfill of one index is starting, after the table's +/// `gsi_creating_phase_delay_ms` test hold, and return the hold so the +/// backfill task can wait it again once the start is recorded. +/// +/// The recovery worker is one loop, so the first wait also delays the indexes +/// queued behind this one; it is zero outside tests. +/// +/// Returns `None` when there is nothing to build: the row is no longer +/// `CREATING`, or it could not be read. The heartbeat is written only to a row +/// read as `CREATING` immediately before, because a CQL `UPDATE` of a deleted +/// row would recreate it in part. +pub(crate) async fn claim( + engine: &CassandraEngine, + table_id: &str, + index_name: &str, + table_name: &str, +) -> Option { + let catalog_ks = engine.catalog_keyspace(); + let session = engine.session_arc(); + let hold = creating_phase_delay(engine, table_name).await; + tokio::time::sleep(hold).await; + let row = crate::cassandra_util::query_optional::( + &session, + &format!( + "SELECT index_status FROM {catalog_ks}.indexes WHERE table_id = ? AND index_name = ?" + ), + cdrs_tokio::query_values!(table_id, index_name), + "gsi_backfill_claim_check", + ) + .await; + let status: Option = match row { + Ok(row) => row.and_then(|r| r.get_by_name("index_status").ok().flatten()), + Err(e) => { + tracing::warn!("gsi_recovery: re-read '{index_name}' before its backfill: {e}"); + return None; + } + }; + if status.as_deref() != Some("CREATING") { + return None; + } + let claimed = crate::cassandra_util::execute::( + &session, + &format!( + "UPDATE {catalog_ks}.indexes SET backfill_heartbeat = toTimestamp(now()) \ + WHERE table_id = ? AND index_name = ?" + ), + cdrs_tokio::query_values!(table_id, index_name), + "gsi_backfill_claim", + ) + .await; + if let Err(e) = claimed { + // The backfill still runs; its first page writes the heartbeat. + tracing::warn!("gsi_recovery: record the backfill start of '{index_name}': {e}"); + } + Some(hold) +} + +/// The `gsi_creating_phase_delay_ms` test hold for one table; zero when unset. +async fn creating_phase_delay(engine: &CassandraEngine, table_name: &str) -> Duration { + let catalog_ks = engine.catalog_keyspace(); + let key = extenddb_core::settings_keys::gsi_creating_phase_delay_key(table_name); + let row = crate::cassandra_util::query_optional::( + &engine.session_arc(), + &format!("SELECT value FROM {catalog_ks}.settings WHERE key = ?"), + cdrs_tokio::query_values!(key.as_str()), + "gsi_creating_phase_delay", + ) + .await + .ok() + .flatten(); + let value: Option = row.and_then(|r| r.get_by_name("value").ok().flatten()); + extenddb_storage::gsi_backfill::phase_delay(value.as_deref()) +} diff --git a/crates/storage-cassandra/src/lib.rs b/crates/storage-cassandra/src/lib.rs index 7dbfe98c4..ed25abab9 100644 --- a/crates/storage-cassandra/src/lib.rs +++ b/crates/storage-cassandra/src/lib.rs @@ -21,6 +21,7 @@ mod credential_store; pub mod data; mod delete_table; pub mod engine; +mod gsi_backfill_phase; pub mod gsi_queue; mod management_store; mod metadata_engine; diff --git a/crates/storage-cassandra/src/table_helpers.rs b/crates/storage-cassandra/src/table_helpers.rs index 94b6eb0cf..6a8b21418 100644 --- a/crates/storage-cassandra/src/table_helpers.rs +++ b/crates/storage-cassandra/src/table_helpers.rs @@ -218,11 +218,19 @@ impl CassandraEngine { if index_type == "GSI" { let pt_str: Option = row.get_r_by_name("provisioned_throughput").ok(); + // The recovery worker writes the heartbeat when it takes the + // index up for its backfill, so a heartbeat on a CREATING index + // means the backfill has started. + let heartbeat: Option = row.get_by_name("backfill_heartbeat").ok().flatten(); gsis.push(GsiDescription { index_name: index_name.clone(), key_schema: ks, projection: proj, + backfilling: extenddb_storage::gsi_backfill::backfilling_member( + &index_status, + heartbeat.is_some(), + ), index_status, provisioned_throughput: gsi_throughput_description(pt_str.as_deref()), index_size_bytes: 0, diff --git a/crates/storage-cassandra/src/workers.rs b/crates/storage-cassandra/src/workers.rs index 520175ed6..f109b944c 100644 --- a/crates/storage-cassandra/src/workers.rs +++ b/crates/storage-cassandra/src/workers.rs @@ -470,17 +470,34 @@ pub(crate) async fn poll_gsi_backfill_recovery( }; for account_ks in &keyspaces { - if let Err(e) = recover_stale_backfills(&engine, account_ks, stale_threshold).await { + if let Err(e) = + recover_stale_backfills(&engine, account_ks, stale_threshold, None).await + { tracing::warn!("gsi_recovery: {account_ks}: {e}"); } } } } +/// One scan of the recovery worker, limited to the `CREATING` indexes of one +/// table. Hidden because no deployment path calls it: a direct test drives the +/// worker's claim with it, and the table filter keeps the scan off the indexes +/// the tests running beside it stage. +#[doc(hidden)] +pub async fn recover_stale_gsi_backfills_of_table( + engine: &Arc, + account_ks: &str, + table_id: &str, + stale_threshold: Duration, +) -> Result<(), extenddb_storage::error::StorageError> { + recover_stale_backfills(engine, account_ks, stale_threshold, Some(table_id)).await +} + async fn recover_stale_backfills( engine: &Arc, account_ks: &str, stale_threshold: Duration, + only_table: Option<&str>, ) -> Result<(), extenddb_storage::error::StorageError> { use cdrs_tokio::types::IntoRustByName as _; @@ -513,6 +530,9 @@ async fn recover_stale_backfills( } let table_id: String = crate::cassandra_util::get_column(&row, "table_id", "gsi_recovery")?; + if only_table.is_some_and(|t| t != table_id) { + continue; + } let index_name: String = crate::cassandra_util::get_column(&row, "index_name", "gsi_recovery")?; let index_id: String = crate::cassandra_util::get_column(&row, "index_id", "gsi_recovery")?; @@ -537,12 +557,12 @@ async fn recover_stale_backfills( } }; - // Fetch base table key_schema + attribute_definitions. + // Fetch the base table's account, key_schema and attribute_definitions. let base_row = match crate::cassandra_util::query_optional::( &engine.session, &format!( - "SELECT key_schema, attribute_definitions \ + "SELECT account_id, key_schema, attribute_definitions, table_name \ FROM {catalog_ks}.tables WHERE table_id = ? ALLOW FILTERING" ), cdrs_tokio::query_values!(table_id.as_str()), @@ -563,6 +583,17 @@ async fn recover_stale_backfills( } }; + // The catalog scan returns the CREATING indexes of every account, and + // the worker runs it once per account keyspace. Only the scan of the + // keyspace that holds the base table can build the index, so the + // others leave the row alone: a claim written by one of them would + // make the scan that can build the index find it taken and skip it. + let owner: String = + crate::cassandra_util::get_column(&base_row, "account_id", "gsi_recovery_base")?; + if engine.account_keyspace(&owner) != account_ks { + continue; + } + let base_ks_json: String = crate::cassandra_util::get_column(&base_row, "key_schema", "gsi_recovery_base")?; let base_ad_json: String = crate::cassandra_util::get_column( @@ -591,6 +622,14 @@ async fn recover_stale_backfills( "gsi_recovery: recovering stale backfill for index '{index_name}' \ (table_id={table_id}, index_id={index_id})" ); + let base_table_name: String = + crate::cassandra_util::get_column(&base_row, "table_name", "gsi_recovery_base")?; + let Some(hold) = + crate::gsi_backfill_phase::claim(engine, &table_id, &index_name, &base_table_name) + .await + else { + continue; + }; // Drop and recreate the data table so we start from a clean slate. if let Err(e) = engine.drop_index_data_table(account_ks, &index_id).await { @@ -635,6 +674,7 @@ async fn recover_stale_backfills( let index_id_owned = index_id.clone(); let index_name_owned = index_name.clone(); tokio::spawn(async move { + tokio::time::sleep(hold).await; let result = crate::data::ddl::backfill_gsi( &session, &account_ks_owned, diff --git a/crates/storage-cassandra/tests/direct/gsi_backfilling.rs b/crates/storage-cassandra/tests/direct/gsi_backfilling.rs new file mode 100644 index 000000000..c900a4552 --- /dev/null +++ b/crates/storage-cassandra/tests/direct/gsi_backfilling.rs @@ -0,0 +1,459 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! A `CREATING` GSI reports `Backfilling` and refuses reads by phase, as +//! measured on 2026-10-26 (see `extenddb_storage::gsi_backfill`). +//! +//! These direct tests run no recovery worker, so the first test puts the +//! catalog row in each state by hand, the way the worker leaves it: `CREATING` +//! with no `backfill_heartbeat` as `UpdateTable` inserts it, `CREATING` with +//! one once the worker has taken the index up, then `ACTIVE` with the heartbeat +//! left in place. The others drive single scans of the worker on their tables. + +#[cfg(test)] +mod tests { + use std::sync::Arc; + use std::time::{Duration, Instant}; + + use cdrs_tokio::query::QueryValues; + use cdrs_tokio::types::IntoRustByName as _; + use extenddb_core::types::{ + AttributeDefinition, AttributeValue, CreateGsiAction, DescribeTableInput, + GlobalSecondaryIndexUpdate, KeySchemaElement, KeyType, Projection, ProjectionType, + ScalarAttributeType, TableKeyInfo, UpdateTableInput, + }; + use extenddb_storage::error::StorageError; + use extenddb_storage::{DataEngine, TableEngine}; + use extenddb_storage_cassandra::CassandraEngine; + use extenddb_storage_cassandra::cassandra_util::{execute, query_lwt, query_optional}; + use extenddb_storage_cassandra::workers::recover_stale_gsi_backfills_of_table; + use uuid::Uuid; + + const BACKFILLING: &str = "Cannot read from backfilling global secondary index: gidx"; + + async fn set_row(engine: &CassandraEngine, table_id: &str, assignments: &str) { + let catalog_ks = engine.catalog_keyspace(); + execute::( + &engine.session_arc(), + &format!( + "UPDATE {catalog_ks}.indexes SET {assignments} \ + WHERE table_id = ? AND index_name = 'gidx'" + ), + cdrs_tokio::query_values!(table_id), + "gsi_backfilling_test", + ) + .await + .expect("set the index row"); + } + + /// The serialised `IndexStatus` and `Backfilling` members of `gidx`. + async fn described( + engine: &CassandraEngine, + account_id: &str, + table_name: &str, + ) -> (String, Option) { + let desc = engine + .describe_table( + account_id, + DescribeTableInput { + table_name: table_name.to_owned(), + }, + ) + .await + .expect("describe the table"); + let gsi = desc + .global_secondary_indexes + .expect("GSIs") + .into_iter() + .find(|g| g.index_name == "gidx") + .expect("gidx"); + let wire = serde_json::to_value(&gsi).expect("serialise the GSI description"); + (gsi.index_status, wire.get("Backfilling").cloned()) + } + + #[tokio::test] + async fn a_creating_gsi_reports_each_phase_and_refuses_reads_by_phase() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = crate::helpers::setup_engine().await; + let table = crate::helpers::TestTable::with_gsi(&engine, "GsiPhases", "gidx", "g").await; + let ki = &table.key_info; + + set_row( + &engine, + &ki.table_id, + "index_status = 'CREATING', backfill_heartbeat = null", + ) + .await; + assert_eq!( + described(&engine, &ki.account_id, &ki.table_name).await, + ("CREATING".to_owned(), Some(serde_json::Value::Bool(false))) + ); + let read = engine.index_info_by_table_id(&ki.table_id, "gidx").await; + assert!( + matches!(read, Err(StorageError::IndexNotFound(_))), + "before the backfill a read is told there is no such index: {read:?}" + ); + + set_row( + &engine, + &ki.table_id, + "backfill_heartbeat = toTimestamp(now())", + ) + .await; + assert_eq!( + described(&engine, &ki.account_id, &ki.table_name).await, + ("CREATING".to_owned(), Some(serde_json::Value::Bool(true))) + ); + let read = engine.index_info_by_table_id(&ki.table_id, "gidx").await; + assert!( + matches!(read, Err(StorageError::Validation(ref m)) if m == BACKFILLING), + "during the backfill a read gets the backfilling sentence: {read:?}" + ); + + set_row(&engine, &ki.table_id, "index_status = 'ACTIVE'").await; + assert_eq!( + described(&engine, &ki.account_id, &ki.table_name).await, + ("ACTIVE".to_owned(), None) + ); + engine + .index_info_by_table_id(&ki.table_id, "gidx") + .await + .expect("an ACTIVE index resolves"); + } + + /// One `i64` column, by its alias, from the first row of a query. + async fn scalar(engine: &CassandraEngine, cql: &str, values: QueryValues) -> Option { + query_optional::(&engine.session_arc(), cql, values, "gsi_claim_test") + .await + .expect("query") + .and_then(|row| row.get_by_name("v").ok().flatten()) + } + + /// The `index_id` of `gidx` on one table. + async fn index_id(engine: &CassandraEngine, table_id: &str) -> String { + let catalog_ks = engine.catalog_keyspace(); + query_optional::( + &engine.session_arc(), + &format!( + "SELECT index_id FROM {catalog_ks}.indexes WHERE table_id = ? AND index_name = 'gidx'" + ), + cdrs_tokio::query_values!(table_id), + "gsi_claim_test", + ) + .await + .expect("read the index row") + .and_then(|row| row.get_by_name("index_id").ok().flatten()) + .expect("index id") + } + + /// The id and the write time of the schema row of `gidx`'s data table. + async fn data_table(engine: &CassandraEngine, account_ks: &str, table: &str) -> (Uuid, i64) { + let row = query_optional::( + &engine.session_arc(), + "SELECT id, WRITETIME(comment) AS v FROM system_schema.tables \ + WHERE keyspace_name = ? AND table_name = ?", + cdrs_tokio::query_values!(account_ks, table), + "gsi_claim_test", + ) + .await + .expect("read the schema row") + .expect("the data table exists"); + ( + row.get_by_name("id").ok().flatten().expect("table id"), + row.get_by_name("v").ok().flatten().expect("write time"), + ) + } + + /// The recovery worker's claim. `UpdateTable` leaves a new index with no + /// heartbeat; the worker writes one before it rebuilds the data table and + /// starts the backfill, which is the start of the second phase, and a + /// later scan that finds the heartbeat fresh leaves the index to the build + /// already running. The table's phase hold keeps that build waiting while + /// the second scan runs. + #[tokio::test] + async fn the_recovery_claim_starts_the_second_phase_and_a_later_scan_leaves_the_build() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = Arc::new(crate::helpers::setup_engine().await); + let table = crate::helpers::TestTable::with_gsi(&engine, "GsiClaim", "gidx", "g").await; + let ki = &table.key_info; + let catalog_ks = engine.catalog_keyspace(); + let account_ks = engine.account_keyspace(&ki.account_id); + for n in 0..3 { + let item = std::collections::BTreeMap::from([ + ("id".to_owned(), AttributeValue::S(format!("p{n}"))), + ("g".to_owned(), AttributeValue::S("v".to_owned())), + ]); + engine + .put_item(ki, item, false, None, &Default::default(), None) + .await + .expect("put"); + } + let hold_key = extenddb_core::settings_keys::gsi_creating_phase_delay_key(&ki.table_name); + execute::( + &engine.session_arc(), + &format!("INSERT INTO {catalog_ks}.settings (key, value) VALUES (?, ?)"), + cdrs_tokio::query_values!(hold_key.as_str(), "3000"), + "gsi_claim_test", + ) + .await + .expect("set the phase hold"); + let index_id = index_id(&engine, &ki.table_id).await; + let data = extenddb_storage_cassandra::data::ddl::index_table_name(&index_id); + + // As UpdateTable leaves an index it adds. + set_row( + &engine, + &ki.table_id, + "index_status = 'CREATING', backfill_heartbeat = null", + ) + .await; + assert_eq!( + described(&engine, &ki.account_id, &ki.table_name).await, + ("CREATING".to_owned(), Some(serde_json::Value::Bool(false))) + ); + + let stale_after = Duration::from_secs(120); + recover_stale_gsi_backfills_of_table(&engine, &account_ks, &ki.table_id, stale_after) + .await + .expect("the first scan"); + assert_eq!( + described(&engine, &ki.account_id, &ki.table_name).await, + ("CREATING".to_owned(), Some(serde_json::Value::Bool(true))), + "the claim records the start of the backfill" + ); + let read = engine.index_info_by_table_id(&ki.table_id, "gidx").await; + assert!( + matches!(read, Err(StorageError::Validation(ref m)) if m == BACKFILLING), + "after the claim a read gets the backfilling sentence: {read:?}" + ); + let heartbeat_at = scalar( + &engine, + &format!( + "SELECT WRITETIME(backfill_heartbeat) AS v FROM {catalog_ks}.indexes \ + WHERE table_id = ? AND index_name = 'gidx'" + ), + cdrs_tokio::query_values!(ki.table_id.as_str()), + ) + .await + .expect("the heartbeat's write time"); + let (rebuilt, rebuilt_at) = data_table(&engine, &account_ks, &data).await; + assert!( + heartbeat_at < rebuilt_at, + "the heartbeat ({heartbeat_at}) is written before the data table is rebuilt ({rebuilt_at})" + ); + + // The backfill the first scan started is waiting out the hold. + recover_stale_gsi_backfills_of_table(&engine, &account_ks, &ki.table_id, stale_after) + .await + .expect("the second scan"); + assert_eq!( + data_table(&engine, &account_ks, &data).await.0, + rebuilt, + "the second scan left the data table to the running build" + ); + assert_eq!( + described(&engine, &ki.account_id, &ki.table_name).await, + ("CREATING".to_owned(), Some(serde_json::Value::Bool(true))) + ); + + let deadline = Instant::now() + Duration::from_secs(60); + while described(&engine, &ki.account_id, &ki.table_name).await + != ("ACTIVE".to_owned(), None) + { + assert!( + Instant::now() < deadline, + "the running build never published the index" + ); + tokio::time::sleep(Duration::from_millis(200)).await; + } + engine + .index_info_by_table_id(&ki.table_id, "gidx") + .await + .expect("an ACTIVE index resolves"); + } + + /// Make a new table ACTIVE with the update the control-plane worker + /// applies, limited to this table so the tests running beside this one + /// keep the state of theirs. + async fn activate(engine: &CassandraEngine, ki: &TableKeyInfo) { + let catalog_ks = engine.catalog_keyspace(); + query_lwt( + &engine.session_arc(), + &format!( + "UPDATE {catalog_ks}.tables SET table_status = 'ACTIVE', \ + status_transition_at = null \ + WHERE account_id = ? AND table_name = ? IF table_status = 'CREATING'" + ), + cdrs_tokio::query_values!(ki.account_id.as_str(), ki.table_name.as_str()), + ) + .await + .expect("activate the table"); + } + + /// Add `gidx`, keyed on `g`, with `UpdateTable`. + async fn add_gsi(engine: &CassandraEngine, ki: &TableKeyInfo) { + let attribute = |name: &str| AttributeDefinition { + attribute_name: name.to_owned(), + attribute_type: ScalarAttributeType::S, + }; + engine + .update_table( + &ki.account_id, + UpdateTableInput { + table_name: ki.table_name.clone(), + billing_mode: None, + table_throughput_mode: None, + provisioned_throughput: None, + deletion_protection_enabled: None, + global_secondary_index_updates: Some(vec![GlobalSecondaryIndexUpdate { + create: Some(CreateGsiAction { + index_name: "gidx".to_owned(), + key_schema: vec![KeySchemaElement { + attribute_name: "g".to_owned(), + key_type: KeyType::Hash, + }], + projection: Projection { + projection_type: ProjectionType::All, + non_key_attributes: None, + }, + provisioned_throughput: None, + }), + update: None, + delete: None, + }]), + attribute_definitions: Some(vec![attribute("id"), attribute("g")]), + stream_specification: None, + table_class: None, + on_demand_throughput: None, + vector_index_updates: None, + }, + ) + .await + .expect("UpdateTable adds gidx"); + } + + /// The number of rows in a keyspace's table, or `None` when the keyspace + /// has no table of that name. + async fn rows_in(engine: &CassandraEngine, keyspace: &str, table: &str) -> Option { + let exists = query_optional::( + &engine.session_arc(), + "SELECT table_name FROM system_schema.tables \ + WHERE keyspace_name = ? AND table_name = ?", + cdrs_tokio::query_values!(keyspace, table), + "gsi_own_keyspace_test", + ) + .await + .expect("read the schema row") + .is_some(); + if !exists { + return None; + } + scalar( + engine, + &format!("SELECT COUNT(*) AS v FROM {keyspace}.{table}"), + cdrs_tokio::query_values!(), + ) + .await + } + + /// One scan of the recovery worker in one account keyspace, limited to + /// the given tables. + async fn scan(engine: &Arc, account_ks: &str, tables: &[&TableKeyInfo]) { + for ki in tables { + recover_stale_gsi_backfills_of_table( + engine, + account_ks, + &ki.table_id, + Duration::from_secs(120), + ) + .await + .expect("a scan of the recovery worker"); + } + } + + /// The recovery worker runs its catalog scan once per account keyspace, + /// and the scan returns the `CREATING` indexes of every account. Two + /// accounts each add an index with `UpdateTable`, and one pass scans the + /// first account's keyspace, then the second's. The scan of the first + /// keyspace leaves the second account's index unclaimed and creates no + /// data table for it there, and the scan of the second keyspace builds + /// it, so both indexes become ACTIVE with their table's items. + #[tokio::test] + async fn a_recovery_pass_builds_each_index_in_its_own_account_keyspace() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = Arc::new(crate::helpers::setup_engine().await); + let first = crate::helpers::TestTable::new(&engine, "GsiOwnKeyspaceFirst", false).await; + let second = crate::helpers::TestTable::new(&engine, "GsiOwnKeyspaceSecond", false).await; + let tables = [&first.key_info, &second.key_info]; + let mut keyspaces = Vec::new(); + let mut data = Vec::new(); + for (label, ki) in ["first", "second"].into_iter().zip(tables) { + activate(&engine, ki).await; + for n in 0..3 { + let item = std::collections::BTreeMap::from([ + ("id".to_owned(), AttributeValue::S(format!("p{n}"))), + ("g".to_owned(), AttributeValue::S("v".to_owned())), + ]); + engine + .put_item(ki, item, false, None, &Default::default(), None) + .await + .expect("put"); + } + add_gsi(&engine, ki).await; + assert_eq!( + described(&engine, &ki.account_id, &ki.table_name).await, + ("CREATING".to_owned(), Some(serde_json::Value::Bool(false))), + "UpdateTable leaves the {label} account's index before its backfill" + ); + keyspaces.push(engine.account_keyspace(&ki.account_id)); + data.push(extenddb_storage_cassandra::data::ddl::index_table_name( + &index_id(&engine, &ki.table_id).await, + )); + } + + scan(&engine, &keyspaces[0], &tables).await; + let ki = tables[1]; + assert_eq!( + described(&engine, &ki.account_id, &ki.table_name).await, + ("CREATING".to_owned(), Some(serde_json::Value::Bool(false))), + "the scan of the first account's keyspace leaves the second account's index unclaimed" + ); + assert_eq!( + rows_in(&engine, &keyspaces[0], &data[1]).await, + None, + "the scan of the first account's keyspace creates no data table for the second \ + account's index" + ); + + scan(&engine, &keyspaces[1], &tables).await; + for (n, ki) in tables.iter().enumerate() { + let deadline = Instant::now() + Duration::from_secs(60); + while described(&engine, &ki.account_id, &ki.table_name).await + != ("ACTIVE".to_owned(), None) + { + assert!( + Instant::now() < deadline, + "{}'s index never became ACTIVE", + ki.table_name + ); + tokio::time::sleep(Duration::from_millis(200)).await; + } + assert_eq!( + rows_in(&engine, &keyspaces[n], &data[n]).await, + Some(3), + "{}'s index holds its table's three items", + ki.table_name + ); + engine + .index_info_by_table_id(&ki.table_id, "gidx") + .await + .expect("an ACTIVE index resolves"); + } + } +} diff --git a/crates/storage-cassandra/tests/direct_integration.rs b/crates/storage-cassandra/tests/direct_integration.rs index 7c12889a0..4cb57936f 100644 --- a/crates/storage-cassandra/tests/direct_integration.rs +++ b/crates/storage-cassandra/tests/direct_integration.rs @@ -31,6 +31,8 @@ mod concurrent_account_creation; mod delete_item; #[path = "direct/groups.rs"] mod groups; +#[path = "direct/gsi_backfilling.rs"] +mod gsi_backfilling; #[path = "direct/index.rs"] mod index; #[path = "direct/metadata_engine.rs"] diff --git a/crates/storage-mongodb/src/gsi_backfill_live_tests.rs b/crates/storage-mongodb/src/gsi_backfill_live_tests.rs new file mode 100644 index 000000000..869ec7569 --- /dev/null +++ b/crates/storage-mongodb/src/gsi_backfill_live_tests.rs @@ -0,0 +1,261 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Live MongoDB tests: a GSI added by `UpdateTable` walks the service's two +//! `CREATING` phases (measured 2026-10-26, see `extenddb_storage::gsi_backfill`), +//! and a backfill resumed after a restart stays in its second phase. +//! +//! Needs a replica set, read from `EXTENDDB_TEST_MONGODB_URI`, and is +//! `#[ignore]`d like the backup live tests; run it with +//! `cargo test -p extenddb-storage-mongodb --lib -- --ignored live_gsi_`. Each +//! test uses its own catalog and data databases and drops them when it passes. + +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use extenddb_core::expression::ExpressionMaps; +use extenddb_core::types::{AttributeValue, DescribeTableInput, TableKeyInfo}; +use extenddb_storage::error::StorageError; +use extenddb_storage::{DataEngine, TableEngine}; +use mongodb::bson::{Document, doc}; + +use crate::MongoEngine; + +const ACCOUNT: &str = "123456789012"; +const TABLE: &str = "t_gsi_backfilling"; +const BACKFILLING: &str = "Cannot read from backfilling global secondary index: gidx"; + +async fn engine() -> MongoEngine { + let uri = std::env::var("EXTENDDB_TEST_MONGODB_URI") + .expect("set EXTENDDB_TEST_MONGODB_URI to a MongoDB replica set"); + let client = mongodb::Client::with_uri_str(uri).await.expect("client"); + let tag = uuid::Uuid::new_v4().simple().to_string(); + let slots = crate::backup_engine::restore_server_slot_count(10); + MongoEngine { + client: client.clone(), + catalog_db: client.database(&format!("g{}_cat", &tag[..12])), + data_db: client.database(&format!("g{}_data", &tag[..12])), + region: "us-east-1".to_owned(), + gsi_cache: Arc::default(), + tx_read_concern: mongodb::options::ReadConcern::snapshot(), + backup_liveness: crate::backup_engine::BackupLiveness::default(), + restore_server_slot_count: slots, + restore_server_slots: Arc::new(tokio::sync::Semaphore::new(slots)), + restore_account_slots: crate::backup_engine::RestoreAccountSlots::default(), + } +} + +/// A three-item table with the phase hold set to `hold_ms`, then `gidx` +/// added by `UpdateTable`, which returns at once on this backend: the index +/// is `CREATING` until the backfill worker builds it. +async fn table_with_a_new_gsi(engine: &MongoEngine, hold_ms: u32) -> TableKeyInfo { + let settings = engine.catalog_db.collection::("settings"); + settings + .insert_one(doc! { "_id": "control_plane_delay_seconds", "value": "0" }) + .await + .expect("settings"); + + let input: extenddb_core::types::CreateTableInput = serde_json::from_value(serde_json::json!({ + "TableName": TABLE, + "KeySchema": [{"AttributeName": "pk", "KeyType": "HASH"}], + "AttributeDefinitions": [{"AttributeName": "pk", "AttributeType": "S"}], + "BillingMode": "PAY_PER_REQUEST" + })) + .expect("input"); + engine + .create_table_impl(ACCOUNT, input, false) + .await + .expect("create table"); + let key_info = engine + .table_key_info(ACCOUNT, TABLE) + .await + .expect("key info"); + for n in 0..3 { + let item = std::collections::BTreeMap::from([ + ("pk".to_owned(), AttributeValue::S(format!("p{n}"))), + ("g".to_owned(), AttributeValue::S("v".to_owned())), + ]); + engine + .put_item( + &key_info, + item, + false, + None, + &ExpressionMaps::default(), + None, + ) + .await + .expect("put"); + } + settings + .insert_one(doc! { + "_id": extenddb_core::settings_keys::gsi_creating_phase_delay_key(TABLE), + "value": hold_ms.to_string(), + }) + .await + .expect("set the phase hold"); + + let update: extenddb_core::types::UpdateTableInput = + serde_json::from_value(serde_json::json!({ + "TableName": TABLE, + "AttributeDefinitions": [{"AttributeName": "g", "AttributeType": "S"}], + "GlobalSecondaryIndexUpdates": [{"Create": { + "IndexName": "gidx", + "KeySchema": [{"AttributeName": "g", "KeyType": "HASH"}], + "Projection": {"ProjectionType": "ALL"} + }}] + })) + .expect("update input"); + engine + .update_table(ACCOUNT, update) + .await + .expect("UpdateTable succeeds"); + key_info +} + +/// What DescribeTable says about `gidx`: its status and its serialised +/// `Backfilling` member, or `None` before the catalog row is visible. +async fn gsi_state(engine: &MongoEngine) -> Option<(String, Option)> { + let desc = engine + .describe_table( + ACCOUNT, + DescribeTableInput { + table_name: TABLE.to_owned(), + }, + ) + .await + .expect("describe the table"); + let gsi = desc + .global_secondary_indexes? + .into_iter() + .find(|g| g.index_name == "gidx")?; + let wire = serde_json::to_value(&gsi).expect("serialise the GSI description"); + Some((gsi.index_status, wire.get("Backfilling").cloned())) +} + +/// The phase of `gidx`: 1 is `Backfilling` false, 2 is true, 3 is ACTIVE. +/// +/// Reads the index between two DescribeTable calls and checks the read gets +/// the answer the description implies. `None` when the index is not visible +/// yet, or when the two calls disagree, so a phase change between them cannot +/// pair one phase's description with the other's refusal. +async fn phase(engine: &MongoEngine, key_info: &TableKeyInfo) -> Option { + let before = gsi_state(engine).await?; + let read = engine + .index_info_by_table_id(&key_info.table_id, "gidx") + .await; + if gsi_state(engine).await.as_ref() != Some(&before) { + return None; + } + Some(match before { + (status, Some(serde_json::Value::Bool(false))) if status == "CREATING" => { + assert!( + matches!(read, Err(StorageError::IndexNotFound(ref n)) if n == "gidx"), + "before the backfill a read is told there is no such index: {read:?}" + ); + 1 + } + (status, Some(serde_json::Value::Bool(true))) if status == "CREATING" => { + assert!( + matches!(read, Err(StorageError::Validation(ref m)) if m == BACKFILLING), + "during the backfill a read gets the backfilling sentence: {read:?}" + ); + 2 + } + (status, None) if status == "ACTIVE" => { + read.expect("an ACTIVE index resolves"); + 3 + } + other => panic!("a state the service does not report: {other:?}"), + }) +} + +#[tokio::test] +#[ignore = "needs EXTENDDB_TEST_MONGODB_URI"] +async fn live_gsi_added_by_update_table_walks_both_creating_phases() { + let engine = engine().await; + let worker = tokio::spawn(crate::ttl_worker::gsi_backfill_worker(Arc::new( + engine.clone(), + ))); + let key_info = table_with_a_new_gsi(&engine, 2000).await; + + let mut seen: Vec = Vec::new(); + let deadline = Instant::now() + Duration::from_secs(60); + while seen.last() != Some(&3) { + assert!(Instant::now() < deadline, "no ACTIVE state; seen {seen:?}"); + if let Some(phase) = phase(&engine, &key_info).await + && seen.last() != Some(&phase) + { + assert!( + seen.last().is_none_or(|&last| last < phase), + "phase {phase} after {seen:?}" + ); + seen.push(phase); + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + assert_eq!(seen, vec![1, 2, 3], "each phase is observed, in order"); + + worker.abort(); + engine.catalog_db.drop().await.expect("drop catalog"); + engine.data_db.drop().await.expect("drop data"); +} + +/// A process that dies during a backfill leaves the index document marked as +/// backfilling. The worker of the next process resumes the job from that +/// mark: the index reports its second phase before and during the resumed +/// backfill, and the job goes straight to its scan instead of starting the +/// first phase again, which with the hold set would wait it twice more. +#[tokio::test] +#[ignore = "needs EXTENDDB_TEST_MONGODB_URI"] +async fn live_gsi_job_resumed_after_a_restart_keeps_its_second_phase() { + const HOLD_MS: u32 = 30_000; + let engine = engine().await; + // No worker yet: the index stays as UpdateTable left it until the mark + // below stands in for a worker that set it and then died. + let key_info = table_with_a_new_gsi(&engine, HOLD_MS).await; + assert_eq!( + phase(&engine, &key_info).await, + Some(1), + "as UpdateTable left it" + ); + let marked = engine + .catalog_db + .collection::("indexes") + .update_one( + doc! { + "_id": { "table_id": &key_info.table_id, "index_name": "gidx" }, + "index_status": "CREATING", + }, + doc! { "$set": { "backfilling": true } }, + ) + .await + .expect("mark the backfill as started"); + assert_eq!(marked.matched_count, 1, "the CREATING index document"); + assert_eq!( + phase(&engine, &key_info).await, + Some(2), + "after the restart" + ); + + let started = Instant::now(); + let worker = tokio::spawn(crate::ttl_worker::gsi_backfill_worker(Arc::new( + engine.clone(), + ))); + loop { + match phase(&engine, &key_info).await { + Some(3) => break, + Some(2) | None => {} + Some(other) => panic!("phase {other} while the resumed job ran"), + } + assert!( + started.elapsed() < Duration::from_millis(u64::from(HOLD_MS) / 2), + "the resumed job waited on the phase hold" + ); + tokio::time::sleep(Duration::from_millis(50)).await; + } + + worker.abort(); + engine.catalog_db.drop().await.expect("drop catalog"); + engine.data_db.drop().await.expect("drop data"); +} diff --git a/crates/storage-mongodb/src/gsi_backfill_phase.rs b/crates/storage-mongodb/src/gsi_backfill_phase.rs new file mode 100644 index 000000000..5a00b9da7 --- /dev/null +++ b/crates/storage-mongodb/src/gsi_backfill_phase.rs @@ -0,0 +1,98 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! The `backfilling` field of a GSI catalog document, which DescribeTable +//! reports as `Backfilling` and which decides a Query or Scan refusal for a +//! `CREATING` GSI (see `extenddb_storage::gsi_backfill`). +//! +//! An index `UpdateTable` adds is inserted without the field, which reads as +//! `false`. The backfill worker sets it to `true` before it scans the first +//! batch, and the `ACTIVE` flip removes it. A job resumed after a restart +//! finds it already set. A restore builds its indexes as part of restoring +//! the table and never sets it. + +use std::time::Duration; + +use bson::{Document, doc}; +use extenddb_storage::error::StorageError; + +use crate::MongoEngine; + +/// Record that the backfill of a `CREATING` index has started, after the +/// table's `gsi_creating_phase_delay_ms` test hold, and then wait the hold +/// again. The worker runs one job at a time, so a hold also delays the jobs +/// queued behind this one. +/// +/// Returns `false` when the index is no longer `CREATING` (deleted, or +/// published by another process), in which case the job has nothing to do. +pub(crate) async fn mark_started( + storage: &MongoEngine, + index_id: &str, + table_name: &str, +) -> Result { + let hold = creating_phase_delay(storage, table_name).await; + tokio::time::sleep(hold).await; + let result = storage + .catalog_db + .collection::("indexes") + .update_one( + doc! { "index_id": index_id, "index_status": "CREATING" }, + doc! { "$set": { "backfilling": true } }, + ) + .await + .map_err(|e| StorageError::Internal(e.to_string()))?; + if result.matched_count == 0 { + return Ok(false); + } + tokio::time::sleep(hold).await; + Ok(true) +} + +/// The `Backfilling` member for a GSI catalog document. +pub(crate) fn member(index_doc: &Document, index_status: &str) -> Option { + if index_doc.get_str("backfill_mode") == Ok("restore") { + return None; + } + extenddb_storage::gsi_backfill::backfilling_member(index_status, started(index_doc)) +} + +/// Whether the worker has recorded the start of this index's backfill. +pub(crate) fn started(index_doc: &Document) -> bool { + index_doc.get_bool("backfilling").unwrap_or(false) +} + +/// The `gsi_creating_phase_delay_ms` test hold for one table; zero when unset. +async fn creating_phase_delay(storage: &MongoEngine, table_name: &str) -> Duration { + let key = extenddb_core::settings_keys::gsi_creating_phase_delay_key(table_name); + let setting = storage + .catalog_db + .collection::("settings") + .find_one(doc! { "_id": key }) + .await + .ok() + .flatten(); + extenddb_storage::gsi_backfill::phase_delay( + setting.as_ref().and_then(|d| d.get_str("value").ok()), + ) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn an_added_index_reports_false_then_true_then_nothing() { + let inserted = doc! { "index_status": "CREATING" }; + assert_eq!(member(&inserted, "CREATING"), Some(false)); + let started = doc! { "index_status": "CREATING", "backfilling": true }; + assert_eq!(member(&started, "CREATING"), Some(true)); + let active = doc! { "index_status": "ACTIVE" }; + assert_eq!(member(&active, "ACTIVE"), None); + } + + #[test] + fn a_restored_index_reports_no_member() { + let restoring = doc! { "index_status": "CREATING", "backfill_mode": "restore" }; + assert_eq!(member(&restoring, "CREATING"), None); + } +} diff --git a/crates/storage-mongodb/src/lib.rs b/crates/storage-mongodb/src/lib.rs index b27cc0baf..bddc35530 100644 --- a/crates/storage-mongodb/src/lib.rs +++ b/crates/storage-mongodb/src/lib.rs @@ -17,6 +17,9 @@ pub mod config; mod credential_store; mod data; mod data_engine; +#[cfg(test)] +mod gsi_backfill_live_tests; +mod gsi_backfill_phase; mod management_store; mod metadata_engine; mod operations; diff --git a/crates/storage-mongodb/src/table_engine.rs b/crates/storage-mongodb/src/table_engine.rs index 77706c3c7..f130b0ba3 100644 --- a/crates/storage-mongodb/src/table_engine.rs +++ b/crates/storage-mongodb/src/table_engine.rs @@ -383,6 +383,7 @@ impl MongoEngine { key_schema: gsi.key_schema.clone(), projection: gsi.projection.clone(), index_status: index_status.to_string(), + backfilling: None, provisioned_throughput: gsi.provisioned_throughput.as_ref().map(|pt| { ProvisionedThroughputDescription { read_capacity_units: pt.read_capacity_units, @@ -1318,6 +1319,16 @@ impl MongoEngine { .map_err(|e| StorageError::Internal(e.to_string()))? .ok_or_else(|| StorageError::IndexNotFound(index_name.to_string()))?; + // A GSI still being built is not readable: its collection is partial. + // The service refuses a Query or Scan naming one in two ways, by phase; + // see `extenddb_storage::gsi_backfill`. + if index_doc.get_str("index_status") == Ok("CREATING") { + return Err(extenddb_storage::gsi_backfill::creating_read_refusal( + index_name, + crate::gsi_backfill_phase::started(&index_doc), + )); + } + let index_id = index_doc .get_str("index_id") .map_err(|_| StorageError::Internal("missing index_id".to_string()))? @@ -1570,14 +1581,13 @@ impl MongoEngine { last_decrease_date_time: None, })); + let idx_status = idx_doc.get_str("index_status").unwrap_or("ACTIVE"); gsis.push(GsiDescription { index_name: idx_name, key_schema: idx_key_schema, projection: idx_projection, - index_status: idx_doc - .get_str("index_status") - .unwrap_or("ACTIVE") - .to_string(), + index_status: idx_status.to_string(), + backfilling: crate::gsi_backfill_phase::member(idx_doc, idx_status), provisioned_throughput: idx_pt, index_size_bytes: 0, item_count: 0, diff --git a/crates/storage-mongodb/src/ttl_worker.rs b/crates/storage-mongodb/src/ttl_worker.rs index e9cd41941..0561f7b83 100644 --- a/crates/storage-mongodb/src/ttl_worker.rs +++ b/crates/storage-mongodb/src/ttl_worker.rs @@ -178,6 +178,15 @@ async fn run_gsi_backfill_job(storage: &MongoEngine, job: &Document) -> Result<( if mode == GsiBackfillMode::Restore && !restore_backfill_is_ready(storage, &table_id).await? { return Ok(()); } + let indexes_coll = storage.catalog_db.collection::("indexes"); + if mode == GsiBackfillMode::Live + && !crate::gsi_backfill_phase::started(job) + && !crate::gsi_backfill_phase::mark_started(storage, &index_id, &key_info.table_name) + .await? + { + return drop_index_collection_if_catalog_entry_missing(storage, &indexes_coll, &index_id) + .await; + } let context = GsiBackfillContext { key_info: &key_info, @@ -187,7 +196,6 @@ async fn run_gsi_backfill_job(storage: &MongoEngine, job: &Document) -> Result<( mode, }; let mut cursor = job.get("backfill_cursor").cloned(); - let indexes_coll = storage.catalog_db.collection::("indexes"); loop { let progress = storage @@ -201,7 +209,7 @@ async fn run_gsi_backfill_job(storage: &MongoEngine, job: &Document) -> Result<( doc! { "index_id": &index_id, "index_status": "CREATING" }, doc! { "$set": { "index_status": "ACTIVE" }, - "$unset": { "backfill_cursor": "" }, + "$unset": { "backfill_cursor": "", "backfilling": "" }, }, ) .await diff --git a/crates/storage-postgres/src/create_table.rs b/crates/storage-postgres/src/create_table.rs index fc41c8c0b..3b6fc8135 100755 --- a/crates/storage-postgres/src/create_table.rs +++ b/crates/storage-postgres/src/create_table.rs @@ -557,6 +557,7 @@ impl PostgresEngine { key_schema: g.key_schema.clone(), projection: g.projection.clone(), index_status: "ACTIVE".to_owned(), + backfilling: None, provisioned_throughput: Some(ProvisionedThroughputDescription { read_capacity_units: g .provisioned_throughput diff --git a/crates/storage-postgres/src/data/ddl.rs b/crates/storage-postgres/src/data/ddl.rs index 16f1a3172..a054f2071 100755 --- a/crates/storage-postgres/src/data/ddl.rs +++ b/crates/storage-postgres/src/data/ddl.rs @@ -565,14 +565,15 @@ impl PostgresEngine { table_id: &str, 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)> = + // A GSI still being built is not readable: its table is partial. The + // service refuses a Query or Scan naming one in two ways, by phase; see + // `extenddb_storage::gsi_backfill`. The backfill set is read before the + // row; see `GsiBackfills::snapshot`. + let started = self.gsi_backfills.snapshot(); + let idx_row: Option<(String, String, serde_json::Value, serde_json::Value, String)> = 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, key_schema, projection, index_status \ + FROM indexes WHERE table_id = $1 AND index_name = $2", ) .bind(table_id) .bind(index_name) @@ -580,8 +581,14 @@ 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, ks_json, proj_json, status) = idx_row.ok_or_else(|| StorageError::IndexNotFound(index_name.to_owned()))?; + if status == "CREATING" { + return Err(extenddb_storage::gsi_backfill::creating_read_refusal( + index_name, + started.contains(&idx_id), + )); + } 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..1acae7e7f 100644 --- a/crates/storage-postgres/src/data/gsi_build.rs +++ b/crates/storage-postgres/src/data/gsi_build.rs @@ -969,6 +969,7 @@ pub(crate) async fn remove_failed_build( data: &sqlx::PgPool, table_id: &str, build: &GsiBuild, + backfills: &extenddb_storage::gsi_backfill::GsiBackfills, ) { // By id, not name: the name may already belong to a newer index. let deleted = sqlx::query( @@ -981,10 +982,14 @@ pub(crate) async fn remove_failed_build( .await; match deleted { Ok(r) if r.rows_affected() == 1 => { + backfills.finish(&build.def.index_id); tokio::time::sleep(GRACE_PUBLISH).await; drop_gsi_tables(data, std::slice::from_ref(build)).await; } - Ok(_) => {} + // No CREATING row: the index was published or deleted meanwhile. + Ok(_) => backfills.finish(&build.def.index_id), + // The row may still be CREATING, so its record stays for the recovery + // pass that will take it up. Err(e) => tracing::error!( index_id = %build.def.index_id, "could not remove the catalog row of a failed GSI build: {e}" @@ -1015,6 +1020,16 @@ async fn index_row_exists( /// that failed with its error; builds before it are ACTIVE, and the caller /// abandons it and the ones after it. A build whose index was deleted while it /// ran, or that someone else published first, is not a failure. +/// +/// Each build is recorded in `backfills` from the start of its backfill until +/// its index leaves `CREATING`, which is what DescribeTable reports as +/// `Backfilling: true`; before that the index reports `false`. A build that +/// returns an error leaves the record for the caller: a passing failure keeps +/// it for the retry, and `remove_failed_build` removes it with the index. +/// `hold` is the table's `gsi_creating_phase_delay_ms` test hold, waited once +/// before the barrier and once more inside each build's recorded phase, +/// between its backfill and its flip to ACTIVE. +#[allow(clippy::too_many_arguments)] pub(crate) async fn build_gsis( catalog: &sqlx::PgPool, data: &sqlx::PgPool, @@ -1022,12 +1037,14 @@ pub(crate) async fn build_gsis( base_ks: &[KeySchemaElement], attr_defs: &[AttributeDefinition], builds: &[GsiBuild], + backfills: &extenddb_storage::gsi_backfill::GsiBackfills, + hold: Duration, ) -> Result { if builds.is_empty() { return Ok(0); } let mut published_here = 0usize; - tokio::time::sleep(GRACE_PUBLISH).await; + tokio::time::sleep(GRACE_PUBLISH + hold).await; if let Err(e) = write_barrier(data, table_id).await { // A table deleted during the wait is not a failed build. let mut gone = true; @@ -1038,13 +1055,26 @@ pub(crate) async fn build_gsis( } } if gone { + for b in builds { + backfills.finish(&b.def.index_id); + } drop_gsi_tables(data, builds).await; return Ok(0); } return Err((0, e)); } for (i, b) in builds.iter().enumerate() { + // Kept until the index leaves CREATING: removed below once the index + // is published or gone, and by `remove_failed_build` after a failure + // that would recur. A passing failure returns with it in place, so the + // index stays in its second phase until the retry publishes it. + backfills.start(&b.def.index_id); let filled = backfill(data, table_id, &b.def, base_ks, attr_defs).await; + if filled.is_ok() && !hold.is_zero() { + // After the backfill, so the end of the second phase, from the last + // page to the flip, is the part the hold lengthens. + tokio::time::sleep(hold).await; + } let published = match filled { Ok(counts) => { tracing::debug!( @@ -1067,13 +1097,17 @@ pub(crate) async fn build_gsis( Err(e) => Err(e), }; match published { - Ok(true) => published_here += 1, + Ok(true) => { + backfills.finish(&b.def.index_id); + published_here += 1; + } outcome => { // Either the flip found no CREATING row or the build failed. If the // row is gone the index was deleted (or its table was) while this // ran: drop what the delete may have raced past, and move on. match index_row_exists(catalog, table_id, &b.def.index_id).await { Ok(false) => { + backfills.finish(&b.def.index_id); drop_gsi_tables(data, std::slice::from_ref(b)).await; tracing::info!( index_id = %b.def.index_id, @@ -1084,7 +1118,10 @@ pub(crate) async fn build_gsis( let e = match outcome { Err(e) => e, // Present but not CREATING: someone else published it. - Ok(_) => continue, + Ok(_) => { + backfills.finish(&b.def.index_id); + continue; + } }; return Err((i, e)); } @@ -1302,6 +1339,8 @@ async fn reconcile_pass( &base_ks, &attr_defs, std::slice::from_ref(&build), + &engine.gsi_backfills, + Duration::ZERO, ) .await { @@ -1337,7 +1376,14 @@ async fn reconcile_pass( index_id = %index_id, "GSI build fails on every attempt; removing the index: {e}" ); - remove_failed_build(&engine.pool, &engine.data_pool, &table_id, &build).await; + remove_failed_build( + &engine.pool, + &engine.data_pool, + &table_id, + &build, + &engine.gsi_backfills, + ) + .await; if let Some(b) = backoff.as_mut() { b.succeeded(&index_id); } diff --git a/crates/storage-postgres/src/delete_table.rs b/crates/storage-postgres/src/delete_table.rs index f03d2c7f0..5e1e0e763 100755 --- a/crates/storage-postgres/src/delete_table.rs +++ b/crates/storage-postgres/src/delete_table.rs @@ -60,7 +60,9 @@ impl PostgresEngine { let row = row.ok_or_else(|| StorageError::TableNotFound(input.table_name.clone()))?; - // Fetch indexes for the response description. + // Fetch indexes for the response description, after the backfill set; + // see `GsiBackfills::snapshot`. + let started = self.gsi_backfills.snapshot(); let index_rows: Vec = sqlx::query_as( r"SELECT index_name, index_id, index_type, key_schema, projection, index_status, provisioned_throughput @@ -78,7 +80,8 @@ impl PostgresEngine { tx.rollback() .await .map_err(|e| StorageError::Internal(e.to_string()))?; - let desc = self.build_table_description_from_row(account_id, row, index_rows)?; + let desc = + self.build_table_description_from_row(account_id, row, index_rows, &started)?; return Ok(TableDescription { table_status: TableStatus::Deleting, ..desc @@ -328,7 +331,7 @@ impl PostgresEngine { } // Build description from the fetched row data - let desc = self.build_table_description_from_row(account_id, row, index_rows)?; + let desc = self.build_table_description_from_row(account_id, row, index_rows, &started)?; Ok(TableDescription { table_status: TableStatus::Deleting, diff --git a/crates/storage-postgres/src/lib.rs b/crates/storage-postgres/src/lib.rs index 9603d68ff..8a101624c 100755 --- a/crates/storage-postgres/src/lib.rs +++ b/crates/storage-postgres/src/lib.rs @@ -304,6 +304,10 @@ pub struct PostgresEngine { /// tests that count them. Shared by clones. #[cfg(feature = "test-hooks")] pub(crate) drop_attempts: Arc, + /// GSI builds in this process whose backfill has started, which is what + /// DescribeTable reports as `Backfilling` and what decides a Query or Scan + /// refusal for a `CREATING` GSI; see `extenddb_storage::gsi_backfill`. + pub(crate) gsi_backfills: extenddb_storage::gsi_backfill::GsiBackfills, } /// Table snapshots one engine may hold open at once: half the pool (at least @@ -430,6 +434,7 @@ impl PostgresEngine { shared_data_pool, #[cfg(feature = "test-hooks")] drop_attempts: Arc::default(), + gsi_backfills: extenddb_storage::gsi_backfill::GsiBackfills::default(), }) } @@ -743,6 +748,22 @@ impl PostgresEngine { } } + /// The `gsi_creating_phase_delay_ms` test hold for one table; zero when unset. + pub(crate) async fn gsi_creating_phase_delay(&self, table_name: &str) -> std::time::Duration { + let live: Result, _> = + sqlx::query_as("SELECT value FROM settings WHERE key = $1") + .bind(extenddb_core::settings_keys::gsi_creating_phase_delay_key( + table_name, + )) + .fetch_optional(&self.pool) + .await; + let value = live.unwrap_or_else(|e| { + tracing::debug!("gsi_creating_phase_delay: live read failed, using 0: {e:?}"); + None + }); + extenddb_storage::gsi_backfill::phase_delay(value.as_ref().map(|(v,)| v.as_str())) + } + /// Whether the data database has pgvector, as probed at construction. /// /// Public so that a deployment check, and the tests that pin the diff --git a/crates/storage-postgres/src/table_helpers.rs b/crates/storage-postgres/src/table_helpers.rs index 545f3701e..180a2036f 100755 --- a/crates/storage-postgres/src/table_helpers.rs +++ b/crates/storage-postgres/src/table_helpers.rs @@ -131,6 +131,8 @@ impl PostgresEngine { account_id: &str, table_name: &str, ) -> Result { + // Before the catalog rows; see `GsiBackfills::snapshot`. + let started = self.gsi_backfills.snapshot(); let row: Option = sqlx::query_as( r"SELECT table_name, key_schema, attribute_definitions, billing_mode, provisioned_throughput, stream_specification, table_status, @@ -169,7 +171,8 @@ impl PostgresEngine { let table_name_owned = row.table_name.clone(); let table_id = row.table_id.clone(); - let mut desc = self.build_table_description_from_row(account_id, row, index_rows)?; + let mut desc = + self.build_table_description_from_row(account_id, row, index_rows, &started)?; desc.restore_summary = Self::restore_summary_on(conn, &table_id, &desc.table_status).await?; desc.vector_indexes = vector_index_descriptions( @@ -181,11 +184,16 @@ impl PostgresEngine { Ok(desc) } + /// `started` is a [`GsiBackfills::snapshot`] the caller took before it read + /// `index_rows`. + /// + /// [`GsiBackfills::snapshot`]: extenddb_storage::gsi_backfill::GsiBackfills::snapshot pub(crate) fn build_table_description_from_row( &self, account_id: &str, row: TableRow, index_rows: Vec, + started: &std::collections::HashSet, ) -> Result { let mut gsis: Vec = Vec::new(); let mut lsis: Vec = Vec::new(); @@ -223,6 +231,10 @@ impl PostgresEngine { index_name: idx.index_name.clone(), key_schema: ks, projection: proj, + backfilling: extenddb_storage::gsi_backfill::backfilling_member( + &idx.index_status, + started.contains(&idx.index_id), + ), index_status: idx.index_status, provisioned_throughput: pt.or(Some(ProvisionedThroughputDescription { read_capacity_units: 0, diff --git a/crates/storage-postgres/src/update_table.rs b/crates/storage-postgres/src/update_table.rs index dec99a93b..931b83f13 100755 --- a/crates/storage-postgres/src/update_table.rs +++ b/crates/storage-postgres/src/update_table.rs @@ -1041,6 +1041,8 @@ impl PostgresEngine { let table_name = input.table_name.clone(); let base_ks = base_key_schema.clone(); let attr_defs = effective_attr_defs.clone(); + let backfills = self.gsi_backfills.clone(); + let hold = self.gsi_creating_phase_delay(&input.table_name).await; let task = tokio::spawn(async move { let built = build_gsis( &catalog, @@ -1049,6 +1051,8 @@ impl PostgresEngine { &base_ks, &attr_defs, &gsi_builds, + &backfills, + hold, ) .await; let Err((first_failed, e)) = built else { @@ -1074,7 +1078,7 @@ impl PostgresEngine { "Failed to build GSI '{}' on '{table_name}', removing it: {e}", b.index_name, ); - remove_failed_build(&catalog, &data, &build_table_id, b).await; + remove_failed_build(&catalog, &data, &build_table_id, b, &backfills).await; } Err(e) }); diff --git a/crates/storage-postgres/tests/gsi_backfilling.rs b/crates/storage-postgres/tests/gsi_backfilling.rs new file mode 100644 index 000000000..1351719b1 --- /dev/null +++ b/crates/storage-postgres/tests/gsi_backfilling.rs @@ -0,0 +1,464 @@ +// SPDX-License-Identifier: Apache-2.0 +//! A GSI added by `UpdateTable` walks the service's two `CREATING` phases. +//! +//! Measured 2026-10-26 (see `extenddb_storage::gsi_backfill`): `Backfilling` +//! is `false` and a read is told the table has no such index until the +//! backfill starts, then `true` with the backfilling sentence until the index +//! is `ACTIVE`, when the member is gone and reads are served. The table's +//! `gsi_creating_phase_delay_ms` hold lengthens both phases so the walk is +//! observable on a three-item table. +//! +//! Each test builds its own throwaway database from the shipped migrations and +//! drops it when it passes; a failing test leaves it behind, named +//! `eddb_gsibf_*`. Requires `EXTENDDB_TEST_PG_CONNECTION_STRING`, a base URL +//! with no database component, pointing at a server whose role may create and +//! drop databases. Without it each test reports a skip and passes. + +use std::collections::BTreeMap; +use std::time::{Duration, Instant}; + +use extenddb_core::expression::ExpressionMaps; +use extenddb_core::types::{ + AttributeDefinition, AttributeValue, BillingMode, CreateTableInput, DescribeTableInput, + KeySchemaElement, KeyType, ScalarAttributeType, TableDescription, TableKeyInfo, + UpdateTableInput, +}; +use extenddb_storage::error::StorageError; +use extenddb_storage::{DataEngine, TableEngine}; +use extenddb_storage_postgres::{PostgresConfig, PostgresEngine}; +use sqlx::postgres::PgPoolOptions; + +const ACCOUNT: &str = "123456789012"; +const TABLE: &str = "t_gsi_backfilling"; +const BACKFILLING: &str = "Cannot read from backfilling global secondary index: gidx"; + +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 +} + +/// A scratch database holding one three-item table, an engine on it running +/// the propagation workers, and the table's phase hold. +struct Scratch { + admin: sqlx::PgPool, + db: sqlx::PgPool, + db_name: String, + engine: PostgresEngine, + key_info: TableKeyInfo, +} + +impl Scratch { + async fn new(base: &str, hold_ms: u32) -> Self { + let db_name = format!("eddb_gsibf_{}", 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(3) + .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( + &PostgresConfig { + connection_string: url, + pool_size: 10, + max_item_size_bytes: 400_000, + }, + "us-east-1", + ) + .await + .expect("open a PostgresEngine on the scratch database") + .start_gsi_workers(); + + engine + .create_table( + ACCOUNT, + CreateTableInput { + table_name: TABLE.to_owned(), + key_schema: vec![KeySchemaElement { + attribute_name: "pk".to_owned(), + key_type: KeyType::Hash, + }], + attribute_definitions: vec![AttributeDefinition { + attribute_name: "pk".to_owned(), + attribute_type: ScalarAttributeType::S, + }], + billing_mode: Some(BillingMode::PayPerRequest), + ..Default::default() + }, + ) + .await + .expect("create the table"); + let key_info = engine + .table_key_info(ACCOUNT, TABLE) + .await + .expect("read the key info"); + for n in 0..3 { + let item = BTreeMap::from([ + ("pk".to_owned(), AttributeValue::S(format!("p{n}"))), + ("g".to_owned(), AttributeValue::S("v".to_owned())), + ]); + engine + .put_item( + &key_info, + item, + false, + None, + &ExpressionMaps::default(), + None, + ) + .await + .expect("put"); + } + + sqlx::query("INSERT INTO settings (key, value) VALUES ($1, $2)") + .bind(extenddb_core::settings_keys::gsi_creating_phase_delay_key( + TABLE, + )) + .bind(hold_ms.to_string()) + .execute(&db) + .await + .expect("set the phase hold"); + + Self { + admin, + db, + db_name, + engine, + key_info, + } + } + + /// Start `UpdateTable` adding `gidx` in its own task. It returns once the + /// index is ACTIVE on this backend, or once a passing build failure has + /// left it to the recovery pass. + fn add_gsi(&self) -> tokio::task::JoinHandle> { + let input: UpdateTableInput = serde_json::from_value(serde_json::json!({ + "TableName": TABLE, + "AttributeDefinitions": [{"AttributeName": "g", "AttributeType": "S"}], + "GlobalSecondaryIndexUpdates": [{"Create": { + "IndexName": "gidx", + "KeySchema": [{"AttributeName": "g", "KeyType": "HASH"}], + "Projection": {"ProjectionType": "ALL"}, + }}], + })) + .expect("build the UpdateTable input"); + let updater = self.engine.clone(); + tokio::spawn(async move { updater.update_table(ACCOUNT, input).await }) + } + + /// What DescribeTable says about `gidx`: its status and its serialised + /// `Backfilling` member, or `None` before the catalog row is visible. + async fn gsi_state(&self) -> Option<(String, Option)> { + let desc = self + .engine + .describe_table( + ACCOUNT, + DescribeTableInput { + table_name: TABLE.to_owned(), + }, + ) + .await + .expect("describe the table"); + let gsi = desc + .global_secondary_indexes? + .into_iter() + .find(|g| g.index_name == "gidx")?; + let wire = serde_json::to_value(&gsi).expect("serialise the GSI description"); + Some((gsi.index_status, wire.get("Backfilling").cloned())) + } + + /// The phase of `gidx`: 1 is `Backfilling` false, 2 is true, 3 is ACTIVE. + /// + /// Reads the index between two DescribeTable calls and checks the read + /// gets the answer the description implies. `None` when the index is not + /// visible yet, or when the two calls disagree, so a phase change between + /// them cannot pair one phase's description with the other's refusal. + async fn phase(&self) -> Option { + let before = self.gsi_state().await?; + let read = self + .engine + .index_info_by_table_id(&self.key_info.table_id, "gidx") + .await; + if self.gsi_state().await.as_ref() != Some(&before) { + return None; + } + Some(match before { + (status, Some(serde_json::Value::Bool(false))) if status == "CREATING" => { + assert!( + matches!(read, Err(StorageError::IndexNotFound(ref n)) if n == "gidx"), + "before the backfill a read is told there is no such index: {read:?}" + ); + 1 + } + (status, Some(serde_json::Value::Bool(true))) if status == "CREATING" => { + assert!( + matches!(read, Err(StorageError::Validation(ref m)) if m == BACKFILLING), + "during the backfill a read gets the backfilling sentence: {read:?}" + ); + 2 + } + (status, None) if status == "ACTIVE" => { + read.expect("an ACTIVE index resolves"); + 3 + } + other => panic!("a state the service does not report: {other:?}"), + }) + } + + async fn cleanup(self) { + drop(self.engine); + self.db.close().await; + sqlx::query(&format!( + "DROP DATABASE IF EXISTS \"{}\" WITH (FORCE)", + self.db_name + )) + .execute(&self.admin) + .await + .expect("drop the scratch database"); + self.admin.close().await; + } +} + +#[tokio::test] +async fn a_gsi_added_by_update_table_walks_both_creating_phases() { + let Some(base) = base_conn() else { + eprintln!("SKIP: EXTENDDB_TEST_PG_CONNECTION_STRING is not set"); + return; + }; + let s = Scratch::new(&base, 1500).await; + let update = s.add_gsi(); + + let mut seen: Vec = Vec::new(); + let deadline = Instant::now() + Duration::from_secs(60); + while seen.last() != Some(&3) { + assert!(Instant::now() < deadline, "no ACTIVE state; seen {seen:?}"); + if let Some(phase) = s.phase().await + && seen.last() != Some(&phase) + { + assert!( + seen.last().is_none_or(|&last| last < phase), + "phase {phase} after {seen:?}" + ); + seen.push(phase); + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + assert_eq!(seen, vec![1, 2, 3], "each phase is observed, in order"); + update + .await + .expect("the UpdateTable task") + .expect("UpdateTable succeeds"); + + s.cleanup().await; +} + +/// From the end of its backfill to the flip to ACTIVE, the index is still in +/// its second phase: the build keeps its record until the row has left +/// CREATING, and DescribeTable and a read take the record set before the row. +/// On this backend the hold sits between the backfill and the flip, so the +/// test reads the index through that window. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn the_index_stays_backfilling_from_the_end_of_its_backfill_to_the_flip() { + let Some(base) = base_conn() else { + eprintln!("SKIP: EXTENDDB_TEST_PG_CONNECTION_STRING is not set"); + return; + }; + let s = Scratch::new(&base, 1500).await; + let update = s.add_gsi(); + + // The backfill has written all three items once the index table holds + // them; nothing else writes to the table in this test. + let deadline = Instant::now() + Duration::from_secs(30); + loop { + let index_id: Option<(String,)> = + sqlx::query_as("SELECT index_id FROM indexes WHERE index_name = 'gidx'") + .fetch_optional(&s.db) + .await + .expect("read the index row"); + let rows: Option = match index_id { + Some((id,)) => sqlx::query_scalar(&format!("SELECT count(*) FROM \"_ddb_{id}\"")) + .fetch_one(&s.db) + .await + .ok(), + None => None, + }; + if rows == Some(3) { + break; + } + assert!( + Instant::now() < deadline, + "the backfill never wrote the items" + ); + tokio::time::sleep(Duration::from_millis(10)).await; + } + + let mut readings = 0; + let deadline = Instant::now() + Duration::from_secs(30); + loop { + match s.phase().await { + Some(3) => break, + Some(2) => readings += 1, + None => {} + Some(phase) => panic!("phase {phase} after the backfill wrote every item"), + } + assert!(Instant::now() < deadline, "no ACTIVE state"); + tokio::time::sleep(Duration::from_millis(20)).await; + } + assert!(readings > 0, "no reading between the backfill and the flip"); + update + .await + .expect("the UpdateTable task") + .expect("UpdateTable succeeds"); + + s.cleanup().await; +} + +/// A build that fails for a passing reason leaves its index CREATING for the +/// recovery pass. The index stays in its second phase through the wait and the +/// retry, until the retry publishes it. +/// +/// The failure is a real one: the test locks the index's catalog row so the +/// flip to ACTIVE waits on it, then cancels the waiting statement, which is +/// `SQLSTATE 57014`, a failure the build treats as passing. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_passing_build_failure_keeps_the_index_backfilling_until_the_retry_publishes_it() { + let Some(base) = base_conn() else { + eprintln!("SKIP: EXTENDDB_TEST_PG_CONNECTION_STRING is not set"); + return; + }; + let s = Scratch::new(&base, 1500).await; + let update = s.add_gsi(); + + let deadline = Instant::now() + Duration::from_secs(30); + while s.phase().await != Some(2) { + assert!(Instant::now() < deadline, "the backfill never started"); + tokio::time::sleep(Duration::from_millis(20)).await; + } + let mut lock = s.db.begin().await.expect("begin the lock transaction"); + sqlx::query("SELECT 1 FROM indexes WHERE index_name = 'gidx' FOR UPDATE") + .execute(&mut *lock) + .await + .expect("lock the index row"); + let flip = loop { + let waiting: Option<(i32,)> = sqlx::query_as( + "SELECT pid FROM pg_stat_activity \ + WHERE datname = current_database() AND wait_event_type = 'Lock' \ + AND query LIKE 'UPDATE indexes SET index_status = ''ACTIVE''%'", + ) + .fetch_optional(&s.db) + .await + .expect("read pg_stat_activity"); + if let Some((pid,)) = waiting { + break pid; + } + assert!( + Instant::now() < deadline, + "the flip to ACTIVE never waited on the lock" + ); + tokio::time::sleep(Duration::from_millis(20)).await; + }; + let cancelled: bool = sqlx::query_scalar("SELECT pg_cancel_backend($1)") + .bind(flip) + .fetch_one(&s.db) + .await + .expect("cancel the flip"); + assert!(cancelled, "the flip's backend was signalled"); + lock.rollback().await.expect("release the index row"); + + // UpdateTable succeeds and reports the index it leaves to the recovery pass. + let returned = update + .await + .expect("the UpdateTable task") + .expect("a passing build failure is not an UpdateTable failure"); + let gsi = returned + .global_secondary_indexes + .unwrap_or_default() + .into_iter() + .find(|g| g.index_name == "gidx") + .expect("gidx in the UpdateTable response"); + assert_eq!( + (gsi.index_status.as_str(), gsi.backfilling), + ("CREATING", Some(true)), + "the UpdateTable response after the failure" + ); + for _ in 0..10 { + assert_eq!( + s.phase().await, + Some(2), + "waiting for the retry, the index is still backfilling" + ); + tokio::time::sleep(Duration::from_millis(50)).await; + } + + // The pass the server's recovery worker runs every minute. + let engine = s.engine.clone(); + let retry = + tokio::spawn( + async move { extenddb_storage_postgres::reconcile_creating_gsis(&engine).await }, + ); + let deadline = Instant::now() + Duration::from_secs(60); + loop { + match s.phase().await { + Some(3) => break, + Some(2) | None => {} + Some(phase) => panic!("phase {phase} during the retry"), + } + assert!( + Instant::now() < deadline, + "the retry never published the index" + ); + tokio::time::sleep(Duration::from_millis(20)).await; + } + assert_eq!( + retry + .await + .expect("the recovery task") + .expect("the recovery pass"), + 1, + "the retry published the index" + ); + + s.cleanup().await; +} diff --git a/crates/storage-postgres/tests/gsi_create_converge.rs b/crates/storage-postgres/tests/gsi_create_converge.rs new file mode 100644 index 000000000..22137c9eb --- /dev/null +++ b/crates/storage-postgres/tests/gsi_create_converge.rs @@ -0,0 +1,460 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 +//! Storage-level test for creating a global secondary index while concurrent +//! writers change the base table. +//! +//! `UpdateTable` creates the index, seeds it from the base table, and publishes +//! it. While that runs, four tasks overwrite existing rows (moving the index +//! key), insert new rows, and delete rows. When the index is `ACTIVE` and the +//! propagation queue has drained, the index table must equal the projection of +//! the base table, row for row: no row the backfill copied and a writer then +//! changed may keep its old image, no row a writer moved may leave its old key +//! behind, and a row the scan meets twice must not fail the creation. +//! +//! The scenario runs ten times, with the propagation delay at 5 ms and again at +//! zero. Before the build of an index added by `UpdateTable` waited, took a +//! write barrier, and backfilled under each base row's lock, a write that +//! landed while the scan ran could be dropped or race the scan, and the index +//! was left disagreeing with the base table. +//! +//! 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_gsir_*`, 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 the test reports +//! a skip and passes, the same convention the other suites use. + +use std::collections::BTreeMap; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::Duration; + +use extenddb_core::expression::ExpressionMaps; +use extenddb_core::types::{ + AttributeDefinition, AttributeValue, BillingMode, CreateTableInput, Item, KeySchemaElement, + KeyType, ScalarAttributeType, TableKeyInfo, UpdateTableInput, +}; +use extenddb_storage::{DataEngine, TableEngine}; +use extenddb_storage_postgres::{PostgresConfig, PostgresEngine}; +use sqlx::PgPool; +use sqlx::postgres::PgPoolOptions; + +const ACCOUNT: &str = "123456789012"; +const REGION: &str = "us-east-1"; +/// Base rows seeded before the index is created. +const SEED_ROWS: i64 = 3000; +/// Distinct index-key values among the seeded rows. +const SEED_GROUPS: i64 = 97; + +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()) +} + +/// Every `.sql` file under `migrations/` and `data_migrations/`, each directory +/// in filename order, so a new migration is picked up here without editing this +/// file. +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() { + eprintln!( + "SKIP gsi_create_converge: EXTENDDB_TEST_PG_CONNECTION_STRING is not set, so there is \ + no PostgreSQL to build a scratch catalog in." + ); +} + +async fn scratch() -> Scratch { + scratch_with_delay("5").await +} + +/// Build a scratch database and engine with a chosen propagation delay. Delay +/// `"5"` keeps writes on the asynchronous queue; `"0"` makes an ACTIVE index +/// maintain writes inline. +async fn scratch_with_delay(delay_ms: &str) -> Scratch { + let base = base_conn().expect("caller checks base_conn() first"); + let db_name = format!("eddb_gsir_{}", 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(8) + .connect(&url) + .await + .expect("connect to the scratch database"); + // One scratch database stands in for both the catalog and the data database, + // so every shipped migration from both directories is applied to it. + for sql in shipped_migrations() { + sqlx::raw_sql(&sql) + .execute(&db) + .await + .expect("apply a shipped migration"); + } + // CreateTable completes inline with no control-plane worker. + sqlx::query("UPDATE settings SET value = '0' WHERE key = 'control_plane_delay_seconds'") + .execute(&db) + .await + .expect("pin the control-plane delay to zero"); + // The propagation delay the caller chose. A non-zero delay keeps writes on + // the asynchronous queue; zero returns an ACTIVE index to inline maintenance. + sqlx::query("UPDATE settings SET value = $1 WHERE key = 'index_propagation_delay_ms'") + .bind(delay_ms) + .execute(&db) + .await + .expect("set the propagation delay"); + 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( + &PostgresConfig { + connection_string: url, + pool_size: 10, + max_item_size_bytes: 400_000, + }, + REGION, + ) + .await + .expect("open a PostgresEngine on the scratch database") + .start_gsi_workers(); + Scratch { + engine, + db, + admin, + db_name, + } +} + +fn item_with(pk: &str, g1: &str) -> Item { + BTreeMap::from([ + ("pk".to_owned(), AttributeValue::S(pk.to_owned())), + ("g1".to_owned(), AttributeValue::S(g1.to_owned())), + ("pad".to_owned(), AttributeValue::S("x".repeat(300))), + ]) +} + +fn key_only(pk: &str) -> Item { + BTreeMap::from([("pk".to_owned(), AttributeValue::S(pk.to_owned()))]) +} + +async fn table_id(db: &PgPool, table: &str) -> String { + sqlx::query_scalar("SELECT table_id FROM tables WHERE account_id = $1 AND table_name = $2") + .bind(ACCOUNT) + .bind(table) + .fetch_one(db) + .await + .expect("look up the table id") +} + +async fn gsi_index_id(db: &PgPool, tid: &str) -> String { + sqlx::query_scalar("SELECT index_id FROM indexes WHERE table_id = $1 AND index_name = 'gsi1'") + .bind(tid) + .fetch_one(db) + .await + .expect("look up the index id") +} + +/// `(pk, g1)` of every base row that carries `g1`, sorted. This is the KEYS_ONLY +/// projection the index must hold. +async fn base_projection(db: &PgPool, tid: &str) -> Vec<(String, String)> { + let mut rows: Vec<(String, String)> = sqlx::query_as(&format!( + "SELECT item_data->'pk'->>'S', item_data->'g1'->>'S' FROM \"_ddb_{tid}\" \ + WHERE item_data ? 'g1'" + )) + .fetch_all(db) + .await + .expect("read the base projection"); + rows.sort(); + rows +} + +/// `(pk, g1)` of every index row, sorted. +async fn index_rows(db: &PgPool, iid: &str) -> Vec<(String, String)> { + let mut rows: Vec<(String, String)> = sqlx::query_as(&format!( + "SELECT item_data->'pk'->>'S', item_data->'g1'->>'S' FROM \"_ddb_{iid}\"" + )) + .fetch_all(db) + .await + .expect("read the index rows"); + rows.sort(); + rows +} + +async fn queue_depth(db: &PgPool, tid: &str) -> i64 { + sqlx::query_scalar("SELECT COUNT(*) FROM gsi_pending WHERE table_id = $1") + .bind(tid) + .fetch_one(db) + .await + .expect("count the queued rows") +} + +async fn hold_count(db: &PgPool, tid: &str) -> i64 { + sqlx::query_scalar("SELECT COUNT(*) FROM vector_index_holds WHERE table_id = $1") + .bind(tid) + .fetch_one(db) + .await + .expect("count the holds") +} + +/// One writer: a mix of overwrites that move the index key, inserts of new rows, +/// and deletes, until `stop` is set. It first waits until the index's catalog +/// row is visible, so every write it makes observes the index being created and +/// maintains it through the queue. A write that ran before the catalog row was +/// visible would miss the index through the catalog read window that every write +/// path has, which is a separate matter from the creation race under test. +async fn writer( + engine: &PostgresEngine, + db: &PgPool, + key_info: &TableKeyInfo, + tid: &str, + stop: &AtomicBool, + seed: i64, +) { + let maps = ExpressionMaps::default(); + // Wait for the index's catalog row to appear, or for the creation to end. + while !stop.load(Ordering::Relaxed) && !gsi_visible(db, tid).await { + tokio::time::sleep(Duration::from_millis(1)).await; + } + let mut i = seed; + while !stop.load(Ordering::Relaxed) { + let kind = i % 10; + if kind < 6 { + // Overwrite an existing row with a new index key (a key move). + let pk = format!("k{}", i % SEED_ROWS); + let item = item_with(&pk, &format!("h{}", i % 13)); + let _ = engine + .put_item(key_info, item, false, None, &maps, None) + .await; + } else if kind < 8 { + // Insert a new row, one that the scan may meet, miss, or meet twice. + let pk = format!("ins{seed}_{i}"); + let item = item_with(&pk, &format!("g{}", i % SEED_GROUPS)); + let _ = engine + .put_item(key_info, item, false, None, &maps, None) + .await; + } else { + // Delete an existing row. + let pk = format!("k{}", i % SEED_ROWS); + let _ = engine + .delete_item(key_info, &key_only(&pk), false, None, &maps, None) + .await; + } + i += 1; + } +} + +/// Whether the GSI's catalog row exists yet, whatever its status. +async fn gsi_visible(db: &PgPool, tid: &str) -> bool { + let n: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM indexes WHERE table_id = $1 AND index_name = 'gsi1'", + ) + .bind(tid) + .fetch_one(db) + .await + .unwrap_or(0); + n > 0 +} + +/// Wait until the propagation queue is empty and no hold is held for the table, +/// then settle once more. Returns false on timeout. +async fn wait_for_drain(db: &PgPool, tid: &str) -> bool { + for _ in 0..600 { + if queue_depth(db, tid).await == 0 && hold_count(db, tid).await == 0 { + tokio::time::sleep(Duration::from_millis(50)).await; + if queue_depth(db, tid).await == 0 && hold_count(db, tid).await == 0 { + return true; + } + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + false +} + +async fn run_once(s: &Scratch, iter: usize) -> Result<(), String> { + let table = format!("gsi_race_{iter}"); + s.engine + .create_table( + ACCOUNT, + CreateTableInput { + table_name: table.clone(), + key_schema: vec![KeySchemaElement { + attribute_name: "pk".to_owned(), + key_type: KeyType::Hash, + }], + attribute_definitions: vec![AttributeDefinition { + attribute_name: "pk".to_owned(), + attribute_type: ScalarAttributeType::S, + }], + billing_mode: Some(BillingMode::PayPerRequest), + ..Default::default() + }, + ) + .await + .map_err(|e| format!("create table: {e}"))?; + let key_info = s + .engine + .table_key_info(ACCOUNT, &table) + .await + .map_err(|e| format!("key info: {e}"))?; + let tid = table_id(&s.db, &table).await; + + // Seed the base table directly, so the setup is one statement rather than + // thousands of writes. No index exists yet, so no maintenance is due. + sqlx::query(&format!( + "INSERT INTO \"_ddb_{tid}\" (pk, item_data) \ + SELECT 'k'||g, jsonb_build_object( \ + 'pk', jsonb_build_object('S', 'k'||g), \ + 'g1', jsonb_build_object('S', 'g'||(g % {SEED_GROUPS})), \ + 'pad', jsonb_build_object('S', repeat('x', 300))) \ + FROM generate_series(0, {}) g", + SEED_ROWS - 1 + )) + .execute(&s.db) + .await + .map_err(|e| format!("seed: {e}"))?; + + let stop = AtomicBool::new(false); + let create_input: UpdateTableInput = serde_json::from_value(serde_json::json!({ + "TableName": table, + "AttributeDefinitions": [{"AttributeName": "g1", "AttributeType": "S"}], + "GlobalSecondaryIndexUpdates": [{"Create": { + "IndexName": "gsi1", + "KeySchema": [{"AttributeName": "g1", "KeyType": "HASH"}], + "Projection": {"ProjectionType": "KEYS_ONLY"}, + }}], + })) + .map_err(|e| format!("build the UpdateTable input: {e}"))?; + + let create = async { + let r = s.engine.update_table(ACCOUNT, create_input).await; + stop.store(true, Ordering::Relaxed); + r + }; + let writers = async { + tokio::join!( + writer(&s.engine, &s.db, &key_info, &tid, &stop, 1), + writer(&s.engine, &s.db, &key_info, &tid, &stop, 777), + writer(&s.engine, &s.db, &key_info, &tid, &stop, 1_500), + writer(&s.engine, &s.db, &key_info, &tid, &stop, 2_600), + ) + }; + let (create_result, _) = tokio::join!(create, writers); + create_result.map_err(|e| format!("UpdateTable creating the GSI failed: {e}"))?; + + if !wait_for_drain(&s.db, &tid).await { + return Err(format!( + "propagation did not drain: queue={}, holds={}", + queue_depth(&s.db, &tid).await, + hold_count(&s.db, &tid).await + )); + } + + let iid = gsi_index_id(&s.db, &tid).await; + let base = base_projection(&s.db, &tid).await; + let index = index_rows(&s.db, &iid).await; + + if base != index { + use std::collections::BTreeSet; + let b: BTreeSet<_> = base.iter().cloned().collect(); + let i: BTreeSet<_> = index.iter().cloned().collect(); + let missing = b.difference(&i).count(); + let stale = i.difference(&b).count(); + return Err(format!( + "index does not equal the base projection: base={}, index={}, missing from index={}, \ + stale in index={}", + base.len(), + index.len(), + missing, + stale + )); + } + eprintln!( + "iteration {iter}: base={} index={} equal", + base.len(), + index.len() + ); + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn gsi_creation_converges_under_concurrent_writers() { + if base_conn().is_none() { + return skip(); + } + let s = scratch().await; + for iter in 0..10 { + if let Err(e) = run_once(&s, iter).await { + panic!("iteration {iter}: {e}"); + } + } + s.cleanup().await; +} + +/// The same race with the propagation delay pinned to zero, so an ACTIVE index +/// maintains writes inline rather than through the queue. Rows queued while the +/// index was building must not drain after an inline write and put back an +/// older image, so the index must still equal the base projection. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn gsi_creation_converges_under_concurrent_writers_zero_delay() { + if base_conn().is_none() { + return skip(); + } + let s = scratch_with_delay("0").await; + for iter in 0..10 { + if let Err(e) = run_once(&s, iter).await { + panic!("iteration {iter}: {e}"); + } + } + s.cleanup().await; +} diff --git a/crates/storage-sqlite/src/create_table.rs b/crates/storage-sqlite/src/create_table.rs index ae4ae05e7..084eea664 100644 --- a/crates/storage-sqlite/src/create_table.rs +++ b/crates/storage-sqlite/src/create_table.rs @@ -520,6 +520,7 @@ impl SqliteEngine { key_schema: g.key_schema.clone(), projection: g.projection.clone(), index_status: "ACTIVE".to_owned(), + backfilling: None, provisioned_throughput: Some(ProvisionedThroughputDescription { read_capacity_units: g .provisioned_throughput diff --git a/crates/storage-sqlite/src/data/ddl.rs b/crates/storage-sqlite/src/data/ddl.rs index 739e1310e..3a6df846f 100644 --- a/crates/storage-sqlite/src/data/ddl.rs +++ b/crates/storage-sqlite/src/data/ddl.rs @@ -525,13 +525,14 @@ impl SqliteEngine { index_name: &str, ) -> Result { // A GSI still being built is not readable: its table is partial, or not - // there yet (a Query hit it with a 500 on main). 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 row: Option<(String, String, String, String)> = sqlx::query_as( - "SELECT index_type, index_id, key_schema, projection \ - FROM indexes WHERE table_id = ? AND index_name = ? AND index_status <> 'CREATING'", + // there yet (a Query hit it with a 500 on main). The service refuses a + // Query or Scan naming one in two ways, by phase; see + // `extenddb_storage::gsi_backfill`. The backfill set is read before the + // row; see `GsiBackfills::snapshot`. + let started = self.gsi_backfills.snapshot(); + let row: Option<(String, String, String, String, String)> = sqlx::query_as( + "SELECT index_type, index_id, key_schema, projection, index_status \ + FROM indexes WHERE table_id = ? AND index_name = ?", ) .bind(table_id) .bind(index_name) @@ -539,8 +540,14 @@ impl SqliteEngine { .await .map_err(|e| StorageError::Internal(e.to_string()))?; - let (idx_type_str, index_id, ks_str, proj_str) = + let (idx_type_str, index_id, ks_str, proj_str, status) = row.ok_or_else(|| StorageError::IndexNotFound(index_name.to_owned()))?; + if status == "CREATING" { + return Err(extenddb_storage::gsi_backfill::creating_read_refusal( + index_name, + started.contains(&index_id), + )); + } let index_type = match idx_type_str.as_str() { "GSI" => IndexType::Gsi, diff --git a/crates/storage-sqlite/src/delete_table.rs b/crates/storage-sqlite/src/delete_table.rs index 476177d6c..f3b4102bb 100644 --- a/crates/storage-sqlite/src/delete_table.rs +++ b/crates/storage-sqlite/src/delete_table.rs @@ -37,7 +37,9 @@ impl SqliteEngine { let row = row.ok_or_else(|| StorageError::TableNotFound(input.table_name.clone()))?; if row.table_status == "DELETING" { // Already scheduled for removal by the control-plane worker: - // nothing to change. + // nothing to change. After the backfill set; see + // `GsiBackfills::snapshot`. + let started = self.gsi_backfills.snapshot(); let index_rows: Vec = sqlx::query_as(&format!( "SELECT {INDEX_COLUMNS} FROM indexes WHERE table_id = ?" )) @@ -45,7 +47,8 @@ impl SqliteEngine { .fetch_all(&self.pool) .await .map_err(|e| StorageError::Internal(e.to_string()))?; - let desc = self.build_table_description_from_row(account_id, row, index_rows)?; + let desc = + self.build_table_description_from_row(account_id, row, index_rows, &started)?; return Ok(TableDescription { table_status: TableStatus::Deleting, ..desc @@ -78,6 +81,8 @@ impl SqliteEngine { } } + // After the backfill set; see `GsiBackfills::snapshot`. + let started = self.gsi_backfills.snapshot(); let index_rows: Vec = sqlx::query_as(&format!( "SELECT {INDEX_COLUMNS} FROM indexes WHERE table_id = ?" )) @@ -172,7 +177,7 @@ impl SqliteEngine { self.control_plane_notify.notify_one(); } - let desc = self.build_table_description_from_row(account_id, row, index_rows)?; + let desc = self.build_table_description_from_row(account_id, row, index_rows, &started)?; Ok(TableDescription { table_status: TableStatus::Deleting, ..desc diff --git a/crates/storage-sqlite/src/gsi_backfilling_tests.rs b/crates/storage-sqlite/src/gsi_backfilling_tests.rs new file mode 100644 index 000000000..5dcb35ba2 --- /dev/null +++ b/crates/storage-sqlite/src/gsi_backfilling_tests.rs @@ -0,0 +1,167 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! A GSI added by `UpdateTable` walks the service's two `CREATING` phases. +//! +//! Measured 2026-10-26 (see `extenddb_storage::gsi_backfill`): `Backfilling` +//! is `false` and a read is told the table has no such index until the +//! backfill starts, then `true` with the backfilling sentence until the index +//! is `ACTIVE`, when the member is gone and reads are served. The table's +//! `gsi_creating_phase_delay_ms` hold lengthens both phases so the walk is +//! observable on a three-item table. + +use std::time::{Duration, Instant}; + +use extenddb_core::expression::ExpressionMaps; +use extenddb_core::types::{AttributeValue, DescribeTableInput}; +use extenddb_storage::error::StorageError; +use extenddb_storage::{DataEngine, TableEngine}; +use serde_json::json; + +use crate::SqliteEngine; + +const ACCOUNT: &str = "000000000000"; +const BACKFILLING: &str = "Cannot read from backfilling global secondary index: gidx"; + +/// What DescribeTable says about `gidx`: its status and its serialised +/// `Backfilling` member, or `None` before the catalog row is visible. +async fn gsi_state(engine: &SqliteEngine) -> Option<(String, Option)> { + let desc = engine + .describe_table( + ACCOUNT, + DescribeTableInput { + table_name: "t".to_owned(), + }, + ) + .await + .expect("describe the table"); + let gsi = desc + .global_secondary_indexes? + .into_iter() + .find(|g| g.index_name == "gidx")?; + let wire = serde_json::to_value(&gsi).expect("serialise the GSI description"); + Some((gsi.index_status, wire.get("Backfilling").cloned())) +} + +#[tokio::test] +async fn a_gsi_added_by_update_table_walks_both_creating_phases() { + let engine = SqliteEngine::new(":memory:", 1, "us-east-1", 409_600) + .await + .expect("engine"); + crate::schema::apply(&engine.pool).await.expect("schema"); + sqlx::query("UPDATE settings SET value = '0' WHERE key = 'control_plane_delay_seconds'") + .execute(&engine.pool) + .await + .expect("delay"); + sqlx::query("INSERT INTO accounts (account_id, account_name) VALUES (?, 'default')") + .bind(ACCOUNT) + .execute(&engine.pool) + .await + .expect("account"); + let input: extenddb_core::types::CreateTableInput = serde_json::from_value(json!({ + "TableName": "t", + "KeySchema": [{"AttributeName": "pk", "KeyType": "HASH"}], + "AttributeDefinitions": [{"AttributeName": "pk", "AttributeType": "S"}], + "BillingMode": "PAY_PER_REQUEST" + })) + .expect("input"); + engine + .create_table_impl(ACCOUNT, input, false) + .await + .expect("create table"); + sqlx::query("UPDATE tables SET table_status = 'ACTIVE' WHERE table_name = 't'") + .execute(&engine.pool) + .await + .expect("activate the table"); + let key_info = engine.table_key_info(ACCOUNT, "t").await.expect("key info"); + for n in 0..3 { + let item = std::collections::BTreeMap::from([ + ("pk".to_owned(), AttributeValue::S(format!("p{n}"))), + ("g".to_owned(), AttributeValue::S("v".to_owned())), + ]); + engine + .put_item( + &key_info, + item, + false, + None, + &ExpressionMaps::default(), + None, + ) + .await + .expect("put"); + } + sqlx::query("INSERT INTO settings (key, value) VALUES (?, '1000')") + .bind(extenddb_core::settings_keys::gsi_creating_phase_delay_key( + "t", + )) + .execute(&engine.pool) + .await + .expect("set the phase hold"); + + // UpdateTable returns once the index is ACTIVE on this backend, so it runs + // in its own task while this one watches DescribeTable. + let update: extenddb_core::types::UpdateTableInput = serde_json::from_value(json!({ + "TableName": "t", + "AttributeDefinitions": [{"AttributeName": "g", "AttributeType": "S"}], + "GlobalSecondaryIndexUpdates": [{"Create": { + "IndexName": "gidx", + "KeySchema": [{"AttributeName": "g", "KeyType": "HASH"}], + "Projection": {"ProjectionType": "ALL"} + }}] + })) + .expect("update input"); + let updater = engine.clone(); + let task = tokio::spawn(async move { updater.update_table_impl(ACCOUNT, update).await }); + + // Phases seen, in order: 1 is Backfilling false, 2 is true, 3 is ACTIVE. + // A reading counts only when DescribeTable agrees before and after the + // read, so a phase change between the two calls cannot pair one phase's + // description with the other's refusal. + let mut seen: Vec = Vec::new(); + let deadline = Instant::now() + Duration::from_secs(30); + while seen.last() != Some(&3) { + assert!(Instant::now() < deadline, "no ACTIVE state; seen {seen:?}"); + let before = gsi_state(&engine).await; + let read = engine + .index_info_by_table_id(&key_info.table_id, "gidx") + .await; + if before.is_none() || gsi_state(&engine).await != before { + tokio::time::sleep(Duration::from_millis(20)).await; + continue; + } + let phase = match before.expect("checked above") { + (status, Some(serde_json::Value::Bool(false))) if status == "CREATING" => { + assert!( + matches!(read, Err(StorageError::IndexNotFound(ref n)) if n == "gidx"), + "before the backfill a read is told there is no such index: {read:?}" + ); + 1 + } + (status, Some(serde_json::Value::Bool(true))) if status == "CREATING" => { + assert!( + matches!(read, Err(StorageError::Validation(ref m)) if m == BACKFILLING), + "during the backfill a read gets the backfilling sentence: {read:?}" + ); + 2 + } + (status, None) if status == "ACTIVE" => { + read.expect("an ACTIVE index resolves"); + 3 + } + other => panic!("a state the service does not report: {other:?}"), + }; + if seen.last() != Some(&phase) { + assert!( + seen.last().is_none_or(|&last| last < phase), + "phase {phase} after {seen:?}" + ); + seen.push(phase); + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + assert_eq!(seen, vec![1, 2, 3], "each phase is observed, in order"); + task.await + .expect("the UpdateTable task") + .expect("UpdateTable succeeds"); +} diff --git a/crates/storage-sqlite/src/lib.rs b/crates/storage-sqlite/src/lib.rs index 7e8085ff9..1327ca5c6 100644 --- a/crates/storage-sqlite/src/lib.rs +++ b/crates/storage-sqlite/src/lib.rs @@ -30,6 +30,8 @@ mod create_table; mod credential_store; mod data; mod delete_table; +#[cfg(test)] +mod gsi_backfilling_tests; mod hooks; mod management_store; mod metadata; diff --git a/crates/storage-sqlite/src/store.rs b/crates/storage-sqlite/src/store.rs index 075d25b28..2d16864b4 100644 --- a/crates/storage-sqlite/src/store.rs +++ b/crates/storage-sqlite/src/store.rs @@ -79,6 +79,10 @@ pub struct SqliteEngine { /// no entry, because nothing else ever will until a restart, and until then /// the per-table queue hold blocks every write's index maintenance. pub(crate) vector_builds_running: Arc>>, + /// GSI builds whose backfill has started, which is what DescribeTable + /// reports as `Backfilling` and what decides a Query or Scan refusal for a + /// `CREATING` GSI; see `extenddb_storage::gsi_backfill`. + pub(crate) gsi_backfills: extenddb_storage::gsi_backfill::GsiBackfills, /// The `extenddb serve` lock on the database file, held for as long as any /// clone of the engine lives. `None` outside `serve` and for in-memory /// databases. @@ -181,6 +185,7 @@ impl SqliteEngine { vector_builds_running: Arc::new( std::sync::Mutex::new(std::collections::HashSet::new()), ), + gsi_backfills: extenddb_storage::gsi_backfill::GsiBackfills::default(), serve_lock: None, backup_server_slot_count, backup_server_slots: Arc::new(tokio::sync::Semaphore::new(backup_server_slot_count)), @@ -388,6 +393,22 @@ impl SqliteEngine { } } + /// The `gsi_creating_phase_delay_ms` test hold for one table; zero when unset. + pub(crate) async fn gsi_creating_phase_delay(&self, table_name: &str) -> std::time::Duration { + let live: Result, _> = + sqlx::query_as("SELECT value FROM settings WHERE key = ?") + .bind(extenddb_core::settings_keys::gsi_creating_phase_delay_key( + table_name, + )) + .fetch_optional(&self.pool) + .await; + let value = live.unwrap_or_else(|e| { + tracing::debug!("gsi_creating_phase_delay: live read failed, using 0: {e:?}"); + None + }); + extenddb_storage::gsi_backfill::phase_delay(value.as_ref().map(|(v,)| v.as_str())) + } + /// Handle to the GSI propagation notifier, woken after an enqueue. pub(crate) fn gsi_notify(&self) -> Arc { Arc::clone(&self.gsi_notify) diff --git a/crates/storage-sqlite/src/table_helpers.rs b/crates/storage-sqlite/src/table_helpers.rs index 4827391f2..6a6f5ac06 100644 --- a/crates/storage-sqlite/src/table_helpers.rs +++ b/crates/storage-sqlite/src/table_helpers.rs @@ -44,7 +44,6 @@ pub(crate) struct TableRow { #[derive(sqlx::FromRow)] pub(crate) struct IndexRow { pub index_name: String, - #[allow(dead_code)] pub index_id: String, pub index_type: String, pub key_schema: String, @@ -158,6 +157,8 @@ impl SqliteEngine { account_id: &str, table_name: &str, ) -> Result { + // Before the catalog rows; see `GsiBackfills::snapshot`. + let started = self.gsi_backfills.snapshot(); let row: Option = sqlx::query_as(&format!( "SELECT {TABLE_COLUMNS} FROM tables WHERE account_id = ? AND table_name = ?" )) @@ -187,7 +188,8 @@ impl SqliteEngine { let table_name_owned = row.table_name.clone(); let table_id = row.table_id.clone(); - let mut desc = self.build_table_description_from_row(account_id, row, index_rows)?; + let mut desc = + self.build_table_description_from_row(account_id, row, index_rows, &started)?; desc.restore_summary = Self::restore_summary_on(conn, &table_id, &desc.table_status).await?; let catalog_rows = vector_rows @@ -199,11 +201,16 @@ impl SqliteEngine { Ok(desc) } + /// `started` is a [`GsiBackfills::snapshot`] the caller took before it read + /// `index_rows`. + /// + /// [`GsiBackfills::snapshot`]: extenddb_storage::gsi_backfill::GsiBackfills::snapshot pub(crate) fn build_table_description_from_row( &self, account_id: &str, row: TableRow, index_rows: Vec, + started: &std::collections::HashSet, ) -> Result { let mut gsis: Vec = Vec::new(); let mut lsis: Vec = Vec::new(); @@ -221,6 +228,10 @@ impl SqliteEngine { index_name: idx.index_name.clone(), key_schema: ks, projection: proj, + backfilling: extenddb_storage::gsi_backfill::backfilling_member( + &idx.index_status, + started.contains(&idx.index_id), + ), index_status: idx.index_status, provisioned_throughput: pt.or(Some(zero_throughput())), index_size_bytes: 0, diff --git a/crates/storage-sqlite/src/update_table.rs b/crates/storage-sqlite/src/update_table.rs index 8adff6173..dbf9af7db 100644 --- a/crates/storage-sqlite/src/update_table.rs +++ b/crates/storage-sqlite/src/update_table.rs @@ -715,6 +715,20 @@ impl SqliteEngine { if let Some(create) = &update.create { let idx_id = created[ci].clone(); ci += 1; + // The row committed above as CREATING reports `Backfilling: + // false` until the record below is taken, and `true` from then + // until the flip to ACTIVE. The test hold lengthens each phase; + // it runs under the engine write lock, so it also delays every + // other write in this process while it lasts. + let hold = self.gsi_creating_phase_delay(&input.table_name).await; + if !hold.is_zero() { + tokio::time::sleep(hold).await; + } + // The record ends with this block. A build that fails below + // deletes its row. If that DELETE fails, or the flip to ACTIVE + // fails, the row stays CREATING with no record and reports the + // first phase until the startup reconciler rebuilds it. + let _backfilling = self.gsi_backfills.start_scoped(&idx_id); let result = async { let mut data_tx = self .pool @@ -763,6 +777,12 @@ impl SqliteEngine { .await; return Err(e); } + // The second hold sits between the build's commit and the flip, + // as on PostgreSQL, so a record that ended before the flip would + // show as the first phase while it lasts. + if !hold.is_zero() { + tokio::time::sleep(hold).await; + } // Backfill succeeded and the data table is populated, so the // index is now queryable: flip CREATING -> ACTIVE. DescribeTable // reports CREATING until this point (matching DynamoDB), so a diff --git a/crates/storage/src/gsi_backfill.rs b/crates/storage/src/gsi_backfill.rs new file mode 100644 index 000000000..93fb919be --- /dev/null +++ b/crates/storage/src/gsi_backfill.rs @@ -0,0 +1,226 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! The two `CREATING` phases of a GSI that `UpdateTable` adds. +//! +//! Measured against the service on 2026-10-26 (us-east-1) with a GSI added to +//! a 3000-item table, which became `ACTIVE` after 512 s and 519 s on two runs: +//! +//! | `IndexStatus` | `Backfilling` | Query or Scan naming the index | +//! |---|---|---| +//! | `CREATING` | `false` | `The table does not have the specified index: ` | +//! | `CREATING` | `true` | `Cannot read from backfilling global secondary index: ` | +//! | `ACTIVE` | absent | served | +//! +//! Both refusals are `ValidationException`. The first phase lasted about 30 s. +//! +//! Each backend decides from its own build state whether the backfill has +//! started and passes that one answer to [`backfilling_member`] for +//! DescribeTable and to [`creating_read_refusal`] for Query and Scan, so the +//! description and the refusal cannot disagree. + +use std::collections::HashSet; +use std::sync::{Arc, Mutex, MutexGuard, PoisonError}; +use std::time::Duration; + +use extenddb_core::types::VECTOR_INDEX_BACKFILLING_SCAN_PREFIX; + +use crate::error::StorageError; + +/// The `Backfilling` member of a GSI whose status is `index_status`: present +/// only while the index is `CREATING`, and then `true` once its backfill has +/// started. +#[must_use] +pub fn backfilling_member(index_status: &str, backfill_started: bool) -> Option { + (index_status == "CREATING").then_some(backfill_started) +} + +/// The refusal for a Query or Scan naming a `CREATING` GSI. +/// +/// Before the backfill starts it is `IndexNotFound`, which the engine reports +/// as "The table does not have the specified index", the same answer as for a +/// name that matches no index. After, it is the backfilling sentence, which the +/// engine passes through verbatim as a `ValidationException`. +#[must_use] +pub fn creating_read_refusal(index_name: &str, backfill_started: bool) -> StorageError { + if backfill_started { + StorageError::Validation(format!( + "{VECTOR_INDEX_BACKFILLING_SCAN_PREFIX}{index_name}" + )) + } else { + StorageError::IndexNotFound(index_name.to_owned()) + } +} + +/// The hold a `gsi_creating_phase_delay_ms:
` setting asks for, from its +/// stored value. Absent or unreadable means no hold. +#[must_use] +pub fn phase_delay(value: Option<&str>) -> Duration { + Duration::from_millis( + value + .and_then(|v| v.trim().parse::().ok()) + .unwrap_or(0), + ) +} + +/// GSI builds in this process whose backfill has started, by index id. +/// +/// For a backend whose build runs inside the server process and records no +/// progress in its catalog (PostgreSQL and SQLite). One process serves a +/// catalog, which is the supported deployment, so the process running a build +/// is also the one answering DescribeTable for it. +/// +/// An id stays in the set until its index leaves `CREATING`: a build removes it +/// once it has published the index, found it deleted, or removed it after a +/// failure that would recur. A build that fails for a passing reason leaves it, +/// so the index keeps reporting `Backfilling: true` and the backfilling refusal +/// while it waits for the retry, which records it again; the service is not +/// known to go back to the first phase. Index ids are not reused, so an id left +/// behind by an index that another path deleted is never read again. Clones +/// share the set. +#[derive(Clone, Debug, Default)] +pub struct GsiBackfills(Arc>>); + +impl GsiBackfills { + /// Record that a backfill of `index_id` has started. The record stays + /// until [`Self::finish`] removes it. + pub fn start(&self, index_id: &str) { + self.lock().insert(index_id.to_owned()); + } + + /// Remove the record of `index_id`, once its index has left `CREATING`. + pub fn finish(&self, index_id: &str) { + self.lock().remove(index_id); + } + + /// [`Self::start`], and [`Self::finish`] when the returned guard drops, for + /// a build that deletes its index in the same call whenever it fails. + #[must_use] + pub fn start_scoped(&self, index_id: &str) -> BackfillGuard { + self.start(index_id); + BackfillGuard { + set: self.clone(), + index_id: index_id.to_owned(), + } + } + + /// Whether a backfill of `index_id` has started in this process and its + /// index is still `CREATING`. + #[must_use] + pub fn started(&self, index_id: &str) -> bool { + self.lock().contains(index_id) + } + + /// The ids recorded now, for a reader to take before it reads the catalog + /// rows it pairs them with. + /// + /// A build removes an id only after the index's row has left `CREATING`, + /// so a set read first can be paired only with the same row or a newer + /// one. A `CREATING` row is then reported with a set that still holds its + /// id, and the answer never goes back to the first phase. Read the other + /// way round, a call could read the row as `CREATING` just before the flip + /// and the set just after the id was removed, and report the first phase + /// for an index whose backfill had finished. + /// + /// SQLite has two exceptions, both after a failure. Its record lasts as + /// long as the build inside UpdateTable, so when the flip to `ACTIVE` + /// fails, or the cleanup `DELETE` after a failed build fails, the id is + /// removed while the row stays `CREATING`, and the index reports the first + /// phase until the startup reconciler rebuilds it. + #[must_use] + pub fn snapshot(&self) -> HashSet { + self.lock().clone() + } + + fn lock(&self) -> MutexGuard<'_, HashSet> { + // A panic while the lock was held leaves a set that is still usable, so + // a poisoned lock is taken as it is. + self.0.lock().unwrap_or_else(PoisonError::into_inner) + } +} + +/// Removes one backfill record from [`GsiBackfills`] when dropped. +#[derive(Debug)] +pub struct BackfillGuard { + set: GsiBackfills, + index_id: String, +} + +impl Drop for BackfillGuard { + fn drop(&mut self) { + self.set.finish(&self.index_id); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn the_member_is_present_only_while_creating() { + assert_eq!(backfilling_member("CREATING", false), Some(false)); + assert_eq!(backfilling_member("CREATING", true), Some(true)); + assert_eq!(backfilling_member("ACTIVE", true), None); + assert_eq!(backfilling_member("ACTIVE", false), None); + assert_eq!(backfilling_member("DELETING", true), None); + } + + #[test] + fn the_refusal_follows_the_phase_byte_for_byte() { + match creating_read_refusal("gsi_slow", false) { + StorageError::IndexNotFound(name) => assert_eq!(name, "gsi_slow"), + other => panic!("first phase: {other:?}"), + } + match creating_read_refusal("gsi_slow", true) { + StorageError::Validation(msg) => assert_eq!( + msg, + "Cannot read from backfilling global secondary index: gsi_slow" + ), + other => panic!("second phase: {other:?}"), + } + } + + #[test] + fn a_record_lasts_until_finish() { + let running = GsiBackfills::default(); + assert!(!running.started("i1")); + running.start("i1"); + assert!(running.clone().started("i1")); + assert!(!running.started("i2")); + // A second build of the same index (the retry after a passing failure) + // records it again; one finish ends it, because finish is called only + // once the index has left CREATING. + running.start("i1"); + running.finish("i1"); + assert!(!running.started("i1")); + } + + #[test] + fn a_scoped_record_lasts_as_long_as_its_guard() { + let running = GsiBackfills::default(); + let guard = running.start_scoped("i1"); + assert!(running.started("i1")); + drop(guard); + assert!(!running.started("i1")); + } + + #[test] + fn a_snapshot_keeps_the_ids_recorded_when_it_was_taken() { + let running = GsiBackfills::default(); + running.start("i1"); + let before = running.snapshot(); + running.finish("i1"); + running.start("i2"); + assert!(before.contains("i1")); + assert!(!before.contains("i2")); + assert!(!running.started("i1")); + } + + #[test] + fn an_absent_or_unreadable_delay_holds_nothing() { + assert_eq!(phase_delay(None), Duration::ZERO); + assert_eq!(phase_delay(Some("not a number")), Duration::ZERO); + assert_eq!(phase_delay(Some("0")), Duration::ZERO); + assert_eq!(phase_delay(Some(" 2500 ")), Duration::from_millis(2500)); + } +} diff --git a/crates/storage/src/lib.rs b/crates/storage/src/lib.rs index b7aec12ff..181ec7ab7 100755 --- a/crates/storage/src/lib.rs +++ b/crates/storage/src/lib.rs @@ -15,6 +15,7 @@ pub mod config; pub mod diagnostics; pub mod diagnostics_store; pub mod error; +pub mod gsi_backfill; pub mod hooks; pub mod management_store; pub mod management_store_factory; diff --git a/devtools/run-mongodb-tests b/devtools/run-mongodb-tests index 2a287798a..c510747e0 100755 --- a/devtools/run-mongodb-tests +++ b/devtools/run-mongodb-tests @@ -279,13 +279,14 @@ fi echo "" if [[ " ${RUN_TESTS_ARGS[*]} " == *" --rust"* ]]; then # The storage-level MongoDB tests that need a live server (the stream - # records page over 16 MB and the shard ownership predicates) are - # #[ignore]d in the crate and read their server from this variable; this - # container is the one place in CI that has one. - echo "=== Running MongoDB storage-level live tests (records_live) ===" + # records page over 16 MB, the shard ownership predicates, and the phases + # of a GSI that UpdateTable adds) are #[ignore]d in the crate and read + # their server from this variable; this container is the one place in CI + # that has one. + echo "=== Running MongoDB storage-level live tests (records_live, live_gsi_) ===" echo "" EXTENDDB_TEST_MONGODB_URI="mongodb://localhost:$MONGO_PORT/?replicaSet=rs0&directConnection=true" \ - cargo test -p extenddb-storage-mongodb --lib -- --ignored records_live + cargo test -p extenddb-storage-mongodb --lib -- --ignored records_live live_gsi_ echo "" fi echo "=== Running test suite via devtools/run-tests --backend mongodb ===" diff --git a/docs/manuals/05-admin-guide.md b/docs/manuals/05-admin-guide.md index a0cf56e7e..cae4e2629 100755 --- a/docs/manuals/05-admin-guide.md +++ b/docs/manuals/05-admin-guide.md @@ -189,6 +189,7 @@ Managed via `extenddb settings set`. Changes take effect within 30 seconds witho | `allow_credential_import` | `true` | Whether `import-access-key` is allowed | | `vector_backfill_batch_delay_ms` | `0` | **Test-oriented.** Milliseconds to pause between batches while a vector index backfills. Zero in production. A test sets it so a write is guaranteed to land while the index is still building. The pause is outside the batch transaction, so writes are still accepted throughout, but it does extend the per-table propagation hold: no index on that table advances while the build runs, GSIs included, and the accepted range goes up to 60 s per batch. | | `vector_allocation_phase_delay_ms` | `0` | **Test-oriented.** Milliseconds to hold a new vector index in the resource-allocation phase (`CREATING` with `Backfilling: false`) before the scan starts. Zero in production. Without it the phase lasts only from the `UpdateTable` transaction, which inserts the row as `CREATING` with `Backfilling: false`, until the detached build task flips the flag, which is a window no client can time reliably rather than one that cannot exist. | +| `gsi_creating_phase_delay_ms:
` | unset | **Test-oriented.** Milliseconds, 0 to 60000, by which a GSI that `UpdateTable` adds to `
` lengthens each of its two `CREATING` phases: before its backfill starts (`Backfilling: false`, reads told the index does not exist) and after (`Backfilling: true`, reads refused as backfilling). Unset in production. On SQLite the hold runs under the engine write lock, so every write to the database waits for it; on MongoDB and Cassandra it also delays the other index builds the worker has queued. The row stays in the catalog after a test sets it back to `0`. | ```bash # View all settings diff --git a/tests/python/test_gsi_backfilling.py b/tests/python/test_gsi_backfilling.py new file mode 100644 index 000000000..8eeee7d74 --- /dev/null +++ b/tests/python/test_gsi_backfilling.py @@ -0,0 +1,167 @@ +# Copyright 2026 ExtendDB contributors +# SPDX-License-Identifier: Apache-2.0 + +"""A GSI added by UpdateTable walks the two CREATING phases the service reports. + +Measured against Amazon DynamoDB on 2026-10-26 (us-east-1, a 3000-item table, +two runs) with whole strings, both refusals ValidationException: + +- CREATING, ``Backfilling: false``: Query and Scan on the index answer + ``The table does not have the specified index: gsi_slow`` +- CREATING, ``Backfilling: true``: both answer + ``Cannot read from backfilling global secondary index: gsi_slow`` +- ACTIVE: ``Backfilling`` is absent and both are served + +The CREATING description carries exactly the members in ``CREATING_MEMBERS``. + +On the service the first phase lasted about 30 s and the second about 480 s. +ExtendDB builds a three-item index in milliseconds, so the test sets the +``gsi_creating_phase_delay_ms:
`` hold, which lengthens both phases, and +is therefore ExtendDB-only. Setting it needs the admin credentials that +``devtools/run-tests`` exports. +""" + +from __future__ import annotations + +import os +import threading +import time + +import botocore.exceptions +import pytest +import requests + +from helpers import make_dynamodb_client + +GSI = "gsi_slow" +NOT_FOUND = f"The table does not have the specified index: {GSI}" +BACKFILLING = f"Cannot read from backfilling global secondary index: {GSI}" +CREATING_MEMBERS = { + "Backfilling", + "IndexArn", + "IndexName", + "IndexSizeBytes", + "IndexStatus", + "ItemCount", + "KeySchema", + "Projection", + "ProvisionedThroughput", +} +HOLD_MS = 2000 +ABSENT = "absent" + +_extenddb_only = pytest.mark.skipif( + os.environ.get("EXTENDDB_VALIDATION_MODE", "").lower() == "true", + reason="extenddb_only: skipped in validation mode", +) + + +def _set_hold(endpoint: str, table: str, ms: int) -> None: + password = os.environ.get("EXTENDDB_ADMIN_PASSWORD", "").strip() + if not password: + pytest.fail("EXTENDDB_ADMIN_PASSWORD is required to set the GSI phase hold") + user = os.environ.get("EXTENDDB_ADMIN_USER", "admin") + ca_cert = os.environ.get("EXTENDDB_CA_CERT", "") + response = requests.put( + f"{endpoint.rstrip('/')}/management/settings/gsi_creating_phase_delay_ms:{table}", + auth=(user, password), + json={"value": str(ms)}, + timeout=30, + verify=ca_cert if ca_cert else False, + ) + assert response.ok, f"setting the phase hold failed: {response.status_code} {response.text}" + + +def _outcome(call, **kwargs): + """``("OK", count)`` for a served read, ``(code, message)`` for a refusal.""" + try: + return ("OK", call(**kwargs)["Count"]) + except botocore.exceptions.ClientError as e: + return (e.response["Error"]["Code"], e.response["Error"]["Message"]) + + +def _gsi(client, table: str): + gsis = client.describe_table(TableName=table)["Table"].get("GlobalSecondaryIndexes", []) + return next((g for g in gsis if g["IndexName"] == GSI), None) + + +@_extenddb_only +def test_added_gsi_reports_backfilling_and_refuses_reads_by_phase( + dynamodb_client, endpoint_url, table_factory +): + if endpoint_url is None: + pytest.skip("the phase hold is an ExtendDB setting") + table = table_factory() + for n in range(3): + dynamodb_client.put_item( + TableName=table, Item={"pk": {"S": f"p{n}"}, "g": {"S": "v"}} + ) + _set_hold(endpoint_url, table, HOLD_MS) + + # UpdateTable answers only once the index is ACTIVE on some backends, so it + # runs on its own client in a thread while this one watches the phases. + failures: list[Exception] = [] + + def add_gsi() -> None: + try: + make_dynamodb_client(endpoint_url).update_table( + TableName=table, + AttributeDefinitions=[{"AttributeName": "g", "AttributeType": "S"}], + GlobalSecondaryIndexUpdates=[{"Create": { + "IndexName": GSI, + "KeySchema": [{"AttributeName": "g", "KeyType": "HASH"}], + "Projection": {"ProjectionType": "ALL"}, + }}], + ) + except Exception as e: # reported after the join + failures.append(e) + + updater = threading.Thread(target=add_gsi) + updater.start() + try: + # Phases seen, in order: 1 is Backfilling false, 2 is true, 3 is ACTIVE. + # A reading counts only when DescribeTable agrees before and after the + # two reads, so a phase change between them is read again. + seen: list[int] = [] + deadline = time.monotonic() + 120 + while seen[-1:] != [3]: + assert time.monotonic() < deadline, f"no ACTIVE state; seen {seen}" + before = _gsi(dynamodb_client, table) + query = _outcome( + dynamodb_client.query, + TableName=table, + IndexName=GSI, + KeyConditionExpression="g = :v", + ExpressionAttributeValues={":v": {"S": "v"}}, + ) + scan = _outcome(dynamodb_client.scan, TableName=table, IndexName=GSI) + after = _gsi(dynamodb_client, table) + if before is None or before != after: + time.sleep(0.05) + continue + state = (before["IndexStatus"], before.get("Backfilling", ABSENT)) + if state == ("CREATING", False): + assert set(before) == CREATING_MEMBERS, sorted(before) + assert query == ("ValidationException", NOT_FOUND), query + assert scan == ("ValidationException", NOT_FOUND), scan + phase = 1 + elif state == ("CREATING", True): + assert set(before) == CREATING_MEMBERS, sorted(before) + assert query == ("ValidationException", BACKFILLING), query + assert scan == ("ValidationException", BACKFILLING), scan + phase = 2 + elif state == ("ACTIVE", ABSENT): + assert query == ("OK", 3), query + assert scan == ("OK", 3), scan + phase = 3 + else: + pytest.fail(f"a state the service does not report: {state}") + if seen[-1:] != [phase]: + assert not seen or seen[-1] < phase, f"phase {phase} after {seen}" + seen.append(phase) + time.sleep(0.05) + assert seen == [1, 2, 3], seen + finally: + updater.join(timeout=120) + _set_hold(endpoint_url, table, 0) + assert not failures, failures diff --git a/tests/rust/src/gsi_on_hash_only_table.rs b/tests/rust/src/gsi_on_hash_only_table.rs index ed4f34d46..d5e40abec 100644 --- a/tests/rust/src/gsi_on_hash_only_table.rs +++ b/tests/rust/src/gsi_on_hash_only_table.rs @@ -14,11 +14,30 @@ use crate::test_base::*; use aws_sdk_dynamodb::types::{ - AttributeDefinition, CreateGlobalSecondaryIndexAction, GlobalSecondaryIndexUpdate, + AttributeDefinition, CreateGlobalSecondaryIndexAction, GlobalSecondaryIndexUpdate, IndexStatus, KeySchemaElement, KeyType, Projection, ProjectionType, ScalarAttributeType, }; use std::time::Duration; +/// Poll DescribeTable until the named index reports ACTIVE (up to 60s). +async fn wait_for_index_active(c: &aws_sdk_dynamodb::Client, table: &str, index_name: &str) { + for _ in 0..60 { + if let Ok(resp) = c.describe_table().table_name(table).send().await { + if let Some(t) = resp.table() { + let active = t.global_secondary_indexes().iter().any(|i| { + i.index_name() == Some(index_name) + && i.index_status() == Some(&IndexStatus::Active) + }); + if active { + return; + } + } + } + tokio::time::sleep(Duration::from_secs(1)).await; + } + panic!("Index {index_name} on {table} did not become ACTIVE within 60s"); +} + const GSI_BACKFILL_TEST_GATE: &str = "gsi_backfill_test_gate"; fn backfill_gate_key(table_name: &str) -> String { @@ -163,6 +182,10 @@ async fn gsi_can_be_added_to_a_table_with_no_sort_key() { ) }); wait_for_active(c, &table).await; + // The table is ACTIVE again before the index is: the service refuses a + // Query naming an index that is still CREATING, and so do the backends + // whose UpdateTable returns before the backfill ends. + wait_for_index_active(c, &table, "hashOnlyIdx").await; // The index must exist on the table description, not merely have been // accepted by UpdateTable.