Integrate atomic SQL bootstrap and deferred browser indexes

crates/cli/src/commands/local_db/cli.rs

@@ -216,6 +216,7 @@ mod tests {
             metadata_rpcs: Vec::new(),
             cfg: sync_config,
             dump_str: None,
+            defer_analyze: false,
             block_number_threshold: 10000,
             manifest_end_block: 1,
         };

crates/cli/src/commands/local_db/pipeline/bootstrap.rs

@@ -167,6 +167,7 @@ mod tests {
 
         let cfg = BootstrapConfig {
             raindex_id: sample_ob_id(),
+            dump_sql: None,
             dump_stmt: None,
             latest_block: 0,
             block_number_threshold: TEST_BLOCK_NUMBER_THRESHOLD,
@@ -198,6 +199,7 @@ mod tests {
 
         let cfg = BootstrapConfig {
             raindex_id: sample_ob_id(),
+            dump_sql: None,
             dump_stmt: None,
             latest_block: 0,
             block_number_threshold: TEST_BLOCK_NUMBER_THRESHOLD,
@@ -231,6 +233,7 @@ mod tests {
 
         let cfg = BootstrapConfig {
             raindex_id: sample_ob_id(),
+            dump_sql: None,
             dump_stmt: Some(SqlStatementBatch::from(vec![dump_stmt.clone()])),
             latest_block: 0,
             block_number_threshold: TEST_BLOCK_NUMBER_THRESHOLD,
@@ -266,6 +269,7 @@ mod tests {
 
         let cfg = BootstrapConfig {
             raindex_id: sample_ob_id(),
+            dump_sql: None,
             dump_stmt: Some(SqlStatementBatch::from(vec![dump_stmt.clone()])),
             latest_block: 0,
             block_number_threshold: TEST_BLOCK_NUMBER_THRESHOLD,
@@ -301,6 +305,7 @@ mod tests {
 
         let cfg = BootstrapConfig {
             raindex_id: sample_ob_id(),
+            dump_sql: None,
             dump_stmt: None,
             latest_block: 0,
             block_number_threshold: 1,

crates/cli/src/commands/local_db/pipeline/runner/environment.rs

@@ -84,6 +84,7 @@ mod tests {
                     window_overrides: WindowOverrides::default(),
                 },
                 dump_str: None,
+                defer_analyze: false,
                 block_number_threshold: 10000,
                 manifest_end_block: 1,
             },

crates/cli/src/commands/local_db/pipeline/runner/export.rs

@@ -149,6 +149,7 @@ mod tests {
                 metadata_rpcs: Vec::new(),
                 cfg: sync_config,
                 dump_str: None,
+                defer_analyze: false,
                 block_number_threshold: 10000,
                 manifest_end_block: 1,
             },
@@ -236,6 +237,7 @@ mod tests {
                 metadata_rpcs: Vec::new(),
                 cfg: sync_config,
                 dump_str: None,
+                defer_analyze: false,
                 block_number_threshold: 10000,
                 manifest_end_block: 1,
             },

crates/cli/src/commands/local_db/pipeline/runner/manifest.rs

@@ -190,6 +190,7 @@ mod tests {
                     window_overrides: WindowOverrides::default(),
                 },
                 dump_str: None,
+                defer_analyze: false,
                 block_number_threshold: 10000,
                 manifest_end_block: 1,
             },

crates/cli/src/commands/local_db/pipeline/runner/mod.rs

@@ -293,7 +293,7 @@ where
                 .await
                 .map_err(|error| mk_failure(TargetStage::DumpDownload, error))?;
             SyncInputs {
-                dump_str: Some(dump_sql),
+                dump_str: Some(std::sync::Arc::new(dump_sql)),
                 ..inputs
             }
         }
@@ -548,8 +548,13 @@ mod tests {
             )));
             db.execute_batch(&batch.ensure_transaction()).await?;
 
-            self.telemetry
-                .record_bootstrap_dump(config.dump_stmt.as_ref().map(dump_sql));
+            self.telemetry.record_bootstrap_dump(
+                config
+                    .dump_sql
+                    .as_ref()
+                    .map(|sql| sql.as_str().to_owned())
+                    .or_else(|| config.dump_stmt.as_ref().map(dump_sql)),
+            );
             Ok(())
         }
 
@@ -694,7 +699,11 @@ mod tests {
                 panic!("stub bootstrap panic");
             }
 
-            let dump_stmt = config.dump_stmt.as_ref().map(dump_sql);
+            let dump_stmt = config
+                .dump_sql
+                .as_ref()
+                .map(|sql| sql.as_str().to_owned())
+                .or_else(|| config.dump_stmt.as_ref().map(dump_sql));
             self.telemetry.record_bootstrap_dump(dump_stmt);
             self.ensure_tables(db).await?;
 

crates/common/src/local_db/mod.rs

@@ -140,6 +140,9 @@ pub enum LocalDbError {
     #[error(transparent)]
     LocalDbQueryError(#[from] LocalDbQueryError),
 
+    #[error("SQL dump import failed: {0}")]
+    DumpImportFailed(#[source] LocalDbQueryError),
+
     #[error(transparent)]
     IoError(#[from] std::io::Error),
 
@@ -292,6 +295,7 @@ impl LocalDbError {
             LocalDbError::ManifestFetch(err) => format!("Failed to fetch manifest: {}", err),
             LocalDbError::TaskJoin(err) => format!("Task join error: {}", err),
             LocalDbError::LocalDbQueryError(err) => format!("Database query error: {}", err),
+            LocalDbError::DumpImportFailed(err) => format!("SQL dump import failed: {}", err),
             LocalDbError::IoError(err) => format!("I/O error: {}", err),
             LocalDbError::FromHexError(err) => format!("Hex decoding error: {}", err),
             LocalDbError::SqlBuildError(err) => format!("SQL build error: {}", err),

crates/common/src/local_db/pipeline/adapters/apply.rs

@@ -22,6 +22,7 @@ pub struct ApplyPipelineTargetInfo {
     pub raindex_id: RaindexIdentifier,
     pub start_block: u64,
     pub target_block: u64,
+    pub defer_analyze: bool,
     pub hash: B256,
 }
 
@@ -104,7 +105,9 @@ pub trait ApplyPipeline {
 
         // Ensure SQLite planner stats are up to date so reads don't suffer from
         // poor query plans.
-        batch.add(SqlStatement::new("ANALYZE"));
+        if !target_info.defer_analyze {
+            batch.add(SqlStatement::new("ANALYZE"));
+        }
 
         Ok(batch)
     }
@@ -166,6 +169,7 @@ mod tests {
             raindex_id: raindex_id.clone(),
             start_block,
             target_block,
+            defer_analyze: false,
             hash: SAMPLE_HASH_B256,
         }
     }
@@ -930,6 +934,24 @@ mod tests {
         );
     }
 
+    #[test]
+    fn deferred_analysis_omits_per_target_analyze() {
+        let pipeline = DefaultApplyPipeline::new();
+        let raindex_id = sample_raindex_id();
+        let mut target = build_target_info(&raindex_id, 1, 100);
+        target.defer_analyze = true;
+
+        let batch = pipeline
+            .build_batch(&target, &[], &[], &[], &[])
+            .expect("batch ok");
+
+        assert!(batch.statements().iter().all(|s| s.sql() != "ANALYZE"));
+        assert!(batch
+            .statements()
+            .iter()
+            .any(|s| s.sql().starts_with("INSERT INTO target_watermarks")));
+    }
+
     #[test]
     fn statements_order_raw_then_tokens_then_decoded_then_watermark() {
         let pipeline = DefaultApplyPipeline::new();

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

@@ -15,10 +15,12 @@ use crate::local_db::RaindexIdentifier;
 use async_trait::async_trait;
 use raindex_app_settings::local_db_manifest::DB_SCHEMA_VERSION;
 use std::collections::HashSet;
+use std::sync::Arc;
 
 #[derive(Debug, Clone)]
 pub struct BootstrapConfig {
     pub raindex_id: RaindexIdentifier,
+    pub dump_sql: Option<Arc<String>>,
     pub dump_stmt: Option<SqlStatementBatch>,
     pub latest_block: u64,
     pub block_number_threshold: u32,
@@ -53,7 +55,7 @@ pub trait BootstrapPipeline {
         DB: LocalDbQueryExecutor + ?Sized,
     {
         let rows = db
-            .query_json::<Vec<DbMetadataRow>>(&fetch_db_metadata_stmt())
+            .query_json_retryable::<Vec<DbMetadataRow>>(&fetch_db_metadata_stmt())
             .await?;
         if let Some(row) = rows.first() {
             let expected = db_schema_version.unwrap_or(DB_SCHEMA_VERSION);
@@ -76,7 +78,7 @@ pub trait BootstrapPipeline {
     where
         DB: LocalDbQueryExecutor + ?Sized,
     {
-        let existing: Vec<TableResponse> = db.query_json(&fetch_tables_stmt()).await?;
+        let existing: Vec<TableResponse> = db.query_json_retryable(&fetch_tables_stmt()).await?;
         let existing_set: HashSet<String> = existing
             .into_iter()
             .map(|t| t.name.to_ascii_lowercase())
@@ -88,7 +90,7 @@ pub trait BootstrapPipeline {
 
         let last_synced_block = if existing_set.contains("target_watermarks") {
             let rows: Vec<TargetWatermarkRow> = db
-                .query_json(&fetch_target_watermark_stmt(raindex_id))
+                .query_json_retryable(&fetch_target_watermark_stmt(raindex_id))
                 .await?;
             rows.first().map(|r| r.last_block)
         } else {
@@ -124,7 +126,7 @@ pub trait BootstrapPipeline {
     where
         DB: LocalDbQueryExecutor + ?Sized,
     {
-        let rows: Vec<IntegrityCheckRow> = db.query_json(&integrity_check_stmt()).await?;
+        let rows: Vec<IntegrityCheckRow> = db.query_json_retryable(&integrity_check_stmt()).await?;
         let is_healthy = rows
             .first()
             .map(|row| row.quick_check.eq_ignore_ascii_case("ok"))
@@ -748,6 +750,7 @@ mod tests {
         let db = MockDb::default();
         let cfg = BootstrapConfig {
             raindex_id: RaindexIdentifier::new(1, Address::ZERO),
+            dump_sql: None,
             dump_stmt: None,
             latest_block: 0,
             block_number_threshold: 10_000,

crates/common/src/local_db/pipeline/adapters/tokens.rs

@@ -44,7 +44,7 @@ impl TokensPipeline for DefaultTokensPipeline {
         let Some(stmt) = build_fetch_stmt(raindex_id, token_addrs_lower)? else {
             return Ok(vec![]);
         };
-        let rows: Vec<Erc20TokenRow> = db.query_json(&stmt).await?;
+        let rows: Vec<Erc20TokenRow> = db.query_json_retryable(&stmt).await?;
         Ok(rows)
     }