feat: implement p2p layer and broadcast flashblocks (#275)

Author: noot

Cargo.lock

@@ -8,7 +8,7 @@ repository = "https://github.com/flashbots/op-rbuilder"
 exclude = [".github/"]
 
 [workspace]
-members = [ "crates/op-rbuilder", "crates/tdx-quote-provider"]
+members = [ "crates/op-rbuilder", "crates/p2p", "crates/tdx-quote-provider"]
 default-members = ["crates/op-rbuilder"]
 resolver = "2"
 
@@ -182,6 +182,9 @@ flate2 = "1.0.37"
 prometheus = "0.13.4"
 ctor = "0.2"
 dashmap = "6.1"
+hex = "0.4"
+futures = "0.3"
+futures-util = "0.3.31"
 
 lazy_static = "1.4.0"
 tikv-jemallocator = { version = "0.6" }

Cargo.toml

@@ -12,6 +12,8 @@ default-run = "op-rbuilder"
 workspace = true
 
 [dependencies]
+p2p = { path = "../p2p" }
+
 reth.workspace = true
 reth-optimism-node.workspace = true
 reth-optimism-cli.workspace = true
@@ -109,10 +111,11 @@ url.workspace = true
 anyhow = "1"
 opentelemetry = { workspace = true, optional = true }
 dashmap.workspace = true
+hex = { workspace = true }
+futures = { workspace = true }
+futures-util = { workspace = true }
 
 tower = "0.5"
-futures = "0.3"
-futures-util = "0.3.31"
 time = { version = "0.3.36", features = ["macros", "formatting", "parsing"] }
 chrono = "0.4"
 uuid = { version = "1.6.1", features = ["serde", "v5", "v4"] }
@@ -124,7 +127,6 @@ serde_yaml = { version = "0.9" }
 moka = "0.12"
 http = "1.0"
 sha3 = "0.10"
-hex = "0.4"
 reqwest = "0.12.23"
 k256 = "0.13.4"
 

crates/op-rbuilder/Cargo.toml

@@ -166,6 +166,10 @@ pub struct FlashblocksArgs {
         env = "FLASHBLOCK_NUMBER_CONTRACT_ADDRESS"
     )]
     pub flashblocks_number_contract_address: Option<Address>,
+
+    /// Flashblocks p2p configuration
+    #[command(flatten)]
+    pub p2p: FlashblocksP2pArgs,
 }
 
 impl Default for FlashblocksArgs {
@@ -178,6 +182,49 @@ impl Default for FlashblocksArgs {
     }
 }
 
+#[derive(Debug, Clone, PartialEq, Eq, clap::Args)]
+pub struct FlashblocksP2pArgs {
+    /// Enable libp2p networking for flashblock propagation
+    #[arg(
+        long = "flashblocks.p2p_enabled",
+        env = "FLASHBLOCK_P2P_ENABLED",
+        default_value = "false"
+    )]
+    pub p2p_enabled: bool,
+
+    /// Port for the flashblocks p2p node
+    #[arg(
+        long = "flashblocks.p2p_port",
+        env = "FLASHBLOCK_P2P_PORT",
+        default_value = "9009"
+    )]
+    pub p2p_port: u16,
+
+    /// Path to the file containing a hex-encoded libp2p private key.
+    /// If the file does not exist, a new key will be generated.
+    #[arg(
+        long = "flashblocks.p2p_private_key_file",
+        env = "FLASHBLOCK_P2P_PRIVATE_KEY_FILE"
+    )]
+    pub p2p_private_key_file: Option<String>,
+
+    /// Comma-separated list of multiaddrs of known Flashblocks peers
+    /// Example: "/ip4/104.131.131.82/tcp/4001/p2p/QmaCpDMGvV2BGHeYERUEnRQAwe3N8SzbUtfsmvsqQLuvuJ,/ip4/104.131.131.82/udp/4001/quic-v1/p2p/QmaCpDMGvV2BGHeYERUEnRQAwe3N8SzbUtfsmvsqQLuvuJ"
+    #[arg(
+        long = "flashblocks.p2p_known_peers",
+        env = "FLASHBLOCK_P2P_KNOWN_PEERS"
+    )]
+    pub p2p_known_peers: Option<String>,
+
+    /// Maximum number of peers for the flashblocks p2p node
+    #[arg(
+        long = "flashblocks.p2p_max_peer_count",
+        env = "FLASHBLOCK_P2P_MAX_PEER_COUNT",
+        default_value = "50"
+    )]
+    pub p2p_max_peer_count: u32,
+}
+
 /// Parameters for telemetry configuration
 #[derive(Debug, Clone, Default, PartialEq, Eq, clap::Args)]
 pub struct TelemetryArgs {

crates/op-rbuilder/src/args/op.rs

@@ -4,7 +4,7 @@ use alloy_evm::Database;
 use alloy_op_evm::OpEvm;
 use alloy_primitives::{
     Address, B256, Bytes, TxKind, U256,
-    map::foldhash::{HashMap, HashSet, HashSetExt},
+    map::{HashMap, HashSet},
 };
 use alloy_sol_types::{ContractError, Revert, SolCall, SolError, SolInterface};
 use core::fmt::Debug;
@@ -192,7 +192,7 @@ pub trait BuilderTransactions<ExtraCtx: Debug + Default = (), Extra: Debug + Def
                 .evm_config
                 .evm_with_env(&mut *db, builder_ctx.evm_env.clone());
 
-            let mut invalid: HashSet<Address> = HashSet::new();
+            let mut invalid = HashSet::new();
 
             for builder_tx in builder_txs.iter() {
                 if builder_tx.is_top_of_block != top_of_block {

crates/op-rbuilder/src/builders/builder_tx.rs

@@ -544,9 +544,9 @@ impl<ExtraCtx: Debug + Default> OpPayloadBuilderCtx<ExtraCtx> {
             info.executed_transactions.push(tx.into_inner());
         }
 
-        let payload_tx_simulation_time = execute_txs_start_time.elapsed();
+        let payload_transaction_simulation_time = execute_txs_start_time.elapsed();
         self.metrics.set_payload_builder_metrics(
-            payload_tx_simulation_time,
+            payload_transaction_simulation_time,
             num_txs_considered,
             num_txs_simulated,
             num_txs_simulated_success,

crates/op-rbuilder/src/builders/context.rs

@@ -38,6 +38,21 @@ pub struct FlashblocksConfig {
     ///
     /// If set a builder tx will be added to the start of every flashblock instead of the regular builder tx.
     pub flashblocks_number_contract_address: Option<Address>,
+
+    /// Whether to enable the p2p node for flashblocks
+    pub p2p_enabled: bool,
+
+    /// Port for the p2p node
+    pub p2p_port: u16,
+
+    /// Optional hex-encoded private key file path for the p2p node
+    pub p2p_private_key_file: Option<String>,
+
+    /// Comma-separated list of multiaddresses of known peers to connect to
+    pub p2p_known_peers: Option<String>,
+
+    /// Maximum number of peers for the p2p node
+    pub p2p_max_peer_count: u32,
 }
 
 impl Default for FlashblocksConfig {
@@ -49,6 +64,11 @@ impl Default for FlashblocksConfig {
             fixed: false,
             calculate_state_root: true,
             flashblocks_number_contract_address: None,
+            p2p_enabled: false,
+            p2p_port: 9009,
+            p2p_private_key_file: None,
+            p2p_known_peers: None,
+            p2p_max_peer_count: 50,
         }
     }
 }
@@ -80,6 +100,11 @@ impl TryFrom<OpRbuilderArgs> for FlashblocksConfig {
             fixed,
             calculate_state_root,
             flashblocks_number_contract_address,
+            p2p_enabled: args.flashblocks.p2p.p2p_enabled,
+            p2p_port: args.flashblocks.p2p.p2p_port,
+            p2p_private_key_file: args.flashblocks.p2p.p2p_private_key_file,
+            p2p_known_peers: args.flashblocks.p2p.p2p_known_peers,
+            p2p_max_peer_count: args.flashblocks.p2p.p2p_max_peer_count,
         })
     }
 }

crates/op-rbuilder/src/builders/flashblocks/config.rs

@@ -6,7 +6,9 @@ use service::FlashblocksServiceBuilder;
 mod best_txs;
 mod builder_tx;
 mod config;
+mod p2p;
 mod payload;
+mod payload_handler;
 mod service;
 mod wspub;
 

crates/op-rbuilder/src/builders/flashblocks/mod.rs

@@ -0,0 +1,60 @@
+use alloy_primitives::U256;
+use reth::{core::primitives::SealedBlock, payload::PayloadId};
+use reth_optimism_payload_builder::OpBuiltPayload as RethOpBuiltPayload;
+use reth_optimism_primitives::OpBlock;
+use serde::{Deserialize, Serialize};
+
+pub(super) const AGENT_VERSION: &str = "op-rbuilder/1.0.0";
+pub(super) const FLASHBLOCKS_STREAM_PROTOCOL: p2p::StreamProtocol =
+    p2p::StreamProtocol::new("/flashblocks/1.0.0");
+
+#[derive(Clone, Debug, PartialEq, Deserialize, Serialize)]
+pub(super) enum Message {
+    OpBuiltPayload(OpBuiltPayload),
+}
+
+impl p2p::Message for Message {
+    fn protocol(&self) -> p2p::StreamProtocol {
+        FLASHBLOCKS_STREAM_PROTOCOL
+    }
+}
+
+/// Internal type analogous to [`reth_optimism_payload_builder::OpBuiltPayload`]
+/// which additionally implements `Serialize` and `Deserialize` for p2p transmission.
+#[derive(Clone, Debug, PartialEq, Deserialize, Serialize)]
+pub(crate) struct OpBuiltPayload {
+    /// Identifier of the payload
+    pub(crate) id: PayloadId,
+    /// Sealed block
+    pub(crate) block: SealedBlock<OpBlock>,
+    /// The fees of the block
+    pub(crate) fees: U256,
+}
+
+impl From<RethOpBuiltPayload> for Message {
+    fn from(value: RethOpBuiltPayload) -> Self {
+        Message::OpBuiltPayload(value.into())
+    }
+}
+
+impl From<OpBuiltPayload> for Message {
+    fn from(value: OpBuiltPayload) -> Self {
+        Message::OpBuiltPayload(value)
+    }
+}
+
+impl From<OpBuiltPayload> for RethOpBuiltPayload {
+    fn from(value: OpBuiltPayload) -> Self {
+        RethOpBuiltPayload::new(value.id, value.block.into(), value.fees, None)
+    }
+}
+
+impl From<RethOpBuiltPayload> for OpBuiltPayload {
+    fn from(value: RethOpBuiltPayload) -> Self {
+        OpBuiltPayload {
+            id: value.id(),
+            block: value.block().clone(),
+            fees: value.fees(),
+        }
+    }
+}