From 6e882f6ca05a0839d2adf695811455ecd2919a76 Mon Sep 17 00:00:00 2001 From: Beinan Wang Date: Wed, 21 Oct 2026 12:47:03 +0000 Subject: [PATCH] fix(master): allow maintenance executors to opt out of WAL claims --- crates/lance-context-master/src/config.rs | 18 +++++ crates/lance-context-master/src/scheduler.rs | 70 +++++++++++++++++++- docs/operations/native-maintenance.md | 12 ++++ 3 files changed, 97 insertions(+), 3 deletions(-) diff --git a/crates/lance-context-master/src/config.rs b/crates/lance-context-master/src/config.rs index 3a86dc31..b13b24c4 100644 --- a/crates/lance-context-master/src/config.rs +++ b/crates/lance-context-master/src/config.rs @@ -321,11 +321,28 @@ mod tests { assert_eq!(config.compaction_max_source_fragments, 32); assert_eq!(config.compaction_max_bytes_per_file, 1024 * 1024 * 1024); } + + #[test] + fn wal_claims_default_on_and_can_be_explicitly_disabled() { + let default = MasterConfig::try_parse_from(["lance-context-master"]).unwrap(); + assert!(default.maintenance.claim_wal_merges); + assert!(MaintenanceConfig::default().claim_wal_merges); + let disabled = + MasterConfig::try_parse_from(["lance-context-master", "--claim-wal-merges=false"]) + .unwrap(); + assert!(!disabled.maintenance.claim_wal_merges); + } } /// Admission and progress deadlines for local master table mutations. #[derive(Debug, Clone, clap::Args)] pub struct MaintenanceConfig { + /// Allow this master's scheduler to claim queued MergeWal tasks. Set false + /// for maintenance-only executors; Compact, IndexId and Repair still run. + /// This is independent of enqueue intervals and concurrency budgets, and + /// is not a runtime drain or proof that a predecessor's writes have ended. + #[arg(long, env = "MASTER_CLAIM_WAL_MERGES", default_value_t = true, action = clap::ArgAction::Set)] + pub claim_wal_merges: bool, /// 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. @@ -367,6 +384,7 @@ pub struct MaintenanceConfig { impl Default for MaintenanceConfig { fn default() -> Self { Self { + claim_wal_merges: true, fence_legacy_maintenance: false, compaction_prepare_targets: Vec::new(), index_prepare_targets: Vec::new(), diff --git a/crates/lance-context-master/src/scheduler.rs b/crates/lance-context-master/src/scheduler.rs index 0a67b7df..95a452f8 100644 --- a/crates/lance-context-master/src/scheduler.rs +++ b/crates/lance-context-master/src/scheduler.rs @@ -1154,7 +1154,8 @@ pub fn spawn_scheduler(state: &Arc) -> JoinHandle<()> { // WAL merge draws from its own budget so a backlog of slow HTTP fan-outs // cannot occupy every general slot and starve compaction. `0` opts back // into the shared pool. - let merge_sem = (state.config.merge_wal_concurrency > 0) + let claim_merges = state.config.maintenance.claim_wal_merges; + let merge_sem = (claim_merges && state.config.merge_wal_concurrency > 0) .then(|| Arc::new(Semaphore::new(state.config.merge_wal_concurrency))); // One poller per pool, each claiming only the kinds its pool runs. @@ -1179,7 +1180,7 @@ pub fn spawn_scheduler(state: &Arc) -> JoinHandle<()> { let general = spawn_pool_poller( state.clone(), sem.clone(), - if merge_sem.is_some() { + if !claim_merges || merge_sem.is_some() { TaskKinds::GENERAL.without_compact() } else { // No separate merge budget: the general pool runs everything, so it @@ -1192,7 +1193,8 @@ pub fn spawn_scheduler(state: &Arc) -> JoinHandle<()> { if let Some(merge) = merge_sem { spawn_pool_poller(state.clone(), merge, TaskKinds::MERGE_WAL, false); } - if crate::local_merge::enabled(&state.config) + if claim_merges + && crate::local_merge::enabled(&state.config) && state.config.append.rollout_append_local_task_concurrency > 0 { // Reserve before claiming. A saturated legacy RPC pool cannot consume @@ -3259,6 +3261,68 @@ mod tests { assert_eq!(state.task_store.list().await.unwrap().len(), 1); } + /// Maintenance-only masters leave merge tasks available to their peers. + #[tokio::test] + #[ignore = "requires ETCD_TEST_ENDPOINTS"] + async fn disabled_merge_claims_preserve_maintenance_and_other_executor_admission() { + // Cover the general-pool fallback, dedicated RPC pool and resident + // pool. A zero concurrency budget alone used to claim MergeWal too. + for (merge_concurrency, resident) in [(0, false), (2, false), (2, true)] { + let dir = TempDir::new().unwrap(); + let mut cfg = config(&dir); + cfg.maintenance.claim_wal_merges = false; + cfg.merge_wal_concurrency = merge_concurrency; + if resident { + cfg.local_merge.master_local_merge_targets = vec!["exp".into()]; + cfg.catchup.shards = vec!["writer".into()]; + cfg.catchup.slice_secs = 2; + cfg.catchup.merge_max_bytes = 1024 * 1024; + } + let state = MasterState::new(cfg).await.unwrap(); + for name in ["exp", "other"] { + let mut store = RolloutStore::open(&state.rollout_uri(name)).await.unwrap(); + store.add(&[rollout_record("row")]).await.unwrap(); + store.cleanup_own_shard().await.unwrap(); + } + let merge = enqueue(&state, TaskKind::MergeWal, "exp").await.unwrap(); + let compact = enqueue(&state, TaskKind::Compact, "exp").await.unwrap(); + let index = enqueue(&state, TaskKind::IndexId, "other").await.unwrap(); + let scheduler = spawn_scheduler(&state); + for task in [&compact, &index] { + let terminal = await_terminal(&state, &task.id).await; + assert_eq!(terminal.state, TaskState::Done, "{terminal:?}"); + } + // Observe more than one poll interval, then close admission and + // join admitted work before checking the durable claim state. + tokio::time::sleep(Duration::from_millis(1100)).await; + state.admission.begin_drain(); + tokio::time::timeout(Duration::from_secs(10), state.admission.wait_drained()) + .await + .unwrap(); + scheduler.abort(); + let queued = state.task_store.get(&merge.id).await.unwrap().unwrap(); + assert_eq!(queued.state, TaskState::Queued, "{queued:?}"); + assert!( + queued.started_at.is_none(), + "disabled executor claimed merge" + ); + // The task and its target remain available to an enabled peer; + // the setting is local dispatch policy, not a global queue pause. + let peer = state + .task_store + .claim_next_of_kinds(TaskKinds::MERGE_WAL) + .await + .unwrap() + .unwrap(); + assert_eq!(peer.task.id, merge.id); + state + .task_store + .finish(peer, Ok("peer accepted".into())) + .await + .unwrap(); + } + } + /// A MergeWal task with no configured endpoints fails fast with a clear msg. #[tokio::test] #[ignore = "requires ETCD_TEST_ENDPOINTS"] diff --git a/docs/operations/native-maintenance.md b/docs/operations/native-maintenance.md index d235916a..9dee7a94 100644 --- a/docs/operations/native-maintenance.md +++ b/docs/operations/native-maintenance.md @@ -1,5 +1,17 @@ # Maintenance beside persistent native catch-up owners +For a master assigned only to Compact, IndexId and Repair, set +`MASTER_CLAIM_WAL_MERGES=false`. The default is `true`. This excludes queued +MergeWal work from its general, dedicated merge and resident merge pollers; +it leaves that work queued for other masters. It does not stop enqueue sweeps, +cancel accepted operations or change ownership/fencing. Set enqueue intervals +separately when the executor should not generate new periodic work. + +`MERGE_WAL_INTERVAL_SECS=0` only disables periodic enqueueing, not claiming. +`MERGE_WAL_CONCURRENCY=0` shares the general pool and also does not disable +claiming. Changing the claim setting requires a new process; retiring an older +process still requires its supported drain and storage-quiescence evidence. + `MAINTENANCE_CATCHUP_TARGETS` is an explicit comma-separated list of rollout targets. It defaults to empty and does not support `*`. A target must also be owned by the merge protocol, non-generic, and not draining. Enable it only