Fix proxied changes dropped when an optimization fails (#10364)

* Add SetFlushInterval op to the model tester

Changes the collection's flush_interval_sec mid-run through the same path
update_collection takes (persist the optimizer-config diff, then recreate
the optimizers in the background). The model is untouched: what it perturbs
is the flush cadence, so how much of the workload is still WAL-only when a
restart hits, plus the worker stop/start race in on_optimizer_config_update.

Kept in FORCE_OFF for now: with the optimizer on it makes stale point state
visible within a few ops of the config change. Narrowed to
recreate_optimizers_background, see the comment on Swarm::BASE.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_011AjmS5GFeGztfP3JqutnXj

* Keep SetFlushInterval enabled in the swarm

Drops it from FORCE_OFF so the divergence it surfaces is reachable without
--enable-force-off (which would also enable the broken vector-name ops).
The evidence moves from the FORCE_OFF comment onto the op's own doc.

The two optimizer-on harness gates now fail whenever the swarm draws the op.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_011AjmS5GFeGztfP3JqutnXj

* Propagate proxied changes when unwrapping proxies on optimization failure

unwrap_proxy puts the wrapped segments back into the segment holder, so the
changes recorded on the proxy while the optimization ran (deleted points,
index and vector-name changes) have to reach the wrapped segment first. They
did not, so every point deleted or overwritten during the optimization kept
its pre-optimization copy live next to the new copy in the write segment, and
reads saw both: counts too high, scroll and search returning the stale copy.

The snapshot unproxy path already does this; the optimizer failure path was
the only place putting a wrapped segment back without it. It is reachable
whenever the shard outlives the cancellation, in particular an update_collection
that recreates the optimizers while an optimization is in flight.

Lock order is holder-then-updates, matching try_unproxy_segment: updates-then-
holder-write deadlocks against the snapshot path.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_011AjmS5GFeGztfP3JqutnXj

* Drop the stale failure note from the SetFlushInterval doc

The divergence it described is fixed in this branch.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_011AjmS5GFeGztfP3JqutnXj

* test as well with 0s as flushing interval

* Fail optimization unwrapping when proxy propagation fails

Losing proxied deletes and index changes is data corruption, so return the
error instead of logging it: no proxy is unwrapped and the changes stay
served by the proxies. The cancelled-segment cleanup moves ahead of
unwrap_proxy so the orphan is still removed when that error fires.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CpoaAtGbAQAEuxEi8ScHc5

* Drop the model tester --flush-interval-sec flag

SetFlushInterval covers the interval now, so the run starts at the shipped
5s default (fixture::INITIAL_FLUSH_INTERVAL_SEC, still traced in the header)
and the ops move it from there. Also documents what 0 does now that it is a
generated value.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CpoaAtGbAQAEuxEi8ScHc5

* Fail snapshot unproxying when proxy propagation fails

Both paths logged the error and unwrapped anyway, dropping the deletes and
index changes that never reached the wrapped segment. Same reasoning as
unwrap_proxy in the optimizer.

try_unproxy_segment hands the lock back and leaves the proxy installed, the
failure mode its doc already describes: the caller keeps it in `proxies` and
unproxy_all_segments retries the propagation right after. unproxy_all_segments
returns before touching the holder, so the temp segment the surviving proxies
write into stays in place (remove_segment_if_not_needed only checks whether it
is empty and appendable, not whether a proxy still references it).

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CpoaAtGbAQAEuxEi8ScHc5

---------

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

lib/collection/src/model_testing/apply/mod.rs

@@ -210,6 +210,7 @@ pub(super) async fn apply(
         Op::CreateSnapshot => {
             unreachable!("CreateSnapshot is handled in the run loop, not apply()")
         }
+        Op::SetFlushInterval(sec) => writes::apply_set_flush_interval(collection, *sec).await,
     }
 }
 

lib/collection/src/model_testing/apply/writes.rs

@@ -13,6 +13,7 @@ use super::super::op::{
 use super::{apply_update, to_named_persisted};
 use crate::collection::Collection;
 use crate::operations::CollectionUpdateOperations;
+use crate::operations::config_diff::OptimizersConfigDiff;
 use crate::operations::payload_ops::{DeletePayloadOp, PayloadOps, SetPayloadOp};
 use crate::operations::point_ops::{
     ConditionalInsertOperationInternal, PointIdsList, PointInsertOperationsInternal,
@@ -305,6 +306,35 @@ pub(super) async fn apply_drop_index(collection: &Collection, field: &JsonPath)
         .expect("drop index failed");
 }
 
+/// Change `flush_interval_sec` on the live collection, mirroring the production
+/// `update_collection` path (`toc::collection_meta_ops`): persist the optimizer-config diff,
+/// then recreate the optimizers so the per-shard update/flush workers pick the new cadence up.
+///
+/// The recreation runs in the background (single-flight, coalescing) and we deliberately don't
+/// wait for it: that's what a real config update does, and it leaves the workers restarting
+/// while the run loop keeps writing. The model isn't touched - flush cadence is invisible to
+/// the collection's observable contents.
+pub(super) async fn apply_set_flush_interval(collection: &Collection, flush_interval_sec: u64) {
+    collection
+        .update_optimizer_params_from_diff(OptimizersConfigDiff {
+            flush_interval_sec: Some(flush_interval_sec),
+            // Every other field stays `None` so the diff touches the flush cadence only
+            // (`OptimizersConfig::update` keeps the current value for each `None`).
+            deleted_threshold: None,
+            vacuum_min_vector_number: None,
+            default_segment_number: None,
+            max_segment_size: None,
+            #[expect(deprecated)]
+            memmap_threshold: None,
+            indexing_threshold: None,
+            max_optimization_threads: None,
+            prevent_unoptimized: None,
+        })
+        .await
+        .expect("update flush_interval_sec failed");
+    collection.recreate_optimizers_background();
+}
+
 pub(super) async fn apply_upsert_conditional(
     collection: &Collection,
     model: &mut Model,

lib/collection/src/model_testing/fixture.rs

@@ -97,6 +97,10 @@ fn quantization_config(kind: QuantizationKind) -> QuantizationConfig {
     }
 }
 
+/// Periodic flush worker interval the collection starts with, in seconds. Matches the shipped
+/// default (see `config/config.yaml`); `Op::SetFlushInterval` moves it around during the run.
+pub(super) const INITIAL_FLUSH_INTERVAL_SEC: u64 = 5;
+
 /// Build a fresh soak collection rooted at `storage_path`. Creates `collection/` and
 /// `snapshots/` subdirs, wiping any pre-existing contents — each soak run starts from a
 /// clean slate. If you need to preserve a crashed run's state for post-mortem, copy the
@@ -107,7 +111,6 @@ pub(super) async fn fixture(
     disable_optimizer: bool,
     max_segment_size_kb: usize,
     indexing_threshold_kb: usize,
-    flush_interval_sec: u64,
     on_disk: bool,
 ) -> (PathBuf, PathBuf, Collection) {
     let collection_dir = storage_path.join("collection");
@@ -190,7 +193,7 @@ pub(super) async fn fixture(
         #[expect(deprecated)]
         memmap_threshold: None,
         indexing_threshold: Some(indexing_threshold_kb),
-        flush_interval_sec,
+        flush_interval_sec: INITIAL_FLUSH_INTERVAL_SEC,
         // `Some(0)` disables the optimizer worker entirely (per `OptimizersConfig::fixture`'s
         // comment); `None` lets it use its default thread count.
         max_optimization_threads: if disable_optimizer { Some(0) } else { Some(1) },

lib/collection/src/model_testing/mod.rs

@@ -516,7 +516,6 @@ pub async fn run(
     disable_optimizer: bool,
     max_segment_size_kb: usize,
     indexing_threshold_kb: usize,
-    flush_interval_sec: u64,
     restart_probability: f64,
     swarm_interval: usize,
     on_disk: bool,
@@ -534,7 +533,6 @@ pub async fn run(
         disable_optimizer,
         max_segment_size_kb,
         indexing_threshold_kb,
-        flush_interval_sec,
         on_disk,
     )
     .await;
@@ -557,7 +555,7 @@ pub async fn run(
         disable_optimizer,
         max_segment_size_kb,
         indexing_threshold_kb,
-        flush_interval_sec,
+        fixture::INITIAL_FLUSH_INTERVAL_SEC,
         restart_probability,
         swarm_interval,
         enable_force_off,
@@ -1019,7 +1017,6 @@ mod tests {
             disable_optimizer,
             10, // max_segment_size_kb
             5,  // indexing_threshold_kb
-            5,  // flush_interval_sec
             restart_probability,
             2500,  // swarm_interval: a few redraws within the run
             false, // on_disk

lib/collection/src/model_testing/op/generators.rs

@@ -202,6 +202,16 @@ pub(super) fn random_slice(rng: &mut impl Rng) -> (NonZeroU32, u32) {
     (total, index)
 }
 
+/// Flush interval for `Op::SetFlushInterval`, in seconds. The range mirrors what operators
+/// realistically configure: the shipped default is 5s (see `config/config.yaml`), and tuned
+/// deployments trade durability for write throughput anywhere between 1s and 30s. `0` is
+/// included too: the flush worker then flushes back to back without sleeping in between,
+/// the most aggressive durability setting and the tightest flush-vs-write interleaving.
+pub(super) fn random_flush_interval_sec(rng: &mut impl Rng) -> u64 {
+    const INTERVALS: &[u64] = &[0, 1, 2, 5, 10, 30];
+    *INTERVALS.choose(rng).unwrap()
+}
+
 /// A filter selector for paginated scroll: no filter, `num == X`, `tag == X`, a `has_id`
 /// matcher, a `has_vector` matcher over a currently-active vector name, a `url` prefix
 /// matcher, a deterministic `slice` of the id space, or composed `num` ∧ `slice`.

lib/collection/src/model_testing/op/mod.rs

@@ -7,12 +7,12 @@ use std::num::NonZeroU32;
 use ahash::AHashSet;
 use api::rest::RecommendStrategy;
 use generators::{
-    random_direction, random_distinct_ids, random_distinct_points, random_existing_ids, random_num,
-    random_partial_named_vectors, random_payload, random_payload_key, random_payload_keys,
-    random_point, random_prefetch, random_query_for_name, random_recommend_strategy,
-    random_scroll_filter, random_slice, random_tag, random_update_mode, random_url_prefix_probe,
-    random_vector_name, random_vector_name_subset, random_with_payload, random_with_vector,
-    upsert_fallback,
+    random_direction, random_distinct_ids, random_distinct_points, random_existing_ids,
+    random_flush_interval_sec, random_num, random_partial_named_vectors, random_payload,
+    random_payload_key, random_payload_keys, random_point, random_prefetch, random_query_for_name,
+    random_recommend_strategy, random_scroll_filter, random_slice, random_tag, random_update_mode,
+    random_url_prefix_probe, random_vector_name, random_vector_name_subset, random_with_payload,
+    random_with_vector, upsert_fallback,
 };
 use rand::distr::weighted::WeightedIndex;
 use rand::prelude::Distribution;
@@ -243,6 +243,15 @@ pub(super) enum Op {
     /// verification ops + final reload catch that). Handled in the run loop (not `apply`) because it
     /// spawns a task against shared collection state.
     CreateSnapshot,
+    /// Change the collection's `flush_interval_sec` mid-run, through the same path a user's
+    /// `update_collection` takes: persist the optimizer-config diff, then recreate the
+    /// optimizers (and with them the per-shard update/flush workers) in the background.
+    ///
+    /// Not a data op: the model is untouched and every verification op must keep passing. What it
+    /// perturbs is the *flush cadence*, which decides how much of the workload is still WAL-only
+    /// when a restart hits, plus the worker stop/start race in `on_optimizer_config_update`.
+    /// The new value is persisted to `config.json`, so it survives the run's restarts.
+    SetFlushInterval(u64),
 }
 
 /// A single prefetch source for `QueryFusion`: a Nearest sub-query over one vector name, with its
@@ -312,7 +321,7 @@ pub(super) struct Swarm {
 }
 
 impl Swarm {
-    const N: usize = 39;
+    const N: usize = 40;
 
     /// Op names, aligned 1:1 with `BASE` and the `match` arms in `Op::random`.
     const NAMES: [&'static str; Self::N] = [
@@ -355,6 +364,7 @@ impl Swarm {
         "ScrollFilteredByUrlPrefix",
         "CreateSnapshot",
         "CountBySlice",
+        "SetFlushInterval",
     ];
 
     /// Each op's *natural* relative weight — the default distribution before swarm masking.
@@ -410,6 +420,9 @@ impl Swarm {
         // blocking thread, so keep the weight modest.
         2, // CreateSnapshot
         4, // CountBySlice
+        // Config change: cheap in itself, but it restarts every shard's update workers, so keep
+        // it rare enough that the workload isn't dominated by worker churn.
+        1, // SetFlushInterval
     ];
 
     /// Indices kept enabled in every swarm config: without a way to insert points the run can't
@@ -771,6 +784,7 @@ impl Op {
                 let (total, index) = random_slice(rng);
                 Op::CountBySlice { total, index }
             }
+            39 => Op::SetFlushInterval(random_flush_interval_sec(rng)),
             n => panic!("unexpected op index {n}"),
         }
     }
@@ -818,6 +832,7 @@ impl Op {
             Op::ScrollPaged { .. } => "ScrollPaged",
             Op::QueryFusion { .. } => "QueryFusion",
             Op::CreateSnapshot => "CreateSnapshot",
+            Op::SetFlushInterval(_) => "SetFlushInterval",
         }
     }
 }

lib/collection/src/model_testing/trace.rs

@@ -32,6 +32,10 @@
 //! with the op events that follow it (the archive is discarded — no recovery). It draws no rng, so
 //! it doesn't perturb the op stream.
 //!
+//! A `SetFlushInterval` op line changes the flush cadence from that point on, so the `Header`'s
+//! `flush_interval_sec` is the run's *initial* value, not necessarily the one in effect at the
+//! failing op.
+//!
 //! `extra` = ids the engine has that the model doesn't.
 //! `missing` = ids the model has that the engine doesn't.
 //! When the run is clean both arrays are empty.
@@ -363,6 +367,9 @@ fn op_payload(op: &Op) -> Value {
         }),
         // The op events that follow a `CreateSnapshot` ran concurrently with the background capture.
         Op::CreateSnapshot => json!({}),
+        // The flush cadence in effect for the op events that follow, until the next
+        // `SetFlushInterval` (the value is persisted, so restarts don't reset it).
+        Op::SetFlushInterval(sec) => json!({ "flush_interval_sec": sec }),
     }
 }
 

lib/shard/src/optimize.rs

@@ -87,6 +87,10 @@ pub trait OptimizationStrategy: Send {
 
 /// Restores original segments from proxies
 ///
+/// Proxied changes (deleted points, index and vector-name changes) are always propagated into the
+/// wrapped segments first, so they are not lost when the proxies are dropped. If that propagation
+/// fails, no proxy is unwrapped and the error is returned.
+///
 /// # Arguments
 ///
 /// * `segments` - segment holder
@@ -94,12 +98,36 @@ pub trait OptimizationStrategy: Send {
 ///
 /// # Result
 ///
-/// Original segments are pushed into `segments`, proxies removed.
+/// Original segments are pushed into `segments`, proxies removed. On a propagation error the
+/// proxies are left in the holder untouched.
 pub fn unwrap_proxy(
     segments: &LockedSegmentHolder,
     proxy_ids: &[SegmentId],
 ) -> OperationResult<()> {
-    let mut segments_lock = segments.write();
+    // Propagate proxied changes back into wrapped segment to not lose these in-memory changes
+    let segments_lock = segments.upgradable_read();
+    let _update_guard = segments.acquire_updates_lock();
+
+    let proxies: Vec<_> = proxy_ids
+        .iter()
+        .filter_map(|&proxy_id| match segments_lock.get(proxy_id).cloned() {
+            Some(LockedSegment::Proxy(proxy_segment)) => Some((proxy_id, proxy_segment)),
+            _ => None,
+        })
+        .collect();
+    for (proxy_id, proxy_segment) in &proxies {
+        // Unwrapping a proxy whose changes did not reach the wrapped segment loses those deletes
+        // and index changes for good, so bail out instead. Every proxy stays installed and keeps
+        // serving its changes; nothing is unwrapped, so nothing is lost.
+        if let Err(err) = proxy_segment.write().propagate_to_wrapped() {
+            log::error!(
+                "Propagating proxy segment {proxy_id} changes to wrapped segment failed: {err}",
+            );
+            return Err(err);
+        }
+    }
+
+    let mut segments_lock = RwLockUpgradableReadGuard::upgrade(segments_lock);
     for &proxy_id in proxy_ids {
         if let Some(proxy_segment_ref) = segments_lock.get(proxy_id) {
             let locked_proxy_segment = proxy_segment_ref.clone();
@@ -903,16 +931,17 @@ pub fn execute_optimization<F: ?Sized + OptimizationStrategy>(
     let (optimized_segment, already_remove_points) = match build_result {
         Ok(result) => result,
         Err(err) => {
-            // Properly cancel optimization on all error kinds
-            // Unwrap proxies and add temp segment to holder
-            unwrap_proxy(&segment_holder, &proxy_ids)?;
             // A graceful cancellation always happens before the optimized segment is swapped into
             // the holder, so the segment `build` already moved into `segments_path` is now an
-            // orphan that `Drop` won't remove. Delete it explicitly. Non-cancellation errors may
-            // occur after the swap, where the segment is live, so they are left untouched.
+            // orphan that `Drop` won't remove. Delete it explicitly, before unwrapping the proxies
+            // so a failure there cannot leave it behind. Non-cancellation errors may occur after
+            // the swap, where the segment is live, so they are left untouched.
             if matches!(err, OperationError::Cancelled { .. }) {
                 cleanup_cancelled_optimized_segment(&paths.segments_path, output_segment_uuid);
             }
+            // Properly cancel optimization on all error kinds
+            // Unwrap proxies and add temp segment to holder
+            unwrap_proxy(&segment_holder, &proxy_ids)?;
             return Err(err);
         }
     };
@@ -930,16 +959,17 @@ pub fn execute_optimization<F: ?Sized + OptimizationStrategy>(
     ) {
         Ok(points_count) => points_count,
         Err(err) => {
-            // Properly cancel optimization on all error kinds
-            // Unwrap proxies and add temp segment to holder
-            unwrap_proxy(&segment_holder, &proxy_ids)?;
             // A graceful cancellation always happens before the optimized segment is swapped into
             // the holder, so the segment `build` already moved into `segments_path` is now an
-            // orphan that `Drop` won't remove. Delete it explicitly. Non-cancellation errors may
-            // occur after the swap, where the segment is live, so they are left untouched.
+            // orphan that `Drop` won't remove. Delete it explicitly, before unwrapping the proxies
+            // so a failure there cannot leave it behind. Non-cancellation errors may occur after
+            // the swap, where the segment is live, so they are left untouched.
             if matches!(err, OperationError::Cancelled { .. }) {
                 cleanup_cancelled_optimized_segment(&paths.segments_path, output_segment_uuid);
             }
+            // Properly cancel optimization on all error kinds
+            // Unwrap proxies and add temp segment to holder
+            unwrap_proxy(&segment_holder, &proxy_ids)?;
             return Err(err);
         }
     };
@@ -950,3 +980,57 @@ pub fn execute_optimization<F: ?Sized + OptimizationStrategy>(
 
     Ok(OptimizationResult { points_count })
 }
+
+#[cfg(test)]
+mod tests {
+    use common::counter::hardware_counter::HardwareCounterCell;
+    use common::types::DeferredBehavior;
+    use tempfile::Builder;
+
+    use super::*;
+    use crate::fixtures::build_segment_1;
+    use crate::proxy_segment::ProxySegment;
+    use crate::segment_holder::SegmentHolder;
+
+    /// A cancelled optimization puts the wrapped segments back into the holder, so the deletions
+    /// recorded on the proxy while the optimization ran must reach the wrapped segment first.
+    /// Without that, the point's pre-optimization copy stays live next to whatever the write
+    /// segment holds for it, and reads see both.
+    #[test]
+    fn unwrap_proxy_propagates_deletes_to_wrapped_segment() {
+        let dir = Builder::new().prefix("segment_dir").tempdir().unwrap();
+        let hw_counter = HardwareCounterCell::new();
+
+        let wrapped = LockedSegment::new(build_segment_1(dir.path()));
+        let mut holder = SegmentHolder::default();
+        let segment_id = holder.add_new_locked(wrapped.clone());
+
+        // Wrap it the way an optimization does, then delete a point through the proxy: the
+        // deletion is recorded on the proxy, the wrapped segment still has the point.
+        let mut proxy = ProxySegment::new(wrapped.clone());
+        proxy.delete_point(100, 1.into(), &hw_counter).unwrap();
+        let holder = LockedSegmentHolder::new(holder);
+        holder
+            .write()
+            .replace(segment_id, LockedSegment::from(proxy))
+            .unwrap();
+        assert!(
+            wrapped
+                .get()
+                .read()
+                .has_point(1.into(), DeferredBehavior::WithDeferred),
+            "wrapped segment should still hold the point while proxied",
+        );
+
+        unwrap_proxy(&holder, &[segment_id]).unwrap();
+
+        assert!(
+            !wrapped
+                .get()
+                .read()
+                .has_point(1.into(), DeferredBehavior::WithDeferred),
+            "deletion recorded on the proxy must reach the wrapped segment before it goes back \
+             into the holder",
+        );
+    }
+}

lib/shard/src/segment_holder/snapshot.rs

@@ -149,13 +149,14 @@ impl SegmentHolder {
             }
         };
 
-        // propagate changes to wrapped segment with segment holder read lock
-        {
-            if let Err(err) = proxy_segment.write().propagate_to_wrapped() {
-                log::error!(
-                    "Propagating proxy segment {segment_id} changes to wrapped segment failed, ignoring: {err}",
-                );
-            }
+        // Propagate changes to wrapped segment with segment holder read lock. On failure keep the
+        // proxy installed rather than unwrapping into a segment that never got the changes;
+        // `unproxy_all_segments` retries the propagation for every proxy left behind.
+        if let Err(err) = proxy_segment.write().propagate_to_wrapped() {
+            log::error!(
+                "Propagating proxy segment {segment_id} changes to wrapped segment failed: {err}",
+            );
+            return Err(segments_lock);
         }
 
         let mut write_segments = RwLockUpgradableReadGuard::upgrade(segments_lock);
@@ -181,16 +182,21 @@ impl SegmentHolder {
         // so it is important, that we don't block reads while doing this.
 
         // propagate changes to wrapped segment with segment holder read lock
-        proxies
-            .iter()
-            .filter_map(|(segment_id, proxy_segment)| match proxy_segment {
-                LockedSegment::Proxy(proxy_segment) => Some((segment_id, proxy_segment)),
-                LockedSegment::Original(_) => None,
-            }).for_each(|(proxy_id, proxy_segment)| {
-            if let Err(err) = proxy_segment.write().propagate_to_wrapped() {
-                log::error!("Propagating proxy segment {proxy_id} changes to wrapped segment failed, ignoring: {err}");
-            }
+        let proxies_to_propagate = proxies.iter().filter_map(|(id, segment)| match segment {
+            LockedSegment::Proxy(proxy_segment) => Some((id, proxy_segment)),
+            LockedSegment::Original(_) => None,
         });
+        for (proxy_id, proxy_segment) in proxies_to_propagate {
+            // Unwrapping a proxy whose changes did not reach the wrapped segment loses them for
+            // good, so bail out before touching the holder. Every proxy stays installed and keeps
+            // serving its changes, and the temp segment they write into is left in place.
+            if let Err(err) = proxy_segment.write().propagate_to_wrapped() {
+                log::error!(
+                    "Propagating proxy segment {proxy_id} changes to wrapped segment failed: {err}",
+                );
+                return Err(err);
+            }
+        }
 
         // Swap out each proxy with wrapped segment once changes are propagated
         let mut write_segments = RwLockUpgradableReadGuard::upgrade(segments_lock);

src/model_testing.rs

@@ -86,13 +86,6 @@ struct Args {
     #[clap(long, default_value_t = 5, value_parser = clap::value_parser!(u64).range(1..))]
     indexing_threshold_kb: u64,
 
-    /// Periodic flush worker interval in seconds. Lower values mean smaller unflushed-WAL
-    /// windows (more production-realistic given the soak's high op rate); larger values
-    /// stress the unflushed-WAL path and surface bugs like the `DeleteVectorName` replay
-    /// issue where historical Upserts referencing a since-deleted vector name fail.
-    #[clap(long, default_value_t = 5, value_parser = clap::value_parser!(u64).range(1..))]
-    flush_interval_sec: u64,
-
     /// Per-iteration probability (0.0..=1.0) of restarting the collection mid-run:
     /// close + reopen + full model verification. Surfaces reload/WAL-replay bugs at the
     /// op where they're introduced rather than only at end-of-run. 0.0 (default) skips
@@ -228,7 +221,7 @@ async fn run_main(args: Args) {
     println!(
         "model_testing: seed={} {stop} shard_count={} id_pool={} uuid_id_fraction={} \
          storage_path={} disable_optimizer={} max_segment_size_kb={} indexing_threshold_kb={} \
-         flush_interval_sec={} restart_probability={} swarm_interval={} \
+         restart_probability={} swarm_interval={} \
          on_disk={} async_scorer={} pre_restart_check={} enable_force_off={} \
          disable_snapshots={}",
         args.seed,
@@ -239,7 +232,6 @@ async fn run_main(args: Args) {
         args.disable_optimizer,
         args.max_segment_size_kb,
         args.indexing_threshold_kb,
-        args.flush_interval_sec,
         args.restart_probability,
         args.swarm_interval,
         args.on_disk,
@@ -259,7 +251,6 @@ async fn run_main(args: Args) {
         args.disable_optimizer,
         args.max_segment_size_kb as usize,
         args.indexing_threshold_kb as usize,
-        args.flush_interval_sec,
         args.restart_probability,
         args.swarm_interval as usize,
         args.on_disk,