feat(rollout): periodic per-shard WAL cleanup + concurrency hardening

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

@@ -33,6 +33,23 @@ pub struct ServerConfig {
     /// unioned at read time.
     #[arg(long, env = "ROLLOUT_MERGE_AFTER_GENERATIONS", default_value = "0")]
     pub rollout_merge_after_generations: usize,
+
+    /// 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.
+    #[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,
 }
 
 impl ServerConfig {

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

@@ -626,6 +626,8 @@ mod tests {
             base_path: dir.path().to_path_buf(),
             instance_id: None,
             rollout_merge_after_generations: 0,
+            rollout_cleanup_interval_secs: 0,
+            rollout_cleanup_min_generations: 1,
         });
         (state, dir)
     }

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

@@ -42,15 +42,24 @@ pub async fn create_rollout_store(
         shard_id: state.instance_id.clone(),
         merge_after_generations: (state.rollout_merge_after_generations > 0)
             .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)
         .await
         .map_err(AppError::from_lance)?;
     let version = store.version();
 
+    let store = Arc::new(RwLock::new(store));
+    // Start the periodic per-shard WAL-cleanup timer (no-op when the interval is
+    // disabled). The handle is detached: it is aborted when the store is dropped
+    // on delete, and otherwise runs for the server's lifetime.
+    let _cleanup = RolloutStore::spawn_periodic_cleanup(store.clone());
+
     let mut stores = state.rollout_stores.write().await;
-    stores.insert(req.name.clone(), Arc::new(RwLock::new(store)));
+    stores.insert(req.name.clone(), store);
 
     Ok((
         StatusCode::CREATED,
@@ -510,6 +519,8 @@ mod tests {
             base_path: dir.path().to_path_buf(),
             instance_id: None,
             rollout_merge_after_generations: 0,
+            rollout_cleanup_interval_secs: 0,
+            rollout_cleanup_min_generations: 1,
         });
         let (_status, _info) = create_rollout_store(
             State(state.clone()),

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

@@ -159,6 +159,8 @@ mod tests {
             base_path: dir.path().to_path_buf(),
             instance_id: None,
             rollout_merge_after_generations: 0,
+            rollout_cleanup_interval_secs: 0,
+            rollout_cleanup_min_generations: 1,
         });
         (state, dir)
     }

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

@@ -18,6 +18,12 @@ pub struct AppState {
     /// Count-triggered self-merge threshold for rollout MemWAL shards; `0`
     /// disables it. See `RolloutStoreOptions::merge_after_generations`.
     pub rollout_merge_after_generations: usize,
+    /// Periodic per-shard WAL-cleanup interval in seconds; `0` disables the
+    /// background timer. See `RolloutStoreOptions::cleanup_interval_secs`.
+    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 {
@@ -29,6 +35,8 @@ impl AppState {
             base_path: PathBuf::from(&config.data_dir),
             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,
         }
     }
 

crates/lance-context/src/unified_rollout.rs

@@ -38,6 +38,11 @@ impl RolloutStore {
             // caller that wants count-triggered self-merge opens the core
             // `RolloutStore` directly with `merge_after_generations` set.
             merge_after_generations: None,
+            // Embedded use has no long-lived background runtime to own a cleanup
+            // timer; leave the periodic WAL cleanup disabled here. A caller that
+            // wants it opens the core `RolloutStore` directly and drives
+            // `spawn_periodic_cleanup`.
+            ..Default::default()
         };
         let store = LocalStore::open_with_options(uri, options)
             .await