Add websocket and rpc request gen to txgen

Cargo.toml

@@ -99,6 +99,7 @@ alloy-eips = "0.8"
 alloy-json-rpc = "0.8"
 alloy-network = "0.8"
 alloy-primitives = "=0.8.18"
+alloy-provider = "0.8"
 alloy-rlp = "0.3"
 alloy-rpc-client = "0.8"
 alloy-rpc-types = "0.8"
@@ -205,6 +206,7 @@ tiny-keccak = "2"
 tokio = { version = "1.39", features = ["sync"] }
 tokio-retry = "0.3.0"
 tokio-stream = "0.1.17"
+tokio-tungstenite = {version="0.24.0", features = ["native-tls"]}
 tokio-util = "0.7"
 toml = "0.7"
 tracing = "0.1"

monad-eth-testutil/Cargo.toml

@@ -23,9 +23,11 @@ monad-validator = { workspace = true }
 alloy-consensus = { workspace = true, features = ["k256"] }
 alloy-eips = { workspace = true }
 alloy-primitives = { workspace = true }
+alloy-provider = { workspace = true, features = ["ws"] }
 alloy-signer = { workspace = true }
 alloy-signer-local = { workspace = true }
 rand = { workspace = true }
+tokio-tungstenite = { workspace = true, features = ["native-tls"]}
 
 [dev-dependencies]
 alloy-json-rpc = { workspace = true }

monad-eth-testutil/examples/txgen/cli.rs

@@ -28,6 +28,9 @@ pub struct CliConfig {
     #[arg(long, global = true)]
     pub rpc_url: Option<Url>,
 
+    #[arg(long, global = true)]
+    pub ws_url: Option<Url>,
+
     /// Target tps of the generator
     #[arg(long, global = true)]
     pub tps: Option<u64>,
@@ -125,6 +128,14 @@ pub struct CliConfig {
     #[arg(long, global = true)]
     pub use_static_tps_interval: Option<bool>,
 
+    /// Spams rpc with common wallet workflow requests
+    #[arg(long, global = true)]
+    pub spam_rpc: Option<bool>,
+
+    /// Compares rpc and websocket responses
+    #[arg(long, global = true)]
+    pub compare_rpc_ws: Option<bool>,
+
     /// Otel endpoint
     #[arg(long, global = true)]
     pub otel_endpoint: Option<String>,

monad-eth-testutil/examples/txgen/config.rs

@@ -30,6 +30,9 @@ pub struct Config {
     #[serde(default)]
     pub rpc_urls: Vec<String>,
 
+    #[serde(default)]
+    pub ws_url: String,
+
     /// Funded private keys used to seed native tokens to sender accounts
     pub root_private_keys: Vec<String>,
 
@@ -80,6 +83,7 @@ impl Default for Config {
     fn default() -> Self {
         Self {
             rpc_urls: vec!["http://localhost:8545".to_string()],
+            ws_url: "ws://localhost:8546".to_string(),
             root_private_keys: vec![
                 "0xac0974bec39a17e36ba4a6b4d238ff944bacb478cbed5efcae784d7bf4f2ff80".to_string(),
                 "0x59c6995e998f97a5a0044966f0945389dc9e86dae88c7a8412f4603b6b78690d".to_string(),
@@ -253,6 +257,12 @@ impl Config {
             })
             .collect()
     }
+
+    pub fn ws_url(&self) -> Result<Url> {
+        self.ws_url
+            .parse()
+            .wrap_err_with(|| format!("Failed to parse WS URL: {}", self.ws_url))
+    }
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
@@ -305,6 +315,12 @@ pub struct TrafficGen {
     /// How many txs should be generated per sender per cycle.
     /// Or put another way, how many txs should be generated before refreshing the nonce from chain state
     pub tx_per_sender: Option<usize>,
+
+    // Should the txgen spam rpc and websocket with wallet workflow requests
+    pub spam_rpc: bool,
+
+    /// Should the txgen compare rpc and websocket responses
+    pub compare_rpc_ws: bool,
 }
 
 impl Default for TrafficGen {
@@ -321,6 +337,8 @@ impl Default for TrafficGen {
             }),
             sender_group_size: None,
             tx_per_sender: None,
+            spam_rpc: false,
+            compare_rpc_ws: false,
         }
     }
 }

monad-eth-testutil/examples/txgen/run.rs

@@ -30,7 +30,10 @@ use serde::{Deserialize, Serialize};
 use crate::{
     config::{Config, DeployedContract, TrafficGen},
     generators::make_generator,
-    prelude::*,
+    prelude::{
+        rpc_request_gen::{RpcWalletSpam, RpcWsCompare},
+        *,
+    },
     shared::{
         ecmul::ECMul, eip7702::EIP7702, erc20::ERC20, eth_json_rpc::EthJsonRpc, uniswap::Uniswap,
     },
@@ -232,7 +235,7 @@ fn run_traffic_gen(
     let refresher = Refresher::new(
         rpc_rx,
         gen_sender,
-        read_client,
+        read_client.clone(),
         Arc::clone(&metrics),
         base_fee,
         Duration::from_secs_f64(config.refresh_delay_secs),
@@ -242,12 +245,40 @@ fn run_traffic_gen(
         Arc::clone(shutdown),
     )?;
 
-    Ok([
-        critical_task("Refresher", tokio::spawn(refresher.run())).boxed(),
-        critical_task("Rpc Sender", tokio::spawn(rpc_sender.run())).boxed(),
-        critical_task("Generator Harness", tokio::spawn(gen.run())).boxed(),
-    ]
-    .into_iter())
+    let mut tasks = Vec::new();
+
+    if traffic_gen.spam_rpc {
+        let spammer = RpcWalletSpam::new(
+            read_client.clone(),
+            config.ws_url().expect("WS URL is not valid"),
+        );
+        tasks.push(
+            critical_task("Spammer", tokio::spawn(async move { spammer.run().await })).boxed(),
+        );
+    }
+
+    if traffic_gen.compare_rpc_ws {
+        let compare_rpc_ws = RpcWsCompare::new(
+            read_client.clone(),
+            config.ws_url().expect("WS URL is not valid"),
+        );
+        tasks.push(
+            critical_task(
+                "Compare RPC WS",
+                tokio::spawn(async move { compare_rpc_ws.run().await }),
+            )
+            .boxed(),
+        );
+    }
+
+    Ok(tasks
+        .into_iter()
+        .chain([
+            critical_task("Refresher", tokio::spawn(refresher.run())).boxed(),
+            critical_task("Rpc Sender", tokio::spawn(rpc_sender.run())).boxed(),
+            critical_task("Generator Harness", tokio::spawn(gen.run())).boxed(),
+        ])
+        .into_iter())
 }
 
 async fn helper_task(

monad-eth-testutil/examples/txgen/workers/mod.rs

@@ -27,6 +27,7 @@ pub mod committed_tx_watcher;
 pub mod gen_harness;
 pub mod metrics;
 pub mod refresher;
+pub mod rpc_request_gen;
 pub mod rpc_sender;
 
 pub use committed_tx_watcher::*;