feat(rollout): concurrent writes with periodic flush

crates/lance-context-core/tests/wal_merge_generation_cleanup.rs

@@ -87,32 +87,51 @@ async fn serial_merge_deletes_merged_generation_dirs() {
         .await
         .unwrap();
 
-    // 30 serial single-row appends. Each append flushes one generation; every
-    // 10th triggers a count-merge that folds the accumulated generations into
-    // the base table and drains them from the manifest.
+    // 30 serial single-row appends. `add` is durable-only now, so we `flush`
+    // each one into its own generation (as an inline seal used to), then run the
+    // count-triggered merge explicitly (it moved off the append path onto the
+    // periodic sweeper). Every 10th accumulated generation is folded into the
+    // base table and drained from the manifest.
     let n = 30;
     for i in 0..n {
         store.add(&[rec(&format!("row-{i}"))]).await.unwrap();
+        store.flush().await.unwrap();
+        store.maybe_merge_own_shard().await.unwrap();
     }
 
     let obs = store.observe().await.unwrap();
     let manifest_pending = obs.pending_wal_generations as usize;
-    let row_count = obs.row_count;
+
+    // Data correctness must be measured through the LSM read path, which
+    // de-duplicates by `id`. `observe().row_count` is a raw `count_rows()` over
+    // the base table and does NOT de-duplicate, so it over-counts the physical
+    // duplicate rows that a WAL merge can leave in the base table (see the
+    // dedup-consistency issue). Assert on the read path instead.
+    let listed = store.list(None, None).await.unwrap();
+    let mut ids: Vec<String> = listed.iter().map(|record| record.id.clone()).collect();
+    ids.sort();
+    ids.dedup();
 
     let on_disk = count_gen_dirs_on_disk(Path::new(&uri));
 
     eprintln!(
-        "appends={n} row_count={row_count} \
+        "appends={n} listed={} unique_ids={} raw_row_count={} \
          manifest_pending_generations={manifest_pending} \
          gen_dirs_on_disk={on_disk} leaked={}",
+        listed.len(),
+        ids.len(),
+        obs.row_count,
         on_disk.saturating_sub(manifest_pending)
     );
 
-    // Data correctness: all rows present exactly once.
+    // Data correctness: every appended row is readable exactly once through the
+    // deduplicating read path.
     assert_eq!(
-        row_count as usize, n,
-        "all rows must be readable exactly once"
+        listed.len(),
+        n,
+        "read path must return each appended row exactly once"
     );
+    assert_eq!(ids.len(), n, "no duplicate ids on the read path");
 
     // The fix: merged generations are drained from the manifest AND their blob
     // dirs are deleted, so on-disk gen dirs never exceed the manifest's pending

crates/lance-context-master/src/routes.rs

@@ -771,7 +771,7 @@ mod tests {
         let dir = TempDir::new().unwrap();
         let state = MasterState::new(test_config(&dir)).await.unwrap();
         let uri = state.rollout_uri("records");
-        let mut store = RolloutStore::open(&uri).await.unwrap();
+        let store = RolloutStore::open(&uri).await.unwrap();
         store
             .add(&[
                 test_record("assistant-1", false),
@@ -841,7 +841,7 @@ mod tests {
         let dir = TempDir::new().unwrap();
         let state = MasterState::new(test_config(&dir)).await.unwrap();
         let uri = state.rollout_uri("records");
-        let mut store = RolloutStore::open(&uri).await.unwrap();
+        let store = RolloutStore::open(&uri).await.unwrap();
         store
             .add(&[
                 test_record("assistant-1", false),
@@ -916,7 +916,7 @@ mod tests {
         let dir = TempDir::new().unwrap();
         let state = MasterState::new(test_config(&dir)).await.unwrap();
         let uri = state.rollout_uri("blobs");
-        let mut store = RolloutStore::open(&uri).await.unwrap();
+        let store = RolloutStore::open(&uri).await.unwrap();
         store
             .add(&[
                 test_record("artifact-1", true),

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

@@ -24,27 +24,40 @@ pub struct ServerConfig {
     #[arg(long, env = "INSTANCE_ID")]
     pub instance_id: Option<String>,
 
-    /// Count-triggered self-merge threshold for rollout MemWAL shards. After an
-    /// append flushes a new generation, if this instance's shard has at least
-    /// this many un-merged flushed generations, the append synchronously folds
-    /// them into the base table and drains the shard (see
+    /// Count-triggered self-merge threshold for rollout MemWAL shards. When this
+    /// instance's shard has at least this many un-merged flushed generations, the
+    /// periodic sweeper folds them into the base table and drains the shard (see
     /// `specs/rollout-deployment.md`). This bounds read amplification. `0`
-    /// (the default) disables self-merge — generations accumulate and are
-    /// unioned at read time.
+    /// (the default) disables the count trigger — generations accumulate and are
+    /// unioned at read time (or reclaimed by the time-based cleanup below).
+    ///
+    /// Note: this merge runs on the periodic sweeper, not on the append path.
+    /// Appends are durable (WAL-persisted) but do not merge inline.
     #[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 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.
+    /// non-zero, the global sweeper folds this instance's flushed MemWAL
+    /// generations into the base table on a schedule — 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,
 
+    /// Interval, in seconds, at which the sweeper flushes each resident rollout
+    /// store's active MemWAL memtable into a queryable generation. Rollout
+    /// appends are durable on return (the WAL entry is persisted to object
+    /// storage) but are not visible to reads until the memtable is flushed, so
+    /// this interval bounds read-after-write latency. Decoupling flush from the
+    /// append path is what lets concurrent appends run without serializing behind
+    /// a per-append seal. Default `30`; `0` disables periodic flush (rows then
+    /// only become visible when the cleanup/merge path flushes them).
+    #[arg(long, env = "ROLLOUT_FLUSH_INTERVAL_SECS", default_value = "30")]
+    pub rollout_flush_interval_secs: u64,
+
     /// 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/main.rs

@@ -47,6 +47,10 @@ async fn main() {
     // the last `Arc<AppState>` is dropped.
     let _sweeper = state.spawn_global_sweeper();
 
+    // Periodic MemWAL flush sweeper: bounds rollout read-after-write latency
+    // without serializing concurrent appends. Detached for the server lifetime.
+    let _flush_sweeper = state.spawn_flush_sweeper();
+
     // Install the Prometheus recorder once, before any metrics are emitted.
     let metrics_handle = lance_context_metrics::install_recorder();
 

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

@@ -274,7 +274,10 @@ pub async fn add_rollouts(
         .collect();
     let count = core_records.len();
 
-    let mut store = store_lock.write().await;
+    // A read lock: `add` is `&self` and MemWAL appends are internally
+    // concurrent, so multiple ingest requests to the same store run in parallel.
+    // Mutating ops (merge, compact, checkout, close) still take the write lock.
+    let store = store_lock.read().await;
     let version = store
         .add(&core_records)
         .await
@@ -824,6 +827,14 @@ mod tests {
             .unwrap()
     }
 
+    /// Flush a rollout store's MemWAL so just-added rows become visible to reads.
+    /// `add` is durable-but-async now (visible only after a flush / the periodic
+    /// flush sweeper), so tests that add then read must flush explicitly.
+    async fn flush_store(state: &Arc<AppState>, name: &str) {
+        let store = state.get_or_open_rollout_store(name).await.unwrap();
+        store.read().await.flush().await.unwrap();
+    }
+
     /// Assemble a `multipart/form-data` body from ordered (name, bytes) parts.
     fn multipart_request(boundary: &str, parts: &[(&str, Vec<u8>)]) -> Request {
         let mut body = Vec::new();
@@ -877,6 +888,7 @@ mod tests {
         assert_eq!(status, StatusCode::CREATED);
         assert_eq!(resp.count, 1);
         assert_eq!(resp.ids, vec!["r0".to_string()]);
+        flush_store(&state, "rl").await;
 
         // A plain get projects the binary column out, reading it back as None...
         let Json(got) = get_rollout(
@@ -917,6 +929,7 @@ mod tests {
             .unwrap();
         assert_eq!(status, StatusCode::CREATED);
         assert_eq!(resp.count, 1);
+        flush_store(&state, "rl").await;
 
         let resp = fetch_rollout_blob(
             State(state.clone()),
@@ -1043,6 +1056,9 @@ mod tests {
         )
         .await
         .expect("append succeeds");
+        // Seal the append into a flushed generation so the merge has something
+        // to reclaim (appends no longer flush inline).
+        flush_store(&state, "rl").await;
 
         let Json(first) = merge_wal(State(state.clone()), Path("rl".to_string()))
             .await
@@ -1082,6 +1098,9 @@ mod tests {
         )
         .await
         .expect("write on instance A");
+        // Flush A's shard so its row is visible to any reader (reads union all
+        // shards' flushed generations).
+        flush_store(&state_a, "rl").await;
 
         // Pod B: a fresh AppState over the SAME data dir, with an empty cache and
         // a different instance id (its own shard). It never saw the `create`.
@@ -1116,6 +1135,8 @@ mod tests {
         )
         .await
         .expect("write on instance B");
+        // Flush B's shard so its row is visible to A's reader too.
+        flush_store(&state_b, "rl").await;
         assert_eq!(count_rollouts(&state_a).await, 2);
     }
 
@@ -1230,6 +1251,7 @@ mod tests {
         )
         .await
         .unwrap();
+        flush_store(&state, "rl").await;
 
         let Json(response) = list_rollouts(
             State(state),

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

@@ -47,6 +47,10 @@ 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,
+    /// Periodic MemWAL flush interval in seconds; `0` disables periodic flush.
+    /// Bounds rollout read-after-write latency (appends are durable but not
+    /// visible until flushed). See [`Self::spawn_global_sweeper`].
+    pub rollout_flush_interval_secs: u64,
     /// Admission budget for in-flight artifact-blob bytes across concurrent
     /// uploads/downloads. `None` disables the budget (unbounded). See
     /// [`BlobBudget`].
@@ -154,6 +158,7 @@ impl AppState {
             instance_id,
             rollout_merge_after_generations: config.rollout_merge_after_generations,
             rollout_cleanup_interval_secs: config.rollout_cleanup_interval_secs,
+            rollout_flush_interval_secs: config.rollout_flush_interval_secs,
             blob_budget,
         })
     }
@@ -191,6 +196,7 @@ impl AppState {
             instance_id,
             rollout_merge_after_generations: 0,
             rollout_cleanup_interval_secs: 0,
+            rollout_flush_interval_secs: 0,
             blob_budget: None,
         }
     }
@@ -426,7 +432,92 @@ impl AppState {
         }))
     }
 
-    /// Gracefully drain every resident rollout writer on shutdown.
+    /// Spawn the process-wide MemWAL flush sweeper.
+    ///
+    /// Rollout appends are durable on return (the WAL entry is persisted) but not
+    /// visible to reads until the active memtable is flushed into a queryable
+    /// generation. Rather than flush on every append — which would serialize
+    /// concurrent writes behind a per-append seal — this sweeper flushes each
+    /// resident store on a fixed interval, bounding read-after-write latency
+    /// while keeping the append path concurrent.
+    ///
+    /// After flushing a store it also runs the count-triggered merge
+    /// ([`RolloutStore::maybe_merge_own_shard`]): the read-amplification bound
+    /// that formerly lived on the append path now rides this timer. The heavier
+    /// time-based cleanup/merge remains on [`Self::spawn_global_sweeper`].
+    ///
+    /// Returns `None` when the flush interval is `0`.
+    pub fn spawn_flush_sweeper(self: &Arc<Self>) -> Option<JoinHandle<()>> {
+        let interval_secs = self.rollout_flush_interval_secs;
+        if interval_secs == 0 {
+            return None;
+        }
+        let interval = Duration::from_secs(interval_secs);
+        // Abandon any single store's flush that outruns five intervals (min 30s)
+        // so one wedged store cannot stall flushing for the rest.
+        let pass_timeout = interval.saturating_mul(5).max(Duration::from_secs(30));
+        let weak = Arc::downgrade(self);
+        Some(tokio::spawn(async move {
+            let mut ticker = tokio::time::interval(interval);
+            ticker.tick().await; // skip the immediate first tick
+            loop {
+                ticker.tick().await;
+                let Some(state) = weak.upgrade() else {
+                    return;
+                };
+                let resident: Vec<(String, Arc<RwLock<RolloutStore>>)> = {
+                    let cache = state.rollout_stores.lock().await;
+                    cache
+                        .iter()
+                        .map(|(name, store)| (name.clone(), store.clone()))
+                        .collect()
+                };
+                for (name, store) in resident {
+                    // Flush under a read lock so concurrent appends are not blocked.
+                    {
+                        let guard = store.read().await;
+                        match tokio::time::timeout(pass_timeout, guard.flush()).await {
+                            Ok(Ok(())) => {}
+                            Ok(Err(e)) => {
+                                metrics::counter!("rollout_wal_flush_total", "result" => "failed")
+                                    .increment(1);
+                                tracing::warn!(store = %name, error = %e, "flush sweeper failed");
+                                continue;
+                            }
+                            Err(_elapsed) => {
+                                metrics::counter!("rollout_wal_flush_total", "result" => "timeout")
+                                    .increment(1);
+                                tracing::warn!(store = %name, "flush sweeper timed out");
+                                continue;
+                            }
+                        }
+                        metrics::counter!("rollout_wal_flush_total", "result" => "ok").increment(1);
+                    }
+                    // Count-triggered merge (no-op unless the threshold is set and
+                    // met). Needs the write lock; excludes concurrent appends.
+                    let mut guard = store.write().await;
+                    match tokio::time::timeout(pass_timeout, guard.maybe_merge_own_shard()).await {
+                        Ok(Ok(0)) => {}
+                        Ok(Ok(n)) => {
+                            metrics::counter!("rollout_wal_generations_reclaimed_total")
+                                .increment(n as u64);
+                            tracing::info!(
+                                store = %name,
+                                reclaimed = n,
+                                "flush sweeper count-merged flushed generations"
+                            );
+                        }
+                        Ok(Err(e)) => {
+                            tracing::warn!(store = %name, error = %e, "flush sweeper merge failed");
+                        }
+                        Err(_elapsed) => {
+                            tracing::warn!(store = %name, "flush sweeper merge timed out");
+                        }
+                    }
+                }
+            }
+        }))
+    }
     ///
     /// [`RolloutStore`]'s writer ([`ShardWriter`]) has no `Drop`, so its
     /// background tasks are only reclaimed by an explicit `close().await`. On the