Coordinate browser bootstrap across networks before rebuilding indexes

crates/common/ARCHITECTURE.md

@@ -144,6 +144,28 @@ Target gating is used extensively:
   validation. When the YAML declares `local-db-sync` sections and DB callbacks
   are provided, the constructor automatically sets up the local DB and starts
   the sync scheduler.
+- Browser startup starts all networks independently when any persisted target
+  exists, retaining indexes and letting each missing network import its own
+  seed. A fresh DB coordinates missing targets when it owns at least two
+  networks, under a shared Web Lock and their per-network leadership locks.
+  Ownership transfers directly to cold runners, including on failure; networks
+  led in another tab are left to their existing runners. Smaller cohorts use
+  their individual runners, retaining ownership and fresh-DB index deferral.
+  Seeds download in parallel; failed downloads leave only those targets for
+  retry, while successful downloads are rechecked against persisted watermarks
+  and imported together in one bounded sqlite-web session. A globally fresh DB
+  drops non-unique secondary indexes and rebuilds them once after all seeds;
+  additions to a populated DB retain existing indexes. A shared import Web Lock
+  covers watermark/index preflight through commit or cancellation for both
+  cohort and per-network imports; without that ownership, or after SDK lease
+  contention, imports retain indexes. PRIMARY KEY/UNIQUE indexes remain
+  enforced. Cold network loops catch up, analyze their applied data and
+  establish readiness; the cohort does not run an additional `ANALYZE`.
+  Import/manifest failures and unsupported coordination fall back to per-network
+  provisioning; stale-target refresh retains its existing behavior. Coordinated
+  manifest and seed requests are bounded to two minutes each; per-network runner
+  requests keep their existing behavior. Scheduler stop cancels
+  coordination/import cleanup.
 - `LocalDbState` encapsulates the local DB handle, scheduler, `SyncReadiness`
   (tracks which chains have completed a sync cycle), and the set of configured
   chain IDs. Query routing uses `QuerySource::LocalDb` vs

crates/common/src/local_db/query/executor.rs

@@ -23,6 +23,13 @@ pub trait LocalDbQueryExecutor {
             .await
     }
 
+    /// Atomically import a browser bootstrap cohort with a single index rebuild.
+    /// Implementations must not commit individual dumps separately.
+    #[cfg(target_family = "wasm")]
+    async fn execute_sql_dumps(&self, _dumps: Vec<Arc<String>>) -> Result<(), LocalDbQueryError> {
+        Err(LocalDbQueryError::not_implemented("execute_sql_dumps"))
+    }
+
     async fn query_json<T>(&self, stmt: &SqlStatement) -> Result<T, LocalDbQueryError>
     where
         T: FromDbJson;

crates/common/src/raindex_client/local_db/mod.rs

@@ -47,6 +47,12 @@ type ExecuteBatchFn = dyn Fn(
 type ExecuteDumpFn =
     dyn Fn(Arc<String>) -> Pin<Box<dyn Future<Output = Result<(), LocalDbQueryError>> + 'static>>;
 
+#[cfg(target_family = "wasm")]
+type ExecuteDumpsFn =
+    dyn Fn(
+        Vec<Arc<String>>,
+    ) -> Pin<Box<dyn Future<Output = Result<(), LocalDbQueryError>> + 'static>>;
+
 #[cfg(target_family = "wasm")]
 type QueryTextFn =
     dyn Fn(
@@ -89,6 +95,8 @@ pub struct LocalDb {
     execute_batch_fn: FnPtr<ExecuteBatchFn>,
     #[cfg(target_family = "wasm")]
     execute_dump_fn: FnPtr<ExecuteDumpFn>,
+    #[cfg(target_family = "wasm")]
+    execute_dumps_fn: FnPtr<ExecuteDumpsFn>,
     query_text_fn: FnPtr<QueryTextFn>,
     query_json_fn: FnPtr<QueryJsonFn>,
     #[cfg(target_family = "wasm")]
@@ -121,6 +129,14 @@ impl LocalDb {
             })
         };
 
+        let execute_dumps_fn: FnPtr<ExecuteDumpsFn> = {
+            let exec = Rc::clone(&exec);
+            Rc::new(move |dumps: Vec<Arc<String>>| {
+                let exec = Rc::clone(&exec);
+                Box::pin(async move { exec.execute_sql_dumps(dumps).await })
+            })
+        };
+
         let query_text_fn: FnPtr<QueryTextFn> = {
             let exec = Rc::clone(&exec);
             Rc::new(move |stmt: &SqlStatement| {
@@ -159,6 +175,7 @@ impl LocalDb {
         Self {
             execute_batch_fn,
             execute_dump_fn,
+            execute_dumps_fn,
             query_text_fn,
             query_json_fn,
             query_json_retryable_fn,
@@ -241,6 +258,11 @@ impl LocalDbQueryExecutor for LocalDb {
         (self.execute_dump_fn)(dump).await
     }
 
+    #[cfg(target_family = "wasm")]
+    async fn execute_sql_dumps(&self, dumps: Vec<Arc<String>>) -> Result<(), LocalDbQueryError> {
+        (self.execute_dumps_fn)(dumps).await
+    }
+
     async fn query_json<T>(&self, stmt: &SqlStatement) -> Result<T, LocalDbQueryError>
     where
         T: FromDbJson,
@@ -394,6 +416,62 @@ mod wasm_tests {
 
     wasm_bindgen_test_configure!(run_in_browser);
 
+    #[wasm_bindgen_test]
+    async fn import_preflight_waits_for_cross_tab_ownership() {
+        use super::pipeline::runner::leadership::acquire_import;
+        use js_sys::{Array, Function, Reflect};
+        use std::sync::Arc;
+        let mut holder = None;
+        for _ in 0..100 {
+            holder = acquire_import().await.unwrap();
+            if holder.is_some() {
+                break;
+            }
+            TimeoutFuture::new(5).await;
+        }
+        let holder = holder.expect("previous test must release import ownership");
+        let local_db = Function::new_no_args(r#"return {
+            reads:0, populated:false, chunks:[],
+            query(sql) {
+                this.reads++;
+                if (!sql.includes('target_watermarks')) throw new Error('must retain populated indexes');
+                return {value:this.populated ? '[{"present":1}]' : '[]'};
+            },
+            transaction() {}, wipeAndRecreate() {},
+            beginSqlDumpImport() { return {value:'session'}; },
+            appendSqlDumpChunk(id, sql) { this.chunks.push(sql); return {value:''}; },
+            finishSqlDumpImport() { return {value:''}; }, cancelSqlDumpImport() {}
+        };"#).call0(&JsValue::UNDEFINED).unwrap();
+        let exec = JsCallbackExecutor::new(local_db.clone()).unwrap();
+        let mut import = exec.execute_sql_dumps(vec![Arc::new(
+            "BEGIN; INSERT INTO t VALUES (2); COMMIT;".into(),
+        )]);
+        for _ in 0..10 {
+            assert!(futures::poll!(import.as_mut()).is_pending());
+            gloo_timers::future::TimeoutFuture::new(1).await;
+        }
+        assert_eq!(
+            Reflect::get(&local_db, &"reads".into()).unwrap().as_f64(),
+            Some(0.0)
+        );
+        Reflect::set(&local_db, &"populated".into(), &JsValue::TRUE).unwrap();
+        drop(holder);
+        import.await.unwrap();
+        assert_eq!(
+            Reflect::get(&local_db, &"reads".into()).unwrap().as_f64(),
+            Some(1.0)
+        );
+        let chunks = Reflect::get(&local_db, &"chunks".into())
+            .unwrap()
+            .dyn_into::<Array>()
+            .unwrap();
+        let sql = chunks
+            .iter()
+            .map(|v| v.as_string().unwrap())
+            .collect::<String>();
+        assert!(!sql.contains("DROP INDEX"));
+    }
+
     fn single_raindex_settings_yaml() -> String {
         format!(
             r#"

crates/common/src/raindex_client/local_db/pipeline/bootstrap.rs

@@ -15,7 +15,8 @@ use std::sync::Arc;
 
 const BOOTSTRAP_CACHE_SIZE_SQL: &str = "PRAGMA cache_size = -25000";
 #[cfg(target_family = "wasm")]
-const BOOTSTRAP_CACHE_SIZE_QUERY_SQL: &str = "PRAGMA cache_size = -25000; SELECT 1 WHERE 0;";
+pub(super) const BOOTSTRAP_CACHE_SIZE_QUERY_SQL: &str =
+    "PRAGMA cache_size = -25000; SELECT 1 WHERE 0;";
 
 #[derive(Debug, Default, Clone, Copy)]
 pub struct ClientBootstrapAdapter;
@@ -272,7 +273,7 @@ mod tests {
         for sql in [
             BOOTSTRAP_CACHE_SIZE_QUERY_SQL,
             "SELECT 1 AS present FROM target_watermarks LIMIT 1",
-            "SELECT name, sql FROM sqlite_master WHERE type = 'index' AND sql IS NOT NULL ORDER BY name",
+            "SELECT m.name, m.sql FROM sqlite_master AS m JOIN pragma_index_list(m.tbl_name) AS i ON i.name = m.name WHERE m.type = 'index' AND m.sql IS NOT NULL AND i.\"unique\" = 0 ORDER BY m.name",
         ] {
             assert_eq!(
                 Reflect::get(&calls, &JsValue::from_str(sql))