Let an EL-enabled node survive a restart instead of dropping every block

Block import is gated on execution, so an execution layer that cannot extend the
consensus head is not degraded — it is terminal. Every arriving block names a
parent it does not have, fails, and is dropped before the store sees it. With the
execution store in memory, that was the state of every restarted node: consensus
resumed at its old slot, execution rewound to genesis, and the node never followed
the chain again. Checkpoint sync made it worse by moving the consensus head
further ahead.

Two parts. The execution store now persists in RocksDB under --data-dir/el, on the
same volume as the consensus store so it survives container recreation rather than
just a restart. And el_sync replays, at startup, whatever the execution layer is
missing: walk back from the consensus head to the last block it can actually
extend, then execute forward. The Lean chain carries every ExecutionPayloadV3 in
its block bodies, so it already is a complete execution history — no new wire
protocol and no peers needed.

Persistence turned out not to be sufficient on its own, which is worth recording:
a reopened datadir keeps the block index and canonical head but NOT executable
state. So the head is present and unbuildable, and the first attempt at gap
detection — "do I have this block?" — answered yes and concluded there was nothing
to do. The predicate is now can_build_on: the block is present AND its post-state
is reachable, which is the same condition ethrex checks before accepting a fork
choice update, and therefore the exact question build_payload will later ask.
Measured on a devnet restart: 44 payloads replayed in 8ms.

Also fixes a latent corruption risk found while testing this. The value seeded
into the consensus genesis anchor was read from head_hash(), which equals the
genesis hash only on a fresh store. With execution state persisted, a node whose
consensus datadir was wiped would have anchored a newly created genesis state to
whatever block its execution layer last executed, producing a chain no peer
agrees with — silently. Added genesis_hash() and used that.

start_p2p now runs after the resync rather than before. It calls set_synced(),
which is what unlocks inbound transaction gossip, and that should not happen while
the execution layer is still behind the chain.

The devnet harness gains --restart-node N. It stops a node, waits out the
gossipsub backoff, starts it, and then asserts what a node on a private fork
cannot fake: blocks imported from peers after the restart, and no synthetic-payload
fallbacks. The first version of this check asserted that the head slot advanced
and that the node was level with its peers — both of which a forked node satisfies
exactly, since a fork advances at the same rate. It passed while the node was
importing nothing. Head slot is now reported rather than asserted. Per-node checks
also count nodes instead of log lines, since a restart logs startup twice and
produced totals like "4/3 nodes".

Verified: 585 tests pass, three new, fmt and clippy -D warnings clean. On a 3-node
devnet, node 2 was stopped at slot 54 and restarted: it replayed 44 payloads,
imported 16 blocks from peers, built payloads with no synthetic fallbacks, and
reached slot 84 level with its peers. The chain finalized at slot 75 throughout,
with cross-node transaction gossip still working.

Author: pablodeymo

Cargo.lock

@@ -207,23 +207,72 @@ async fn main() -> eyre::Result<()> {
     let (execution_engine, el_genesis_hash) = match options.el_genesis.as_deref() {
         None => (None, None),
         Some(path) => {
-            let engine = EthrexEngine::from_genesis_path(path)
+            // Execution state lives beside the consensus store, so a restarted
+            // node keeps the blocks it has already executed. Without this it
+            // rewinds to genesis while consensus resumes at its old slot, and
+            // since block import is gated on execution, the node then drops
+            // every block it receives — permanently.
+            let el_store_dir = options.data_dir.join("el");
+            let engine = EthrexEngine::from_genesis_path(path, Some(el_store_dir.as_path()))
                 .await
                 .map_err(|err| eyre::eyre!("failed to bootstrap embedded ethrex: {err}"))?;
-            let hash = engine
-                .head_hash()
+            // The *genesis* hash, not the head. They coincide only on a fresh
+            // store; with execution state persisted, the head after a restart is
+            // whatever block was last executed, and anchoring a newly created
+            // consensus genesis to that would silently build a chain no peer
+            // agrees with.
+            let genesis_hash = engine
+                .genesis_hash()
                 .await
                 .map_err(|err| eyre::eyre!("failed to read EL genesis block hash: {err}"))?;
-            info!(genesis = %path.display(), el_genesis_hash = %hash, "Embedded ethrex enabled");
-            (Some(Arc::new(engine)), Some(hash))
+            let head = engine
+                .head_number()
+                .await
+                .map_err(|err| eyre::eyre!("failed to read EL head: {err}"))?;
+            info!(
+                genesis = %path.display(),
+                el_genesis_hash = %genesis_hash,
+                el_head_block = head,
+                "Embedded ethrex enabled"
+            );
+            (Some(Arc::new(engine)), Some(genesis_hash))
         }
     };
 
+    let clean_checkpoint_urls: Vec<String> = options
+        .checkpoint_sync_url
+        .into_iter()
+        .map(|url| url.trim().to_string())
+        .filter(|url| !url.is_empty())
+        .collect();
+
+    let store = fetch_initial_state(
+        &clean_checkpoint_urls,
+        &genesis_config,
+        backend.clone(),
+        el_genesis_hash,
+    )
+    .await
+    .inspect_err(|err| error!(%err, "Failed to initialize state"))?;
+
+    // Catch the execution layer up to the consensus chain before anything else
+    // runs. On a fresh genesis this is a no-op; on a restart it replays whatever
+    // the persistent execution store is missing. It has to happen before the
+    // blockchain actor starts, because the actor's first tick may propose a block
+    // and would build on the wrong parent.
+    if let Some(engine) = execution_engine.as_ref() {
+        ethlambda_blockchain::el_sync::resync_execution_layer(&store, engine).await;
+    }
+
     // Join the execution layers into their own transaction-gossip mesh. Without
     // it each mempool is isolated, so a submitted transaction waits for the turn
     // of the node that received it; with it, whichever node proposes next can
     // include it. Independent of consensus gossip in every respect — own key,
     // own port, own peer set.
+    //
+    // After the resync, deliberately: joining marks the execution layer as
+    // synced, which is what unlocks inbound transaction gossip, and that should
+    // not happen while it is still behind the chain.
     if let (Some(engine), Some(port)) = (execution_engine.as_ref(), options.el_p2p_port) {
         let mut el_p2p = P2PConfig::loopback(derive_el_node_key(&node_p2p_key), port);
         el_p2p.bootnodes = options.el_bootnodes.clone();
@@ -236,22 +285,6 @@ async fn main() -> eyre::Result<()> {
         }
     }
 
-    let clean_checkpoint_urls: Vec<String> = options
-        .checkpoint_sync_url
-        .into_iter()
-        .map(|url| url.trim().to_string())
-        .filter(|url| !url.is_empty())
-        .collect();
-
-    let store = fetch_initial_state(
-        &clean_checkpoint_urls,
-        &genesis_config,
-        backend.clone(),
-        el_genesis_hash,
-    )
-    .await
-    .inspect_err(|err| error!(%err, "Failed to initialize state"))?;
-
     let validator_ids: Vec<u64> = validator_keys.keys().copied().collect();
 
     // Shared, runtime-mutable aggregator flag. Seeded from the CLI and

bin/ethlambda/src/main.rs

@@ -0,0 +1,164 @@
+//! Bring a restarted node's execution layer back in step with the consensus
+//! chain, by replaying the payloads it has not executed yet.
+//!
+//! # Why this exists
+//!
+//! Block import is gated on execution: `import_gossiped_block` drops any block
+//! whose payload the execution layer rejects. So an execution layer that is
+//! behind the consensus chain is not a degraded state — it is terminal. Every
+//! arriving block names a parent the execution layer does not have, fails with
+//! `ParentNotFound`, and is dropped before the store sees it. The node stops
+//! following the chain permanently, and no amount of checkpoint syncing helps:
+//! that moves the consensus head *further* ahead, widening the gap.
+//!
+//! Persisting execution state alongside the consensus store removes the common
+//! cause (a restart), but not every cause: the two stores are written
+//! independently, so a crash between them, an operator wiping one, or a
+//! consensus store restored from a checkpoint can all leave a gap.
+//!
+//! # How
+//!
+//! The Lean chain already contains every `ExecutionPayloadV3` in its block
+//! bodies, so it *is* a complete execution-layer history — no new wire protocol
+//! and no peers needed. Walk back from the consensus head until reaching a block
+//! the execution layer already has, then replay forward from there.
+//!
+//! The cost is bounded by the size of the gap, not the length of the chain: a
+//! node whose execution state is current walks back exactly one block and
+//! replays nothing.
+
+use ethlambda_ethrex_engine::EthrexEngine;
+use ethlambda_storage::Store;
+use ethlambda_types::{ShortRoot, block::Block};
+use tracing::{info, warn};
+
+/// Outcome of a resync attempt, for logging and tests.
+#[derive(Debug, Default, PartialEq, Eq)]
+pub struct ResyncReport {
+    /// Payloads executed to close the gap.
+    pub replayed: usize,
+    /// Blocks skipped because their payload carries no execution-layer block —
+    /// the pass-through shape a proposer emits when its own build failed.
+    pub skipped: usize,
+    /// True when the walk ran out of consensus blocks before reaching one the
+    /// execution layer had, so the gap could not be closed.
+    pub incomplete: bool,
+}
+
+/// Replay the payloads between the execution layer's head and the consensus
+/// head.
+///
+/// Returns what it did. Never fails the caller: a node that cannot close the gap
+/// should start and say so loudly rather than refuse to boot, since the operator
+/// may be deliberately reusing a datadir.
+pub async fn resync_execution_layer(store: &Store, engine: &EthrexEngine) -> ResyncReport {
+    let mut report = ResyncReport::default();
+
+    let head_root = match store.head() {
+        Ok(root) if !root.is_zero() => root,
+        // No consensus head yet (fresh genesis): nothing to replay.
+        _ => return report,
+    };
+
+    // Walk back along parent links, collecting blocks the execution layer is
+    // missing. Stops at the first block it already has, which is the common
+    // ancestor of the two views.
+    let mut missing: Vec<Block> = Vec::new();
+    let mut root = head_root;
+    loop {
+        let block = match store.get_block(&root) {
+            Ok(Some(block)) => block,
+            Ok(None) | Err(_) => {
+                // Ran out of history without meeting the execution layer. Only
+                // reachable when the consensus store does not go back far enough
+                // — a checkpoint-synced node, or a partially pruned datadir.
+                report.incomplete = true;
+                break;
+            }
+        };
+
+        let payload_hash = block.body.execution_payload.block_hash;
+        match engine.can_build_on(payload_hash).await {
+            Ok(true) => break,
+            Ok(false) => {}
+            Err(err) => {
+                warn!(%err, "Could not query the execution layer; skipping resync");
+                return report;
+            }
+        }
+
+        let parent_root = block.parent_root;
+        missing.push(block);
+        if parent_root.is_zero() {
+            // Reached the anchor. Its payload is the EL genesis, which the
+            // engine always has, so this means the chain does not link back to
+            // the execution layer's genesis at all.
+            report.incomplete = true;
+            break;
+        }
+        root = parent_root;
+    }
+
+    if missing.is_empty() && !report.incomplete {
+        // Logged even though there is nothing to do. "No output" is the same
+        // thing an accidentally-skipped resync produces, and telling those two
+        // apart from a log file afterwards is worth one line at startup.
+        info!(
+            head_slot = store.head_slot(),
+            "Execution layer is in step with the consensus chain"
+        );
+        return report;
+    }
+
+    info!(
+        blocks = missing.len(),
+        head_slot = store.head_slot(),
+        "Execution layer is behind the consensus chain; replaying payloads"
+    );
+
+    // Oldest first: each payload extends the one before it.
+    for block in missing.iter().rev() {
+        let payload = &block.body.execution_payload;
+
+        // A pass-through payload repeats its parent hash instead of naming a new
+        // execution-layer block: the proposer's build failed and no block was
+        // produced. There is nothing to execute, and trying would fail the
+        // block-hash check.
+        if payload.block_hash == payload.parent_hash {
+            report.skipped += 1;
+            continue;
+        }
+
+        if let Err(err) = engine.execute_payload(payload, block.parent_root) {
+            // Stop at the first failure rather than pressing on: every later
+            // payload builds on this one, so they would all fail too and bury
+            // the real error in noise.
+            warn!(
+                slot = block.slot,
+                block_root = %ShortRoot(&payload.block_hash.0),
+                %err,
+                "Replay failed; execution layer left behind the consensus chain"
+            );
+            report.incomplete = true;
+            return report;
+        }
+        report.replayed += 1;
+    }
+
+    if report.incomplete {
+        warn!(
+            replayed = report.replayed,
+            "Could not fully resync the execution layer from the consensus chain. \
+             This node will drop every block it receives. Wipe its data directory \
+             and restart to rebuild from genesis."
+        );
+    } else {
+        info!(
+            replayed = report.replayed,
+            skipped = report.skipped,
+            "Execution layer resynced with the consensus chain"
+        );
+    }
+
+    report
+}

crates/blockchain/src/el_sync.rs

@@ -40,6 +40,7 @@ pub mod aggregation;
 pub mod block_builder;
 pub(crate) mod coverage;
 mod el_integration;
+pub mod el_sync;
 pub mod events;
 pub(crate) mod fork_choice_tree;
 pub mod key_manager;

crates/blockchain/src/lib.rs

@@ -7,7 +7,7 @@ rust-version.workspace = true
 
 [dependencies]
 ethrex-common.workspace = true
-ethrex-storage.workspace = true
+ethrex-storage = { workspace = true, features = ["rocksdb"] }
 # `c-kzg` gates the blob-transaction mempool API. It already arrives through
 # ethrex-p2p's default features, but naming it here keeps `cargo check -p
 # ethlambda-ethrex-engine` and any --no-default-features build honest.

crates/net/ethrex-engine/Cargo.toml

@@ -100,24 +100,54 @@ impl EthrexEngine {
     /// The genesis must be **Cancun**: a Prague genesis makes ethrex require a
     /// `requests_hash` in the block header that the Cancun-shaped
     /// [`ExecutionPayloadV3`] cannot carry, and every payload is then rejected.
-    pub async fn from_genesis_path(path: impl AsRef<std::path::Path>) -> Result<Self, EngineError> {
+    ///
+    /// `store_dir` selects where execution state lives. `Some(dir)` persists it
+    /// in RocksDB, so a restarted node keeps the blocks it has already executed;
+    /// `None` keeps it in memory, which is what tests want. A persistent store
+    /// re-reads its own genesis on reopen and rejects a *different* one, so the
+    /// directory doubles as a genesis fingerprint.
+    pub async fn from_genesis_path(
+        path: impl AsRef<std::path::Path>,
+        store_dir: Option<&std::path::Path>,
+    ) -> Result<Self, EngineError> {
         let path = path.as_ref();
         let file = std::fs::File::open(path)
             .map_err(|err| EngineError::GenesisLoad(format!("open {}: {err}", path.display())))?;
         let genesis: Genesis = serde_json::from_reader(std::io::BufReader::new(file))
             .map_err(|err| EngineError::GenesisLoad(format!("parse {}: {err}", path.display())))?;
-        Self::from_genesis(genesis).await
+        Self::build(genesis, store_dir).await
     }
 
-    /// Bootstrap an engine with an in-memory store initialised from `genesis`.
+    /// Bootstrap an engine with an **in-memory** store initialised from
+    /// `genesis`. Execution state does not survive the process.
     pub async fn from_genesis(genesis: Genesis) -> Result<Self, EngineError> {
+        Self::build(genesis, None).await
+    }
+
+    async fn build(
+        genesis: Genesis,
+        store_dir: Option<&std::path::Path>,
+    ) -> Result<Self, EngineError> {
         // Load the KZG trusted setup on a background thread. It is compiled in,
         // so there is no file to find, but the first blob verification would
         // otherwise pay a multi-second initialisation — and the call that
         // triggers it could be a payload build at interval 4.
         ethrex_crypto::kzg::warm_up_trusted_setup();
 
-        let mut store = Store::new("", EngineType::InMemory)?;
+        let mut store = match store_dir {
+            Some(dir) => {
+                std::fs::create_dir_all(dir).map_err(|err| {
+                    EngineError::Store(StoreError::Custom(format!(
+                        "create EL store dir {}: {err}",
+                        dir.display()
+                    )))
+                })?;
+                Store::new(dir.to_string_lossy().as_ref(), EngineType::RocksDB)?
+            }
+            None => Store::new("", EngineType::InMemory)?,
+        };
+        // Idempotent: on an existing datadir this recognises its own genesis and
+        // returns, and rejects a mismatched one rather than corrupting the chain.
         store.add_initial_state(genesis).await?;
         let blockchain = Arc::new(Blockchain::default_with_store(store.clone()));
         Ok(Self {
@@ -129,6 +159,41 @@ impl EthrexEngine {
         })
     }
 
+    /// Whether the execution layer can extend the block with this hash — it has
+    /// the block *and* the block's post-state is reachable in the database.
+    ///
+    /// Both halves matter, and testing only the first is a trap. An unclean
+    /// shutdown can leave a block's header durable while its state trie is not,
+    /// and such a block looks present while every attempt to build on it fails
+    /// with `StateNotReachable`. This is the same condition ethrex checks before
+    /// accepting a fork-choice update, so it answers exactly the question
+    /// [`Self::build_payload`] will later ask.
+    ///
+    /// Used to find where a restarted node's execution state effectively stopped,
+    /// so only the unusable payloads are replayed.
+    pub async fn can_build_on(&self, hash: LeanH256) -> Result<bool, EngineError> {
+        let Some(header) = self.store.get_block_header_by_hash(H256(hash.0))? else {
+            return Ok(false);
+        };
+        Ok(self.store.has_state_root(header.state_root)?)
+    }
+
+    /// Hash of the execution layer's **genesis** block.
+    ///
+    /// Distinct from [`Self::head_hash`], which is only the same thing on a fresh
+    /// store. This is what seeds the consensus genesis anchor, so a node with a
+    /// persistent execution store must not use the head: after a restart that is
+    /// whatever block it last executed, and anchoring a genesis state to it would
+    /// silently produce a chain no peer agrees with.
+    pub async fn genesis_hash(&self) -> Result<LeanH256, EngineError> {
+        let hash = self
+            .store
+            .get_block_header(0)?
+            .ok_or(EngineError::NoCanonicalHead)?
+            .hash();
+        Ok(LeanH256(hash.0))
+    }
+
     /// Hash of the current canonical head block.
     ///
     /// Immediately after [`Self::from_genesis`] this is the EL genesis block