P2P Flashblocks Propagation

Cargo.toml

@@ -5,10 +5,14 @@ members = [
     "crates/rollup-boost",
     "crates/websocket-proxy",
     "crates/flashblocks-rpc",
+    "crates/flashblocks-p2p",
+    "crates/flashblocks-node",
 ]
 
 [workspace.dependencies]
 rollup-boost = { path = "crates/rollup-boost" }
+flashblocks-p2p = { path = "crates/flashblocks-p2p" }
+flashblocks-rpc = { path = "crates/flashblocks-rpc" }
 
 tracing = "0.1.4"
 tracing-subscriber = { version = "0.3.11", features = ["env-filter", "json"] }
@@ -20,12 +24,22 @@ serde_json = "1.0.96"
 metrics = "0.24.0"
 metrics-derive = "0.1"
 tokio = { version = "1", features = ["full"] }
+tokio-stream = "0.1.17"
 eyre = "0.6.12"
 url = "2.2.0"
 sha2 = { version = "0.10", default-features = false }
 
 # Reth deps
-reth-optimism-payload-builder = { git = "https://github.com/paradigmxyz/reth", tag = "v1.5.0" }
+reth-optimism-payload-builder = { git = "https://github.com/paradigmxyz/reth", tag = "v1.5.1" }
+reth = { git = "https://github.com/paradigmxyz/reth", tag = "v1.5.1" }
+reth-ethereum = { git = "https://github.com/paradigmxyz/reth", tag = "v1.5.1" }
+reth-eth-wire = { git = "https://github.com/paradigmxyz/reth", tag = "v1.5.1" }
+reth-network = { git = "https://github.com/paradigmxyz/reth", tag = "v1.5.1" }
+reth-network-peers = { git = "https://github.com/paradigmxyz/reth", tag = "v1.5.1" }
+reth-node-api = { git = "https://github.com/paradigmxyz/reth", tag = "v1.5.1" }
+reth-node-builder = { git = "https://github.com/paradigmxyz/reth", tag = "v1.5.1" }
+reth-provider = { git = "https://github.com/paradigmxyz/reth", tag = "v1.5.1" }
+reth-transaction-pool = { git = "https://github.com/paradigmxyz/reth", tag = "v1.5.1" }
 
 # Alloy libraries
 alloy-rpc-types-engine = "1.0.13"
@@ -38,6 +52,7 @@ alloy-consensus = "1.0.13"
 alloy-rpc-types = "1.0.13"
 alloy-genesis = "1.0.13"
 alloy-rpc-client = "1.0.13"
+alloy-rlp = "0.3.12"
 alloy-provider = "1.0.13"
 op-alloy-network = "0.18.7"
 op-alloy-rpc-types-engine = "0.18.7"

crates/flashblocks-node/Cargo.toml

@@ -0,0 +1,66 @@
+[package]
+name = "flashblocks-node"
+version = "0.1.0"
+edition = "2024"
+license = "MIT"
+
+[dependencies]
+rollup-boost.workspace = true
+flashblocks-p2p.workspace = true
+flashblocks-rpc.workspace = true
+
+reth-optimism-node = { git = "https://github.com/paradigmxyz/reth", tag = "v1.5.1" }
+reth-optimism-cli = { git = "https://github.com/paradigmxyz/reth", tag = "v1.5.1" }
+reth-rpc-eth-api = { git = "https://github.com/paradigmxyz/reth", tag = "v1.5.1" }
+reth-optimism-primitives = { git = "https://github.com/paradigmxyz/reth", tag = "v1.5.1" }
+reth-optimism-chainspec = { git = "https://github.com/paradigmxyz/reth", tag = "v1.5.1" }
+reth-optimism-rpc = { git = "https://github.com/paradigmxyz/reth", tag = "v1.5.1" }
+reth-optimism-evm = { git = "https://github.com/paradigmxyz/reth", tag = "v1.5.1" }
+reth-optimism-forks = { git = "https://github.com/paradigmxyz/reth", tag = "v1.5.1" }
+reth-provider = { git = "https://github.com/paradigmxyz/reth", tag = "v1.5.1" }
+reth-primitives = { git = "https://github.com/paradigmxyz/reth", tag = "v1.5.1" }
+reth-primitives-traits = { git = "https://github.com/paradigmxyz/reth", tag = "v1.5.1" }
+reth-node-builder = { git = "https://github.com/paradigmxyz/reth", tag = "v1.5.1", features = [
+    "test-utils",
+] }
+reth-node-api = { git = "https://github.com/paradigmxyz/reth", tag = "v1.5.1" }
+reth-tasks = { git = "https://github.com/paradigmxyz/reth", tag = "v1.5.1" }
+reth-node-core = { git = "https://github.com/paradigmxyz/reth", tag = "v1.5.1" }
+reth-db = { git = "https://github.com/paradigmxyz/reth", tag = "v1.5.1" }
+reth-tracing = { git = "https://github.com/paradigmxyz/reth", tag = "v1.5.1" }
+reth-rpc-server-types = { git = "https://github.com/paradigmxyz/reth", tag = "v1.5.1" }
+
+reth-ethereum = { workspace = true, features = ["node", "network", "cli"] }
+reth-eth-wire = { workspace = true }
+reth-network = { workspace = true }
+reth-network-peers = { workspace = true }
+ed25519-dalek = { version = "2", features = ["serde"] }
+
+alloy-eips.workspace = true
+alloy-primitives.workspace = true
+alloy-json-rpc.workspace = true
+alloy-rpc-types.workspace = true
+alloy-rpc-types-engine.workspace = true
+alloy-rpc-types-eth.workspace = true
+alloy-consensus.workspace = true
+alloy-genesis.workspace = true
+alloy-rpc-client.workspace = true
+alloy-provider.workspace = true
+op-alloy-network.workspace = true
+op-alloy-consensus.workspace = true
+op-alloy-rpc-types.workspace = true
+
+tokio.workspace = true
+tokio-tungstenite.workspace = true
+url.workspace = true
+tracing.workspace = true
+clap.workspace = true
+eyre.workspace = true
+metrics.workspace = true
+metrics-derive.workspace = true
+serde_json.workspace = true
+serde.workspace = true
+
+jsonrpsee = { version = "0.25.1" }
+futures-util = "0.3.31"
+brotli = "8.0.1"

crates/flashblocks-rpc/src/bin/main.rs → crates/flashblocks-node/src/bin/main.rs

@@ -1,39 +1,41 @@
 #![allow(missing_docs, rustdoc::missing_crate_level_docs)]
-
 use clap::Parser;
+use ed25519_dalek::VerifyingKey;
+use flashblocks_node::FlashblocksNodeArgs;
+use flashblocks_p2p::protocol::handler::FlashblocksHandler;
 use flashblocks_rpc::{EthApiOverrideServer, FlashblocksApiExt, FlashblocksOverlay};
+use reth_ethereum::network::{NetworkProtocols, protocol::IntoRlpxSubProtocol};
 use reth_optimism_cli::{Cli, chainspec::OpChainSpecParser};
 use reth_optimism_node::{OpNode, args::RollupArgs};
+use tokio::sync::broadcast;
 use tracing::info;
 
-#[derive(Debug, Clone, PartialEq, Eq, clap::Args)]
+#[derive(Debug, Clone, clap::Args)]
 #[command(next_help_heading = "Rollup")]
 struct FlashblocksRollupArgs {
     #[command(flatten)]
-    rollup_args: RollupArgs,
-
-    #[arg(long = "flashblocks.enabled", default_value = "false")]
-    flashblocks_enabled: bool,
+    pub rollup_args: RollupArgs,
 
-    #[arg(long = "flashblocks.websocket-url", value_name = "WEBSOCKET_URL")]
-    websocket_url: url::Url,
+    #[command(flatten)]
+    pub flashblock_args: Option<FlashblocksNodeArgs>,
 }
 
-fn main() {
+pub fn main() {
     if let Err(err) =
         Cli::<OpChainSpecParser, FlashblocksRollupArgs>::parse().run(async move |builder, args| {
             let rollup_args = args.rollup_args;
             let chain_spec = builder.config().chain.clone();
+            let (inbound_tx, inbound_rx) = broadcast::channel(100);
+            let (_publish_tx, publish_rx) = broadcast::channel(100);
+
+            let flashblocks_overlay = FlashblocksOverlay::new(chain_spec, inbound_rx);
+            flashblocks_overlay.clone().start()?;
 
             info!(target: "reth::cli", "Launching Flashblocks RPC overlay node");
             let handle = builder
                 .node(OpNode::new(rollup_args))
                 .extend_rpc_modules(move |ctx| {
-                    if args.flashblocks_enabled {
-                        let mut flashblocks_overlay =
-                            FlashblocksOverlay::new(args.websocket_url, chain_spec);
-                        flashblocks_overlay.start()?;
-
+                    if args.flashblock_args.is_some() {
                         let eth_api = ctx.registry.eth_api().clone();
                         let api_ext = FlashblocksApiExt::new(eth_api.clone(), flashblocks_overlay);
 
@@ -43,6 +45,18 @@ fn main() {
                 })
                 .launch_with_debug_capabilities()
                 .await?;
+
+            let custom_rlpx_handler = FlashblocksHandler::new(
+                handle.node.network.clone(),
+                VerifyingKey::default(),
+                inbound_tx,
+                publish_rx,
+            );
+
+            handle
+                .node
+                .network
+                .add_rlpx_sub_protocol(custom_rlpx_handler.into_rlpx_sub_protocol());
             handle.node_exit_future.await
         })
     {

crates/flashblocks-node/src/lib.rs

@@ -0,0 +1,19 @@
+use clap::Args;
+use ed25519_dalek::VerifyingKey;
+use rollup_boost::parse_vk;
+
+#[derive(Args, Clone, Debug)]
+#[group(requires = "flashblocks")]
+pub struct FlashblocksNodeArgs {
+    /// Enable Flashblocks client
+    #[arg(long, env, required = false)]
+    pub flashblocks: bool,
+
+    #[arg(
+        long = "flashblocks.authorizor_vk",
+        env = "FLASHBLOCKS_AUTHORIZOR_VK",
+        value_parser = parse_vk,
+        required = false,
+    )]
+    pub authorizor_vk: VerifyingKey,
+}