diff --git a/debian/etc/monad-services/env b/debian/etc/monad-services/env index 506d8f8..b4d3a0a 100644 --- a/debian/etc/monad-services/env +++ b/debian/etc/monad-services/env @@ -1,3 +1,38 @@ -# Configuration for the monad-services units. Variables set here are -# expanded into each unit's ExecStart line; a unit fails to start when a -# variable it names is missing. +# Configuration for the monad-services units. The monad-archive binary reads +# these from its environment; a commented-out variable falls back to its +# built-in default, and a unit fails to start when a variable it requires is +# unset. + +## Required by the archiver, indexer, and block-writer. +## For triedb: 'triedb ' +#BLOCK_DATA_SOURCE= + +## Required by the archiver and indexer. +## For aws: 'aws ' +#ARCHIVE_SINK= + +## Required by the archive-checker: the bucket holding checker state, and +## the comma-separated replica set to validate, +## e.g. 'aws bucket-1 20,aws bucket-2 20,aws bucket-3 20'. +#ARCHIVE_BUCKET= +#INIT_REPLICAS= + +## Required by the block-writer. +#BLOCK_WRITER_DEST_PATH= + +## Optional fallback read source for the archiver. +#FALLBACK_BLOCK_DATA_SOURCE= + +## Optional path to bft blocks the archiver uploads to the archive sink. +#BFT_BLOCK_PATH= + +## Archiver concurrency (default 15). +#MAX_CONCURRENT_BLOCKS= + +# Metrics collector for every unit; metrics are disabled when unset. +OTEL_ENDPOINT=http://127.0.0.1:4317 + +# Checker settings carried over from the previous unit command lines +# (defaults: unlimited compute threads, 15 minute recheck). +MAX_COMPUTE_THREADS=2 +RECHECK_FREQ_MIN=50 diff --git a/debian/usr/lib/systemd/system/monad-services-archive-checker.service b/debian/usr/lib/systemd/system/monad-services-archive-checker.service index a329031..da3f79d 100644 --- a/debian/usr/lib/systemd/system/monad-services-archive-checker.service +++ b/debian/usr/lib/systemd/system/monad-services-archive-checker.service @@ -7,11 +7,19 @@ Wants=network-online.target Type=simple User=monad Group=monad +StateDirectory=monad-services +WorkingDirectory=/var/lib/monad-services EnvironmentFile=-/etc/monad-services/env -ExecStart=/usr/lib/monad-services/monad-archive-checker +ExecStart=/usr/lib/monad-services/monad-archive check checker Restart=always RestartSec=5 +DeviceAllow=/dev/triedb rw LimitNOFILE=1048576 +# Hugemem configs below +LimitMEMLOCK=infinity +MemoryDenyWriteExecute=false +ProtectKernelModules=no +ProtectKernelLogs=no [Install] WantedBy=multi-user.target diff --git a/debian/usr/lib/systemd/system/monad-services-archiver.service b/debian/usr/lib/systemd/system/monad-services-archiver.service index 701fa8a..9263153 100644 --- a/debian/usr/lib/systemd/system/monad-services-archiver.service +++ b/debian/usr/lib/systemd/system/monad-services-archiver.service @@ -7,11 +7,19 @@ Wants=network-online.target Type=simple User=monad Group=monad +StateDirectory=monad-services +WorkingDirectory=/var/lib/monad-services EnvironmentFile=-/etc/monad-services/env -ExecStart=/usr/lib/monad-services/monad-archiver +ExecStart=/usr/lib/monad-services/monad-archive archive Restart=always RestartSec=5 +DeviceAllow=/dev/triedb rw LimitNOFILE=1048576 +# Hugemem configs below +LimitMEMLOCK=infinity +MemoryDenyWriteExecute=false +ProtectKernelModules=no +ProtectKernelLogs=no [Install] WantedBy=multi-user.target diff --git a/debian/usr/lib/systemd/system/monad-services-block-writer.service b/debian/usr/lib/systemd/system/monad-services-block-writer.service index 514bee4..fbdb9f1 100644 --- a/debian/usr/lib/systemd/system/monad-services-block-writer.service +++ b/debian/usr/lib/systemd/system/monad-services-block-writer.service @@ -7,11 +7,19 @@ Wants=network-online.target Type=simple User=monad Group=monad +StateDirectory=monad-services +WorkingDirectory=/var/lib/monad-services EnvironmentFile=-/etc/monad-services/env -ExecStart=/usr/lib/monad-services/monad-block-writer +ExecStart=/usr/lib/monad-services/monad-archive block-writer stream Restart=always RestartSec=5 +DeviceAllow=/dev/triedb rw LimitNOFILE=1048576 +# Hugemem configs below +LimitMEMLOCK=infinity +MemoryDenyWriteExecute=false +ProtectKernelModules=no +ProtectKernelLogs=no [Install] WantedBy=multi-user.target diff --git a/debian/usr/lib/systemd/system/monad-services-indexer.service b/debian/usr/lib/systemd/system/monad-services-indexer.service index 4896c47..85fad7a 100644 --- a/debian/usr/lib/systemd/system/monad-services-indexer.service +++ b/debian/usr/lib/systemd/system/monad-services-indexer.service @@ -7,11 +7,19 @@ Wants=network-online.target Type=simple User=monad Group=monad +StateDirectory=monad-services +WorkingDirectory=/var/lib/monad-services EnvironmentFile=-/etc/monad-services/env -ExecStart=/usr/lib/monad-services/monad-indexer +ExecStart=/usr/lib/monad-services/monad-archive index Restart=always RestartSec=5 +DeviceAllow=/dev/triedb rw LimitNOFILE=1048576 +# Hugemem configs below +LimitMEMLOCK=infinity +MemoryDenyWriteExecute=false +ProtectKernelModules=no +ProtectKernelLogs=no [Install] WantedBy=multi-user.target diff --git a/monad-archive/Cargo.toml b/monad-archive/Cargo.toml index 9e8dd3b..d8644aa 100644 --- a/monad-archive/Cargo.toml +++ b/monad-archive/Cargo.toml @@ -3,6 +3,8 @@ name = "monad-archive" description = "Monad archive solution" version = "0.1.0" edition = "2021" +# src/bin/ holds the modules of the single monad-archive binary, not a binary each. +autobins = false # this is necessary for criterion bench options # https://bheisler.github.io/criterion.rs/book/faq.html#cargo-bench-gives-unrecognized-option-errors-for-valid-command-line-options @@ -10,19 +12,8 @@ edition = "2021" bench = false [[bin]] -name = "monad-archive-checker" -bench = false - -[[bin]] -name = "monad-archiver" -bench = false - -[[bin]] -name = "monad-block-writer" -bench = false - -[[bin]] -name = "monad-indexer" +name = "monad-archive" +path = "src/bin/main.rs" bench = false [features] @@ -48,7 +39,7 @@ aws-sdk-s3 = { workspace = true } base64 = { workspace = true } bytes = { workspace = true } chrono = { workspace = true } -clap = { workspace = true, features = ["derive"] } +clap = { workspace = true, features = ["derive", "env"] } dashmap = { workspace = true } enum_dispatch = { workspace = true } eyre = { workspace = true } diff --git a/monad-archive/src/bin/monad-archiver/bft_archive_worker.rs b/monad-archive/src/bin/archive/bft_archive_worker.rs similarity index 100% rename from monad-archive/src/bin/monad-archiver/bft_archive_worker.rs rename to monad-archive/src/bin/archive/bft_archive_worker.rs diff --git a/monad-archive/src/bin/monad-archiver/block_archive_worker.rs b/monad-archive/src/bin/archive/block_archive_worker.rs similarity index 100% rename from monad-archive/src/bin/monad-archiver/block_archive_worker.rs rename to monad-archive/src/bin/archive/block_archive_worker.rs diff --git a/monad-archive/src/bin/monad-archiver/cli.rs b/monad-archive/src/bin/archive/cli.rs similarity index 93% rename from monad-archive/src/bin/monad-archiver/cli.rs rename to monad-archive/src/bin/archive/cli.rs index f52fc2a..57ac6c5 100644 --- a/monad-archive/src/bin/monad-archiver/cli.rs +++ b/monad-archive/src/bin/archive/cli.rs @@ -24,19 +24,19 @@ use eyre::{eyre, Context, Result}; use monad_archive::cli::{ArchiveArgs, BlockDataReaderArgs}; use serde::Deserialize; -/// Runtime configuration for the `monad-archiver` binary. +/// Runtime configuration for the `monad-archive archive` subcommand. /// /// Values can come from either a `--config path/to/config.toml` file or from /// CLI flags. When both are supplied, CLI arguments win. For example: /// /// ```text -/// monad-archiver --config config.toml --max-blocks-per-iteration 50 +/// monad-archive archive --config config.toml --max-blocks-per-iteration 50 /// ``` /// /// will load every field from `config.toml` and then replace the /// `max_blocks_per_iteration` value with `50` before execution. #[derive(Debug, Deserialize)] -pub struct Cli { +pub struct ArchiveRunCli { /// Where blocks, receipts and traces are read from /// For triedb: 'triedb ' pub block_data_source: BlockDataReaderArgs, @@ -143,43 +143,42 @@ pub struct Cli { } /// Result of parsing CLI arguments - either a subcommand or daemon config -pub enum ParsedCli { +pub enum ArchiveRunParsedCli { /// A subcommand was provided - handle and exit Command(Commands), /// No subcommand - run as daemon with this config - Daemon(Cli), + Daemon(ArchiveRunCli), } -impl Cli { - pub fn parse() -> ParsedCli { - Self::try_parse().unwrap_or_else(|err| { +impl ArchiveRunCli { + pub fn parse(args: ArchiveRunCliArgs) -> ArchiveRunParsedCli { + Self::try_parse(args).unwrap_or_else(|err| { eprintln!("failed to load monad-archiver configuration: {err:?}"); process::exit(2); }) } - pub fn try_parse() -> Result { - let args = CliArgs::parse(); + pub fn try_parse(args: ArchiveRunCliArgs) -> Result { // If a subcommand is provided, return it without requiring daemon args if let Some(command) = args.command { - return Ok(ParsedCli::Command(command)); + return Ok(ArchiveRunParsedCli::Command(command)); } // No subcommand - parse full daemon config (this requires block_data_source, archive_sink, etc.) - let (_, cli) = CliArgs { + let (_, cli) = ArchiveRunCliArgs { command: None, ..args } .into_cli()?; - Ok(ParsedCli::Daemon(cli)) + Ok(ArchiveRunParsedCli::Daemon(cli)) } - fn from_sources(config: Option, overrides: CliOverrides) -> Result { + fn from_sources(config: Option, overrides: CliOverrides) -> Result { match config { Some(mut cli) => { cli.apply_overrides(overrides); Ok(cli) } - None => Cli::from_overrides(overrides), + None => Self::from_overrides(overrides), } } @@ -356,8 +355,7 @@ pub enum Commands { } #[derive(Debug, Parser)] -#[command(name = "monad-archive", about, long_about = None)] -struct CliArgs { +pub struct ArchiveRunCliArgs { #[command(subcommand)] pub command: Option, @@ -367,23 +365,23 @@ struct CliArgs { /// Where blocks, receipts and traces are read from /// For triedb: 'triedb ' - #[arg(long, value_parser = clap::value_parser!(BlockDataReaderArgs))] + #[arg(long, env = "BLOCK_DATA_SOURCE", value_parser = clap::value_parser!(BlockDataReaderArgs))] block_data_source: Option, /// If reading from --block-data-source fails, attempts to read from /// this optional fallback - #[arg(long, value_parser = clap::value_parser!(BlockDataReaderArgs))] + #[arg(long, env = "FALLBACK_BLOCK_DATA_SOURCE", value_parser = clap::value_parser!(BlockDataReaderArgs))] fallback_block_data_source: Option, /// Where archive data is written to /// For aws: 'aws ' - #[arg(long, value_parser = clap::value_parser!(ArchiveArgs))] + #[arg(long, env = "ARCHIVE_SINK", value_parser = clap::value_parser!(ArchiveArgs))] archive_sink: Option, #[arg(long)] max_blocks_per_iteration: Option, - #[arg(long)] + #[arg(long, env = "MAX_CONCURRENT_BLOCKS")] max_concurrent_blocks: Option, /// Override block number to stop at @@ -411,7 +409,7 @@ struct CliArgs { /// Path to folder containing bft blocks /// If set, archiver will upload these files to blob store provided in archive_sink - #[arg(long)] + #[arg(long, env = "BFT_BLOCK_PATH")] bft_block_path: Option, #[arg(long)] @@ -467,7 +465,7 @@ struct CliArgs { #[arg(long, action = ArgAction::SetTrue)] unsafe_allow_traces_overwrite: bool, - #[arg(long)] + #[arg(long, env = "OTEL_ENDPOINT")] otel_endpoint: Option, #[arg(long)] @@ -477,14 +475,14 @@ struct CliArgs { skip_connectivity_check: bool, } -impl CliArgs { - fn into_cli(self) -> Result<(Option, Cli)> { +impl ArchiveRunCliArgs { + fn into_cli(self) -> Result<(Option, ArchiveRunCli)> { let (command, config_path, overrides) = self.into_parts(); let config = match config_path { Some(path) => Some(load_config(&path)?), None => None, }; - Ok((command, Cli::from_sources(config, overrides)?)) + Ok((command, ArchiveRunCli::from_sources(config, overrides)?)) } fn into_parts(self) -> (Option, Option, CliOverrides) { @@ -588,7 +586,7 @@ struct CliOverrides { unsafe_allow_traces_overwrite: Option, } -fn load_config(path: &Path) -> Result { +fn load_config(path: &Path) -> Result { let contents = fs::read_to_string(path) .wrap_err_with(|| format!("failed to read config file {}", path.display()))?; toml::from_str(&contents) @@ -683,7 +681,7 @@ mod tests { db = "sink-db" "#; - let cli: Cli = toml::from_str(config).expect("toml should deserialize"); + let cli: ArchiveRunCli = toml::from_str(config).expect("toml should deserialize"); assert_eq!(cli.max_blocks_per_iteration, 250); assert_eq!(cli.max_concurrent_blocks, 32); @@ -750,7 +748,7 @@ mod tests { concurrency = 50 "#; - let cli: Cli = toml::from_str(config).expect("toml should deserialize"); + let cli: ArchiveRunCli = toml::from_str(config).expect("toml should deserialize"); assert_eq!(cli.max_blocks_per_iteration, 100); assert_eq!(cli.max_concurrent_blocks, 15); @@ -802,10 +800,13 @@ mod tests { ) .unwrap(); - let (_, cli) = - CliArgs::parse_from(["monad-archiver", "--config", file.path().to_str().unwrap()]) - .into_cli() - .expect("config file should load"); + let (_, cli) = ArchiveRunCliArgs::parse_from([ + "monad-archiver", + "--config", + file.path().to_str().unwrap(), + ]) + .into_cli() + .expect("config file should load"); assert_eq!(cli.max_blocks_per_iteration, 222); assert_eq!(cli.max_concurrent_blocks, 16); @@ -843,7 +844,7 @@ mod tests { ) .unwrap(); - let (_, cli) = CliArgs::parse_from([ + let (_, cli) = ArchiveRunCliArgs::parse_from([ "monad-archiver", "--config", file.path().to_str().unwrap(), diff --git a/monad-archive/src/bin/monad-archiver/file_checkpointer.rs b/monad-archive/src/bin/archive/file_checkpointer.rs similarity index 100% rename from monad-archive/src/bin/monad-archiver/file_checkpointer.rs rename to monad-archive/src/bin/archive/file_checkpointer.rs diff --git a/monad-archive/src/bin/monad-archiver/generic_folder_archiver.rs b/monad-archive/src/bin/archive/generic_folder_archiver.rs similarity index 100% rename from monad-archive/src/bin/monad-archiver/generic_folder_archiver.rs rename to monad-archive/src/bin/archive/generic_folder_archiver.rs diff --git a/monad-archive/src/bin/monad-archiver/main.rs b/monad-archive/src/bin/archive/mod.rs similarity index 95% rename from monad-archive/src/bin/monad-archiver/main.rs rename to monad-archive/src/bin/archive/mod.rs index 3e89824..4cf0f46 100644 --- a/monad-archive/src/bin/monad-archiver/main.rs +++ b/monad-archive/src/bin/archive/mod.rs @@ -24,26 +24,24 @@ mod generic_folder_archiver; use bft_archive_worker::bft_block_archive_worker; use block_archive_worker::{archive_worker, ArchiveWorkerOpts}; -use cli::{Commands, ParsedCli}; +use cli::{ArchiveRunCli, ArchiveRunCliArgs, ArchiveRunParsedCli, Commands}; use file_checkpointer::file_checkpoint_worker; use generic_folder_archiver::recursive_dir_archiver; use tokio::task::JoinHandle; use tracing::Level; -mod cli; +pub mod cli; -#[tokio::main(flavor = "multi_thread")] -async fn main() -> Result<()> { +pub async fn run(args: ArchiveRunCliArgs) -> Result<()> { tracing_subscriber::fmt().with_max_level(Level::INFO).init(); - let parsed = cli::Cli::parse(); - // Handle subcommands - if let ParsedCli::Command(cmd) = parsed { + let parsed = ArchiveRunCli::parse(args); + if let ArchiveRunParsedCli::Command(cmd) = parsed { return handle_command(cmd).await; } - let ParsedCli::Daemon(args) = parsed else { + let ArchiveRunParsedCli::Daemon(args) = parsed else { unreachable!() }; info!(?args, "Cli Arguments: "); diff --git a/monad-archive/src/bin/monad-block-writer/cli.rs b/monad-archive/src/bin/block_writer/cli.rs similarity index 93% rename from monad-archive/src/bin/monad-block-writer/cli.rs rename to monad-archive/src/bin/block_writer/cli.rs index 9694b79..3b940e6 100644 --- a/monad-archive/src/bin/monad-block-writer/cli.rs +++ b/monad-archive/src/bin/block_writer/cli.rs @@ -19,8 +19,8 @@ use clap::{Parser, Subcommand}; use monad_archive::cli::BlockDataReaderArgs; #[derive(Debug, Parser)] -#[command(name = "monad-archive", about, long_about = None)] -pub struct Cli { +#[command(about, long_about = None)] +pub struct ArchiveBlockWriterCli { #[command(subcommand)] pub mode: Mode, } @@ -28,7 +28,7 @@ pub struct Cli { #[derive(Debug, Parser)] pub struct SharedArgs { /// Source to read block data that will be indexed - #[arg(long, value_parser = clap::value_parser!(BlockDataReaderArgs))] + #[arg(long, env = "BLOCK_DATA_SOURCE", value_parser = clap::value_parser!(BlockDataReaderArgs))] pub block_data_source: BlockDataReaderArgs, /// If reading from --block-data-source fails, attempts to read from @@ -36,7 +36,7 @@ pub struct SharedArgs { #[arg(long, value_parser = clap::value_parser!(BlockDataReaderArgs))] pub fallback_block_data_source: Option, - #[arg(long)] + #[arg(long, env = "BLOCK_WRITER_DEST_PATH")] pub dest_path: PathBuf, /// Maximum number of retries for failed blocks diff --git a/monad-archive/src/bin/monad-block-writer/main.rs b/monad-archive/src/bin/block_writer/mod.rs similarity index 99% rename from monad-archive/src/bin/monad-block-writer/main.rs rename to monad-archive/src/bin/block_writer/mod.rs index be724e9..50a25e4 100644 --- a/monad-archive/src/bin/monad-block-writer/main.rs +++ b/monad-archive/src/bin/block_writer/mod.rs @@ -17,12 +17,11 @@ use std::vec::IntoIter; use alloy_consensus::Block as AlloyBlock; use alloy_rlp::Encodable; -use clap::Parser; use monad_archive::{kvstore::WritePolicy, prelude::*}; use monad_compress::{brotli::BrotliCompression, CompressionAlgo}; use tracing::Level; -mod cli; +pub mod cli; async fn process_block( reader: &BlockDataReaderErased, @@ -75,11 +74,8 @@ async fn process_block( Ok(()) } -#[tokio::main(flavor = "current_thread")] -async fn main() -> Result<()> { +pub async fn run(args: cli::ArchiveBlockWriterCli) -> Result<()> { tracing_subscriber::fmt().with_max_level(Level::INFO).init(); - - let args = cli::Cli::parse(); info!(?args, "Cli Arguments: "); // Handle SetStartBlock separately since it doesn't need shared args diff --git a/monad-archive/src/bin/monad-archive-checker/checker.rs b/monad-archive/src/bin/check/checker.rs similarity index 94% rename from monad-archive/src/bin/monad-archive-checker/checker.rs rename to monad-archive/src/bin/check/checker.rs index 526b411..e5ea9f8 100644 --- a/monad-archive/src/bin/monad-archive-checker/checker.rs +++ b/monad-archive/src/bin/check/checker.rs @@ -17,10 +17,10 @@ use std::collections::HashMap; use alloy_consensus::proofs::{calculate_receipt_root, calculate_transaction_root}; use eyre::Result; -use futures::stream; +use futures::{stream, Stream, StreamExt}; use monad_archive::prelude::*; -use crate::{ +use crate::check::{ model::{CheckerModel, Fault, FaultKind, GoodBlocks, InconsistentBlockReason}, CHUNK_SIZE, }; @@ -101,15 +101,11 @@ async fn process_block_batch( "Fetching block data from replicas" ); - let data_by_block_num = - fetch_block_data(model, start_block..=end_block, &replicas, concurrency).await; - - debug!("Fetched data for {} blocks", data_by_block_num.len()); - - // Process blocks to find faults and good blocks - info!("Processing blocks to find faults and good blocks"); + // Fetch and check in one pass, so only `concurrency` blocks are held at a + // time rather than the whole chunk. + info!("Fetching and processing blocks to find faults and good blocks"); let (faults_by_replica, good_blocks) = - process_blocks(&data_by_block_num, start_block, end_block); + stream_and_process_blocks(model, start_block..=end_block, &replicas, concurrency).await; // Count total faults and good blocks let total_faults: usize = faults_by_replica.values().map(|v| v.len()).sum(); @@ -144,8 +140,29 @@ pub async fn fetch_block_data( ) -> HashMap>> { debug!("Fetching block data for {} replicas", replicas.len()); + fetch_block_data_stream(model, block_nums, replicas, concurrency) + .collect() + .await +} + +/// Yields each block's per-replica data as it arrives, in block order. +/// +/// `buffered` keeps at most `concurrency` blocks in flight, so a consumer that +/// handles each block as it lands never holds more than that, however long the +/// requested range is. +fn fetch_block_data_stream<'a>( + model: &'a CheckerModel, + block_nums: impl IntoIterator + 'a, + replicas: &'a [&'a str], + concurrency: usize, +) -> impl Stream< + Item = ( + u64, + HashMap>, + ), +> + 'a { stream::iter(block_nums) - .map(|block_num| async move { + .map(move |block_num| async move { let mut block_data = HashMap::new(); debug!(block_num, "Fetching data for block"); @@ -162,45 +179,102 @@ pub async fn fetch_block_data( (block_num, block_data) }) .buffered(concurrency) - .collect::>, - )>>() - .await - .into_iter() - .collect() +} + +/// Fetches and checks a range of blocks in one pass. +/// +/// Equivalent to `fetch_block_data` followed by `process_blocks`, but each block +/// is dropped once it has been checked instead of the whole range being held in +/// memory first. Only the faults and good-block markers are retained, and those +/// are a few bytes per block. +pub async fn stream_and_process_blocks( + model: &CheckerModel, + block_nums: impl IntoIterator, + replicas: &[&str], + concurrency: usize, +) -> (HashMap>, GoodBlocks) { + let mut processor = BlockProcessor::default(); + let blocks = fetch_block_data_stream(model, block_nums, replicas, concurrency); + let mut blocks = std::pin::pin!(blocks); + + while let Some((block_num, replica_data)) = blocks.next().await { + debug!(block_num, "Processing block"); + processor.process(block_num, &replica_data); + } + + processor.finish() +} + +/// Checks blocks one at a time, carrying the previous block's headers forward so +/// the parent-hash check still works without keeping earlier blocks around. +#[derive(Default)] +struct BlockProcessor { + faults_by_replica: HashMap>, + good_blocks: GoodBlocks, + prev_headers: HashMap, +} + +impl BlockProcessor { + fn process( + &mut self, + block_num: u64, + replica_data: &HashMap>, + ) { + let parents = std::mem::take(&mut self.prev_headers); + process_single_block( + block_num, + replica_data, + parents, + &mut self.faults_by_replica, + &mut self.good_blocks, + ); + self.prev_headers = replica_data + .iter() + .filter_map(|(replica_name, block_data)| { + Some((replica_name.clone(), block_data.as_ref()?.0.header.clone())) + }) + .collect(); + } + + /// Drops the carried headers, so the next block is treated as having no + /// parent -- what a gap in the range means. + #[cfg(test)] + fn skip(&mut self) { + self.prev_headers.clear(); + } + + fn finish(self) -> (HashMap>, GoodBlocks) { + (self.faults_by_replica, self.good_blocks) + } } /// Processes blocks to find faults and good blocks by comparing data across replicas. +/// +/// Retained for tests that drive the checking logic from a hand-built map; +/// production paths use [`stream_and_process_blocks`], which shares the same +/// [`BlockProcessor`] core. +#[cfg(test)] pub fn process_blocks( data_by_block_num: &HashMap>>, start_block: u64, end_block: u64, ) -> (HashMap>, GoodBlocks) { - let mut faults_by_replica: HashMap> = HashMap::new(); - let mut good_blocks = GoodBlocks { - block_num_to_replica: HashMap::new(), - }; + let mut processor = BlockProcessor::default(); debug!(start_block, end_block, "Processing blocks"); for block_num in start_block..=end_block { if let Some(replica_data) = data_by_block_num.get(&block_num) { - let prev_headers = get_prev_header(block_num, data_by_block_num); - debug!(block_num, "Processing block"); - process_single_block( - block_num, - replica_data, - prev_headers, - &mut faults_by_replica, - &mut good_blocks, - ); + processor.process(block_num, replica_data); } else { debug!(block_num, "No data found for block"); + processor.skip(); } } + let (faults_by_replica, good_blocks) = processor.finish(); + // Log summary of processed blocks let total_faults: usize = faults_by_replica.values().map(|v| v.len()).sum(); let good_block_count = good_blocks.block_num_to_replica.len(); @@ -210,24 +284,6 @@ pub fn process_blocks( (faults_by_replica, good_blocks) } -fn get_prev_header( - block_num: u64, - data_by_block_num: &HashMap>>, -) -> HashMap { - let Some(prev_block_num) = block_num.checked_sub(1) else { - return HashMap::new(); - }; - let Some(prev_block_data) = data_by_block_num.get(&prev_block_num) else { - return HashMap::new(); - }; - prev_block_data - .iter() - .filter_map(|(replica_name, block_data)| { - Some((replica_name.clone(), block_data.as_ref()?.0.header.clone())) - }) - .collect() -} - /// Processes a single block across all replicas pub fn process_single_block( block_num: u64, @@ -841,7 +897,7 @@ pub mod tests { }; use super::*; - use crate::model::InconsistentBlockReason; + use crate::check::model::InconsistentBlockReason; #[test] fn test_process_blocks_empty() { diff --git a/monad-archive/src/bin/monad-archive-checker/cli.rs b/monad-archive/src/bin/check/cli.rs similarity index 87% rename from monad-archive/src/bin/monad-archive-checker/cli.rs rename to monad-archive/src/bin/check/cli.rs index d33dc2e..eaa89c9 100644 --- a/monad-archive/src/bin/monad-archive-checker/cli.rs +++ b/monad-archive/src/bin/check/cli.rs @@ -18,50 +18,49 @@ use monad_archive::cli::ArchiveArgs; #[derive(Debug, Parser)] #[command( - name = "monad-archive-checker", about = "Archive consistency checker for validating blockchain data across multiple replicas", long_about = "Archive consistency checker for validating blockchain data across multiple replicas.\n\n\ EXAMPLES:\n\n\ # Start main checker with 3 replicas (runs continuously)\n\ - monad-archive-checker --bucket checker-state --region us-east-1 checker \\\n\ + monad-archive check --bucket checker-state --region us-east-1 checker \\\n\ --init-replicas 'aws archive-1 20,aws archive-2 20,aws archive-3 20'\n\n\ # Inspect specific faults\n\ - monad-archive-checker --bucket checker-state inspector list-faults\n\ - monad-archive-checker --bucket checker-state inspector inspect-block 12345 --format all\n\n\ + monad-archive check --bucket checker-state inspector list-faults\n\ + monad-archive check --bucket checker-state inspector inspect-block 12345 --format all\n\n\ # Fix faults by copying from good replicas (dry run first)\n\ - monad-archive-checker --bucket checker-state fault-fixer\n\ - monad-archive-checker --bucket checker-state fault-fixer --commit-changes --verify\n\n\ + monad-archive check --bucket checker-state fault-fixer\n\ + monad-archive check --bucket checker-state fault-fixer --commit-changes --verify\n\n\ # Run standalone rechecker to fix false positives (runs once and exits)\n\ - monad-archive-checker --bucket checker-state rechecker\n\n\ + monad-archive check --bucket checker-state rechecker\n\n\ # Run rechecker in worker mode (runs periodically)\n\ - monad-archive-checker --bucket checker-state rechecker --worker --recheck-freq-min 5\n\n\ + monad-archive check --bucket checker-state rechecker --worker --recheck-freq-min 5\n\n\ # Advanced: Recheck specific block range with dry run\n\ - monad-archive-checker --bucket checker-state rechecker \\\n\ + monad-archive check --bucket checker-state rechecker \\\n\ --start-block 1000 --end-block 5000 --dry-run\n\n\ # Force recheck all chunks even without faults\n\ - monad-archive-checker --bucket checker-state rechecker \\\n\ + monad-archive check --bucket checker-state rechecker \\\n\ --start-block 0 --end-block 10000 --force-recheck --dry-run\n\n\ " )] -pub struct Cli { +pub struct ArchiveCheckCli { #[command(subcommand)] pub mode: Mode, /// S3 bucket name for storing checker state - #[arg(long)] + #[arg(long, env = "ARCHIVE_BUCKET")] pub bucket: String, /// AWS region #[arg(long, global = true)] pub region: Option, - #[arg(long, global = true)] + #[arg(long, global = true, env = "OTEL_ENDPOINT")] pub otel_endpoint: Option, #[arg(long, global = true)] pub otel_replica_name_override: Option, - #[arg(long, global = true)] + #[arg(long, global = true, env = "MAX_COMPUTE_THREADS")] pub max_compute_threads: Option, } @@ -81,7 +80,7 @@ pub enum Mode { pub struct CheckerArgs { /// Comma-separated list of replicas to check /// Format: 'aws bucket1 [concurrency1] [region1],aws bucket2 [concurrency2] [region2],...' - #[arg(long, value_delimiter = ',', value_parser = clap::value_parser!(ArchiveArgs))] + #[arg(long, env = "INIT_REPLICAS", value_delimiter = ',', value_parser = clap::value_parser!(ArchiveArgs))] pub init_replicas: Option>, /// Flag to disable running rechecker worker to determine if existing faults still exist @@ -96,7 +95,7 @@ pub struct CheckerArgs { pub min_lag_from_tip: u64, /// How frequently to recheck faults in minutes - #[arg(long, default_value_t = 15.)] + #[arg(long, env = "RECHECK_FREQ_MIN", default_value_t = 15.)] pub recheck_freq_min: f64, /// How many blocks to process in parallel @@ -228,7 +227,7 @@ mod tests { #[test] fn global_args_work_after_subcommand() { - let cli = Cli::try_parse_from([ + let cli = ArchiveCheckCli::try_parse_from([ "monad-archive-checker", "--bucket", "test-bucket", diff --git a/monad-archive/src/bin/monad-archive-checker/fault_fixer.rs b/monad-archive/src/bin/check/fault_fixer.rs similarity index 99% rename from monad-archive/src/bin/monad-archive-checker/fault_fixer.rs rename to monad-archive/src/bin/check/fault_fixer.rs index bba9c28..7270995 100644 --- a/monad-archive/src/bin/monad-archive-checker/fault_fixer.rs +++ b/monad-archive/src/bin/check/fault_fixer.rs @@ -21,7 +21,7 @@ use monad_archive::{kvstore::WritePolicy, prelude::*}; use opentelemetry::KeyValue; use tracing::{debug, error, info, warn}; -use crate::{ +use crate::check::{ model::{CheckerModel, Fault, FaultKind}, rechecker_v2::recheck_chunk_from_scratch, }; @@ -391,7 +391,7 @@ mod tests { use monad_archive::prelude::LatestKind; use super::*; - use crate::{ + use crate::check::{ checker::tests::{create_test_block_data, create_test_block_data_range, setup_test_model}, model::{GoodBlocks, InconsistentBlockReason}, }; diff --git a/monad-archive/src/bin/monad-archive-checker/inspector.rs b/monad-archive/src/bin/check/inspector.rs similarity index 99% rename from monad-archive/src/bin/monad-archive-checker/inspector.rs rename to monad-archive/src/bin/check/inspector.rs index 6f0767f..b6de530 100644 --- a/monad-archive/src/bin/monad-archive-checker/inspector.rs +++ b/monad-archive/src/bin/check/inspector.rs @@ -21,7 +21,7 @@ use std::{ use eyre::Result; use monad_archive::prelude::*; -use crate::{ +use crate::check::{ checker::fetch_block_data, model::{CheckerModel, Fault}, CHUNK_SIZE, @@ -65,7 +65,7 @@ pub async fn status(model: &CheckerModel) -> Result<()> { println!("\nFault Chunks:"); let fault_keys = model .store - .scan_prefix(crate::model::FAULTS_CHUNK_PREFIX) + .scan_prefix(crate::check::model::FAULTS_CHUNK_PREFIX) .await?; // Group by replica @@ -270,7 +270,7 @@ fn format_ranges(ranges: &[(u64, u64)]) -> String { } // Re-export the OutputFormat from cli module to avoid duplication -pub use crate::cli::InspectorOutputFormat as OutputFormat; +pub use crate::check::cli::InspectorOutputFormat as OutputFormat; /// Inspects a single block across all replicas pub async fn inspect_block( diff --git a/monad-archive/src/bin/monad-archive-checker/main.rs b/monad-archive/src/bin/check/mod.rs similarity index 97% rename from monad-archive/src/bin/monad-archive-checker/main.rs rename to monad-archive/src/bin/check/mod.rs index eed06e4..428ab63 100644 --- a/monad-archive/src/bin/monad-archive-checker/main.rs +++ b/monad-archive/src/bin/check/mod.rs @@ -19,14 +19,13 @@ //! blocks, receipts, and traces across multiple archival replicas. It detects //! and reports inconsistencies, missing data, and corrupted blocks. -use clap::Parser; use eyre::Result; use model::CheckerModel; use monad_archive::{cli::get_aws_config, prelude::*}; use tracing_subscriber::EnvFilter; mod checker; -mod cli; +pub mod cli; mod fault_fixer; mod inspector; mod model; @@ -37,19 +36,17 @@ mod rechecker_v2; /// A chunk is the smallest unit of blocks that can be rechecked or have faults cleared. pub const CHUNK_SIZE: u64 = 1000; -#[tokio::main] -async fn main() -> Result<()> { +pub async fn run(args: cli::ArchiveCheckCli) -> Result<()> { // Initialize logging tracing_subscriber::fmt() .with_env_filter( EnvFilter::builder() - .with_default_directive("monad_archive_checker=debug".parse()?) + .with_default_directive("monad_archive::check=debug".parse()?) .from_env_lossy(), ) .init(); // Parse command line arguments - let args = cli::Cli::parse(); info!("Starting monad-archive-checker with mode: {:?}", args.mode); if let Some(max_compute_threads) = args.max_compute_threads.as_ref() { diff --git a/monad-archive/src/bin/monad-archive-checker/model.rs b/monad-archive/src/bin/check/model.rs similarity index 99% rename from monad-archive/src/bin/monad-archive-checker/model.rs rename to monad-archive/src/bin/check/model.rs index 6e7ca61..4c3865e 100644 --- a/monad-archive/src/bin/monad-archive-checker/model.rs +++ b/monad-archive/src/bin/check/model.rs @@ -282,7 +282,7 @@ impl CheckerModel { .filter(|chunk_start| { // A chunk starting at position X contains blocks from X to X+CHUNK_SIZE-1 // Include the chunk if it could contain any blocks in the range [start, end] - use crate::CHUNK_SIZE; + use crate::check::CHUNK_SIZE; if let Some(end) = end { // Chunk should start before or at the end block @@ -309,7 +309,7 @@ impl CheckerModel { /// Find a common prefix for chunk starts in a given range /// Returns empty string if no useful prefix exists fn find_chunk_range_prefix(start: Option, end: Option) -> String { - use crate::CHUNK_SIZE; + use crate::check::CHUNK_SIZE; match (start, end) { (Some(start), Some(end)) => { diff --git a/monad-archive/src/bin/monad-archive-checker/rechecker_v2.rs b/monad-archive/src/bin/check/rechecker_v2.rs similarity index 99% rename from monad-archive/src/bin/monad-archive-checker/rechecker_v2.rs rename to monad-archive/src/bin/check/rechecker_v2.rs index 0e5af92..fe7eb07 100644 --- a/monad-archive/src/bin/monad-archive-checker/rechecker_v2.rs +++ b/monad-archive/src/bin/check/rechecker_v2.rs @@ -18,8 +18,8 @@ use monad_archive::{kvstore::WritePolicy, prelude::*}; use opentelemetry::KeyValue; use tokio::time::interval; -use crate::{ - checker::{fetch_block_data, process_blocks, store_checking_results}, +use crate::check::{ + checker::{store_checking_results, stream_and_process_blocks}, model::{CheckerModel, Fault, FaultKind, GoodBlocks}, CHUNK_SIZE, }; @@ -280,12 +280,9 @@ pub async fn recheck_chunk_from_scratch( .map(String::as_str) .collect::>(); - let data_by_block_num = - fetch_block_data(model, chunk_start..=end_block, &replicas, concurrency).await; - - // Process blocks to find faults and good blocks using original checker logic + // Fetch and process blocks in one pass using original checker logic let (new_faults_by_replica, new_good_blocks) = - process_blocks(&data_by_block_num, chunk_start, end_block); + stream_and_process_blocks(model, chunk_start..=end_block, &replicas, concurrency).await; // Get old results for comparison let old_faults_by_replica = model.get_faults_chunks_all_replicas(chunk_start).await?; @@ -537,7 +534,7 @@ mod tests { use monad_archive::{kvstore::WritePolicy, prelude::LatestKind}; use super::*; - use crate::{ + use crate::check::{ checker::tests::{create_test_block_data, create_test_block_data_range, setup_test_model}, model::{GoodBlocks, InconsistentBlockReason}, }; diff --git a/monad-archive/src/bin/cli.rs b/monad-archive/src/bin/cli.rs new file mode 100644 index 0000000..1aed91a --- /dev/null +++ b/monad-archive/src/bin/cli.rs @@ -0,0 +1,40 @@ +// Copyright (C) 2025 Category Labs, Inc. +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU General Public License for more details. +// +// You should have received a copy of the GNU General Public License +// along with this program. If not, see . + +use clap::{Parser, Subcommand}; + +use crate::{ + archive::cli::ArchiveRunCliArgs, block_writer::cli::ArchiveBlockWriterCli, + check::cli::ArchiveCheckCli, index::cli::ArchiveIndexCliArgs, +}; + +#[derive(Debug, Parser)] +#[command(name = "monad-archive", about = "Monad archive tools")] +pub struct ArchiveCli { + #[command(subcommand)] + pub command: ArchiveCliCommand, +} + +#[derive(Debug, Subcommand)] +pub enum ArchiveCliCommand { + /// Archive blocks, receipts and traces to the archive sink + Archive(ArchiveRunCliArgs), + /// Index archived blocks for RPC queries + Index(ArchiveIndexCliArgs), + /// Check archive consistency across replicas + Check(ArchiveCheckCli), + /// Write block data out to a filesystem path + BlockWriter(ArchiveBlockWriterCli), +} diff --git a/monad-archive/src/bin/monad-indexer/cli.rs b/monad-archive/src/bin/index/cli.rs similarity index 90% rename from monad-archive/src/bin/monad-indexer/cli.rs rename to monad-archive/src/bin/index/cli.rs index 07f5493..1e92406 100644 --- a/monad-archive/src/bin/monad-indexer/cli.rs +++ b/monad-archive/src/bin/index/cli.rs @@ -20,7 +20,7 @@ use eyre::{eyre, Result}; use monad_archive::cli::{ArchiveArgs, BlockDataReaderArgs}; #[derive(Debug)] -pub struct Cli { +pub struct ArchiveIndexCli { pub block_data_source: BlockDataReaderArgs, pub fallback_block_data_source: Option, pub archive_sink: ArchiveArgs, @@ -36,37 +36,26 @@ pub struct Cli { pub enable_logs_indexing: bool, } -pub enum ParsedCli { - Command { command: Commands, args: CliArgs }, - Daemon(Cli), +pub enum ArchiveIndexParsedCli { + Command { + command: Commands, + args: ArchiveIndexCliArgs, + }, + Daemon(ArchiveIndexCli), } -impl Cli { - pub fn parse() -> ParsedCli { - let mut args = match CliArgs::try_parse() { - Ok(args) => args, - Err(err) => match err.kind() { - ErrorKind::DisplayHelp | ErrorKind::DisplayVersion => { - let _ = err.print(); - process::exit(0); - } - _ => { - eprintln!("failed to parse monad-indexer args: {err:?}"); - process::exit(2); - } - }, - }; - +impl ArchiveIndexCli { + pub fn parse(mut args: ArchiveIndexCliArgs) -> ArchiveIndexParsedCli { if let Some(command) = std::mem::take(&mut args.command) { if let Err(err) = args.validate_command(&command) { let _ = err.print(); process::exit(2); } - return ParsedCli::Command { command, args }; + return ArchiveIndexParsedCli::Command { command, args }; } match args.into_cli() { - Ok(cli) => ParsedCli::Daemon(cli), + Ok(cli) => ArchiveIndexParsedCli::Daemon(cli), Err(err) => { eprintln!("failed to load monad-indexer configuration: {err:?}"); process::exit(2); @@ -76,13 +65,12 @@ impl Cli { } #[derive(Debug, Parser)] -#[command(name = "monad-indexer", about, long_about = None)] -pub struct CliArgs { +pub struct ArchiveIndexCliArgs { #[command(subcommand)] pub command: Option, /// Source to read block data that will be indexed - #[arg(long, value_parser = clap::value_parser!(BlockDataReaderArgs))] + #[arg(long, env = "BLOCK_DATA_SOURCE", value_parser = clap::value_parser!(BlockDataReaderArgs))] pub block_data_source: Option, /// If reading from --block-data-source fails, attempts to read from @@ -92,7 +80,7 @@ pub struct CliArgs { /// Where archive data is written to /// For aws: 'aws ' - #[arg(long, global = true, value_parser = clap::value_parser!(ArchiveArgs))] + #[arg(long, global = true, env = "ARCHIVE_SINK", value_parser = clap::value_parser!(ArchiveArgs))] pub archive_sink: Option, /// If set, indexer will perform an asynchronous backfill of the index @@ -115,7 +103,7 @@ pub struct CliArgs { pub stop_block: Option, /// Endpoint to push metrics to - #[arg(long)] + #[arg(long, env = "OTEL_ENDPOINT")] pub otel_endpoint: Option, #[arg(long)] @@ -135,9 +123,9 @@ pub struct CliArgs { pub enable_logs_indexing: bool, } -impl CliArgs { - pub fn into_cli(self) -> Result { - let CliArgs { +impl ArchiveIndexCliArgs { + pub fn into_cli(self) -> Result { + let ArchiveIndexCliArgs { command: _, block_data_source, fallback_block_data_source, @@ -154,7 +142,7 @@ impl CliArgs { enable_logs_indexing, } = self; - Ok(Cli { + Ok(ArchiveIndexCli { block_data_source: block_data_source .ok_or_else(|| eyre!("block_data_source must be provided"))?, fallback_block_data_source, @@ -185,7 +173,7 @@ impl CliArgs { } fn missing_archive_sink_error() -> clap::Error { - CliArgs::command().error( + ArchiveIndexCliArgs::command().error( ErrorKind::MissingRequiredArgument, "The following required argument was not provided: --archive-sink ", ) @@ -268,8 +256,8 @@ mod tests { }) } - fn make_cli_args_with_defaults() -> CliArgs { - CliArgs { + fn make_cli_args_with_defaults() -> ArchiveIndexCliArgs { + ArchiveIndexCliArgs { command: None, block_data_source: None, fallback_block_data_source: None, @@ -343,7 +331,7 @@ mod tests { #[test] fn clap_parses_set_start_block_with_global_archive_sink() { - let args = CliArgs::try_parse_from([ + let args = ArchiveIndexCliArgs::try_parse_from([ "monad-indexer", "--archive-sink", "aws sink-bucket", @@ -365,7 +353,7 @@ mod tests { #[test] fn clap_parses_set_start_block_with_archive_sink_after_subcommand() { - let args = CliArgs::try_parse_from([ + let args = ArchiveIndexCliArgs::try_parse_from([ "monad-indexer", "set-start-block", "--block", @@ -387,7 +375,7 @@ mod tests { #[test] fn clap_parses_set_start_block_with_async_backfill() { - let args = CliArgs::try_parse_from([ + let args = ArchiveIndexCliArgs::try_parse_from([ "monad-indexer", "--archive-sink", "aws sink-bucket", @@ -409,7 +397,7 @@ mod tests { #[test] fn clap_parses_daemon_mode_with_all_flags() { - let args = CliArgs::try_parse_from([ + let args = ArchiveIndexCliArgs::try_parse_from([ "monad-indexer", "--block-data-source", "aws source-bucket", @@ -431,7 +419,12 @@ mod tests { #[test] fn validate_set_start_block_requires_archive_sink() { - let args = CliArgs::try_parse_from(["monad-indexer", "set-start-block", "--block", "100"]); + let args = ArchiveIndexCliArgs::try_parse_from([ + "monad-indexer", + "set-start-block", + "--block", + "100", + ]); let args = args.expect("clap should parse, validation happens later"); let command = args.command.as_ref().expect("command should be set"); @@ -443,7 +436,7 @@ mod tests { #[test] fn validate_migrate_capped_without_block_data_source() { - let args = CliArgs::try_parse_from([ + let args = ArchiveIndexCliArgs::try_parse_from([ "monad-indexer", "migrate-capped", "--db-name", @@ -463,7 +456,7 @@ mod tests { #[test] fn validate_migrate_capped_requires_archive_sink() { - let args = CliArgs::try_parse_from([ + let args = ArchiveIndexCliArgs::try_parse_from([ "monad-indexer", "migrate-capped", "--db-name", diff --git a/monad-archive/src/bin/monad-indexer/index_worker.rs b/monad-archive/src/bin/index/index_worker.rs similarity index 100% rename from monad-archive/src/bin/monad-indexer/index_worker.rs rename to monad-archive/src/bin/index/index_worker.rs diff --git a/monad-archive/src/bin/monad-indexer/migrate_capped.rs b/monad-archive/src/bin/index/migrate_capped.rs similarity index 100% rename from monad-archive/src/bin/monad-indexer/migrate_capped.rs rename to monad-archive/src/bin/index/migrate_capped.rs diff --git a/monad-archive/src/bin/monad-indexer/migrate_logs.rs b/monad-archive/src/bin/index/migrate_logs.rs similarity index 98% rename from monad-archive/src/bin/monad-indexer/migrate_logs.rs rename to monad-archive/src/bin/index/migrate_logs.rs index e95b9c4..4e5a15c 100644 --- a/monad-archive/src/bin/monad-indexer/migrate_logs.rs +++ b/monad-archive/src/bin/index/migrate_logs.rs @@ -17,7 +17,7 @@ use futures::TryStreamExt; use monad_archive::{model::logs_index::LogsIndexArchiver, prelude::*}; pub async fn run_migrate_logs( - args: crate::cli::Cli, + args: crate::index::cli::ArchiveIndexCli, start_block: u64, stop_block_override: Option, ) -> Result<()> { diff --git a/monad-archive/src/bin/monad-indexer/main.rs b/monad-archive/src/bin/index/mod.rs similarity index 93% rename from monad-archive/src/bin/monad-indexer/main.rs rename to monad-archive/src/bin/index/mod.rs index f10b2f3..db20870 100644 --- a/monad-archive/src/bin/monad-indexer/main.rs +++ b/monad-archive/src/bin/index/mod.rs @@ -23,18 +23,17 @@ mod index_worker; use index_worker::index_worker; use tracing::{info, Level}; -use crate::{migrate_capped::migrate_to_uncapped, migrate_logs::run_migrate_logs}; +use crate::index::{migrate_capped::migrate_to_uncapped, migrate_logs::run_migrate_logs}; -mod cli; +pub mod cli; mod migrate_capped; mod migrate_logs; -#[tokio::main] -async fn main() -> Result<()> { +pub async fn run(args: cli::ArchiveIndexCliArgs) -> Result<()> { tracing_subscriber::fmt().with_max_level(Level::INFO).init(); - match cli::Cli::parse() { - cli::ParsedCli::Command { command, args } => match command { + match cli::ArchiveIndexCli::parse(args) { + cli::ArchiveIndexParsedCli::Command { command, args } => match command { cli::Commands::MigrateLogs { start_block, stop_block, @@ -71,14 +70,14 @@ async fn main() -> Result<()> { run_set_start_block(block, archive_sink, async_backfill).await } }, - cli::ParsedCli::Daemon(args) => { + cli::ArchiveIndexParsedCli::Daemon(args) => { info!(?args, "Cli Arguments: "); run_indexer(args).await } } } -async fn run_indexer(args: cli::Cli) -> Result<()> { +async fn run_indexer(args: cli::ArchiveIndexCli) -> Result<()> { let metrics = Metrics::new( args.otel_endpoint, "monad-indexer", diff --git a/monad-archive/src/bin/main.rs b/monad-archive/src/bin/main.rs new file mode 100644 index 0000000..0231982 --- /dev/null +++ b/monad-archive/src/bin/main.rs @@ -0,0 +1,35 @@ +// Copyright (C) 2025 Category Labs, Inc. +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU General Public License for more details. +// +// You should have received a copy of the GNU General Public License +// along with this program. If not, see . + +use clap::Parser; +use eyre::Result; + +mod archive; +mod block_writer; +mod check; +mod cli; +mod index; + +#[tokio::main(flavor = "multi_thread")] +async fn main() -> Result<()> { + // Each subcommand installs its own tracing subscriber; the checker needs an + // env filter rather than a fixed level. + match cli::ArchiveCli::parse().command { + cli::ArchiveCliCommand::Archive(args) => archive::run(args).await, + cli::ArchiveCliCommand::Index(args) => index::run(args).await, + cli::ArchiveCliCommand::Check(args) => check::run(args).await, + cli::ArchiveCliCommand::BlockWriter(args) => block_writer::run(args).await, + } +} diff --git a/scripts/build-deb b/scripts/build-deb index b4d3573..1f1e205 100755 --- a/scripts/build-deb +++ b/scripts/build-deb @@ -30,9 +30,7 @@ cd "$root_dir" lib_dir="$stage_dir/usr/lib/monad-services" mkdir -p "$stage_dir/DEBIAN" "$lib_dir" -for binary in monad-archiver monad-archive-checker monad-indexer monad-block-writer; do - install -m 0755 "target/release/$binary" "$lib_dir/$binary" -done +install -m 0755 target/release/monad-archive "$lib_dir/monad-archive" # --- stage libraries --- # Bundle libtriedb_driver.so, which is loaded with dlopen at runtime and so