Add support for subsecond spam intervals

crates/cli/src/commands/campaign.rs

@@ -362,6 +362,8 @@ fn create_spam_cli_args(
             } else {
                 None
             },
+            txs_per_period: None,
+            period_millis: 1000,
             txs_per_block: if matches!(spam_mode, CampaignMode::Tpb) {
                 Some(spam_rate)
             } else {

crates/cli/src/commands/common.rs

@@ -335,6 +335,37 @@ pub struct SendSpamCliArgs {
     )]
     pub txs_per_second: Option<u64>,
 
+    /// The number of txs to send per period (a tick of `--period-millis`) using
+    /// the timed spammer. Like `--txs-per-second`, but the period length is
+    /// configurable. May not be set if `txs_per_block` or `txs_per_second` is set.
+    #[arg(
+        global = true,
+        long,
+        long_help =
+"Number of txs to send per period, where one period is --period-millis long (default 1000ms).
+This is the count sent on each tick, sent as-is with no per-second division, so the rate
+is exactly --txs-per-period txs every --period-millis. Use it to pace load at sub-second
+granularity (e.g. --tpp 4 --period-millis 100 sends 4 txs every 100ms). Must not be set if
+--txs-per-second or --txs-per-block is set. With --period-millis 1000 it is identical to --tps.",
+        visible_aliases = ["tpp"],
+        help_heading = HELP_HEADING_COMMON,
+    )]
+    pub txs_per_period: Option<u64>,
+
+    /// Length of one timed-spammer period in milliseconds. Only meaningful with
+    /// `--txs-per-period`. Defaults to 1000 (one batch per second).
+    #[arg(
+        global = true,
+        long,
+        default_value_t = 1000,
+        long_help =
+"Length of one timed-spammer period in milliseconds. Each period, --txs-per-period txs are sent.
+Only meaningful with --txs-per-period; ignored otherwise. Defaults to 1000ms, which makes
+--txs-per-period behave exactly like --txs-per-second.",
+        help_heading = HELP_HEADING_COMMON,
+    )]
+    pub period_millis: u64,
+
     /// The number of txs to send per block using the blockwise spammer.
     /// May not be set if `txs_per_second` is set. Requires `prv_keys` to be set.
     #[arg(
@@ -393,11 +424,16 @@ Requires --priv-key to be set for each 'from' address in the given testfile.",
 
 impl SendSpamCliArgs {
     pub fn spam_rate(&self) -> Result<SpamRate, ArgsError> {
-        match (self.txs_per_second, self.txs_per_block) {
-            (Some(_), Some(_)) => Err(ArgsError::SpamRateNotFound),
-            (None, None) => Err(ArgsError::SpamRateNotFound),
-            (Some(tps), None) => Ok(SpamRate::TxsPerSecond(tps)),
-            (None, Some(tpb)) => Ok(SpamRate::TxsPerBlock(tpb)),
+        // --tpp is a timed rate like --tps but with a configurable period. For
+        // rate purposes (builtin scenarios, campaigns) it is the per-period count,
+        // mapped to TxsPerSecond; the period length is applied separately where
+        // the timed spammer is constructed.
+        match (self.txs_per_second, self.txs_per_period, self.txs_per_block) {
+            (None, None, None) => Err(ArgsError::SpamRateNotFound),
+            (Some(tps), None, None) => Ok(SpamRate::TxsPerSecond(tps)),
+            (None, Some(tpp), None) => Ok(SpamRate::TxsPerSecond(tpp)),
+            (None, None, Some(tpb)) => Ok(SpamRate::TxsPerBlock(tpb)),
+            _ => Err(ArgsError::SpamRateNotFound),
         }
     }
 }

crates/cli/src/commands/spam.rs

@@ -334,6 +334,8 @@ impl SpamCommandArgs {
         let SendSpamCliArgs {
             builder_url,
             txs_per_second,
+            txs_per_period,
+            period_millis,
             txs_per_block,
             duration,
             pending_timeout,
@@ -352,7 +354,10 @@ impl SpamCommandArgs {
 
         let mut testconfig = self.testconfig().await?;
         let spam_len = testconfig.spam.as_ref().map(|s| s.len()).unwrap_or(0);
-        let txs_per_duration = txs_per_block.unwrap_or(txs_per_second.unwrap_or(spam_len as u64));
+        let txs_per_duration = txs_per_block
+            .or(txs_per_second)
+            .or(txs_per_period)
+            .unwrap_or(spam_len as u64);
         let engine_params = self.engine_params().await?;
 
         // If send_raw_tx_sync is set alongside rpc_batch_size, warn and disable batching.
@@ -383,8 +388,17 @@ impl SpamCommandArgs {
             }
         }
 
+        // Number of spammer ticks over the run: for the sub-second timed path
+        // (--tpp + --period-millis) this exceeds `duration` seconds; otherwise
+        // it equals `duration` (seconds for --tps, blocks for --tpb).
+        let num_periods = if txs_per_period.is_some() {
+            duration.saturating_mul(1000) / period_millis.max(1)
+        } else {
+            duration
+        };
+
         // check if txs_per_duration is enough to cover the spam requests
-        if (txs_per_duration * duration) < spam_len as u64 {
+        if (txs_per_duration * num_periods) < spam_len as u64 {
             return Err(ArgsError::TransactionsPerDurationInsufficient {
                 min_tpd: spam_len as u64,
                 tpd: txs_per_duration,
@@ -481,14 +495,28 @@ impl SpamCommandArgs {
         let block_time = get_block_time(&rpc_client).await?;
 
         check_private_keys(&testconfig, &user_signers);
-        if txs_per_block.is_some() && txs_per_second.is_some() {
-            panic!("Cannot set both --txs-per-block and --txs-per-second");
+        let rate_flags_set = [
+            txs_per_block.is_some(),
+            txs_per_second.is_some(),
+            txs_per_period.is_some(),
+        ]
+        .iter()
+        .filter(|set| **set)
+        .count();
+        if rate_flags_set > 1 {
+            panic!(
+                "Set only one of {}, {}, or {}",
+                bold("--txs-per-block (--tpb)"),
+                bold("--txs-per-second (--tps)"),
+                bold("--txs-per-period (--tpp)")
+            );
         }
-        if txs_per_block.is_none() && txs_per_second.is_none() {
+        if rate_flags_set == 0 {
             panic!(
-                "Must set either {} or {}",
+                "Must set one of {}, {}, or {}",
                 bold("--txs-per-block (--tpb)"),
-                bold("--txs-per-second (--tps)")
+                bold("--txs-per-second (--tps)"),
+                bold("--txs-per-period (--tpp)")
             );
         }
 
@@ -605,9 +633,11 @@ impl SpamCommandArgs {
         done_fcu.store(true, std::sync::atomic::Ordering::SeqCst);
 
         // estimate spam cost. contracts must be deployed at this point,
-        // otherwise you'll get "contract not found" errors
+        // otherwise you'll get "contract not found" errors. Cost scales with the
+        // number of batches sent (num_periods), which for sub-second periods
+        // exceeds the wall-clock duration in seconds.
         let total_cost =
-            U256::from(duration) * test_scenario.get_max_spam_cost(&user_signers).await?;
+            U256::from(num_periods) * test_scenario.get_max_spam_cost(&user_signers).await?;
         if min_balance < U256::from(total_cost) {
             return Err(ArgsError::MinBalanceInsufficient {
                 min_balance,
@@ -616,10 +646,10 @@ impl SpamCommandArgs {
             .into());
         }
 
-        let duration_unit = if txs_per_second.is_some() {
-            "second"
-        } else {
+        let duration_unit = if txs_per_block.is_some() {
             "block"
+        } else {
+            "second"
         };
         let duration_units = if duration > 1 {
             format!("{duration_unit}s")
@@ -757,6 +787,8 @@ where
     } = spam_args.to_owned();
     let SendSpamCliArgs {
         txs_per_second,
+        txs_per_period,
+        period_millis,
         txs_per_block,
         duration,
         pending_timeout,
@@ -813,18 +845,41 @@ where
         _ => err,
     };
 
-    let (spammer, txs_per_batch, spam_duration) = if let Some(txs_per_block) = txs_per_block {
+    // `num_periods` is the number of spammer ticks (what spam_rpc consumes);
+    // `spam_duration` is the run-record label, kept in the spammer's natural unit
+    // (blocks or wall-clock seconds) regardless of sub-second period length.
+    let (spammer, txs_per_batch, num_periods, spam_duration) = if let Some(txs_per_block) =
+        txs_per_block
+    {
         info!("Blockwise spammer starting. Sending {txs_per_block} txs per block.");
         (
             TypedSpammer::Blockwise(BlockwiseSpammer::new()),
             txs_per_block,
+            duration,
             SpamDuration::Blocks(duration),
         )
     } else if let Some(txs_per_second) = txs_per_second {
         info!("Timed spammer starting. Sending {txs_per_second} txs per second.");
         (
             TypedSpammer::Timed(TimedSpammer::new(std::time::Duration::from_secs(1))),
             txs_per_second,
+            duration,
+            SpamDuration::Seconds(duration),
+        )
+    } else if let Some(txs_per_period) = txs_per_period {
+        // floor(duration_secs * 1000 / period_millis): if the period doesn't
+        // divide the wall-clock duration evenly, run slightly under rather
+        // than over. The run record still reports the requested seconds.
+        let num_periods = duration.saturating_mul(1000) / period_millis.max(1);
+        info!(
+                "Timed spammer starting. Sending {txs_per_period} txs every {period_millis}ms ({num_periods} periods over ~{duration}s)."
+            );
+        (
+            TypedSpammer::Timed(TimedSpammer::new(std::time::Duration::from_millis(
+                period_millis,
+            ))),
+            txs_per_period,
+            num_periods,
             SpamDuration::Seconds(duration),
         )
     } else {
@@ -847,7 +902,7 @@ where
             .as_millis();
         let run = SpamRunRequest {
             timestamp: timestamp as usize,
-            tx_count: (txs_per_batch * duration) as usize,
+            tx_count: (txs_per_batch * num_periods) as usize,
             scenario_name,
             campaign_id: campaign_id.clone(),
             campaign_name: campaign_name.clone(),
@@ -880,7 +935,7 @@ where
     // Spawn periodic reporting task if --report-interval is set and we have a run_id
     let report_interval = args.spam_args.report_interval;
     let report_cancel = if let (Some(interval_secs), Some(rid)) = (report_interval, run_id) {
-        let planned_tx_count = txs_per_batch * duration;
+        let planned_tx_count = txs_per_batch * num_periods;
         Some(spawn_spam_report_task(
             db,
             rid,
@@ -935,7 +990,7 @@ where
                 .spam_rpc(
                     test_scenario,
                     txs_per_batch,
-                    duration,
+                    num_periods,
                     run_id,
                     callback.clone(),
                 )

crates/cli/src/server/static/openrpc.json

@@ -16,7 +16,7 @@
       "BuilderParams": {"file":"crates/cli/src/server/rpc_server/types.rs","line":227},
       "BuiltinScenarioCli": {"file":"crates/cli/src/default_scenarios/builtin.rs","line":33},
       "BundleCallDefinition": {"file":"crates/core/src/generator/function_def.rs","line":61},
-      "BundleTypeCli": {"file":"crates/cli/src/commands/common.rs","line":438},
+      "BundleTypeCli": {"file":"crates/cli/src/commands/common.rs","line":474},
       "CompiledContract": {"file":"crates/core/src/generator/create_def.rs","line":8},
       "ContenderSessionInfo": {"file":"crates/cli/src/server/sessions.rs","line":128},
       "CreateDefinition": {"file":"crates/core/src/generator/create_def.rs","line":65},
@@ -45,7 +45,7 @@
       "TestConfig": {"file":"crates/cli/src/server/rpc_server/types.rs","line":83},
       "TestConfigSource": {"file":"crates/cli/src/server/rpc_server/types.rs","line":83},
       "TransferStressCliArgs": {"file":"crates/cli/src/default_scenarios/transfers.rs","line":12},
-      "TxTypeCli": {"file":"crates/cli/src/commands/common.rs","line":407},
+      "TxTypeCli": {"file":"crates/cli/src/commands/common.rs","line":443},
       "UniV2CliArgs": {"file":"crates/cli/src/default_scenarios/uni_v2.rs","line":15}
     }
   },

crates/core/src/spammer/timed.rs

@@ -73,3 +73,93 @@ where
         &self.context
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use alloy::{network::AnyNetwork, providers::ProviderBuilder};
+    use contender_bundle_provider::bundle::BundleType;
+    use std::sync::Arc;
+    use std::time::Instant;
+    use tokio::sync::OnceCell;
+    use tokio_util::sync::CancellationToken;
+
+    use crate::{
+        db::MockDb,
+        generator::{agent_pools::AgentSpec, util::test::spawn_anvil, RandSeed},
+        test_scenario::{tests::MockConfig, TestScenario, TestScenarioParams},
+    };
+
+    static PROM: OnceCell<prometheus::Registry> = OnceCell::const_new();
+    static HIST: OnceCell<prometheus::HistogramVec> = OnceCell::const_new();
+
+    async fn mock_scenario() -> TestScenario<MockDb, RandSeed, MockConfig> {
+        let anvil = spawn_anvil();
+        let _provider = Arc::new(
+            ProviderBuilder::new()
+                .network::<AnyNetwork>()
+                .connect_http(anvil.endpoint_url()),
+        );
+        let seed = RandSeed::seed_from_str("777777777777");
+        TestScenario::new(
+            MockConfig,
+            MockDb.into(),
+            seed,
+            TestScenarioParams {
+                rpc_url: anvil.endpoint_url(),
+                builder_rpc_url: None,
+                txs_rpc_url: None,
+                signers: crate::util::default_signers(),
+                agent_spec: AgentSpec::default(),
+                tx_type: alloy::consensus::TxType::Legacy,
+                bundle_type: BundleType::default(),
+                pending_tx_timeout: Duration::from_secs(12),
+                extra_msg_handles: None,
+                sync_nonces_after_batch: true,
+                rpc_batch_size: 0,
+                gas_price: None,
+                scenario_label: None,
+                send_raw_tx_sync: false,
+                flashblocks_ws_url: None,
+            },
+            None,
+            (&PROM, &HIST).into(),
+            &CancellationToken::new(),
+        )
+        .await
+        .unwrap()
+    }
+
+    // The timed spammer's trigger stream must fire at its configured interval,
+    // not assume a fixed one-second period. A 100ms interval should emit ~10
+    // ticks in the time a 1000ms interval emits one, proving sub-second pacing.
+    #[tokio::test]
+    async fn ticks_fire_at_configured_sub_second_interval() {
+        let mut scenario = mock_scenario().await;
+
+        let spammer = TimedSpammer::new(Duration::from_millis(100));
+        let stream = Spammer::<crate::spammer::NilCallback, MockDb, RandSeed, MockConfig>::on_spam(
+            &spammer,
+            &mut scenario,
+        )
+        .await
+        .unwrap();
+
+        let start = Instant::now();
+        let ticks: Vec<_> = stream.take(5).collect().await;
+        let elapsed = start.elapsed();
+
+        assert_eq!(ticks.len(), 5);
+        // 5 ticks at 100ms each: first tick after the initial wait, so ~5
+        // intervals. Allow generous slack for scheduling, but it must be far
+        // under the ~5s a one-second period would take.
+        assert!(
+            elapsed >= Duration::from_millis(450),
+            "expected >=450ms for 5x100ms ticks, got {elapsed:?}"
+        );
+        assert!(
+            elapsed < Duration::from_millis(2000),
+            "5x100ms ticks took {elapsed:?}; pacing is not sub-second"
+        );
+    }
+}