From def4050577baf25bdd0f287b9e6cf743c9fb4b79 Mon Sep 17 00:00:00 2001 From: Beinan Wang Date: Wed, 21 Oct 2026 06:38:31 +0000 Subject: [PATCH 1/4] Recover durable own-shard WAL before cleanup after restart --- .../lance-context-core/src/generic_store.rs | 44 ++++++++++ .../lance-context-core/src/rollout_store.rs | 88 +++++++++++++++++++ crates/lance-context-core/src/store_base.rs | 85 +++++++++++++++++- 3 files changed, 215 insertions(+), 2 deletions(-) diff --git a/crates/lance-context-core/src/generic_store.rs b/crates/lance-context-core/src/generic_store.rs index 4ddff56..6862fdb 100644 --- a/crates/lance-context-core/src/generic_store.rs +++ b/crates/lance-context-core/src/generic_store.rs @@ -1383,6 +1383,50 @@ mod tests { }); } + #[test] + fn cleanup_recovers_own_wal_updates_without_new_ingest_after_runtime_exit() { + let dir = TempDir::new().unwrap(); + let uri = dir.path().to_string_lossy().to_string(); + let options = GenericStoreOptions { + shard_id: Some("restarted-generic-writer".into()), + ..Default::default() + }; + let old_runtime = tokio::runtime::Runtime::new().unwrap(); + let old = old_runtime.block_on(async { + let store = GenericStore::open(&uri, spec(), options.clone()) + .await + .unwrap(); + store + .add(&[row(json!({"id":"shared", "score":1.0}))]) + .await + .unwrap(); + store.flush().await.unwrap(); + store + .add(&[row(json!({"id":"shared", "score":2.0}))]) + .await + .unwrap(); + store + }); + drop(old_runtime); + drop(old); + let runtime = tokio::runtime::Runtime::new().unwrap(); + runtime.block_on(async { + let mut recovered = GenericStore::open_existing(&uri, options).await.unwrap(); + assert_eq!( + recovered.get("shared", None).await.unwrap().unwrap()["score"], + json!(1.0) + ); + assert!(recovered.cleanup_wal().await.unwrap() > 0); + assert_eq!( + recovered.get("shared", None).await.unwrap().unwrap()["score"], + json!(2.0) + ); + assert_eq!(recovered.list(None, None).await.unwrap().len(), 1); + assert_eq!(recovered.cleanup_wal().await.unwrap(), 0); + recovered.close().await.unwrap(); + }); + } + #[test] fn seal_mode_survives_a_reopen() { // The mode is a property of the store, not of whoever opens it. Before diff --git a/crates/lance-context-core/src/rollout_store.rs b/crates/lance-context-core/src/rollout_store.rs index 940fd9c..f79e655 100644 --- a/crates/lance-context-core/src/rollout_store.rs +++ b/crates/lance-context-core/src/rollout_store.rs @@ -4686,6 +4686,94 @@ mod tests { }); } + #[test] + fn cleanup_recovers_own_wal_without_new_ingest_after_runtime_exit() { + let dir = TempDir::new().unwrap(); + let uri = dir.path().to_string_lossy().to_string(); + let options = RolloutStoreOptions { + shard_id: Some("restarted-writer".into()), + ..Default::default() + }; + let payload = vec![37; 64 * 1024]; + let old_runtime = tokio::runtime::Runtime::new().unwrap(); + let old = old_runtime.block_on(async { + let store = RolloutStore::open_with_options(&uri, options.clone()) + .await + .unwrap(); + store + .add(&[artifact_record("durable-before-exit", &payload)]) + .await + .unwrap(); + assert_eq!(flushed_generation_count(&store).await, 0); + store + }); + // Stop every old background task before dropping the store outside a + // runtime. Do not run its graceful close (which would hide this bug). + drop(old_runtime); + drop(old); + + let runtime = tokio::runtime::Runtime::new().unwrap(); + runtime.block_on(async { + let mut other = RolloutStore::open_with_options( + &uri, + RolloutStoreOptions { + shard_id: Some("different-live-writer".into()), + ..Default::default() + }, + ) + .await + .unwrap(); + assert_eq!(other.cleanup_own_shard().await.unwrap(), 0); + assert!(other.own_shard_watermark().await.unwrap().is_none()); + assert!(other + .get_by_id("durable-before-exit") + .await + .unwrap() + .is_none()); + + let mut recovered = RolloutStore::open_with_options(&uri, options) + .await + .unwrap(); + recovered.flush().await.unwrap(); + assert!(recovered + .get_by_id("durable-before-exit") + .await + .unwrap() + .is_none()); + // Exercise the split-lock route used by the actual HTTP merge. + let (manifests, manifest, prepared) = recovered + .prepare_cleanup_merge() + .await + .unwrap() + .expect("cleanup must replay acknowledged but unsealed WAL"); + assert!( + recovered + .commit_prepared_merge(&manifests, &manifest, prepared) + .await + .unwrap() + > 0 + ); + assert_eq!( + recovered.get_blob("durable-before-exit").await.unwrap(), + Some(payload) + ); + assert_eq!(recovered.cleanup_own_shard().await.unwrap(), 0); + let epoch = shard_writer_epoch(&recovered).await; + recovered + .add(&[assistant_record("after-recovery")]) + .await + .unwrap(); + recovered.cleanup_own_shard().await.unwrap(); + assert_eq!( + shard_writer_epoch(&recovered).await, + epoch, + "live writer must be reused" + ); + assert_eq!(recovered.list(None, None).await.unwrap().len(), 2); + recovered.close().await.unwrap(); + }); + } + #[test] fn flush_is_noop_without_writes() { // flush() with no resident writer / nothing buffered must be a harmless diff --git a/crates/lance-context-core/src/store_base.rs b/crates/lance-context-core/src/store_base.rs index c4766bd..14b8bc4 100644 --- a/crates/lance-context-core/src/store_base.rs +++ b/crates/lance-context-core/src/store_base.rs @@ -81,6 +81,10 @@ pub(crate) const DEFAULT_MANIFEST_SCAN_BATCH_SIZE: usize = 16; /// concurrently while collecting observability metrics. pub(crate) const DEFAULT_OBSERVE_CONCURRENCY: usize = 16; +// Cold WAL replay decodes payloads. Serialize maintenance-triggered recovery +// across tables rather than opening a writer for every cold store at once. +static OWN_WAL_RECOVERY_SLOT: tokio::sync::Semaphore = tokio::sync::Semaphore::const_new(1); + /// Flushed generations folded into the base table by one merge pass, by default. /// /// A merge buffers every row of every generation it takes before appending, and @@ -763,6 +767,61 @@ impl StorageBase { result.map(|_| ()) } + /// Recover a restarted instance's durable WAL before its cleanup merge. + /// Ordinary flush sweeps stay cheap: they only touch resident writers. + /// This path is for this handle's OWN write shard, never a foreign shard; + /// as with `put`, the caller must have exclusive ownership of that writer ID. + /// Opening a writer claims its epoch, so this is not a mechanism for taking + /// over an instance that is still alive. + async fn recover_and_flush_own_wal(&self) -> LanceResult<()> { + if self.is_version_pinned() || self.deleted { + return Ok(()); + } + if self.write_writer.lock().await.is_some() { + return self.flush().await; + } + let manifest_store = ShardManifestStore::new( + self.dataset.object_store(None).await?, + &self.dataset.branch_location().path, + self.write_shard, + DEFAULT_MANIFEST_SCAN_BATCH_SIZE, + ); + // A worker with no own shard must not queue behind another table's + // recovery. Serial master fan-out visits many such workers. + if manifest_store.refresh_latest().await?.is_none() { + return Ok(()); + } + let _recovery_slot = OWN_WAL_RECOVERY_SLOT + .acquire() + .await + .expect("WAL recovery semaphore is never closed"); + { + let mut guard = self.write_writer.lock().await; + // An append may have opened/replayed it while we waited. + if guard.is_none() { + // Manifest WAL-tip hints can lag acknowledged PUTs. An existing + // own shard must be replayed even if its hint says it is empty. + // Lance streams replay and seals along the way. These are soft + // byte ceilings: one indivisible WAL entry can exceed them. + let config = ShardWriterConfig { + shard_id: self.write_shard, + max_memtable_size: 32 * 1024 * 1024, + max_unflushed_memtable_bytes: 128 * 1024 * 1024, + ..Default::default() + }; + info!(shard = %self.write_shard, "recovering own WAL before cleanup merge"); + let writer = self + .dataset + .mem_wal_writer(self.write_shard, config) + .await?; + *guard = Some(Arc::new(writer)); + } + } + // Keep the recovered writer resident so concurrent ingest shares it; + // a later cheap flush can also finish sealing if this caller is cancelled. + self.flush().await + } + async fn flush_inner(&self) -> LanceResult { let writer = { let guard = self.write_writer.lock().await; @@ -871,7 +930,7 @@ impl StorageBase { /// the merge — leaving rows durable but permanently invisible until a /// process restart replayed the WAL. pub async fn cleanup_own_shard(&mut self) -> LanceResult { - self.flush().await?; + self.recover_and_flush_own_wal().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. @@ -994,7 +1053,7 @@ impl StorageBase { } if seal_first { // Materialize anything buffered so it is eligible for this pass. - self.flush().await?; + self.recover_and_flush_own_wal().await?; } let object_store = self.dataset.object_store(None).await?; let branch_location = self.dataset.branch_location(); @@ -2739,3 +2798,25 @@ pub fn derive_shard_id(instance_id: Option<&str>) -> Uuid { let input = instance_id.unwrap_or("default"); Uuid::new_v5(&Uuid::NAMESPACE_OID, input.as_bytes()) } + +#[cfg(test)] +mod cold_recovery_tests { + use super::OWN_WAL_RECOVERY_SLOT; + use crate::RolloutStore; + use std::time::Duration; + + #[tokio::test] + async fn empty_shard_cleanup_does_not_wait_for_other_recovery() { + let dir = tempfile::TempDir::new().unwrap(); + let mut store = RolloutStore::open(dir.path().to_str().unwrap()) + .await + .unwrap(); + let _occupied = OWN_WAL_RECOVERY_SLOT.acquire().await.unwrap(); + let reclaimed = tokio::time::timeout(Duration::from_secs(2), store.cleanup_own_shard()) + .await + .expect("absent own shard must bypass cold recovery admission") + .unwrap(); + assert_eq!(reclaimed, 0); + assert!(store.own_shard_watermark().await.unwrap().is_none()); + } +} From ed7b4d0b1237a6cbc420e47dd59d28434b8f8ff1 Mon Sep 17 00:00:00 2001 From: Beinan Wang Date: Wed, 21 Oct 2026 06:40:47 +0000 Subject: [PATCH 2/4] Reuse normal writer configuration during cold recovery --- crates/lance-context-core/src/store_base.rs | 31 ++++++--------------- 1 file changed, 9 insertions(+), 22 deletions(-) diff --git a/crates/lance-context-core/src/store_base.rs b/crates/lance-context-core/src/store_base.rs index 14b8bc4..2f38570 100644 --- a/crates/lance-context-core/src/store_base.rs +++ b/crates/lance-context-core/src/store_base.rs @@ -795,28 +795,15 @@ impl StorageBase { .acquire() .await .expect("WAL recovery semaphore is never closed"); - { - let mut guard = self.write_writer.lock().await; - // An append may have opened/replayed it while we waited. - if guard.is_none() { - // Manifest WAL-tip hints can lag acknowledged PUTs. An existing - // own shard must be replayed even if its hint says it is empty. - // Lance streams replay and seals along the way. These are soft - // byte ceilings: one indivisible WAL entry can exceed them. - let config = ShardWriterConfig { - shard_id: self.write_shard, - max_memtable_size: 32 * 1024 * 1024, - max_unflushed_memtable_bytes: 128 * 1024 * 1024, - ..Default::default() - }; - info!(shard = %self.write_shard, "recovering own WAL before cleanup merge"); - let writer = self - .dataset - .mem_wal_writer(self.write_shard, config) - .await?; - *guard = Some(Arc::new(writer)); - } - } + // Manifest WAL-tip hints can lag acknowledged PUTs, so an existing + // own shard needs replay even if its hint says it is empty. The normal + // writer open rechecks under its mutex if ingest won the race. Reuse its + // normal replay limits (including Lance's streaming seals); changing + // those here would also change future ingest's generation sizes because + // this writer remains resident. One indivisible entry can exceed the + // configured soft memory ceiling. + info!(shard = %self.write_shard, "recovering own WAL before cleanup merge"); + self.resident_writer().await?; // Keep the recovered writer resident so concurrent ingest shares it; // a later cheap flush can also finish sealing if this caller is cancelled. self.flush().await From cfb359e6cc216ecc8e8609e594a042067b152d76 Mon Sep 17 00:00:00 2001 From: Beinan Wang Date: Wed, 21 Oct 2026 06:43:39 +0000 Subject: [PATCH 3/4] Avoid a duplicate manifest scan for empty worker shards --- crates/lance-context-core/src/store_base.rs | 24 +++++++++++++-------- 1 file changed, 15 insertions(+), 9 deletions(-) diff --git a/crates/lance-context-core/src/store_base.rs b/crates/lance-context-core/src/store_base.rs index 2f38570..b33cec8 100644 --- a/crates/lance-context-core/src/store_base.rs +++ b/crates/lance-context-core/src/store_base.rs @@ -773,12 +773,14 @@ impl StorageBase { /// as with `put`, the caller must have exclusive ownership of that writer ID. /// Opening a writer claims its epoch, so this is not a mechanism for taking /// over an instance that is still alive. - async fn recover_and_flush_own_wal(&self) -> LanceResult<()> { + /// Returns false when there is no own shard to merge. + async fn recover_and_flush_own_wal(&self) -> LanceResult { if self.is_version_pinned() || self.deleted { - return Ok(()); + return Ok(false); } if self.write_writer.lock().await.is_some() { - return self.flush().await; + self.flush().await?; + return Ok(true); } let manifest_store = ShardManifestStore::new( self.dataset.object_store(None).await?, @@ -789,7 +791,7 @@ impl StorageBase { // A worker with no own shard must not queue behind another table's // recovery. Serial master fan-out visits many such workers. if manifest_store.refresh_latest().await?.is_none() { - return Ok(()); + return Ok(false); } let _recovery_slot = OWN_WAL_RECOVERY_SLOT .acquire() @@ -806,7 +808,8 @@ impl StorageBase { self.resident_writer().await?; // Keep the recovered writer resident so concurrent ingest shares it; // a later cheap flush can also finish sealing if this caller is cancelled. - self.flush().await + self.flush().await?; + Ok(true) } async fn flush_inner(&self) -> LanceResult { @@ -917,7 +920,9 @@ impl StorageBase { /// the merge — leaving rows durable but permanently invisible until a /// process restart replayed the WAL. pub async fn cleanup_own_shard(&mut self) -> LanceResult { - self.recover_and_flush_own_wal().await?; + if !self.recover_and_flush_own_wal().await? { + return Ok(0); + } // 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. @@ -1038,9 +1043,10 @@ impl StorageBase { if self.is_version_pinned() || self.deleted { return Ok(None); } - if seal_first { - // Materialize anything buffered so it is eligible for this pass. - self.recover_and_flush_own_wal().await?; + // Materialize anything buffered so it is eligible for this pass. An + // absent own shard has already been checked; do not scan it twice. + if seal_first && !self.recover_and_flush_own_wal().await? { + return Ok(None); } let object_store = self.dataset.object_store(None).await?; let branch_location = self.dataset.branch_location(); From 3e24444797968f5aef3cf611ee911b5b173d5867 Mon Sep 17 00:00:00 2001 From: Beinan Wang Date: Wed, 21 Oct 2026 07:02:44 +0000 Subject: [PATCH 4/4] Keep cold WAL recovery exclusive to ingestion-owner cleanup --- .../lance-context-core/src/generic_store.rs | 71 ++++++++++++++++++- .../lance-context-core/src/rollout_store.rs | 19 ++++- crates/lance-context-core/src/store_base.rs | 38 ++++++---- .../src/routes/generic.rs | 2 +- .../src/routes/rollouts.rs | 2 +- crates/lance-context-server/src/sweeper.rs | 4 +- 6 files changed, 115 insertions(+), 21 deletions(-) diff --git a/crates/lance-context-core/src/generic_store.rs b/crates/lance-context-core/src/generic_store.rs index 6862fdb..b966499 100644 --- a/crates/lance-context-core/src/generic_store.rs +++ b/crates/lance-context-core/src/generic_store.rs @@ -700,6 +700,15 @@ impl GenericStore { self.base.prepare_cleanup_merge().await } + /// Recover this instance's unsealed WAL, then prepare cleanup. The caller + /// must exclusively own the ingestion writer ID, as when appending. Never + /// use this to merge a live worker's shard from a maintenance master. + pub async fn prepare_recovered_cleanup_merge( + &self, + ) -> LanceResult> { + self.base.prepare_recovered_cleanup_merge().await + } + /// Commit a merge prepared by [`Self::prepare_cleanup_merge`]. pub async fn commit_prepared_merge( &mut self, @@ -1383,6 +1392,55 @@ mod tests { }); } + #[tokio::test] + async fn cold_maintenance_cleanup_does_not_claim_a_live_ingestion_epoch() { + let dir = TempDir::new().unwrap(); + let uri = dir.path().to_str().unwrap(); + let options = GenericStoreOptions { + shard_id: Some("live-ingestion-writer".into()), + ..Default::default() + }; + let writer = GenericStore::open(uri, spec(), options.clone()) + .await + .unwrap(); + writer + .add(&[row(json!({"id":"shared", "score":1.0}))]) + .await + .unwrap(); + writer.flush().await.unwrap(); + let before = writer + .own_shard_watermark() + .await + .unwrap() + .unwrap() + .writer_epoch; + let mut maintenance = GenericStore::open_existing(uri, options).await.unwrap(); + let (manifests, manifest, prepared) = + maintenance.prepare_cleanup_merge().await.unwrap().unwrap(); + maintenance + .commit_prepared_merge(&manifests, &manifest, prepared) + .await + .unwrap(); + writer + .add(&[row(json!({"id":"shared", "score":2.0}))]) + .await + .unwrap(); + writer.flush().await.unwrap(); + assert_eq!( + writer + .own_shard_watermark() + .await + .unwrap() + .unwrap() + .writer_epoch, + before + ); + assert_eq!( + writer.get("shared", None).await.unwrap().unwrap()["score"], + json!(2.0) + ); + } + #[test] fn cleanup_recovers_own_wal_updates_without_new_ingest_after_runtime_exit() { let dir = TempDir::new().unwrap(); @@ -1416,7 +1474,18 @@ mod tests { recovered.get("shared", None).await.unwrap().unwrap()["score"], json!(1.0) ); - assert!(recovered.cleanup_wal().await.unwrap() > 0); + let (manifests, manifest, prepared) = recovered + .prepare_recovered_cleanup_merge() + .await + .unwrap() + .expect("own WAL must recover without a new append"); + assert!( + recovered + .commit_prepared_merge(&manifests, &manifest, prepared) + .await + .unwrap() + > 0 + ); assert_eq!( recovered.get("shared", None).await.unwrap().unwrap()["score"], json!(2.0) diff --git a/crates/lance-context-core/src/rollout_store.rs b/crates/lance-context-core/src/rollout_store.rs index f79e655..1be83be 100644 --- a/crates/lance-context-core/src/rollout_store.rs +++ b/crates/lance-context-core/src/rollout_store.rs @@ -702,6 +702,15 @@ impl RolloutStore { self.base.prepare_cleanup_merge().await } + /// Recover this instance's unsealed WAL, then prepare cleanup. The caller + /// must exclusively own the ingestion writer ID, as when appending. Never + /// use this to merge a live worker's shard from a maintenance master. + pub async fn prepare_recovered_cleanup_merge( + &self, + ) -> LanceResult> { + self.base.prepare_recovered_cleanup_merge().await + } + /// Commit a merge prepared by [`Self::prepare_merge_if_ready`]. pub async fn commit_prepared_merge( &mut self, @@ -4714,7 +4723,7 @@ mod tests { let runtime = tokio::runtime::Runtime::new().unwrap(); runtime.block_on(async { - let mut other = RolloutStore::open_with_options( + let other = RolloutStore::open_with_options( &uri, RolloutStoreOptions { shard_id: Some("different-live-writer".into()), @@ -4723,7 +4732,11 @@ mod tests { ) .await .unwrap(); - assert_eq!(other.cleanup_own_shard().await.unwrap(), 0); + assert!(other + .prepare_recovered_cleanup_merge() + .await + .unwrap() + .is_none()); assert!(other.own_shard_watermark().await.unwrap().is_none()); assert!(other .get_by_id("durable-before-exit") @@ -4742,7 +4755,7 @@ mod tests { .is_none()); // Exercise the split-lock route used by the actual HTTP merge. let (manifests, manifest, prepared) = recovered - .prepare_cleanup_merge() + .prepare_recovered_cleanup_merge() .await .unwrap() .expect("cleanup must replay acknowledged but unsealed WAL"); diff --git a/crates/lance-context-core/src/store_base.rs b/crates/lance-context-core/src/store_base.rs index b33cec8..ed29a90 100644 --- a/crates/lance-context-core/src/store_base.rs +++ b/crates/lance-context-core/src/store_base.rs @@ -920,9 +920,7 @@ impl StorageBase { /// the merge — leaving rows durable but permanently invisible until a /// process restart replayed the WAL. pub async fn cleanup_own_shard(&mut self) -> LanceResult { - if !self.recover_and_flush_own_wal().await? { - return Ok(0); - } + 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. @@ -1034,6 +1032,19 @@ impl StorageBase { self.prepare_merge_if_ready_inner(1, true).await } + /// Recover and seal this instance's durable WAL before preparing cleanup. + /// The caller MUST own the ingestion writer ID exclusively. A maintenance + /// master merging a live worker's sealed shard must use the ordinary + /// preparation methods, which never acquire an ingestion epoch. + pub async fn prepare_recovered_cleanup_merge( + &self, + ) -> LanceResult> { + if !self.recover_and_flush_own_wal().await? { + return Ok(None); + } + self.prepare_merge_if_ready_inner(1, false).await + } + async fn prepare_merge_if_ready_inner( &self, threshold: usize, @@ -1043,10 +1054,8 @@ impl StorageBase { if self.is_version_pinned() || self.deleted { return Ok(None); } - // Materialize anything buffered so it is eligible for this pass. An - // absent own shard has already been checked; do not scan it twice. - if seal_first && !self.recover_and_flush_own_wal().await? { - return Ok(None); + if seal_first { + self.flush().await?; } let object_store = self.dataset.object_store(None).await?; let branch_location = self.dataset.branch_location(); @@ -2801,15 +2810,18 @@ mod cold_recovery_tests { #[tokio::test] async fn empty_shard_cleanup_does_not_wait_for_other_recovery() { let dir = tempfile::TempDir::new().unwrap(); - let mut store = RolloutStore::open(dir.path().to_str().unwrap()) + let store = RolloutStore::open(dir.path().to_str().unwrap()) .await .unwrap(); let _occupied = OWN_WAL_RECOVERY_SLOT.acquire().await.unwrap(); - let reclaimed = tokio::time::timeout(Duration::from_secs(2), store.cleanup_own_shard()) - .await - .expect("absent own shard must bypass cold recovery admission") - .unwrap(); - assert_eq!(reclaimed, 0); + let prepared = tokio::time::timeout( + Duration::from_secs(2), + store.prepare_recovered_cleanup_merge(), + ) + .await + .expect("absent own shard must bypass cold recovery admission") + .unwrap(); + assert!(prepared.is_none()); assert!(store.own_shard_watermark().await.unwrap().is_none()); } } diff --git a/crates/lance-context-server/src/routes/generic.rs b/crates/lance-context-server/src/routes/generic.rs index 287ba02..eaece28 100644 --- a/crates/lance-context-server/src/routes/generic.rs +++ b/crates/lance-context-server/src/routes/generic.rs @@ -338,7 +338,7 @@ pub(crate) async fn merge_generic_wal_owned( let prepared = { let guard = store.read().await; guard - .prepare_cleanup_merge() + .prepare_recovered_cleanup_merge() .await .map_err(AppError::from_lance)? }; diff --git a/crates/lance-context-server/src/routes/rollouts.rs b/crates/lance-context-server/src/routes/rollouts.rs index 1b00fc1..6e01e72 100644 --- a/crates/lance-context-server/src/routes/rollouts.rs +++ b/crates/lance-context-server/src/routes/rollouts.rs @@ -756,7 +756,7 @@ pub(crate) async fn merge_wal_owned( let store = store_lock.read().await; ::metrics::histogram!("rollout_wal_merge_lock_wait_seconds") .record(lock_start.elapsed().as_secs_f64()); - match store.prepare_cleanup_merge().await { + match store.prepare_recovered_cleanup_merge().await { Ok(prepared) => prepared, Err(e) => { ::metrics::counter!("rollout_wal_cleanup_total", "result" => "failed").increment(1); diff --git a/crates/lance-context-server/src/sweeper.rs b/crates/lance-context-server/src/sweeper.rs index 783980e..3ca5690 100644 --- a/crates/lance-context-server/src/sweeper.rs +++ b/crates/lance-context-server/src/sweeper.rs @@ -116,7 +116,7 @@ impl Sweepable for Arc> { let prepared = { let guard = self.read().await; guard - .prepare_cleanup_merge() + .prepare_recovered_cleanup_merge() .await .map_err(|e| e.to_string())? }; @@ -205,7 +205,7 @@ impl Sweepable for Arc> { let prepared = { let guard = self.read().await; guard - .prepare_cleanup_merge() + .prepare_recovered_cleanup_merge() .await .map_err(|e| e.to_string())? };