From e0f69b32c6e0c6749950bf6927064dc49145c18d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Tue, 3 Nov 2026 18:07:45 -0300 Subject: [PATCH 1/2] feat(beacon): rebuild held blocks' data columns from the execution client's blob pool A fulu block whose custody columns have not all arrived waits on gossip and on per-slot peer asks. The blob transactions it commits to usually passed through the local execution client's pool first, so the node can often build those columns itself, sooner. Fulu's p2p-interface asks honest nodes to do so with engine_getBlobsV2, as Lighthouse and the other clients already do. On a fulu block's first data-availability hold, with an execution client configured, the chain actor spawns a task that asks engine_getBlobsV2 once (1 s, the spec's timeout, outside the retry ladder: the call races gossip), builds the custody columns still missing without verifying the KZG proofs (the spec says to skip verification for local-EL blobs), and hands them to p2p through the existing check_data_column_sidecars, now tagged with a SidecarOrigin. For ExecutionLayer sidecars p2p runs no checks; it records them in the seen cache, publishes them on their subnets, and forwards them to the chain actor, which stores them and releases the block. Those two steps are the spec's MUSTs for importing local-EL blobs, and p2p owns both. V2 only: V3 answers partially, and a column needs a cell from every blob. Gloas is not covered, since availability moves to the payload envelope. --- CLAUDE.md | 2 +- crates/blockchain/src/beacon_blobs.rs | 479 ++++++++++++++++++ crates/blockchain/src/beacon_columns.rs | 3 +- crates/blockchain/src/lib.rs | 29 ++ crates/blockchain/src/metrics.rs | 52 ++ .../src/beacon/data_columns.rs | 239 +++++++-- crates/net/api/src/lib.rs | 37 +- crates/net/engine/src/building.rs | 4 +- crates/net/engine/src/client.rs | 42 +- crates/net/engine/src/lib.rs | 15 +- crates/net/engine/src/types.rs | 115 ++++- crates/net/engine/tests/wire_smoke.rs | 85 +++- crates/net/p2p/src/beacon/engine_columns.rs | 209 ++++++++ crates/net/p2p/src/beacon/mod.rs | 1 + crates/net/p2p/src/lib.rs | 15 +- crates/net/p2p/src/metrics.rs | 22 + docs/beacon_engine.md | 94 +++- docs/metrics.md | 11 + 18 files changed, 1381 insertions(+), 73 deletions(-) create mode 100644 crates/blockchain/src/beacon_blobs.rs create mode 100644 crates/net/p2p/src/beacon/engine_columns.rs diff --git a/CLAUDE.md b/CLAUDE.md index a171704a8..5c33f6a61 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -337,7 +337,7 @@ actual_slot = finalized_slot + 1 + relative_index - Mesh size: 8 (6-12 bounds), heartbeat: 700ms - Beacon wire: `validate_messages()` is on, so every beacon message waits for a verdict (~4.2s before gossipsub's cache evicts it). Rules in `state_transition::beacon::gossip` (cheap half inline, stateful half on a bounded `spawn_blocking` task); plumbing in `p2p/src/beacon/verdict.rs`. Lean gossip still auto-forwards - Beacon peer scoring: lighthouse's parameters ported in `p2p/src/beacon/scoring.rs` (block, aggregate, 64 attestation subnets, exits, slashings; columns and sync contributions unscored), refreshed every slot from the head's active validator count. Mesh-delivery scoring (P3) is gated off while the head lags more than 4 slots and for an epoch after, because a catching-up node `Ignore`s every aggregate and would otherwise graylist its own mesh. Every held fork digest is parametrized before `transition::apply` joins it (gossipsub never scores a mesh grafted before its topic had parameters); the previous digest's weights are zeroed at the switch, and the switch restarts the P3 warmup. The swarm disconnects peers below the graylist (-16000) every 10 s. Lean runs unscored. See [`docs/beacon_wire.md`](docs/beacon_wire.md#peer-scoring) - - Data columns: every check runs in p2p. A column gossip did not accept (`Queue`/`Overloaded`), every fetched column, and parked columns replayed after their parent imports go through `column::chain_checks` in `p2p/src/beacon/column_checks.rs`. The chain actor stores what it gets unchecked; only debug builds re-run `chain_checks` there + - Data columns: every check runs in p2p. A column gossip did not accept (`Queue`/`Overloaded`), every fetched column, and parked columns replayed after their parent imports go through `column::chain_checks` in `p2p/src/beacon/column_checks.rs`. The chain actor stores what it gets unchecked; only debug builds re-run `chain_checks` there. A held fulu block's custody columns can also come from the execution client (`engine_getBlobsV2`, `crates/blockchain/src/beacon_blobs.rs`), built without KZG verification and handed to p2p as `SidecarOrigin::ExecutionLayer` through `check_data_column_sidecars`, which records them seen, publishes them on their subnets and forwards them; see `docs/beacon_engine.md` - Beacon subscribes seven global topics plus two node-id-derived subnet families: custody columns and backbone attestation subnets. One of the seven is `sync_committee_contribution_and_proof`; the four `sync_committee_{0..3}` subnets are diff --git a/crates/blockchain/src/beacon_blobs.rs b/crates/blockchain/src/beacon_blobs.rs new file mode 100644 index 000000000..55d2820b9 --- /dev/null +++ b/crates/blockchain/src/beacon_blobs.rs @@ -0,0 +1,479 @@ +//! Data columns from this node's own execution client (`engine_getBlobsV2`). +//! +//! A fulu block held for its custody columns (see +//! `BlockChainServer::hold_block_for_columns`) is usually answered by gossip +//! within a fraction of a second. When it is not, the execution client may +//! already hold every blob the block commits to, since the transactions that +//! carry them went through its pool first. This module asks it once and turns +//! the answer into the custody columns still missing. +//! +//! # Flow +//! +//! ```text +//! hold (first time) -> spawn_fetch +//! task: engine_getBlobsV2(versioned hashes) 1 s ceiling +//! null / error -> count, give up (gossip and peer asks carry on) +//! every blob -> spawn_blocking: +//! block imported meanwhile? -> stop +//! custody columns still missing, none? -> stop +//! cells + sidecars (no KZG verification) +//! check_data_column_sidecars(.., ExecutionLayer) +//! p2p: records the columns as seen, publishes the new ones on their +//! subnets, forwards them to the chain actor, which stores them and +//! releases the held block +//! ``` +//! +//! # Why it is shaped this way +//! +//! - **Spec basis.** Fulu `p2p-interface.md`, "Distributed blob publishing +//! using blobs retrieved from local execution-layer client": honest nodes +//! SHOULD query `engine_getBlobsV2` as soon as they receive a valid +//! `beacon_block` or `data_column_sidecar` from gossip, convert a complete +//! answer to data columns and import them as if received on gossip. This +//! node asks on the first data-availability hold of a block, whether the +//! block came from gossip or from sync; it does not yet ask when a sidecar +//! arrives. Hence the hand-over to p2p rather than a direct store write: +//! publishing on subscribed subnets and updating the anti-equivocation cache +//! are MUSTs, and p2p owns both. +//! - **One attempt.** The call races gossip delivery of the same columns, the +//! specification's own timeout for the method is 1 s, and a retry after the +//! engine ladder's backoff would land after that race is decided. A miss +//! leaves the block to gossip and the per-slot peer asks, which is where it +//! was anyway. (The execution client can learn a blob a moment later, since +//! blob transactions keep arriving in its pool; a later ask is not useless, +//! just not worth the delay.) +//! - **V2, not V3.** V3 answers partially, but a column needs one cell from +//! every blob, so a partial blob set builds no complete column and this node +//! has no partial-column assembly. +//! - **No KZG verification.** The specification lets a node skip verifying +//! blobs from its own execution client, so the proofs are trusted as given +//! and the build is [`data_columns::unverified_sidecars`]. A mismatch with +//! the block still fails the shape checks there. +//! - **Not gloas.** From gloas on, data availability moves from the block to +//! the payload envelope, so a held gloas block has no columns to fill this +//! way. +//! +//! The cell computation is CPU-heavy, so it runs on tokio's blocking pool: off +//! the async runtime and off the chain actor, which only ever sees the +//! finished columns, through p2p. + +use std::time::Instant; + +use ethlambda_engine::EngineClient; +use ethlambda_network_api::{BlockChainToP2PRef, SidecarOrigin}; +use ethlambda_state_transition::beacon::data_columns; +use ethlambda_state_transition::beacon::stf::deneb::kzg_commitment_to_versioned_hash; +use ethlambda_storage::Store; +use ethlambda_types::ShortRoot; +use ethlambda_types::beacon::containers::{DataColumnSidecar, deneb, electra}; +use ethlambda_types::beacon::primitives::KzgProof; +use ethlambda_types::primitives::H256; +use tracing::{debug, error, info, warn}; + +use crate::metrics; + +/// Ask the execution client for a held fulu block's blobs on a task of its +/// own, and hand the custody columns built from them to the p2p layer. +pub(crate) fn spawn_fetch( + engine: EngineClient, + p2p: BlockChainToP2PRef, + store: Store, + block: electra::SignedBeaconBlock, + block_root: H256, + custody_columns: Vec, +) { + tokio::spawn(fetch_and_publish( + engine, + p2p, + store, + block, + block_root, + custody_columns, + )); +} + +async fn fetch_and_publish( + engine: EngineClient, + p2p: BlockChainToP2PRef, + store: Store, + block: electra::SignedBeaconBlock, + block_root: H256, + custody_columns: Vec, +) { + let slot = block.message.slot; + let versioned_hashes: Vec = block + .message + .body + .blob_kzg_commitments + .iter() + .map(kzg_commitment_to_versioned_hash) + .collect(); + + let started = Instant::now(); + let answer = engine.get_blobs_v2(&versioned_hashes).await; + metrics::observe_engine_get_blobs_duration(started.elapsed()); + + let entries = match answer { + Err(err) => { + debug!(%slot, block_root = %ShortRoot(&block_root.0), %err, "engine_getBlobsV2 failed"); + metrics::inc_engine_get_blobs("error"); + return; + } + Ok(None) => { + debug!(%slot, block_root = %ShortRoot(&block_root.0), "Execution client is missing blobs for the block"); + metrics::inc_engine_get_blobs("miss"); + return; + } + Ok(Some(entries)) => { + metrics::inc_engine_get_blobs("hit"); + entries + } + }; + + let blob_count = entries.len(); + let work = tokio::task::spawn_blocking(move || { + // The block was imported while the execution client answered (gossip + // completed it first): nothing left to supply. + match store.has_state(&block_root) { + Ok(false) => {} + Ok(true) => return None, + Err(err) => { + error!(%slot, block_root = %ShortRoot(&block_root.0), %err, "Failed to check whether a held block was imported"); + return None; + } + } + let present = match store.data_column_indices_for(slot, &block_root) { + Ok(present) => present, + Err(err) => { + error!(%slot, block_root = %ShortRoot(&block_root.0), %err, "Failed to read a held block's stored data columns"); + return None; + } + }; + let missing: Vec = custody_columns + .iter() + .copied() + .filter(|index| !present.contains(index)) + .collect(); + if missing.is_empty() { + return None; + } + + let converted = Instant::now(); + let mut blobs = Vec::with_capacity(entries.len()); + let mut proofs: Vec = Vec::new(); + for entry in entries { + match deneb::Blob::try_from(entry.blob) { + Ok(blob) => blobs.push(blob), + Err(_) => { + warn!(%slot, block_root = %ShortRoot(&block_root.0), "Execution client's blob has the wrong size"); + return None; + } + } + proofs.extend(entry.proofs); + } + let sidecars = match data_columns::unverified_sidecars(&block, &blobs, &proofs, &missing) { + Ok(sidecars) => sidecars, + Err(err) => { + warn!(%slot, block_root = %ShortRoot(&block_root.0), %err, "Execution client's blobs do not fit the block"); + return None; + } + }; + metrics::observe_engine_blobs_to_columns(converted.elapsed()); + Some(sidecars) + }) + .await; + + let sidecars = match work { + Ok(Some(sidecars)) if !sidecars.is_empty() => sidecars, + Ok(_) => return, + Err(err) => { + error!(%slot, block_root = %ShortRoot(&block_root.0), %err, "Blob-to-column task failed"); + return; + } + }; + + let columns = sidecars.len(); + let sidecars: Vec = + sidecars.into_iter().map(DataColumnSidecar::Fulu).collect(); + let _ = p2p + .check_data_column_sidecars(sidecars, SidecarOrigin::ExecutionLayer) + .inspect_err(|err| error!(%slot, %err, "Failed to hand execution-client columns to P2P")); + + info!( + %slot, + block_root = %ShortRoot(&block_root.0), + blobs = blob_count, + columns, + elapsed_ms = started.elapsed().as_millis() as u64, + "Built data columns from the execution client's blobs" + ); +} + +#[cfg(test)] +mod tests { + use std::sync::Arc; + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::time::Duration; + + use ethlambda_engine::{EngineClient, JwtSecret}; + use ethlambda_state_transition::beacon::kzg; + use ethlambda_types::beacon::containers::SignedBeaconBlock; + use ethlambda_types::beacon::primitives::KzgCommitment; + use ethlambda_types::primitives::HashTreeRoot as _; + use libssz_types::SszList; + use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _}; + use tokio::net::TcpListener; + + use super::*; + use crate::import_timing::ImportTimings; + use crate::tests::{beacon_server_recording, beacon_store, fulu_block}; + use ethlambda_types::beacon::preset::BYTES_PER_BLOB; + + /// A blob whose every field element is below the BLS modulus. + fn blob_bytes(seed: u8) -> Vec { + let mut bytes = vec![0u8; BYTES_PER_BLOB]; + for (i, element) in bytes.chunks_mut(32).enumerate() { + element[1] = seed; + element[2] = i as u8; + element[3] = (i >> 8) as u8; + } + bytes + } + + /// The `result` of an `engine_getBlobsV2` answer for `blobs`, with real + /// cell proofs. + fn answer_for(blobs: &[Vec]) -> String { + let entries: Vec = blobs + .iter() + .map(|blob| { + let (_, proofs) = kzg::compute_cells_and_kzg_proofs(blob).unwrap(); + let proofs: Vec = proofs + .iter() + .map(|proof| format!("\"0x{}\"", hex::encode(proof.0))) + .collect(); + format!( + r#"{{"blob":"0x{}","proofs":[{}]}}"#, + hex::encode(blob), + proofs.join(",") + ) + }) + .collect(); + format!("[{}]", entries.join(",")) + } + + /// A loopback JSON-RPC server answering every request with `result`, and + /// counting them. + async fn mock_engine(result: String) -> (EngineClient, Arc) { + let requests = Arc::new(AtomicUsize::new(0)); + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("a loopback port"); + let addr = listener.local_addr().expect("bound address"); + let counter = Arc::clone(&requests); + let payload = format!(r#"{{"jsonrpc":"2.0","id":1,"result":{result}}}"#); + tokio::spawn(async move { + loop { + let Ok((mut socket, _)) = listener.accept().await else { + return; + }; + let counter = Arc::clone(&counter); + let payload = payload.clone(); + tokio::spawn(async move { + let mut pending: Vec = Vec::new(); + loop { + let (head_end, body_len) = loop { + if let Some(end) = + pending.windows(4).position(|window| window == b"\r\n\r\n") + { + let head = String::from_utf8_lossy(&pending[..end]).to_lowercase(); + let len = head + .lines() + .find_map(|line| line.strip_prefix("content-length:")) + .and_then(|value| value.trim().parse::().ok()) + .unwrap_or(0); + break (end + 4, len); + } + let mut chunk = [0u8; 16 * 1024]; + match socket.read(&mut chunk).await { + Ok(0) | Err(_) => return, + Ok(read) => pending.extend_from_slice(&chunk[..read]), + } + }; + while pending.len() < head_end + body_len { + let mut chunk = [0u8; 16 * 1024]; + match socket.read(&mut chunk).await { + Ok(0) | Err(_) => return, + Ok(read) => pending.extend_from_slice(&chunk[..read]), + } + } + pending.drain(..head_end + body_len); + counter.fetch_add(1, Ordering::SeqCst); + + let response = format!( + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n{payload}", + payload.len() + ); + if socket.write_all(response.as_bytes()).await.is_err() { + return; + } + } + }); + } + }); + let client = EngineClient::new(format!("http://{addr}"), JwtSecret::new([0x0f; 32])) + .expect("a client"); + (client, requests) + } + + /// A fulu block at slot 1 committing to each of `blobs`. + fn block_committing_to(blobs: &[Vec]) -> SignedBeaconBlock { + let mut block = fulu_block(H256::ZERO, 1, 0); + let commitments: Vec = blobs + .iter() + .map(|blob| kzg::blob_to_kzg_commitment(blob).unwrap()) + .collect(); + let SignedBeaconBlock::Fulu(inner) = &mut block else { + unreachable!("fulu_block builds a fulu block"); + }; + inner.message.body.blob_kzg_commitments = SszList::try_from(commitments).unwrap(); + block + } + + const CUSTODY: [u64; 3] = [0, 5, 77]; + + /// Polls until `RecordingP2P` has a hand-over, or a few seconds pass. + async fn wait_for_hand_over(p2p: &crate::tests::RecordingP2P) { + for _ in 0..100 { + if !p2p.checks.lock().unwrap().is_empty() { + return; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + } + + #[tokio::test] + async fn a_full_answer_hands_over_exactly_the_custody_columns() { + let blobs = vec![blob_bytes(1), blob_bytes(2)]; + let block = block_committing_to(&blobs); + let root = block.message_hash_tree_root(); + let (engine, requests) = mock_engine(answer_for(&blobs)).await; + let (mut server, p2p) = beacon_server_recording(beacon_store(0, 0)); + server.engine = Some(engine); + server.custody_columns = CUSTODY.to_vec(); + + server.hold_block_for_columns(block.clone(), 1, ImportTimings::default()); + wait_for_hand_over(&p2p).await; + + let checks = p2p.checks.lock().unwrap(); + assert_eq!(checks.len(), 1, "one hand-over"); + assert_eq!( + *p2p.check_origins.lock().unwrap(), + vec![ethlambda_network_api::SidecarOrigin::ExecutionLayer] + ); + let indices: Vec = checks[0] + .iter() + .map(|sidecar| match sidecar { + DataColumnSidecar::Fulu(sidecar) => sidecar.index, + DataColumnSidecar::Gloas(_) => panic!("a fulu block yields fulu sidecars"), + }) + .collect(); + assert_eq!(indices, CUSTODY.to_vec()); + for sidecar in checks[0].iter() { + let DataColumnSidecar::Fulu(sidecar) = sidecar else { + unreachable!() + }; + let header = &sidecar.signed_block_header.message; + assert_eq!(header.slot, block.slot()); + assert_eq!(header.parent_root, block.parent_root()); + assert_eq!(header.hash_tree_root(), root); + assert_eq!(sidecar.column.len(), blobs.len()); + } + assert_eq!(requests.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + async fn a_stored_custody_column_is_not_handed_over_again() { + let blobs = vec![blob_bytes(3)]; + let block = block_committing_to(&blobs); + let root = block.message_hash_tree_root(); + let (engine, _requests) = mock_engine(answer_for(&blobs)).await; + let (mut server, p2p) = beacon_server_recording(beacon_store(0, 0)); + server.engine = Some(engine); + server.custody_columns = CUSTODY.to_vec(); + server + .store + .put_data_column_sidecar(block.slot(), &root, 0, vec![1]) + .unwrap(); + + server.hold_block_for_columns(block, 1, ImportTimings::default()); + wait_for_hand_over(&p2p).await; + + let checks = p2p.checks.lock().unwrap(); + assert_eq!(checks.len(), 1); + let indices: Vec = checks[0] + .iter() + .map(|sidecar| match sidecar { + DataColumnSidecar::Fulu(sidecar) => sidecar.index, + DataColumnSidecar::Gloas(_) => panic!("a fulu block yields fulu sidecars"), + }) + .collect(); + assert_eq!(indices, vec![5, 77]); + } + + #[tokio::test] + async fn a_null_answer_hands_nothing_over() { + let blobs = vec![blob_bytes(4)]; + let block = block_committing_to(&blobs); + let (engine, requests) = mock_engine("null".to_string()).await; + let (mut server, p2p) = beacon_server_recording(beacon_store(0, 0)); + server.engine = Some(engine); + server.custody_columns = CUSTODY.to_vec(); + + server.hold_block_for_columns(block, 1, ImportTimings::default()); + for _ in 0..40 { + if requests.load(Ordering::SeqCst) > 0 { + break; + } + tokio::time::sleep(Duration::from_millis(25)).await; + } + tokio::time::sleep(Duration::from_millis(300)).await; + + assert_eq!(requests.load(Ordering::SeqCst), 1, "the engine was asked"); + assert!(p2p.checks.lock().unwrap().is_empty()); + } + + #[tokio::test] + async fn without_an_engine_nothing_is_asked_or_handed_over() { + let blobs = vec![blob_bytes(5)]; + let block = block_committing_to(&blobs); + let (mut server, p2p) = beacon_server_recording(beacon_store(0, 0)); + server.custody_columns = CUSTODY.to_vec(); + + server.hold_block_for_columns(block, 1, ImportTimings::default()); + tokio::time::sleep(Duration::from_millis(200)).await; + + assert!(p2p.checks.lock().unwrap().is_empty()); + } + + #[tokio::test] + async fn a_second_hold_of_the_same_root_does_not_ask_again() { + let blobs = vec![blob_bytes(6)]; + let block = block_committing_to(&blobs); + let (engine, requests) = mock_engine("null".to_string()).await; + let (mut server, _p2p) = beacon_server_recording(beacon_store(0, 0)); + server.engine = Some(engine); + server.custody_columns = CUSTODY.to_vec(); + + server.hold_block_for_columns(block.clone(), 1, ImportTimings::default()); + for _ in 0..40 { + if requests.load(Ordering::SeqCst) > 0 { + break; + } + tokio::time::sleep(Duration::from_millis(25)).await; + } + server.hold_block_for_columns(block, 1, ImportTimings::default()); + tokio::time::sleep(Duration::from_millis(300)).await; + + assert_eq!(requests.load(Ordering::SeqCst), 1); + } +} diff --git a/crates/blockchain/src/beacon_columns.rs b/crates/blockchain/src/beacon_columns.rs index 88d035a2a..2ae4fd3c5 100644 --- a/crates/blockchain/src/beacon_columns.rs +++ b/crates/blockchain/src/beacon_columns.rs @@ -18,6 +18,7 @@ use std::collections::{HashMap, HashSet}; +use ethlambda_network_api::SidecarOrigin; use ethlambda_state_transition::beacon::gossip::column::{ is_parkable_gloas, judgeable_block_gloas, }; @@ -141,7 +142,7 @@ impl BlockChainServer { let Some(ref p2p) = self.p2p else { return; }; - let _ = p2p.check_data_column_sidecars(sidecars).inspect_err( + let _ = p2p.check_data_column_sidecars(sidecars, SidecarOrigin::Parked).inspect_err( |err| error!(%err, "Failed to send data column sidecars to the p2p layer for checks"), ); } diff --git a/crates/blockchain/src/lib.rs b/crates/blockchain/src/lib.rs index 251048b7a..59d52b2d7 100644 --- a/crates/blockchain/src/lib.rs +++ b/crates/blockchain/src/lib.rs @@ -55,6 +55,7 @@ pub use events::{ChainEvent, EventBus, Topic, UnknownTopic}; pub mod aggregation; mod beacon_aggregates; +mod beacon_blobs; mod beacon_columns; pub mod beacon_engine; mod beacon_envelope; @@ -3142,6 +3143,13 @@ impl BlockChainServer { /// arrives in; leaving it to the redrive cadence instead made a follower /// recovering from a checkpoint sync import roughly one such block per /// slot. + /// + /// The execution client is asked for the block's blobs once, on the first + /// hold of a fulu block that commits to any (see [`beacon_blobs`]). That + /// does not depend on the peer-ask timing above: it is a local call that + /// races nobody, and a hit hands the custody columns back through the same + /// path gossip uses, so it is as useful for a block at the tip as for an + /// old one. fn hold_block_for_columns( &mut self, block: SignedBeaconBlock, @@ -3169,6 +3177,23 @@ impl BlockChainServer { "Holding block: its custody columns have not all arrived yet" ); + // Before the block is moved into the store. Only the first hold asks: + // a redelivery of a held root must not re-query the execution client. + if let (Some(engine), Some(p2p), SignedBeaconBlock::Fulu(inner)) = + (&self.engine, &self.p2p, &block) + && !inner.message.body.blob_kzg_commitments.is_empty() + && !self.blocks_awaiting_columns.contains_key(&block_root) + { + beacon_blobs::spawn_fetch( + engine.clone(), + p2p.clone(), + self.store.clone(), + inner.clone(), + block_root, + self.custody_columns.clone(), + ); + } + // Written the same way a parent-missing block is (see // `process_or_pend_block`): no `LiveChain` entry, so it stays // invisible to fork choice until it is re-admitted, but readable back @@ -4665,6 +4690,8 @@ mod tests { #[derive(Default)] pub(crate) struct RecordingP2P { pub(crate) checks: std::sync::Mutex>>, + /// The origin each entry of `checks` was handed over with, in order. + pub(crate) check_origins: std::sync::Mutex>, pub(crate) fetches: std::sync::Mutex>, } @@ -4697,8 +4724,10 @@ mod tests { fn check_data_column_sidecars( &self, sidecars: Vec, + origin: ethlambda_network_api::SidecarOrigin, ) -> Result<(), spawned_concurrency::error::ActorError> { self.checks.lock().unwrap().push(sidecars); + self.check_origins.lock().unwrap().push(origin); Ok(()) } } diff --git a/crates/blockchain/src/metrics.rs b/crates/blockchain/src/metrics.rs index 34c67398f..6823eeb85 100644 --- a/crates/blockchain/src/metrics.rs +++ b/crates/blockchain/src/metrics.rs @@ -1650,3 +1650,55 @@ pub fn observe_beacon_aggregate_end_to_end(duration: Duration) { }); LEAN_BEACON_AGGREGATE_END_TO_END_SECONDS.observe(duration.as_secs_f64()); } + +/// Count one `engine_getBlobsV2` attempt for a held fulu block, by what came +/// back: `hit` (every blob), `miss` (the execution client answered `null`, so +/// at least one blob is not in its pool) or `error` (transport, timeout or a +/// malformed answer). +pub fn inc_engine_get_blobs(outcome: &str) { + static LEAN_ENGINE_GET_BLOBS_TOTAL: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + register_int_counter_vec!( + "lean_engine_get_blobs_total", + "engine_getBlobsV2 attempts for held blocks by outcome (hit, miss, error)", + &["outcome"] + ) + .unwrap() + }); + LEAN_ENGINE_GET_BLOBS_TOTAL + .with_label_values(&[outcome]) + .inc(); +} + +/// How long one `engine_getBlobsV2` round trip took, whatever its outcome. +/// +/// The buckets stop at the call's own timeout, so a mass in the last bucket +/// means the execution client is not answering inside the window the +/// specification gives it. +pub fn observe_engine_get_blobs_duration(duration: Duration) { + static LEAN_ENGINE_GET_BLOBS_DURATION_SECONDS: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + register_histogram!( + "lean_engine_get_blobs_duration_seconds", + "Time for one engine_getBlobsV2 round trip", + vec![0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0] + ) + .unwrap() + }); + LEAN_ENGINE_GET_BLOBS_DURATION_SECONDS.observe(duration.as_secs_f64()); +} + +/// How long turning an execution client's blobs into custody column sidecars +/// took: the cell computation plus the sidecar build, off the actor thread. +pub fn observe_engine_blobs_to_columns(duration: Duration) { + static LEAN_ENGINE_BLOBS_TO_COLUMNS_SECONDS: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + register_histogram!( + "lean_engine_blobs_to_columns_seconds", + "Time to build custody data column sidecars from execution-client blobs", + vec![0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.0] + ) + .unwrap() + }); + LEAN_ENGINE_BLOBS_TO_COLUMNS_SECONDS.observe(duration.as_secs_f64()); +} diff --git a/crates/blockchain/state_transition/src/beacon/data_columns.rs b/crates/blockchain/state_transition/src/beacon/data_columns.rs index 26cc347e8..9707fb977 100644 --- a/crates/blockchain/state_transition/src/beacon/data_columns.rs +++ b/crates/blockchain/state_transition/src/beacon/data_columns.rs @@ -148,31 +148,50 @@ pub fn get_data_column_sidecars( ); (0..NUMBER_OF_COLUMNS) .map(|column_index| { - let column: Vec = cells_and_proofs - .iter() - .map(|(cells, _)| { - SszVector::try_from(cells[column_index].to_bytes().to_vec()) - .expect("a c-kzg cell is BYTES_PER_CELL bytes") - }) - .collect(); - let kzg_proofs: Vec = cells_and_proofs - .iter() - .map(|(_, proofs)| proofs[column_index]) - .collect(); - DataColumnSidecar { - index: column_index as u64, - column: SszList::try_from(column) - .expect("a block holds at most MAX_BLOB_COMMITMENTS_PER_BLOCK blobs"), - kzg_commitments: kzg_commitments.clone(), - kzg_proofs: SszList::try_from(kzg_proofs) - .expect("a block holds at most MAX_BLOB_COMMITMENTS_PER_BLOCK blobs"), - signed_block_header: signed_block_header.clone(), - kzg_commitments_inclusion_proof: inclusion_proof.clone(), - } + column_sidecar( + column_index, + signed_block_header, + kzg_commitments, + inclusion_proof, + cells_and_proofs, + ) }) .collect() } +/// The sidecar for one column: that column's cell and proof from every blob, +/// under the block's header, commitments and inclusion proof. The one place a +/// sidecar is assembled, so the full set and a requested subset cannot differ. +fn column_sidecar( + column_index: usize, + signed_block_header: &SignedBeaconBlockHeader, + kzg_commitments: &KzgCommitments, + inclusion_proof: &KzgCommitmentsInclusionProof, + cells_and_proofs: &[(Box, &ProofsPerBlob)], +) -> DataColumnSidecar { + let column: Vec = cells_and_proofs + .iter() + .map(|(cells, _)| { + SszVector::try_from(cells[column_index].to_bytes().to_vec()) + .expect("a c-kzg cell is BYTES_PER_CELL bytes") + }) + .collect(); + let kzg_proofs: Vec = cells_and_proofs + .iter() + .map(|(_, proofs)| proofs[column_index]) + .collect(); + DataColumnSidecar { + index: column_index as u64, + column: SszList::try_from(column) + .expect("a block holds at most MAX_BLOB_COMMITMENTS_PER_BLOCK blobs"), + kzg_commitments: kzg_commitments.clone(), + kzg_proofs: SszList::try_from(kzg_proofs) + .expect("a block holds at most MAX_BLOB_COMMITMENTS_PER_BLOCK blobs"), + signed_block_header: signed_block_header.clone(), + kzg_commitments_inclusion_proof: inclusion_proof.clone(), + } +} + /// The specification's `get_data_column_sidecars_from_block`. /// /// # Panics @@ -205,25 +224,12 @@ pub fn verified_sidecars( blobs: &[Blob], cell_proofs: &[KzgProof], ) -> Result, SidecarError> { - let commitments = &block.message.body.blob_kzg_commitments; - if blobs.len() != commitments.len() { - return Err(SidecarError::Shape( - "the number of blobs differs from the block's commitments", - )); - } - if cell_proofs.len() != blobs.len() * CELLS_PER_EXT_BLOB { - return Err(SidecarError::Shape( - "the number of cell proofs is not blobs * CELLS_PER_EXT_BLOB", - )); - } + check_shape(block, blobs, cell_proofs)?; if blobs.is_empty() { return Ok(Vec::new()); } - - let all_cells: Vec> = blobs - .par_iter() - .map(|blob| kzg::compute_cells(&blob[..]).map_err(|_| SidecarError::InvalidBlob)) - .collect::>()?; + let commitments = &block.message.body.blob_kzg_commitments; + let all_cells = compute_all_cells(blobs)?; let mut batch_commitments = Vec::with_capacity(cell_proofs.len()); let mut batch_indices = Vec::with_capacity(cell_proofs.len()); @@ -246,16 +252,106 @@ pub fn verified_sidecars( return Err(SidecarError::InvalidProofs); } - // The shape check above made the length an exact multiple, so no remainder. - let (proof_chunks, _) = cell_proofs.as_chunks::(); - let cells_and_proofs: Vec<(Box, &ProofsPerBlob)> = - all_cells.into_iter().zip(proof_chunks).collect(); + let cells_and_proofs = pair_cells_with_proofs(all_cells, cell_proofs); Ok(get_data_column_sidecars_from_block( block, &cells_and_proofs, )) } +/// The sidecars for `columns` of `block`, built from blobs and cell proofs +/// this node's own execution client returned (`engine_getBlobsV2`), without +/// verifying the proofs. +/// +/// Fulu `p2p-interface.md` ("Distributed blob publishing using blobs +/// retrieved from local execution-layer client") says a client SHOULD skip +/// verification of blobs it got from its own execution layer, which is why +/// this is not [`verified_sidecars`]. The caller is the chain actor's +/// `beacon_blobs` module; blobs from any other source must go through +/// [`verified_sidecars`]. +/// +/// The shape checks are the same as [`verified_sidecars`]. Only the requested +/// columns are built, one per distinct index in ascending order, whatever the +/// order or repetition in `columns`; an index at or past `NUMBER_OF_COLUMNS` +/// is a [`SidecarError::Shape`]. A block with zero blobs, or an empty +/// `columns`, yields no sidecars and computes no cells. +pub fn unverified_sidecars( + block: &electra::SignedBeaconBlock, + blobs: &[Blob], + cell_proofs: &[KzgProof], + columns: &[u64], +) -> Result, SidecarError> { + check_shape(block, blobs, cell_proofs)?; + let mut columns = columns.to_vec(); + columns.sort_unstable(); + columns.dedup(); + if columns + .last() + .is_some_and(|&last| last >= NUMBER_OF_COLUMNS as u64) + { + return Err(SidecarError::Shape( + "a requested column index is not below NUMBER_OF_COLUMNS", + )); + } + if blobs.is_empty() || columns.is_empty() { + return Ok(Vec::new()); + } + + let cells_and_proofs = pair_cells_with_proofs(compute_all_cells(blobs)?, cell_proofs); + let commitments = &block.message.body.blob_kzg_commitments; + let (inclusion_proof, body_root) = inclusion_proof_and_body_root(&block.message.body); + let header = header_with_body_root(block, body_root); + Ok(columns + .into_iter() + .map(|column_index| { + column_sidecar( + column_index as usize, + &header, + commitments, + &inclusion_proof, + &cells_and_proofs, + ) + }) + .collect()) +} + +/// The count checks both sidecar builders start with, before any KZG work. +fn check_shape( + block: &electra::SignedBeaconBlock, + blobs: &[Blob], + cell_proofs: &[KzgProof], +) -> Result<(), SidecarError> { + if blobs.len() != block.message.body.blob_kzg_commitments.len() { + return Err(SidecarError::Shape( + "the number of blobs differs from the block's commitments", + )); + } + if cell_proofs.len() != blobs.len() * CELLS_PER_EXT_BLOB { + return Err(SidecarError::Shape( + "the number of cell proofs is not blobs * CELLS_PER_EXT_BLOB", + )); + } + Ok(()) +} + +/// Every blob's cells, computed in parallel (one blob per task). +fn compute_all_cells(blobs: &[Blob]) -> Result>, SidecarError> { + blobs + .par_iter() + .map(|blob| kzg::compute_cells(&blob[..]).map_err(|_| SidecarError::InvalidBlob)) + .collect() +} + +/// Blob `i`'s cells next to its `CELLS_PER_EXT_BLOB` proofs. `check_shape` +/// made the proof count an exact multiple, so there is no remainder. +fn pair_cells_with_proofs( + all_cells: Vec>, + cell_proofs: &[KzgProof], +) -> Vec<(Box, &ProofsPerBlob)> { + let (proof_chunks, _) = cell_proofs.as_chunks::(); + all_cells.into_iter().zip(proof_chunks).collect() +} + #[cfg(test)] mod tests { use super::*; @@ -382,4 +478,65 @@ mod tests { Err(SidecarError::InvalidBlob) )); } + + #[test] + fn requested_columns_match_the_verified_set_in_ascending_order() { + let blobs = [blob(1), blob(2)]; + let (block, proofs) = block_with(&blobs); + let all = verified_sidecars(&block, &blobs, &proofs).unwrap(); + let last = NUMBER_OF_COLUMNS as u64 - 1; + let got = unverified_sidecars(&block, &blobs, &proofs, &[last, 5, 0, 5]).unwrap(); + let expected: Vec<_> = [0, 5, last as usize] + .into_iter() + .map(|i| all[i].clone()) + .collect(); + assert_eq!(got, expected); + } + + #[test] + fn no_requested_columns_or_no_blobs_yield_nothing() { + let blobs = [blob(1)]; + let (block, proofs) = block_with(&blobs); + assert!( + unverified_sidecars(&block, &blobs, &proofs, &[]) + .unwrap() + .is_empty() + ); + let (empty_block, empty_proofs) = block_with(&[]); + assert!( + unverified_sidecars(&empty_block, &[], &empty_proofs, &[0, 1]) + .unwrap() + .is_empty() + ); + } + + #[test] + fn unverified_sidecars_refuse_bad_shapes() { + let blobs = [blob(1)]; + let (block, proofs) = block_with(&blobs); + assert!(matches!( + unverified_sidecars(&block, &[], &proofs, &[0]), + Err(SidecarError::Shape(_)) + )); + assert!(matches!( + unverified_sidecars(&block, &blobs, &proofs[..proofs.len() - 1], &[0]), + Err(SidecarError::Shape(_)) + )); + assert!(matches!( + unverified_sidecars(&block, &blobs, &proofs, &[0, NUMBER_OF_COLUMNS as u64]), + Err(SidecarError::Shape(_)) + )); + } + + #[test] + fn a_tampered_proof_still_builds_because_nothing_is_verified() { + let blobs = [blob(1), blob(2)]; + let (block, mut proofs) = block_with(&blobs); + proofs.swap(0, 1); + let swapped = proofs[0]; + let got = unverified_sidecars(&block, &blobs, &proofs, &[0]).unwrap(); + assert_eq!(got.len(), 1); + assert_eq!(got[0].kzg_proofs[0], swapped); + assert!(!verify_data_column_sidecar_kzg_proofs(&got[0]).unwrap()); + } } diff --git a/crates/net/api/src/lib.rs b/crates/net/api/src/lib.rs index a0aa347d2..0308c430b 100644 --- a/crates/net/api/src/lib.rs +++ b/crates/net/api/src/lib.rs @@ -33,20 +33,43 @@ pub trait BlockChainToP2P: Send + Sync { ) -> Result<(), ActorError>; /// Ask peers for whatever of one block this node is missing. fn fetch_block(&self, request: FetchRequest) -> Result<(), ActorError>; - /// Run the chain checks on sidecars the chain actor had parked, now that - /// the block they are checked against has a post-state: their parent for a - /// fulu sidecar, their own block for a gloas one. + /// Hand sidecars the chain actor holds to the p2p layer, which decides + /// what they need before they are stored by where they came from; see + /// [`SidecarOrigin`]. /// - /// The chain actor keeps a sidecar without checking it, so it hands these - /// back rather than judging them itself: the p2p layer runs every column - /// check, off both actors, and sends the ones that pass back through - /// [`P2PToBlockChain::new_data_column_sidecars`]. + /// For [`SidecarOrigin::Parked`] that is the chain checks: sidecars the + /// chain actor had parked, now that the block they are checked against has + /// a post-state (their parent for a fulu sidecar, their own block for a + /// gloas one). The chain actor keeps a sidecar without checking it, so it + /// hands these back rather than judging them itself: the p2p layer runs + /// every column check, off both actors, and sends the ones that pass back + /// through [`P2PToBlockChain::new_data_column_sidecars`]. fn check_data_column_sidecars( &self, sidecars: Vec, + origin: SidecarOrigin, ) -> Result<(), ActorError>; } +/// Where the sidecars handed to [`BlockChainToP2P::check_data_column_sidecars`] +/// came from, which decides what the p2p layer does with them before they +/// reach [`P2PToBlockChain::new_data_column_sidecars`]. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum SidecarOrigin { + /// From the network, parked by the chain actor unchecked until the block + /// they are judged against had a post-state: run the chain checks. + Parked, + /// Built by this node from the blobs its own execution client returned + /// for a held block (`engine_getBlobsV2`). + /// + /// Not checked: the fulu `p2p-interface.md` says a client "SHOULD skip + /// verification" of blobs from its local execution layer. But it "MUST + /// behave as if the `data_column_sidecar` had been received via gossip": + /// publish each on its subnet if and only if this node is subscribed to + /// it, and record it in the anti-equivocation (seen) cache. + ExecutionLayer, +} + /// What one block is missing, from the chain actor's point of view. /// /// One message rather than a block fetch and a column fetch, because the diff --git a/crates/net/engine/src/building.rs b/crates/net/engine/src/building.rs index 6ff109248..03e239aed 100644 --- a/crates/net/engine/src/building.rs +++ b/crates/net/engine/src/building.rs @@ -362,7 +362,7 @@ fn decode(message: &str) -> EngineError { } /// A `DATA` of any length. -fn parse_data(text: &str) -> Result, EngineError> { +pub(crate) fn parse_data(text: &str) -> Result, EngineError> { let digits = text .strip_prefix("0x") .ok_or_else(|| decode("DATA must be 0x-prefixed"))?; @@ -370,7 +370,7 @@ fn parse_data(text: &str) -> Result, EngineError> { } /// A `DATA` of exactly `N` bytes. -fn parse_fixed(text: &str) -> Result<[u8; N], EngineError> { +pub(crate) fn parse_fixed(text: &str) -> Result<[u8; N], EngineError> { parse_data(text)?.try_into().map_err(|bytes: Vec| { EngineError::Decode(format!("expected {N} bytes, got {}", bytes.len())) }) diff --git a/crates/net/engine/src/client.rs b/crates/net/engine/src/client.rs index fb16ba6db..79797381d 100644 --- a/crates/net/engine/src/client.rs +++ b/crates/net/engine/src/client.rs @@ -1,4 +1,4 @@ -//! The JSON-RPC client, its retry ladder, and the four methods. +//! The JSON-RPC client, its retry ladder, and the methods a follower calls. use std::time::Duration; @@ -10,8 +10,9 @@ use tracing::{debug, warn}; use crate::auth::JwtSecret; use crate::error::EngineError; use crate::types::{ - ClientVersionV1, CustodyColumns, ExecutionPayloadV3, ExecutionPayloadV4, ForkchoiceStateV1, - ForkchoiceUpdatedResponse, PayloadStatusV1, data, + BlobAndProofV2, BlobAndProofV2Json, ClientVersionV1, CustodyColumns, ExecutionPayloadV3, + ExecutionPayloadV4, ForkchoiceStateV1, ForkchoiceUpdatedResponse, PayloadStatusV1, data, + decode_blobs_v2, }; /// Per-attempt timeout. @@ -26,6 +27,10 @@ pub const ENGINE_TIMEOUT: Duration = Duration::from_secs(8); /// for the older methods. pub const ENGINE_NEW_PAYLOAD_V5_TIMEOUT: Duration = Duration::from_secs(6); +/// Per-attempt timeout for `engine_getBlobsV2`, the value `osaka.md` gives that +/// method. +pub const ENGINE_GET_BLOBS_TIMEOUT: Duration = Duration::from_secs(1); + /// How many times one call is attempted before it is given up on. /// /// Three, not the ten `ethlambda-p2p` uses for a peer fetch. Those are different @@ -341,6 +346,37 @@ impl EngineClient { response.try_into() } + /// `engine_getBlobsV2`: every blob named by `versioned_hashes`, in order, + /// with its cell proofs, or `None` when the execution client is missing + /// any of them. + /// + /// One attempt, bounded by [`ENGINE_GET_BLOBS_TIMEOUT`], and deliberately + /// not through the retry ladder: this call races the gossip delivery of the + /// same data columns, and a retry after the ladder's first backoff would + /// land after that race is lost. A failure leaves the block to gossip and + /// peers, which is where it was anyway. + /// + /// An answer that is not shaped as the specification requires (wrong + /// length, blob size or proof count, or a `null` element) is a + /// [`EngineError::Decode`]. + pub async fn get_blobs_v2( + &self, + versioned_hashes: &[Bytes32], + ) -> Result>, EngineError> { + let hashes: Vec = versioned_hashes + .iter() + .map(|hash| data(hash.as_slice())) + .collect(); + let result: Option>> = self + .call_once( + "engine_getBlobsV2", + &json!([hashes]), + ENGINE_GET_BLOBS_TIMEOUT, + ) + .await?; + decode_blobs_v2(result, versioned_hashes.len()) + } + /// `engine_exchangeCapabilities`. Returns what the execution client says it /// supports. pub async fn exchange_capabilities(&self, ours: &[&str]) -> Result, EngineError> { diff --git a/crates/net/engine/src/lib.rs b/crates/net/engine/src/lib.rs index 742cd0bbf..2ac3c74a9 100644 --- a/crates/net/engine/src/lib.rs +++ b/crates/net/engine/src/lib.rs @@ -1,6 +1,6 @@ //! The Ethereum Engine API, as a consensus-layer follower needs it. //! -//! Wire concerns only: JWT authentication, the JSON-RPC envelope, and the four +//! Wire concerns only: JWT authentication, the JSON-RPC envelope, and the //! methods a follower calls. Nothing here knows what a beacon block is beyond //! the containers it serializes, and nothing here decides what an answer means. //! Assembling a request from a block, and reading a status as a fork choice @@ -17,9 +17,11 @@ //! //! Payload building (`PayloadAttributesV3` on `forkchoiceUpdated`, and //! Osaka's `engine_getPayloadV5`) is in [`building`], for the Beacon API's -//! block production. Two method families a full client would have are -//! deliberately absent: `engine_getBlobs*`, because there is no blob-pool -//! fetch path and data columns come from peers; and +//! block production. `engine_getBlobsV2` is called for held fulu blocks, to +//! rebuild data columns from the execution client's blob pool. `V3` is not: +//! it answers partially, a column needs one cell from every blob, and without +//! partial-column assembly a partial blob set yields no complete column. One +//! method family a full client would have is deliberately absent: //! `engine_getPayloadBodies*`, because nothing consumes them. pub mod auth; @@ -31,7 +33,9 @@ pub mod types; pub use auth::JwtSecret; pub use client::EngineClient; pub use error::EngineError; -pub use types::{CustodyColumns, ForkchoiceStateV1, PayloadStatusV1, PayloadStatusValue}; +pub use types::{ + BlobAndProofV2, CustodyColumns, ForkchoiceStateV1, PayloadStatusV1, PayloadStatusValue, +}; /// The methods this client will call, sent in the `engine_exchangeCapabilities` /// handshake. @@ -45,5 +49,6 @@ pub const ETHLAMBDA_ENGINE_CAPABILITIES: &[&str] = &[ "engine_forkchoiceUpdatedV4", "engine_getPayloadV5", "engine_getPayloadV6", + "engine_getBlobsV2", "engine_getClientVersionV1", ]; diff --git a/crates/net/engine/src/types.rs b/crates/net/engine/src/types.rs index 6e142812e..e677598e9 100644 --- a/crates/net/engine/src/types.rs +++ b/crates/net/engine/src/types.rs @@ -12,10 +12,13 @@ use ethlambda_types::beacon::containers::capella::Withdrawal; use ethlambda_types::beacon::containers::{deneb, gloas}; use ethlambda_types::beacon::preset; -use ethlambda_types::beacon::primitives::{ExecutionBlockHash, Uint256}; +use ethlambda_types::beacon::primitives::{ExecutionBlockHash, KzgProof, Uint256}; use ethlambda_types::beacon::serde_helpers::HexPrefixed; use serde::{Deserialize, Serialize, Serializer}; +use crate::building::{parse_data, parse_fixed}; +use crate::error::EngineError; + /// `PayloadStatusV1.status`. #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "SCREAMING_SNAKE_CASE")] @@ -255,6 +258,76 @@ pub fn uint256(value: &Uint256) -> String { format!("0x{trimmed}") } +/// `BlobAndProofV2` (Osaka): one blob from the execution client's blob pool, +/// with its `CELLS_PER_EXT_BLOB` cell proofs. +/// +/// Produced only by `decode_blobs_v2`, which checks the wire shape, so a value +/// of this type always has a `BYTES_PER_BLOB` blob and exactly +/// `CELLS_PER_EXT_BLOB` proofs. The proofs are not verified: the specification +/// lets a consensus client skip that for blobs from its own execution client. +#[derive(Debug, Clone)] +pub struct BlobAndProofV2 { + /// The blob, `BYTES_PER_BLOB` bytes. + pub blob: Vec, + /// The blob's cell proofs, exactly `CELLS_PER_EXT_BLOB` of them. + pub proofs: Vec, +} + +#[derive(Deserialize)] +pub(crate) struct BlobAndProofV2Json { + blob: String, + proofs: Vec, +} + +/// Decodes an `engine_getBlobsV2` result for a request of `requested` hashes. +/// +/// `null` is `Ok(None)`: the execution client is missing at least one blob, or +/// cannot serve its pool. A `null` element is a decode error, since V2 never +/// sends one (V3 does, for a partial answer). +pub(crate) fn decode_blobs_v2( + result: Option>>, + requested: usize, +) -> Result>, EngineError> { + let Some(entries) = result else { + return Ok(None); + }; + if entries.len() != requested { + return Err(EngineError::Decode(format!( + "asked for {requested} blobs, got {}", + entries.len() + ))); + } + + let mut blobs = Vec::with_capacity(entries.len()); + for (index, entry) in entries.into_iter().enumerate() { + let json = entry + .ok_or_else(|| EngineError::Decode(format!("blob {index} is null in a V2 answer")))?; + + let blob = parse_data(&json.blob)?; + if blob.len() != preset::BYTES_PER_BLOB { + return Err(EngineError::Decode(format!( + "blob {index} is {} bytes, expected BYTES_PER_BLOB ({})", + blob.len(), + preset::BYTES_PER_BLOB + ))); + } + if json.proofs.len() != preset::CELLS_PER_EXT_BLOB { + return Err(EngineError::Decode(format!( + "blob {index} has {} proofs, expected CELLS_PER_EXT_BLOB ({})", + json.proofs.len(), + preset::CELLS_PER_EXT_BLOB + ))); + } + let proofs = json + .proofs + .iter() + .map(|text| parse_fixed(text).map(KzgProof)) + .collect::, _>>()?; + blobs.push(BlobAndProofV2 { blob, proofs }); + } + Ok(Some(blobs)) +} + #[cfg(test)] mod tests { use ethlambda_types::beacon::containers::bellatrix; @@ -497,4 +570,44 @@ mod tests { format!("0xa{}", "0".repeat(62)) ); } + + fn blob_json(blob_len: usize, proof_count: usize) -> BlobAndProofV2Json { + BlobAndProofV2Json { + blob: format!("0x{}", "00".repeat(blob_len)), + proofs: vec![format!("0x{}", "11".repeat(48)); proof_count], + } + } + + #[test] + fn a_blob_of_the_wrong_size_is_a_decode_error() { + let answer = Some(vec![Some(blob_json( + preset::BYTES_PER_BLOB - 1, + preset::CELLS_PER_EXT_BLOB, + ))]); + assert!(matches!( + decode_blobs_v2(answer, 1), + Err(EngineError::Decode(_)) + )); + } + + #[test] + fn a_null_element_is_a_decode_error() { + let answer = Some(vec![None]); + assert!(matches!( + decode_blobs_v2(answer, 1), + Err(EngineError::Decode(_)) + )); + } + + #[test] + fn a_well_formed_blob_decodes() { + let answer = Some(vec![Some(blob_json( + preset::BYTES_PER_BLOB, + preset::CELLS_PER_EXT_BLOB, + ))]); + let blobs = decode_blobs_v2(answer, 1) + .expect("well formed") + .expect("some"); + assert_eq!(blobs[0].proofs.len(), preset::CELLS_PER_EXT_BLOB); + } } diff --git a/crates/net/engine/tests/wire_smoke.rs b/crates/net/engine/tests/wire_smoke.rs index 72d943efd..1211dde5e 100644 --- a/crates/net/engine/tests/wire_smoke.rs +++ b/crates/net/engine/tests/wire_smoke.rs @@ -11,18 +11,20 @@ use std::sync::Arc; use ethlambda_engine::types::{CustodyColumns, ForkchoiceStateV1, PayloadStatusValue}; use ethlambda_engine::{EngineClient, EngineError, JwtSecret}; -use ethlambda_types::beacon::primitives::ExecutionBlockHash; +use ethlambda_types::beacon::preset; +use ethlambda_types::beacon::primitives::{Bytes32, ExecutionBlockHash}; use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _}; use tokio::net::TcpListener; use tokio::sync::Mutex; /// Serves exactly one request, answering with `body`, and hands the caller the /// request it saw. -async fn serve_once(body: &'static str) -> (String, Arc>>) { +async fn serve_once(body: impl Into) -> (String, Arc>>) { let listener = TcpListener::bind("127.0.0.1:0") .await .expect("an ephemeral loopback port"); let addr = listener.local_addr().expect("the bound address"); + let body: String = body.into(); let seen = Arc::new(Mutex::new(None)); let seen_writer = Arc::clone(&seen); @@ -138,8 +140,7 @@ async fn an_rpc_error_envelope_surfaces_typed_and_is_not_retried() { #[tokio::test] async fn new_payload_v5_sends_four_params_under_its_own_method() { use ethlambda_types::beacon::containers::{bellatrix, gloas}; - use ethlambda_types::beacon::preset; - use ethlambda_types::beacon::primitives::{Bytes32, ExecutionAddress, Root, Uint256}; + use ethlambda_types::beacon::primitives::{ExecutionAddress, Root, Uint256}; let (endpoint, seen) = serve_once( r#"{"jsonrpc":"2.0","id":1,"result":{"status":"VALID","latestValidHash":"0x0505050505050505050505050505050505050505050505050505050505050505","validationError":null}}"#, @@ -221,3 +222,79 @@ async fn forkchoice_updated_v4_sends_three_params_and_parses_syncing() { // The attributes are null and the bitmap follows them. assert!(request.contains(",null,\"0x01020000000000000000000000000000\"]")); } + +/// A `getBlobsV2` result body with `blobs` blobs of `proof_count` proofs each. +/// Proof `i` is 48 bytes of `i + 1`, so a reordering shows up. +fn blobs_v2_body(blobs: usize, proof_count: usize) -> String { + let blob = format!("0x{}", "00".repeat(preset::BYTES_PER_BLOB)); + let proofs: Vec = (0..proof_count) + .map(|i| format!("0x{}", format!("{:02x}", i + 1).repeat(48))) + .collect(); + let entry = serde_json::json!({ "blob": blob, "proofs": proofs }); + serde_json::json!({ "jsonrpc": "2.0", "id": 1, "result": vec![entry; blobs] }).to_string() +} + +#[tokio::test] +async fn get_blobs_v2_null_is_none_and_the_request_carries_the_hashes() { + let (endpoint, seen) = serve_once(r#"{"jsonrpc":"2.0","id":1,"result":null}"#).await; + + let hashes = [Bytes32::repeat_byte(1), Bytes32::repeat_byte(2)]; + let answer = client(endpoint) + .get_blobs_v2(&hashes) + .await + .expect("null is an answer, not an error"); + assert!(answer.is_none()); + + let request = seen.lock().await.clone().expect("the mock saw a request"); + let (_, body) = request.split_once("\r\n\r\n").expect("a request body"); + let body: serde_json::Value = serde_json::from_str(body).expect("a JSON body"); + assert_eq!(body["method"], "engine_getBlobsV2"); + assert_eq!( + body["params"], + serde_json::json!([[ + format!("0x{}", "01".repeat(32)), + format!("0x{}", "02".repeat(32)), + ]]) + ); +} + +#[tokio::test] +async fn get_blobs_v2_decodes_a_full_answer_in_order() { + let (endpoint, _seen) = serve_once(blobs_v2_body(1, preset::CELLS_PER_EXT_BLOB)).await; + + let blobs = client(endpoint) + .get_blobs_v2(&[Bytes32::repeat_byte(1)]) + .await + .expect("a well-formed answer") + .expect("every blob is available"); + + assert_eq!(blobs.len(), 1); + assert_eq!(blobs[0].blob.len(), preset::BYTES_PER_BLOB); + assert!(blobs[0].blob.iter().all(|byte| *byte == 0)); + assert_eq!(blobs[0].proofs.len(), preset::CELLS_PER_EXT_BLOB); + for (i, proof) in blobs[0].proofs.iter().enumerate() { + assert_eq!(proof.0, [(i + 1) as u8; 48], "proof {i} out of order"); + } +} + +#[tokio::test] +async fn get_blobs_v2_with_the_wrong_proof_count_is_a_decode_error() { + let (endpoint, _seen) = serve_once(blobs_v2_body(1, preset::CELLS_PER_EXT_BLOB - 1)).await; + + let err = client(endpoint) + .get_blobs_v2(&[Bytes32::repeat_byte(1)]) + .await + .expect_err("a short proof list is malformed"); + assert!(matches!(err, EngineError::Decode(_)), "got {err:?}"); +} + +#[tokio::test] +async fn get_blobs_v2_with_a_different_length_than_requested_is_a_decode_error() { + let (endpoint, _seen) = serve_once(blobs_v2_body(1, preset::CELLS_PER_EXT_BLOB)).await; + + let err = client(endpoint) + .get_blobs_v2(&[Bytes32::repeat_byte(1), Bytes32::repeat_byte(2)]) + .await + .expect_err("one blob for two hashes is malformed"); + assert!(matches!(err, EngineError::Decode(_)), "got {err:?}"); +} diff --git a/crates/net/p2p/src/beacon/engine_columns.rs b/crates/net/p2p/src/beacon/engine_columns.rs new file mode 100644 index 000000000..b9df3e3ab --- /dev/null +++ b/crates/net/p2p/src/beacon/engine_columns.rs @@ -0,0 +1,209 @@ +//! Data column sidecars this node built from its own execution client's blob +//! pool (`engine_getBlobsV2`), handed over by the chain actor with +//! `SidecarOrigin::ExecutionLayer`. +//! +//! The fulu p2p spec says a client that takes blobs from its local execution +//! layer SHOULD skip verifying them, and MUST then behave as if the sidecar had +//! arrived by gossip. In particular it MUST publish the sidecar on its +//! `data_column_sidecar_{subnet_id}` topic if and only if it is subscribed to +//! that topic, and MUST update the gossip rule data structures (the +//! anti-equivocation cache, here [`P2PServer::seen_columns`]). +//! +//! That is all this module does, which is why no checks run: the sidecars are +//! built from the block the chain actor already holds, with proofs the +//! execution client vouched for, so the chain checks `beacon::column_checks` +//! runs (header signature, inclusion proof, KZG) have nothing to add. Debug +//! builds still re-run them in the chain actor. +//! +//! Recording comes before publishing on purpose. A copy of the same column that +//! gossip delivers afterwards then finds its key taken and is ignored rather +//! than re-forwarded, and a column gossip already accepted is not published a +//! second time. + +use ethlambda_state_transition::beacon::das; +use ethlambda_types::ShortRoot; +use ethlambda_types::beacon::containers::DataColumnSidecar; +use ethlambda_types::primitives::HashTreeRoot as _; +use libp2p::gossipsub::IdentTopic; +use tracing::{debug, error, info, warn}; + +use crate::beacon::topics as beacon_topics; +use crate::gossipsub::compress_message; +use crate::{P2PServer, metrics}; + +/// Record each sidecar in the seen cache, publish the newly recorded ones on +/// their subnets, and hand them to the chain actor. +pub(crate) fn publish_and_forward(server: &mut P2PServer, sidecars: Vec) { + let Some(beacon) = server.wire.beacon() else { + error!( + columns = sidecars.len(), + "Execution-layer columns reached a lean node; dropping them" + ); + return; + }; + // Subnets are derived from the custody set (see `build_swarm`), so a column + // is subscribed to exactly when it is custodied. + let custody_columns = beacon.custody_columns.clone(); + + let mut fresh: Vec<(DataColumnSidecar, Option)> = Vec::new(); + let mut already_seen = 0u64; + let mut not_custody = 0u64; + let mut missing_digest = false; + let mut batch_header = None; + for sidecar in sidecars { + let DataColumnSidecar::Fulu(fulu) = &sidecar else { + debug!("Skipping a gloas sidecar from the execution layer"); + continue; + }; + let index = fulu.index; + if !custody_columns.contains(&index) { + not_custody += 1; + continue; + } + let header = &fulu.signed_block_header.message; + if !server + .seen_columns + .record(header.slot, header.proposer_index, index) + { + already_seen += 1; + continue; + } + batch_header.get_or_insert_with(|| (header.slot, header.hash_tree_root())); + let topic = server + .wire + .beacon() + .and_then(|beacon| beacon.publish_digest(header.slot)) + .map(|digest| { + let subnet = das::compute_subnet_for_data_column_sidecar(index); + IdentTopic::new(beacon_topics::data_column_topic_name(digest, subnet)) + }); + missing_digest |= topic.is_none(); + fresh.push((sidecar, topic)); + } + + let mut published = 0u64; + let mut forwarded = Vec::with_capacity(fresh.len()); + for (sidecar, topic) in fresh { + if let Some(topic) = topic { + server + .swarm_handle + .publish(topic, compress_message(&sidecar.to_ssz())); + published += 1; + } + forwarded.push(sidecar); + } + let unpublished = forwarded.len() as u64 - published; + + if let Some((slot, block_root)) = batch_header { + if missing_digest { + warn!( + slot, + "No held fork digest covers these columns' slot; not publishing them" + ); + } + info!( + slot, + block_root = %ShortRoot(&block_root.0), + published, + no_digest = unpublished, + already_seen, + not_custody, + "Execution-layer data columns" + ); + } + metrics::inc_data_columns_from_execution_layer("published", published); + metrics::inc_data_columns_from_execution_layer("no_digest", unpublished); + metrics::inc_data_columns_from_execution_layer("already_seen", already_seen); + metrics::inc_data_columns_from_execution_layer("not_custody", not_custody); + + if forwarded.is_empty() { + return; + } + if let Some(blockchain) = &server.blockchain { + let _ = blockchain.new_data_column_sidecars(forwarded).inspect_err( + |err| error!(%err, "Failed to hand the execution-layer columns to the chain"), + ); + } +} + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use ethlambda_types::beacon::config::Config; + + use super::*; + use crate::test_support::{RecordingChain, unconnected_beacon_server, valid_shaped_sidecar}; + + const SLOT: u64 = 5; + const PROPOSER: u64 = 7; + + /// A server whose node custodies `custody`, wired to a recording chain. + async fn server_custodying(custody: &[u64]) -> (P2PServer, Arc) { + let mut server = unconnected_beacon_server(Config::mainnet(), 0).await; + let crate::Wire::Beacon(beacon) = &mut server.wire else { + unreachable!("unconnected_beacon_server builds a beacon wire"); + }; + beacon.custody_columns = custody.to_vec(); + // The test server's digest is a placeholder; a held digest is what + // lets the columns be published. + beacon.topics.fork_digest = beacon.digest_for_slot(SLOT); + let chain = Arc::new(RecordingChain::default()); + server.blockchain = Some(chain.clone()); + (server, chain) + } + + fn sidecar(index: u64) -> DataColumnSidecar { + DataColumnSidecar::Fulu(valid_shaped_sidecar(SLOT, index)) + } + + fn forwarded_indices(chain: &RecordingChain) -> Vec> { + chain + .sidecar_batches + .lock() + .unwrap() + .iter() + .map(|batch| batch.iter().map(DataColumnSidecar::index).collect()) + .collect() + } + + #[tokio::test] + async fn newly_seen_custody_columns_are_forwarded_in_one_batch_and_recorded() { + let (mut server, chain) = server_custodying(&[1, 2]).await; + + publish_and_forward(&mut server, vec![sidecar(1), sidecar(2)]); + + assert_eq!(forwarded_indices(&chain), [vec![1, 2]]); + assert!(server.seen_columns.contains(SLOT, PROPOSER, 1)); + assert!(server.seen_columns.contains(SLOT, PROPOSER, 2)); + } + + #[tokio::test] + async fn a_column_gossip_already_accepted_is_not_forwarded() { + let (mut server, chain) = server_custodying(&[1, 2]).await; + assert!(server.seen_columns.record(SLOT, PROPOSER, 1)); + + publish_and_forward(&mut server, vec![sidecar(1), sidecar(2)]); + + assert_eq!(forwarded_indices(&chain), [vec![2]]); + } + + #[tokio::test] + async fn a_column_outside_the_custody_set_is_dropped_and_not_recorded() { + let (mut server, chain) = server_custodying(&[1]).await; + + publish_and_forward(&mut server, vec![sidecar(1), sidecar(9)]); + + assert_eq!(forwarded_indices(&chain), [vec![1]]); + assert!(!server.seen_columns.contains(SLOT, PROPOSER, 9)); + } + + #[tokio::test] + async fn a_batch_with_nothing_new_sends_the_chain_nothing() { + let (mut server, chain) = server_custodying(&[1]).await; + + publish_and_forward(&mut server, vec![sidecar(9)]); + + assert!(forwarded_indices(&chain).is_empty()); + } +} diff --git a/crates/net/p2p/src/beacon/mod.rs b/crates/net/p2p/src/beacon/mod.rs index 33944b4f7..25b4d00ac 100644 --- a/crates/net/p2p/src/beacon/mod.rs +++ b/crates/net/p2p/src/beacon/mod.rs @@ -17,6 +17,7 @@ pub mod builder_market; pub mod column_checks; pub mod decode; pub mod encoding; +pub mod engine_columns; pub mod envelope_checks; pub mod fork_schedule; pub mod handler; diff --git a/crates/net/p2p/src/lib.rs b/crates/net/p2p/src/lib.rs index cc9ca2bd1..5fbbc5d8b 100644 --- a/crates/net/p2p/src/lib.rs +++ b/crates/net/p2p/src/lib.rs @@ -38,7 +38,7 @@ use std::{ use either::Either; use ethlambda_network_api::{ - FetchRequest, InitBlockChain, MetaDataController, P2PToBlockChainRef, + FetchRequest, InitBlockChain, MetaDataController, P2PToBlockChainRef, SidecarOrigin, block_chain_to_p2p::{ CheckDataColumnSidecars, FetchBlock, PublishAggregatedAttestation, PublishAttestation, PublishBlock, @@ -1853,7 +1853,12 @@ impl Handler for P2PServer { impl Handler for P2PServer { async fn handle(&mut self, msg: CheckDataColumnSidecars, _ctx: &Context) { - beacon::column_checks::check_and_forward(self, msg.sidecars); + match msg.origin { + SidecarOrigin::Parked => beacon::column_checks::check_and_forward(self, msg.sidecars), + SidecarOrigin::ExecutionLayer => { + beacon::engine_columns::publish_and_forward(self, msg.sidecars) + } + } } } @@ -2840,6 +2845,9 @@ pub(crate) mod test_support { Vec, )>, >, + /// Each `new_data_column_sidecars` batch, in arrival order. + pub(crate) sidecar_batches: + std::sync::Mutex>>, } impl ethlambda_network_api::P2PToBlockChain for RecordingChain { @@ -2867,8 +2875,9 @@ pub(crate) mod test_support { } fn new_data_column_sidecars( &self, - _sidecars: Vec, + sidecars: Vec, ) -> Result<(), spawned_concurrency::error::ActorError> { + self.sidecar_batches.lock().unwrap().push(sidecars); Ok(()) } fn data_column_sidecars_awaiting_parent( diff --git a/crates/net/p2p/src/metrics.rs b/crates/net/p2p/src/metrics.rs index 643300967..086aa8965 100644 --- a/crates/net/p2p/src/metrics.rs +++ b/crates/net/p2p/src/metrics.rs @@ -511,6 +511,28 @@ pub fn inc_data_column_fetch_failure(reason: &str) { .inc(); } +/// Count data column sidecars this node built from its own execution client's +/// blobs, by what `beacon::engine_columns` did with them. `result` is +/// `"published"` (newly recorded and sent to its subnet topic), `"no_digest"` +/// (newly recorded and handed to the chain, but no held digest covers its slot +/// so nothing was published),`"already_seen"` (gossip delivered that column +/// first) or `"not_custody"` (not one of this node's columns, so not +/// subscribed and dropped). +pub fn inc_data_columns_from_execution_layer(result: &str, count: u64) { + static LEAN_DATA_COLUMNS_FROM_EXECUTION_LAYER_TOTAL: LazyLock = + LazyLock::new(|| { + register_int_counter_vec!( + "lean_data_columns_from_execution_layer_total", + "Data column sidecars built from the local execution client's blobs, by outcome", + &["result"] + ) + .unwrap() + }); + LEAN_DATA_COLUMNS_FROM_EXECUTION_LAYER_TOTAL + .with_label_values(&[result]) + .inc_by(count); +} + /// Test-only readback of [`inc_data_column_fetch_failure`]'s counter. The /// metric otherwise has no consumer inside the crate itself (Prometheus /// scrapes it), so this is the only way a test can observe that a failure was diff --git a/docs/beacon_engine.md b/docs/beacon_engine.md index 1aa13b184..11fc0b095 100644 --- a/docs/beacon_engine.md +++ b/docs/beacon_engine.md @@ -14,7 +14,7 @@ The wire is ## Which methods, and why so few -Six: +Seven: | Method | Introduced | Why | |---|---|---| @@ -24,14 +24,16 @@ Six: | `engine_forkchoiceUpdatedV4` | amsterdam | The same, from a gloas head on; also carries the node's custody columns | | `engine_exchangeCapabilities` | common | Startup handshake | | `engine_getClientVersionV1` | identification | Log what the execution client is | +| `engine_getBlobsV2` | osaka | Rebuild a held fulu block's custody columns from the execution client's blob pool | Osaka introduces **no new `newPayload`**. Its own document adds only -`engine_getPayloadV5` and `engine_getBlobsV2`/`V3`; `engine_newPayloadV5` belongs +`engine_getPayloadV5` and `engine_getBlobsV2`/`V3`, of which this node calls +`engine_getBlobsV2` (see [Blobs from the execution client](#blobs-from-the-execution-client)); `engine_newPayloadV5` belongs to Amsterdam, where gloas's envelope import uses it. So the Osaka-current call for a payload is prague's V4, and the Osaka-current fork choice notification is cancun's V3. Amsterdam's pair (newPayload V5, forkchoiceUpdated V4) is used from gloas on; the startup handshake warns when the execution client does not advertise either, but only on a network whose schedule includes gloas. -Three method families a full client would have are deliberately absent: +Two method families a full client would have are deliberately absent: - **`engine_getPayload*` outside the gloas path.** The follower itself never asks an execution client to start building a block, so its own @@ -43,8 +45,6 @@ Three method families a full client would have are deliberately absent: `engine_getPayloadV6` (`ExecutionPayloadV4`, with `blockAccessList` and `slotNumber`) from gloas on. The handshake warns when a gloas network's execution client does not advertise `engine_getPayloadV6`. -- **`engine_getBlobs*`.** There is no blob-pool fetch path here; data columns come - from peers over the network. - **`engine_getPayloadBodies*`.** Nothing consumes them. Blocks before electra are not asked about at all. This node checkpoint-syncs onto @@ -239,6 +239,88 @@ costs one retry ladder per slot however many envelopes wait. This is the gloas counterpart of the limitation below, and unlike a block the envelope is not dropped. +## Blobs from the execution client + +A fulu block is held until its custody data columns are stored. Gossip usually +delivers them within a fraction of a second, but the execution client may +already have every blob the block commits to, since the blob transactions went +through its pool first. The fulu p2p spec ("Distributed blob publishing using +blobs retrieved from local execution-layer client") says honest nodes SHOULD +ask `engine_getBlobsV2` on receiving a valid `beacon_block` or +`data_column_sidecar` from gossip, and turn a complete answer into data columns. +The code is `crates/blockchain/src/beacon_blobs.rs`. + +**When it fires.** On the first data-availability hold of a block +(`hold_block_for_columns`), when all of these hold: the block is fulu, it +commits to at least one blob, and `--execution-endpoint` is set. Any slot and +any source (gossip or sync) qualifies. A redelivery of a held root does not ask +again. + +**The call.** One attempt with a 1 s timeout (`ENGINE_GET_BLOBS_TIMEOUT`, the +value `osaka.md` gives the method), outside the retry ladder. The call races +gossip delivery of the same columns, and a retry after the ladder's first +backoff would land after that race is decided. A miss leaves the block to +gossip and the per-slot peer asks, which is where it was anyway. The answer is +all-or-nothing: `null` means the execution client lacks at least one blob (or is +syncing), and counts as a miss. + +**V2, not V3.** V3 answers partially, but a column holds one cell from every +blob, so a partial blob set builds no complete column, and this node has no +partial-column assembly. + +**The build.** On a full answer, a blocking task first re-checks that the block +was not imported meanwhile and which custody columns are still missing from the +store, then builds only those (cells and sidecars) off the async runtime and off +the chain actor. It does **not** verify the KZG proofs: the spec says a node +SHOULD skip verifying blobs from its own execution client. + +**The hand-over.** The sidecars go to p2p through the existing +`check_data_column_sidecars`, with `SidecarOrigin::ExecutionLayer`. That origin +runs no checks. p2p records each column in the gossip seen cache first (so a +copy gossip delivers later is ignored, and a column gossip already accepted is +not published twice), publishes the newly recorded ones on their subnet topics +if and only if this node is subscribed to them, which here means its custody +columns, and forwards one batch to the chain actor. The chain actor stores the +columns and releases the held block through the usual completeness check. + +``` +hold (first time) -> engine_getBlobsV2 1 s, one attempt + |-- null / error -> count; gossip and peer asks carry on + '-- every blob -> build missing custody columns + (block imported meanwhile? stop) + | + check_data_column_sidecars(ExecutionLayer) + | + p2p: seen cache -> publish on subscribed subnets -> forward one batch + | + chain actor: store columns -> release the held block +``` + +**Trust.** The proofs are taken as the execution client gives them. One that +returns wrong proofs makes this node store and publish columns its peers' +gossip validation rejects, which costs this node peer score. Debug builds +re-run the chain checks on everything `new_data_column_sidecars` delivers and +would catch it. + +**Not done.** + +- No ask when a sidecar arrives, only on a block's first hold. +- Gloas: data availability moves to the payload envelope, so a held gloas block + has no columns to fill this way. +- Partial columns (see V3 above). + +**Metrics.** + +| Metric | Reads | +|---|---| +| `lean_engine_get_blobs_total{outcome}` | Attempts: `hit`, `miss`, `error` | +| `lean_engine_get_blobs_duration_seconds` | One round trip, whatever the outcome | +| `lean_engine_blobs_to_columns_seconds` | Cell computation plus sidecar build | +| `lean_data_columns_from_execution_layer_total{result}` | What p2p did with each column: `published`, `no_digest`, `already_seen`, `not_custody` | + +A high `miss` rate means the execution client's pool does not have the blobs. +That is normal for old blocks and while the execution client is syncing. + ## `forkchoiceUpdated` Sent once after every head recompute, which is once per import cascade and once @@ -298,6 +380,8 @@ authentication, so an endpoint without a secret is refused at startup, before anything binds or connects. Clap cannot express "both or neither" across two optional flags, so the pairing is checked once, in `From`. +Blob fetching has no flag of its own: it is on whenever `--execution-endpoint` is set. + With **neither** flag, the follower contacts no execution client and every block gets `PayloadValidity::NotRequired`, which is exactly what it did before any of this existed. diff --git a/docs/metrics.md b/docs/metrics.md index 7107778d1..f8554ea4d 100644 --- a/docs/metrics.md +++ b/docs/metrics.md @@ -600,6 +600,17 @@ spec. | `lean_envelopes_awaiting_block` | Gauge | Gloas execution payload envelopes held until their block is imported | On every hold, release, and finality eviction of the held envelopes | | | | `lean_envelopes_awaiting_columns` | Gauge | Gloas execution payload envelopes held until every sampled column is stored | On every hold, release, and finality eviction of the held envelopes | | | | `lean_blocks_awaiting_parent_payload` | Gauge | Gloas blocks held until their FULL parent's payload envelope is verified | On every hold, release, and finality eviction of the held blocks | | | +| `lean_engine_get_blobs_total` | Counter | `engine_getBlobsV2` attempts for held fulu blocks, by outcome | On each answer of the one attempt made at a block's first hold | outcome=hit,miss,error | | +| `lean_engine_get_blobs_duration_seconds` | Histogram | Time for one `engine_getBlobsV2` round trip | On each attempt, whatever its outcome | | 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0 | +| `lean_engine_blobs_to_columns_seconds` | Histogram | Time to build custody data column sidecars from execution-client blobs | On each build that reaches the cell computation, off the actor thread | | 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.0 | +| `lean_data_columns_from_execution_layer_total` | Counter | Data column sidecars built from the execution client's blobs, by what p2p did with them | On each hand-over, once per result, incremented by the column count | result=published,no_digest,already_seen,not_custody | | + +`lean_engine_get_blobs_total` and its siblings cover the execution-client path +(`docs/beacon_engine.md`, "Blobs from the execution client"). A high `miss` +share means the execution client's blob pool lacks the blobs, which is normal +for old blocks and for a syncing execution client; `already_seen` in +`lean_data_columns_from_execution_layer_total` counts columns gossip delivered +first. `lean_data_columns_rejected_total` counts the chain checks, which run in the p2p layer on every sidecar gossip did not accept: every fetched sidecar, a From 3ef0e531785d257bd72e26aa66b92fb7369ac561 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Wed, 11 Nov 2026 16:15:49 -0300 Subject: [PATCH 2/2] fix(beacon): ask the execution client for blobs only for a block its proposer signed A held fulu block's blob fetch builds columns that carry the block's own signed header, and p2p records and publishes those with no check (SidecarOrigin::ExecutionLayer). A block reaches its hold unverified when it comes from range sync or a by-root answer, since only gossip checks the proposer signature before data availability. A peer serving a forged copy whose commitments name blobs in the local pool therefore had this node publish columns its peers reject, and take the seen-cache key of the real block's columns from the same proposer and slot. The first hold now runs precheck_block against the parent's post-state (expected proposer and signature) before asking. The tests sign their blocks over a keyed parent state, and a new one checks that an altered copy is not asked about. --- crates/blockchain/Cargo.toml | 3 + crates/blockchain/src/beacon_blobs.rs | 151 +++++++++++++++++++++++--- crates/blockchain/src/lib.rs | 6 +- docs/beacon_engine.md | 16 ++- 4 files changed, 158 insertions(+), 18 deletions(-) diff --git a/crates/blockchain/Cargo.toml b/crates/blockchain/Cargo.toml index f6aaa1136..3f761d8c2 100644 --- a/crates/blockchain/Cargo.toml +++ b/crates/blockchain/Cargo.toml @@ -35,6 +35,9 @@ tracing.workspace = true hex.workspace = true [dev-dependencies] +# `test_state`'s validator keys, so a test can sign a block that has to pass +# the proposer-signature precheck before its blobs are asked for. +ethlambda-state-transition = { workspace = true, features = ["test-utils"] } hex = { workspace = true } libssz.workspace = true libssz-types.workspace = true diff --git a/crates/blockchain/src/beacon_blobs.rs b/crates/blockchain/src/beacon_blobs.rs index 55d2820b9..8c5bc8cc9 100644 --- a/crates/blockchain/src/beacon_blobs.rs +++ b/crates/blockchain/src/beacon_blobs.rs @@ -31,10 +31,11 @@ //! `beacon_block` or `data_column_sidecar` from gossip, convert a complete //! answer to data columns and import them as if received on gossip. This //! node asks on the first data-availability hold of a block, whether the -//! block came from gossip or from sync; it does not yet ask when a sidecar -//! arrives. Hence the hand-over to p2p rather than a direct store write: -//! publishing on subscribed subnets and updating the anti-equivocation cache -//! are MUSTs, and p2p owns both. +//! block came from gossip or from sync, once its proposer signature verifies +//! against its parent's state (see [`signed_by_its_proposer`]); it does not +//! yet ask when a sidecar arrives. Hence the hand-over to p2p rather than a +//! direct store write: publishing on subscribed subnets and updating the +//! anti-equivocation cache are MUSTs, and p2p owns both. //! - **One attempt.** The call races gossip delivery of the same columns, the //! specification's own timeout for the method is 1 s, and a retry after the //! engine ladder's backoff would land after that race is decided. A miss @@ -62,16 +63,65 @@ use std::time::Instant; use ethlambda_engine::EngineClient; use ethlambda_network_api::{BlockChainToP2PRef, SidecarOrigin}; use ethlambda_state_transition::beacon::data_columns; +use ethlambda_state_transition::beacon::precheck::{Reference, precheck_block}; use ethlambda_state_transition::beacon::stf::deneb::kzg_commitment_to_versioned_hash; use ethlambda_storage::Store; use ethlambda_types::ShortRoot; -use ethlambda_types::beacon::containers::{DataColumnSidecar, deneb, electra}; +use ethlambda_types::beacon::containers::{DataColumnSidecar, SignedBeaconBlock, deneb, electra}; use ethlambda_types::beacon::primitives::KzgProof; use ethlambda_types::primitives::H256; use tracing::{debug, error, info, warn}; use crate::metrics; +/// Whether a held block may have its columns built from the execution +/// client's blobs: its proposer signature, and the rest of +/// [`precheck_block`]'s rules, hold against its parent's post-state. +/// +/// The columns built here carry the block's own signed header and skip every +/// check on their way out (see `SidecarOrigin::ExecutionLayer`): p2p records +/// them in the seen cache under `(slot, proposer, index)` and publishes them. +/// A block can reach its hold unverified, since range sync and by-root +/// answers never pass gossip validation and the import checks the signature +/// only after data availability. Without this, a peer serving a forged copy +/// whose commitments name blobs in the local pool would have this node publish +/// columns its peers reject, and take the seen-cache key of the real block's +/// columns from the same proposer and slot. +/// +/// The parent has a post-state whenever the import reaches a hold; a hold +/// without one is not asked about. +pub(crate) fn signed_by_its_proposer( + store: &Store, + block: &SignedBeaconBlock, + block_root: H256, +) -> bool { + let slot = block.slot(); + let Some(parent_state) = store + .get_state(&block.parent_root()) + .expect("DB read should succeed") + else { + debug!( + %slot, + block_root = %ShortRoot(&block_root.0), + "Not asking the execution client for blobs: the block's parent has no post-state" + ); + return false; + }; + let config = store.config(); + precheck_block(block, block_root, Reference::Parent(&parent_state), &config) + .inspect_err(|err| { + let proposer = block.proposer_index(); + warn!( + %slot, + proposer, + block_root = %ShortRoot(&block_root.0), + %err, + "Not asking the execution client for blobs: the block fails its precheck" + ) + }) + .is_ok() +} + /// Ask the execution client for a held fulu block's blobs on a task of its /// own, and hand the custody columns built from them to the p2p layer. pub(crate) fn spawn_fetch( @@ -226,8 +276,52 @@ mod tests { use super::*; use crate::import_timing::ImportTimings; - use crate::tests::{beacon_server_recording, beacon_store, fulu_block}; - use ethlambda_types::beacon::preset::BYTES_PER_BLOB; + use crate::tests::{bare_block, bare_state, beacon_server_recording, beacon_store, fulu_block}; + use ethlambda_state_transition::beacon::helpers::test_state; + use ethlambda_types::beacon::config::Config; + use ethlambda_types::beacon::constants::DOMAIN_BEACON_PROPOSER; + use ethlambda_types::beacon::fork::ForkName; + use ethlambda_types::beacon::preset::{self, BYTES_PER_BLOB}; + use ethlambda_types::beacon::signing::{ + compute_domain, compute_epoch_at_slot, compute_signing_root, + }; + + /// `beacon_store(0, 0)` whose anchor, the parent of every block here, has + /// a post-state registering validator 0 under the key [`sign`] uses, so a + /// signed block passes the precheck a hold runs before it asks the + /// execution client. + fn store_with_keyed_parent() -> Store { + let keyed = test_state::with_signing_validators_at(ForkName::Phase0, 1); + let validator = keyed.validator(0).expect("one validator").clone(); + let mut state = bare_state(); + state + .push_validator(validator, preset::MAX_EFFECTIVE_BALANCE) + .expect("one validator fits the registry"); + state.apply_pending_mutations(); + + let mut store = beacon_store(0, 0); + store + .insert_signed_block(H256::ZERO, bare_block(0, H256::repeat_byte(0xcc))) + .expect("insert"); + store.insert_state(H256::ZERO, state).expect("insert"); + store + } + + /// `block` signed by validator 0, its proposer, under the mainnet fork + /// schedule `beacon_store` runs on. + fn sign(mut block: SignedBeaconBlock) -> SignedBeaconBlock { + let config = Config::mainnet(); + let fork_version = + config.fork_version(config.fork_at_epoch(compute_epoch_at_slot(block.slot()))); + // `bare_state`'s genesis validators root. + let domain = compute_domain(DOMAIN_BEACON_PROPOSER, fork_version, H256::ZERO); + let signing_root = compute_signing_root(block.message_hash_tree_root(), domain); + let SignedBeaconBlock::Fulu(inner) = &mut block else { + unreachable!("these tests build fulu blocks"); + }; + inner.signature = test_state::sign_for(inner.message.proposer_index as usize, signing_root); + block + } /// A blob whose every field element is below the BLS modulus. fn blob_bytes(seed: u8) -> Vec { @@ -325,7 +419,8 @@ mod tests { (client, requests) } - /// A fulu block at slot 1 committing to each of `blobs`. + /// A fulu block at slot 1 committing to each of `blobs`, signed by its + /// proposer. fn block_committing_to(blobs: &[Vec]) -> SignedBeaconBlock { let mut block = fulu_block(H256::ZERO, 1, 0); let commitments: Vec = blobs @@ -336,7 +431,7 @@ mod tests { unreachable!("fulu_block builds a fulu block"); }; inner.message.body.blob_kzg_commitments = SszList::try_from(commitments).unwrap(); - block + sign(block) } const CUSTODY: [u64; 3] = [0, 5, 77]; @@ -357,7 +452,7 @@ mod tests { let block = block_committing_to(&blobs); let root = block.message_hash_tree_root(); let (engine, requests) = mock_engine(answer_for(&blobs)).await; - let (mut server, p2p) = beacon_server_recording(beacon_store(0, 0)); + let (mut server, p2p) = beacon_server_recording(store_with_keyed_parent()); server.engine = Some(engine); server.custody_columns = CUSTODY.to_vec(); @@ -397,7 +492,7 @@ mod tests { let block = block_committing_to(&blobs); let root = block.message_hash_tree_root(); let (engine, _requests) = mock_engine(answer_for(&blobs)).await; - let (mut server, p2p) = beacon_server_recording(beacon_store(0, 0)); + let (mut server, p2p) = beacon_server_recording(store_with_keyed_parent()); server.engine = Some(engine); server.custody_columns = CUSTODY.to_vec(); server @@ -425,7 +520,7 @@ mod tests { let blobs = vec![blob_bytes(4)]; let block = block_committing_to(&blobs); let (engine, requests) = mock_engine("null".to_string()).await; - let (mut server, p2p) = beacon_server_recording(beacon_store(0, 0)); + let (mut server, p2p) = beacon_server_recording(store_with_keyed_parent()); server.engine = Some(engine); server.custody_columns = CUSTODY.to_vec(); @@ -446,7 +541,7 @@ mod tests { async fn without_an_engine_nothing_is_asked_or_handed_over() { let blobs = vec![blob_bytes(5)]; let block = block_committing_to(&blobs); - let (mut server, p2p) = beacon_server_recording(beacon_store(0, 0)); + let (mut server, p2p) = beacon_server_recording(store_with_keyed_parent()); server.custody_columns = CUSTODY.to_vec(); server.hold_block_for_columns(block, 1, ImportTimings::default()); @@ -460,7 +555,7 @@ mod tests { let blobs = vec![blob_bytes(6)]; let block = block_committing_to(&blobs); let (engine, requests) = mock_engine("null".to_string()).await; - let (mut server, _p2p) = beacon_server_recording(beacon_store(0, 0)); + let (mut server, _p2p) = beacon_server_recording(store_with_keyed_parent()); server.engine = Some(engine); server.custody_columns = CUSTODY.to_vec(); @@ -476,4 +571,32 @@ mod tests { assert_eq!(requests.load(Ordering::SeqCst), 1); } + + /// An altered copy of a block (its body changed after signing, the + /// signature kept) whose commitments name blobs the execution client + /// holds: the columns would carry a header nobody signed, so the client is + /// not asked and nothing reaches p2p's seen cache or the wire. + #[tokio::test] + async fn a_block_whose_signature_does_not_verify_is_not_asked_about() { + let blobs = vec![blob_bytes(7)]; + let mut block = block_committing_to(&blobs); + let SignedBeaconBlock::Fulu(inner) = &mut block else { + unreachable!("fulu_block builds a fulu block"); + }; + inner.message.body.graffiti = H256::repeat_byte(0x01); + let (engine, requests) = mock_engine(answer_for(&blobs)).await; + let (mut server, p2p) = beacon_server_recording(store_with_keyed_parent()); + server.engine = Some(engine); + server.custody_columns = CUSTODY.to_vec(); + + server.hold_block_for_columns(block, 1, ImportTimings::default()); + tokio::time::sleep(Duration::from_millis(300)).await; + + assert_eq!( + requests.load(Ordering::SeqCst), + 0, + "the engine was not asked" + ); + assert!(p2p.checks.lock().unwrap().is_empty()); + } } diff --git a/crates/blockchain/src/lib.rs b/crates/blockchain/src/lib.rs index 59d52b2d7..79d0dddcd 100644 --- a/crates/blockchain/src/lib.rs +++ b/crates/blockchain/src/lib.rs @@ -3145,7 +3145,8 @@ impl BlockChainServer { /// slot. /// /// The execution client is asked for the block's blobs once, on the first - /// hold of a fulu block that commits to any (see [`beacon_blobs`]). That + /// hold of a fulu block that commits to any and whose proposer signature + /// verifies against its parent's state (see [`beacon_blobs`]). That /// does not depend on the peer-ask timing above: it is a local call that /// races nobody, and a hit hands the custody columns back through the same /// path gossip uses, so it is as useful for a block at the tip as for an @@ -3179,10 +3180,13 @@ impl BlockChainServer { // Before the block is moved into the store. Only the first hold asks: // a redelivery of a held root must not re-query the execution client. + // Only a block whose proposer signature verifies does, since the + // columns built from its blobs carry its header unchecked. if let (Some(engine), Some(p2p), SignedBeaconBlock::Fulu(inner)) = (&self.engine, &self.p2p, &block) && !inner.message.body.blob_kzg_commitments.is_empty() && !self.blocks_awaiting_columns.contains_key(&block_root) + && beacon_blobs::signed_by_its_proposer(&self.store, &block, block_root) { beacon_blobs::spawn_fetch( engine.clone(), diff --git a/docs/beacon_engine.md b/docs/beacon_engine.md index 11fc0b095..e398aff27 100644 --- a/docs/beacon_engine.md +++ b/docs/beacon_engine.md @@ -252,9 +252,19 @@ The code is `crates/blockchain/src/beacon_blobs.rs`. **When it fires.** On the first data-availability hold of a block (`hold_block_for_columns`), when all of these hold: the block is fulu, it -commits to at least one blob, and `--execution-endpoint` is set. Any slot and -any source (gossip or sync) qualifies. A redelivery of a held root does not ask -again. +commits to at least one blob, `--execution-endpoint` is set, and the block +passes `precheck_block` against its parent's post-state (expected proposer and +proposer signature). Any slot and any source (gossip or sync) qualifies. A +redelivery of a held root does not ask again. + +The precheck is there because the built columns carry the block's own signed +header and skip every check on their way out (see the hand-over below). Range +sync and by-root answers never pass gossip validation, and the import checks +the signature only after data availability, so without it a peer serving a +forged copy of a block, whose commitments name blobs in this node's pool, would +have this node publish columns its peers reject, and take the seen-cache key of +the real block's columns from the same proposer and slot. It costs one BLS +verification per first hold of a block that would be asked about. **The call.** One attempt with a 1 s timeout (`ENGINE_GET_BLOBS_TIMEOUT`, the value `osaka.md` gives the method), outside the retry ladder. The call races