refactor: async pending tx cache processing

CHANGELOG.md

@@ -7,6 +7,35 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ---
 
+## [Unreleased]
+
+### Changed
+
+- **BREAKING**: Replaced `--loops` flag with `--indefinite` flag for spam command ([#395](https://github.com/flashbots/contender/issues/395))
+  - Use `--indefinite` to run spam continuously until manually stopped
+  - Default behavior (without `--indefinite`) runs spam once
+  - Previous `--loops N` functionality replaced with running spamd N times via script
+  - Migration: If you used `--loops` without a value for infinite loops, use `--indefinite`. If you used `--loops N` for a specific count, run the command N times or use a wrapper script.
+
+### Added
+
+- **Auto-flush configuration**: New `--cache-flush-interval` flag to control how often (in blocks) the pending transaction cache is flushed to the database ([#395](https://github.com/flashbots/contender/issues/395))
+  - Default: 5 blocks
+  - Lower values reduce memory usage but increase DB writes
+  - Higher values batch more efficiently but use more memory
+
+### Improved
+
+- **Spammer now processes receipts in background**: TxActor automatically flushes pending transaction cache in the background while spamming continues ([#395](https://github.com/flashbots/contender/issues/395))
+  - Spamming no longer pauses to collect receipts
+  - Cache is automatically flushed at configurable intervals (default: every 5 blocks)
+  - More realistic RPC traffic patterns
+  - Better performance for long-running spam operations
+  - Improved error handling with automatic retry on transient failures
+  - Intelligent logging to avoid spam during extended RPC issues
+
+---
+
 ## [0.6.0](https://github.com/flashbots/contender/releases/tag/v0.6.0) - 2025-11-25
 
 Features:

crates/cli/src/commands/campaign.rs

@@ -308,7 +308,8 @@ fn create_spam_cli_args(
             },
             duration: spam_duration,
             pending_timeout: args.pending_timeout,
-            loops: Some(Some(1)),
+            indefinite: false,
+            cache_flush_interval: 5, // Use default for campaigns
             accounts_per_agent: args.accounts_per_agent,
         },
         ignore_receipts: args.ignore_receipts,

crates/cli/src/commands/common.rs

@@ -294,13 +294,18 @@ 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."
+        long_help = "Run spam indefinitely until manually stopped. If not set, the spam run will be executed once."
     )]
-    pub loops: Option<Option<u64>>,
-
+    pub indefinite: bool,
+    /// How often (in blocks) to flush the pending transaction cache to the database.
+    /// Lower values reduce memory usage but increase DB writes. Higher values batch more efficiently.
+    #[arg(
+        long,
+        default_value_t = 5,
+        long_help = "Number of blocks between automatic cache flushes. Controls memory usage vs DB write frequency."
+    )]
+    pub cache_flush_interval: u64,
     /// The number of accounts to generate for each agent (`from_pool` in scenario files)
     #[arg(
         short,

crates/cli/src/commands/spam.rs

@@ -205,7 +205,8 @@ impl SpamCommandArgs {
             txs_per_block,
             duration,
             pending_timeout,
-            loops,
+            indefinite,
+            cache_flush_interval,
             accounts_per_agent,
         } = self.spam_args.spam_args.clone();
         let SendTxsCliArgsInner {
@@ -478,7 +479,7 @@ impl SpamCommandArgs {
         }
         done_fcu.store(true, std::sync::atomic::Ordering::SeqCst);
 
-        if loops.is_some_and(|inner_loops| inner_loops.is_none()) {
+        if indefinite {
             warn!("Spammer agents will eventually run out of funds.");
             println!(
                 "Make sure you add plenty of funds with {} (set your pre-funded account with {}).",
@@ -487,7 +488,7 @@ impl SpamCommandArgs {
             );
         }
 
-        let total_cost = U256::from(duration * loops.flatten().unwrap_or(1))
+        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 {
@@ -497,6 +498,9 @@ impl SpamCommandArgs {
             .into());
         }
 
+        // Set the cache flush interval from CLI args
+        test_scenario.cache_flush_interval_blocks = cache_flush_interval;
+
         Ok(test_scenario)
     }
 

crates/cli/src/commands/spamd.rs

@@ -16,7 +16,7 @@ use tracing::{error, info, warn};
 
 /// Runs spam in a loop, potentially executing multiple spam runs.
 ///
-/// If `limit_loops` is `None`, it will run indefinitely.
+/// If `limit_loops` is `None`, it will run indefinitely (--indefinite flag).
 ///
 /// If `limit_loops` is `Some(n)`, it will run `n` times.
 ///

crates/cli/src/main.rs

@@ -94,7 +94,7 @@ async fn run() -> Result<(), CliError> {
                 ..
             } = *args.to_owned();
 
-            let SendSpamCliArgs { loops, .. } = spam_args.to_owned();
+            let SendSpamCliArgs { indefinite, .. } = spam_args.to_owned();
 
             let client = ClientBuilder::default()
                 .http(Url::from_str(&rpc_args.rpc_url).map_err(ArgsError::UrlParse)?);
@@ -119,15 +119,15 @@ async fn run() -> Result<(), CliError> {
                 )
             };
 
-            let real_loops = if let Some(loops) = loops {
-                // loops flag is set; spamd will interpret a None value as infinite
-                loops
+            let limit_loops = if indefinite {
+                // indefinite flag is set; spamd will interpret None as infinite
+                None
             } else {
-                // loops flag is not set, so only loop once
+                // indefinite flag is not set, so only loop once
                 Some(1)
             };
             let spamd_args = SpamCommandArgs::new(scenario, *args)?;
-            commands::spamd(&db, spamd_args, gen_report, real_loops).await?;
+            commands::spamd(&db, spamd_args, gen_report, limit_loops).await?;
         }
 
         ContenderSubcommand::Replay { args } => {

crates/core/src/spammer/spammer_trait.rs

@@ -117,6 +117,22 @@ where
                 scenario.sync_nonces().await?;
             }
 
+            // ═══════════════════════════════════════════════════════════════
+            // START AUTO-FLUSH: Begin background receipt processing
+            // ═══════════════════════════════════════════════════════════════
+            // From this point forward, pending transactions will be automatically
+            // flushed to the database every N blocks (configurable via --cache-flush-interval).
+            // This allows spamming to continue uninterrupted while receipts are collected.
+            // 
+            // The auto-flush process:
+            // 1. TxActor monitors block numbers via periodic checks
+            // 2. When interval blocks have passed, it fetches receipts for those blocks
+            // 3. Confirmed transactions are saved to DB, unconfirmed remain in cache
+            // 4. Process repeats automatically until auto-flush is stopped
+            for msg_handle in scenario.msg_handles.values() {
+                msg_handle.start_auto_flush(run_id, scenario.cache_flush_interval_blocks).await?;
+            }
+
             // calling cancel() on cancel_token should stop all running tasks
             // (as long as each task checks for it)
             let cancel_token = self.context().do_quit.clone();
@@ -143,6 +159,22 @@ where
                 .done_sending
                 .store(true, std::sync::atomic::Ordering::SeqCst);
 
+            // ═══════════════════════════════════════════════════════════════
+            // STOP AUTO-FLUSH: Transition to final manual flush
+            // ═══════════════════════════════════════════════════════════════
+            // Now that spamming is complete, we stop the background auto-flush
+            // and do a final comprehensive manual flush to ensure all remaining
+            // transactions are processed.
+            //
+            // Why stop auto-flush before final flush:
+            // 1. Prevents interference between auto-flush and manual flush
+            // 2. Manual flush has more sophisticated timeout/stall detection
+            // 3. Ensures we wait for all blocks to be available
+            // 4. Cleaner separation of concerns
+            for msg_handle in scenario.msg_handles.values() {
+                let _ = msg_handle.stop_auto_flush().await;
+            }
+
             // collect results from cached pending txs
             let flush_finished: bool = tokio::select! {
                 _ = tokio::signal::ctrl_c() => {