feat(rooch-da): add transaction stats tracking to unpack (#3162)

## Summary

1. Added support for tracking and displaying L2 transaction size statistics in the unpack command, including histograms and top-N transaction sizes. Also introduced a `--stats-only` flag to unpack for stats-only processing.

2. fix(rooch-da): handle state root mismatch error in execution
    
    Add explicit error handling for cases where the execution state root does not match RoochNetwork. This ensures more accurate error reporting and avoids unintended behavior during L2 transaction execution.

Cargo.toml

@@ -182,6 +182,7 @@ eyre = "0.6.8"
 fastcrypto = { git = "https://github.com/MystenLabs/fastcrypto", rev = "56f6223b84ada922b6cb2c672c69db2ea3dc6a13" }
 futures = "0.3.31"
 futures-util = "0.3.31"
+hdrhistogram = "7.5.4"
 hex = "0.4.3"
 itertools = "0.13.0"
 jsonrpsee = { version = "0.23.2", features = ["full"] }

crates/rooch/Cargo.toml

@@ -34,6 +34,7 @@ async-trait = { workspace = true }
 codespan-reporting = { workspace = true }
 termcolor = { workspace = true }
 itertools = { workspace = true }
+hdrhistogram = { workspace = true }
 hex = { workspace = true }
 regex = { workspace = true }
 parking_lot = { workspace = true }

crates/rooch/src/commands/da/commands/exec.rs

@@ -465,6 +465,14 @@ impl ExecInner {
                 );
                 Err(error)
             } else {
+                // return error if is state root not equal to RoochNetwork
+                if error
+                    .to_string()
+                    .contains("Execution state root is not equal to RoochNetwork")
+                {
+                    return Err(error);
+                }
+
                 tracing::warn!(
                     "L2 Tx execution failed with a non-VM panic error. Ignoring and returning Ok; tx_order: {}, error: {:?}",
                     tx_order,

crates/rooch/src/commands/da/commands/unpack.rs

@@ -4,9 +4,11 @@
 use crate::commands::da::commands::{collect_chunks, get_tx_list_from_chunk};
 use clap::Parser;
 use rooch_types::error::RoochResult;
-use std::collections::{HashMap, HashSet};
+use std::collections::{BinaryHeap, HashMap, HashSet};
+
+use std::cmp::Reverse;
 use std::fs;
-use std::io::{BufRead, BufWriter, Write};
+use std::io::{BufWriter, Write};
 use std::path::PathBuf;
 
 /// Unpack batches to human-readable LedgerTransaction List from segments directory.
@@ -16,11 +18,8 @@ pub struct UnpackCommand {
     pub segment_dir: PathBuf,
     #[clap(long = "batch-dir")]
     pub batch_dir: PathBuf,
-    #[clap(
-        long = "verify-order",
-        help = "Verify the order of transactions for all batches"
-    )]
-    pub verify_order: bool,
+    #[clap(long = "stats-only", help = "Only print L2Tx size stats, no unpacking")]
+    pub stats_only: bool,
 }
 
 impl UnpackCommand {
@@ -30,11 +29,9 @@ impl UnpackCommand {
             chunks: Default::default(),
             segment_dir: self.segment_dir,
             batch_dir: self.batch_dir,
+            stats_only: self.stats_only,
         };
         unpacker.unpack()?;
-        if self.verify_order {
-            unpacker.verify_order()?;
-        }
 
         Ok(())
     }
@@ -45,48 +42,10 @@ struct UnpackInner {
     chunks: HashMap<u128, Vec<u64>>,
     segment_dir: PathBuf,
     batch_dir: PathBuf,
+    stats_only: bool,
 }
 
 impl UnpackInner {
-    fn verify_order(&self) -> anyhow::Result<()> {
-        let mut max_block_number = 0;
-        let mut last_tx_order = 0;
-        // start from block_number 0,
-        // read from batch_dir/<block_number> and verify the order of transactions, until no file found.
-        loop {
-            let batch_file_path = self.batch_dir.join(max_block_number.to_string());
-            if !batch_file_path.exists() {
-                break;
-            }
-
-            let file = fs::File::open(batch_file_path)?;
-            let reader = std::io::BufReader::new(file);
-            for line in reader.lines() {
-                let line = line?;
-                let tx: rooch_types::transaction::LedgerTransaction = serde_json::from_str(&line)?;
-                let tx_order = tx.sequence_info.tx_order;
-                if tx_order != last_tx_order + 1 {
-                    return Err(anyhow::anyhow!(
-                        "Transaction order is not strictly incremental for block {}: last_tx_order: {}, tx_order: {}",
-                        max_block_number, last_tx_order, tx_order
-                    ));
-                }
-                last_tx_order = tx_order;
-            }
-
-            if max_block_number % 1000 == 0 && max_block_number > 0 {
-                println!("Verified block: {}", max_block_number);
-            }
-
-            max_block_number += 1;
-        }
-        println!(
-            "All transactions are strictly incremental for blocks: [0, {}). last_tx_order: {}",
-            max_block_number, last_tx_order
-        );
-        Ok(())
-    }
-
     // batch_dir is a directory that stores all the unpacked batches.
     // each batch is stored in a file named by the block number (each batch maps to a block).
     // we collect all the block numbers to avoid unpacking the same batch multiple times.
@@ -126,11 +85,19 @@ impl UnpackInner {
     // unpack batches from segment_dir to batch_dir.
     // warn: ChunkV0 only in present
     fn unpack(&mut self) -> anyhow::Result<()> {
+        const TOP_N: usize = 20;
+
         self.collect_unpacked()?;
         self.collect_chunks()?;
 
         let mut new_unpacked = HashSet::new();
 
+        let mut l2tx_hist = TxStats {
+            hist: hdrhistogram::Histogram::<u64>::new_with_bounds(1, 4_096_000, 3)?,
+            tops: BinaryHeap::new(),
+            top_n: TOP_N,
+        };
+
         for (chunk_id, segment_numbers) in &self.chunks {
             if self.unpacked.contains(chunk_id) {
                 // For ChunkV0, chunk_id is block_number
@@ -143,6 +110,25 @@ impl UnpackInner {
                 segment_numbers.clone(),
             )?;
 
+            let mut last_tx_order = 0; // the first tx_order in DA is 1
+            for tx in &tx_list {
+                let tx_order = tx.sequence_info.tx_order;
+                if last_tx_order != 0 && tx_order != last_tx_order + 1 {
+                    return Err(anyhow::anyhow!(
+                            "Transaction order is not strictly incremental for block {}: last_tx_order: {}, tx_order: {}",
+                            chunk_id, last_tx_order, tx_order
+                        ));
+                }
+                last_tx_order = tx_order;
+                if let rooch_types::transaction::LedgerTxData::L2Tx(tx) = &tx.data {
+                    let tx_size = tx.tx_size();
+                    l2tx_hist.record(tx_order, tx_size)?;
+                }
+            }
+
+            if self.stats_only {
+                continue;
+            }
             // write LedgerTx in batch to file, each line is a tx in json
             let batch_file_path = self.batch_dir.join(chunk_id.to_string());
             let file = fs::OpenOptions::new()
@@ -163,6 +149,74 @@ impl UnpackInner {
         }
 
         println!("Unpacked batches(block_number): {:?}", new_unpacked);
+
+        l2tx_hist.print();
+
         Ok(())
     }
 }
+
+struct TxStats {
+    hist: hdrhistogram::Histogram<u64>,
+    tops: BinaryHeap<Reverse<(u64, u64)>>, // (tx_size, tx_order) Use Reverse to keep the smallest element at the top
+    top_n: usize,
+}
+
+impl TxStats {
+    fn record(&mut self, tx_order: u64, tx_size: u64) -> anyhow::Result<()> {
+        self.hist.record(tx_size)?;
+
+        if self.tops.len() < self.top_n {
+            // Add the new item directly if space is available
+            self.tops.push(Reverse((tx_size, tx_order)));
+        } else if let Some(&Reverse((smallest_size, _))) = self.tops.peek() {
+            // Compare with the smallest item in the heap
+            if tx_size > smallest_size {
+                self.tops.pop(); // Remove the smallest
+                self.tops.push(Reverse((tx_size, tx_order))); // Add the new larger item
+            }
+        }
+        // Keep only top-N
+        Ok(())
+    }
+
+    /// Returns the top N items, sorted by `tx_size` in descending order
+    pub fn get_top(&self) -> Vec<(u64, u64)> {
+        let mut sorted: Vec<_> = self.tops.iter().map(|&Reverse(x)| x).collect();
+        sorted.sort_by(|a, b| b.0.cmp(&a.0)); // Sort by tx_size in descending order
+        sorted
+    }
+
+    fn print(&mut self) {
+        let hist = &self.hist;
+
+        let min_size = hist.min();
+        let max_size = hist.max();
+        let mean_size = hist.mean();
+
+        println!("-----------------L2Tx Size Stats-----------------");
+        println!(
+            "Tx Size Percentiles distribution(count: {}): min={}, max={}, mean={:.2}, stdev={:.2}: ",
+            hist.len(),
+            min_size,
+            max_size,
+            mean_size,
+            hist.stdev()
+        );
+        let percentiles = [
+            1.00, 5.00, 10.00, 20.00, 30.00, 40.00, 50.00, 60.00, 70.00, 80.00, 90.00, 95.00,
+            99.00, 99.50, 99.90, 99.95, 99.99,
+        ];
+        for &p in &percentiles {
+            let v = hist.value_at_percentile(p);
+            println!("| {:6.2}th=[{}]", p, v);
+        }
+
+        // each pair one line
+        println!("-------------Top{} transactions--------------", self.top_n);
+        let tops = self.get_top();
+        for (tx_size, tx_order) in &tops {
+            println!("tx_order: {}, tx_size: {}", tx_order, tx_size);
+        }
+    }
+}