feat: run owned WAL catch-up inside native masters

crates/lance-context-core/src/rollout_append.rs

@@ -360,6 +360,37 @@ pub(crate) async fn reconcile_shard(dataset: &Dataset, shard: Uuid) -> Result<us
     Ok(count)
 }
 
+/// Metadata-only WAL demand probe for rollout and generic tables. This opens no
+/// writer, evolves no schema, and reads no generation payloads.
+pub async fn has_sealed_wal(
+    uri: &str,
+    session: Option<Arc<Session>>,
+    shards: &[String],
+) -> Result<bool> {
+    let dataset = StorageBase::load_with_options(uri, None, session).await?;
+    dataset_has_sealed_wal(&dataset, shards).await
+}
+
+async fn dataset_has_sealed_wal(dataset: &Dataset, shards: &[String]) -> Result<bool> {
+    if shards.is_empty() || shards.len() > 256 {
+        return Err(Error::invalid_input(
+            "expected 1..=256 WAL shard identities",
+        ));
+    }
+    let mut seen = HashSet::new();
+    for name in shards {
+        let id = derive_shard_id(Some(name));
+        if seen.insert(id) {
+            if let Some(manifest) = shard_store(dataset, id).await?.refresh_latest().await? {
+                if !manifest.sstables.is_empty() {
+                    return Ok(true);
+                }
+            }
+        }
+    }
+    Ok(false)
+}
+
 /// The only publisher. Open this inside the existing MergeWriteScope so both
 /// table commits and shard drains retain ownership checks and version fencing.
 pub struct AppendCoordinator {
@@ -395,27 +426,7 @@ impl AppendCoordinator {
     /// No writer opens, raw WAL replay, generation payload reads, or publication.
     /// Committed-but-undrained generations also need a maintenance pass.
     pub async fn has_sealed_wal(&self, shards: &[String]) -> Result<bool> {
-        if shards.is_empty() || shards.len() > 256 {
-            return Err(Error::invalid_input(
-                "expected 1..=256 WAL shard identities",
-            ));
-        }
-        let mut seen = HashSet::new();
-        for name in shards {
-            let id = derive_shard_id(Some(name));
-            if seen.insert(id) {
-                if let Some(manifest) = shard_store(&self.dataset, id)
-                    .await?
-                    .refresh_latest()
-                    .await?
-                {
-                    if !manifest.sstables.is_empty() {
-                        return Ok(true);
-                    }
-                }
-            }
-        }
-        Ok(false)
+        dataset_has_sealed_wal(&self.dataset, shards).await
     }
 
     pub async fn plan(

crates/lance-context-master/src/catchup/executor.rs

@@ -120,16 +120,37 @@ async fn merge_passes(state: &Arc<MasterState>, target: &str) -> Result<String>
     if state.config.append.enabled(target) {
         return crate::rollout_append::run(state, target).await;
     }
-    let config = &state.config.catchup;
     let session = RolloutStore::build_session(96 * 1024 * 1024, 32 * 1024 * 1024);
-    let budget = MergeMemoryBudget::new(config.merge_memory_bytes);
+    let budget = MergeMemoryBudget::new(state.config.catchup.merge_memory_bytes);
+    merge_local(state, target, session, budget).await
+}
+
+/// Same bounded storage implementation for resident masters and dedicated legacy executors.
+/// Resources are supplied by the caller so resident tasks share one process budget.
+pub(crate) async fn merge_local(
+    state: &Arc<MasterState>,
+    target: &str,
+    session: Arc<lance::session::Session>,
+    budget: Arc<MergeMemoryBudget>,
+) -> Result<String> {
+    let config = &state.config.catchup;
     let mut total = 0usize;
+    // Legacy supervisors consume these progress messages and task details.
+    let dedicated = config.target.is_some();
+    let detail = |total| {
+        let prefix = if dedicated {
+            "dedicated catch-up merged"
+        } else {
+            "local WAL merge reclaimed"
+        };
+        format!("{prefix} {total} generations")
+    };
     let started = tokio::time::Instant::now();
     // This is only an admission time slice. A progressing operation already
     // admitted may run beyond it; cancellation is based on lack of progress.
     let admit_for = Duration::from_secs(config.slice_secs);
-    // Bound each slice even under continuous ingestion; later fresh stats may
-    // request another Job. Visit every shard before repeating any hot shard.
+    // Bound each slice even under continuous ingestion. Resident masters publish
+    // continuation demand before releasing their claim; no full scan is required.
     for _ in 0..16 {
         let before = total;
         total += run_pipeline(&config.shards, config.pipeline_enabled, |shard| {
@@ -150,18 +171,37 @@ async fn merge_passes(state: &Arc<MasterState>, target: &str) -> Result<String>
                     let commit_secs = commit_started.elapsed().as_secs_f64();
                     metrics::counter!("master_catchup_generations_reclaimed_total")
                         .increment(reclaimed as u64);
-                    tracing::info!(%target, %shard, reclaimed, prepare_secs, commit_secs,
-                        "dedicated catch-up committed");
+                    if dedicated {
+                        tracing::info!(%target, %shard, reclaimed, prepare_secs, commit_secs,
+                            "dedicated catch-up committed");
+                    } else {
+                        tracing::info!(%target, %shard, reclaimed, prepare_secs, commit_secs,
+                            "local WAL merge committed");
+                    }
                     Ok(reclaimed)
                 }) as BoxFuture<'_, Result<usize>>)
             }
         })
         .await?;
-        if total == before {
+        if total == before && (total == 0 || started.elapsed() < admit_for) {
+            // Empty metadata reads can consume the admission slice too. They
+            // must not create a self-sustaining chain of zero-work tasks;
+            // bounded resident discovery will reassert any remaining demand.
+            return Ok(detail(total));
+        }
+        if started.elapsed() >= admit_for {
             break;
         }
     }
-    Ok(format!("dedicated catch-up merged {total} generations"))
+    if crate::local_merge::selected(&state.config, target) && config.target.is_none() {
+        state
+            .task_store
+            .merge_coordinator()
+            .request_merge(target)
+            .await?;
+        metrics::counter!("master_resident_merge_continuations_total").increment(1);
+    }
+    Ok(detail(total))
 }
 
 /// Read sealed WAL without acquiring a writer epoch. The returned future owns

crates/lance-context-master/src/catchup/mod.rs

@@ -12,6 +12,7 @@ use axum::{
     extract::{Query, State},
     Json,
 };
+pub(crate) use executor::merge_local;
 pub use executor::{execute, ExecuteOutcome};
 use kubernetes::Kubernetes;
 use serde::{Deserialize, Serialize};

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

@@ -15,6 +15,8 @@ pub struct MasterConfig {
     #[command(flatten)]
     pub append: crate::rollout_append::AppendConfig,
     #[command(flatten)]
+    pub local_merge: crate::local_merge::LocalMergeConfig,
+    #[command(flatten)]
     pub wal_tail: crate::wal_tail::WalTailConfig,
     #[command(flatten)]
     pub planner: crate::planner::PlannerConfig,

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

@@ -17,8 +17,7 @@ use lance_context_merge::rollout::MergeRollout;
 pub enum Skip {
     /// The poller's kind set does not include this task kind.
     KindNotRunnable,
-    /// A resident-only poller saw a task that is not a local append merge
-    /// on an owned, non-draining, opted-in rollout target.
+    /// A resident-only poller saw a task outside the owned, non-draining local scope.
     NotResidentEligible,
     /// A prepared compaction or index is waiting for its commit turn on
     /// this table; merge yields one admission turn.
@@ -90,10 +89,10 @@ pub fn pre_dependency_filter(
     }
     if let Some(targets) = resident_targets {
         let eligible = kind == TaskKind::MergeWal
-            && !is_generic(target)
             && policy.rollout.owned(target)
             && !policy.rollout.draining(target)
-            && listed(targets, target);
+            && (targets.iter().any(|t| t == target)
+                || (!is_generic(target) && listed(targets, target)));
         if !eligible {
             return Err(Skip::NotResidentEligible);
         }
@@ -232,10 +231,7 @@ mod tests {
             check(TaskKind::MergeWal, "excluded"),
             Err(Skip::NotResidentEligible)
         );
-        assert_eq!(
-            check(TaskKind::MergeWal, "generic:gs"),
-            Err(Skip::NotResidentEligible)
-        );
+        assert_eq!(check(TaskKind::MergeWal, "generic:gs"), Ok(()));
         assert_eq!(check(TaskKind::Compact, "hot"), Err(Skip::KindNotRunnable));
         let wildcard = strings(&["*"]);
         assert_eq!(

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

@@ -213,7 +213,7 @@ pub(crate) fn local_heartbeat(state: &MasterState) -> Heartbeat {
         cfg.merge_wal_concurrency
     } else {
         cfg.task_concurrency.max(1)
-    } + if cfg.append.rollout_append_local {
+    } + if crate::local_merge::enabled(cfg) {
         cfg.append.rollout_append_local_task_concurrency
     } else {
         0
@@ -222,7 +222,7 @@ pub(crate) fn local_heartbeat(state: &MasterState) -> Heartbeat {
         kinds.push(TaskKind::MergeWal);
         slots_total.insert(TaskKind::MergeWal, merge_slots as u32);
     }
-    let bytes_total = if cfg.append.rollout_append_local {
+    let bytes_total = if crate::local_merge::enabled(cfg) {
         cfg.append.rollout_append_local_memory_bytes as u64
     } else {
         0

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

@@ -11,6 +11,7 @@ pub mod discovery;
 pub mod eligibility;
 pub mod error;
 pub mod executors;
+pub mod local_merge;
 mod maintenance_execution;
 mod merge_execution;
 mod resident_recovery;

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

@@ -0,0 +1,137 @@
+//! Native maintenance execution inside each master, sharing its local buffer budget.
+use crate::{config::MasterConfig, state::MasterState};
+use std::sync::Arc;
+
+#[derive(Clone, Debug, Default, clap::Args)]
+pub struct LocalMergeConfig {
+    /// Execute sealed WAL merges in the master, including generic upsert tables.
+    /// '*' selects only explicitly owned targets; it never migrates legacy writers.
+    #[arg(long, env = "MASTER_LOCAL_MERGE_TARGETS", value_delimiter = ',')]
+    pub master_local_merge_targets: Vec<String>,
+}
+
+impl LocalMergeConfig {
+    pub(crate) fn selected(&self, target: &str) -> bool {
+        self.master_local_merge_targets
+            .iter()
+            .any(|t| t == "*" || t == target)
+    }
+}
+
+pub(crate) fn enabled(config: &MasterConfig) -> bool {
+    config.append.rollout_append_local || !config.local_merge.master_local_merge_targets.is_empty()
+}
+
+pub(crate) fn selected(config: &MasterConfig, target: &str) -> bool {
+    config.merge_rollout.owned(target)
+        && !config.merge_rollout.draining(target)
+        && (config.append.local(target) || config.local_merge.selected(target))
+}
+
+/// Resolve the local selection before claiming. In particular an append wildcard
+/// cannot accidentally reserve a slot for a generic table that still needs RPCs.
+pub(crate) fn targets(config: &MasterConfig) -> Vec<String> {
+    config
+        .merge_rollout
+        .owned_targets
+        .iter()
+        .filter(|target| selected(config, target))
+        .cloned()
+        .collect()
+}
+
+pub(crate) fn validate(config: &MasterConfig) -> Result<(), String> {
+    if config.local_merge.master_local_merge_targets.is_empty() {
+        return Ok(());
+    }
+    let catchup = &config.catchup;
+    if catchup.enabled || catchup.pod_template.is_some() || catchup.target.is_some() {
+        return Err("master local merges cannot be combined with Kubernetes catch-up mode".into());
+    }
+    if catchup.shards.is_empty()
+        || catchup.shards.len() > 256
+        || catchup.shards.iter().any(|s| s.is_empty())
+        || catchup.slice_secs == 0
+        || catchup.merge_max_bytes == 0
+        || config.append.rollout_append_local_memory_bytes < catchup.merge_max_bytes
+        || config.merge_wal_concurrency == 0
+    {
+        return Err("master local merges require stable shards, bounded passes and a sufficient shared memory budget".into());
+    }
+    if targets(config)
+        .iter()
+        .any(|target| config.append.enabled(target))
+    {
+        config.append.validate()?;
+        if config.append.rollout_append_local_memory_bytes < config.append.rollout_append_max_bytes
+        {
+            return Err("local append batch exceeds shared memory budget".into());
+        }
+    }
+    for target in &config.local_merge.master_local_merge_targets {
+        if target != "*" && !selected(config, target) {
+            return Err(format!(
+                "master local merge requires a non-draining owned target: {target}"
+            ));
+        }
+    }
+    Ok(())
+}
+
+/// The caller owns the canonical task and maintenance fence. No private queue,
+/// Pod, writer epoch, or independent per-table memory budget is created here.
+pub(crate) async fn run(state: &Arc<MasterState>, target: &str) -> Result<String, String> {
+    if !selected(&state.config, target) {
+        return Err("target is not configured for master local merge".into());
+    }
+    let local = state
+        .local_append
+        .as_ref()
+        .ok_or("missing shared local merge resources")?;
+    crate::catchup::merge_local(state, target, local.session.clone(), local.budget.clone()).await
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use clap::Parser;
+
+    #[test]
+    fn local_selection_preserves_legacy_and_append_boundaries() {
+        let mut config = MasterConfig::parse_from(["test"]);
+        config.merge_rollout.owned_targets = vec!["rollout".into(), "generic:rows".into()];
+        config.append.rollout_append_local = true;
+        config.append.rollout_append_targets = vec!["*".into()];
+        assert_eq!(targets(&config), ["rollout"]);
+        config.local_merge.master_local_merge_targets = vec!["*".into()];
+        assert_eq!(targets(&config), ["rollout", "generic:rows"]);
+        assert!(!selected(&config, "legacy"));
+        config.merge_rollout.drain_targets = vec!["generic:rows".into()];
+        assert_eq!(targets(&config), ["rollout"]);
+    }
+
+    #[test]
+    fn local_validation_requires_safe_resources_and_ownership() {
+        let mut config = MasterConfig::parse_from(["test"]);
+        config.local_merge.master_local_merge_targets = vec!["generic:rows".into()];
+        config.catchup.shards = vec!["writer".into()];
+        assert!(validate(&config).is_err());
+        config.merge_rollout.owned_targets = vec!["generic:rows".into()];
+        assert!(validate(&config).is_ok());
+        config.append.rollout_append_local_memory_bytes = config.catchup.merge_max_bytes - 1;
+        assert!(validate(&config).is_err());
+        config.append.rollout_append_local_memory_bytes = config.catchup.merge_max_bytes;
+        config.catchup.enabled = true;
+        assert!(validate(&config).is_err());
+        config.catchup.enabled = false;
+        config.catchup.shards.clear();
+        assert!(validate(&config).is_err());
+        config.catchup.shards = vec!["writer".into()];
+        config.local_merge.master_local_merge_targets = vec!["*".into()];
+        config.merge_rollout.owned_targets.push("rollout".into());
+        config.append.rollout_append_targets = vec!["rollout".into()];
+        config.append.rollout_append_max_bytes =
+            config.append.rollout_append_local_memory_bytes + 1;
+        assert!(validate(&config).is_err());
+    }
+}