feat: add engine_getBlobsV3 method

crates/node/builder/src/rpc.rs

@@ -1381,6 +1381,7 @@ where
             version: version_metadata().cargo_pkg_version.to_string(),
             commit: version_metadata().vergen_git_sha.to_string(),
         };
+
         Ok(EngineApi::new(
             ctx.node.provider().clone(),
             ctx.config.chain.clone(),
@@ -1392,6 +1393,7 @@ where
             EngineCapabilities::default(),
             engine_validator,
             ctx.config.engine.accept_execution_requests_hash,
+            ctx.node.network().clone(),
         ))
     }
 }

crates/optimism/node/src/rpc.rs

@@ -146,6 +146,7 @@ where
             EngineCapabilities::new(OP_ENGINE_CAPABILITIES.iter().copied()),
             engine_validator,
             ctx.config.engine.accept_execution_requests_hash,
+            ctx.node.network().clone(),
         );
 
         Ok(OpEngineApi::new(inner))

crates/rpc/rpc-api/src/engine.rs

@@ -240,6 +240,18 @@ pub trait EngineApi<Engine: EngineTypes> {
         &self,
         versioned_hashes: Vec<B256>,
     ) -> RpcResult<Option<Vec<BlobAndProofV2>>>;
+
+    /// Fetch blobs for the consensus layer from the blob store.
+    ///
+    /// Returns a response of the same length as the request. Missing or older-version blobs are
+    /// returned as `null` elements.
+    ///
+    /// Returns `null` if syncing.
+    #[method(name = "getBlobsV3")]
+    async fn get_blobs_v3(
+        &self,
+        versioned_hashes: Vec<B256>,
+    ) -> RpcResult<Option<Vec<Option<BlobAndProofV2>>>>;
 }
 
 /// A subset of the ETH rpc interface: <https://ethereum.github.io/execution-apis/api-documentation>

crates/rpc/rpc-builder/tests/it/utils.rs

@@ -54,6 +54,7 @@ pub async fn launch_auth(secret: JwtSecret) -> AuthServerHandle {
         EngineCapabilities::default(),
         EthereumEngineValidator::new(MAINNET.clone()),
         false,
+        NoopNetwork::default(),
     );
     let module = AuthRpcModule::new(engine_api);
     module.start_server(config).await.unwrap()

crates/rpc/rpc-engine-api/Cargo.toml

@@ -23,6 +23,7 @@ reth-tasks.workspace = true
 reth-engine-primitives.workspace = true
 reth-transaction-pool.workspace = true
 reth-primitives-traits.workspace = true
+reth-network-api.workspace = true
 
 # ethereum
 alloy-eips.workspace = true

crates/rpc/rpc-engine-api/src/capabilities.rs

@@ -19,6 +19,7 @@ pub const CAPABILITIES: &[&str] = &[
     "engine_getPayloadBodiesByRangeV1",
     "engine_getBlobsV1",
     "engine_getBlobsV2",
+    "engine_getBlobsV3",
 ];
 
 // The list of all supported Engine capabilities available over the engine endpoint.

crates/rpc/rpc-engine-api/src/engine_api.rs

@@ -18,6 +18,7 @@ use async_trait::async_trait;
 use jsonrpsee_core::{server::RpcModule, RpcResult};
 use reth_chainspec::EthereumHardforks;
 use reth_engine_primitives::{ConsensusEngineHandle, EngineApiValidator, EngineTypes};
+use reth_network_api::NetworkInfo;
 use reth_payload_builder::PayloadStore;
 use reth_payload_primitives::{
     validate_payload_timestamp, EngineApiMessageVersion, MessageValidationKind,
@@ -94,7 +95,9 @@ where
         capabilities: EngineCapabilities,
         validator: Validator,
         accept_execution_requests_hash: bool,
+        network: impl NetworkInfo + 'static,
     ) -> Self {
+        let is_syncing = Arc::new(move || network.is_syncing());
         let inner = Arc::new(EngineApiInner {
             provider,
             chain_spec,
@@ -107,6 +110,7 @@ where
             tx_pool,
             validator,
             accept_execution_requests_hash,
+            is_syncing,
         });
         Self { inner }
     }
@@ -792,6 +796,35 @@ where
             .map_err(|err| EngineApiError::Internal(Box::new(err)))
     }
 
+    fn get_blobs_v3(
+        &self,
+        versioned_hashes: Vec<B256>,
+    ) -> EngineApiResult<Option<Vec<Option<BlobAndProofV2>>>> {
+        // Check if Osaka fork is active
+        let current_timestamp =
+            SystemTime::now().duration_since(SystemTime::UNIX_EPOCH).unwrap_or_default().as_secs();
+        if !self.inner.chain_spec.is_osaka_active_at_timestamp(current_timestamp) {
+            return Err(EngineApiError::EngineObjectValidationError(
+                reth_payload_primitives::EngineObjectValidationError::UnsupportedFork,
+            ));
+        }
+
+        if versioned_hashes.len() > MAX_BLOB_LIMIT {
+            return Err(EngineApiError::BlobRequestTooLarge { len: versioned_hashes.len() })
+        }
+
+        // Spec requires returning `null` if syncing.
+        if (*self.inner.is_syncing)() {
+            return Ok(None)
+        }
+
+        self.inner
+            .tx_pool
+            .get_blobs_for_versioned_hashes_v3(&versioned_hashes)
+            .map(Some)
+            .map_err(|err| EngineApiError::Internal(Box::new(err)))
+    }
+
     /// Metered version of `get_blobs_v2`.
     pub fn get_blobs_v2_metered(
         &self,
@@ -827,6 +860,27 @@ where
 
         res
     }
+
+    /// Metered version of `get_blobs_v3`.
+    pub fn get_blobs_v3_metered(
+        &self,
+        versioned_hashes: Vec<B256>,
+    ) -> EngineApiResult<Option<Vec<Option<BlobAndProofV2>>>> {
+        let hashes_len = versioned_hashes.len();
+        let start = Instant::now();
+        let res = Self::get_blobs_v3(self, versioned_hashes);
+        self.inner.metrics.latency.get_blobs_v3.record(start.elapsed());
+
+        if let Ok(Some(blobs)) = &res {
+            let blobs_found = blobs.iter().flatten().count();
+            let blobs_missed = hashes_len - blobs_found;
+
+            self.inner.metrics.blob_metrics.blob_count.increment(blobs_found as u64);
+            self.inner.metrics.blob_metrics.blob_misses.increment(blobs_missed as u64);
+        }
+
+        res
+    }
 }
 
 // This is the concrete ethereum engine API implementation.
@@ -1099,6 +1153,14 @@ where
         trace!(target: "rpc::engine", "Serving engine_getBlobsV2");
         Ok(self.get_blobs_v2_metered(versioned_hashes)?)
     }
+
+    async fn get_blobs_v3(
+        &self,
+        versioned_hashes: Vec<B256>,
+    ) -> RpcResult<Option<Vec<Option<BlobAndProofV2>>>> {
+        trace!(target: "rpc::engine", "Serving engine_getBlobsV3");
+        Ok(self.get_blobs_v3_metered(versioned_hashes)?)
+    }
 }
 
 impl<Provider, EngineT, Pool, Validator, ChainSpec> IntoEngineApiRpcModule
@@ -1155,17 +1217,22 @@ struct EngineApiInner<Provider, PayloadT: PayloadTypes, Pool, Validator, ChainSp
     /// Engine validator.
     validator: Validator,
     accept_execution_requests_hash: bool,
+    /// Returns `true` if the node is currently syncing.
+    is_syncing: Arc<dyn Fn() -> bool + Send + Sync>,
 }
 
 #[cfg(test)]
 mod tests {
     use super::*;
     use alloy_rpc_types_engine::{ClientCode, ClientVersionV1};
     use assert_matches::assert_matches;
-    use reth_chainspec::{ChainSpec, MAINNET};
+    use reth_chainspec::{ChainSpec, ChainSpecBuilder, MAINNET};
     use reth_engine_primitives::BeaconEngineMessage;
     use reth_ethereum_engine_primitives::EthEngineTypes;
     use reth_ethereum_primitives::Block;
+    use reth_network_api::{
+        noop::NoopNetwork, EthProtocolInfo, NetworkError, NetworkInfo, NetworkStatus,
+    };
     use reth_node_ethereum::EthereumEngineValidator;
     use reth_payload_builder::test_utils::spawn_test_payload_service;
     use reth_provider::test_utils::MockEthProvider;
@@ -1206,6 +1273,7 @@ mod tests {
             EngineCapabilities::default(),
             EthereumEngineValidator::new(chain_spec.clone()),
             false,
+            NoopNetwork::default(),
         );
         let handle = EngineApiTestHandle { chain_spec, provider, from_api: engine_rx };
         (handle, api)
@@ -1247,6 +1315,76 @@ mod tests {
         assert_matches!(handle.from_api.recv().await, Some(BeaconEngineMessage::NewPayload { .. }));
     }
 
+    #[derive(Clone)]
+    struct TestNetworkInfo {
+        syncing: bool,
+    }
+
+    impl NetworkInfo for TestNetworkInfo {
+        fn local_addr(&self) -> std::net::SocketAddr {
+            (std::net::Ipv4Addr::UNSPECIFIED, 0).into()
+        }
+
+        async fn network_status(&self) -> Result<NetworkStatus, NetworkError> {
+            #[allow(deprecated)]
+            Ok(NetworkStatus {
+                client_version: "test".to_string(),
+                protocol_version: 5,
+                eth_protocol_info: EthProtocolInfo {
+                    network: 1,
+                    difficulty: None,
+                    genesis: Default::default(),
+                    config: Default::default(),
+                    head: Default::default(),
+                },
+                capabilities: vec![],
+            })
+        }
+
+        fn chain_id(&self) -> u64 {
+            1
+        }
+
+        fn is_syncing(&self) -> bool {
+            self.syncing
+        }
+
+        fn is_initially_syncing(&self) -> bool {
+            self.syncing
+        }
+    }
+
+    #[tokio::test]
+    async fn get_blobs_v3_returns_null_when_syncing() {
+        let chain_spec: Arc<ChainSpec> =
+            Arc::new(ChainSpecBuilder::mainnet().osaka_activated().build());
+        let provider = Arc::new(MockEthProvider::default());
+        let payload_store = spawn_test_payload_service::<EthEngineTypes>();
+        let (to_engine, _engine_rx) = unbounded_channel::<BeaconEngineMessage<EthEngineTypes>>();
+
+        let api = EngineApi::new(
+            provider,
+            chain_spec.clone(),
+            ConsensusEngineHandle::new(to_engine),
+            payload_store.into(),
+            NoopTransactionPool::default(),
+            Box::<TokioTaskExecutor>::default(),
+            ClientVersionV1 {
+                code: ClientCode::RH,
+                name: "Reth".to_string(),
+                version: "v0.0.0-test".to_string(),
+                commit: "test".to_string(),
+            },
+            EngineCapabilities::default(),
+            EthereumEngineValidator::new(chain_spec),
+            false,
+            TestNetworkInfo { syncing: true },
+        );
+
+        let res = api.get_blobs_v3_metered(vec![B256::ZERO]);
+        assert_matches!(res, Ok(None));
+    }
+
     // tests covering `engine_getPayloadBodiesByRange` and `engine_getPayloadBodiesByHash`
     mod get_payload_bodies {
         use super::*;

crates/rpc/rpc-engine-api/src/metrics.rs

@@ -46,6 +46,8 @@ pub(crate) struct EngineApiLatencyMetrics {
     pub(crate) get_blobs_v1: Histogram,
     /// Latency for `engine_getBlobsV2`
     pub(crate) get_blobs_v2: Histogram,
+    /// Latency for `engine_getBlobsV3`
+    pub(crate) get_blobs_v3: Histogram,
 }
 
 #[derive(Metrics)]