Merge pull request #24 from emeraldpay/feature/target-json

Author: splix

Cargo.lock

@@ -48,3 +48,4 @@ shadow-rs = "1.4"
 [dev-dependencies]
 testcontainers = "0.27"
 rand = "0.10"
+tempfile = "3"

Cargo.toml

@@ -47,61 +47,103 @@ Arguments:
 Options:
   -b, --blockchain <BLOCKCHAIN>
           Blockchain
+
       --dry-run
           Do not modify the storage
+
   -c, --connection <HOST:PORT>
           Connection (host:port)
+
       --connection.notls
           Disable TLS
+
       --parallel <PARALLEL>
           How many API requests to make in parallel. Range: 1..512. Default: 16
+
       --notify.dir <NOTIFY_DIR>
           Write notifications as JSON line to the specified dir in a file <dshackle-archive-%STARTTIME.jsonl>
+
       --notify.pulsar.topic <PULSAR_TOPIC>
           Send notifications as JSON to the Pulsar to the specified topic (notify.pulsar.url must be specified)
+
       --notify.pulsar.url <PULSAR_URL>
           Send notifications as JSON to the Pulsar with specified URL (notify.pulsar.topic must be specified)
+
       --auth.aws.access-key <ACCESS_KEY>
           AWS / S3 Access Key
+
       --auth.aws.secret-key <SECRET_KEY>
           AWS / S3 Secret Key
+
       --aws.endpoint <ENDPOINT>
           AWS / S3 endpoint url instead of the default one
+
       --aws.region <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 <DIR>
           Target directory
+
       --continue
           [Stream Command] Continue from the last archived block; i.e., not the latest in blockchain
+
       --tail <TAIL>
           [Verify/Fix Commands] Use the latest T blocks instead of a range
+
   -r, --range <RANGE>
           Blocks Range (`N...M`)
+
       --range.chunk <RANGE_CHUNK>
           Range chunk size (default 1000)
+
   -t, --tables <TABLES>
           Types of tables to archive (comma-separated list of `blocks`, `txes`, `traces`). Default: blocks,txes
+
       --fields.trace <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>
-          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 <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 <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-<HASH>.json, receipt-<HASH>.json, raw-<HASH>.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 <HOST:PORT>
           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

README.adoc

@@ -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-<HASH>.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)

src/archiver/filenames.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-<HASH>.json, receipt-<HASH>.json,
+    /// raw-<HASH>.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,

src/args.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)

src/blockchain/ethereum.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,15 +9,15 @@ use crate::archiver::datakind::DataOptions;
 /// It checks the archive for the specified range and add missing data
 ///
 #[derive(Clone)]
-pub struct FixCommand<B: BlockchainTypes, TS: ReadTarget> {
+pub struct FixCommand<B: BlockchainTypes, TS: ScanTarget> {
     b: PhantomData<B>,
     blocks: Blocks,
     chunk_size: usize,
     archiver: Archiver<B, TS>,
     tx_options: DataOptions,
 }
 
-impl<B: BlockchainTypes, TS: ReadTarget> FixCommand<B, TS> {
+impl<B: BlockchainTypes, TS: ScanTarget> FixCommand<B, TS> {
     pub fn new(config: &Args,
                archiver: Archiver<B, TS>) -> anyhow::Result<Self> {
 
@@ -34,7 +34,7 @@ impl<B: BlockchainTypes, TS: ReadTarget> FixCommand<B, TS> {
 }
 
 #[async_trait]
-impl<B: BlockchainTypes, TS: ReadTarget> CommandExecutor for FixCommand<B, TS> {
+impl<B: BlockchainTypes, TS: ScanTarget> CommandExecutor for FixCommand<B, TS> {
 
     async fn execute(&self) -> anyhow::Result<()> {
         let shutdown = global::get_shutdown();

src/command/fix.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<B: BlockchainTypes, TS: ReadTarget> {
+pub struct StreamCommand<B: BlockchainTypes, TS: ScanTarget> {
     b: PhantomData<B>,
     blockchain: Arc<Blockchain>,
     continue_blocks: Option<u64>,
@@ -34,7 +34,7 @@ pub struct StreamCommand<B: BlockchainTypes, TS: ReadTarget> {
     follow: Follow,
 }
 
-impl<B: BlockchainTypes, TS: ReadTarget> StreamCommand<B, TS> {
+impl<B: BlockchainTypes, TS: ScanTarget> StreamCommand<B, TS> {
     pub async fn new(config: &Args,
                      archiver: Archiver<B, TS>
     ) -> Result<Self> {
@@ -87,7 +87,7 @@ impl<B: BlockchainTypes, TS: ReadTarget> StreamCommand<B, TS> {
 }
 
 #[async_trait]
-impl<B: BlockchainTypes, TS: ReadTarget> CommandExecutor for StreamCommand<B, TS> {
+impl<B: BlockchainTypes, TS: ScanTarget> CommandExecutor for StreamCommand<B, TS> {
 
     async fn execute(&self) -> Result<()> {