feat(beacon): rebuild held blocks' data columns from the execution client's blob pool

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

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

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"),
         );
     }

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,14 @@ 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 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
+    /// old one.
     fn hold_block_for_columns(
         &mut self,
         block: SignedBeaconBlock,
@@ -3169,6 +3178,26 @@ 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.
+        // 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(),
+                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 +4694,8 @@ mod tests {
     #[derive(Default)]
     pub(crate) struct RecordingP2P {
         pub(crate) checks: std::sync::Mutex<Vec<Vec<DataColumnSidecar>>>,
+        /// The origin each entry of `checks` was handed over with, in order.
+        pub(crate) check_origins: std::sync::Mutex<Vec<ethlambda_network_api::SidecarOrigin>>,
         pub(crate) fetches: std::sync::Mutex<Vec<FetchRequest>>,
     }
 
@@ -4697,8 +4728,10 @@ mod tests {
         fn check_data_column_sidecars(
             &self,
             sidecars: Vec<DataColumnSidecar>,
+            origin: ethlambda_network_api::SidecarOrigin,
         ) -> Result<(), spawned_concurrency::error::ActorError> {
             self.checks.lock().unwrap().push(sidecars);
+            self.check_origins.lock().unwrap().push(origin);
             Ok(())
         }
     }

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<IntCounterVec> =
+        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<Histogram> =
+        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<Histogram> =
+        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());
+}