Land verified corpus and offline-import stack

bin/bitcoin-rs/src/config.rs

@@ -71,6 +71,26 @@ macro_rules! emit_cli {
             pub(crate) config: Option<PathBuf>,
             #[arg(long = "bitcoin-conf")]
             pub(crate) bitcoin_conf: Option<PathBuf>,
+            /// Single regular Core-framed archive file for foreground offline import.
+            ///
+            /// Meta flag: never reaches `UserConfig`. All three import flags
+            /// must be present together to run the offline import; otherwise
+            /// the daemon runs. The file must be the custody archive pin
+            /// whose bytes equal the indexed logical stream by parity.
+            #[arg(long)]
+            pub(crate) corpus: Option<PathBuf>,
+            /// Archive manifest binding the corpus identity and stop tip.
+            ///
+            /// Meta flag: never reaches `UserConfig`. Required with
+            /// `--corpus` and `--state` for the offline import.
+            #[arg(long)]
+            pub(crate) manifest: Option<PathBuf>,
+            /// Output certified-state JSON for the offline import.
+            ///
+            /// Meta flag: never reaches `UserConfig`. Required with
+            /// `--corpus` and `--manifest` for the offline import.
+            #[arg(long)]
+            pub(crate) state: Option<PathBuf>,
             $(
                 $(#[$fdoc])*
                 $($fcli)*

bin/bitcoin-rs/src/main.rs

@@ -27,8 +27,22 @@ fn main() -> ExitCode {
         Ok(cli) => cli,
         Err(error) => error.exit(),
     };
-    let result = config::resolve(cli, std::env::vars_os())
-        .and_then(|config| bitcoin_rs_node::run(config, bitcoin_rs_node::RuntimeInputs::default()));
+    // Foreground offline import shares the timed `{corpus_path}` contract:
+    // one regular file, never the indexed `sources` directory. All three
+    // flags must appear together; a partial set is a caller error, not a
+    // daemon run.
+    let import = (cli.corpus.clone(), cli.manifest.clone(), cli.state.clone());
+    let result = config::resolve(cli, std::env::vars_os()).and_then(|config| match import {
+        (Some(corpus), Some(manifest), Some(state)) => {
+            bitcoin_rs_node::offline_import::run_offline_import(config, &corpus, &manifest, &state)
+        }
+        (None, None, None) => {
+            bitcoin_rs_node::run(config, bitcoin_rs_node::RuntimeInputs::default())
+        }
+        _ => Err(anyhow::anyhow!(
+            "--corpus, --manifest, and --state must be given together"
+        )),
+    });
     match result {
         Ok(()) => ExitCode::SUCCESS,
         Err(error) => {

crates/index/src/index.rs

@@ -19,10 +19,9 @@ pub use filters::BasicFilter;
 pub use prepared::{PreparedBatch, PreparedBatchLimits, PreparedBlock};
 pub use reader::Indexer;
 pub use rows::IndexRowCounts;
-#[cfg(test)]
-pub(crate) use snapshot::RowVisitor;
 pub use snapshot::{
-    IndexReader, ScriptLiveScan, TxIndexFamily, TxIndexScan, TxIndexScanRow, TxIndexSnapshot,
+    IndexReader, ScriptLiveScan, TxIndexFamily, TxIndexPage, TxIndexScan, TxIndexScanRow,
+    TxIndexSnapshot,
 };
 pub use state::{ConsumerCursorUpdate, IndexWriteFence};
 pub use write::IndexWriter;

crates/index/src/index/snapshot.rs

@@ -10,14 +10,9 @@ use crate::{
 };
 use bitcoin_rs_primitives::{OutPoint, Txid};
 use bitcoin_rs_storage::{ColumnFamily, KvSnapshot, KvStore, PrefixScanLimit};
-use std::ops::ControlFlow;
 
-/// Per-row visitor of [`TxIndexSnapshot::for_each_row`]: the parsed key, the
-/// raw value, and [`ControlFlow::Break`] to stop the stream early.
-pub(crate) type RowVisitor<'a> =
-    dyn FnMut(HashPrefixRow, &[u8]) -> Result<ControlFlow<()>, IndexError> + 'a;
-
-/// One persisted `TxIndex` row family that [`TxIndexSnapshot::for_each_row`] streams.
+/// One persisted `TxIndex` row family that [`TxIndexSnapshot::scan_family_page`]
+/// pages.
 ///
 /// `ScriptHistory` occurrences are intentionally absent: that family has no
 /// qualified portable layout yet, and a variant is added only with its codec.
@@ -56,6 +51,17 @@ pub struct TxIndexScan {
     pub complete: bool,
 }
 
+/// One bounded page of a `TxIndex` family in ascending stored-key order.
+#[derive(Debug)]
+pub struct TxIndexPage {
+    /// Parsed rows of this page in ascending stored-key order.
+    pub rows: Vec<TxIndexScanRow>,
+    /// Whether the family ends with this page. With a positive `max_rows`
+    /// limit, a page that is not complete is always non-empty, so callers
+    /// advance with the last returned key until `complete`.
+    pub complete: bool,
+}
+
 /// Result of one bounded `ScriptLive` prefix scan.
 #[derive(Debug)]
 pub struct ScriptLiveScan {
@@ -78,18 +84,19 @@ pub trait TxIndexSnapshot: Send + Sync {
         &self,
         capability: IndexCapability,
     ) -> Result<Option<IndexWatermark>, IndexError>;
-    /// Streams every row of `family` in ascending stored-key order from this
-    /// snapshot, one row at a time.
+    /// Scans one bounded page of `family` in ascending stored-key order.
     ///
-    /// There is no seek, paging cursor, or family-sized vector: `visit`
-    /// receives each parsed key and its raw value and answers
-    /// [`ControlFlow::Break`] to stop early. A malformed key or a storage
-    /// failure ends the stream with an error.
-    fn for_each_row(
+    /// `start_after = None` starts at the family head; otherwise the page
+    /// begins at the first key strictly above `start_after`. Backends seek
+    /// the snapshot to the cursor, so no family-sized copy is made on any
+    /// backend. A malformed key or a storage failure ends the page with an
+    /// error.
+    fn scan_family_page(
         &self,
         family: TxIndexFamily,
-        visit: &mut RowVisitor<'_>,
-    ) -> Result<(), IndexError>;
+        start_after: Option<&[u8]>,
+        limit: PrefixScanLimit,
+    ) -> Result<TxIndexPage, IndexError>;
     /// Loads one capability's coverage floor: the first height its committed
     /// rows cover. `0` means complete coverage from genesis.
     fn capability_floor(&self, capability: IndexCapability) -> Result<u32, IndexError> {
@@ -169,20 +176,40 @@ impl TxIndexSnapshot for StoreTxIndexSnapshot<'_> {
         crate::index::capability::read_coverage_floor(self.snapshot.as_ref(), capability)
     }
 
-    fn for_each_row(
+    fn scan_family_page(
         &self,
         family: TxIndexFamily,
-        visit: &mut RowVisitor<'_>,
-    ) -> Result<(), IndexError> {
-        for item in self.snapshot.iter_prefix(family.column_family(), &[])? {
-            let (key, value) = item?;
+        start_after: Option<&[u8]>,
+        limit: PrefixScanLimit,
+    ) -> Result<TxIndexPage, IndexError> {
+        // Resume after `cursor` with `cursor || 0x00`: stored keys are
+        // fixed-width, so the suffixed key is the least key strictly above
+        // `cursor`. No `start_after` begins at the family head.
+        let start: Vec<u8> = match start_after {
+            Some(cursor) => {
+                let mut start = Vec::with_capacity(cursor.len().saturating_add(1));
+                start.extend_from_slice(cursor);
+                start.push(0);
+                start
+            }
+            None => Vec::new(),
+        };
+        let scan =
+            self.snapshot
+                .scan_prefix_from_bounded(family.column_family(), &[], &start, limit)?;
+        let mut rows = Vec::with_capacity(scan.rows.len());
+        for (key, value) in scan.rows {
+            if key.len() != crate::HASH_PREFIX_ROW_SIZE {
+                return Err(IndexError::InvalidPrefixRowLength { len: key.len() });
+            }
             let row = zerocopy::FromBytes::read_from_bytes(&key)
                 .map_err(|_| IndexError::InvalidPrefixRowLength { len: key.len() })?;
-            if visit(row, &value)?.is_break() {
-                break;
-            }
+            rows.push(TxIndexScanRow { row, value });
         }
-        Ok(())
+        Ok(TxIndexPage {
+            rows,
+            complete: scan.complete,
+        })
     }
 
     fn transaction_rows(

crates/index/src/lib.rs

@@ -33,7 +33,7 @@ pub use index::{
     BasicFilter, ConsumerCursorUpdate, IndexCapabilities, IndexCapability, IndexError, IndexReader,
     IndexRowCounts, IndexWatermark, IndexWatermarks, IndexWriteFence, IndexWriter, Indexer,
     MAX_LIVE_SCRIPT_SIZE, PreparedBatch, PreparedBatchLimits, PreparedBlock, ScriptLiveScan,
-    SpentCoinScripts, TxIndexFamily, TxIndexScan, TxIndexScanRow, TxIndexSnapshot,
+    SpentCoinScripts, TxIndexFamily, TxIndexPage, TxIndexScan, TxIndexScanRow, TxIndexSnapshot,
 };
 pub use query_api::{
     CompactFilterQuery, DerivedIndexInfo, DerivedIndexQuery, RollbackWarningSource,

crates/index/src/portable/chain.rs

@@ -348,7 +348,7 @@ impl<'a> BodyReader<'a> {
     }
 }
 
-fn decode(bytes: &[u8], expected_len: u32) -> Located {
+pub(super) fn decode(bytes: &[u8], expected_len: u32) -> Located {
     if usize::try_from(expected_len) != Ok(bytes.len()) {
         return Located::Mismatch;
     }