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();