fix(storage-cassandra): recover a GSI backfill only in its owner's keyspace

.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

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;

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<CassandraEngine>,
     notify: Arc<tokio::sync::Notify>,
     stale_threshold: Duration,
     scan_interval: Duration,
+    only_keyspace: Option<String>,
     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<CassandraEngine>,
+    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<CassandraEngine>,
+    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<CassandraEngine>,
     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<extenddb_core::types::KeySchemaElement> =
-            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::<extenddb_storage::error::StorageError>(
                 &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<extenddb_core::types::KeySchemaElement> =
+            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(