feat: implement p2p layer and broadcast flashblocks

Cargo.toml

@@ -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"
 
@@ -179,6 +179,9 @@ flate2 = "1.0.35"
 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" }

crates/op-rbuilder/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,7 @@ serde_yaml = { version = "0.9" }
 moka = "0.12"
 http = "1.0"
 sha3 = "0.10"
-hex = "0.4"
+ureq = "2.10"
 reqwest = "0.12.23"
 k256 = "0.13.4"
 

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

@@ -157,6 +157,38 @@ pub struct FlashblocksArgs {
     )]
     pub flashblocks_calculate_state_root: bool,
 
+    /// Enable libp2p networking for flashblock propagation
+    #[arg(
+        long = "flashblocks.p2p_enabled",
+        env = "FLASHBLOCK_P2P_ENABLED",
+        default_value = "false"
+    )]
+    pub flashblocks_p2p_enabled: bool,
+
+    /// Port for the flashblocks p2p node
+    #[arg(
+        long = "flashblocks.p2p_port",
+        env = "FLASHBLOCK_P2P_PORT",
+        default_value = "9009"
+    )]
+    pub flashblocks_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 flashblocks_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 flashblocks_known_peers: Option<String>,
+
     /// Flashblocks number contract address
     ///
     /// This is the address of the contract that will be used to increment the flashblock number.

crates/op-rbuilder/src/builders/flashblocks/mod.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/p2p.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(),
+        }
+    }
+}