feat(l1): add tx broadcasting time interval flag to cli

.github/workflows/pr-main_l1.yaml

@@ -196,7 +196,7 @@ jobs:
             test_pattern: engine-(auth|exchange-capabilities)/
           - name: "Cancun Engine tests"
             simulation: ethereum/engine
-            test_pattern: "engine-cancun/Blob Transactions On Block 1|Blob Transaction Ordering|Parallel Blob Transactions|ForkchoiceUpdatedV3|ForkchoiceUpdatedV2|ForkchoiceUpdated Version|GetPayload|NewPayloadV3 After Cancun|NewPayloadV3 Before Cancun|NewPayloadV3 Versioned Hashes|Incorrect BlobGasUsed|ParentHash equals BlockHash|RPC:|in ForkchoiceState|Unknown SafeBlockHash|Unknown FinalizedBlockHash|Unique|Re-Execute Payload|Multiple New Payloads|NewPayload with|Build Payload with|Re-org to Previously|Safe Re-Org to Side Chain|Transaction Re-Org|Re-Org Back into Canonical Chain|Suggested Fee Recipient Test|PrevRandao Opcode|Fork ID: *|Request Blob Pooled Transactions Single|Invalid NewPayload, Incomplete Transactions|Re-Org Back to Canonical Chain*|Invalid PayloadAttributes*|Invalid NewPayload, VersionedHashes|Invalid NewPayload, Incomplete VersionedHashes|Invalid NewPayload, Extra VersionedHashes|Bad Hash on NewPayload|Unknown HeadBlockHash|In-Order Consecutive Payload Execution|Valid NewPayload->ForkchoiceUpdated|Invalid NewPayload, ParentHash|Syncing=False|Payload Build after New Invalid Payload|Invalid NewPayload|Invalid Missing Ancestor ReOrg|Invalid Missing Ancestor Syncing ReOrG"
+            test_pattern: "engine-cancun"
           - name: "Paris Engine tests"
             simulation: ethereum/engine
             test_pattern: "engine-api/RPC|Bad Hash on NewPayload|Build Payload|Fork ID|In-Order Consecutive Payload Execution|Inconsistent|Invalid Missing Ancestor ReOrg|Invalid NewPayload|Invalid PayloadAttributes|Multiple New Payloads|NewPayload with|ParentHash equals BlockHash on NewPayload|Payload Build|PrevRandao Opcode Transactions|Re-Execute Payload|Re-Org Back|Re-org to Previously Validated Sidechain Payload|RPC:|Safe Re-Org|Suggested Fee|Transaction Re-Org|Unique Payload ID|Unknown|Valid NewPayload->ForkchoiceUpdated" # |Invalid P9 -> flaky

cmd/ethrex/cli.rs

@@ -175,6 +175,14 @@ pub struct Options {
         help_heading = "P2P options"
     )]
     pub discovery_port: String,
+    #[arg(
+        long = "p2p.tx-broadcasting-interval",
+        default_value = "1000",
+        value_name = "INTERVAL_MS",
+        help = "Transaction Broadcasting Time Interval (ms) for batching transactions before broadcasting them.",
+        help_heading = "P2P options"
+    )]
+    pub tx_broadcasting_time_interval: u64,
     #[arg(
         long = "block-producer.extra-data",
         default_value = get_minimal_client_version(),
@@ -246,6 +254,7 @@ impl Default for Options {
             dev: Default::default(),
             force: false,
             mempool_max_size: Default::default(),
+            tx_broadcasting_time_interval: Default::default(),
             extra_data: get_minimal_client_version(),
         }
     }

cmd/ethrex/initializers.rs

@@ -197,6 +197,7 @@ pub async fn init_network(
         blockchain.clone(),
         get_client_version(),
         based_context,
+        opts.tx_broadcasting_time_interval,
     )
     .await
     .expect("P2P context could not be created");

crates/networking/p2p/network.rs

@@ -60,17 +60,22 @@ impl P2PContext {
         blockchain: Arc<Blockchain>,
         client_version: String,
         based_context: Option<P2PBasedContext>,
+        tx_broadcasting_time_interval: u64,
     ) -> Result<Self, NetworkError> {
         let (channel_broadcast_send_end, _) = tokio::sync::broadcast::channel::<(
             tokio::task::Id,
             Arc<Message>,
         )>(MAX_MESSAGES_TO_BROADCAST);
 
-        let tx_broadcaster = TxBroadcaster::spawn(peer_table.clone(), blockchain.clone())
-            .await
-            .inspect_err(|e| {
-                error!("Failed to start Tx Broadcaster: {e}");
-            })?;
+        let tx_broadcaster = TxBroadcaster::spawn(
+            peer_table.clone(),
+            blockchain.clone(),
+            tx_broadcasting_time_interval,
+        )
+        .await
+        .inspect_err(|e| {
+            error!("Failed to start Tx Broadcaster: {e}");
+        })?;
 
         Ok(P2PContext {
             local_node,

crates/networking/p2p/tx_broadcaster.rs

@@ -34,9 +34,6 @@ const PRUNE_WAIT_TIME_SECS: u64 = 600; // 10 minutes
 // Amount of seconds between each prune
 const PRUNE_INTERVAL_SECS: u64 = 360; // 6 minutes
 
-// Amount of seconds between each broadcast
-const BROADCAST_INTERVAL_SECS: u64 = 1; // 1 second
-
 #[derive(Debug, Clone, Default)]
 struct PeerMask {
     bits: Vec<u64>,
@@ -115,6 +112,7 @@ impl TxBroadcaster {
     pub async fn spawn(
         kademlia: PeerTableHandle,
         blockchain: Arc<Blockchain>,
+        tx_broadcasting_time_interval: u64,
     ) -> Result<GenServerHandle<TxBroadcaster>, TxBroadcasterError> {
         info!("Starting Transaction Broadcaster");
 
@@ -129,7 +127,7 @@ impl TxBroadcaster {
         let server = state.clone().start();
 
         send_interval(
-            Duration::from_secs(BROADCAST_INTERVAL_SECS),
+            Duration::from_millis(tx_broadcasting_time_interval),
             server.clone(),
             InMessage::BroadcastTxs,
         );

docs/CLI.md

@@ -83,6 +83,11 @@ P2P options:
           UDP port for P2P discovery.
 
           [default: 30303]
+
+        --p2p.tx-broadcasting-interval <INTERVAL_MS>
+            Transaction Broadcasting Time Interval (ms) for batching transactions before broadcasting them.
+
+            [default: 1000]
 
 RPC options:
       --http.addr <ADDRESS>