feat(model_testing): add --duration-sec to run for a fixed wall-clock time (#9506)

* feat(model_testing): add --duration-sec to run for a fixed wall-clock time

Bound the soak by wall-clock time instead of op count: when --duration-sec is set, the loop runs until the deadline (or Ctrl-C) and --op-num is ignored.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* Fix clippy

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

lib/collection/src/model_testing/mod.rs

@@ -8,6 +8,7 @@ use std::collections::{BTreeMap, BTreeSet};
 use std::path::Path;
 use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, Ordering};
+use std::time::{Duration, Instant};
 
 use indicatif::{ProgressBar, ProgressDrawTarget, ProgressStyle};
 use rand::rngs::SmallRng;
@@ -107,8 +108,12 @@ pub(super) enum VectorValue {
     MultiDense(Vec<Vec<f32>>),
 }
 
-/// Soak entrypoint — drives `op_num` randomized operations against a fresh collection,
-/// continuously verifying it against an in-memory model. Reproducible for a given `seed`.
+/// Soak entrypoint — drives randomized operations against a fresh collection, continuously
+/// verifying it against an in-memory model. Reproducible for a given `seed`.
+///
+/// The run stops at whichever bound applies: by default after `op_num` ops, or — when
+/// `duration` is `Some` — after that much wall-clock time elapses (`op_num` is then ignored
+/// as the stop condition). Either way an early Ctrl-C still ends it cleanly.
 ///
 /// Storage is rooted at `storage_path`. Any pre-existing `collection/` and `snapshots/`
 /// subdirectories are wiped at the start of each run; copy them out beforehand if you
@@ -133,6 +138,7 @@ pub async fn run(
     on_disk: bool,
     pre_restart_check: bool,
     enable_force_off: bool,
+    duration: Option<Duration>,
     shutdown: Arc<AtomicBool>,
 ) {
     let (collection_dir, snapshots_dir, mut collection) = fixture::fixture(
@@ -164,6 +170,7 @@ pub async fn run(
         restart_probability,
         swarm_interval,
         enable_force_off,
+        duration,
     );
 
     let mut model: Model = Model::new();
@@ -182,27 +189,67 @@ pub async fn run(
     trace.swarm(0, &initial_enabled);
     println!("model_testing: op:0 swarm -> {initial_enabled:?}");
 
-    let bar = ProgressBar::new(op_num as u64);
-    bar.set_style(
-        ProgressStyle::default_bar()
-            // Single line with a fixed-width bar: a multi-line (`{msg}\n…`) template plus a
-            // full-width `{wide_bar}` can't be repainted in place — the bar line hits the
-            // terminal-width auto-wrap boundary and occupies one more physical row than indicatif
-            // clears, so stale `{msg}` lines stick and smear together. A fixed-width `{bar}` never
-            // reaches that boundary; `{msg:24}` is width-padded so a changing op kind (longest is
-            // "OverwritePayloadByFilter", 24 chars) doesn't shift the bar left/right each frame.
-            .template("{msg:24} [{elapsed_precise}] {bar:40} {pos}/{len} ({per_sec}, eta:{eta})")
-            .expect("Failed to create progress style"),
-    );
+    // Two stop modes. By default the run is bounded by `op_num` and the bar fills toward that
+    // op count. With `--duration` the op count is unbounded, so the bar instead fills toward the
+    // wall-clock deadline (length = total seconds, position = elapsed seconds): `{pos}/{len}` and
+    // `eta` read in seconds and the bar still visibly advances to completion.
+    let loop_start = Instant::now();
+    let deadline = duration.map(|d| loop_start + d);
+    let bar = match duration {
+        // `max(1)` avoids a zero-length bar (which indicatif renders as already-complete) for a
+        // sub-second `--duration`.
+        Some(d) => {
+            let b = ProgressBar::new(d.as_secs().max(1));
+            b.set_style(
+                ProgressStyle::default_bar()
+                    .template("{msg:24} [{elapsed_precise}] {bar:40} {pos}s/{len}s (eta:{eta})")
+                    .expect("Failed to create progress style"),
+            );
+            b
+        }
+        None => {
+            let b = ProgressBar::new(op_num as u64);
+            b.set_style(
+                ProgressStyle::default_bar()
+                    // Single line with a fixed-width bar: a multi-line (`{msg}\n…`) template plus a
+                    // full-width `{wide_bar}` can't be repainted in place — the bar line hits the
+                    // terminal-width auto-wrap boundary and occupies one more physical row than
+                    // indicatif clears, so stale `{msg}` lines stick and smear together. A
+                    // fixed-width `{bar}` never reaches that boundary; `{msg:24}` is width-padded so
+                    // a changing op kind (longest is "OverwritePayloadByFilter", 24 chars) doesn't
+                    // shift the bar left/right each frame.
+                    .template(
+                        "{msg:24} [{elapsed_precise}] {bar:40} {pos}/{len} ({per_sec}, eta:{eta})",
+                    )
+                    .expect("Failed to create progress style"),
+            );
+            b
+        }
+    };
 
     let mut applied = 0usize;
-    for i in 0..op_num {
+    let mut i = 0usize;
+    loop {
         if shutdown.load(Ordering::Relaxed) {
             log::info!("shutdown received at op:{i}, exiting loop");
             break;
         }
+        // Stop condition: deadline reached in duration mode, op count reached otherwise.
+        match deadline {
+            Some(dl) => {
+                if Instant::now() >= dl {
+                    log::info!("duration elapsed at op:{i}, exiting loop");
+                    break;
+                }
+            }
+            None => {
+                if i >= op_num {
+                    break;
+                }
+            }
+        }
         // Recompute the swarm config at each interval boundary (op 0 was drawn before the loop).
-        if i > 0 && i % swarm_interval == 0 {
+        if i > 0 && i.is_multiple_of(swarm_interval) {
             let prev = swarm.enabled_ops();
             swarm = op::Swarm::random(rng, enable_force_off);
             let next = swarm.enabled_ops();
@@ -284,8 +331,18 @@ pub async fn run(
             trace.op(i, &op);
             apply::apply(&collection, &mut model, &mut active_names, &op).await;
         }
-        bar.inc(1);
+        i += 1;
         applied += 1;
+        // In duration mode the bar tracks elapsed wall-clock seconds (capped at the total so a
+        // final op finishing just past the deadline doesn't overshoot the bar); otherwise it
+        // tracks the op count.
+        match deadline {
+            Some(_) => {
+                let total = duration.map_or(0, |d| d.as_secs().max(1));
+                bar.set_position(loop_start.elapsed().as_secs().min(total));
+            }
+            None => bar.inc(1),
+        }
     }
     bar.finish();
 
@@ -381,6 +438,7 @@ mod tests {
             false, // on_disk
             false, // pre_restart_check
             false, // enable_force_off
+            None,  // duration: bounded by op_num
             Arc::new(AtomicBool::new(false)),
         )
         .await;
@@ -412,6 +470,7 @@ mod tests {
             false, // on_disk
             false, // pre_restart_check
             false, // enable_force_off
+            None,  // duration: bounded by op_num
             Arc::new(AtomicBool::new(false)),
         )
         .await;

lib/collection/src/model_testing/trace.rs

@@ -22,7 +22,7 @@
 //!
 //! | `kind`         | When                                            | Extra fields |
 //! |----------------|-------------------------------------------------|---|
-//! | `Header`       | First line — run configuration                  | `seed`, `op_num`, `shard_count`, `id_pool`, `disable_optimizer`, `max_segment_size_kb`, `indexing_threshold_kb`, `flush_interval_sec`, `restart_probability` |
+//! | `Header`       | First line — run configuration                  | `seed`, `op_num`, `shard_count`, `id_pool`, `disable_optimizer`, `max_segment_size_kb`, `indexing_threshold_kb`, `flush_interval_sec`, `restart_probability`, `swarm_interval`, `enable_force_off`, `duration_sec` (null unless `--duration`) |
 //! | *(op variant)* | Each `Op` from the workload generator           | See `op_payload` below — `id`/`ids`/etc. depending on variant |
 //! | `Restart`      | Mid-run close+reopen+verify                     | `pre_points`, `pre_segments` |
 //! | `LiveVerify`   | End-of-run scroll vs model, before reload       | `model_points`, `engine_points`, `segments`, `optimized_points`, `extra`, `missing` |
@@ -54,6 +54,7 @@
 
 use std::io::{BufWriter, Write};
 use std::path::Path;
+use std::time::Duration;
 
 use fs_err::File;
 use segment::types::PointIdType;
@@ -86,6 +87,7 @@ impl Trace {
         restart_probability: f64,
         swarm_interval: usize,
         enable_force_off: bool,
+        duration: Option<Duration>,
     ) {
         self.write(&json!({
             "kind": "Header",
@@ -100,6 +102,8 @@ impl Trace {
             "restart_probability": restart_probability,
             "swarm_interval": swarm_interval,
             "enable_force_off": enable_force_off,
+            // null in op-count mode; seconds when the run is time-bounded (`--duration`).
+            "duration_sec": duration.map(|d| d.as_secs()),
         }));
     }
 

src/model_testing.rs

@@ -1,7 +1,7 @@
 use std::path::PathBuf;
 use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, Ordering};
-use std::time::Instant;
+use std::time::{Duration, Instant};
 
 use clap::Parser;
 use collection::profiling::interface::init_requests_profile_collector;
@@ -30,10 +30,20 @@ struct Args {
     #[clap(long, default_value_t = 0)]
     seed: u64,
 
-    /// Number of randomized operations to apply.
+    /// Number of randomized operations to apply. Ignored as the stop condition when
+    /// `--duration` is set (the run is then bounded by wall-clock time instead).
     #[clap(long, default_value_t = 10_000, value_parser = clap::value_parser!(u64).range(1..))]
     op_num: u64,
 
+    /// Run continuously for this many wall-clock seconds instead of stopping after `--op-num` ops.
+    /// When set, `--op-num` is ignored as the stop condition and the run ends at the deadline (or
+    /// on Ctrl-C, whichever comes first). The post-run live verification + final close/reopen
+    /// reload check run as usual on whatever ops were applied. Use this for time-boxed soak runs
+    /// (e.g. an overnight or per-CI-slot budget) where the interesting variable is "how long"
+    /// rather than "how many ops".
+    #[clap(long, value_parser = clap::value_parser!(u64).range(1..))]
+    duration_sec: Option<u64>,
+
     /// Number of shards in the test collection.
     #[clap(long, default_value_t = 3, value_parser = clap::value_parser!(u32).range(1..))]
     shard_count: u32,
@@ -157,13 +167,17 @@ async fn main() {
     // Process-global flag, read when each on-disk dense vector storage is opened — set it before
     // the fixture builds any segments.
     segment::vector_storage::common::set_async_scorer(args.async_scorer);
+    // Show the active stop condition: the duration when time-bounded, else the op count.
+    let stop = match args.duration_sec {
+        Some(s) => format!("duration_sec={s}"),
+        None => format!("op_num={}", args.op_num),
+    };
     println!(
-        "model_testing: seed={} op_num={} shard_count={} id_pool={} storage_path={} \
+        "model_testing: seed={} {stop} shard_count={} id_pool={} storage_path={} \
          disable_optimizer={} max_segment_size_kb={} indexing_threshold_kb={} \
          flush_interval_sec={} restart_probability={} swarm_interval={} on_disk={} \
          async_scorer={} pre_restart_check={} enable_force_off={}",
         args.seed,
-        args.op_num,
         args.shard_count,
         args.id_pool,
         args.storage_path.display(),
@@ -194,6 +208,7 @@ async fn main() {
         args.on_disk,
         args.pre_restart_check,
         args.enable_force_off,
+        args.duration_sec.map(Duration::from_secs),
         shutdown,
     )
     .await;