diff --git a/Cargo.lock b/Cargo.lock index 1be5726..7928d95 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2045,6 +2045,7 @@ dependencies = [ "shadow-rs", "shutdown", "strum", + "tempfile", "testcontainers", "thiserror 2.0.18", "tokio", diff --git a/Cargo.toml b/Cargo.toml index 1764422..5132246 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -48,3 +48,4 @@ shadow-rs = "1.4" [dev-dependencies] testcontainers = "0.27" rand = "0.10" +tempfile = "3" diff --git a/README.adoc b/README.adoc index 5f0061c..6dcea50 100644 --- a/README.adoc +++ b/README.adoc @@ -47,61 +47,103 @@ Arguments: Options: -b, --blockchain Blockchain + --dry-run Do not modify the storage + -c, --connection Connection (host:port) + --connection.notls Disable TLS + --parallel How many API requests to make in parallel. Range: 1..512. Default: 16 + --notify.dir Write notifications as JSON line to the specified dir in a file + --notify.pulsar.topic Send notifications as JSON to the Pulsar to the specified topic (notify.pulsar.url must be specified) + --notify.pulsar.url Send notifications as JSON to the Pulsar with specified URL (notify.pulsar.topic must be specified) + --auth.aws.access-key AWS / S3 Access Key + --auth.aws.secret-key AWS / S3 Secret Key + --aws.endpoint AWS / S3 endpoint url instead of the default one + --aws.region AWS / S3 region ID to use for requests + --aws.s3.path-style Enable S3 Path Style access (default is false). Use this flag for a no-AWS service + --aws.trust-tls Trust any TLS certificate for AWS / S3 (default is false) + -d, --dir Target directory + --continue [Stream Command] Continue from the last archived block; i.e., not the latest in blockchain + --tail [Verify/Fix Commands] Use the latest T blocks instead of a range + -r, --range Blocks Range (`N...M`) + --range.chunk Range chunk size (default 1000) + -t, --tables Types of tables to archive (comma-separated list of `blocks`, `txes`, `traces`). Default: blocks,txes + --fields.trace List of data to include into tracing archive table (comma-separated list of `calls`, `stateDiff`). Default: calls,stateDiff; Used only if `traces` are included into the archived tables (see `--tables` option); Details: `calls` - debug_traceTransaction with `callTracer` tracing; `stateDiff` - debug_traceTransaction with `prestateTracer` tracing + --fix.clean [Fix Command] Set to remove any existing data in whole chunk if any of tables is missing a block in the chunk or has broken values. Default is `false`, which deleted only tables with missing / corrupted data + --compression - Compression algorithm to use when writing new Avro files. Default is `zstd` [possible values: snappy, zstd] + Compression algorithm to use when writing new Avro files. Default is `zstd` + + [possible values: snappy, zstd] + --follow - [Stream Command] Follow mode for new blocks: `latest` - follow the latest blocks (default); `finalized` - follow only finalized blocks [default: latest] [possible values: latest, finalized] + [Stream Command] Follow mode for new blocks: `latest` - follow the latest blocks (default); `finalized` - follow only finalized blocks + + [default: latest] + [possible values: latest, finalized] + + --format + Output format. `avro` (default) writes one row-batched Avro file per (kind, range) under the historical layout. `json` writes one JSON file per field (block.json, tx-.json, receipt-.json, raw-.hex, …) inside a directory per height. `compact` and `verify` are not supported with `json` + + Possible values: + - avro: One Avro file per (kind, range). Historical layout — supports all commands + - json: One JSON file per field under a per-height directory. `archive`, `stream`, and `fix` are supported; `compact` and `verify` are rejected at startup + + [default: avro] + --metrics Start a Prometheus-compatible metrics server on the given address (e.g., 127.0.0.1:8080). Metrics are served at http://HOST:PORT/metrics + --metrics.await After the main command finishes, keep the metrics server running until one final scrape completes (or 60 seconds elapse). Useful for short-lived commands (fix, verify, compact) to ensure Prometheus collects the final metrics before the process exits + -h, --help - Print help + Print help (see a summary with '-h') + -V, --version Print version + ---- === Commands diff --git a/src/archiver/filenames.rs b/src/archiver/filenames.rs index a50c71e..fc88152 100644 --- a/src/archiver/filenames.rs +++ b/src/archiver/filenames.rs @@ -168,6 +168,19 @@ impl Filenames { self.full_path(self.relative_path(kind, range)) } + /// Directory that holds the per-field files for a single block height. + /// Used by the JSON layout (one directory per height, each containing + /// `block.json`, `tx-.json`, etc.) — reuses the same two-level + /// height bucketing as [`Self::path`]. + pub fn height_dir(&self, height: u64) -> String { + self.full_path(format!( + "{}/{}/{}", + self.level_1(height), + self.level_2(height), + self.range_padded(height) + )) + } + fn level_1(&self, value: u64) -> String { let number = value / self.dir_block_size_l1 * self.dir_block_size_l1; self.range_padded(number) diff --git a/src/args.rs b/src/args.rs index 4e59ec2..e92d2d2 100644 --- a/src/args.rs +++ b/src/args.rs @@ -108,6 +108,14 @@ pub struct Args { #[arg(long = "follow", default_value = "latest")] pub follow: Follow, + /// Output format. `avro` (default) writes one row-batched Avro file per + /// (kind, range) under the historical layout. `json` writes one JSON file + /// per field (block.json, tx-.json, receipt-.json, + /// raw-.hex, …) inside a directory per height. `compact` and `verify` + /// are not supported with `json`. + #[arg(long = "format", default_value = "avro")] + pub format: Format, + /// Start a Prometheus-compatible metrics server on the given address (e.g., 127.0.0.1:8080). /// Metrics are served at http://HOST:PORT/metrics #[arg(long = "metrics", value_name = "HOST:PORT")] @@ -139,6 +147,7 @@ impl Default for Args { fix_clean: false, compression: None, follow: Follow::Latest, + format: Format::Avro, metrics: None, metrics_await: false, } @@ -269,6 +278,16 @@ pub enum Compression { Zstd, } +/// Output format selected via `--format`. +#[derive(clap::ValueEnum, Debug, Clone, Copy, PartialEq, Eq)] +pub enum Format { + /// One Avro file per (kind, range). Historical layout — supports all commands. + Avro, + /// One JSON file per field under a per-height directory. `archive`, `stream`, + /// and `fix` are supported; `compact` and `verify` are rejected at startup. + Json, +} + #[derive(clap::ValueEnum, Debug, Clone, PartialEq)] pub enum Follow { Latest, diff --git a/src/blockchain/ethereum.rs b/src/blockchain/ethereum.rs index b604765..8ff5c6f 100644 --- a/src/blockchain/ethereum.rs +++ b/src/blockchain/ethereum.rs @@ -94,6 +94,8 @@ impl EthereumData { if data_as_json == b"null" { return Err(anyhow!("Transaction not found: 0x{:x}", hash)); } + // Wire format is a JSON string like "0xabcdef…"; strip the surrounding + // quotes and the `0x` prefix, then hex-decode to bytes. let data_as_hex = String::from_utf8( data_as_json[3..(data_as_json.len() - 1)].to_vec() ).map_err(|_| anyhow!("Invalid hex"))?; @@ -136,7 +138,7 @@ impl EthereumData { .take(10); Retry::spawn(retry_strategy, async || { self.get_tx_raw(hash).await - .and_then(|value| if value == b"null" { + .and_then(|value| if value.is_empty() { Err(anyhow!("Transaction Raw not found: 0x{:x}", hash)) } else { Ok(value) diff --git a/src/command/fix.rs b/src/command/fix.rs index 4ede4a9..ec73cbb 100644 --- a/src/command/fix.rs +++ b/src/command/fix.rs @@ -1,6 +1,6 @@ use std::marker::PhantomData; use async_trait::async_trait; -use crate::{archiver::{ArchiveAll, Archiver}, args::Args, blockchain::BlockchainTypes, command::CommandExecutor, global, notify::RunMode, storage::ReadTarget}; +use crate::{archiver::{ArchiveAll, Archiver}, args::Args, blockchain::BlockchainTypes, command::CommandExecutor, global, notify::RunMode, storage::ScanTarget}; use crate::archiver::blocks_config::Blocks; use crate::archiver::datakind::DataOptions; @@ -9,7 +9,7 @@ use crate::archiver::datakind::DataOptions; /// It checks the archive for the specified range and add missing data /// #[derive(Clone)] -pub struct FixCommand { +pub struct FixCommand { b: PhantomData, blocks: Blocks, chunk_size: usize, @@ -17,7 +17,7 @@ pub struct FixCommand { tx_options: DataOptions, } -impl FixCommand { +impl FixCommand { pub fn new(config: &Args, archiver: Archiver) -> anyhow::Result { @@ -34,7 +34,7 @@ impl FixCommand { } #[async_trait] -impl CommandExecutor for FixCommand { +impl CommandExecutor for FixCommand { async fn execute(&self) -> anyhow::Result<()> { let shutdown = global::get_shutdown(); diff --git a/src/command/stream.rs b/src/command/stream.rs index d2eaecc..b1ec102 100644 --- a/src/command/stream.rs +++ b/src/command/stream.rs @@ -11,7 +11,7 @@ use crate::{ command::CommandExecutor, global, notify::RunMode, - storage::ReadTarget + storage::ScanTarget }; use anyhow::Result; use crate::archiver::datakind::DataOptions; @@ -25,7 +25,7 @@ use crate::notify::Maturity; /// It appends fresh blocks one by one to the archive /// #[derive(Clone)] -pub struct StreamCommand { +pub struct StreamCommand { b: PhantomData, blockchain: Arc, continue_blocks: Option, @@ -34,7 +34,7 @@ pub struct StreamCommand { follow: Follow, } -impl StreamCommand { +impl StreamCommand { pub async fn new(config: &Args, archiver: Archiver ) -> Result { @@ -87,7 +87,7 @@ impl StreamCommand { } #[async_trait] -impl CommandExecutor for StreamCommand { +impl CommandExecutor for StreamCommand { async fn execute(&self) -> Result<()> { diff --git a/src/formats/json.rs b/src/formats/json.rs new file mode 100644 index 0000000..a36a699 --- /dev/null +++ b/src/formats/json.rs @@ -0,0 +1,292 @@ +// Copyright 2026 EmeraldPay Ltd +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. + +//! JSON-per-field file format. +//! +//! Each [`Field`] in an [`ArchiveRow`] becomes its own file. The file extension +//! is `.json` when the payload is the original JSON response from the node, and +//! `.hex` for the raw-transaction field (which is a hex string, not JSON). +//! +//! Filenames are derived from the field name (plus, for per-tx fields, the +//! txid): +//! +//! | Field | Filename | +//! |--------------|---------------------------| +//! | BlockJson | `block.json` | +//! | Uncle{N} | `uncle-N.json` | +//! | TxJson | `tx-.json` | +//! | TxRaw | `raw-.hex` | +//! | Receipt | `receipt-.json` | +//! | Trace | `trace-.json` | +//! | StateDiff | `statediff-.json` | +//! +//! `From` and `To` are convenience fields surfaced as separate Avro columns; they +//! are already part of the transaction JSON, so the JSON file layout skips them. +//! +//! Payloads are written **byte-for-byte** as the node returned them — no +//! re-encoding, no whitespace normalization. + +use std::collections::HashMap; + +use crate::archiver::datakind::{DataKind, DataOptions}; +use crate::archiver::range::Range; +use crate::archiver::range_bag::RangeBag; +use crate::record::{ArchiveRow, BlockchainType, Field}; + +/// A single file produced by [`encode_row`]. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct JsonFieldFile { + /// Bare filename (no directory component). + pub filename: String, + /// Exact bytes to write to the file. + pub payload: Vec, +} + +/// Convert an [`ArchiveRow`] into the set of per-field files it produces under +/// the JSON layout. The caller decides where (which directory) to write them. +pub fn encode_row(row: &ArchiveRow) -> Vec { + let tx_id = row.tx_id.as_deref(); + let blockchain_type = row.blockchain_type; + row.fields + .iter() + .filter_map(|f| encode_field(f, tx_id, blockchain_type)) + .collect() +} + +fn encode_field( + field: &Field, + tx_id: Option<&str>, + blockchain_type: BlockchainType, +) -> Option { + match field { + Field::BlockJson(bytes) => Some(JsonFieldFile { + filename: "block.json".to_string(), + payload: bytes.clone(), + }), + Field::Uncle { index, json } => Some(JsonFieldFile { + filename: format!("uncle-{}.json", index), + payload: json.clone(), + }), + Field::TxJson(bytes) => tx_id.map(|id| JsonFieldFile { + filename: format!("tx-{}.json", id), + payload: bytes.clone(), + }), + // Raw transaction is stored decoded to save memory; re-encode to the + // node's wire format here (Ethereum prefixes with `0x`, Bitcoin does not). + Field::TxRaw(bytes) => tx_id.map(|id| JsonFieldFile { + filename: format!("raw-{}.hex", id), + payload: encode_tx_raw(bytes, blockchain_type), + }), + Field::Receipt(bytes) => tx_id.map(|id| JsonFieldFile { + filename: format!("receipt-{}.json", id), + payload: bytes.clone(), + }), + // Convenience-only fields that are already present inside the parent JSON. + Field::From(_) | Field::To(_) => None, + Field::Trace(bytes) => tx_id.map(|id| JsonFieldFile { + filename: format!("trace-{}.json", id), + payload: bytes.clone(), + }), + Field::StateDiff(bytes) => tx_id.map(|id| JsonFieldFile { + filename: format!("statediff-{}.json", id), + payload: bytes.clone(), + }), + } +} + +fn encode_tx_raw(bytes: &[u8], blockchain_type: BlockchainType) -> Vec { + let hex_str = hex::encode(bytes); + match blockchain_type { + BlockchainType::Ethereum => format!("0x{}", hex_str).into_bytes(), + BlockchainType::Bitcoin => hex_str.into_bytes(), + } +} + +/// Per-height completeness check used by the JSON storage backends. +/// +/// Given the set of filenames present in a single height's directory, return +/// the [`DataKind`]s that are missing or incomplete given the requested +/// [`DataOptions`]. This is the heuristic the Fix command relies on: +/// +/// - **Blocks** is missing when `block.json` is absent. +/// - **Transactions** is missing when no `tx-*.json` or no `raw-*.hex` is +/// present. (Partial-tx coverage is not detected — Fix re-archives the +/// height and the writer skips files that already exist.) +/// - **TransactionTraces** is missing when the requested trace kinds +/// (`trace-*.json` and/or `statediff-*.json` depending on +/// [`DataOptions::trace`]) are absent. +pub fn missing_kinds_at_height(entries: I, options: &DataOptions) -> Vec +where + I: IntoIterator, + S: AsRef, +{ + let entries: Vec = entries + .into_iter() + .map(|s| s.as_ref().to_string()) + .collect(); + let mut missing = Vec::new(); + + if options.include_block() && !entries.iter().any(|n| n == "block.json") { + missing.push(DataKind::Blocks); + } + if options.include_tx() { + let has_any_tx = entries + .iter() + .any(|n| n.starts_with("tx-") && n.ends_with(".json")); + let has_any_raw = entries + .iter() + .any(|n| n.starts_with("raw-") && n.ends_with(".hex")); + if !has_any_tx || !has_any_raw { + missing.push(DataKind::Transactions); + } + } + if options.include_trace() { + // include_trace() guarantees `trace` is Some. + let trace_opts = options.trace.as_ref().unwrap(); + let need_trace = trace_opts.include_trace; + let need_state = trace_opts.include_state_diff; + let has_any_trace = entries + .iter() + .any(|n| n.starts_with("trace-") && n.ends_with(".json")); + let has_any_state = entries + .iter() + .any(|n| n.starts_with("statediff-") && n.ends_with(".json")); + if (need_trace && !has_any_trace) || (need_state && !has_any_state) { + missing.push(DataKind::TransactionTraces); + } + } + + missing +} + +/// Collapse a per-height missing-kinds map into the shape that +/// [`crate::storage::ScanTarget::find_incomplete_tables`] returns: one +/// `(Range, Vec)` entry per contiguous run of heights that share the +/// same missing-kinds set. +pub fn collapse_missing(per_height: HashMap>) -> Vec<(Range, Vec)> { + let mut by_kinds: HashMap, RangeBag> = HashMap::new(); + for (height, kinds) in per_height { + by_kinds + .entry(kinds) + .or_insert_with(RangeBag::new) + .append(Range::Single(height.into())); + } + let mut result = Vec::new(); + for (kinds, bag) in by_kinds { + for range in bag.compact().ranges { + result.push((range, kinds.clone())); + } + } + result +} + +#[cfg(test)] +mod tests { + use super::*; + use chrono::{TimeZone, Utc}; + + use crate::archiver::datakind::DataKind; + use crate::record::BlockchainType; + + fn row(kind: DataKind, tx_id: Option<&str>, fields: Vec) -> ArchiveRow { + ArchiveRow { + kind, + blockchain_type: BlockchainType::Ethereum, + blockchain_id: "ETH".to_string(), + archive_ts: Utc::now(), + height: 100, + block_id: "0xblock".to_string(), + timestamp: Utc.timestamp_millis_opt(0).unwrap(), + parent_id: Some("0xparent".to_string()), + tx_index: tx_id.map(|_| 0), + tx_id: tx_id.map(|s| s.to_string()), + fields, + } + } + + #[test] + fn block_row_emits_block_and_uncle_files() { + let r = row( + DataKind::Blocks, + None, + vec![ + Field::BlockJson(b"BLOCK".to_vec()), + Field::Uncle { index: 0, json: b"U0".to_vec() }, + Field::Uncle { index: 1, json: b"U1".to_vec() }, + ], + ); + let files = encode_row(&r); + assert_eq!(files.len(), 3); + assert_eq!(files[0].filename, "block.json"); + assert_eq!(files[0].payload, b"BLOCK"); + assert_eq!(files[1].filename, "uncle-0.json"); + assert_eq!(files[2].filename, "uncle-1.json"); + } + + #[test] + fn tx_row_emits_per_field_files_keyed_by_txid() { + let r = row( + DataKind::Transactions, + Some("0xabc"), + vec![ + Field::TxJson(b"TX".to_vec()), + Field::TxRaw(vec![0xde, 0xad, 0xbe, 0xef]), + Field::Receipt(b"R".to_vec()), + Field::From("0xfrom".to_string()), + Field::To("0xto".to_string()), + ], + ); + let files = encode_row(&r); + // From/To are intentionally skipped. + assert_eq!(files.len(), 3); + let names: Vec<_> = files.iter().map(|f| f.filename.as_str()).collect(); + assert!(names.contains(&"tx-0xabc.json")); + assert!(names.contains(&"raw-0xabc.hex")); + assert!(names.contains(&"receipt-0xabc.json")); + let raw = files.iter().find(|f| f.filename == "raw-0xabc.hex").unwrap(); + // Ethereum row → `0x` prefix added on the way out. + assert_eq!(raw.payload, b"0xdeadbeef"); + } + + #[test] + fn bitcoin_raw_tx_has_no_0x_prefix() { + let mut r = row( + DataKind::Transactions, + Some("abc"), + vec![Field::TxRaw(vec![0xde, 0xad, 0xbe, 0xef])], + ); + r.blockchain_type = BlockchainType::Bitcoin; + let files = encode_row(&r); + let raw = files.iter().find(|f| f.filename == "raw-abc.hex").unwrap(); + assert_eq!(raw.payload, b"deadbeef"); + } + + #[test] + fn trace_row_emits_trace_and_statediff_files() { + let r = row( + DataKind::TransactionTraces, + Some("0xabc"), + vec![ + Field::Trace(b"T".to_vec()), + Field::StateDiff(b"S".to_vec()), + ], + ); + let files = encode_row(&r); + assert_eq!(files.len(), 2); + let names: Vec<_> = files.iter().map(|f| f.filename.as_str()).collect(); + assert!(names.contains(&"trace-0xabc.json")); + assert!(names.contains(&"statediff-0xabc.json")); + } + + #[test] + fn per_tx_fields_without_tx_id_are_skipped() { + let r = row( + DataKind::Transactions, + None, + vec![Field::TxJson(b"X".to_vec())], + ); + assert!(encode_row(&r).is_empty()); + } +} diff --git a/src/formats/mod.rs b/src/formats/mod.rs index 15c8438..ff29d6c 100644 --- a/src/formats/mod.rs +++ b/src/formats/mod.rs @@ -10,3 +10,4 @@ //! future modules will add JSON-per-field files and Pulsar/Kafka stream encoding. pub mod avro; +pub mod json; diff --git a/src/main.rs b/src/main.rs index 5de48d8..1b83d18 100644 --- a/src/main.rs +++ b/src/main.rs @@ -28,11 +28,12 @@ use crate::{ }, args::{ Command, - Args + Args, + Format, }, blockchain::{BitcoinType, BlockchainTypes, EthereumType}, notify::Notifier, - storage::ReadTarget, + storage::{ReadTarget, ScanTarget, WriteTarget}, archiver::Archiver, }; @@ -94,6 +95,18 @@ async fn main_inner() -> Result<()> { tracing::info!("Dry run mode enabled, no changes will be made"); } + if args.format == Format::Json { + match args.command { + Command::Compact | Command::Verify => { + return Err(anyhow!( + "{:?} is not supported with --format json (the per-height JSON layout has no ranges to compact or verify)", + args.command + )); + } + _ => {} + } + } + let chain_ref = ChainRef::from_str(&args.blockchain) .map_err(|_| anyhow!("Unsupported blockchain: {}", args.blockchain))?; let chain_type = BlockchainType::try_from(chain_ref) @@ -114,48 +127,74 @@ async fn main_inner() -> Result<()> { async fn run(builder: Builder, args: &Args) -> Result<()> { if storage::is_fs(&args) { - let target = storage::create_fs(&args)?; - run_with_target(builder, target, args).await + match args.format { + Format::Avro => run_with_read_target(builder, storage::create_fs(&args)?, args).await, + Format::Json => run_with_scan_target(builder, storage::create_fs_json(&args)?, args).await, + } } else if storage::is_s3(&args) { - let target = storage::create_aws(&args)?; - run_with_target(builder, target, args).await + match args.format { + Format::Avro => run_with_read_target(builder, storage::create_aws(&args)?, args).await, + Format::Json => run_with_scan_target(builder, storage::create_aws_json(&args)?, args).await, + } } else { return Err(anyhow!("Unsupported storage")); } } -async fn run_with_target(builder: Builder, target: TS, args: &Args) -> Result<()> { +/// +/// Dispatch path for targets that support full read access (Avro on FS/S3 today). +/// All five commands are valid. +async fn run_with_read_target( + builder: Builder, + target: TS, + args: &Args, +) -> Result<()> { + let builder = build_with_target(builder, target, args).await?; + match args.command { + Command::Stream => builder.stream(args).await.execute().await, + Command::Fix => builder.fix(args).execute().await, + Command::Archive => builder.archive(args).execute().await, + Command::Verify => builder.verify(args).execute().await, + Command::Compact => builder.compact(args).execute().await, + } +} + +/// +/// Dispatch path for scan-only targets (JSON-per-field today; streaming targets +/// in Phase 3 will use a separate write-only path). `verify`/`compact` are +/// rejected upfront in [`main_inner`], so the runtime path here is unreachable +/// for those commands; we still error defensively in case the upfront check is +/// ever loosened. +async fn run_with_scan_target( + builder: Builder, + target: TS, + args: &Args, +) -> Result<()> { + let builder = build_with_target(builder, target, args).await?; + match args.command { + Command::Stream => builder.stream(args).await.execute().await, + Command::Fix => builder.fix(args).execute().await, + Command::Archive => builder.archive(args).execute().await, + Command::Verify | Command::Compact => Err(anyhow!( + "{:?} requires a read-capable target", + args.command + )), + } +} + +async fn build_with_target( + builder: Builder, + target: TS, + args: &Args, +) -> Result> { let chain_ref = ChainRef::from_str(&args.blockchain) .map_err(|_| anyhow!("Unsupported blockchain: {}", args.blockchain))?; let blockchain = Blockchain::new(&args.connection, args.as_dshackle_blockchain()?, chain_ref.code()).await?; - let builder = builder + Ok(builder .with_notifier(notify::create_notifier(&args).await?) .with_target(target) - .with_data(blockchain, chain_ref.code()); - - match args.command { - Command::Stream => { - builder.stream(args).await - .execute().await - }, - Command::Fix => { - builder.fix(args) - .execute().await - }, - Command::Verify => { - builder.verify(args) - .execute().await - }, - Command::Archive => { - builder.archive(args) - .execute().await - }, - Command::Compact => { - builder.compact(args) - .execute().await - }, - } + .with_data(blockchain, chain_ref.code())) } struct Builder { @@ -163,12 +202,12 @@ struct Builder { notifier: Option>, } -struct BuilderWithTarget { +struct BuilderWithTarget { parent: Builder, target: TS, } -struct BuilderWithData { +struct BuilderWithData { parent: BuilderWithTarget, data: B::DataProvider, } @@ -188,7 +227,7 @@ impl Builder where B: BlockchainTypes { } } - fn with_target(self, target: TS) -> BuilderWithTarget where TS: ReadTarget { + fn with_target(self, target: TS) -> BuilderWithTarget where TS: WriteTarget { BuilderWithTarget { target, parent: self, @@ -196,7 +235,7 @@ impl Builder where B: BlockchainTypes { } } -impl BuilderWithTarget where B: BlockchainTypes, TS: ReadTarget { +impl BuilderWithTarget where B: BlockchainTypes, TS: WriteTarget { fn with_data(self, blockchain: Blockchain, id: String) -> BuilderWithData { BuilderWithData { parent: self, @@ -205,9 +244,14 @@ impl BuilderWithTarget where B: BlockchainTypes, TS: ReadTarget { } } -impl BuilderWithData where B: BlockchainTypes + 'static, TS: ReadTarget + 'static { +impl BuilderWithData where B: BlockchainTypes + 'static, TS: WriteTarget + 'static { - async fn stream(self, args: &Args) -> StreamCommand { + /// `stream` reads existing data to honour `--continue`, so it requires + /// [`ScanTarget`]. + async fn stream(self, args: &Args) -> StreamCommand + where + TS: ScanTarget, + { let notifier = self.parent.parent.notifier.unwrap(); let notifications = notifier.start(); let archiver = Archiver::new( @@ -217,7 +261,11 @@ impl BuilderWithData where B: BlockchainTypes + 'static, TS: ReadT command } - fn fix(self, args: &Args) -> FixCommand { + /// `fix` enumerates missing data via [`ScanTarget::find_incomplete_tables`]. + fn fix(self, args: &Args) -> FixCommand + where + TS: ScanTarget, + { let notifier = self.parent.parent.notifier.unwrap(); let notifications = notifier.start(); let archiver = Archiver::new( @@ -227,7 +275,11 @@ impl BuilderWithData where B: BlockchainTypes + 'static, TS: ReadT command } - fn verify(self, args: &Args) -> VerifyCommand { + /// `verify` opens existing files to inspect them — requires [`ReadTarget`]. + fn verify(self, args: &Args) -> VerifyCommand + where + TS: ReadTarget, + { let notifier = self.parent.parent.notifier.unwrap(); let notifications = notifier.start(); let archiver = Archiver::new( @@ -237,6 +289,7 @@ impl BuilderWithData where B: BlockchainTypes + 'static, TS: ReadT command } + /// `archive` only writes; [`WriteTarget`] is enough. fn archive(self, args: &Args) -> ArchiveCommand { let notifier = self.parent.parent.notifier.unwrap(); let notifications = notifier.start(); @@ -247,7 +300,11 @@ impl BuilderWithData where B: BlockchainTypes + 'static, TS: ReadT command } - fn compact(self, args: &Args) -> CompactCommand { + /// `compact` reads existing range files and rewrites them — requires [`ReadTarget`]. + fn compact(self, args: &Args) -> CompactCommand + where + TS: ReadTarget, + { let notifier = self.parent.parent.notifier.unwrap(); let notifications = notifier.start(); let archiver = Archiver::new( diff --git a/src/record.rs b/src/record.rs index f801422..63b0096 100644 --- a/src/record.rs +++ b/src/record.rs @@ -95,7 +95,14 @@ pub enum Field { // ---- Transaction-kind fields ---- /// JSON of the transaction as returned by the node. TxJson(Vec), - /// Raw bytes of the transaction (binary, not JSON). + /// Raw transaction bytes (binary, not JSON). + /// + /// Stored decoded to save memory — the wire format is a hex string, but its + /// presence/absence of an `0x` prefix is fully determined by the blockchain + /// (Ethereum uses `0x`, Bitcoin does not), so format adapters can reconstruct + /// the wire form from the row's [`crate::record::BlockchainType`] when + /// writing it back out (e.g., the JSON layout writes `raw-.hex` as a + /// hex string with the appropriate prefix). TxRaw(Vec), /// Ethereum-only: the transaction receipt JSON. Receipt(Vec), diff --git a/src/storage/fs.rs b/src/storage/fs.rs index de5caa4..75daf82 100644 --- a/src/storage/fs.rs +++ b/src/storage/fs.rs @@ -5,12 +5,15 @@ use std::sync::{Mutex}; use apache_avro::types::Record; use apache_avro::{Writer}; use async_trait::async_trait; -use crate::archiver::datakind::DataKind; +use crate::archiver::datakind::{DataKind, DataOptions}; use crate::archiver::filenames::{Filenames, Level, LevelDouble}; use crate::archiver::range::Range; use crate::formats::avro; use crate::record::ArchiveRow; -use crate::storage::{avro_reader, copy, FileReference, ReadTarget, TargetFile, TargetFileReader, TargetFileWriter, WriteTarget}; +use crate::storage::{ + avro_reader, copy, find_incomplete_by_listing, FileReference, ReadTarget, ScanTarget, + TargetFile, TargetFileReader, TargetFileWriter, WriteTarget, +}; use anyhow::{anyhow, Context, Result}; use tokio::sync::mpsc::Receiver; use crate::global; @@ -40,6 +43,17 @@ impl WriteTarget for FsStorage { } } +#[async_trait] +impl ScanTarget for FsStorage { + async fn find_incomplete_tables( + &self, + blocks: Range, + tx_options: &DataOptions, + ) -> Result)>> { + find_incomplete_by_listing(self, blocks, tx_options).await + } +} + #[async_trait] impl ReadTarget for FsStorage { diff --git a/src/storage/json_fs.rs b/src/storage/json_fs.rs new file mode 100644 index 0000000..c2da34f --- /dev/null +++ b/src/storage/json_fs.rs @@ -0,0 +1,434 @@ +// Copyright 2026 EmeraldPay Ltd +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. + +//! Filesystem storage that writes one JSON file per [`crate::record::Field`]. +//! +//! Layout matches the existing two-level height bucketing +//! (`////`). Each archive row produces a set of +//! files inside its height's directory; nothing is grouped into ranges, which is +//! why `compact` and `verify` are not supported for this layout. + +use std::collections::HashMap; +use std::fs; +use std::path::PathBuf; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Mutex; + +use anyhow::{anyhow, Result}; +use async_trait::async_trait; + +use crate::archiver::datakind::{DataKind, DataOptions}; +use crate::archiver::filenames::Filenames; +use crate::archiver::range::Range; +use crate::formats::json; +use crate::record::ArchiveRow; +use crate::storage::{ScanTarget, TargetFile, TargetFileWriter, WriteTarget}; + +/// Filesystem-backed JSON-per-field target. +/// +/// Reuses [`Filenames`] only for its level-1/level-2 height bucketing +/// (`height_dir`); the file extension on [`Filenames`] is unused. +pub struct JsonFsStorage { + parent_dir: PathBuf, + filenames: Filenames, +} + +impl JsonFsStorage { + pub fn new(dir: PathBuf, filenames: Filenames) -> Self { + Self { parent_dir: dir, filenames } + } + + fn height_dir(&self, height: u64) -> PathBuf { + self.parent_dir.join(self.filenames.height_dir(height)) + } + + fn read_dir_entries(&self, height: u64) -> Result> { + let dir = self.height_dir(height); + if !dir.exists() { + return Ok(Vec::new()); + } + let entries = fs::read_dir(&dir) + .map_err(|e| anyhow!("Cannot read dir {:?}: {}", dir, e))?; + let mut names = Vec::new(); + for entry in entries.flatten() { + if let Some(name) = entry.file_name().to_str() { + names.push(name.to_string()); + } + } + Ok(names) + } +} + +#[async_trait] +impl WriteTarget for JsonFsStorage { + type Writer = JsonFsWriter; + + async fn create( + &self, + kind: DataKind, + range: &Range, + overwrite: bool, + ) -> Result> { + // For per-field JSON we don't pre-create a file; the writer fans out at + // append time and decides per-file whether to overwrite. + Ok(Some(JsonFsWriter { + parent_dir: self.parent_dir.clone(), + filenames: self.filenames.clone(), + kind, + range: range.clone(), + overwrite, + written_files: Mutex::new(Vec::new()), + closed: AtomicBool::new(false), + })) + } +} + +#[async_trait] +impl ScanTarget for JsonFsStorage { + /// Per-height completeness check; the shared rules live in + /// [`json::missing_kinds_at_height`]. The simple heuristic trades exhaustive + /// per-tx coverage for not having to parse `block.json` on every Fix run; + /// heights with partial data are re-archived wholesale by Fix, which then + /// skips files that already exist (see [`JsonFsWriter::append`]). + async fn find_incomplete_tables( + &self, + blocks: Range, + tx_options: &DataOptions, + ) -> Result)>> { + let mut per_height: HashMap> = HashMap::new(); + for height in blocks.iter() { + let entries = self.read_dir_entries(height)?; + let missing = json::missing_kinds_at_height(entries, tx_options); + if !missing.is_empty() { + per_height.insert(height, missing); + } + } + Ok(json::collapse_missing(per_height)) + } +} + +/// Writer for one (kind, range) session. Each [`Self::append`] call fans the row +/// into the per-field files that live inside the row's height directory. +/// +/// Drop semantics: if [`Self::close`] isn't called (e.g., the archive run was +/// aborted), every file written through this session is removed so that +/// partial output doesn't get mistaken for a complete archive by the Fix +/// command's completeness check. +pub struct JsonFsWriter { + parent_dir: PathBuf, + filenames: Filenames, + kind: DataKind, + range: Range, + overwrite: bool, + written_files: Mutex>, + closed: AtomicBool, +} + +impl TargetFile for JsonFsWriter { + /// For a single-height session, the height directory; for a multi-height + /// session, the level-2 directory that contains all written heights. + /// Notifications carry this so consumers know where the new data landed. + fn get_url(&self) -> String { + let path = match &self.range { + Range::Single(h) => self.parent_dir.join(self.filenames.height_dir(h.height)), + Range::Multiple(start, _) => { + // Drop the trailing height segment to get the level-2 directory. + let dir = self.filenames.height_dir(start.height); + let dir = dir.rsplit_once('/').map(|(p, _)| p.to_string()).unwrap_or(dir); + self.parent_dir.join(dir) + } + }; + let canonical = path.canonicalize().unwrap_or(path); + format!("file://{}", canonical.to_str().unwrap_or("invalid")) + } +} + +#[async_trait] +impl TargetFileWriter for JsonFsWriter { + async fn append(&self, row: ArchiveRow) -> Result<()> { + let dir = self.parent_dir.join(self.filenames.height_dir(row.height)); + fs::create_dir_all(&dir) + .map_err(|e| anyhow!("Failed to create dir {:?}: {}", dir, e))?; + + let files = json::encode_row(&row); + for file in files { + let path = dir.join(&file.filename); + if !self.overwrite && path.exists() { + tracing::debug!("Skipping existing JSON file: {:?}", path); + continue; + } + fs::write(&path, &file.payload) + .map_err(|e| anyhow!("Failed to write {:?}: {}", path, e))?; + crate::progress::on_bytes(file.payload.len()); + crate::metrics::add_bytes( + &self.kind, + crate::metrics::Direction::Write, + file.payload.len(), + ); + self.written_files.lock().unwrap().push(path); + } + crate::progress::on_record(); + crate::metrics::add_items(&self.kind, crate::metrics::Direction::Write, 1); + Ok(()) + } + + async fn close(self) -> Result<()> { + self.closed.store(true, Ordering::Relaxed); + Ok(()) + } +} + +impl Drop for JsonFsWriter { + fn drop(&mut self) { + if self.closed.load(Ordering::Relaxed) { + return; + } + // Session aborted: roll back every file we touched so partial data + // doesn't get mistaken for a complete archive. + for path in self.written_files.lock().unwrap().drain(..) { + if let Err(e) = fs::remove_file(&path) { + tracing::error!( + "Failed to remove uncommitted JSON file {:?}: {:?}", + path, + e + ); + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use chrono::{TimeZone, Utc}; + use tempfile::tempdir; + + use crate::archiver::datakind::DataKind; + use crate::record::{ArchiveRow, BlockchainType, Field}; + + fn block_row(height: u64) -> ArchiveRow { + ArchiveRow { + kind: DataKind::Blocks, + blockchain_type: BlockchainType::Ethereum, + blockchain_id: "ETH".to_string(), + archive_ts: Utc::now(), + height, + block_id: format!("0xblock{}", height), + timestamp: Utc.timestamp_millis_opt(0).unwrap(), + parent_id: Some(format!("0xparent{}", height - 1)), + tx_index: None, + tx_id: None, + fields: vec![ + Field::BlockJson(format!("{{\"h\":{}}}", height).into_bytes()), + Field::Uncle { index: 0, json: b"U0".to_vec() }, + ], + } + } + + fn tx_row(height: u64, tx_id: &str) -> ArchiveRow { + ArchiveRow { + kind: DataKind::Transactions, + blockchain_type: BlockchainType::Ethereum, + blockchain_id: "ETH".to_string(), + archive_ts: Utc::now(), + height, + block_id: format!("0xblock{}", height), + timestamp: Utc.timestamp_millis_opt(0).unwrap(), + parent_id: None, + tx_index: Some(0), + tx_id: Some(tx_id.to_string()), + fields: vec![ + Field::TxJson(b"{}".to_vec()), + Field::TxRaw(vec![0xde, 0xad, 0xbe, 0xef]), + Field::Receipt(b"R".to_vec()), + ], + } + } + + #[tokio::test] + async fn writes_per_field_files_under_height_dir() { + let tmp = tempdir().unwrap(); + let storage = JsonFsStorage::new(tmp.path().to_path_buf(), Filenames::with_dir("eth".to_string())); + + let writer = storage + .create(DataKind::Blocks, &Range::Single(21596362.into()), true) + .await + .unwrap() + .unwrap(); + writer.append(block_row(21596362)).await.unwrap(); + writer.close().await.unwrap(); + + let dir = tmp.path().join("eth/021000000/021596000/021596362"); + assert!(dir.join("block.json").exists()); + assert!(dir.join("uncle-0.json").exists()); + assert_eq!(fs::read(dir.join("block.json")).unwrap(), b"{\"h\":21596362}"); + } + + #[tokio::test] + async fn tx_files_use_txid_in_filename_and_keep_raw_hex() { + let tmp = tempdir().unwrap(); + let storage = JsonFsStorage::new(tmp.path().to_path_buf(), Filenames::with_dir("eth".to_string())); + + let writer = storage + .create(DataKind::Transactions, &Range::Single(21596362.into()), true) + .await + .unwrap() + .unwrap(); + writer.append(tx_row(21596362, "0xabc")).await.unwrap(); + writer.close().await.unwrap(); + + let dir = tmp.path().join("eth/021000000/021596000/021596362"); + assert!(dir.join("tx-0xabc.json").exists()); + assert!(dir.join("receipt-0xabc.json").exists()); + let raw = dir.join("raw-0xabc.hex"); + assert!(raw.exists()); + assert_eq!(fs::read(raw).unwrap(), b"0xdeadbeef"); + } + + #[tokio::test] + async fn overwrite_false_keeps_existing_file() { + let tmp = tempdir().unwrap(); + let storage = JsonFsStorage::new(tmp.path().to_path_buf(), Filenames::with_dir("eth".to_string())); + + // First write + let w = storage + .create(DataKind::Blocks, &Range::Single(21596362.into()), true) + .await + .unwrap() + .unwrap(); + w.append(block_row(21596362)).await.unwrap(); + w.close().await.unwrap(); + + // Second write with overwrite=false and different payload — should not change file + let mut row = block_row(21596362); + row.fields = vec![Field::BlockJson(b"REPLACED".to_vec())]; + let w = storage + .create(DataKind::Blocks, &Range::Single(21596362.into()), false) + .await + .unwrap() + .unwrap(); + w.append(row).await.unwrap(); + w.close().await.unwrap(); + + let block = tmp.path().join("eth/021000000/021596000/021596362/block.json"); + assert_eq!(fs::read(block).unwrap(), b"{\"h\":21596362}"); + } + + #[tokio::test] + async fn drop_without_close_removes_written_files() { + let tmp = tempdir().unwrap(); + let storage = JsonFsStorage::new(tmp.path().to_path_buf(), Filenames::with_dir("eth".to_string())); + { + let w = storage + .create(DataKind::Blocks, &Range::Single(21596362.into()), true) + .await + .unwrap() + .unwrap(); + w.append(block_row(21596362)).await.unwrap(); + // Drop without closing. + } + let block = tmp.path().join("eth/021000000/021596000/021596362/block.json"); + assert!(!block.exists()); + } + + #[tokio::test(flavor = "multi_thread")] + async fn archive_command_end_to_end_writes_per_field_files() { + use crate::archiver::Archiver; + use crate::args::Args; + use crate::blockchain::mock::{MockBlock, MockData, MockTx, MockType}; + use crate::command::archive::ArchiveCommand; + use crate::command::CommandExecutor; + use std::sync::Arc; + + crate::testing::start_test(); + let tmp = tempdir().unwrap(); + + // Mock blockchain with three blocks, each with two transactions. + let data = MockData::new("TEST"); + let data_provider: Arc = Arc::new(data); + for h in 100..103u64 { + let txs = vec![format!("0xTX{}-A", h), format!("0xTX{}-B", h)]; + data_provider.add_block(MockBlock { + height: h, + hash: format!("0xB{}", h), + parent: format!("0xB{}", h - 1), + transactions: txs.clone(), + }); + for tx in &txs { + data_provider.add_tx(MockTx { hash: tx.clone() }); + } + } + + let storage = JsonFsStorage::new( + tmp.path().to_path_buf(), + Filenames::with_dir("test".to_string()), + ); + let archiver: Archiver = + Archiver::new_simple(Arc::new(storage), data_provider); + + let args = Args { + range: Some("100..102".to_string()), + range_chunk: Some(10), + ..Default::default() + }; + let cmd = ArchiveCommand::new(&args, archiver).unwrap(); + cmd.execute().await.unwrap(); + + // Every height should have block.json plus per-tx files. + for h in 100..103u64 { + let dir = tmp + .path() + .join(format!("test/000000000/000000000/{:09}", h)); + assert!(dir.join("block.json").exists(), "missing block.json for {}", h); + for letter in ["A", "B"] { + let tx_id = format!("0xTX{}-{}", h, letter); + assert!( + dir.join(format!("tx-{}.json", tx_id)).exists(), + "missing tx-{}.json", + tx_id + ); + assert!( + dir.join(format!("raw-{}.hex", tx_id)).exists(), + "missing raw-{}.hex", + tx_id + ); + } + } + } + + #[tokio::test] + async fn find_incomplete_tables_flags_missing_block_files() { + let tmp = tempdir().unwrap(); + let storage = JsonFsStorage::new(tmp.path().to_path_buf(), Filenames::with_dir("eth".to_string())); + + // Write height 100 only (block + tx + raw); leave 101 and 102 missing. + let writer = storage + .create(DataKind::Blocks, &Range::Single(100.into()), true) + .await + .unwrap() + .unwrap(); + writer.append(block_row(100)).await.unwrap(); + writer.close().await.unwrap(); + let writer = storage + .create(DataKind::Transactions, &Range::Single(100.into()), true) + .await + .unwrap() + .unwrap(); + writer.append(tx_row(100, "0xabc")).await.unwrap(); + writer.close().await.unwrap(); + + let options = DataOptions::default(); // blocks + tx + let incomplete = storage + .find_incomplete_tables(Range::new(100, 102), &options) + .await + .unwrap(); + // 101 and 102 are both missing both kinds — should merge into a single range. + assert_eq!(incomplete.len(), 1); + assert_eq!(incomplete[0].0, Range::new(101, 102)); + let mut kinds = incomplete[0].1.clone(); + kinds.sort(); + assert_eq!(kinds, vec![DataKind::Blocks, DataKind::Transactions]); + } +} diff --git a/src/storage/json_objects.rs b/src/storage/json_objects.rs new file mode 100644 index 0000000..666a926 --- /dev/null +++ b/src/storage/json_objects.rs @@ -0,0 +1,331 @@ +// Copyright 2026 EmeraldPay Ltd +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. + +//! `object_store`-backed JSON-per-field target. Same layout as +//! [`crate::storage::json_fs`] but on S3 (or any other `ObjectStore`). +//! +//! Unlike the filesystem variant, dropping a [`JsonObjectsWriter`] without +//! calling [`close`](TargetFileWriter::close) does **not** roll back already-written +//! objects: `object_store`'s async API can't be driven from `Drop` reliably, so we +//! leave the files in place and rely on [`JsonObjectsStorage::find_incomplete_tables`] +//! to detect partial heights on the next Fix run. + +use std::collections::HashMap; +use std::sync::Arc; + +use anyhow::{anyhow, Result}; +use async_trait::async_trait; +use bytes::Bytes; +use futures_util::StreamExt; +use object_store::path::Path; +use object_store::{ObjectStore, ObjectStoreExt, PutPayload}; +use crate::archiver::datakind::{DataKind, DataOptions}; +use crate::archiver::filenames::Filenames; +use crate::archiver::range::Range; +use crate::formats::json; +use crate::record::ArchiveRow; +use crate::storage::{ScanTarget, TargetFile, TargetFileWriter, WriteTarget}; + +pub struct JsonObjectsStorage { + os: Arc, + bucket: String, + filenames: Filenames, +} + +impl JsonObjectsStorage { + pub fn new(os: Arc, bucket: String, filenames: Filenames) -> Self { + Self { os, bucket, filenames } + } + + fn height_prefix(&self, height: u64) -> String { + // height_dir returns no trailing slash; add one so `list` scopes the + // prefix to the height's directory rather than matching sibling + // directories that happen to share a prefix. + format!("{}/", self.filenames.height_dir(height)) + } + + async fn list_filenames(&self, height: u64) -> Result> { + let prefix = Path::from(self.height_prefix(height)); + let mut stream = self.os.list(Some(&prefix)); + let mut out = Vec::new(); + while let Some(meta) = stream.next().await { + let meta = meta.map_err(|e| anyhow!("list error: {:?}", e))?; + if let Some(name) = meta.location.filename() { + out.push(name.to_string()); + } + } + Ok(out) + } +} + +#[async_trait] +impl WriteTarget for JsonObjectsStorage { + type Writer = JsonObjectsWriter; + + async fn create( + &self, + kind: DataKind, + range: &Range, + overwrite: bool, + ) -> Result> { + Ok(Some(JsonObjectsWriter { + os: self.os.clone(), + bucket: self.bucket.clone(), + filenames: self.filenames.clone(), + kind, + range: range.clone(), + overwrite, + })) + } +} + +#[async_trait] +impl ScanTarget for JsonObjectsStorage { + /// Same heuristic as [`crate::storage::json_fs::JsonFsStorage`]; the per-height + /// completeness rules live in [`json::missing_kinds_at_height`]. + async fn find_incomplete_tables( + &self, + blocks: Range, + tx_options: &DataOptions, + ) -> Result)>> { + let mut per_height: HashMap> = HashMap::new(); + // Collect heights up front — Range::iter() returns a non-Send iterator + // we can't hold across an .await. + let heights: Vec = blocks.iter().collect(); + for height in heights { + let entries = self.list_filenames(height).await?; + let missing = json::missing_kinds_at_height(entries, tx_options); + if !missing.is_empty() { + per_height.insert(height, missing); + } + } + Ok(json::collapse_missing(per_height)) + } +} + +pub struct JsonObjectsWriter { + os: Arc, + bucket: String, + filenames: Filenames, + kind: DataKind, + range: Range, + overwrite: bool, +} + +impl TargetFile for JsonObjectsWriter { + fn get_url(&self) -> String { + let path = match &self.range { + Range::Single(h) => self.filenames.height_dir(h.height), + Range::Multiple(start, _) => { + let dir = self.filenames.height_dir(start.height); + dir.rsplit_once('/') + .map(|(p, _)| p.to_string()) + .unwrap_or(dir) + } + }; + format!("s3://{}/{}", self.bucket, path) + } +} + +#[async_trait] +impl TargetFileWriter for JsonObjectsWriter { + async fn append(&self, row: ArchiveRow) -> Result<()> { + let dir = self.filenames.height_dir(row.height); + let files = json::encode_row(&row); + for file in files { + let key = format!("{}/{}", dir, file.filename); + let path = Path::from(key); + if !self.overwrite { + if self.os.head(&path).await.is_ok() { + tracing::debug!("Skipping existing JSON object: {}", path); + continue; + } + } + let payload_len = file.payload.len(); + self.os + .put(&path, PutPayload::from(Bytes::from(file.payload))) + .await + .map_err(|e| anyhow!("Failed to put {}: {:?}", path, e))?; + crate::progress::on_bytes(payload_len); + crate::metrics::add_bytes(&self.kind, crate::metrics::Direction::Write, payload_len); + } + crate::progress::on_record(); + crate::metrics::add_items(&self.kind, crate::metrics::Direction::Write, 1); + Ok(()) + } + + async fn close(self) -> Result<()> { + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use chrono::{TimeZone, Utc}; + use object_store::memory::InMemory; + + use crate::archiver::datakind::DataKind; + use crate::record::{BlockchainType, Field}; + + fn block_row(height: u64) -> ArchiveRow { + ArchiveRow { + kind: DataKind::Blocks, + blockchain_type: BlockchainType::Ethereum, + blockchain_id: "ETH".to_string(), + archive_ts: Utc::now(), + height, + block_id: format!("0xblock{}", height), + timestamp: Utc.timestamp_millis_opt(0).unwrap(), + parent_id: Some(format!("0xparent{}", height.saturating_sub(1))), + tx_index: None, + tx_id: None, + fields: vec![Field::BlockJson(format!("{{\"h\":{}}}", height).into_bytes())], + } + } + + fn tx_row(height: u64, tx_id: &str) -> ArchiveRow { + ArchiveRow { + kind: DataKind::Transactions, + blockchain_type: BlockchainType::Ethereum, + blockchain_id: "ETH".to_string(), + archive_ts: Utc::now(), + height, + block_id: format!("0xblock{}", height), + timestamp: Utc.timestamp_millis_opt(0).unwrap(), + parent_id: None, + tx_index: Some(0), + tx_id: Some(tx_id.to_string()), + fields: vec![ + Field::TxJson(b"{}".to_vec()), + Field::TxRaw(vec![0xde, 0xad, 0xbe, 0xef]), + Field::Receipt(b"R".to_vec()), + ], + } + } + + async fn list_paths(mem: &InMemory) -> Vec { + let mut stream = mem.list(None); + let mut out = Vec::new(); + while let Some(meta) = stream.next().await { + if let Ok(meta) = meta { + out.push(meta.location.to_string()); + } + } + out.sort(); + out + } + + #[tokio::test] + async fn writes_per_field_objects_under_height_prefix() { + let mem = Arc::new(InMemory::new()); + let storage = JsonObjectsStorage::new( + mem.clone(), + "bucket".to_string(), + Filenames::with_dir("eth".to_string()), + ); + + let writer = storage + .create(DataKind::Blocks, &Range::Single(21596362.into()), true) + .await + .unwrap() + .unwrap(); + writer.append(block_row(21596362)).await.unwrap(); + writer.close().await.unwrap(); + + let paths = list_paths(&mem).await; + assert_eq!(paths, vec!["eth/021000000/021596000/021596362/block.json"]); + } + + #[tokio::test] + async fn tx_files_use_txid_in_filename_and_keep_raw_hex() { + let mem = Arc::new(InMemory::new()); + let storage = JsonObjectsStorage::new( + mem.clone(), + "bucket".to_string(), + Filenames::with_dir("eth".to_string()), + ); + + let writer = storage + .create(DataKind::Transactions, &Range::Single(21596362.into()), true) + .await + .unwrap() + .unwrap(); + writer.append(tx_row(21596362, "0xabc")).await.unwrap(); + writer.close().await.unwrap(); + + let raw_path = Path::from("eth/021000000/021596000/021596362/raw-0xabc.hex"); + let raw = mem.get(&raw_path).await.unwrap().bytes().await.unwrap(); + assert_eq!(&raw[..], b"0xdeadbeef"); + } + + #[tokio::test] + async fn overwrite_false_keeps_existing_object() { + let mem = Arc::new(InMemory::new()); + let storage = JsonObjectsStorage::new( + mem.clone(), + "bucket".to_string(), + Filenames::with_dir("eth".to_string()), + ); + + let w = storage + .create(DataKind::Blocks, &Range::Single(21596362.into()), true) + .await + .unwrap() + .unwrap(); + w.append(block_row(21596362)).await.unwrap(); + w.close().await.unwrap(); + + let mut row = block_row(21596362); + row.fields = vec![Field::BlockJson(b"REPLACED".to_vec())]; + let w = storage + .create(DataKind::Blocks, &Range::Single(21596362.into()), false) + .await + .unwrap() + .unwrap(); + w.append(row).await.unwrap(); + w.close().await.unwrap(); + + let path = Path::from("eth/021000000/021596000/021596362/block.json"); + let bytes = mem.get(&path).await.unwrap().bytes().await.unwrap(); + assert_eq!(&bytes[..], b"{\"h\":21596362}"); + } + + #[tokio::test] + async fn find_incomplete_tables_detects_missing_heights() { + let mem = Arc::new(InMemory::new()); + let storage = JsonObjectsStorage::new( + mem.clone(), + "bucket".to_string(), + Filenames::with_dir("eth".to_string()), + ); + + let writer = storage + .create(DataKind::Blocks, &Range::Single(100.into()), true) + .await + .unwrap() + .unwrap(); + writer.append(block_row(100)).await.unwrap(); + writer.close().await.unwrap(); + let writer = storage + .create(DataKind::Transactions, &Range::Single(100.into()), true) + .await + .unwrap() + .unwrap(); + writer.append(tx_row(100, "0xabc")).await.unwrap(); + writer.close().await.unwrap(); + + let options = DataOptions::default(); + let incomplete = storage + .find_incomplete_tables(Range::new(100, 102), &options) + .await + .unwrap(); + assert_eq!(incomplete.len(), 1); + assert_eq!(incomplete[0].0, Range::new(101, 102)); + let mut kinds = incomplete[0].1.clone(); + kinds.sort(); + assert_eq!(kinds, vec![DataKind::Blocks, DataKind::Transactions]); + } +} diff --git a/src/storage/mod.rs b/src/storage/mod.rs index 6f73818..e6bf078 100644 --- a/src/storage/mod.rs +++ b/src/storage/mod.rs @@ -5,6 +5,8 @@ use async_trait::async_trait; use crate::{ storage::{ fs::FsStorage, + json_fs::JsonFsStorage, + json_objects::JsonObjectsStorage, objects::ObjectsStorage }, archiver::{ @@ -28,6 +30,8 @@ use url::Url; use itertools::Itertools; pub mod fs; +pub mod json_fs; +pub mod json_objects; pub mod objects; mod avro_reader; mod copy; @@ -41,33 +45,37 @@ pub fn is_fs(args: &Args) -> bool { args.dir.is_some() && !is_s3(args) } -pub fn create_aws(value: &Args) -> Result> { - // inside the archive we create a subdirectory for each blockchain +/// +/// Compute the S3 key prefix (without bucket) under which the archive lives. +/// +/// Joins the user-supplied `--dir` URL path with the lowercased blockchain code +/// (e.g. `eth`), trimming any leading/trailing slashes on the URL path so +/// `s3://bucket/`, `s3://bucket/archive`, and `s3://bucket/archive/` all +/// produce the expected `archive/eth` (or just `eth` for the bucket root). +fn s3_parent_dir(value: &Args) -> Result { let blockchain_dir = value.get_blockchain()?.code().to_lowercase(); - let aws = value.aws.as_ref().unwrap(); - - tracing::info!("Using S3 storage"); - - let parent_dir = if let Some(dir) = &value.dir { - url::Url::parse(dir) - .map_err(|_| anyhow!("Please specify a --dir as s3://bucket/path"))? - .path() - .to_string() - } else { - "".to_string() - }; - - let parent_dir = if parent_dir.ends_with('/') { + if value.dir.is_none() { + return Err(anyhow!("Please set target dir as a s3://bucket/path")); + } + let raw = url::Url::parse(value.dir.as_ref().unwrap()) + .map_err(|_| anyhow!("Please specify a --dir as s3://bucket/path"))? + .path() + .to_string(); + let trimmed = raw.trim_matches('/'); + Ok(if trimmed.is_empty() { blockchain_dir } else { - format!("{}/{}", parent_dir, blockchain_dir) - }; + format!("{}/{}", trimmed, blockchain_dir) + }) +} - let filenames = Filenames::with_dir(parent_dir); +pub fn create_aws(value: &Args) -> Result> { + let aws = value.aws.as_ref().unwrap(); - if value.dir.is_none() { - return Err(anyhow!("Please set target dir as a s3://bucket/path")); - } + tracing::info!("Using S3 storage"); + + // s3_parent_dir errors if --dir is missing, so subsequent unwraps are safe. + let filenames = Filenames::with_dir(s3_parent_dir(value)?); let mut builder = AmazonS3Builder::new() .with_access_key_id(aws.access_key.clone()) @@ -114,15 +122,66 @@ pub fn create_fs(value: &Args) -> Result { Ok(FsStorage::new(PathBuf::from(dir), Filenames::with_dir(blockchain_dir))) } +/// Build a [`JsonFsStorage`] using the same root-dir and blockchain-subdirectory +/// conventions as [`create_fs`]. +pub fn create_fs_json(value: &Args) -> Result { + let blockchain_dir = value.get_blockchain()?.code().to_lowercase(); + tracing::info!("Using Filesystem storage (JSON per-field layout)"); + if value.dir.is_none() { + return Err(anyhow!("No target dir set for a Filesystem based storage")); + } + let dir = value.dir.as_ref().unwrap().clone(); + Ok(JsonFsStorage::new( + PathBuf::from(dir), + Filenames::with_dir(blockchain_dir), + )) +} + +/// Build a [`JsonObjectsStorage`] backed by AWS S3 (or any S3-compatible service), +/// sharing the same authentication and URL-parsing logic as [`create_aws`]. +pub fn create_aws_json(value: &Args) -> Result> { + let aws = value.aws.as_ref().unwrap(); + + tracing::info!("Using S3 storage (JSON per-field layout)"); + + // s3_parent_dir errors if --dir is missing, so subsequent unwraps are safe. + let filenames = Filenames::with_dir(s3_parent_dir(value)?); + + let mut builder = AmazonS3Builder::new() + .with_access_key_id(aws.access_key.clone()) + .with_secret_access_key(aws.secret_key.clone()) + .with_url(value.dir.clone().unwrap()) + .with_client_options( + ClientOptions::default() + .with_allow_http(true) + .with_allow_invalid_certificates(aws.trust_tls), + ); + if let Some(endpoint) = &aws.endpoint { + builder = builder.with_endpoint(endpoint.clone()); + } else { + builder = builder.with_endpoint("https://s3.amazonaws.com".to_string()); + } + if let Some(region) = &aws.region { + builder = builder.with_region(region.clone()); + } + builder = builder.with_virtual_hosted_style_request(!aws.path_style); + + let bucket = { + let parsed = Url::parse(value.dir.clone().unwrap().as_str())?; + parsed.host_str().unwrap().to_string() + }; + let os = builder + .build() + .map_err(|e| anyhow!("Invalid S3 connection options: {}", e))?; + + Ok(JsonObjectsStorage::new(Arc::new(os), bucket, filenames)) +} + /// /// Capability trait for targets that can have new records appended. /// -/// All targets (Avro/JSON files, future Pulsar/Kafka topics) implement this. -/// `archive` requires only this capability. The `stream` command currently also -/// requires [`ReadTarget`] (it uses `find_incomplete_tables` to honour -/// `--continue`); when streaming-only targets land in Phase 3 it will branch on -/// the target type and the file-based path will be the one that needs read -/// capabilities. +/// Every target (Avro/JSON files, future Pulsar/Kafka topics) implements this. +/// The `archive` command requires only this capability. #[async_trait] pub trait WriteTarget: Send + Sync { /// @@ -137,14 +196,30 @@ pub trait WriteTarget: Send + Sync { } /// -/// Capability trait for targets that can list, open, and delete existing files. +/// Capability trait for targets that can report which heights/kinds are not +/// yet fully archived. Required by the `stream --continue` and `fix` commands. +/// +/// Both file-based targets (Avro, JSON) implement this; streaming targets +/// (Pulsar, Kafka) will get a separate resume mechanism in Phase 3 and +/// intentionally do not implement [`ScanTarget`]. +#[async_trait] +pub trait ScanTarget: WriteTarget { + async fn find_incomplete_tables( + &self, + blocks: Range, + tx_options: &DataOptions, + ) -> Result)>>; +} + +/// +/// Capability trait for targets whose existing data can be enumerated, opened, +/// and deleted file-by-file. Required by the `verify` and `compact` commands. /// -/// Implemented by file-based targets (filesystem, S3) but **not** by streaming -/// targets (Pulsar, Kafka), which are append-only logs. Commands that need to -/// inspect or rewrite existing data (`verify`, `fix`, `compact`) bound on this -/// trait. +/// Implemented by the row-batched file backends (Avro on FS/S3, future Parquet) +/// but **not** by the per-field JSON layout (which has no row-batched files to +/// open or merge) or by streaming targets. #[async_trait] -pub trait ReadTarget: WriteTarget { +pub trait ReadTarget: ScanTarget { /// /// A type used for reading the existing records. type Reader: TargetFileReader + Send + Sync; @@ -160,72 +235,85 @@ pub trait ReadTarget: WriteTarget { /// /// List all files in the range fn list(&self, range: Range) -> Result>; +} - async fn find_incomplete_tables(&self, blocks: Range, tx_options: &DataOptions) -> Result)>> { - tracing::info!("Check if blocks are fully archived in range: {}", blocks); - let mut existing = self.list(blocks.clone())?; - let mut archived = ArchivesList::new(tx_options.files().clone()); - - // Track which ranges have no files at all - let mut missing_ranges = RangeBag::new(); - missing_ranges.append(blocks.clone()); - - let shutdown = global::get_shutdown(); - while let Some(file) = existing.recv().await { - // Remove this file's range from the missing ranges - missing_ranges.remove(&file.range); - archived.append(file)?; - if shutdown.is_signalled() { - tracing::info!("Shutdown signal received"); - return Ok(vec![]); - } +/// +/// Default [`ScanTarget::find_incomplete_tables`] implementation for row-batched +/// file-based targets — uses [`ReadTarget::list`] to enumerate existing files and +/// computes the missing kinds via [`ArchivesList`]. Called by [`fs::FsStorage`] +/// and [`objects::ObjectsStorage`]; per-field JSON backends override +/// `find_incomplete_tables` with their own heuristic and don't go through here. +pub async fn find_incomplete_by_listing( + target: &T, + blocks: Range, + tx_options: &DataOptions, +) -> Result)>> +where + T: ReadTarget, +{ + tracing::info!("Check if blocks are fully archived in range: {}", blocks); + let mut existing = target.list(blocks.clone())?; + let mut archived = ArchivesList::new(tx_options.files().clone()); + + // Track which ranges have no files at all + let mut missing_ranges = RangeBag::new(); + missing_ranges.append(blocks.clone()); + + let shutdown = global::get_shutdown(); + while let Some(file) = existing.recv().await { + // Remove this file's range from the missing ranges + missing_ranges.remove(&file.range); + archived.append(file)?; + if shutdown.is_signalled() { + tracing::info!("Shutdown signal received"); + return Ok(vec![]); } + } - let incomplete = archived.list_incomplete(); - let mut result: Vec<(Range, Vec)> = Vec::new(); - - // Add ranges that have incomplete files (some files present but not all) - if !incomplete.is_empty() { - // We do not process ranges here (like joining them) because we don't want to break any existing layout (i.e, chunk size, alignment, etc) - let ranges = incomplete.iter() - .map(|g| g.range.clone()) - .collect::>(); - - tracing::debug!("Incomplete blocks {} (sample: {},...)", - ranges.len(), - ranges.iter().take(5).map(|r| r.to_string()).join(",") - ); - - let incomplete_by_range: Vec<(Range, Vec)> = ranges.iter() - .filter_map(|r| { - // we clone all file groups here b/c if it split by different ranges each range would download its part - incomplete.iter() - .find(|g| g.range.eq(r)) - .map(|g| (r.clone(), g.get_incomplete_kinds())) - }) - .collect(); - result.extend(incomplete_by_range); - } + let incomplete = archived.list_incomplete(); + let mut result: Vec<(Range, Vec)> = Vec::new(); - // Add completely missing ranges (no files at all) - if !missing_ranges.is_empty() { - let all_kinds = tx_options.files().files.clone(); - tracing::debug!("Missing blocks (no files at all) {} (sample: {},...)", - missing_ranges.len(), - missing_ranges.ranges.iter().take(5).map(|r| r.to_string()).join(",") - ); - - for range in missing_ranges.ranges { - result.push((range, all_kinds.clone())); - } - } + // Add ranges that have incomplete files (some files present but not all) + if !incomplete.is_empty() { + // We do not process ranges here (like joining them) because we don't want to break any existing layout (i.e, chunk size, alignment, etc) + let ranges = incomplete.iter() + .map(|g| g.range.clone()) + .collect::>(); + + tracing::debug!("Incomplete blocks {} (sample: {},...)", + ranges.len(), + ranges.iter().take(5).map(|r| r.to_string()).join(",") + ); + + let incomplete_by_range: Vec<(Range, Vec)> = ranges.iter() + .filter_map(|r| { + // we clone all file groups here b/c if it split by different ranges each range would download its part + incomplete.iter() + .find(|g| g.range.eq(r)) + .map(|g| (r.clone(), g.get_incomplete_kinds())) + }) + .collect(); + result.extend(incomplete_by_range); + } - if result.is_empty() { - tracing::info!("All blocks are fully archived"); + // Add completely missing ranges (no files at all) + if !missing_ranges.is_empty() { + let all_kinds = tx_options.files().files.clone(); + tracing::debug!("Missing blocks (no files at all) {} (sample: {},...)", + missing_ranges.len(), + missing_ranges.ranges.iter().take(5).map(|r| r.to_string()).join(",") + ); + + for range in missing_ranges.ranges { + result.push((range, all_kinds.clone())); } + } - Ok(result) + if result.is_empty() { + tracing::info!("All blocks are fully archived"); } + + Ok(result) } pub trait TargetFile { @@ -310,6 +398,60 @@ mod tests { use object_store::path::Path; use object_store::{ObjectStoreExt, PutPayload}; + fn args_with_dir(dir: &str) -> Args { + Args { + blockchain: "ethereum".to_string(), + dir: Some(dir.to_string()), + ..Default::default() + } + } + + #[test] + fn s3_parent_dir_trims_trailing_slash() { + assert_eq!( + s3_parent_dir(&args_with_dir("s3://bucket/archive/")).unwrap(), + "archive/eth" + ); + } + + #[test] + fn s3_parent_dir_keeps_prefix_without_trailing_slash() { + assert_eq!( + s3_parent_dir(&args_with_dir("s3://bucket/archive")).unwrap(), + "archive/eth" + ); + } + + #[test] + fn s3_parent_dir_handles_nested_prefix() { + assert_eq!( + s3_parent_dir(&args_with_dir("s3://bucket/foo/bar/")).unwrap(), + "foo/bar/eth" + ); + } + + #[test] + fn s3_parent_dir_bucket_root() { + assert_eq!( + s3_parent_dir(&args_with_dir("s3://bucket/")).unwrap(), + "eth" + ); + assert_eq!( + s3_parent_dir(&args_with_dir("s3://bucket")).unwrap(), + "eth" + ); + } + + #[test] + fn s3_parent_dir_errors_without_dir() { + let args = Args { + blockchain: "ethereum".to_string(), + dir: None, + ..Default::default() + }; + assert!(s3_parent_dir(&args).is_err()); + } + /// Helper to create a test storage with specified files async fn create_test_storage(files: Vec<&str>) -> ObjectsStorage { let mem = Arc::new(InMemory::new()); diff --git a/src/storage/objects.rs b/src/storage/objects.rs index 747d771..551c718 100644 --- a/src/storage/objects.rs +++ b/src/storage/objects.rs @@ -19,13 +19,16 @@ use tokio::{ use tokio::sync::mpsc::Sender; use tokio::sync::Mutex; use tokio_util::io::{StreamReader, SyncIoBridge}; -use crate::archiver::datakind::DataKind; +use crate::archiver::datakind::{DataKind, DataOptions}; use crate::archiver::filenames::{Filenames, Level, LevelDouble, LevelSingle}; use crate::archiver::range::Range; use crate::formats::avro; use crate::global; use crate::record::ArchiveRow; -use crate::storage::{avro_reader, copy, sorted_files, FileReference, ReadTarget, TargetFile, TargetFileReader, TargetFileWriter, WriteTarget}; +use crate::storage::{ + avro_reader, copy, find_incomplete_by_listing, sorted_files, FileReference, ReadTarget, + ScanTarget, TargetFile, TargetFileReader, TargetFileWriter, WriteTarget, +}; pub struct ObjectsStorage { os: Arc, @@ -56,6 +59,17 @@ impl WriteTarget for ObjectsStorage { } } +#[async_trait] +impl ScanTarget for ObjectsStorage { + async fn find_incomplete_tables( + &self, + blocks: Range, + tx_options: &DataOptions, + ) -> anyhow::Result)>> { + find_incomplete_by_listing(self, blocks, tx_options).await + } +} + #[async_trait] impl ReadTarget for ObjectsStorage {