fix(rollout): make WAL time trigger a true OR with count trigger

crates/lance-context-core/src/rollout_store.rs

@@ -148,13 +148,13 @@ pub struct RolloutStoreOptions {
     ///
     /// Spawn the task with [`RolloutStore::spawn_periodic_cleanup`] once the
     /// store is behind an `Arc<RwLock<_>>` (the server's ownership model).
+    ///
+    /// The two triggers are a strict **OR**: whichever fires first merges. The
+    /// time trigger is *not* gated by any generation count — once the interval
+    /// elapses, every pending generation is folded into the base table even if
+    /// the count trigger's threshold was never crossed. (A pass with nothing
+    /// pending is a no-op.)
     pub cleanup_interval_secs: Option<u64>,
-    /// The *count* half of the periodic cleanup trigger: the timer only merges
-    /// when this instance's shard has at least this many flushed generations,
-    /// skipping the pass otherwise (avoids rewriting the base table to reclaim a
-    /// single small generation). `None` defaults to `1` (clean up whenever any
-    /// generation is present on a tick).
-    pub cleanup_min_generations: Option<usize>,
 }
 
 /// A Lance-backed store for RL rollout trajectories.
@@ -171,10 +171,6 @@ pub struct RolloutStore {
     /// Periodic-cleanup interval in seconds; `0` disables the timer. See
     /// [`RolloutStoreOptions::cleanup_interval_secs`].
     cleanup_interval_secs: u64,
-    /// Minimum flushed generations before a periodic-cleanup tick merges.
-    /// Normalized to at least `1`. See
-    /// [`RolloutStoreOptions::cleanup_min_generations`].
-    cleanup_min_generations: usize,
     /// Timestamp of the last successful [`Self::compact`] on this handle.
     last_compaction: Option<DateTime<Utc>>,
     /// Number of successful compactions performed by this handle.
@@ -236,7 +232,6 @@ impl RolloutStore {
             storage_options,
             merge_after_generations: options.merge_after_generations.unwrap_or(0),
             cleanup_interval_secs: options.cleanup_interval_secs.unwrap_or(0),
-            cleanup_min_generations: options.cleanup_min_generations.unwrap_or(1).max(1),
             last_compaction: None,
             total_compactions: 0,
             last_compaction_error: None,
@@ -320,9 +315,10 @@ impl RolloutStore {
     ///
     /// This is the shared core of both triggers: the synchronous count trigger
     /// in [`Self::add`] (with `threshold = merge_after_generations`) and the
-    /// periodic timer in [`Self::spawn_periodic_cleanup`] (with
-    /// `threshold = cleanup_min_generations`). Both merge only the shard this
-    /// instance owns and writes, so the epoch claim never fences another writer.
+    /// periodic timer in [`Self::spawn_periodic_cleanup`] (with `threshold = 1`,
+    /// i.e. merge whatever is pending once the interval elapses). Both merge only
+    /// the shard this instance owns and writes, so the epoch claim never fences
+    /// another writer.
     async fn merge_own_shard_if_ready(&mut self, threshold: usize) -> LanceResult<usize> {
         let object_store = self.dataset.object_store(None).await?;
         let branch_location = self.dataset.branch_location();
@@ -343,10 +339,13 @@ impl RolloutStore {
         Ok(pending)
     }
 
-    /// Run one periodic WAL-cleanup pass over this instance's own shard: fold any
-    /// flushed generations into the base table once at least
-    /// `cleanup_min_generations` have accumulated. Returns the number of
-    /// generations reclaimed (`0` if below the threshold or nothing pending).
+    /// Run one periodic WAL-cleanup pass over this instance's own shard: fold
+    /// **every** flushed generation into the base table. This is the *time* half
+    /// of the "time OR count" trigger, so it is deliberately *not* gated by any
+    /// generation-count threshold: once the interval elapses, whatever is pending
+    /// gets merged even if the count trigger
+    /// ([`RolloutStoreOptions::merge_after_generations`]) never fired. Returns the
+    /// number of generations reclaimed (`0` only when nothing was pending).
     ///
     /// Exposed for callers that drive cleanup on their own schedule instead of
     /// (or in addition to) the built-in timer from
@@ -355,18 +354,21 @@ impl RolloutStore {
     /// with this instance's own appends but must not target another instance's
     /// shard.
     pub async fn cleanup_own_shard(&mut self) -> LanceResult<usize> {
-        self.merge_own_shard_if_ready(self.cleanup_min_generations)
-            .await
+        // Threshold `1`: merge whenever at least one generation is pending. The
+        // time trigger must not depend on the count threshold — that is what
+        // makes the two triggers a true OR.
+        self.merge_own_shard_if_ready(1).await
     }
 
     /// Spawn a background timer that periodically reclaims this instance's
     /// flushed MemWAL generations into the base table (the *time* trigger).
     ///
     /// Every `cleanup_interval_secs` seconds the task acquires the write lock and
-    /// calls [`Self::cleanup_own_shard`], which merges only when at least
-    /// `cleanup_min_generations` are pending. This bounds read amplification and
-    /// reclaims stale generation datasets even on shards that never cross the
-    /// synchronous count threshold ([`RolloutStoreOptions::merge_after_generations`]).
+    /// calls [`Self::cleanup_own_shard`], which merges whatever generations are
+    /// pending (no count threshold — time and count are a strict OR). This bounds
+    /// read amplification and reclaims stale generation datasets even on shards
+    /// that never cross the synchronous count threshold
+    /// ([`RolloutStoreOptions::merge_after_generations`]).
     ///
     /// Returns `None` (spawning nothing) when `cleanup_interval_secs` is `0`
     /// (disabled). Otherwise returns the task handle.
@@ -1952,7 +1954,6 @@ mod tests {
                 RolloutStoreOptions {
                     storage_options: None,
                     shard_id: Some("rollout-1".to_string()),
-                    cleanup_min_generations: Some(1),
                     ..Default::default()
                 },
             )
@@ -2106,11 +2107,12 @@ mod tests {
     }
 
     #[test]
-    fn cleanup_own_shard_merges_only_at_min_generations() {
+    fn cleanup_own_shard_merges_whatever_is_pending() {
         // The periodic-cleanup entry point (`cleanup_own_shard`) is the time
-        // trigger's per-pass body. With count self-merge disabled, generations
-        // accumulate; a cleanup pass below `cleanup_min_generations` is a no-op,
-        // and one at/above the threshold drains the shard into the base table.
+        // trigger's per-pass body. Time and count are a strict OR, so the time
+        // trigger is NOT gated by any generation count: with count self-merge
+        // disabled, a single pending generation is merged the moment the pass
+        // runs. A pass with nothing pending is a no-op.
         let dir = TempDir::new().unwrap();
         let uri = dir.path().to_string_lossy().to_string();
         let runtime = tokio::runtime::Runtime::new().unwrap();
@@ -2121,21 +2123,21 @@ mod tests {
                     storage_options: None,
                     shard_id: Some("rollout-0".to_string()),
                     merge_after_generations: None, // count trigger off
-                    cleanup_min_generations: Some(2),
                     ..Default::default()
                 },
             )
             .await
             .unwrap();
 
             store.add(&[assistant_record("a-0")]).await.unwrap();
-            // One generation pending, below min (2): cleanup is a no-op.
-            assert_eq!(store.cleanup_own_shard().await.unwrap(), 0);
-            assert_eq!(flushed_generation_count(&store).await, 1);
+            // One generation pending: the time trigger merges it immediately —
+            // it does not wait for a count threshold.
+            assert_eq!(store.cleanup_own_shard().await.unwrap(), 1);
+            assert_eq!(flushed_generation_count(&store).await, 0);
 
             store.add(&[assistant_record("a-1")]).await.unwrap();
-            // Two generations pending, at min: cleanup merges both.
-            assert_eq!(store.cleanup_own_shard().await.unwrap(), 2);
+            // Next generation is likewise merged on the following pass.
+            assert_eq!(store.cleanup_own_shard().await.unwrap(), 1);
             assert_eq!(flushed_generation_count(&store).await, 0);
 
             // Rows survive the merge, readable exactly once.
@@ -2169,7 +2171,6 @@ mod tests {
                     shard_id: Some("rollout-0".to_string()),
                     merge_after_generations: None, // only the timer merges
                     cleanup_interval_secs: Some(1),
-                    cleanup_min_generations: Some(1),
                 },
             )
             .await
@@ -2238,7 +2239,6 @@ mod tests {
                         RolloutStoreOptions {
                             storage_options: None,
                             shard_id: Some(shard),
-                            cleanup_min_generations: Some(1),
                             ..Default::default()
                         },
                     )
@@ -2309,7 +2309,6 @@ mod tests {
                 RolloutStoreOptions {
                     storage_options: None,
                     shard_id: Some("rollout-0".to_string()),
-                    cleanup_min_generations: Some(1),
                     ..Default::default()
                 },
             )

crates/lance-context-server/src/config.rs

@@ -37,20 +37,14 @@ pub struct ServerConfig {
     /// Interval, in seconds, for the periodic per-shard WAL cleanup task. When
     /// non-zero, each rollout store spawns a background timer that folds this
     /// instance's flushed MemWAL generations into the base table on a schedule —
-    /// the *time* half of the "time + count" trigger, complementing
-    /// `--rollout-merge-after-generations`. It reclaims stale generations even on
-    /// low-traffic shards that never cross the count threshold. `0` (the default)
-    /// disables the timer.
+    /// the *time* half of the "time OR count" trigger, complementing
+    /// `--rollout-merge-after-generations`. Whichever fires first merges: the
+    /// timer reclaims whatever is pending regardless of count, so stale
+    /// generations are folded in even on low-traffic shards that never cross the
+    /// count threshold. `0` (the default) disables the timer.
     #[arg(long, env = "ROLLOUT_CLEANUP_INTERVAL_SECS", default_value = "0")]
     pub rollout_cleanup_interval_secs: u64,
 
-    /// Minimum flushed generations a periodic cleanup tick requires before it
-    /// merges (avoids rewriting the base table to reclaim a single small
-    /// generation). Only meaningful when `--rollout-cleanup-interval-secs` is
-    /// set. Defaults to `1` (clean up whenever any generation is present).
-    #[arg(long, env = "ROLLOUT_CLEANUP_MIN_GENERATIONS", default_value = "1")]
-    pub rollout_cleanup_min_generations: usize,
-
     /// Upper bound on the number of resident rollout-store handles kept in
     /// memory (an LRU). With one physical dataset per experiment a deployment
     /// may hold hundreds of thousands of stores; this bounds how many stay open

crates/lance-context-server/src/routes/rollouts.rs

@@ -54,7 +54,6 @@ pub async fn create_rollout_store(
             .then_some(state.rollout_merge_after_generations),
         cleanup_interval_secs: (state.rollout_cleanup_interval_secs > 0)
             .then_some(state.rollout_cleanup_interval_secs),
-        cleanup_min_generations: Some(state.rollout_cleanup_min_generations),
     };
 
     let store = RolloutStore::open_with_options(&uri, options)

crates/lance-context-server/src/state.rs

@@ -47,9 +47,6 @@ pub struct AppState {
     /// Periodic per-shard WAL-cleanup interval in seconds; `0` disables the
     /// global sweeper. See [`Self::spawn_global_sweeper`].
     pub rollout_cleanup_interval_secs: u64,
-    /// Minimum flushed generations before a periodic cleanup tick merges. See
-    /// `RolloutStoreOptions::cleanup_min_generations`.
-    pub rollout_cleanup_min_generations: usize,
 }
 
 impl AppState {
@@ -72,7 +69,6 @@ impl AppState {
             instance_id,
             rollout_merge_after_generations: config.rollout_merge_after_generations,
             rollout_cleanup_interval_secs: config.rollout_cleanup_interval_secs,
-            rollout_cleanup_min_generations: config.rollout_cleanup_min_generations,
         })
     }
 
@@ -109,7 +105,6 @@ impl AppState {
             instance_id,
             rollout_merge_after_generations: 0,
             rollout_cleanup_interval_secs: 0,
-            rollout_cleanup_min_generations: 1,
         }
     }
 
@@ -130,7 +125,6 @@ impl AppState {
                 .then_some(self.rollout_merge_after_generations),
             cleanup_interval_secs: (self.rollout_cleanup_interval_secs > 0)
                 .then_some(self.rollout_cleanup_interval_secs),
-            cleanup_min_generations: Some(self.rollout_cleanup_min_generations),
         }
     }