feat(l1): engine_getBlobsV2 request endpoint

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

@@ -349,6 +349,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::*;
 
@@ -385,10 +392,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

@@ -167,6 +167,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>,
@@ -233,6 +234,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)
     }
@@ -259,7 +264,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,167 @@
+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()))
+    }
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct BlobsV2Request {
+    blob_versioned_hashes: Vec<H256>,
+}
+
+#[derive(Clone, Debug, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct BlobAndProofV2 {
+    #[serde(with = "serde_utils::blob")]
+    pub blob: Blob,
+    #[serde(with = "serde_utils::bytes48::vec")]
+    pub proof: Vec<Proof>,
+}
+
+impl RpcHandler for BlobsV2Request {
+    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(BlobsV2Request {
+            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);
+        }
+
+        let mut res: Vec<Option<BlobAndProofV2>> = 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;
+
+            let cells_per_ext_blob = proofs_in_bundle.len() / blobs_in_bundle.len();
+
+            // Go over all the commitments in each blobs bundle to calculate the blobs versioned hash.
+            for (i, (commitment, blob)) in commitments_in_bundle
+                .iter()
+                .zip(blobs_in_bundle.iter())
+                .enumerate()
+            {
+                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 its proofs in the returned vector. We store them in the same position as the versioned hash was received.
+                    res[index] = Some(BlobAndProofV2 {
+                        blob: *blob,
+                        proof: proofs_in_bundle
+                            [i * cells_per_ext_blob..(i + 1) * cells_per_ext_blob]
+                            .to_vec(),
+                    });
+                }
+            }
+        }
+
+        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; 16] = [
     "engine_forkchoiceUpdatedV1",
     "engine_forkchoiceUpdatedV2",
     "engine_forkchoiceUpdatedV3",
@@ -28,6 +29,8 @@ pub const CAPABILITIES: [&str; 14] = [
     "engine_exchangeTransitionConfigurationV1",
     "engine_getPayloadBodiesByHashV1",
     "engine_getPayloadBodiesByRangeV1",
+    "engine_getBlobsV1",
+    "engine_getBlobsV2",
 ];
 
 impl From<ExchangeCapabilitiesRequest> for RpcRequest {

crates/networking/rpc/rpc.rs

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