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..b21ba7f 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,34 @@ 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) { + 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(); let prefix = format!("/maintenance-test/{}", lance_context_core::generate_id()); @@ -393,12 +431,15 @@ 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; 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()) } @@ -410,6 +451,83 @@ 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 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 = 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 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( + 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. + let mut observed_io = false; + for step in 0..14 { + let store = GenericStore::open_existing(&uri, GenericStoreOptions::default()) + .await + .map_err(|e| e.to_string())?; + let rows = store + .list_from(ListSource::Fragments) + .await + .map_err(|e| e.to_string())?; + assert_eq!(rows.len(), 512); + if step > 3 { + let execution = coordinator.get("table").await?.unwrap(); + 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()) + }), + ) + .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() { @@ -478,13 +596,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 +633,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 +824,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 +896,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