feat(l1): engine_getBlobsV1 request endpoint (#3636)

**Motivation**

We need to implement the RPC endpoint as it will be needed for Fusaka.

**Description**

This pr incorporates a new module to handle RPC endpoint for the
**engine_getBlobsV1** request according to the following spec's
description from
[here](https://ethereum.github.io/execution-apis/api-documentation/) and
[here](https://github.com/ethereum/execution-apis/blob/main/src/engine/cancun.md#engine_getblobsv1).

To check the messages are sent correctly you can set the
`fixtures/network/network_params.yaml` like this:
```bash
participants:
  # NOTE: Both erigon and geth work on this example, but they provide wrong nodes information on discovery protocol
  # - el_type: erigon
  #   el_image: ethpandaops/erigon:main-764a2c50
  #   cl_type: lighthouse
  #   cl_image: sigp/lighthouse:v7.0.0-beta.0
  #   validator_count: 32
  # - el_type: reth
  #   el_image: ghcr.io/paradigmxyz/reth:v1.2.2
  #   cl_type: lighthouse
  #   cl_image: sigp/lighthouse:v7.0.0-beta.0
  #   validator_count: 32
  - el_type: besu
    el_image: ethpandaops/besu:main-142a5e6
    cl_type: lighthouse
    cl_image: sigp/lighthouse:v7.0.0-beta.0
    validator_count: 32
  - el_type: geth
    el_image: ethereum/client-go:v1.15.2
    cl_type: lighthouse
    cl_image: sigp/lighthouse:v7.0.0-beta.0
    validator_count: 32
    count: 1
  - el_type: ethrex
    cl_type: lighthouse
    cl_image: sigp/lighthouse:v7.0.0-beta.0
    validator_count: 32
    snooper_enabled: true

network_params:
  electra_fork_epoch: 1

  # The address of the staking contract address on the Eth1 chain
  deposit_contract_address: "0x4242424242424242424242424242424242424242"

ethereum_metrics_exporter_enabled: true

additional_services:
  - dora
  - forkmon
  - spamoor

blockscout_params:
  image: "blockscout/blockscout:latest"
  verif_image: "ghcr.io/blockscout/smart-contract-verifier:latest"
  frontend_image: "ghcr.io/blockscout/frontend:latest"

prometheus_params:
  # TODO: switch to latest when it points to v3.x
  image: "prom/prometheus:v3.2.1"
```

Then run in a console:
```bash
make localnet
```

And run in another console:
```bash
docker logs -f $(docker ps -q --filter name=snooper-engine-3-lighthouse-ethrex)
```

This last console will show all the rpc requests and the responses. It's
overwhelming. Here I left some pictures of my logs:

<img width="785" height="411" alt="Screenshot 2025-07-22 at 16 07 22"
src="https://github.com/user-attachments/assets/474ee749-0721-4ae6-8934-fc8c00fae70d"
/>
<img width="855" height="195" alt="Screenshot 2025-07-22 at 16 07 42"
src="https://github.com/user-attachments/assets/be879e7c-244c-4130-9ae9-c4e2976f513d"
/>
<img width="898" height="155" alt="Screenshot 2025-07-22 at 16 07 59"
src="https://github.com/user-attachments/assets/7f8922a0-077f-4361-9eed-2b42dd89dd99"
/>
<img width="843" height="144" alt="Screenshot 2025-07-22 at 16 08 18"
src="https://github.com/user-attachments/assets/eb7e1246-1aba-417f-8217-34cbfec6b95c"
/>
<img width="861" height="129" alt="Screenshot 2025-07-22 at 16 08 34"
src="https://github.com/user-attachments/assets/eee3c6c5-17fd-44f8-8e14-6bff8aba4a63"
/>
<img width="831" height="117" alt="Screenshot 2025-07-22 at 16 08 52"
src="https://github.com/user-attachments/assets/3f450e2a-cd8d-4374-b87c-9f83107c09c3"
/>
<img width="830" height="105" alt="Screenshot 2025-07-22 at 16 09 18"
src="https://github.com/user-attachments/assets/b7a8c40a-02c2-44f3-8f96-129a682a5046"
/>

I haven't found a better way of testing it. I think you can check it
with the dora service but I'm not sure how.

Closes #3428

Author: cdiielsi

cmd/ethrex_replay/src/constants.rs

@@ -33,6 +33,7 @@ pub fn make_chainconfig(chain_id: u64) -> ChainConfig {
         prague_time: Some(0),
         terminal_total_difficulty_passed: false,
         verkle_time: None,
+        osaka_time: None,
         blob_schedule: BlobSchedule::default(),
         // Mainnet address
         deposit_contract_address: H160::from_str("0x00000000219ab540356cbb839cbe05303d7705fa")

crates/blockchain/mempool.rs

@@ -232,6 +232,15 @@ impl Mempool {
             .collect())
     }
 
+    /// Returns all blobs bundles currently in the pool
+    pub fn get_blobs_bundle_pool(&self) -> Result<Vec<BlobsBundle>, MempoolError> {
+        let blobs_bundle_pool = self
+            .blobs_bundle_pool
+            .lock()
+            .map_err(|error| StoreError::MempoolReadLock(error.to_string()))?;
+        Ok(blobs_bundle_pool.values().cloned().collect())
+    }
+
     /// Returns the status of the mempool, which is the number of transactions currently in
     /// the pool. Until we add "queue" transactions.
     pub fn status(&self) -> Result<usize, MempoolError> {

crates/common/serde_utils.rs

@@ -388,6 +388,13 @@ pub mod bool {
 pub mod bytes48 {
     use super::*;
 
+    pub fn serialize<S>(value: &[u8; 48], serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        serializer.serialize_str(&format!("0x{}", hex::encode(value)))
+    }
+
     pub mod vec {
         use super::*;
 
@@ -424,10 +431,16 @@ pub mod bytes48 {
 
 pub mod blob {
     use super::*;
+    use crate::types::BYTES_PER_BLOB;
 
-    pub mod vec {
-        use crate::types::BYTES_PER_BLOB;
+    pub fn serialize<S>(value: &[u8; BYTES_PER_BLOB], serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        serializer.serialize_str(&format!("0x{}", hex::encode(value)))
+    }
 
+    pub mod vec {
         use super::*;
 
         pub fn serialize<S>(

crates/common/types/genesis.rs

@@ -186,6 +186,7 @@ pub struct ChainConfig {
     pub cancun_time: Option<u64>,
     pub prague_time: Option<u64>,
     pub verkle_time: Option<u64>,
+    pub osaka_time: Option<u64>,
 
     /// Amount of total difficulty reached by the network that triggers the consensus upgrade.
     pub terminal_total_difficulty: Option<u128>,
@@ -253,6 +254,10 @@ impl From<Fork> for &str {
 }
 
 impl ChainConfig {
+    pub fn is_osaka_activated(&self, block_timestamp: u64) -> bool {
+        self.osaka_time.is_some_and(|time| time <= block_timestamp)
+    }
+
     pub fn is_prague_activated(&self, block_timestamp: u64) -> bool {
         self.prague_time.is_some_and(|time| time <= block_timestamp)
     }
@@ -279,7 +284,9 @@ impl ChainConfig {
     }
 
     pub fn get_fork(&self, block_timestamp: u64) -> Fork {
-        if self.is_prague_activated(block_timestamp) {
+        if self.is_osaka_activated(block_timestamp) {
+            Fork::Osaka
+        } else if self.is_prague_activated(block_timestamp) {
             Fork::Prague
         } else if self.is_cancun_activated(block_timestamp) {
             Fork::Cancun

crates/networking/rpc/engine/blobs.rs

@@ -0,0 +1,97 @@
+use ethrex_common::{
+    H256,
+    serde_utils::{self},
+    types::{Blob, Proof, blobs_bundle::kzg_commitment_to_versioned_hash},
+};
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+use tracing::info;
+
+use crate::{
+    rpc::{RpcApiContext, RpcHandler},
+    utils::RpcErr,
+};
+
+// -> https://github.com/ethereum/execution-apis/blob/d41fdf10fabbb73c4d126fb41809785d830acace/src/engine/cancun.md?plain=1#L186
+const GET_BLOBS_V1_REQUEST_MAX_SIZE: usize = 128;
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct BlobsV1Request {
+    blob_versioned_hashes: Vec<H256>,
+}
+
+#[derive(Clone, Debug, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct BlobAndProofV1 {
+    #[serde(with = "serde_utils::blob")]
+    pub blob: Blob,
+    #[serde(with = "serde_utils::bytes48")]
+    pub proof: Proof,
+}
+
+impl RpcHandler for BlobsV1Request {
+    fn parse(params: &Option<Vec<Value>>) -> Result<Self, RpcErr> {
+        let params = params
+            .as_ref()
+            .ok_or(RpcErr::BadParams("No params provided".to_owned()))?;
+        if params.len() != 1 {
+            return Err(RpcErr::BadParams("Expected 1 param".to_owned()));
+        };
+        Ok(BlobsV1Request {
+            blob_versioned_hashes: serde_json::from_value(params[0].clone())?,
+        })
+    }
+
+    async fn handle(&self, context: RpcApiContext) -> Result<Value, RpcErr> {
+        info!("Received new engine request: Requested Blobs");
+        if self.blob_versioned_hashes.len() >= GET_BLOBS_V1_REQUEST_MAX_SIZE {
+            return Err(RpcErr::TooLargeRequest);
+        }
+
+        if let Some(current_block_header) = context
+            .storage
+            .get_block_header(context.storage.get_latest_block_number().await?)?
+        {
+            if context
+                .storage
+                .get_chain_config()?
+                .is_osaka_activated(current_block_header.timestamp)
+            {
+                // validation requested in https://github.com/ethereum/execution-apis/blob/a1d95fb555cd91efb3e0d6555e4ab556d9f5dd06/src/engine/osaka.md?plain=1#L130
+                return Err(RpcErr::UnsuportedFork(
+                    "getBlobsV1 engine request not supported for Osaka".to_string(),
+                ));
+            }
+        };
+
+        let mut res: Vec<Option<BlobAndProofV1>> = vec![None; self.blob_versioned_hashes.len()];
+
+        for blobs_bundle in context.blockchain.mempool.get_blobs_bundle_pool()? {
+            // Go over all blobs bundles from the blobs bundle pool.
+            let blobs_in_bundle = blobs_bundle.blobs;
+            let commitments_in_bundle = blobs_bundle.commitments;
+            let proofs_in_bundle = blobs_bundle.proofs;
+
+            // Go over all the commitments in each blobs bundle to calculate the blobs versioned hash.
+            for (commitment, (blob, proof)) in commitments_in_bundle
+                .iter()
+                .zip(blobs_in_bundle.iter().zip(proofs_in_bundle.iter()))
+            {
+                let current_versioned_hash = kzg_commitment_to_versioned_hash(commitment);
+                if let Some(index) = self
+                    .blob_versioned_hashes
+                    .iter()
+                    .position(|&hash| hash == current_versioned_hash)
+                {
+                    // If the versioned hash is one of the requested we save its corresponding blob and proof in the returned vector. We store them in the same position as the versioned hash was received.
+                    res[index] = Some(BlobAndProofV1 {
+                        blob: *blob,
+                        proof: *proof,
+                    });
+                }
+            }
+        }
+
+        serde_json::to_value(res).map_err(|error| RpcErr::Internal(error.to_string()))
+    }
+}

crates/networking/rpc/engine/mod.rs

@@ -1,3 +1,4 @@
+pub mod blobs;
 pub mod exchange_transition_config;
 pub mod fork_choice;
 pub mod payload;
@@ -13,7 +14,7 @@ pub type ExchangeCapabilitiesRequest = Vec<String>;
 
 /// List of capabilities that the execution layer client supports. Add new capabilities here.
 /// More info: https://github.com/ethereum/execution-apis/blob/main/src/engine/common.md#engine_exchangecapabilities
-pub const CAPABILITIES: [&str; 14] = [
+pub const CAPABILITIES: [&str; 15] = [
     "engine_forkchoiceUpdatedV1",
     "engine_forkchoiceUpdatedV2",
     "engine_forkchoiceUpdatedV3",
@@ -28,6 +29,7 @@ pub const CAPABILITIES: [&str; 14] = [
     "engine_exchangeTransitionConfigurationV1",
     "engine_getPayloadBodiesByHashV1",
     "engine_getPayloadBodiesByRangeV1",
+    "engine_getBlobsV1",
 ];
 
 impl From<ExchangeCapabilitiesRequest> for RpcRequest {

crates/networking/rpc/rpc.rs

@@ -1,6 +1,7 @@
 use crate::authentication::authenticate;
 use crate::engine::{
     ExchangeCapabilitiesRequest,
+    blobs::BlobsV1Request,
     exchange_transition_config::ExchangeTransitionConfigV1Req,
     fork_choice::{ForkChoiceUpdatedV1, ForkChoiceUpdatedV2, ForkChoiceUpdatedV3},
     payload::{
@@ -376,6 +377,7 @@ pub async fn map_engine_requests(
         "engine_getPayloadBodiesByRangeV1" => {
             GetPayloadBodiesByRangeV1Request::call(req, context).await
         }
+        "engine_getBlobsV1" => BlobsV1Request::call(req, context).await,
         unknown_engine_method => Err(RpcErr::MethodNotFound(unknown_engine_method.to_owned())),
     }
 }
@@ -509,6 +511,7 @@ mod tests {
                         "cancunTime": 0,
                         "pragueTime": 1718232101,
                         "verkleTime": null,
+                        "osakaTime": null,
                         "terminalTotalDifficulty": 0,
                         "terminalTotalDifficultyPassed": true,
                         "blobSchedule": blob_schedule,