fix(master): let maintenance executors opt out of WAL merge claims

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(),

crates/lance-context-master/src/scheduler.rs

@@ -1154,7 +1154,8 @@ pub fn spawn_scheduler(state: &Arc<MasterState>) -> 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<MasterState>) -> 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<MasterState>) -> 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"]

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