refactor: --catchup-delay arg for bench (#203)

Author: shikhar

Cargo.lock

@@ -14,40 +14,40 @@ path = "src/main.rs"
 
 [dependencies]
 async-stream = "0.3.6"
-base64ct = { version = "1.7.3", features = ["alloc"] }
-bytes = "1.10.1"
-clap = { version = "4.5.38", features = ["derive"] }
+base64ct = { version = "1.8.3", features = ["alloc"] }
+bytes = "1.11.0"
+clap = { version = "4.5.54", features = ["derive"] }
 color-print = "0.3.7"
-colored = "3.0.0"
-config = "0.15.11"
+colored = "3.1.1"
+config = "0.15.19"
 dirs = "6.0.0"
 futures = "0.3.31"
-http = "1.3.1"
-humantime = "2.2.0"
+http = "1.4.0"
+humantime = "2.3.0"
 indicatif = "0.18.3"
 json_to_table = "0.12.0"
 miette = { version = "7.6.0", features = ["fancy"] }
-rand = "0.9.1"
+rand = "0.9.2"
 s2-sdk = { version = "0.22.4", features = ["_hidden"] }
-serde = { version = "1.0.219", features = ["derive"] }
-serde_json = { version = "1.0.140", features = ["preserve_order"] }
+serde = { version = "1.0.228", features = ["derive"] }
+serde_json = { version = "1.0.149", features = ["preserve_order"] }
 strum = { version = "0.27", features = ["derive"] }
 tabled = "0.20.0"
-thiserror = "2.0.12"
-tokio = { version = "1.45.1", features = ["full"] }
-tokio-stream = { version = "0.1.17", features = ["io-util"] }
+thiserror = "2.0.18"
+tokio = { version = "1.49.0", features = ["full"] }
+tokio-stream = { version = "0.1.18", features = ["io-util"] }
 toml = "0.9.11"
-tracing = "0.1.41"
-tracing-subscriber = { version = "0.3.20", features = ["env-filter"] }
+tracing = "0.1.44"
+tracing-subscriber = { version = "0.3.22", features = ["env-filter"] }
 uuid = { version = "1.19.0", features = ["v4"] }
 xxhash-rust = { version = "0.8.15", features = ["xxh3"] }
 
 [dev-dependencies]
-assert_cmd = "2.0"
+assert_cmd = "2.1"
 predicates = "3.1"
-rstest = "0.26.0"
-serial_test = "3.2"
-tempfile = "3.15"
+rstest = "0.26.1"
+serial_test = "3.3"
+tempfile = "3.24"
 
 [profile.release]
 lto = true

Cargo.toml

@@ -424,7 +424,7 @@ pub async fn run(
     record_size: usize,
     target_mibps: NonZeroU64,
     duration: Duration,
-    catchup: bool,
+    catchup_delay: Duration,
 ) -> Result<(), CliError> {
     assert!(record_size <= RECORD_BATCH_MAX.bytes);
 
@@ -634,66 +634,61 @@ pub async fn run(
         )));
     }
 
-    if catchup {
-        eprintln!();
-        eprintln!("Waiting 20s before catchup read...");
-        tokio::time::sleep(Duration::from_secs(20)).await;
-
-        let catchup_bar = ProgressBar::no_length()
-            .with_prefix(format!("{:width$}", "catchup", width = prefix_width))
-            .with_style(
-                ProgressStyle::default_bar()
-                    .template("{prefix:.bold.cyan} {msg}")
-                    .expect("valid template"),
-            );
-        let mut catchup_sample: Option<BenchReadSample> = None;
-        let mut catchup_run_hash: Option<u64> = None;
-        let catchup_stream = bench_read_catchup(stream.clone(), record_size, bench_start);
-        let mut catchup_stream = std::pin::pin!(catchup_stream);
-        while let Some(result) = catchup_stream.next().await {
-            match result {
-                Ok(sample) => {
-                    update_bench_bar(&catchup_bar, &sample);
-                    if let Some(hash) = sample.run_hash {
-                        catchup_run_hash = Some(hash);
-                    }
-                    catchup_sample = Some(sample);
-                }
-                Err(e) => {
-                    catchup_bar.finish_and_clear();
-                    return Err(e);
+    eprintln!();
+    eprintln!("Waiting {:?} before catchup read...", catchup_delay);
+    tokio::time::sleep(catchup_delay).await;
+
+    let catchup_bar = ProgressBar::no_length()
+        .with_prefix(format!("{:width$}", "catchup", width = prefix_width))
+        .with_style(
+            ProgressStyle::default_bar()
+                .template("{prefix:.bold.cyan} {msg}")
+                .expect("valid template"),
+        );
+    let mut catchup_sample: Option<BenchReadSample> = None;
+    let mut catchup_run_hash: Option<u64> = None;
+    let catchup_stream = bench_read_catchup(stream.clone(), record_size, bench_start);
+    let mut catchup_stream = std::pin::pin!(catchup_stream);
+    while let Some(result) = catchup_stream.next().await {
+        match result {
+            Ok(sample) => {
+                update_bench_bar(&catchup_bar, &sample);
+                if let Some(hash) = sample.run_hash {
+                    catchup_run_hash = Some(hash);
                 }
+                catchup_sample = Some(sample);
+            }
+            Err(e) => {
+                catchup_bar.finish_and_clear();
+                return Err(e);
             }
         }
+    }
 
-        catchup_bar.finish_and_clear();
-        if let Some(sample) = catchup_sample {
-            eprintln!(
-                "{}: {:.2} MiB/s, {:.0} records/s ({} bytes, {} records in {:.2}s)",
-                "Catchup".bold().cyan(),
-                sample.mib_per_sec(),
-                sample.records_per_sec(),
-                sample.bytes,
-                sample.records,
-                sample.elapsed.as_secs_f64()
-            );
-        } else {
-            eprintln!(
-                "{}: no records available for catchup read",
-                "Catchup".bold().cyan()
-            );
-        }
-
-        if let (Some(expected), Some(actual)) = (write_run_hash, catchup_run_hash)
-            && expected != actual
-        {
-            return Err(CliError::BenchVerification(format!(
-                "catchup read hash mismatch: expected {expected}, got {actual}"
-            )));
-        }
+    catchup_bar.finish_and_clear();
+    if let Some(sample) = catchup_sample {
+        eprintln!(
+            "{}: {:.2} MiB/s, {:.0} records/s ({} bytes, {} records in {:.2}s)",
+            "Catchup".bold().cyan(),
+            sample.mib_per_sec(),
+            sample.records_per_sec(),
+            sample.bytes,
+            sample.records,
+            sample.elapsed.as_secs_f64()
+        );
     } else {
-        eprintln!();
-        eprintln!("Skipping catchup read.");
+        eprintln!(
+            "{}: no records available for catchup read",
+            "Catchup".bold().cyan()
+        );
+    }
+
+    if let (Some(expected), Some(actual)) = (write_run_hash, catchup_run_hash)
+        && expected != actual
+    {
+        return Err(CliError::BenchVerification(format!(
+            "catchup read hash mismatch: expected {expected}, got {actual}"
+        )));
     }
 
     Ok(())

src/bench.rs

@@ -531,9 +531,9 @@ pub struct BenchArgs {
     #[arg(short = 'd', long, default_value = "60s")]
     pub duration: humantime::Duration,
 
-    /// Skip the catchup read after the live run.
-    #[arg(long, default_value_t = false)]
-    pub no_catchup: bool,
+    /// Delay before starting the catchup read.
+    #[arg(short = 'w', long, default_value = "20s")]
+    pub catchup_delay: humantime::Duration,
 }
 
 /// Time range args for gauge metrics (no interval).

src/cli.rs

@@ -541,7 +541,7 @@ async fn run() -> Result<(), CliError> {
                 args.record_size as usize,
                 args.target_mibps,
                 *args.duration,
-                !args.no_catchup,
+                *args.catchup_delay,
             )
             .await;
             let delete_res = basin

src/main.rs

@@ -1044,7 +1044,8 @@ fn bench_stream() {
         "1s",
         "--target-mibps",
         "1",
-        "--no-catchup",
+        "--catchup-delay",
+        "0s",
     ])
     .assert()
     .success();