feat(l1): overlay + read cascade for deep reorgs (#6687)

> Part of the deep-reorg support effort tracked in #6685. This is PR 2
of 4, stacked on #6686. Adds the in-memory consumer of the journal; no
orchestration yet.

## Summary

- Adds `Overlay` struct in `crates/storage/layering.rs` that aggregates
reverse-diffs from `STATE_HISTORY[D..=T]` (descending, older-entry-wins)
to reconstruct virtual state at pivot `T-1`.
- Extends `TrieLayerCache` with optional overlay field and wires
`TrieWrapper::get` to cascade: layer cache -> overlay -> disk.
- 14 new unit tests; 30 storage tests total pass on both InMemory and
RocksDB backends.

## What's in this PR

- `Overlay` struct: walks the journal range descending, applies
older-entry-wins semantics, holds a bloom filter for fast miss
short-circuits.
- `OverlayCf` enum + `classify_by_key_length`: routes CF entries to the
right column family using the same key-length logic as PR 1's
`classify_trie_key`.
- `OverlayError::InvalidRange { from_block, to_block }`: hard error for
swapped caller args (not an I/O error variant).
- `TrieLayerCache` set/clear/lookup accessors for the overlay field.
- `TrieWrapper::get` cascade hook integrated into the existing read
path.
- 14 unit tests: descending-range load, older-wins, bloom miss, hash
mismatch, missing entry, swapped-args hard error,
single-entry-at-genesis, empty-but-present round trip, CF
classification, no-overlay short-circuit, set/clear, iter_all_entries.

## What's NOT in this PR

- Orchestration: `apply_fork_choice_with_deep_reorg`,
`install_overlay_for_reorg`, and the reorg-in-progress flag are in PR 3.
- The 128-block reorg cap is not lifted; that's PR 4.
- `from_block` accessor rename (triggers clippy `wrong_self_convention`,
suppressed); deferred to PR 3.
- Swapping `AtomicBloomFilter` for a non-atomic variant during
construction for perf; deferred to PR 3 or PR 4.

## Test plan

- `cargo fmt` clean.
- `cargo clippy` clean.
- 14 new tests added in `crates/storage/layering.rs`.
- `cargo test -p ethrex-storage` passes all 30 tests on both InMemory
and RocksDB backends.

## References

- Parent issue: #6685
- PR 1 (state history journal): #6686

---------

Co-authored-by: ilitteri <ivanlitteri@hotmail.com>
Co-authored-by: Ivan Litteri <67517699+ilitteri@users.noreply.github.com>

Author: edg-l

.github/workflows/deep_reorg.yaml

@@ -0,0 +1,70 @@
+name: Deep reorg (manual)
+
+# Manual trigger only -- this scenario takes ~1 hour and requires a Kurtosis
+# cluster. It is NOT a PR gate.
+on:
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+env:
+  CARGO_NET_GIT_FETCH_WITH_CLI: "true"
+  CARGO_NET_RETRY: "10"
+
+jobs:
+  deep-reorg:
+    name: Deep reorg partition test
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout sources
+        uses: actions/checkout@v6
+
+      - name: Free Disk Space
+        uses: ./.github/actions/free-disk
+
+      - name: Setup Rust Environment
+        uses: ./.github/actions/setup-rust
+
+      - name: Build ethrex (release, rocksdb + metrics)
+        run: |
+          cargo build --release --bin ethrex --features rocksdb,metrics
+
+      - name: Build ethrex Docker image
+        run: |
+          docker build -t ethrex:local .
+
+      - name: Install Kurtosis CLI
+        run: |
+          echo "deb [trusted=yes] https://apt.fury.io/kurtosis-tech/ /" \
+            | sudo tee /etc/apt/sources.list.d/kurtosis.list
+          sudo apt-get update
+          sudo apt-get install -y kurtosis-cli
+          kurtosis version
+
+      - name: Start Kurtosis engine
+        run: kurtosis engine start
+
+      - name: Run deep-reorg enclave
+        id: kurtosis-run
+        run: |
+          kurtosis run --enclave deep-reorg \
+            github.com/ethpandaops/ethereum-package \
+            --args-file tooling/reorgs/disruptoor/network-partition-deep-reorg.yaml
+
+      - name: Collect enclave logs on failure
+        if: failure()
+        run: |
+          kurtosis enclave dump deep-reorg /tmp/enclave-logs || true
+
+      - name: Upload enclave logs as artifact
+        if: failure()
+        uses: actions/upload-artifact@v6
+        with:
+          name: deep-reorg-enclave-logs
+          path: /tmp/enclave-logs
+          if-no-files-found: ignore
+
+      - name: Tear down enclave
+        if: always()
+        run: kurtosis enclave rm deep-reorg --force || true

Cargo.lock

@@ -460,6 +460,14 @@ pub struct Options {
         env = "ETHREX_PRECOMPUTE_WITNESSES"
     )]
     pub precompute_witnesses: bool,
+    #[arg(
+        long = "max-reorg-depth",
+        value_name = "MAX_REORG_DEPTH",
+        help = "Optional operator override for the maximum reorg depth. Omit for finality-bounded cap. Set to 0 to disable deep reorgs entirely. Set to d to reject reorgs of depth > d.",
+        help_heading = "Node options",
+        env = "ETHREX_MAX_REORG_DEPTH"
+    )]
+    pub max_reorg_depth: Option<u64>,
 }
 
 impl Options {
@@ -554,6 +562,7 @@ impl Default for Options {
             no_bal_parallel_exec: false,
             no_bal_prefetch: false,
             no_bal_parallel_trie: false,
+            max_reorg_depth: None,
         }
     }
 }

cmd/ethrex/cli.rs

@@ -784,6 +784,7 @@ pub async fn init_l1(
             bal_parallel_exec_enabled: !opts.no_bal_parallel_exec,
             bal_prefetch_enabled: !opts.no_bal_prefetch,
             bal_parallel_trie_enabled: !opts.no_bal_parallel_trie,
+            max_reorg_depth: opts.max_reorg_depth,
             gap_admit_occupancy_threshold: opts.mempool_gap_admit_occupancy_threshold,
         },
     );

cmd/ethrex/initializers.rs

@@ -494,6 +494,7 @@ impl Command {
                         latest_hash_on_batch,
                         latest_hash_on_batch,
                         latest_hash_on_batch,
+                        None,
                     )
                     .await?;
 

cmd/ethrex/l2/command.rs

@@ -256,6 +256,7 @@ pub async fn init_l2(
         bal_parallel_exec_enabled: true,
         bal_prefetch_enabled: true,
         bal_parallel_trie_enabled: true,
+        max_reorg_depth: opts.node_opts.max_reorg_depth,
         gap_admit_occupancy_threshold: opts.node_opts.mempool_gap_admit_occupancy_threshold,
     };
 

cmd/ethrex/l2/initializers.rs

@@ -36,6 +36,7 @@ tokio-util.workspace = true
 [dev-dependencies]
 tokio = { workspace = true, features = ["full"] }
 ethrex-levm.workspace = true
+tempfile.workspace = true
 
 [lib]
 path = "./blockchain.rs"
@@ -45,7 +46,7 @@ default = ["secp256k1"]
 rayon = ["ethrex-vm/rayon"]
 secp256k1 = ["ethrex-common/secp256k1", "ethrex-vm/secp256k1", "rayon"]
 c-kzg = ["ethrex-common/c-kzg", "ethrex-vm/c-kzg"]
-metrics = ["ethrex-metrics/transactions"]
+metrics = ["ethrex-metrics/transactions", "ethrex-metrics/metrics"]
 eip-8025 = ["ethrex-common/eip-8025"]
 testing = []
 

crates/blockchain/Cargo.toml

@@ -224,6 +224,12 @@ pub struct Blockchain {
     /// Set to true after initial sync completes, never reset to false.
     /// Does not reflect whether an ongoing sync is in progress.
     is_synced: AtomicBool,
+    /// Set while a deep-reorg apply pass is in flight. Concurrent
+    /// FCUs from the engine API short-circuit to SYNCING while this is set,
+    /// and journal pruning in `forkchoice_update_inner` defers until the apply
+    /// pass clears it. Managed via [`enter_reorg`](Self::enter_reorg) which
+    /// returns a [`ReorgGuard`] RAII that clears the flag on drop.
+    reorg_in_progress: AtomicBool,
     /// Configuration options for blockchain behavior.
     pub options: BlockchainOptions,
     /// Cache of recently built payloads.
@@ -273,6 +279,22 @@ impl std::fmt::Debug for PrewarmedCache {
     }
 }
 
+/// RAII guard that clears [`Blockchain::reorg_in_progress`] on drop. Returned by
+/// [`Blockchain::enter_reorg`]. Ensures the flag is reset on every exit path of
+/// the apply pass, including panics that unwind through the guard.
+pub struct ReorgGuard<'a> {
+    blockchain: &'a Blockchain,
+}
+
+impl Drop for ReorgGuard<'_> {
+    fn drop(&mut self) {
+        self.blockchain.storage.set_journal_pruning_paused(false);
+        self.blockchain
+            .reorg_in_progress
+            .store(false, Ordering::Release);
+    }
+}
+
 /// Configuration options for the blockchain.
 #[derive(Debug, Clone)]
 pub struct BlockchainOptions {
@@ -311,6 +333,13 @@ pub struct BlockchainOptions {
     /// `--no-bal-parallel-trie`) to fall back to streaming `AccountUpdate`s from
     /// the executor and merkleizing post-execution.
     pub bal_parallel_trie_enabled: bool,
+    /// Optional operator override for the maximum reorg depth. `None` ; cap is purely
+    /// physical (layer-cache retention plus journal reach; bounded indirectly by finality
+    /// because finality advances prune the journal). `Some(d)` ; reject reorgs of
+    /// depth `> d` with `-38006 TooDeepReorg`. WARNING: `Some(0)` rejects EVERY reorg,
+    /// including routine 1-2 block reorgs on a healthy network — it is NOT the old
+    /// 128-block cap. Use a small positive value to approximate the pre-stack behavior.
+    pub max_reorg_depth: Option<u64>,
     /// Mempool occupancy percentage (0-100) at or above which incoming
     /// transactions with a nonce gap relative to the sender's on-chain nonce
     /// are rejected. Setting to 100 disables the check.
@@ -330,6 +359,7 @@ impl Default for BlockchainOptions {
             bal_parallel_exec_enabled: true,
             bal_prefetch_enabled: true,
             bal_parallel_trie_enabled: true,
+            max_reorg_depth: None,
             gap_admit_occupancy_threshold: DEFAULT_GAP_ADMIT_OCCUPANCY_THRESHOLD,
         }
     }
@@ -438,6 +468,7 @@ impl Blockchain {
             storage: store,
             mempool: Mempool::new(blockchain_opts.max_mempool_size),
             is_synced: AtomicBool::new(false),
+            reorg_in_progress: AtomicBool::new(false),
             payloads: Arc::new(TokioMutex::new(Vec::new())),
             options: blockchain_opts,
             merkle_pool: Self::build_merkle_pool(),
@@ -459,6 +490,7 @@ impl Blockchain {
             storage: store,
             mempool: Mempool::new(MAX_MEMPOOL_SIZE_DEFAULT),
             is_synced: AtomicBool::new(false),
+            reorg_in_progress: AtomicBool::new(false),
             payloads: Arc::new(TokioMutex::new(Vec::new())),
             options: BlockchainOptions::default(),
             merkle_pool: pool,
@@ -471,6 +503,7 @@ impl Blockchain {
             storage: store,
             mempool: Mempool::new(MAX_MEMPOOL_SIZE_DEFAULT),
             is_synced: AtomicBool::new(false),
+            reorg_in_progress: AtomicBool::new(false),
             payloads: Arc::new(TokioMutex::new(Vec::new())),
             options: BlockchainOptions::default(),
             merkle_pool: Self::build_merkle_pool(),
@@ -498,6 +531,41 @@ impl Blockchain {
         Ok(())
     }
 
+    /// Returns a reference to the underlying [`Store`]. Used by the deep-reorg
+    /// orchestrator to drive the storage-side primitives.
+    pub fn store(&self) -> &Store {
+        &self.storage
+    }
+
+    /// Returns `true` while a deep-reorg apply pass is in flight. Set by
+    /// [`enter_reorg`](Self::enter_reorg); the engine API's FCU handler should
+    /// short-circuit to SYNCING when this is `true`.
+    pub fn is_reorg_in_progress(&self) -> bool {
+        self.reorg_in_progress.load(Ordering::Acquire)
+    }
+
+    /// Attempts to mark the start of a deep-reorg apply pass. Returns a RAII
+    /// guard that clears the flag on drop (success, error, or panic) if no other
+    /// pass is in flight, or `None` if one already is. This is a test-and-set:
+    /// the flag transition `false -> true` is atomic, so two concurrent FCUs
+    /// that both reach the deep path can never both acquire the guard. The loser
+    /// short-circuits to SYNCING and the CL retries. At most one apply pass runs
+    /// at a time.
+    pub fn enter_reorg(&self) -> Option<ReorgGuard<'_>> {
+        self.reorg_in_progress
+            .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
+            .is_ok()
+            .then(|| {
+                // Also pause STATE_HISTORY pruning for the whole pass:
+                // `Overlay::from_journal` reads entries with no snapshot
+                // isolation, and syncer-driven forkchoice updates bypass this
+                // mutex, so a concurrent finality advance could otherwise prune
+                // the journal out from under overlay construction.
+                self.storage.set_journal_pruning_paused(true);
+                ReorgGuard { blockchain: self }
+            })
+    }
+
     /// Executes a block withing a new vm instance and state
     fn execute_block(
         &self,
@@ -4452,3 +4520,58 @@ pub fn compute_sharded_storage_root(
 
     Ok((root_hash.finalize(&NativeCrypto), nodes))
 }
+
+#[cfg(test)]
+mod reorg_guard_tests {
+    use super::*;
+    use ethrex_storage::{EngineType, Store};
+
+    fn make_blockchain() -> Blockchain {
+        let dir = tempfile::tempdir().unwrap();
+        let store = Store::new(dir.path(), EngineType::InMemory).unwrap();
+        Blockchain::default_with_store(store)
+    }
+
+    /// `enter_reorg` SHALL set the flag; the returned guard SHALL clear it on drop.
+    #[test]
+    fn reorg_guard_sets_and_clears_flag() {
+        let blockchain = make_blockchain();
+        assert!(!blockchain.is_reorg_in_progress(), "flag starts false");
+
+        {
+            let guard = blockchain.enter_reorg();
+            assert!(guard.is_some(), "first enter_reorg must acquire the guard");
+            assert!(
+                blockchain.is_reorg_in_progress(),
+                "flag must be set while guard is alive"
+            );
+            // A second attempt while the first guard is alive must fail the
+            // test-and-set and return None.
+            assert!(
+                blockchain.enter_reorg().is_none(),
+                "concurrent enter_reorg must not acquire a second guard"
+            );
+        }
+        assert!(
+            !blockchain.is_reorg_in_progress(),
+            "flag must clear when guard drops"
+        );
+    }
+
+    /// The guard SHALL clear the flag even when its scope ends via panic unwinding.
+    #[test]
+    fn reorg_guard_clears_flag_on_panic() {
+        let blockchain = make_blockchain();
+        let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+            let _guard = blockchain.enter_reorg();
+            assert!(blockchain.is_reorg_in_progress());
+            panic!("simulated apply failure");
+        }));
+        // guard is dropped by the unwind; flag must be clear below.
+        assert!(result.is_err(), "panic must propagate out of the scope");
+        assert!(
+            !blockchain.is_reorg_in_progress(),
+            "flag must be cleared after a panicking guard scope"
+        );
+    }
+}