feat: Can fetch headers from peer

We fetch headers from peer to store them next to the snapshots.
The bootstrap process will start from them instead of fetching
them on demand. That way, we can move towards a peer-less
bootstrap.

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

Author: pgrange

Makefile

@@ -82,6 +82,12 @@ download-haskell-config: ## Download Cardano Haskell configuration for $NETWORK
 clear-dbs: ## Clear the databases
 	@rm -rf $(LEDGER_DIR) $(CHAIN_DIR)
 
+fetch-chain-headers: $(CONFIG_FOLDER)/$(NETWORK)/ ## Fetch chain headers from the network
+	cargo run --profile $(BUILD_PROFILE) -- fetch-chain-headers \
+		--peer-address $(PEER_ADDRESS) \
+		--config-dir $(CONFIG_FOLDER) \
+		--network $(NETWORK)
+
 bootstrap: clear-dbs ## Bootstrap the node from scratch
 	cargo run --profile $(BUILD_PROFILE) -- bootstrap \
 		--peer-address $(PEER_ADDRESS) \

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

@@ -0,0 +1,265 @@
+// Copyright 2025 PRAGMA
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::cmd::{DEFAULT_NETWORK, connect_to_peer};
+use amaru::stages::pull;
+use amaru_consensus::IsHeader;
+use amaru_kernel::{Header, Point, from_cbor, network::NetworkName, peer::Peer};
+use amaru_network::session::PeerSession;
+use amaru_progress_bar::{ProgressBar, new_terminal_progress_bar};
+use clap::{Parser, arg};
+use gasket::framework::{AsWorkError, WorkerError};
+use pallas_network::miniprotocols::chainsync::{self, HeaderContent, NextResponse};
+use std::{
+    error::Error,
+    fs::File,
+    io::Write,
+    path::{Path, PathBuf},
+    sync::{Arc, RwLock},
+    time::Duration,
+};
+use tokio::{sync::Mutex, time::timeout};
+
+use tracing::info;
+
+#[derive(Debug, Parser)]
+pub struct Args {
+    /// Network to fetch chain headers for.
+    ///
+    /// Should be one of 'mainnet', 'preprod', 'preview' or 'testnet:<magic>' where
+    /// `magic` is a 32-bits unsigned value denoting a particular testnet.
+    #[arg(
+        long,
+        value_name = "NETWORK",
+        default_value_t = DEFAULT_NETWORK,
+    )]
+    network: NetworkName,
+
+    /// Path to directory containing per-network bootstrap configuration files.
+    ///
+    /// This path will be used as a prefix to resolve per-network configuration files
+    /// needed for bootstrapping and to store fetched chain headers. Given a source
+    /// directory `data`, and a network name of `preview`, the expected layout for
+    /// configuration files would be:
+    ///
+    /// * `data/preview/snapshots.json`: a list of `Snapshot` vaalues,
+    /// * `data/preview/nonces.json`: a list of `InitialNonces` values,
+    /// * `data/preview/headers.json`: a list of `Point`s,
+    /// * `data/preview/headers`: a directory where the fetched chain headers will be stored.
+    #[arg(
+        long,
+        value_name = "DIRECTORY",
+        default_value = "data",
+        verbatim_doc_comment
+    )]
+    config_dir: PathBuf,
+
+    /// 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.
+    ///
+    /// Address is given in the usual `host:port` format, for example: "1.2.3.4:3000".
+    #[arg(
+        long,
+        value_name = "NETWORK_ADDRESS",
+        default_value = "127.0.0.1:3001",
+        verbatim_doc_comment
+    )]
+    peer_address: String,
+}
+
+pub async fn run(args: Args) -> Result<(), Box<dyn Error>> {
+    info!(config=?args.config_dir, peer=%args.peer_address, network=%args.network,
+          "fetching chain headers",
+    );
+    let network = args.network;
+    let network_dir = args.config_dir.join(&*network.to_string());
+
+    fetch_headers_for_network(network, &args.peer_address, &network_dir).await?;
+
+    Ok(())
+}
+
+async fn fetch_headers_for_network(
+    network: NetworkName,
+    peer_address: &str,
+    config_dir: &Path,
+) -> Result<(), Box<dyn Error>> {
+    let headers_file: PathBuf = config_dir.join("headers.json");
+    let content = tokio::fs::read_to_string(headers_file).await?;
+    let points: Vec<String> = serde_json::from_str(&content)?;
+    let mut initial_headers = Vec::new();
+    for point in points {
+        match Point::try_from(point.as_str()) {
+            Ok(point) => initial_headers.push(point),
+            Err(e) => tracing::warn!("Ignoring malformed header point '{}': {}", point, e),
+        }
+    }
+    for hdr in initial_headers {
+        // FIXME: why do we only fetch 2 headers for each header listed in the
+        // config file? The 2 headers make sense, but why starting from more than
+        // one header?
+        const NUM_HEADERS_TO_FETCH: usize = 2;
+        fetch_headers(peer_address, network, config_dir, hdr, NUM_HEADERS_TO_FETCH).await?;
+    }
+
+    Ok(())
+}
+
+pub(crate) async fn fetch_headers(
+    peer_address: &str,
+    network_name: NetworkName,
+    config_dir: &Path,
+    point: Point,
+    max: usize,
+) -> Result<(), Box<dyn Error>> {
+    let peer_client = Arc::new(Mutex::new(
+        connect_to_peer(peer_address, &network_name).await?,
+    ));
+
+    let peer_session = PeerSession {
+        peer: Peer::new(peer_address),
+        peer_client,
+    };
+
+    let pull = pull::Stage::new(
+        peer_session.clone(),
+        vec![point.clone()],
+        Arc::new(RwLock::new(true)),
+    );
+
+    pull.find_intersection().await?;
+
+    let mut peer_client = peer_session.lock().await;
+    let mut count = 0;
+
+    let client = (*peer_client).chainsync();
+
+    let mut progress: Option<Box<dyn ProgressBar>> = None;
+
+    // TODO: implement a proper pipelined client because this one is super slow
+    // Pipelining in Haskell is single threaded which implies the code handles
+    // scheduling between sending burst of MsgRequest and collecting responses.
+    // Here we can do better thanks to gasket's workers: just spawn 2 workers,
+    // one for sending requests and the other for handling responses, along
+    // with a shared counter.
+    // Pipelining stops when we reach the tip of the peer's chain.
+    loop {
+        let what = if client.has_agency() {
+            request_next_block(client, config_dir, &mut count, &mut progress, max).await?
+        } else {
+            await_for_next_block(client, config_dir, &mut count, &mut progress, max).await?
+        };
+        match what {
+            Continue => continue,
+            Stop => {
+                if let Some(progress) = progress {
+                    progress.clear()
+                }
+                break;
+            }
+        }
+    }
+    info!(total = count, "header_fetched");
+    Ok(())
+}
+enum What {
+    Continue,
+    Stop,
+}
+
+use What::*;
+
+async fn request_next_block(
+    client: &mut chainsync::Client<HeaderContent>,
+    config_dir: &Path,
+    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, config_dir, count, progress, max)
+}
+
+async fn await_for_next_block(
+    client: &mut chainsync::Client<HeaderContent>,
+    config_dir: &Path,
+    count: &mut usize,
+    progress: &mut Option<Box<dyn ProgressBar>>,
+    max: usize,
+) -> Result<What, WorkerError> {
+    match timeout(Duration::from_secs(1), client.recv_while_must_reply()).await {
+        Ok(result) => result
+            .map_err(|_| WorkerError::Recv)
+            .and_then(|next| handle_response(next, config_dir, count, progress, max)),
+        Err(_) => Err(WorkerError::Retry)?,
+    }
+}
+
+#[allow(clippy::unwrap_used)]
+fn handle_response(
+    next: NextResponse<HeaderContent>,
+    config_dir: &Path,
+    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();
+            let slot = header.slot();
+
+            let filename = format!("chain.{}.{}.cbor", slot, hex::encode(hash));
+            let headers_dir = config_dir.join("headers");
+            std::fs::create_dir_all(&headers_dir)
+                .inspect_err(|reason| tracing::error!(dir = %headers_dir.display(), reason = %reason, "Failed to create headers directory"))
+                .map_err(|_| WorkerError::Panic)?;
+            let filepath = headers_dir.join(&filename);
+            let mut file = File::create(&filepath)
+                .inspect_err(|reason|tracing::error!(file = %filepath.display(), reason = %reason, "Failed to create file"))
+                .map_err(|_| WorkerError::Panic)?;
+            file.write_all(&content.cbor)
+                .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)
+            }
+        }
+        #[allow(clippy::unwrap_used)]
+        NextResponse::RollBackward(point, tip) => {
+            info!(?point, ?tip, "roll_backward");
+            if progress.is_none() {
+                *progress = Some(new_terminal_progress_bar(
+                    max,
+                    " fetching headers (~{eta} left) {bar:70} {pos:>7}/{len:7} ({percent_precise}%)",
+                ));
+            }
+            Ok(Continue)
+        }
+        NextResponse::Await => Ok(Continue),
+    }
+}

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

@@ -16,6 +16,7 @@ use amaru_kernel::network::NetworkName;
 use pallas_network::facades::PeerClient;
 
 pub(crate) mod bootstrap;
+pub(crate) mod fetch_chain_headers;
 pub(crate) mod daemon;
 pub(crate) mod import_headers;
 pub(crate) mod import_ledger_state;

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

@@ -37,6 +37,8 @@ enum Command {
     /// **NOTE**: Only `preprod` network is supported for now.
     Bootstrap(cmd::bootstrap::Args),
 
+    FetchChainHeaders(cmd::fetch_chain_headers::Args),
+
     /// Run the node in all its glory.
     Daemon(cmd::daemon::Args),
 
@@ -114,6 +116,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         Command::ImportHeaders(args) => cmd::import_headers::run(args).await,
         Command::ImportNonces(args) => cmd::import_nonces::run(args).await,
         Command::Bootstrap(args) => cmd::bootstrap::run(args).await,
+        Command::FetchChainHeaders(args) => cmd::fetch_chain_headers::run(args).await,
     };
 
     // TODO: we might also want to integrate this into a graceful shutdown system, and into a panic hook

data/README.md

@@ -84,3 +84,14 @@ For example, for PreProd, that means removing all immutable chunks after `03149`
 3. Start your node.
 
 Both Ogmios and the fetch script will automatically connect to the node once ready and capture data from it when it becomes available. This should terminate once all the points have been processed.
+
+4. Dump chain headers as CBOR
+
+Go to home directory `cd ..` and make fetch-chain-header :
+
+```
+cd ..
+make fetch-chain-header
+```
+
+This will add `chain.<slot>.<hash>.cbor` files in the snapshot directory that can then be used to bootstraped the amaru node.