From 99d71d341647b6f0c52c9c47acd80892faaaab5f Mon Sep 17 00:00:00 2001 From: Beinan Date: Sat, 25 Jul 2026 06:45:42 +0000 Subject: [PATCH] fix(rollout): seal the memtable in cleanup_own_shard before merging `ROLLOUT_FLUSH_INTERVAL_SECS=0` is documented as falling back to the cleanup/merge path for visibility, but that fallback could not work. `cleanup_own_shard` delegates to `merge_own_shard_if_ready(1)`, which reads the shard manifest and returns early when `flushed_generations` is empty. With no periodic flush nothing ever seals the active memtable, so that list stayed empty forever: the merge was never reached, and rows that `add` had durably persisted stayed invisible until a process restart replayed the WAL. Flush before the threshold check so cleanup is a genuine standalone fallback. Confirmed by a probe: with a row added and no flush, `cleanup_own_shard` returned 0 and the row stayed invisible; it now returns 1 and the row is readable. Also warn at startup when both the flush and cleanup intervals are 0, since that combination still leaves nothing to seal, and correct the config doc which promised the fallback unconditionally. The new test fails on the parent commit (`reclaimed` 0 != 1) and passes here. Closes #184 Co-Authored-By: Claude --- .../lance-context-core/src/rollout_store.rs | 63 +++++++++++++++++++ crates/lance-context-server/src/config.rs | 11 +++- crates/lance-context-server/src/main.rs | 14 +++++ 3 files changed, 86 insertions(+), 2 deletions(-) diff --git a/crates/lance-context-core/src/rollout_store.rs b/crates/lance-context-core/src/rollout_store.rs index 22d09f4..c5b4ce7 100644 --- a/crates/lance-context-core/src/rollout_store.rs +++ b/crates/lance-context-core/src/rollout_store.rs @@ -722,7 +722,21 @@ impl RolloutStore { /// only on the shard this instance owns, so it is safe to call concurrently /// with this instance's own appends but must not target another instance's /// shard. + /// + /// # Seals first + /// + /// This flushes the active memtable before looking at the manifest. Without + /// that, a deployment with the periodic flush sweeper disabled + /// (`ROLLOUT_FLUSH_INTERVAL_SECS=0`) could never make progress: nothing + /// would seal the memtable, so `flushed_generations` would stay empty, so + /// the threshold check below would return `0` and never reach the merge — + /// leaving rows durable but permanently invisible until a process restart + /// replayed the WAL. Sealing here makes the cleanup path a genuine + /// standalone fallback, as `ROLLOUT_FLUSH_INTERVAL_SECS`'s documentation + /// already promised. pub async fn cleanup_own_shard(&mut self) -> LanceResult { + // Materialize anything buffered so it is eligible for this pass. + self.flush().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. @@ -3060,6 +3074,55 @@ mod tests { }); } + #[test] + fn cleanup_own_shard_seals_before_merging() { + // Regression guard for the `ROLLOUT_FLUSH_INTERVAL_SECS=0` trap: with no + // periodic flush, nothing seals the active memtable, so + // `flushed_generations` stays empty and the threshold check in + // `merge_own_shard_if_ready` used to return 0 without ever merging — + // leaving rows durable but permanently invisible. + // + // `cleanup_own_shard` now flushes first, making it a genuine standalone + // fallback, which is what the config docs already claimed. + let dir = TempDir::new().unwrap(); + let uri = dir.path().to_string_lossy().to_string(); + + let runtime = tokio::runtime::Runtime::new().unwrap(); + runtime.block_on(async { + let mut store = RolloutStore::open_with_options( + &uri, + RolloutStoreOptions { + storage_options: None, + session: None, + shard_id: Some("cleanup-seal-0".to_string()), + // Count trigger disabled: cleanup is the only path that can + // make this row visible, exactly as with flush interval 0. + merge_after_generations: Some(0), + }, + ) + .await + .unwrap(); + + store.add(&[assistant_record("c-0")]).await.unwrap(); + + // Nothing sealed yet: invisible, and no generation pending. + assert!(store.list(None, None).await.unwrap().is_empty()); + assert_eq!( + store.observe().await.unwrap().pending_wal_generations, + 0, + "precondition: the memtable is unsealed, so no generation exists" + ); + + // A single cleanup pass must seal, merge, and expose the row. + let reclaimed = store.cleanup_own_shard().await.unwrap(); + assert_eq!(reclaimed, 1, "cleanup must seal then merge the generation"); + + let seen = store.list(None, None).await.unwrap(); + assert_eq!(seen.len(), 1, "row must be visible after cleanup alone"); + assert_eq!(seen[0].id, "c-0"); + }); + } + #[test] fn distinct_shards_share_one_dataset() { // Two instances writing distinct shards of the same dataset both diff --git a/crates/lance-context-server/src/config.rs b/crates/lance-context-server/src/config.rs index 4078139..e386bfe 100644 --- a/crates/lance-context-server/src/config.rs +++ b/crates/lance-context-server/src/config.rs @@ -53,8 +53,15 @@ pub struct ServerConfig { /// 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). + /// a per-append seal. Default `30`. + /// + /// `0` disables periodic flush, leaving the cleanup sweeper + /// (`ROLLOUT_CLEANUP_INTERVAL_SECS`) as the only thing that seals memtables + /// — it flushes before merging, so it is a sufficient fallback, but + /// read-after-write latency is then bounded by the *cleanup* interval + /// instead. Setting **both** to `0` means nothing ever seals: appends stay + /// durable but invisible until the process restarts and replays the WAL. + /// The server warns at startup in that configuration. #[arg(long, env = "ROLLOUT_FLUSH_INTERVAL_SECS", default_value = "30")] pub rollout_flush_interval_secs: u64, diff --git a/crates/lance-context-server/src/main.rs b/crates/lance-context-server/src/main.rs index 9adf25b..0ba9f32 100644 --- a/crates/lance-context-server/src/main.rs +++ b/crates/lance-context-server/src/main.rs @@ -51,6 +51,20 @@ async fn main() { // without serializing concurrent appends. Detached for the server lifetime. let _flush_sweeper = state.spawn_flush_sweeper(); + // With both sweepers off nothing ever seals the active memtable, so rollout + // appends stay durable-but-invisible until the process restarts and replays + // the WAL. `cleanup_own_shard` flushes before merging, so the cleanup + // sweeper alone is a sufficient fallback — but if neither runs, warn loudly + // rather than leaving the operator to discover it as missing rows. + if state.rollout_flush_interval_secs == 0 && state.rollout_cleanup_interval_secs == 0 { + tracing::warn!( + "ROLLOUT_FLUSH_INTERVAL_SECS=0 and ROLLOUT_CLEANUP_INTERVAL_SECS=0: \ + nothing will seal MemWAL memtables, so rollout appends will be durable \ + but invisible to reads until this process restarts. Set at least one \ + of them to a non-zero interval." + ); + } + // Install the Prometheus recorder once, before any metrics are emitted. let metrics_handle = lance_context_metrics::install_recorder();