feat: forward txns over p2p

bin/odyssey/src/main.rs

@@ -30,6 +30,7 @@ use eyre::Context;
 use odyssey_node::{
     broadcaster::periodic_broadcaster,
     chainspec::OdysseyChainSpecParser,
+    forwarder::forward_raw_transactions,
     node::OdysseyNode,
     rpc::{EthApiExt, EthApiOverrideServer},
 };
@@ -124,6 +125,12 @@ fn main() {
                 })
                 .await?;
 
+            // spawn raw transaction forwarding
+            let txhandle = node.node.network.transactions_handle().await.unwrap();
+            let raw_txs =
+                node.node.add_ons_handle.eth_api().eth_api().subscribe_to_raw_transactions();
+            node.node.task_executor.spawn(Box::pin(forward_raw_transactions(txhandle, raw_txs)));
+
             node.wait_for_node_exit().await
         })
     {

crates/node/Cargo.toml

@@ -12,9 +12,7 @@ categories.workspace = true
 [dependencies]
 odyssey-common.workspace = true
 
-revm-precompile.workspace = true
-revm-primitives.workspace = true
-
+reth-primitives-traits.workspace = true
 reth-cli.workspace = true
 reth-errors.workspace = true
 reth-node-api.workspace = true
@@ -44,6 +42,8 @@ alloy-rpc-types-eth.workspace  = true
 
 op-alloy-consensus.workspace = true
 
+revm-precompile.workspace = true
+revm-primitives.workspace = true
 
 serde_json.workspace = true
 tokio.workspace = true

crates/node/src/forwarder.rs

@@ -0,0 +1,23 @@
+//! P2P transaction forwarding
+
+use alloy_eips::eip2718::Decodable2718;
+use alloy_primitives::Bytes;
+use reth_network::{transactions::TransactionsHandle, NetworkPrimitives};
+use reth_primitives_traits::transaction::signed::SignedTransaction;
+use tokio::sync::broadcast::Receiver;
+use tracing::trace;
+
+/// Forwards raw transactions to the network.
+pub async fn forward_raw_transactions<N: NetworkPrimitives>(
+    txn: TransactionsHandle<N>,
+    mut raw_txs: Receiver<Bytes>,
+) {
+    loop {
+        if let Ok(raw_tx) = raw_txs.recv().await {
+            if let Ok(tx) = N::BroadcastedTransaction::decode_2718(&mut raw_tx.as_ref()) {
+                trace!(target: "rpc::rpc", tx=%tx.tx_hash(), "Forwarding raw transaction over p2p");
+                txn.broadcast_transactions(Some(tx));
+            }
+        }
+    }
+}

crates/node/src/lib.rs

@@ -18,5 +18,6 @@
 pub mod broadcaster;
 pub mod chainspec;
 pub mod evm;
+pub mod forwarder;
 pub mod node;
 pub mod rpc;

crates/node/src/node.rs

@@ -235,8 +235,8 @@ where
         network_config.peers_config.backoff_durations.medium = Duration::from_secs(5);
         network_config.peers_config.backoff_durations.high = Duration::from_secs(5);
         network_config.peers_config.max_backoff_count = u8::MAX;
-        network_config.sessions_config.session_command_buffer = 500;
-        network_config.sessions_config.session_event_buffer = 500;
+        network_config.sessions_config.session_command_buffer = 750;
+        network_config.sessions_config.session_event_buffer = 750;
 
         let txconfig = TransactionsManagerConfig {
             propagation_mode: TransactionPropagationMode::All,