From e8c30325876e916704a60de2eb5d0b9bf202e369 Mon Sep 17 00:00:00 2001 From: Beinan Wang Date: Tue, 20 Oct 2026 22:43:12 +0000 Subject: [PATCH 1/4] Fence master maintenance before worker merge migration --- crates/lance-context-master/src/config.rs | 8 +- .../src/maintenance_execution.rs | 83 +++++++++++++++++-- docs/merge-recovery.md | 22 +++++ 3 files changed, 105 insertions(+), 8 deletions(-) diff --git a/crates/lance-context-master/src/config.rs b/crates/lance-context-master/src/config.rs index 0852416..3a86dc3 100644 --- a/crates/lance-context-master/src/config.rs +++ b/crates/lance-context-master/src/config.rs @@ -323,9 +323,14 @@ mod tests { } } -/// Deadlines for local table mutations on explicitly owned targets. +/// Admission and progress deadlines for local master table mutations. #[derive(Debug, Clone, clap::Args)] pub struct MaintenanceConfig { + /// Fence new master compact/index/repair executions even for rollout and + /// generic targets still using legacy worker merges. This does not migrate + /// worker admission or retroactively fence pre-upgrade writes. + #[arg(long, env = "MASTER_FENCE_LEGACY_MAINTENANCE", default_value_t = false)] + pub fence_legacy_maintenance: bool, /// Prepare rollout compaction files without holding the table write lock. /// Explicit rollout targets or '*'; empty keeps legacy behavior. Updates to /// source fragments invalidate preparation instead of being overwritten. @@ -362,6 +367,7 @@ pub struct MaintenanceConfig { impl Default for MaintenanceConfig { fn default() -> Self { Self { + fence_legacy_maintenance: false, compaction_prepare_targets: Vec::new(), index_prepare_targets: Vec::new(), maintenance_catchup_targets: Vec::new(), diff --git a/crates/lance-context-master/src/maintenance_execution.rs b/crates/lance-context-master/src/maintenance_execution.rs index 2ac2a79..d02b250 100644 --- a/crates/lance-context-master/src/maintenance_execution.rs +++ b/crates/lance-context-master/src/maintenance_execution.rs @@ -34,6 +34,9 @@ pub(crate) async fn enqueue_recovery( }; if !state.config.merge_rollout.owned(&execution.target) && !state.config.merge_rollout.draining(&execution.target) + // A recorded fenced execution needs reconciliation even if the + // enrollment/maintenance flag was disabled after its admission. + && !(execution.protocol == 2 && execution.maintenance.is_some()) { return Ok(()); } @@ -207,13 +210,18 @@ pub(crate) async fn run( where F: Future>, { - if !state.config.merge_rollout.owned(&claim.task.target) { + if !fenced(&state.config, &claim.task.target) { return work.await; } let maintenance = kind(claim.task.kind).ok_or("invalid local maintenance kind")?; run_as(state, claim, maintenance, work).await } +fn fenced(config: &crate::config::MasterConfig, target: &str) -> bool { + config.merge_rollout.owned(target) + || (config.maintenance.fence_legacy_maintenance && !target.starts_with("datagen:")) +} + pub(crate) async fn run_as( state: &Arc, claim: &TaskClaim, @@ -258,7 +266,10 @@ where .start(&reserved) .await? .ok_or("maintenance start lost")?; - let scope = MergeWriteScope::with_pinned_authorizer(Arc::new(Guard { + // These handles are ephemeral to this operation. Observe successful file + // IO as well as logical checkpoints so a long legacy compact/index scan + // does not look idle merely because Lance has not completed a whole phase. + let scope = MergeWriteScope::with_preparation_authorizer(Arc::new(Guard { coordinator: coordinator.clone(), execution: running.clone(), uri: uri.clone(), @@ -380,7 +391,26 @@ mod tests { use clap::Parser; use std::sync::atomic::{AtomicBool, Ordering}; + #[test] + fn legacy_maintenance_fencing_does_not_enroll_worker_merges() { + let default = MasterConfig::parse_from(["master"]); + assert!(!fenced(&default, "new-hot-table")); + let enabled = MasterConfig::parse_from(["master", "--fence-legacy-maintenance"]); + assert!(fenced(&enabled, "new-hot-table")); + assert!(fenced(&enabled, "generic:new-hot-table")); + assert!(!fenced(&enabled, "datagen:legacy")); + assert!(!enabled.merge_rollout.owned("new-hot-table")); + assert!(!enabled.merge_rollout.owned("generic:new-hot-table")); + } + async fn fixture() -> (tempfile::TempDir, Arc) { + configured_fixture(true, false).await + } + + async fn configured_fixture( + owned: bool, + fence_legacy: bool, + ) -> (tempfile::TempDir, Arc) { let dir = tempfile::tempdir().unwrap(); let endpoints = std::env::var("ETCD_TEST_ENDPOINTS").unwrap(); let prefix = format!("/maintenance-test/{}", lance_context_core::generate_id()); @@ -393,7 +423,10 @@ mod tests { "--etcd-prefix", &prefix, ]); - config.merge_rollout.owned_targets = vec!["table".into()]; + if owned { + config.merge_rollout.owned_targets = vec!["table".into()]; + } + config.maintenance.fence_legacy_maintenance = fence_legacy; config.stats_scan_interval_secs = 0; config.compaction_interval_secs = 0; config.merge_wal_interval_secs = 0; @@ -478,13 +511,23 @@ mod tests { #[tokio::test] #[ignore = "requires ETCD_TEST_ENDPOINTS"] async fn stalled_local_kinds_cancel_and_release_without_stopping_other_tables() { + assert_stalled_local_kinds(true, false).await; + } + + #[tokio::test] + #[ignore = "requires ETCD_TEST_ENDPOINTS"] + async fn legacy_stalled_local_kinds_cancel_and_release_without_stopping_other_tables() { + assert_stalled_local_kinds(false, true).await; + } + + async fn assert_stalled_local_kinds(owned: bool, fence_legacy: bool) { struct Active(Arc); impl Drop for Active { fn drop(&mut self) { self.0.store(false, Ordering::SeqCst); } } - let (_dir, state) = fixture().await; + let (_dir, state) = configured_fixture(owned, fence_legacy).await; for kind in [TaskKind::IndexId, TaskKind::Compact, TaskKind::Repair] { let owned = claim(&state, kind).await; let active = Arc::new(AtomicBool::new(true)); @@ -505,7 +548,11 @@ mod tests { .await .unwrap(); }; - let (result, ()) = tokio::join!(result, other); + let (result, ()) = tokio::time::timeout(Duration::from_secs(15), async { + tokio::join!(result, other) + }) + .await + .expect("a no-progress operation must not strand the master"); assert!(result.as_ref().unwrap_err().contains("no-progress")); assert!(!active.load(Ordering::SeqCst)); assert!(state @@ -692,7 +739,17 @@ mod tests { #[tokio::test] #[ignore = "requires ETCD_TEST_ENDPOINTS"] async fn lost_master_index_is_fenced_before_merge_takes_over() { - let (_dir, state) = fixture().await; + assert_lost_master_fenced(true, false).await; + } + + #[tokio::test] + #[ignore = "requires ETCD_TEST_ENDPOINTS"] + async fn legacy_lost_master_index_is_fenced_before_merge_takes_over() { + assert_lost_master_fenced(false, true).await; + } + + async fn assert_lost_master_fenced(owned: bool, fence_legacy: bool) { + let (_dir, state) = configured_fixture(owned, fence_legacy).await; let uri = state.rollout_uri("table"); let store = lance_context_core::RolloutStore::open(&uri).await.unwrap(); let high = store.version() + 1; @@ -754,7 +811,19 @@ mod tests { #[tokio::test] #[ignore = "requires ETCD_TEST_ENDPOINTS"] async fn failed_storage_barrier_keeps_ownership_and_retries_do_not_reset_budget() { - let (_dir, state) = fixture().await; + assert_failed_barrier(true).await; + } + + #[tokio::test] + #[ignore = "requires ETCD_TEST_ENDPOINTS"] + async fn legacy_fenced_records_recover_after_flag_rollback_and_retain_failed_barriers() { + // A previous process admitted this record with fencing enabled; the + // restarted master has neither owned targets nor the legacy flag. + assert_failed_barrier(false).await; + } + + async fn assert_failed_barrier(owned: bool) { + let (_dir, state) = configured_fixture(owned, false).await; let old_claim = claim(&state, TaskKind::Repair).await; let coordinator = state.task_store.merge_coordinator(); let mut old = Execution::new("table", "master:repair", "old-master", 60); diff --git a/docs/merge-recovery.md b/docs/merge-recovery.md index 838f5d8..aed381b 100644 --- a/docs/merge-recovery.md +++ b/docs/merge-recovery.md @@ -164,6 +164,28 @@ Other maintenance tasks retain their existing repair policy. ## Default-off, per-table rollout +### Fencing master maintenance before worker migration + +`MASTER_FENCE_LEGACY_MAINTENANCE=true` opts a master into the existing durable +execution/commit-permit protocol for every new rollout or generic compact, +index, and repair task it executes, including targets absent from +`MERGE_OWNED_TARGETS`. It defaults to false and does not include datagen. +Prepared compaction and index work still runs without table ownership; only +publication takes its existing commit turn and storage fence. + +This closes the master-side untracked-write path without changing worker merge +admission or stopping ingestion. A master cancels an operation only after its +real completed-work sequence stops advancing, drains admitted writes, and +retains ownership until ambiguous commits are fenced. Recovery scans also +re-enqueue these recorded executions after the flag is turned off: disabling +future admission must not strand an existing storage fence. + +This does not make untracked old master/worker/helper writes recoverable or +enroll a table into native merging. Use the migration barriers below for that +transition. Deploy capable masters before enabling this option; measure etcd +commit-authorization overhead and real compaction/index progress in a canary +before expanding it. No worker protocol/configuration change is required. + ### Legacy worker progress evidence Workers advertising `legacy_progress_protocol: 1` expose From b508e27fa114f0cf4826115d942d81432d97256f Mon Sep 17 00:00:00 2001 From: Beinan Wang Date: Tue, 20 Oct 2026 22:46:11 +0000 Subject: [PATCH 2/4] Verify real file IO sustains legacy maintenance beyond idle windows --- .../src/maintenance_execution.rs | 81 ++++++++++++++++++- 1 file changed, 80 insertions(+), 1 deletion(-) diff --git a/crates/lance-context-master/src/maintenance_execution.rs b/crates/lance-context-master/src/maintenance_execution.rs index d02b250..b7561a0 100644 --- a/crates/lance-context-master/src/maintenance_execution.rs +++ b/crates/lance-context-master/src/maintenance_execution.rs @@ -410,6 +410,14 @@ mod tests { async fn configured_fixture( owned: bool, fence_legacy: bool, + ) -> (tempfile::TempDir, Arc) { + fixture_with_idle(owned, fence_legacy, 1).await + } + + async fn fixture_with_idle( + owned: bool, + fence_legacy: bool, + idle: u64, ) -> (tempfile::TempDir, Arc) { let dir = tempfile::tempdir().unwrap(); let endpoints = std::env::var("ETCD_TEST_ENDPOINTS").unwrap(); @@ -431,7 +439,7 @@ mod tests { config.compaction_interval_secs = 0; config.merge_wal_interval_secs = 0; config.maintenance.maintenance_timeout_secs = 60; - config.maintenance.maintenance_idle_timeout_secs = 1; + config.maintenance.maintenance_idle_timeout_secs = idle; config.maintenance.maintenance_drain_timeout_secs = 1; (dir, MasterState::new(config).await.unwrap()) } @@ -443,6 +451,77 @@ mod tests { state.task_store.claim_next().await.unwrap().unwrap() } + #[tokio::test] + #[ignore = "requires ETCD_TEST_ENDPOINTS; ~8 s"] + async fn legacy_maintenance_file_io_progress_survives_multiple_idle_windows() { + use arrow_array::{RecordBatch, RecordBatchIterator, StringArray}; + use arrow_schema::{DataType, Field, Schema}; + use futures::TryStreamExt; + let (_dir, state) = fixture_with_idle(false, true, 3).await; + let uri = state.rollout_uri("table"); + let schema = Arc::new(Schema::new(vec![ + Field::new("id", DataType::Utf8, false), + Field::new("content", DataType::Utf8, false), + ])); + let batch = RecordBatch::try_new( + schema.clone(), + vec![ + Arc::new(StringArray::from_iter_values( + (0..512).map(|n| n.to_string()), + )), + Arc::new(StringArray::from_iter_values( + (0..512).map(|_| "x".repeat(4096)), + )), + ], + ) + .unwrap(); + let mut dataset = + lance_context_core::StorageBase::create_with_options(&uri, schema.clone(), None, None) + .await + .unwrap(); + dataset + .append(RecordBatchIterator::new([Ok(batch)], schema), None) + .await + .unwrap(); + let claim = claim(&state, TaskKind::IndexId).await; + let coordinator = state.task_store.merge_coordinator(); + let result = tokio::time::timeout( + Duration::from_secs(30), + run(&state, &claim, async { + // No logical checkpoint calls: only completed reads through the + // scoped stores may keep this operation alive. Reopen to avoid + // turning later scans into memory-cache hits with no IO. + for step in 0..14 { + let dataset = + lance_context_core::StorageBase::load_with_options(&uri, None, None) + .await + .map_err(|e| e.to_string())?; + let batches = dataset + .scan() + .try_into_stream() + .await + .map_err(|e| e.to_string())? + .try_collect::>() + .await + .map_err(|e| e.to_string())?; + assert_eq!(batches.iter().map(|b| b.num_rows()).sum::(), 512); + if step > 3 { + let execution = coordinator.get("table").await?.unwrap(); + let report = coordinator.work_progress(&execution).await?.unwrap(); + assert!(report["sequence"].as_u64().unwrap() > 0); + } + tokio::time::sleep(Duration::from_millis(500)).await; + } + Ok("file IO kept the operation alive".into()) + }), + ) + .await + .unwrap(); + assert!(result.is_ok(), "{result:?}"); + assert!(coordinator.get("table").await.unwrap().is_none()); + state.task_store.finish(claim, result).await.unwrap(); + } + #[tokio::test] #[ignore = "requires ETCD_TEST_ENDPOINTS"] async fn work_report_heartbeats_do_not_extend_actual_progress_deadline() { From 5f1217613284cc5b7c4661c5348e047c4c4fd178 Mon Sep 17 00:00:00 2001 From: Beinan Wang Date: Tue, 20 Oct 2026 22:46:53 +0000 Subject: [PATCH 3/4] Allow asynchronous progress publication in the file IO regression test --- crates/lance-context-master/src/maintenance_execution.rs | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/crates/lance-context-master/src/maintenance_execution.rs b/crates/lance-context-master/src/maintenance_execution.rs index b7561a0..cd9e0ea 100644 --- a/crates/lance-context-master/src/maintenance_execution.rs +++ b/crates/lance-context-master/src/maintenance_execution.rs @@ -491,6 +491,7 @@ mod tests { // No logical checkpoint calls: only completed reads through the // scoped stores may keep this operation alive. Reopen to avoid // turning later scans into memory-cache hits with no IO. + let mut observed_io = false; for step in 0..14 { let dataset = lance_context_core::StorageBase::load_with_options(&uri, None, None) @@ -507,11 +508,13 @@ mod tests { assert_eq!(batches.iter().map(|b| b.num_rows()).sum::(), 512); if step > 3 { let execution = coordinator.get("table").await?.unwrap(); - let report = coordinator.work_progress(&execution).await?.unwrap(); - assert!(report["sequence"].as_u64().unwrap() > 0); + if let Some(report) = coordinator.work_progress(&execution).await? { + observed_io |= report["sequence"].as_u64().unwrap() > 0; + } } tokio::time::sleep(Duration::from_millis(500)).await; } + assert!(observed_io, "the watchdog must publish completed file IO"); Ok("file IO kept the operation alive".into()) }), ) From e780119c06f667695564ca95e946d97173f64b36 Mon Sep 17 00:00:00 2001 From: Beinan Wang Date: Tue, 20 Oct 2026 22:53:51 +0000 Subject: [PATCH 4/4] test: exercise maintenance IO through public store API --- .../src/maintenance_execution.rs | 73 ++++++++++--------- 1 file changed, 38 insertions(+), 35 deletions(-) diff --git a/crates/lance-context-master/src/maintenance_execution.rs b/crates/lance-context-master/src/maintenance_execution.rs index cd9e0ea..b21ba7f 100644 --- a/crates/lance-context-master/src/maintenance_execution.rs +++ b/crates/lance-context-master/src/maintenance_execution.rs @@ -454,35 +454,43 @@ mod tests { #[tokio::test] #[ignore = "requires ETCD_TEST_ENDPOINTS; ~8 s"] async fn legacy_maintenance_file_io_progress_survives_multiple_idle_windows() { - use arrow_array::{RecordBatch, RecordBatchIterator, StringArray}; - use arrow_schema::{DataType, Field, Schema}; - use futures::TryStreamExt; + use lance_context_core::{ + ColumnSpec, ColumnType, GenericStore, GenericStoreOptions, ListSource, SchemaSpec, + }; let (_dir, state) = fixture_with_idle(false, true, 3).await; let uri = state.rollout_uri("table"); - let schema = Arc::new(Schema::new(vec![ - Field::new("id", DataType::Utf8, false), - Field::new("content", DataType::Utf8, false), - ])); - let batch = RecordBatch::try_new( - schema.clone(), - vec![ - Arc::new(StringArray::from_iter_values( - (0..512).map(|n| n.to_string()), - )), - Arc::new(StringArray::from_iter_values( - (0..512).map(|_| "x".repeat(4096)), - )), - ], + let schema = SchemaSpec::new(vec![ + ( + "id".into(), + ColumnSpec::required(ColumnType::String { large: false }), + ), + ( + "content".into(), + ColumnSpec::required(ColumnType::String { large: false }), + ), + ]); + let mut store = GenericStore::open( + &uri, + schema, + GenericStoreOptions { + merge_after_generations: Some(0), + ..Default::default() + }, ) + .await .unwrap(); - let mut dataset = - lance_context_core::StorageBase::create_with_options(&uri, schema.clone(), None, None) - .await - .unwrap(); - dataset - .append(RecordBatchIterator::new([Ok(batch)], schema), None) - .await - .unwrap(); + let rows: Vec<_> = (0..512) + .map(|n| { + serde_json::json!({"id": n.to_string(), "content": "x".repeat(4096)}) + .as_object() + .unwrap() + .clone() + }) + .collect(); + store.add(&rows).await.unwrap(); + store.flush().await.unwrap(); + store.cleanup_wal().await.unwrap(); + store.close().await.unwrap(); let claim = claim(&state, TaskKind::IndexId).await; let coordinator = state.task_store.merge_coordinator(); let result = tokio::time::timeout( @@ -493,19 +501,14 @@ mod tests { // turning later scans into memory-cache hits with no IO. let mut observed_io = false; for step in 0..14 { - let dataset = - lance_context_core::StorageBase::load_with_options(&uri, None, None) - .await - .map_err(|e| e.to_string())?; - let batches = dataset - .scan() - .try_into_stream() + let store = GenericStore::open_existing(&uri, GenericStoreOptions::default()) .await - .map_err(|e| e.to_string())? - .try_collect::>() + .map_err(|e| e.to_string())?; + let rows = store + .list_from(ListSource::Fragments) .await .map_err(|e| e.to_string())?; - assert_eq!(batches.iter().map(|b| b.num_rows()).sum::(), 512); + assert_eq!(rows.len(), 512); if step > 3 { let execution = coordinator.get("table").await?.unwrap(); if let Some(report) = coordinator.work_progress(&execution).await? {