diff --git a/CLAUDE.md b/CLAUDE.md
index d08ee404f..404f21495 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/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
new file mode 100644
index 000000000..8c5bc8cc9
--- /dev/null
+++ b/crates/blockchain/src/beacon_blobs.rs
@@ -0,0 +1,602 @@
+//! 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, 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
+//!   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::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, 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(
+    engine: EngineClient,
+    p2p: BlockChainToP2PRef,
+    store: Store,
+    block: electra::SignedBeaconBlock,
+    block_root: H256,
+    custody_columns: Vec<u64>,
+) {
+    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<u64>,
+) {
+    let slot = block.message.slot;
+    let versioned_hashes: Vec<H256> = 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<u64> = 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<KzgProof> = 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<DataColumnSidecar> =
+        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::{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<u8> {
+        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<u8>]) -> String {
+        let entries: Vec<String> = blobs
+            .iter()
+            .map(|blob| {
+                let (_, proofs) = kzg::compute_cells_and_kzg_proofs(blob).unwrap();
+                let proofs: Vec<String> = 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<AtomicUsize>) {
+        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<u8> = 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::<usize>().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`, signed by its
+    /// proposer.
+    fn block_committing_to(blobs: &[Vec<u8>]) -> SignedBeaconBlock {
+        let mut block = fulu_block(H256::ZERO, 1, 0);
+        let commitments: Vec<KzgCommitment> = 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();
+        sign(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(store_with_keyed_parent());
+        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<u64> = 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(store_with_keyed_parent());
+        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<u64> = 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(store_with_keyed_parent());
+        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(store_with_keyed_parent());
+        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(store_with_keyed_parent());
+        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);
+    }
+
+    /// 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/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..79d0dddcd 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,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(())
         }
     }
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<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());
+}
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<fulu::Cell> = 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<KzgProof> = 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<CellsPerExtBlob>, &ProofsPerBlob)],
+) -> DataColumnSidecar {
+    let column: Vec<fulu::Cell> = 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<KzgProof> = 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<Vec<DataColumnSidecar>, 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<Box<CellsPerExtBlob>> = blobs
-        .par_iter()
-        .map(|blob| kzg::compute_cells(&blob[..]).map_err(|_| SidecarError::InvalidBlob))
-        .collect::<Result<_, _>>()?;
+    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::<CELLS_PER_EXT_BLOB>();
-    let cells_and_proofs: Vec<(Box<CellsPerExtBlob>, &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<Vec<DataColumnSidecar>, 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<Vec<Box<CellsPerExtBlob>>, 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<Box<CellsPerExtBlob>>,
+    cell_proofs: &[KzgProof],
+) -> Vec<(Box<CellsPerExtBlob>, &ProofsPerBlob)> {
+    let (proof_chunks, _) = cell_proofs.as_chunks::<CELLS_PER_EXT_BLOB>();
+    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<DataColumnSidecar>,
+        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<Vec<u8>, EngineError> {
+pub(crate) fn parse_data(text: &str) -> Result<Vec<u8>, 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<Vec<u8>, EngineError> {
 }
 
 /// A `DATA` of exactly `N` bytes.
-fn parse_fixed<const N: usize>(text: &str) -> Result<[u8; N], EngineError> {
+pub(crate) fn parse_fixed<const N: usize>(text: &str) -> Result<[u8; N], EngineError> {
     parse_data(text)?.try_into().map_err(|bytes: Vec<u8>| {
         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<Option<Vec<BlobAndProofV2>>, EngineError> {
+        let hashes: Vec<String> = versioned_hashes
+            .iter()
+            .map(|hash| data(hash.as_slice()))
+            .collect();
+        let result: Option<Vec<Option<BlobAndProofV2Json>>> = 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<Vec<String>, 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<u8>,
+    /// The blob's cell proofs, exactly `CELLS_PER_EXT_BLOB` of them.
+    pub proofs: Vec<ethlambda_types::beacon::primitives::KzgProof>,
+}
+
+#[derive(Deserialize)]
+pub(crate) struct BlobAndProofV2Json {
+    blob: String,
+    proofs: Vec<String>,
+}
+
+/// 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<Vec<Option<BlobAndProofV2Json>>>,
+    requested: usize,
+) -> Result<Option<Vec<BlobAndProofV2>>, 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::<Result<Vec<_>, _>>()?;
+        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<Mutex<Option<String>>>) {
+async fn serve_once(body: impl Into<String>) -> (String, Arc<Mutex<Option<String>>>) {
     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<String> = (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<DataColumnSidecar>) {
+    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<IdentTopic>)> = 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<RecordingChain>) {
+        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<Vec<u64>> {
+        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<FetchBlock> for P2PServer {
 
 impl Handler<CheckDataColumnSidecars> for P2PServer {
     async fn handle(&mut self, msg: CheckDataColumnSidecars, _ctx: &Context<Self>) {
-        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<ethlambda_types::beacon::primitives::ValidatorIndex>,
             )>,
         >,
+        /// Each `new_data_column_sidecars` batch, in arrival order.
+        pub(crate) sidecar_batches:
+            std::sync::Mutex<Vec<Vec<ethlambda_types::beacon::containers::DataColumnSidecar>>>,
     }
 
     impl ethlambda_network_api::P2PToBlockChain for RecordingChain {
@@ -2867,8 +2875,9 @@ pub(crate) mod test_support {
         }
         fn new_data_column_sidecars(
             &self,
-            _sidecars: Vec<ethlambda_types::beacon::containers::DataColumnSidecar>,
+            sidecars: Vec<ethlambda_types::beacon::containers::DataColumnSidecar>,
         ) -> 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<IntCounterVec> =
+        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..e398aff27 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,98 @@ 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, `--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
+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 +390,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<BeaconOptions>`.
 
+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 9ac92fd4f..c9735224e 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