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}" ); }