feat/async cache processing

crates/cli/CHANGELOG.md

@@ -5,6 +5,21 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## Unreleased
+
+- cli is now solely responsible for intercepting CTRL-C signals
+  - to shutdown background tasks, we rely on [`CancellationToken`s](https://docs.rs/tokio-util/latest/tokio_util/sync/struct.CancellationToken.html)
+  - we no longer require two-phase cancellation (CTRL-C once to stop spamming, CTRL-C again to stop result collection)
+    - result collection happens async, so when the user cancels, most results will have already been collected
+    - stopping quickly is a better UX than two-phase
+
+### Breaking changes
+
+- `commands::spam::spam` removes the `&mut TestScenario` param, creates a `TestScenario` from `spam_args` instead
+- `SendSpamCliArgs` replaces `--loops [NUM_LOOPS]` (optional `usize`) with `--forever` (`bool`)
+- some functions are moved from `utils` to `commands::spam`
+- `commands::spamd` has been deleted (it was just a junk wrapper for `spam`)
+
 ## [0.6.0](https://github.com/flashbots/contender/releases/tag/v0.6.0) - 2025-11-25
 
 - support new ENV vars ([#376](https://github.com/flashbots/contender/pull/376))

crates/cli/src/commands/campaign.rs

@@ -308,7 +308,7 @@ fn create_spam_cli_args(
             },
             duration: spam_duration,
             pending_timeout: args.pending_timeout,
-            loops: Some(Some(1)),
+            run_forever: false,
             accounts_per_agent: args.accounts_per_agent,
         },
         ignore_receipts: args.ignore_receipts,
@@ -374,7 +374,6 @@ async fn execute_stage(
         };
 
         let spam_args = SpamCommandArgs::new(spam_scenario, spam_cli_args)?;
-        let scenario = spam_args.init_scenario(db).await?;
         let duration = stage.duration;
         let db = db.clone();
         let campaign_id_owned = campaign_id.to_owned();
@@ -403,8 +402,7 @@ async fn execute_stage(
             // Wait for all parallel scenarios to be ready before starting
             barrier_clone.wait().await;
 
-            let mut scenario = scenario;
-            let run_res = commands::spam(&db, &spam_args, &mut scenario, ctx).await;
+            let run_res = commands::spam(&db, &spam_args, ctx).await;
             match run_res {
                 Ok(Some(run_id)) => {
                     info!(

crates/cli/src/commands/common.rs

@@ -294,12 +294,13 @@ Requires --priv-key to be set for each 'from' address in the given testfile.",
     /// If set without a value, the spam run will be repeated indefinitely.
     /// If not set, the spam run will be executed once.
     #[arg(
-        short,
-        long,
-        num_args = 0..=1,
-        long_help = "The number of times to repeat the spam run. If set with a value, the spam run will be repeated this many times. If set without a value, the spam run will be repeated indefinitely. If not set, the spam run will be repeated once."
+        global = true,
+        default_value_t = false, // pretty sure this line is unnecessary but it makes me feel safe
+        long = "forever",
+        long_help = "Run spammer indefinitely.",
+        visible_aliases = ["indefinite", "indefinitely", "infinite"]
     )]
-    pub loops: Option<Option<u64>>,
+    pub run_forever: bool,
 
     /// The number of accounts to generate for each agent (`from_pool` in scenario files)
     #[arg(

crates/cli/src/commands/mod.rs

@@ -7,13 +7,11 @@ pub mod error;
 pub mod replay;
 mod setup;
 mod spam;
-mod spamd;
 
 use clap::Parser;
-pub use contender_subcommand::{ContenderSubcommand, DbCommand};
-pub use setup::{setup, SetupCommandArgs};
-pub use spam::{spam, EngineArgs, SpamCampaignContext, SpamCliArgs, SpamCommandArgs, SpamScenario};
-pub use spamd::spamd;
+pub use contender_subcommand::*;
+pub use setup::*;
+pub use spam::*;
 
 use crate::error::CliError;
 

crates/cli/src/commands/spam.rs

@@ -9,17 +9,26 @@ use crate::{
     error::CliError,
     util::{
         bold, check_private_keys, fund_accounts, load_seedfile, load_testconfig, parse_duration,
-        provider::AuthClient, spam_callback_default, TypedSpamCallback,
+        provider::AuthClient,
     },
     LATENCY_HIST as HIST, PROM,
 };
-use alloy::{consensus::TxType, primitives::U256, transports::http::reqwest::Url};
+use alloy::{
+    consensus::TxType,
+    primitives::{utils::format_ether, U256},
+    transports::http::reqwest::Url,
+};
 use contender_core::{
     agent_controller::AgentStore,
     db::{DbOps, SpamDuration, SpamRunRequest},
     error::{RuntimeErrorKind, RuntimeParamErrorKind},
-    generator::{seeder::Seeder, templater::Templater, types::SpamRequest, PlanConfig, RandSeed},
-    spammer::{BlockwiseSpammer, Spammer, TimedSpammer},
+    generator::{
+        seeder::Seeder,
+        templater::Templater,
+        types::{AnyProvider, SpamRequest},
+        PlanConfig, RandSeed,
+    },
+    spammer::{BlockwiseSpammer, LogCallback, NilCallback, Spammer, TimedSpammer},
     test_scenario::{TestScenario, TestScenarioParams},
     util::get_block_time,
 };
@@ -205,7 +214,7 @@ impl SpamCommandArgs {
             txs_per_block,
             duration,
             pending_timeout,
-            loops,
+            run_forever,
             accounts_per_agent,
         } = self.spam_args.spam_args.clone();
         let SendTxsCliArgsInner {
@@ -478,17 +487,8 @@ impl SpamCommandArgs {
         }
         done_fcu.store(true, std::sync::atomic::Ordering::SeqCst);
 
-        if loops.is_some_and(|inner_loops| inner_loops.is_none()) {
-            warn!("Spammer agents will eventually run out of funds.");
-            println!(
-                "Make sure you add plenty of funds with {} (set your pre-funded account with {}).",
-                bold("spam --min-balance"),
-                bold("spam -p"),
-            );
-        }
-
-        let total_cost = U256::from(duration * loops.flatten().unwrap_or(1))
-            * test_scenario.get_max_spam_cost(&user_signers).await?;
+        let total_cost =
+            U256::from(duration) * test_scenario.get_max_spam_cost(&user_signers).await?;
         if min_balance < U256::from(total_cost) {
             return Err(ArgsError::MinBalanceInsufficient {
                 min_balance,
@@ -497,6 +497,26 @@ impl SpamCommandArgs {
             .into());
         }
 
+        let duration_unit = if txs_per_second.is_some() {
+            "second"
+        } else {
+            "block"
+        };
+        let duration_units = if duration > 1 {
+            format!("{duration_unit}s")
+        } else {
+            duration_unit.to_owned()
+        };
+        if run_forever {
+            warn!("Spammer agents will eventually run out of funds. Each batch of spam (sent over {duration} {duration_units}) will cost {} ETH.", format_ether(total_cost));
+            // we use println! after warn! because warn! doesn't properly format bold strings
+            println!(
+                "Make sure you add plenty of funds with {} (set your pre-funded account with {}).",
+                bold("spam --min-balance"),
+                bold("spam -p"),
+            );
+        }
+
         Ok(test_scenario)
     }
 
@@ -509,6 +529,39 @@ impl SpamCommandArgs {
     }
 }
 
+pub fn spam_callback_default(
+    log_txs: bool,
+    send_fcu: bool,
+    rpc_client: Option<Arc<AnyProvider>>,
+    auth_client: Option<Arc<dyn ControlChain + Send + Sync + 'static>>,
+    cancel_token: tokio_util::sync::CancellationToken,
+) -> TypedSpamCallback {
+    if let Some(rpc_client) = rpc_client {
+        if log_txs {
+            let log_callback = LogCallback {
+                rpc_provider: rpc_client.clone(),
+                auth_provider: auth_client,
+                send_fcu,
+                cancel_token,
+            };
+            return TypedSpamCallback::Log(log_callback);
+        }
+    }
+    TypedSpamCallback::Nil(NilCallback)
+}
+
+#[derive(Clone)]
+pub enum TypedSpamCallback {
+    Log(LogCallback),
+    Nil(NilCallback),
+}
+
+impl TypedSpamCallback {
+    pub fn is_log(&self) -> bool {
+        matches!(self, TypedSpamCallback::Log(_))
+    }
+}
+
 enum TypedSpammer {
     Blockwise(BlockwiseSpammer),
     Timed(TimedSpammer),
@@ -564,16 +617,12 @@ impl TypedSpammer {
 }
 
 /// Runs spammer and returns run ID.
-pub async fn spam<
-    D: DbOps + Clone + Send + Sync + 'static,
-    S: Seeder + Send + Sync + Clone,
-    P: PlanConfig<String> + Templater<String> + Send + Sync + Clone,
->(
+pub async fn spam<D: DbOps + Clone + Send + Sync + 'static>(
     db: &D,
     args: &SpamCommandArgs,
-    test_scenario: &mut TestScenario<D, S, P>,
     run_context: SpamCampaignContext,
 ) -> Result<Option<u64>> {
+    let mut test_scenario = args.init_scenario(db).await?;
     let SpamCommandArgs {
         scenario,
         spam_args,
@@ -690,10 +739,29 @@ pub async fn spam<
         }
     }
 
-    spammer
-        .spam_rpc(test_scenario, txs_per_batch, duration, run_id, callback)
-        .await
-        .map_err(err_parse)?;
+    loop {
+        tokio::select! {
+            res = spammer
+            .spam_rpc(
+                &mut test_scenario,
+                txs_per_batch,
+                duration,
+                run_id,
+                callback.clone(),
+            ) => {
+                res.map_err(err_parse)?;
+            }
+
+            _ = tokio::signal::ctrl_c() => {
+                println!("\nCTRL-C received, shutting down spam run.");
+                test_scenario.shutdown().await;
+            }
+        }
+
+        if !args.spam_args.spam_args.run_forever || test_scenario.ctx.cancel_token.is_cancelled() {
+            break;
+        }
+    }
 
     Ok(run_id)
 }