From 7e0b808821b36a725e3540d5a92b4442bad10230 Mon Sep 17 00:00:00 2001 From: Beinan Wang Date: Thu, 8 Oct 2026 07:09:38 +0000 Subject: [PATCH 1/2] Keep resident WAL catch-up scheduled without external refill --- .../lance-context-core/src/rollout_append.rs | 23 + crates/lance-context-master/src/lib.rs | 1 + .../src/resident_recovery.rs | 499 ++++++++++++++++++ .../src/rollout_append.rs | 24 +- crates/lance-context-master/src/scheduler.rs | 3 +- docs/resident-wal-recovery.md | 52 ++ 6 files changed, 599 insertions(+), 3 deletions(-) create mode 100644 crates/lance-context-master/src/resident_recovery.rs create mode 100644 docs/resident-wal-recovery.md diff --git a/crates/lance-context-core/src/rollout_append.rs b/crates/lance-context-core/src/rollout_append.rs index a8ac9825..26e912b7 100644 --- a/crates/lance-context-core/src/rollout_append.rs +++ b/crates/lance-context-core/src/rollout_append.rs @@ -377,6 +377,29 @@ impl AppendCoordinator { self.dataset.version().version } + /// Read only latest manifests for explicitly configured writer identities. + /// 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 { + 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?.read_latest().await? { + if !manifest.flushed_generations.is_empty() { + return Ok(true); + } + } + } + } + Ok(false) + } + pub async fn plan( &mut self, shards: &[String], diff --git a/crates/lance-context-master/src/lib.rs b/crates/lance-context-master/src/lib.rs index 6157d7a5..5bac303f 100644 --- a/crates/lance-context-master/src/lib.rs +++ b/crates/lance-context-master/src/lib.rs @@ -7,6 +7,7 @@ pub mod discovery; pub mod error; mod maintenance_execution; mod merge_execution; +mod resident_recovery; pub mod rollout_append; pub mod routes; pub mod scanner; diff --git a/crates/lance-context-master/src/resident_recovery.rs b/crates/lance-context-master/src/resident_recovery.rs new file mode 100644 index 00000000..82cac85d --- /dev/null +++ b/crates/lance-context-master/src/resident_recovery.rs @@ -0,0 +1,499 @@ +//! Resident catch-up demand discovery, independent of the full stats scanner. +//! A durable fleet cursor bounds metadata reads even when masters are scaled up. +use crate::{config::MasterConfig, state::MasterState, stats_store::StatRow}; +use etcd_client::{Compare, CompareOp, Txn, TxnOp}; +use futures::{stream, StreamExt}; +use lance_context_api::TaskKind; +use lance_context_core::rollout_append::AppendCoordinator; +use serde::{Deserialize, Serialize}; +use std::{ + hash::{Hash, Hasher}, + sync::Arc, + time::Duration, +}; + +const BATCH: usize = 4; +const PROBE_TIMEOUT: Duration = Duration::from_secs(10); + +#[derive(Default, Serialize, Deserialize)] +struct Cursor { + after: Option, + next_batch_ms: i64, +} + +fn scope(config: &MasterConfig) -> u64 { + // Namespace only, not an ownership token. A Rust hash-version change merely + // restarts discovery; the separate fleet gate still limits total probes. + let mut hash = std::collections::hash_map::DefaultHasher::new(); + for names in [ + &config.merge_rollout.owned_targets, + &config.merge_rollout.drain_targets, + &config.append.rollout_append_targets, + ] { + let mut sorted = names.clone(); + sorted.sort_unstable(); + sorted.dedup(); + sorted.hash(&mut hash); + } + hash.finish() +} + +fn enabled(config: &MasterConfig) -> bool { + config.append.rollout_append_local + && config.append.rollout_append_reconcile_interval_secs > 0 + && config.merge_wal_interval_secs > 0 +} + +fn candidates(config: &MasterConfig, snapshot: &[StatRow], after: Option<&str>) -> Vec { + // Explicitly owned tables do not depend on scanner freshness or existence. + // Wildcard installations additionally use known names from the stats cache; + // its counts are never used as evidence of current WAL demand. + let mut names: Vec<_> = config + .merge_rollout + .owned_targets + .iter() + .chain(config.append.rollout_append_targets.iter()) + .map(String::as_str) + .chain(snapshot.iter().map(|row| row.name.as_str())) + .filter(|name| { + *name != "*" + && config.append.local(name) + && config.merge_rollout.owned(name) + && !config.merge_rollout.draining(name) + && after.is_none_or(|after| *name > after) + }) + .map(str::to_owned) + .collect(); + names.sort_unstable(); + names.dedup(); + names.truncate(BATCH + 1); + names +} + +pub(crate) fn spawn(state: &Arc) { + if !enabled(&state.config) { + return; + } + let weak = Arc::downgrade(state); + tokio::spawn(async move { + loop { + let Some(state) = weak.upgrade() else { return }; + // Also bound etcd waits. Dropping this read-only discovery never + // cancels a writer or authorizes a replacement execution. + match tokio::time::timeout( + Duration::from_secs(25), + tick(&state, chrono::Utc::now().timestamp_millis()), + ) + .await + { + Ok(Ok(_)) => {} + outcome => tracing::warn!(?outcome, "resident WAL demand discovery incomplete"), + } + drop(state); + tokio::time::sleep(Duration::from_secs(5)).await; + } + }); +} + +async fn tick(state: &Arc, now: i64) -> Result { + if !enabled(&state.config) { + return Ok(0); + } + let Some(_admission) = state.admission.try_admit() else { + return Ok(0); + }; + let key = format!( + "{}/resident-wal-discovery", + state.config.etcd.etcd_prefix.trim_end_matches('/') + ); + let cursor_key = format!("{key}/cursor/{:016x}", scope(&state.config)); + let mut client = state.task_store.etcd_client().clone(); + let prior = client + .get(key.as_str(), None) + .await + .map_err(|e| e.to_string())?; + let old = prior.kvs().first(); + let next_batch_ms: i64 = old + .map(|kv| serde_json::from_slice(kv.value())) + .transpose() + .map_err(|e| e.to_string())? + .unwrap_or_default(); + if next_batch_ms > now { + return Ok(0); + } + let previous_cursor = client + .get(cursor_key.as_str(), None) + .await + .map_err(|e| e.to_string())?; + let old_cursor = previous_cursor.kvs().first(); + let cursor: Cursor = old_cursor + .map(|kv| serde_json::from_slice(kv.value())) + .transpose() + .map_err(|e| e.to_string())? + .unwrap_or_default(); + let snapshot = state.stats_cache.read().await.clone(); + if candidates(&state.config, &snapshot, None).is_empty() { + return Ok(0); + } + let mut names = candidates(&state.config, &snapshot, cursor.after.as_deref()); + if names.is_empty() && cursor.after.is_none() { + return Ok(0); + } + let more = names.len() > BATCH; + names.truncate(BATCH); + let next = Cursor { + after: if more { names.last().cloned() } else { None }, + next_batch_ms: now.saturating_add( + state + .config + .append + .rollout_append_reconcile_interval_secs + .clamp(30, 86_400) as i64 + * 1000, + ), + }; + // Reserve before object-store reads. Lost responses/crashes defer one page; + // other replicas cannot duplicate its probes. No lease expiry enables writes. + let compare = old.map_or_else( + || Compare::version(key.as_str(), CompareOp::Equal, 0), + |kv| Compare::mod_revision(key.as_str(), CompareOp::Equal, kv.mod_revision()), + ); + let compare_cursor = old_cursor.map_or_else( + || Compare::version(cursor_key.as_str(), CompareOp::Equal, 0), + |kv| Compare::mod_revision(cursor_key.as_str(), CompareOp::Equal, kv.mod_revision()), + ); + if !client + .txn(Txn::new().when([compare, compare_cursor]).and_then([ + TxnOp::put( + key.as_str(), + serde_json::to_vec(&next.next_batch_ms).map_err(|e| e.to_string())?, + None, + ), + TxnOp::put( + cursor_key.as_str(), + serde_json::to_vec(&next).map_err(|e| e.to_string())?, + None, + ), + ])) + .await + .map_err(|e| e.to_string())? + .succeeded() + { + return Ok(0); + } + let mut probes = stream::iter(names.into_iter().map(|target| async move { + let outcome = tokio::time::timeout(PROBE_TIMEOUT, probe(state, &target)).await; + match outcome { + Ok(Ok(requested)) => requested, + outcome => { + metrics::counter!("master_resident_wal_probes_total", "result" => "failed").increment(1); + tracing::warn!(%target, ?outcome, "resident WAL metadata probe failed; continuing other targets"); + false + } + } + })).buffer_unordered(2); + let mut requested = 0; + while let Some(true_or_false) = probes.next().await { + requested += usize::from(true_or_false); + } + Ok(requested) +} + +async fn probe(state: &Arc, target: &str) -> Result { + if state + .task_store + .get_active_id(TaskKind::MergeWal, target) + .await + .map_err(|e| e.to_string())? + .is_some() + { + return Ok(false); + } + // Merge retry policy lives in the durable per-executor failure ledger, + // not the generic task cooldown (which deliberately excludes MergeWal). + if state + .task_store + .merge_coordinator() + .failure( + target, + lance_context_merge::MaintenanceKind::Catchup.endpoint(), + ) + .await? + .is_some_and(|failure| failure.next_retry_ms > lance_context_merge::failure::now_ms()) + { + return Ok(false); + } + let owner = crate::catchup::store::active_key(&state.config.etcd.etcd_prefix, target); + if !state + .task_store + .etcd_client() + .clone() + .get(owner, None) + .await + .map_err(|e| e.to_string())? + .kvs() + .is_empty() + { + return Ok(false); + } + let coordinator = AppendCoordinator::open( + &state.rollout_uri(target), + // Reuse this master's small local-staging session, never default caches. + state + .local_append + .as_ref() + .map(|local| local.session.clone()), + ) + .await + .map_err(|e| e.to_string())?; + let pending = coordinator + .has_sealed_wal(&state.config.catchup.shards) + .await + .map_err(|e| e.to_string())?; + if pending { + state + .task_store + .merge_coordinator() + .request_merge(target) + .await?; + } + metrics::counter!("master_resident_wal_probes_total", "result" => if pending { "requested" } else { "empty" }).increment(1); + Ok(pending) +} + +#[cfg(test)] +mod tests { + use super::*; + use clap::Parser; + use lance_context_core::{RolloutStore, RolloutStoreOptions}; + + async fn fixture() -> (tempfile::TempDir, Arc) { + let dir = tempfile::tempdir().unwrap(); + let mut config = MasterConfig::parse_from(["test"]); + config.data_dir = dir.path().to_string_lossy().into(); + config.etcd.etcd_endpoints = std::env::var("ETCD_TEST_ENDPOINTS") + .unwrap() + .split(',') + .map(str::to_owned) + .collect(); + config.etcd.etcd_prefix = format!( + "/resident-recovery-test/{}", + lance_context_core::generate_id() + ); + config.append.rollout_append_local = true; + config.append.rollout_append_targets = vec!["*".into()]; + config.append.rollout_append_max_generations = 1; + config.merge_rollout.owned_targets = ["a", "b", "c", "d", "e"].map(str::to_owned).to_vec(); + config.catchup.shards = vec!["writer".into()]; + config.worker_endpoints.clear(); + config.stats_scan_interval_secs = 0; + let state = MasterState::new(config).await.unwrap(); + (dir, state) + } + + async fn write(state: &MasterState, target: &str, rows: usize) -> RolloutStore { + let writer = RolloutStore::open_with_options( + &state.rollout_uri(target), + RolloutStoreOptions { + shard_id: Some("writer".into()), + merge_after_generations: Some(0), + ..Default::default() + }, + ) + .await + .unwrap(); + for id in 0..rows { + let dto = serde_json::from_value( + serde_json::json!({"id":id.to_string(), "rollout_id":"r", "content":"test"}), + ) + .unwrap(); + writer + .add(&[lance_context_core::rollout_record_from_add_request(&dto)]) + .await + .unwrap(); + writer.flush().await.unwrap(); + } + writer + } + + #[tokio::test] + #[ignore = "requires ETCD_TEST_ENDPOINTS"] + async fn replicas_discover_without_stats_and_restart_rotates_past_missing_tables() { + let (_dir, state) = fixture().await; + let _b = write(&state, "b", 1).await; + let _e = write(&state, "e", 1).await; + let second = MasterState::new(state.config.clone()).await.unwrap(); + assert!(state.stats_cache.read().await.is_empty()); + let now = chrono::Utc::now().timestamp_millis(); + let (one, two) = tokio::join!(tick(&state, now), tick(&second, now)); + assert_eq!(one.unwrap() + two.unwrap(), 1); + let coordinator = state.task_store.merge_coordinator(); + assert_eq!( + coordinator + .request_page(None) + .await + .unwrap() + .0 + .iter() + .map(|r| r.target.as_str()) + .collect::>(), + ["b"] + ); + assert_eq!(tick(&second, now + 1000).await.unwrap(), 0); + let restarted = MasterState::new(state.config.clone()).await.unwrap(); + assert_eq!(tick(&restarted, now + 30_000).await.unwrap(), 1); + assert_eq!(coordinator.request_page(None).await.unwrap().0.len(), 2); + } + + #[tokio::test] + #[ignore = "requires ETCD_TEST_ENDPOINTS"] + async fn discovery_preserves_external_owner_and_existing_canonical_task() { + let (_dir, state) = fixture().await; + let _a = write(&state, "a", 1).await; + let _b = write(&state, "b", 1).await; + let owner = crate::catchup::store::active_key(&state.config.etcd.etcd_prefix, "a"); + let mut client = state.task_store.etcd_client().clone(); + client + .put(owner.as_str(), "external-publisher", None) + .await + .unwrap(); + let task = state + .task_store + .enqueue(TaskKind::MergeWal, "b", vec![]) + .await + .unwrap(); + assert!(!probe(&state, "a").await.unwrap()); + assert!(!probe(&state, "b").await.unwrap()); + assert!(state + .task_store + .merge_coordinator() + .request_page(None) + .await + .unwrap() + .0 + .is_empty()); + assert_eq!( + client.get(owner, None).await.unwrap().kvs()[0].value(), + b"external-publisher" + ); + assert_eq!( + state + .task_store + .get_active_id(TaskKind::MergeWal, "b") + .await + .unwrap(), + Some(task.id) + ); + } + + #[tokio::test] + #[ignore = "requires ETCD_TEST_ENDPOINTS"] + async fn discovery_does_not_reset_or_enqueue_through_failure_backoff() { + use crate::task_store::TaskKinds; + let (_dir, state) = fixture().await; + let _a = write(&state, "a", 1).await; + state + .task_store + .enqueue(TaskKind::MergeWal, "a", vec![]) + .await + .unwrap(); + let claim = state + .task_store + .claim_next_of_kinds(TaskKinds::MERGE_WAL) + .await + .unwrap() + .unwrap(); + let coordinator = state.task_store.merge_coordinator(); + coordinator + .record_failure( + &state.task_store.merge_claim(&claim), + "a", + "master:catchup", + "invalid rollout schema", + ) + .await + .unwrap(); + state + .task_store + .finish(claim, Err("invalid rollout schema".into())) + .await + .unwrap(); + let before = coordinator + .failure("a", "master:catchup") + .await + .unwrap() + .unwrap(); + assert!(before.needs_attention); + assert!(!probe(&state, "a").await.unwrap()); + let after = coordinator + .failure("a", "master:catchup") + .await + .unwrap() + .unwrap(); + assert_eq!(after.consecutive_attempts, before.consecutive_attempts); + assert_eq!(after.next_retry_ms, before.next_retry_ms); + assert!(coordinator.request_page(None).await.unwrap().0.is_empty()); + } + + #[tokio::test] + #[ignore = "requires ETCD_TEST_ENDPOINTS"] + async fn bounded_slice_leaves_durable_demand_and_successor_finishes_without_scanner() { + use crate::task_store::TaskKinds; + let (_dir, state) = fixture().await; + let writer = write(&state, "a", 17).await; + state + .task_store + .enqueue(TaskKind::MergeWal, "a", vec![]) + .await + .unwrap(); + let claim = state + .task_store + .claim_next_of_kinds(TaskKinds::MERGE_WAL) + .await + .unwrap() + .unwrap(); + let outcome = crate::merge_execution::run_merge_wal(&state, &claim).await; + assert!(outcome.as_ref().unwrap().contains("16 generations")); + let coordinator = state.task_store.merge_coordinator(); + let request = coordinator.request_page(None).await.unwrap().0.remove(0); + // A consumer visiting the request while this task runs cannot erase it. + crate::scheduler::enqueue_merge_request(&state, &coordinator, &request) + .await + .unwrap(); + assert_eq!(coordinator.request_page(None).await.unwrap().0.len(), 1); + state.task_store.finish(claim, outcome).await.unwrap(); + let successor = MasterState::new(state.config.clone()).await.unwrap(); + crate::scheduler::enqueue_merge_request(&successor, &coordinator, &request) + .await + .unwrap(); + assert!(coordinator.request_page(None).await.unwrap().0.is_empty()); + let next = successor + .task_store + .claim_next_of_kinds(TaskKinds::MERGE_WAL) + .await + .unwrap() + .unwrap(); + let result = crate::merge_execution::run_merge_wal(&successor, &next).await; + assert!(result.as_ref().unwrap().contains("1 generations")); + successor.task_store.finish(next, result).await.unwrap(); + assert_eq!(writer.pending_wal_generations().await.unwrap(), 0); + let reader = lance::Dataset::open(&state.rollout_uri("a")).await.unwrap(); + assert_eq!(reader.count_rows(None).await.unwrap(), 17); + assert!(!probe(&successor, "a").await.unwrap()); + } + + #[test] + fn explicit_owned_targets_rotate_without_any_stats_and_exclude_legacy() { + let mut config = MasterConfig::parse_from(["test"]); + config.append.rollout_append_local = true; + config.append.rollout_append_targets = vec!["*".into()]; + config.merge_rollout.owned_targets = ["f", "a", "b", "c", "d", "e", "generic:x"] + .map(str::to_owned) + .to_vec(); + config.merge_rollout.drain_targets = vec!["b".into()]; + assert_eq!(candidates(&config, &[], None), ["a", "c", "d", "e", "f"]); + assert_eq!(candidates(&config, &[], Some("e")), ["f"]); + config.merge_wal_interval_secs = 0; + assert!(!enabled(&config)); + } +} diff --git a/crates/lance-context-master/src/rollout_append.rs b/crates/lance-context-master/src/rollout_append.rs index cfaea9de..69db5061 100644 --- a/crates/lance-context-master/src/rollout_append.rs +++ b/crates/lance-context-master/src/rollout_append.rs @@ -34,6 +34,14 @@ pub struct AppendConfig { default_value_t = 2_147_483_648 )] pub rollout_append_local_memory_bytes: usize, + /// Fleet-wide manifest discovery interval for resident owned targets; 0 disables. + /// Independent of full stats scans. At most four tables and two readers per batch. + #[arg( + long, + env = "ROLLOUT_APPEND_RECONCILE_INTERVAL_SECS", + default_value_t = 30 + )] + pub rollout_append_reconcile_interval_secs: u64, } impl Default for AppendConfig { fn default() -> Self { @@ -44,6 +52,7 @@ impl Default for AppendConfig { rollout_append_max_bytes: 67_108_864, rollout_append_local: false, rollout_append_local_memory_bytes: 2_147_483_648, + rollout_append_reconcile_interval_secs: 30, } } } @@ -112,7 +121,7 @@ impl AppendConfig { #[derive(Clone)] pub(crate) struct LocalStaging { pub(crate) budget: Arc, - session: Arc, + pub(crate) session: Arc, } impl LocalStaging { @@ -288,9 +297,20 @@ pub(crate) async fn run(state: &Arc, target: &str) -> Result, coordinator: &lance_context_merge::Coordinator, request: &lance_context_merge::rollout::MergeRequest, @@ -958,6 +958,7 @@ async fn enqueue_merge_request( /// `MasterState` immediately after. pub fn spawn_scheduler(state: &Arc) -> JoinHandle<()> { crate::wal_tail::spawn(state); + crate::resident_recovery::spawn(state); // Retry metadata independently of coarse stats sweeps. Every replica may // enqueue; the existing task dedupe/claim transaction elects one executor. let retry_state = state.clone(); diff --git a/docs/resident-wal-recovery.md b/docs/resident-wal-recovery.md new file mode 100644 index 00000000..3264463d --- /dev/null +++ b/docs/resident-wal-recovery.md @@ -0,0 +1,52 @@ +# Autonomous resident WAL demand + +Resident masters already execute bounded append-only merges using the ordinary +durable queue, shared process memory budget, progress watchdog and publication +fences. Two scheduling gaps could still require an external enqueue loop: + +* A successful slice stops after 16 passes or its configured service duration, + even when another slice could reclaim more WAL. +* Discovery through the full stats scan can lag behind newly sealed generations. + +A resident slice that exhausts its service budget now writes a durable merge +request before returning. The request consumer only acknowledges demand when a +task is queued, never when it finds the current task still running. A replacement +master can consume this demand. The successor yields through the ordinary queue +and existing maintenance admission; a hot table does not retain its claim +indefinitely. If the last pass drained everything, at most one empty successor +is needed; a zero-work pass does not perpetuate itself. + +With `ROLLOUT_APPEND_LOCAL=true`, `MERGE_WAL_INTERVAL_SECS>0`, and +`ROLLOUT_APPEND_RECONCILE_INTERVAL_SECS>0` (default 30), masters also discover WAL +demand separately from the stats scanner. One fleet-wide etcd compare-and-swap +reserves four targets per batch, with two metadata readers, a ten-second limit +per target, and at least 30 seconds between batch reservations. Each reader +loads the base manifest and latest manifests for up to 256 configured +`CATCHUP_SHARDS`; it does not read generation payloads, open shard writers, +replay raw WAL, or modify table/WAL manifests. Local staging's small shared +session bounds metadata caches. Missing or slow tables consume their bounded +visit and do not prevent rotation to later names. A reserved batch lost to a +crash is revisited on the next rotation. + +Explicit owned/append allowlists supply names even when stats are absent or +stale. Wildcard configurations additionally use known stats-cache names; this +does not replace registry discovery. Different rollout configurations have +separate durable cursors and share the fleet-wide read budget. Configured +historical shards participate; unconfigured historical shards still depend on +ordinary stats discovery or a merge pass's shard enumeration. + +Active canonical tasks, external catch-up owners, draining/unowned tables and +unexpired catch-up failure backoff are skipped. Repeated failures keep their +original attempt counters and retry deadlines. Demand only schedules an +ordinary task: it never removes an execution record, bypasses a target lock, +changes ownership allowlists, or authorizes a competing publisher. + +Metrics: `master_resident_merge_continuations_total` counts durable continuation +requests; `master_resident_wal_probes_total{result="requested|empty|failed"}` +describes discovery outcomes. Neither counts committed generations; use actual +merge completion metrics and pending observations for throughput. + +This removes external refill for already-owned resident tables. It does not +certify cancellation of legacy unfenced worker handlers, fix serial worker +flush scheduling, or replay an unsealed historical raw WAL tail. Those cases +must retain their existing recovery protection until separately validated. From ccfe7c7bf47f6bb6fc0427beae826fb20d9a4d26 Mon Sep 17 00:00:00 2001 From: Beinan Wang Date: Thu, 8 Oct 2026 07:10:29 +0000 Subject: [PATCH 2/2] Limit continuation requests to resident masters --- crates/lance-context-master/src/rollout_append.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/lance-context-master/src/rollout_append.rs b/crates/lance-context-master/src/rollout_append.rs index 69db5061..f7d48061 100644 --- a/crates/lance-context-master/src/rollout_append.rs +++ b/crates/lance-context-master/src/rollout_append.rs @@ -300,7 +300,7 @@ pub(crate) async fn run(state: &Arc, target: &str) -> Result