From 2447c32b6dc2e55be75e0044899d7e6f1c9e33a2 Mon Sep 17 00:00:00 2001 From: Lee Hannigan <29015211+LeeroyHannigan@users.noreply.github.com> Date: Thu, 5 Nov 2026 18:22:50 +0000 Subject: [PATCH] fix(storage-cassandra): recover a GSI backfill only in its owner's keyspace recover_stale_backfills runs once for each account keyspace. Each run read every CREATING index of every account. For an index of another account, it dropped and created the index data table in its own keyspace. It also took a propagation hold there and started a backfill. That backfill failed, because the keyspace has no base table, and the data table stayed. The pass now skips an index whose base table belongs to another account. It skips it before it parses the index definition. poll_gsi_backfill_recovery takes the keyspace to scan as an Option. serve passes None and scans every account keyspace, as before. spawn_gsi_backfill_recovery_for_tests takes an account and scans only that account's keyspace. Before, a worker that one test started could run the backfill of an index that another test created. The other test never fired that worker's token. That made a_backfill_the_recovery_worker_started_ends_before_the_worker_returns fail in the cassandra-storage job. test_ttl_enable_rejects_asynchronous_gsi looked for a refusal wording that #384 replaced. It now checks the table, the index and the delay in the current refusal. The TTL step no longer continues on error. The direct step runs after a failed TTL step, so a run reports both suites. --- .github/workflows/integration-cassandra.yml | 15 +- crates/storage-cassandra/src/lib.rs | 1 + crates/storage-cassandra/src/workers.rs | 96 ++++-- .../storage-cassandra/tests/direct/index.rs | 286 +++++++++++++----- .../tests/ttl_integration.rs | 11 +- 5 files changed, 303 insertions(+), 106 deletions(-) diff --git a/.github/workflows/integration-cassandra.yml b/.github/workflows/integration-cassandra.yml index 3d64445b7..536ffae26 100644 --- a/.github/workflows/integration-cassandra.yml +++ b/.github/workflows/integration-cassandra.yml @@ -106,16 +106,19 @@ jobs: env: EXTENDDB_TEST_CASSANDRA: required - name: TTL integration tests - continue-on-error: true # TTL integration tests are known-flaky under CI load; tracked in #338 run: cargo test -p extenddb-storage-cassandra --test ttl_integration -- --test-threads=1 env: EXTENDDB_TEST_CASSANDRA: required - name: Direct storage-trait integration tests - # Ported from the plug-in repo's tests/rust suite. Parallel-safe: - # each test provisions its own account and keyspace; the shared - # control-plane setting test restores what it mutates. The - # test-hooks feature compiles the tests that pause an operation or - # shorten a heartbeat (crates/storage-cassandra/src/test_hooks.rs). + # Ported from the plug-in repo's tests/rust suite. Each test + # provisions its own account and keyspace, and a GSI backfill + # recovery worker a test starts scans only that keyspace. The GSI + # propagation workers that some index tests start read the queue of + # every keyspace. The shared control-plane setting test restores what + # it mutates. The test-hooks feature compiles the tests that pause an + # operation or shorten a heartbeat (crates/storage-cassandra/src/test_hooks.rs). + # It runs after a failed TTL step too, so a run reports both suites. + if: ${{ !cancelled() }} run: cargo test -p extenddb-storage-cassandra --features test-hooks --test direct_integration env: EXTENDDB_TEST_CASSANDRA: required diff --git a/crates/storage-cassandra/src/lib.rs b/crates/storage-cassandra/src/lib.rs index 3f0489402..995abc3d4 100644 --- a/crates/storage-cassandra/src/lib.rs +++ b/crates/storage-cassandra/src/lib.rs @@ -132,6 +132,7 @@ impl ServerRuntimeHooks for CassandraRuntimeHooks { gsi_backfill_notify, std::time::Duration::from_secs(120), // stale threshold std::time::Duration::from_secs(60), // scan interval backstop + None, // every account keyspace token, ) .await; diff --git a/crates/storage-cassandra/src/workers.rs b/crates/storage-cassandra/src/workers.rs index 32c9c665d..cc23fea0f 100644 --- a/crates/storage-cassandra/src/workers.rs +++ b/crates/storage-cassandra/src/workers.rs @@ -515,11 +515,14 @@ async fn gsi_apply_index( /// same way: drop+recreate the data table and rerun the backfill. Once /// `shutdown` fires it takes no further index, and it returns only after the /// backfills it started have ended, so the server's stop covers them too. +/// +/// It scans every account keyspace, or only `only_keyspace` when one is given. pub(crate) async fn poll_gsi_backfill_recovery( engine: Arc, notify: Arc, stale_threshold: Duration, scan_interval: Duration, + only_keyspace: Option, shutdown: CancellationToken, ) { let mut backfills = tokio::task::JoinSet::new(); @@ -532,13 +535,16 @@ pub(crate) async fn poll_gsi_backfill_recovery( // Collect the backfills that have ended. while backfills.try_join_next().is_some() {} - let keyspaces = match list_account_keyspaces(&engine).await { + let mut keyspaces = match list_account_keyspaces(&engine).await { Ok(ks) => ks, Err(e) => { tracing::warn!("gsi_recovery: list keyspaces failed: {e}"); continue; } }; + if let Some(only) = &only_keyspace { + keyspaces.retain(|account_ks| account_ks == only); + } for account_ks in &keyspaces { if shutdown.is_cancelled() { @@ -563,28 +569,62 @@ pub(crate) async fn poll_gsi_backfill_recovery( } /// The GSI backfill recovery worker `serve` runs, with the stale threshold -/// and scan interval given, until `shutdown` fires; for tests. +/// and scan interval given, until `shutdown` fires; for tests. It scans only +/// the keyspace of `account_id`: the tests of one binary share one catalog, +/// and a worker that scanned every keyspace would also run the backfills of +/// the indexes other tests create. #[cfg(any(test, feature = "test-util"))] #[doc(hidden)] #[must_use] pub fn spawn_gsi_backfill_recovery_for_tests( engine: Arc, + account_id: &str, stale_threshold: Duration, scan_interval: Duration, shutdown: CancellationToken, ) -> tokio::task::JoinHandle<()> { let notify = engine.gsi_backfill_notify.clone(); + let keyspace = engine.account_keyspace(account_id); tokio::spawn(poll_gsi_backfill_recovery( engine, notify, stale_threshold, scan_interval, + Some(keyspace), shutdown, )) } -/// Rerun the stale backfills of `account_ks`, each in a task of -/// `backfills` that stops once `shutdown` fires. +/// One pass of the GSI backfill recovery over the keyspace of `account_id`, +/// returning once the backfills it started have ended; for tests. +/// +/// # Errors +/// +/// The error that ended the pass, after its backfills have ended. +#[cfg(any(test, feature = "test-util"))] +#[doc(hidden)] +pub async fn recover_stale_gsi_backfills_for_tests( + engine: &Arc, + account_id: &str, + stale_threshold: Duration, +) -> Result<(), extenddb_storage::error::StorageError> { + let mut backfills = tokio::task::JoinSet::new(); + let pass = recover_stale_backfills( + engine, + &engine.account_keyspace(account_id), + stale_threshold, + &CancellationToken::new(), + &mut backfills, + ) + .await; + while backfills.join_next().await.is_some() {} + pass +} + +/// Rerun the stale backfills of the indexes of tables in `account_ks`, each +/// in a task of `backfills` that stops once `shutdown` fires. An index of a +/// table in another account's keyspace is left to the pass over that +/// keyspace. async fn recover_stale_backfills( engine: &Arc, account_ks: &str, @@ -634,28 +674,12 @@ async fn recover_stale_backfills( let proj_json: String = crate::cassandra_util::get_column(&row, "projection", "gsi_recovery")?; - let index_key_schema: Vec = - match serde_json::from_str(&ks_json) { - Ok(v) => v, - Err(e) => { - tracing::warn!("gsi_recovery: bad key_schema for {index_name}: {e}"); - continue; - } - }; - let projection: extenddb_core::types::Projection = match serde_json::from_str(&proj_json) { - Ok(v) => v, - Err(e) => { - tracing::warn!("gsi_recovery: bad projection for {index_name}: {e}"); - continue; - } - }; - - // 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 \ FROM {catalog_ks}.tables WHERE table_id = ? ALLOW FILTERING" ), cdrs_tokio::query_values!(table_id.as_str()), @@ -676,6 +700,34 @@ async fn recover_stale_backfills( } }; + // The catalog holds the indexes of every account. Another account's + // index is recovered by the pass over its own keyspace; recovering it + // here would create its data table in this keyspace and run a + // backfill of a base table this keyspace does not hold. It is + // skipped before its definition is parsed, so a malformed row is + // reported by its owner's pass alone. + let owner: String = + crate::cassandra_util::get_column(&base_row, "account_id", "gsi_recovery_base")?; + if engine.account_keyspace(&owner) != account_ks { + continue; + } + + let index_key_schema: Vec = + match serde_json::from_str(&ks_json) { + Ok(v) => v, + Err(e) => { + tracing::warn!("gsi_recovery: bad key_schema for {index_name}: {e}"); + continue; + } + }; + let projection: extenddb_core::types::Projection = match serde_json::from_str(&proj_json) { + Ok(v) => v, + Err(e) => { + tracing::warn!("gsi_recovery: bad projection for {index_name}: {e}"); + 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( diff --git a/crates/storage-cassandra/tests/direct/index.rs b/crates/storage-cassandra/tests/direct/index.rs index 42fd7455e..d7fa605c9 100644 --- a/crates/storage-cassandra/tests/direct/index.rs +++ b/crates/storage-cassandra/tests/direct/index.rs @@ -1547,15 +1547,60 @@ async fn index_rows(engine: &CassandraEngine, account_keyspace: &str, index_id: count.expect("a count") } +/// The `UpdateTable` input that creates the GSI `StatusIndex`, keyed on +/// `status`, on the table `table_name`. +fn status_index_creation(table_name: &str) -> extenddb_core::types::UpdateTableInput { + use extenddb_core::types::{CreateGsiAction, GlobalSecondaryIndexUpdate, UpdateTableInput}; + + UpdateTableInput { + table_name: table_name.to_owned(), + 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: "StatusIndex".to_owned(), + key_schema: vec![KeySchemaElement { + attribute_name: "status".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![ + AttributeDefinition { + attribute_name: "id".to_owned(), + attribute_type: ScalarAttributeType::S, + }, + AttributeDefinition { + attribute_name: "status".to_owned(), + attribute_type: ScalarAttributeType::S, + }, + ]), + stream_specification: None, + table_class: None, + on_demand_throughput: None, + vector_index_updates: None, + } +} + /// A backfill the GSI recovery worker started ends before the worker /// returns: once the worker's token fires, the backfill writes no more index /// rows and does not mark the index ACTIVE, so the server's stop at a lost /// registration, which waits for its workers, covers it. The table holds 40 /// pages, so the backfill is still running when the token fires after its -/// first page; the test fails, rather than passing, if it is not. +/// first page; the test fails, rather than passing, if it is not. The worker +/// scans only this test's keyspace: a worker another test started would +/// also run this backfill, under a token this test does not fire. #[tokio::test] async fn a_backfill_the_recovery_worker_started_ends_before_the_worker_returns() { - use extenddb_core::types::{CreateGsiAction, GlobalSecondaryIndexUpdate, UpdateTableInput}; use extenddb_storage::TableEngine as _; use std::time::Duration; @@ -1611,6 +1656,7 @@ async fn a_backfill_the_recovery_worker_started_ends_before_the_worker_returns() let shutdown = extenddb_storage::CancellationToken::new(); let worker = extenddb_storage_cassandra::workers::spawn_gsi_backfill_recovery_for_tests( engine.clone(), + &table.key_info.account_id, Duration::from_secs(120), Duration::from_secs(60), shutdown.clone(), @@ -1618,43 +1664,7 @@ async fn a_backfill_the_recovery_worker_started_ends_before_the_worker_returns() engine .update_table( &table.key_info.account_id, - UpdateTableInput { - table_name: table.key_info.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: "StatusIndex".to_owned(), - key_schema: vec![KeySchemaElement { - attribute_name: "status".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![ - AttributeDefinition { - attribute_name: "id".to_owned(), - attribute_type: ScalarAttributeType::S, - }, - AttributeDefinition { - attribute_name: "status".to_owned(), - attribute_type: ScalarAttributeType::S, - }, - ]), - stream_specification: None, - table_class: None, - on_demand_throughput: None, - vector_index_updates: None, - }, + status_index_creation(&table.key_info.table_name), ) .await .expect("add the index"); @@ -1832,7 +1842,6 @@ async fn test_a_delete_removes_the_index_row_a_backfill_wrote() { // time, the backfilled row would sit above that tombstone and outlive the // item. use cdrs_tokio::types::IntoRustByName as _; - use extenddb_core::types::{CreateGsiAction, GlobalSecondaryIndexUpdate, UpdateTableInput}; use extenddb_storage::TableEngine as _; use std::time::Duration; @@ -1869,6 +1878,7 @@ async fn test_a_delete_removes_the_index_row_a_backfill_wrote() { let shutdown = extenddb_storage::CancellationToken::new(); let _worker = extenddb_storage_cassandra::workers::spawn_gsi_backfill_recovery_for_tests( engine.clone(), + &table.key_info.account_id, Duration::from_secs(120), Duration::from_secs(60), shutdown.clone(), @@ -1876,43 +1886,7 @@ async fn test_a_delete_removes_the_index_row_a_backfill_wrote() { engine .update_table( &table.key_info.account_id, - UpdateTableInput { - table_name: table.key_info.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: "StatusIndex".to_owned(), - key_schema: vec![KeySchemaElement { - attribute_name: "status".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![ - AttributeDefinition { - attribute_name: "id".to_owned(), - attribute_type: ScalarAttributeType::S, - }, - AttributeDefinition { - attribute_name: "status".to_owned(), - attribute_type: ScalarAttributeType::S, - }, - ]), - stream_specification: None, - table_class: None, - on_demand_throughput: None, - vector_index_updates: None, - }, + status_index_creation(&table.key_info.table_name), ) .await .expect("add the index"); @@ -1968,6 +1942,164 @@ async fn test_a_delete_removes_the_index_row_a_backfill_wrote() { shutdown.cancel(); } +/// The tables of `keyspace`, from the schema. +async fn keyspace_tables(engine: &CassandraEngine, keyspace: &str) -> Vec { + use cdrs_tokio::types::IntoRustByName as _; + + extenddb_storage_cassandra::cassandra_util::query_rows::( + &engine.session_arc(), + "SELECT table_name FROM system_schema.tables WHERE keyspace_name = ?", + cdrs_tokio::query_values!(keyspace), + "keyspace_tables", + ) + .await + .expect("read the keyspace's tables") + .iter() + .filter_map(|row| row.get_by_name("table_name").ok().flatten()) + .collect() +} + +/// The schema id of the table `table` in `keyspace`, or `None` when the +/// table does not exist. A table dropped and created again has a new id. +async fn schema_table_id( + engine: &CassandraEngine, + keyspace: &str, + table: &str, +) -> Option { + use cdrs_tokio::types::IntoRustByName as _; + + extenddb_storage_cassandra::cassandra_util::query_rows::( + &engine.session_arc(), + "SELECT id FROM system_schema.tables WHERE keyspace_name = ? AND table_name = ?", + cdrs_tokio::query_values!(keyspace, table), + "schema_table_id", + ) + .await + .expect("read the table's schema id") + .first() + .and_then(|row| row.get_by_name("id").ok().flatten()) +} + +/// Recovery for one account leaves the indexes of other accounts alone. The +/// catalog holds the indexes of every account. A test's worker that scanned +/// every keyspace would run another test's backfill under its own token, and +/// a pass that took another account's index would create the index's data +/// table in its own keyspace and run a backfill of a base table that +/// keyspace does not hold. The pass over the owner's keyspace then runs the +/// backfill. +#[tokio::test] +async fn recovery_for_one_account_leaves_the_indexes_of_other_accounts_alone() { + use extenddb_storage::TableEngine as _; + use std::time::Duration; + + if crate::helpers::skip_without_cassandra() { + return; + } + let config = test_config(); + let engine = Arc::new(CassandraEngine::new(&config, "us-east-1").await.unwrap()); + let owner = crate::helpers::TestTable::new(&engine, "RecoveryOwnerTable", false).await; + let other = crate::helpers::TestTable::new(&engine, "RecoveryOtherTable", false).await; + tokio::time::sleep(Duration::from_millis(350)).await; + engine + .process_control_plane_transitions() + .await + .expect("activate the tables"); + let mut item = Item::new(); + item.insert("id".to_owned(), AttributeValue::S("x".to_owned())); + item.insert("status".to_owned(), AttributeValue::S("s".to_owned())); + engine + .put_item( + &owner.key_info, + item, + false, + None, + &ExpressionMaps::default(), + None, + ) + .await + .expect("put_item"); + + // No worker runs, so the index stays CREATING with no heartbeat. + engine + .update_table( + &owner.key_info.account_id, + status_index_creation(&owner.key_info.table_name), + ) + .await + .expect("add the index"); + let table_id = owner.key_info.table_id.clone(); + let (status, heartbeat, index_id) = index_state(&engine, &table_id, "StatusIndex").await; + assert_eq!((status.as_str(), heartbeat), ("CREATING", None)); + let index_table = extenddb_storage_cassandra::data::ddl::index_table_name(&index_id); + let owner_keyspace = engine.account_keyspace(&owner.key_info.account_id); + let other_keyspace = engine.account_keyspace(&other.key_info.account_id); + let created = schema_table_id(&engine, &owner_keyspace, &index_table).await; + + // A worker started for the other account passes every second. A recovery + // of the index would drop and create its data table first, then write + // its heartbeat after the first page. + let shutdown = extenddb_storage::CancellationToken::new(); + let worker = extenddb_storage_cassandra::workers::spawn_gsi_backfill_recovery_for_tests( + engine.clone(), + &other.key_info.account_id, + Duration::from_secs(120), + Duration::from_secs(1), + shutdown.clone(), + ); + tokio::time::sleep(Duration::from_secs(3)).await; + let (status, heartbeat, _) = index_state(&engine, &table_id, "StatusIndex").await; + let after = schema_table_id(&engine, &owner_keyspace, &index_table).await; + shutdown.cancel(); + tokio::time::timeout(Duration::from_secs(30), worker) + .await + .expect("the worker returns within 30 s of its token") + .expect("the worker does not panic"); + assert_eq!( + ((status.as_str(), heartbeat), after), + (("CREATING", None), created), + "a worker started for another account recovered the index" + ); + + tokio::time::timeout( + Duration::from_secs(180), + extenddb_storage_cassandra::workers::recover_stale_gsi_backfills_for_tests( + &engine, + &other.key_info.account_id, + Duration::from_secs(120), + ), + ) + .await + .expect("the pass over the other account's keyspace ends within 180 s") + .expect("the pass over the other account's keyspace"); + assert!( + !keyspace_tables(&engine, &other_keyspace) + .await + .contains(&index_table), + "the pass over another account's keyspace created {index_table} there" + ); + let (status, heartbeat, _) = index_state(&engine, &table_id, "StatusIndex").await; + assert_eq!( + (status.as_str(), heartbeat), + ("CREATING", None), + "the pass over another account's keyspace ran the backfill" + ); + + tokio::time::timeout( + Duration::from_secs(180), + extenddb_storage_cassandra::workers::recover_stale_gsi_backfills_for_tests( + &engine, + &owner.key_info.account_id, + Duration::from_secs(120), + ), + ) + .await + .expect("the pass over the owner's keyspace ends within 180 s") + .expect("the pass over the owner's keyspace"); + let (status, _, _) = index_state(&engine, &table_id, "StatusIndex").await; + assert_eq!(status, "ACTIVE", "the pass over the owner's keyspace"); + assert_eq!(index_rows(&engine, &owner_keyspace, &index_id).await, 1); +} + #[tokio::test] async fn test_a_queue_entry_written_over_a_version_ahead_of_the_clock_is_drained() { if crate::helpers::skip_without_cassandra() { diff --git a/crates/storage-cassandra/tests/ttl_integration.rs b/crates/storage-cassandra/tests/ttl_integration.rs index 4f712918a..4b1f4230c 100644 --- a/crates/storage-cassandra/tests/ttl_integration.rs +++ b/crates/storage-cassandra/tests/ttl_integration.rs @@ -2147,8 +2147,17 @@ async fn test_ttl_enable_rejects_asynchronous_gsi() { true, ) .await; + let Err(StorageError::Validation(message)) = result else { + panic!("expected a validation error, got {result:?}"); + }; + // The refusal names the table, the index and its delay; the unit test of + // `async_gsi_ttl_refusal` pins the rest of the wording. + let table_name = &table.key_info.table_name; assert!( - matches!(result, Err(StorageError::Validation(message)) if message.contains("asynchronously propagated GSIs")) + message.contains(&format!("cannot be enabled on table {table_name}:")) + && message.contains("asynchronously propagated global secondary index") + && message.contains("index StatusIndex propagates after 1000 ms"), + "{message}" ); }