From 59f8f1a6bc5889fa09f06c3b6b52085276f7aa38 Mon Sep 17 00:00:00 2001 From: Beinan Date: Sat, 25 Jul 2026 06:54:41 +0000 Subject: [PATCH] fix(rollout): stop swallowing failures in RolloutStore's detached close MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `RolloutStore::Drop` spawns a detached `ShardWriter::close` and discards its result with `let _ =`. Because `close` seals the active memtable, this path is what keeps an LRU-evicted store's unflushed rows from being stranded — so every way it can fail to run is a case where rows stay durable in the WAL but invisible to reads until the next process restart, with no signal anywhere. Log all three: a failing close, a writer still shared by an in-flight append (so the close no-ops), and a drop with no Tokio runtime (so nothing is spawned at all). Behavior is unchanged — this is purely observability on a path that was silent by construction. Also documents the eviction guarantee on `AppState::rollout_stores`, since the flush sweeper only visits resident stores and the reason eviction is nonetheless safe was not written down anywhere. Adds a test asserting end to end that dropping a store with an unsealed memtable makes its rows visible, so the eviction path cannot regress. Closes #185 Co-Authored-By: Claude --- .../lance-context-core/src/rollout_store.rs | 82 ++++++++++++++++++- crates/lance-context-server/src/state.rs | 14 ++++ 2 files changed, 94 insertions(+), 2 deletions(-) diff --git a/crates/lance-context-core/src/rollout_store.rs b/crates/lance-context-core/src/rollout_store.rs index c565e7d..4931921 100644 --- a/crates/lance-context-core/src/rollout_store.rs +++ b/crates/lance-context-core/src/rollout_store.rs @@ -2162,17 +2162,49 @@ impl Drop for RolloutStore { /// Tokio runtime is available we move the writer into a detached task that /// closes it; otherwise (no runtime, e.g. some teardown paths) we can only /// drop it. Callers that can `await` should prefer [`Self::close`]. + /// + /// `ShardWriter::close` seals the active memtable, so this path is also + /// what keeps an evicted-but-unflushed store's rows from being stranded. + /// Each way it can fail to do that is logged rather than swallowed: a + /// stranded memtable leaves rows durable in the WAL but invisible to reads + /// until the next process restart replays it, which is near-impossible to + /// diagnose without a signal here. fn drop(&mut self) { // `&mut self` in drop → exclusive access, so `get_mut` avoids a lock. if let Some(writer) = self.write_writer.get_mut().take() { + let shard = self.write_shard; if let Ok(handle) = tokio::runtime::Handle::try_current() { handle.spawn(async move { // Only the sole owner can close; if an append still shares the // Arc it will drop last. Best-effort either way. - if let Ok(writer) = Arc::try_unwrap(writer) { - let _ = writer.close().await; + match Arc::try_unwrap(writer) { + Ok(writer) => { + if let Err(err) = writer.close().await { + tracing::warn!( + shard = %shard, + error = %err, + "detached close of a dropped rollout writer failed; \ + unflushed rows stay durable in the WAL but remain \ + invisible until the next replay" + ); + } + } + Err(_shared) => { + tracing::debug!( + shard = %shard, + "dropped rollout writer is still shared by an in-flight \ + append; close deferred to the last Arc holder" + ); + } } }); + } else { + tracing::warn!( + shard = %shard, + "rollout writer dropped with no Tokio runtime available; its \ + background tasks leak and unflushed rows remain invisible until \ + the next WAL replay" + ); } } } @@ -3136,6 +3168,52 @@ mod tests { }); } + #[test] + fn dropping_a_store_seals_its_unflushed_rows() { + // The server's flush sweeper only visits LRU-resident stores, so a store + // evicted while holding an unsealed memtable depends entirely on `Drop` + // to seal it. `Drop` spawns a detached `ShardWriter::close`, which does + // seal — this asserts that end to end, so the eviction path cannot + // silently regress into stranding rows. + 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 options = || RolloutStoreOptions { + storage_options: None, + session: None, + shard_id: Some("evicted-0".to_string()), + merge_after_generations: None, + }; + + { + let store = RolloutStore::open_with_options(&uri, options()) + .await + .unwrap(); + store.add(&[assistant_record("e-0")]).await.unwrap(); + // Deliberately no flush() and no close(): this models an LRU + // eviction dropping the last handle. + assert!(store.list(None, None).await.unwrap().is_empty()); + } + + // The detached close is spawned, not awaited, so poll for the row + // rather than assuming it has landed by now. + let mut seen = 0; + for _ in 0..100 { + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + let reader = RolloutStore::open_with_options(&uri, options()) + .await + .unwrap(); + seen = reader.list(None, None).await.unwrap().len(); + if seen > 0 { + break; + } + } + assert_eq!(seen, 1, "Drop must seal the memtable of an evicted store"); + }); + } + #[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/state.rs b/crates/lance-context-server/src/state.rs index e844159..e0711b6 100644 --- a/crates/lance-context-server/src/state.rs +++ b/crates/lance-context-server/src/state.rs @@ -30,6 +30,20 @@ pub struct AppState { /// residency: on overflow the least-recently-used handle is evicted and its /// `Arc>` dropped. Existence is tracked durably by /// [`Self::rollout_registry`], not by membership in this cache. + /// + /// # Eviction does not strand unflushed rows + /// + /// The flush sweeper only visits *resident* stores, so it is worth being + /// explicit about why evicting a store that still has an unsealed memtable + /// is safe: dropping the last handle runs `RolloutStore`'s `Drop`, which + /// spawns a detached `ShardWriter::close`, and that seals the active + /// memtable before draining. An evicted store's rows therefore still become + /// visible without the sweeper ever seeing it. + /// + /// The residual risk is that the detached close is best-effort — it cannot + /// be awaited from `drop`, and it no-ops if an in-flight append still shares + /// the `Arc`. Both cases are logged (see that `Drop` impl) rather than + /// silently stranding rows. pub rollout_stores: Mutex>>>, /// Durable directory of which rollout stores exist. Consulted on a cache /// miss (existence check) and to back the list endpoint. Guarded by a lock