Implement engine_getBlobsV1

crates/node/builder/src/launch/mod.rs

@@ -295,6 +295,7 @@ where
             ctx.chain_spec(),
             beacon_engine_handle,
             ctx.components().payload_builder().clone().into(),
+            ctx.components().pool().clone(),
             Box::new(ctx.task_executor().clone()),
             client,
             EngineCapabilities::default(),

crates/primitives/src/transaction/mod.rs

@@ -415,7 +415,7 @@ impl Transaction {
 
         // Check if max_fee_per_gas is less than base_fee
         if max_fee_per_gas < base_fee {
-            return None
+            return None;
         }
 
         // Calculate the difference between max_fee_per_gas and base_fee
@@ -840,7 +840,7 @@ impl TransactionSignedNoHash {
         // `from` address.
         #[cfg(feature = "optimism")]
         if let Transaction::Deposit(TxDeposit { from, .. }) = self.transaction {
-            return Some(from)
+            return Some(from);
         }
 
         let signature_hash = self.signature_hash();
@@ -869,15 +869,15 @@ impl TransactionSignedNoHash {
         #[cfg(feature = "optimism")]
         {
             if let Transaction::Deposit(TxDeposit { from, .. }) = self.transaction {
-                return Some(from)
+                return Some(from);
             }
 
             // pre bedrock system transactions were sent from the zero address as legacy
             // transactions with an empty signature
             //
             // NOTE: this is very hacky and only relevant for op-mainnet pre bedrock
             if self.is_legacy() && self.signature == Signature::optimism_deposit_tx_signature() {
-                return Some(Address::ZERO)
+                return Some(Address::ZERO);
             }
         }
 
@@ -1094,7 +1094,7 @@ impl TransactionSigned {
         // `from` address.
         #[cfg(feature = "optimism")]
         if let Transaction::Deposit(TxDeposit { from, .. }) = self.transaction {
-            return Some(from)
+            return Some(from);
         }
         let signature_hash = self.signature_hash();
         self.signature.recover_signer(signature_hash)
@@ -1110,7 +1110,7 @@ impl TransactionSigned {
         // `from` address.
         #[cfg(feature = "optimism")]
         if let Transaction::Deposit(TxDeposit { from, .. }) = self.transaction {
-            return Some(from)
+            return Some(from);
         }
         let signature_hash = self.signature_hash();
         self.signature.recover_signer_unchecked(signature_hash)
@@ -1287,7 +1287,7 @@ impl TransactionSigned {
         let transaction_payload_len = header.payload_length;
 
         if transaction_payload_len > remaining_len {
-            return Err(RlpError::InputTooShort)
+            return Err(RlpError::InputTooShort);
         }
 
         let mut transaction = TxLegacy {
@@ -1305,7 +1305,7 @@ impl TransactionSigned {
         // check the new length, compared to the original length and the header length
         let decoded = remaining_len - data.len();
         if decoded != transaction_payload_len {
-            return Err(RlpError::UnexpectedLength)
+            return Err(RlpError::UnexpectedLength);
         }
 
         let tx_length = header.payload_length + header.length();
@@ -1350,7 +1350,7 @@ impl TransactionSigned {
         // decode the list header for the rest of the transaction
         let header = Header::decode(data)?;
         if !header.list {
-            return Err(RlpError::Custom("typed tx fields must be encoded as a list"))
+            return Err(RlpError::Custom("typed tx fields must be encoded as a list"));
         }
 
         let remaining_len = data.len();
@@ -1360,7 +1360,7 @@ impl TransactionSigned {
 
         // decode common fields
         let Ok(tx_type) = TxType::try_from(tx_type) else {
-            return Err(RlpError::Custom("unsupported typed transaction type"))
+            return Err(RlpError::Custom("unsupported typed transaction type"));
         };
 
         let transaction = match tx_type {
@@ -1385,7 +1385,7 @@ impl TransactionSigned {
 
         let bytes_consumed = remaining_len - data.len();
         if bytes_consumed != header.payload_length {
-            return Err(RlpError::UnexpectedLength)
+            return Err(RlpError::UnexpectedLength);
         }
 
         let hash = keccak256(&original_encoding_without_header[..tx_length]);
@@ -1412,7 +1412,7 @@ impl TransactionSigned {
     /// of bytes in input data.
     pub fn decode_enveloped(input_data: &mut &[u8]) -> alloy_rlp::Result<Self> {
         if input_data.is_empty() {
-            return Err(RlpError::InputTooShort)
+            return Err(RlpError::InputTooShort);
         }
 
         // Check if the tx is a list
@@ -1424,7 +1424,7 @@ impl TransactionSigned {
         };
 
         if !input_data.is_empty() {
-            return Err(RlpError::UnexpectedLength)
+            return Err(RlpError::UnexpectedLength);
         }
 
         Ok(output_data)
@@ -1505,7 +1505,7 @@ impl Decodable for TransactionSigned {
     /// string header if the first byte is less than `0xf7`.
     fn decode(buf: &mut &[u8]) -> alloy_rlp::Result<Self> {
         if buf.is_empty() {
-            return Err(RlpError::InputTooShort)
+            return Err(RlpError::InputTooShort);
         }
 
         // decode header
@@ -1523,7 +1523,7 @@ impl Decodable for TransactionSigned {
             // string Header with payload_length of 1, we need to make sure this check is only
             // performed for transactions with a string header
             if bytes_consumed != header.payload_length && original_encoding[0] > EMPTY_STRING_CODE {
-                return Err(RlpError::UnexpectedLength)
+                return Err(RlpError::UnexpectedLength);
             }
 
             Ok(tx)

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

@@ -13,7 +13,7 @@ use reth_rpc_types::{
         ForkchoiceState, ForkchoiceUpdated, PayloadId, PayloadStatus, TransitionConfiguration,
     },
     state::StateOverride,
-    BlockOverrides, Filter, Log, RichBlock, SyncStatus, TransactionRequest,
+    BlobAndProofV1, BlockOverrides, Filter, Log, RichBlock, SyncStatus, TransactionRequest,
 };
 // NOTE: We can't use associated types in the `EngineApi` trait because of jsonrpsee, so we use a
 // generic here. It would be nice if the rpc macro would understand which types need to have serde.
@@ -211,6 +211,13 @@ pub trait EngineApi<Engine: EngineTypes> {
     /// See also <https://github.com/ethereum/execution-apis/blob/6452a6b194d7db269bf1dbd087a267251d3cc7f8/src/engine/common.md#capabilities>
     #[method(name = "exchangeCapabilities")]
     async fn exchange_capabilities(&self, capabilities: Vec<String>) -> RpcResult<Vec<String>>;
+
+    /// Fetch blobs for the consensus layer from the in-memory blob cache.
+    #[method(name = "getBlobsV1")]
+    async fn get_blobs_v1(
+        &self,
+        transaction_ids: Vec<B256>,
+    ) -> RpcResult<Vec<Option<BlobAndProofV1>>>;
 }
 
 /// A subset of the ETH rpc interface: <https://ethereum.github.io/execution-apis/api-documentation/>

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

@@ -25,6 +25,7 @@ reth-tasks.workspace = true
 reth-rpc-types-compat.workspace = true
 reth-engine-primitives.workspace = true
 reth-evm.workspace = true
+reth-transaction-pool.workspace = true
 
 # async
 tokio = { workspace = true, features = ["sync"] }

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

@@ -16,17 +16,21 @@ use reth_primitives::{
     Block, BlockHash, BlockHashOrNumber, BlockNumber, EthereumHardfork, B256, U64,
 };
 use reth_rpc_api::EngineApiServer;
-use reth_rpc_types::engine::{
-    CancunPayloadFields, ClientVersionV1, ExecutionPayload, ExecutionPayloadBodiesV1,
-    ExecutionPayloadBodiesV2, ExecutionPayloadInputV2, ExecutionPayloadV1, ExecutionPayloadV3,
-    ExecutionPayloadV4, ForkchoiceState, ForkchoiceUpdated, PayloadId, PayloadStatus,
-    TransitionConfiguration,
+use reth_rpc_types::{
+    engine::{
+        CancunPayloadFields, ClientVersionV1, ExecutionPayload, ExecutionPayloadBodiesV1,
+        ExecutionPayloadBodiesV2, ExecutionPayloadInputV2, ExecutionPayloadV1, ExecutionPayloadV3,
+        ExecutionPayloadV4, ForkchoiceState, ForkchoiceUpdated, PayloadId, PayloadStatus,
+        TransitionConfiguration,
+    },
+    BlobAndProofV1,
 };
 use reth_rpc_types_compat::engine::payload::{
     convert_payload_input_v2_to_payload, convert_to_payload_body_v1, convert_to_payload_body_v2,
 };
 use reth_storage_api::{BlockReader, HeaderProvider, StateProviderFactory};
 use reth_tasks::TaskSpawner;
+use reth_transaction_pool::TransactionPool;
 use std::{sync::Arc, time::Instant};
 use tokio::sync::oneshot;
 use tracing::{trace, warn};
@@ -39,11 +43,11 @@ const MAX_PAYLOAD_BODIES_LIMIT: u64 = 1024;
 
 /// The Engine API implementation that grants the Consensus layer access to data and
 /// functions in the Execution layer that are crucial for the consensus process.
-pub struct EngineApi<Provider, EngineT: EngineTypes> {
-    inner: Arc<EngineApiInner<Provider, EngineT>>,
+pub struct EngineApi<Provider, EngineT: EngineTypes, Pool> {
+    inner: Arc<EngineApiInner<Provider, EngineT, Pool>>,
 }
 
-struct EngineApiInner<Provider, EngineT: EngineTypes> {
+struct EngineApiInner<Provider, EngineT: EngineTypes, Pool> {
     /// The provider to interact with the chain.
     provider: Provider,
     /// Consensus configuration
@@ -60,19 +64,23 @@ struct EngineApiInner<Provider, EngineT: EngineTypes> {
     client: ClientVersionV1,
     /// The list of all supported Engine capabilities available over the engine endpoint.
     capabilities: EngineCapabilities,
+    /// Transaction pool.
+    tx_pool: Pool,
 }
 
-impl<Provider, EngineT> EngineApi<Provider, EngineT>
+impl<Provider, EngineT, Pool> EngineApi<Provider, EngineT, Pool>
 where
     Provider: HeaderProvider + BlockReader + StateProviderFactory + EvmEnvProvider + 'static,
     EngineT: EngineTypes + 'static,
+    Pool: TransactionPool + 'static,
 {
     /// Create new instance of [`EngineApi`].
     pub fn new(
         provider: Provider,
         chain_spec: Arc<ChainSpec>,
         beacon_consensus: BeaconConsensusEngineHandle<EngineT>,
         payload_store: PayloadStore<EngineT>,
+        tx_pool: Pool,
         task_spawner: Box<dyn TaskSpawner>,
         client: ClientVersionV1,
         capabilities: EngineCapabilities,
@@ -86,6 +94,7 @@ where
             metrics: EngineApiMetrics::default(),
             client,
             capabilities,
+            tx_pool,
         });
         Self { inner }
     }
@@ -609,10 +618,11 @@ where
 }
 
 #[async_trait]
-impl<Provider, EngineT> EngineApiServer<EngineT> for EngineApi<Provider, EngineT>
+impl<Provider, EngineT, Pool> EngineApiServer<EngineT> for EngineApi<Provider, EngineT, Pool>
 where
     Provider: HeaderProvider + BlockReader + StateProviderFactory + EvmEnvProvider + 'static,
     EngineT: EngineTypes + 'static,
+    Pool: TransactionPool + 'static,
 {
     /// Handler for `engine_newPayloadV1`
     /// See also <https://github.com/ethereum/execution-apis/blob/3d627c95a4d3510a8187dd02e0250ecb4331d27e/src/engine/paris.md#engine_newpayloadv1>
@@ -904,9 +914,21 @@ where
     async fn exchange_capabilities(&self, _capabilities: Vec<String>) -> RpcResult<Vec<String>> {
         Ok(self.inner.capabilities.list())
     }
+
+    async fn get_blobs_v1(
+        &self,
+        versioned_hashes: Vec<B256>,
+    ) -> RpcResult<Vec<Option<BlobAndProofV1>>> {
+        // FIXME(sproul): work out error wrapping or make infallible
+        Ok(self
+            .inner
+            .tx_pool
+            .get_blobs_for_versioned_hashes(&versioned_hashes)
+            .expect("get_blobs_for_versioned_hashes is infallible"))
+    }
 }
 
-impl<Provider, EngineT> std::fmt::Debug for EngineApi<Provider, EngineT>
+impl<Provider, EngineT, Pool> std::fmt::Debug for EngineApi<Provider, EngineT, Pool>
 where
     EngineT: EngineTypes,
 {

crates/rpc/rpc-types/Cargo.toml

@@ -14,6 +14,7 @@ workspace = true
 [dependencies]
 
 # ethereum
+alloy-eips = { workspace = true }
 alloy-primitives = { workspace = true, features = ["rand", "rlp", "serde"] }
 alloy-rpc-types = { workspace = true, features = ["jsonrpsee-types"] }
 alloy-rpc-types-admin.workspace = true
@@ -42,4 +43,4 @@ serde_json.workspace = true
 
 [features]
 default = ["jsonrpsee-types"]
-arbitrary = ["alloy-primitives/arbitrary", "alloy-rpc-types/arbitrary"]
+arbitrary = ["alloy-primitives/arbitrary", "alloy-rpc-types/arbitrary"]

crates/rpc/rpc-types/src/lib.rs

@@ -55,3 +55,17 @@ pub use eth::{
 
 pub use peer::*;
 pub use rpc::*;
+
+use alloy_eips::eip4844::BYTES_PER_BLOB;
+use alloy_primitives::FixedBytes;
+use serde::{Deserialize, Serialize};
+
+/// Blob type returned in responses to `engine_getBlobsV1`.
+// FIXME(sproul): move to alloy?
+#[derive(Debug, Serialize, Deserialize, Clone)]
+pub struct BlobAndProofV1 {
+    /// The blob data.
+    pub blob: FixedBytes<BYTES_PER_BLOB>,
+    /// The KZG proof for the blob.
+    pub proof: FixedBytes<48>,
+}

crates/transaction-pool/Cargo.toml

@@ -17,6 +17,7 @@ reth-chainspec.workspace = true
 reth-eth-wire-types.workspace = true
 reth-primitives.workspace = true
 reth-execution-types.workspace = true
+reth-rpc-types.workspace = true
 reth-fs-util.workspace = true
 reth-provider.workspace = true
 reth-tasks.workspace = true