Merge pull request #20 from Irys-xyz/fix/p2p

Fix/p2p

Author: JesseTheRobot

Cargo.toml

@@ -609,8 +609,8 @@ tracy-client = "0.17.3"
 irys-primitives = { path = "../../crates/primitives" }
 
 [patch.crates-io]
-revm = { path = "../.../revm/crates/revm" }
-revm-primitives = { path = "../.../revm/crates/primitives" }
+revm = { path = "../revm/crates/revm" }
+revm-primitives = { path = "../revm/crates/primitives" }
 
 #alloy-consensus = { git = "https://github.com/alloy-rs/alloy", rev = "8c499409"}
 #alloy-eips = { git = "https://github.com/alloy-rs/alloy", rev = "8c499409"}

crates/e2e-test-utils/src/engine_api.rs

@@ -41,6 +41,19 @@ impl<E: EngineTypes> EngineApiTestContext<E> {
         Ok(self.engine_api_client.request("engine_getPayloadV1Irys", (payload_id,)).await?)
     }
 
+    pub async fn build_payload_v1_irys(
+        &self,
+        parent: B256,
+        payload_attributes: E::PayloadAttributes,
+    ) -> eyre::Result<E::ExecutionPayloadV1Irys> {
+        Ok(EngineApiClient::<E>::build_new_payload_irys(
+            &self.engine_api_client,
+            parent,
+            payload_attributes,
+        )
+        .await?)
+    }
+
     /// Submits a payload to the engine api
     pub async fn submit_payload(
         &self,

crates/e2e-test-utils/src/node.rs

@@ -129,6 +129,71 @@ where
         Ok((built_payload, eth_attr))
     }
 
+    pub async fn new_payload_irys(
+        &mut self,
+        // attributes: Engine::PayloadBuilderAttributes,
+        attributes_generator: impl Fn(u64) -> Engine::PayloadBuilderAttributes,
+    ) -> eyre::Result<(Engine::BuiltPayload, Engine::PayloadBuilderAttributes)>
+    where
+        <Engine as EngineTypes>::ExecutionPayloadV1Irys:
+            From<Engine::BuiltPayload> + PayloadEnvelopeExt,
+    {
+        // trigger new payload building draining the pool
+        // self.payload.payload_builder.new_payload(attributes.clone()).await?;
+
+        // let eth_attr = attributes;
+
+        let eth_attr = self.payload.new_payload(attributes_generator).await.unwrap();
+
+        // first event is the payload attributes
+        // self.payload.expect_attr_event(eth_attr.clone()).await?;
+        // wait for the payload builder to have finished building
+        let p2 = self.payload.wait_for_built_payload(eth_attr.payload_id()).await;
+        // trigger resolve payload via engine api
+        let _ = self.engine_api.get_payload_v1_irys(eth_attr.payload_id()).await?;
+        // ensure we're also receiving the built payload as event
+        // let built_payload = self.payload.expect_built_payload().await?;
+        // self.
+        // Ok((built_payload, eth_attr))
+        Ok((/* execution_payload */ p2, eth_attr))
+    }
+
+    pub async fn new_payload_irys2(
+        &mut self,
+        parent: B256,
+        attributes: Engine::PayloadAttributes,
+    ) -> eyre::Result<(
+        Engine::ExecutionPayloadV1Irys,
+        Engine::BuiltPayload,
+        Engine::PayloadBuilderAttributes,
+    )>
+    where
+        <Engine as EngineTypes>::ExecutionPayloadV1Irys:
+            From<Engine::BuiltPayload> + PayloadEnvelopeExt,
+    {
+        let payload_builder_attributes =
+            <Engine as reth_node_builder::PayloadTypes>::PayloadBuilderAttributes::try_new(
+                parent, attributes,
+            )
+            .expect("unable to build PayloadBuilderAttributes");
+        use reth_payload_primitives::PayloadBuilder;
+        // trigger new payload building draining the pool
+        self.payload.payload_builder.new_payload(payload_builder_attributes.clone()).await?;
+
+        let eth_attr = payload_builder_attributes;
+        // first event is the payload attributes
+        // self.payload.expect_attr_event(eth_attr.clone()).await?;
+        // wait for the payload builder to have finished building
+        let built_payload = self.payload.wait_for_built_payload(eth_attr.payload_id()).await;
+        // trigger resolve payload via engine api
+        let execution_payload = self.engine_api.get_payload_v1_irys(eth_attr.payload_id()).await?;
+        // ensure we're also receiving the built payload as event
+        // let built_payload = self.payload.expect_built_payload().await?;
+        // self.
+        // Ok((built_payload, eth_attr))
+        Ok((execution_payload, built_payload, eth_attr))
+    }
+
     /// Advances the node forward one block
     pub async fn advance_block(
         &mut self,
@@ -157,6 +222,34 @@ where
         Ok((payload, eth_attr))
     }
 
+    /// Advances the node forward one block
+    pub async fn advance_block_irys(
+        &mut self,
+        versioned_hashes: Vec<B256>,
+        attributes_generator: impl Fn(u64) -> Engine::PayloadBuilderAttributes,
+    ) -> eyre::Result<(Engine::BuiltPayload, Engine::PayloadBuilderAttributes)>
+    where
+        <Engine as EngineTypes>::ExecutionPayloadV1Irys:
+            From<Engine::BuiltPayload> + PayloadEnvelopeExt,
+    {
+        let (payload, eth_attr) = self.new_payload_irys(attributes_generator).await?;
+
+        let block_hash = self
+            .engine_api
+            .submit_payload(
+                payload.clone(),
+                eth_attr.clone(),
+                PayloadStatusEnum::Valid,
+                versioned_hashes,
+            )
+            .await?;
+
+        // trigger forkchoice update via engine api to commit the block to the blockchain
+        self.engine_api.update_forkchoice(block_hash, block_hash).await?;
+
+        Ok((payload, eth_attr))
+    }
+
     /// Waits for block to be available on node.
     pub async fn wait_block(
         &self,
@@ -221,6 +314,37 @@ where
         let tx = head.tip().transactions().next();
         assert_eq!(tx.unwrap().hash().as_slice(), tip_tx_hash.as_slice());
 
+        loop {
+            // wait for the block to commit
+            tokio::time::sleep(std::time::Duration::from_millis(20)).await;
+            if let Some(latest_block) =
+                self.inner.provider.block_by_number_or_tag(BlockNumberOrTag::Latest)?
+            {
+                if latest_block.number == block_number {
+                    // make sure the block hash we submitted via FCU engine api is the new latest
+                    // block using an RPC call
+                    assert_eq!(latest_block.hash_slow(), block_hash);
+                    break;
+                }
+            }
+        }
+        Ok(())
+    }
+    /// Asserts that a new block has been added to the blockchain
+    /// and the tx has been included in the block.
+    ///
+    /// Does NOT work for pipeline since there's no stream notification!
+    pub async fn assert_new_block2(
+        &self,
+        block_hash: B256,
+        block_number: BlockNumber,
+    ) -> eyre::Result<()> {
+        // get head block from notifications stream and verify the tx has been pushed to the
+        // pool is actually present in the canonical block
+        // let head = self.engine_api.canonical_stream.next().await.unwrap();
+        // let tx = head.tip().transactions().next();
+        // assert_eq!(tx.unwrap().hash().as_slice(), tip_tx_hash.as_slice());
+
         loop {
             // wait for the block to commit
             tokio::time::sleep(std::time::Duration::from_millis(20)).await;

crates/e2e-test-utils/src/payload.rs

@@ -48,7 +48,10 @@ impl<E: EngineTypes> PayloadTestContext<E> {
     }
 
     /// Wait until the best built payload is ready
-    pub async fn wait_for_built_payload(&self, payload_id: PayloadId) {
+    pub async fn wait_for_built_payload(
+        &self,
+        payload_id: PayloadId,
+    ) -> <E as reth_node_builder::PayloadTypes>::BuiltPayload {
         loop {
             // let payload = self.payload_builder.best_payload(payload_id).await.unwrap().unwrap();
             match self.payload_builder.best_payload(payload_id).await {
@@ -59,7 +62,7 @@ impl<E: EngineTypes> PayloadTestContext<E> {
                             continue;
                         } else {
                             trace!("got payload");
-                            break;
+                            return v;
                         }
                     }
                     Err(e) => {

crates/ethereum/evm/src/execute.rs

@@ -166,19 +166,20 @@ where
         //     Default::default(),
         // ))
         // .build();
-        let prev = evm.context.evm.inner.journaled_state.checkpoint();
-        // TODO: fix this
+
+        // let prev = evm.context.evm.inner.journaled_state.checkpoint();
+
         let shadow_exec =
             apply_block_shadows(block.body.shadows.as_ref(), &mut evm).map_err(move |err| {
                 let new_err = err.map_db_err(|e| e.into());
                 BlockValidationError::EVM { hash: B256::ZERO, error: Box::new(new_err) }
             })?;
-        info!("shadow exec: {:#?}", &shadow_exec);
+        info!("shadow exec2: {:#?}", &shadow_exec);
         let ss = evm.context.evm.inner.journaled_state.state.clone();
 
         evm.db_mut().commit(ss);
 
-        evm.context.evm.inner.journaled_state.checkpoint_revert(prev);
+        // evm.context.evm.inner.journaled_state.checkpoint_revert(prev);
 
         // execute transactions
         let mut cumulative_gas_used = 0;

crates/net/eth-wire-types/Cargo.toml

@@ -16,7 +16,6 @@ workspace = true
 reth-chainspec.workspace = true
 reth-codecs-derive.workspace = true
 reth-primitives.workspace = true
-
 # ethereum
 alloy-chains = { workspace = true, features = ["rlp"] }
 alloy-eips.workspace = true

crates/net/eth-wire-types/src/message.rs

@@ -12,7 +12,6 @@ use super::{
     NewPooledTransactionHashes68, NodeData, PooledTransactions, Receipts, Status, Transactions,
 };
 use crate::{EthVersion, SharedTransactions};
-
 use alloy_primitives::bytes::{Buf, BufMut};
 use alloy_rlp::{length_of_length, Decodable, Encodable, Header};
 use std::{fmt::Debug, sync::Arc};
@@ -46,7 +45,6 @@ impl ProtocolMessage {
     /// Create a new `ProtocolMessage` from a message type and message rlp bytes.
     pub fn decode_message(version: EthVersion, buf: &mut &[u8]) -> Result<Self, MessageError> {
         let message_type = EthMessageID::decode(buf)?;
-
         let message = match message_type {
             EthMessageID::Status => EthMessage::Status(Status::decode(buf)?),
             EthMessageID::NewBlockHashes => {
@@ -91,14 +89,14 @@ impl ProtocolMessage {
             }
             EthMessageID::GetNodeData => {
                 if version >= EthVersion::Eth67 {
-                    return Err(MessageError::Invalid(version, EthMessageID::GetNodeData))
+                    return Err(MessageError::Invalid(version, EthMessageID::GetNodeData));
                 }
                 let request_pair = RequestPair::<GetNodeData>::decode(buf)?;
                 EthMessage::GetNodeData(request_pair)
             }
             EthMessageID::NodeData => {
                 if version >= EthVersion::Eth67 {
-                    return Err(MessageError::Invalid(version, EthMessageID::GetNodeData))
+                    return Err(MessageError::Invalid(version, EthMessageID::GetNodeData));
                 }
                 let request_pair = RequestPair::<NodeData>::decode(buf)?;
                 EthMessage::NodeData(request_pair)
@@ -219,7 +217,7 @@ pub enum EthMessage {
 
 impl EthMessage {
     /// Returns the message's ID.
-    pub const fn message_id(&self) -> EthMessageID {
+    pub fn message_id(&self) -> EthMessageID {
         match self {
             Self::Status(_) => EthMessageID::Status,
             Self::NewBlockHashes(_) => EthMessageID::NewBlockHashes,
@@ -483,7 +481,7 @@ where
         // RequestPair
         let consumed_len = initial_length - buf.len();
         if consumed_len != header.payload_length {
-            return Err(alloy_rlp::Error::UnexpectedLength)
+            return Err(alloy_rlp::Error::UnexpectedLength);
         }
 
         Ok(Self { request_id, message })

crates/net/eth-wire/src/p2pstream.rs

@@ -620,10 +620,10 @@ where
                 Poll::Ready(Err(err)) => break Poll::Ready(Err(err.into())),
                 Poll::Ready(Ok(())) => {
                     let Some(message) = this.outgoing_messages.pop_front() else {
-                        break Poll::Ready(Ok(()))
+                        break Poll::Ready(Ok(()));
                     };
                     if let Err(err) = this.inner.as_mut().start_send(message) {
-                        break Poll::Ready(Err(err.into()))
+                        break Poll::Ready(Err(err.into()));
                     }
                 }
             }

crates/net/network/src/eth_requests.rs

@@ -21,6 +21,7 @@ use reth_primitives::{BlockBody, Header};
 use reth_storage_api::{BlockReader, HeaderProvider, ReceiptProvider};
 use tokio::sync::{mpsc::Receiver, oneshot};
 use tokio_stream::wrappers::ReceiverStream;
+use tracing::{debug, warn};
 
 use crate::{
     budget::DEFAULT_BUDGET_TRY_DRAIN_DOWNLOADERS, metered_poll_nested_stream_with_budget,
@@ -133,7 +134,7 @@ where
                 headers.push(header);
 
                 if headers.len() >= MAX_HEADERS_SERVE || total_bytes > SOFT_RESPONSE_LIMIT {
-                    break
+                    break;
                 }
             } else {
                 break;
@@ -164,18 +165,19 @@ where
         let mut bodies = Vec::new();
 
         let mut total_bytes = 0;
+        debug!("Getting block bodies {:?}", &request.0);
 
         for hash in request.0 {
             if let Some(block) = self.client.block_by_hash(hash).unwrap_or_default() {
                 let body: BlockBody = block.into();
-
                 total_bytes += body.length();
                 bodies.push(body);
 
                 if bodies.len() >= MAX_BODIES_SERVE || total_bytes > SOFT_RESPONSE_LIMIT {
-                    break
+                    break;
                 }
             } else {
+                warn!("Unable to get block body {}", &hash);
                 break;
             }
         }
@@ -208,7 +210,7 @@ where
                 receipts.push(receipt);
 
                 if receipts.len() >= MAX_RECEIPTS_SERVE || total_bytes > SOFT_RESPONSE_LIMIT {
-                    break
+                    break;
                 }
             } else {
                 break;