chore: factorize header importation

import-headers and bootstrap both use the same peerless tactic
to import headers from source `data` directory.

Signed-off-by: Pascal Grange <pascal@grange.nom.fr>

Author: pgrange

Makefile

@@ -54,9 +54,7 @@ import-headers: ## Import headers from $PEER_ADDRESS for demo
 		cargo run --profile $(BUILD_PROFILE) -- import-headers \
 			--network $(NETWORK) \
 			--chain-dir $(CHAIN_DIR) \
-			--peer-address $(PEER_ADDRESS) \
-			--starting-point $$HEADER \
-			--count 2; \
+			--config-dir $(CONFIG_FOLDER); \
 	done
 
 import-nonces: ## Import nonces for demo

crates/amaru/src/bin/amaru/cmd/bootstrap.rs

@@ -13,21 +13,16 @@
 // limitations under the License.
 
 use super::{
+    import_headers::import_headers_for_network,
     import_ledger_state::import_all_from_directory,
     import_nonces::{InitialNonces, import_nonces},
 };
 use crate::cmd::DEFAULT_NETWORK;
 use amaru::snapshots_dir;
-use amaru_consensus::IsHeader;
-use amaru_consensus::consensus::store::ChainStore;
-use amaru_kernel::{
-    Header, default_chain_dir, default_ledger_dir, from_cbor, network::NetworkName,
-};
-use amaru_stores::rocksdb::consensus::RocksDBStore;
+use amaru_kernel::{default_chain_dir, default_ledger_dir, network::NetworkName};
 use async_compression::tokio::bufread::GzipDecoder;
 use clap::{Parser, arg};
 use futures_util::TryStreamExt;
-use gasket::framework::WorkerError;
 use serde::Deserialize;
 use std::{
     error::Error,
@@ -37,7 +32,7 @@ use std::{
 use thiserror::Error;
 use tokio::{
     fs::{self, File},
-    io::{AsyncReadExt, BufReader},
+    io::BufReader,
 };
 use tokio_util::io::StreamReader;
 use tracing::info;
@@ -112,40 +107,6 @@ pub async fn run(args: Args) -> Result<(), Box<dyn Error>> {
     Ok(())
 }
 
-#[allow(clippy::unwrap_used)]
-async fn import_headers_for_network(
-    network: NetworkName,
-    config_dir: &Path,
-    chain_dir: &PathBuf,
-) -> Result<(), Box<dyn Error>> {
-    let era_history = network.into();
-    let mut db = RocksDBStore::new(chain_dir, era_history)?;
-
-    for entry in std::fs::read_dir(config_dir.join("headers"))? {
-        let entry = entry?;
-        let path = entry.path();
-        if path.is_file()
-            && let Some(filename) = path.file_name().and_then(|f| f.to_str())
-            && filename.starts_with("header.")
-            && filename.ends_with(".cbor")
-        {
-            let mut file = File::open(&path).await
-                    .inspect_err(|reason| tracing::error!(file = %path.display(), reason = %reason, "Failed to open header file"))
-                    .map_err(|_| WorkerError::Panic)?;
-            let mut cbor_data = Vec::new();
-            file.read_to_end(&mut cbor_data).await
-                    .inspect_err(|reason| tracing::error!(file = %path.display(), reason = %reason, "Failed to read header file"))
-                    .map_err(|_| WorkerError::Panic)?;
-            let header_from_file: Header = from_cbor(&cbor_data).unwrap();
-            let hash = header_from_file.hash();
-            db.store_header(&hash, &header_from_file)
-                .map_err(|_| WorkerError::Panic)?;
-        }
-    }
-
-    Ok(())
-}
-
 async fn import_nonces_for_network(
     network: NetworkName,
     config_dir: &Path,

crates/amaru/src/bin/amaru/cmd/import_headers.rs

@@ -12,31 +12,20 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::cmd::connect_to_peer;
 use amaru_consensus::{IsHeader, consensus::store::ChainStore};
-use amaru_kernel::{Header, Point, default_chain_dir, from_cbor, network::NetworkName, peer::Peer};
-use amaru_network::chain_sync_client::ChainSyncClient;
-use amaru_progress_bar::{ProgressBar, new_terminal_progress_bar};
+use amaru_kernel::{Header, default_chain_dir, from_cbor, network::NetworkName};
 use amaru_stores::rocksdb::consensus::RocksDBStore;
 use clap::Parser;
 use gasket::framework::*;
-use pallas_network::miniprotocols::chainsync::{HeaderContent, NextResponse};
-use std::{error::Error, path::PathBuf, time::Duration};
-use tokio::time::timeout;
-use tracing::info;
+use std::{
+    error::Error,
+    path::{Path, PathBuf},
+};
+use tokio::{fs::File, io::AsyncReadExt};
 
 #[derive(Debug, Parser)]
 pub struct Args {
-    /// Address of the node to connect to for retrieving chain data.
-    /// The node should be accessible via the node-2-node protocol, which
-    /// means the remote node should be running as a validator and not
-    /// as a client node.
-    ///
-    /// Addressis given in the usual `host:port` format, for example: "1.2.3.4:3000".
-    #[arg(long, value_name = "NETWORK_ADDRESS", verbatim_doc_comment)]
-    peer_address: String,
-
-    /// Network to use for the connection.
+    /// Network for which we are importing headers.
     ///
     /// Should be one of 'mainnet', 'preprod', 'preview' or 'testnet:<magic>' where
     /// `magic` is a 32-bits unsigned value denoting a particular testnet.
@@ -54,152 +43,61 @@ pub struct Args {
     #[arg(long, value_name = "DIR", verbatim_doc_comment)]
     chain_dir: Option<PathBuf>,
 
-    /// Starting point of import.
+    /// Path to directory containing per-network configuration files.
+    ///
+    /// This path will be used as a prefix to resolve per-network configuration files
+    /// needed for importing headers. Given a source directory `data`, and a
+    /// a network name of `preview`, the expected layout for header files would be:
     ///
-    /// This is the "intersection" point which will be given to the peer as a starting point
-    /// to import the chain database.
-    #[arg(long, value_name = "POINT", verbatim_doc_comment, value_parser = |s: &str| Point::try_from(s)
+    /// `data/preview/headers/header.*.*.cbor`
+    #[arg(
+        long,
+        value_name = "DIRECTORY",
+        default_value = "data",
+        verbatim_doc_comment
     )]
-    starting_point: Point,
-
-    /// Number of headers to import.
-    /// Maximum number of headers to import from the `peer`.
-    /// By default, it will retrieve all headers until it reaches the tip of the peer's chain.
-    #[arg(long, value_name = "UINT", verbatim_doc_comment, default_value_t = usize::MAX)]
-    count: usize,
-}
-
-enum What {
-    Continue,
-    Stop,
+    config_dir: PathBuf,
 }
 
-use What::*;
-
 pub async fn run(args: Args) -> Result<(), Box<dyn std::error::Error>> {
+    let network = args.network;
+    let network_dir = args.config_dir.join(&*network.to_string());
     let chain_dir = args
         .chain_dir
         .unwrap_or_else(|| default_chain_dir(args.network).into());
-    import_headers(
-        &args.peer_address,
-        args.network,
-        &chain_dir,
-        args.starting_point,
-        args.count,
-    )
-    .await
+    import_headers_for_network(args.network, &network_dir, &chain_dir).await
 }
 
-pub(crate) async fn import_headers(
-    peer_address: &str,
-    network_name: NetworkName,
-    chain_db_dir: &PathBuf,
-    point: Point,
-    max: usize,
+#[allow(clippy::unwrap_used)]
+pub(crate) async fn import_headers_for_network(
+    network: NetworkName,
+    config_dir: &Path,
+    chain_dir: &PathBuf,
 ) -> Result<(), Box<dyn Error>> {
-    let era_history = network_name.into();
-    let mut db = RocksDBStore::new(chain_db_dir, era_history)?;
-
-    let peer_client = connect_to_peer(peer_address, &network_name).await?;
-    let mut client = ChainSyncClient::new(
-        Peer::new(peer_address),
-        peer_client.chainsync,
-        vec![point.clone()],
-    );
-
-    client.find_intersection().await?;
-
-    let mut count = 0;
-
-    let mut progress: Option<Box<dyn ProgressBar>> = None;
-
-    loop {
-        let what = if client.has_agency() {
-            request_next_block(&mut client, &mut db, &mut count, &mut progress, max).await?
-        } else {
-            await_for_next_block(&mut client, &mut db, &mut count, &mut progress, max).await?
-        };
-        match what {
-            Continue => continue,
-            Stop => {
-                if let Some(progress) = progress {
-                    progress.clear()
-                }
-                break;
-            }
-        }
-    }
-    info!(total = count, "header_imported");
-    Ok(())
-}
-
-async fn request_next_block(
-    client: &mut ChainSyncClient,
-    db: &mut RocksDBStore,
-    count: &mut usize,
-    progress: &mut Option<Box<dyn ProgressBar>>,
-    max: usize,
-) -> Result<What, WorkerError> {
-    let next = client.request_next().await.or_restart()?;
-    handle_response(next, db, count, progress, max)
-}
-
-async fn await_for_next_block(
-    client: &mut ChainSyncClient,
-    db: &mut RocksDBStore,
-    count: &mut usize,
-    progress: &mut Option<Box<dyn ProgressBar>>,
-    max: usize,
-) -> Result<What, WorkerError> {
-    match timeout(Duration::from_secs(1), client.await_next()).await {
-        Ok(result) => result
-            .map_err(|_| WorkerError::Recv)
-            .and_then(|next| handle_response(next, db, count, progress, max)),
-        Err(_) => Err(WorkerError::Retry)?,
-    }
-}
-
-#[expect(clippy::unwrap_used)]
-fn handle_response(
-    next: NextResponse<HeaderContent>,
-    db: &mut RocksDBStore,
-    count: &mut usize,
-    progress: &mut Option<Box<dyn ProgressBar>>,
-    max: usize,
-) -> Result<What, WorkerError> {
-    match next {
-        NextResponse::RollForward(content, tip) => {
-            let header: Header = from_cbor(&content.cbor).unwrap();
-            let hash = header.hash();
-
-            db.store_header(&hash, &header)
+    let era_history = network.into();
+    let mut db = RocksDBStore::new(chain_dir, era_history)?;
+
+    for entry in std::fs::read_dir(config_dir.join("headers"))? {
+        let entry = entry?;
+        let path = entry.path();
+        if path.is_file()
+            && let Some(filename) = path.file_name().and_then(|f| f.to_str())
+            && filename.starts_with("header.")
+            && filename.ends_with(".cbor")
+        {
+            let mut file = File::open(&path).await
+                    .inspect_err(|reason| tracing::error!(file = %path.display(), reason = %reason, "Failed to open header file"))
+                    .map_err(|_| WorkerError::Panic)?;
+            let mut cbor_data = Vec::new();
+            file.read_to_end(&mut cbor_data).await
+                    .inspect_err(|reason| tracing::error!(file = %path.display(), reason = %reason, "Failed to read header file"))
+                    .map_err(|_| WorkerError::Panic)?;
+            let header_from_file: Header = from_cbor(&cbor_data).unwrap();
+            let hash = header_from_file.hash();
+            db.store_header(&hash, &header_from_file)
                 .map_err(|_| WorkerError::Panic)?;
-
-            *count += 1;
-
-            let slot = header.slot();
-            let tip_slot = tip.0.slot_or_default();
-
-            if let Some(progress) = progress {
-                progress.tick(1)
-            }
-
-            if *count >= max || slot == tip_slot {
-                Ok(Stop)
-            } else {
-                Ok(Continue)
-            }
-        }
-        NextResponse::RollBackward(point, tip) => {
-            info!(?point, ?tip, "roll_backward");
-            if progress.is_none() {
-                *progress = Some(new_terminal_progress_bar(
-                    max,
-                    " importing headers (~{eta} left) {bar:70} {pos:>7}/{len:7} ({percent_precise}%)",
-                ));
-            }
-            Ok(Continue)
         }
-        NextResponse::Await => Ok(Continue),
     }
+
+    Ok(())
 }

crates/amaru/src/bin/amaru/main.rs

@@ -46,7 +46,7 @@ enum Command {
     #[clap(alias = "import")]
     ImportLedgerState(cmd::import_ledger_state::Args),
 
-    /// Import block headers from another (live) node.
+    /// Import block headers from `${config_dir}/${network name}/`
     #[clap(alias = "import-chain-db")]
     ImportHeaders(cmd::import_headers::Args),