From 3204a6a978894a3de4f65353f73c249347fa4e08 Mon Sep 17 00:00:00 2001 From: Beinan Wang Date: Tue, 20 Oct 2026 23:16:33 +0000 Subject: [PATCH 01/10] Add durable admission primitives for worker-triggered WAL merges --- crates/lance-context-merge/src/background.rs | 376 +++++++++++++++++++ crates/lance-context-merge/src/failure.rs | 4 +- crates/lance-context-merge/src/lib.rs | 11 + crates/lance-context-merge/src/rollout.rs | 2 +- 4 files changed, 390 insertions(+), 3 deletions(-) create mode 100644 crates/lance-context-merge/src/background.rs diff --git a/crates/lance-context-merge/src/background.rs b/crates/lance-context-merge/src/background.rs new file mode 100644 index 00000000..a1ed00a3 --- /dev/null +++ b/crates/lance-context-merge/src/background.rs @@ -0,0 +1,376 @@ +//! Durable admission for worker count/timer triggers. A native handoff may +//! close this gate, but closure alone never proves old writers are quiescent. +use crate::{ + encode, execution_key, execution_owner, target_lock_key, ClaimProof, Coordinator, Execution, + Phase, Result, +}; +use etcd_client::{Compare, CompareOp, TxnOp}; + +/// No lease: losing a controller cannot reopen background admission. +pub fn admission_key(prefix: &str, target: &str) -> String { + execution_key(prefix, target).replace("/merge-executions/", "/background-merge-closed/") +} + +impl Coordinator { + /// Reserve before polling any storage work. The persistent request is in + /// the same transaction so a crash cannot hide an admitted execution from + /// the master. False means contention, a closed gate, or a retry deadline; + /// it never authorizes an untracked fallback. + pub async fn reserve_background(&self, execution: &Execution) -> Result { + if !execution.background + || execution.maintenance.is_some() + || execution.phase != Phase::Reserved + || execution.protocol != 2 + || execution.timeout_secs == 0 + || execution.queue_timeout_secs == 0 + || execution.idle_timeout_secs == 0 + || execution.target.is_empty() + || execution.target.starts_with("datagen:") + || execution.instance.is_empty() + || uuid::Uuid::parse_str(&execution.id).is_err() + { + return Err("invalid background merge reservation".into()); + } + let key = execution_key(&self.prefix, &execution.target); + let failure_key = self.failure_key(&execution.target, &execution.endpoint); + let prior = self + .client + .clone() + .get(failure_key.clone(), None) + .await + .map_err(|e| e.to_string())?; + let failure = prior.kvs().first(); + if let Some(row) = failure { + let previous: crate::failure::ShardFailure = + serde_json::from_slice(row.value()).map_err(|e| e.to_string())?; + if previous.corrected().next_retry_ms > crate::failure::now_ms() { + return Ok(false); + } + } + self.transact( + vec![ + Compare::version(key.as_str(), CompareOp::Equal, 0), + Compare::version( + target_lock_key(&self.prefix, &execution.target), + CompareOp::Equal, + 0, + ), + Compare::version( + admission_key(&self.prefix, &execution.target), + CompareOp::Equal, + 0, + ), + Compare::version( + key.replace("/merge-executions/", "/catchup-active/"), + CompareOp::Equal, + 0, + ), + Compare::version( + key.replace("/merge-executions/", "/merge-claims/"), + CompareOp::Equal, + 0, + ), + Compare::version( + key.replace("/merge-executions/", "/compact-commit-ready/"), + CompareOp::Equal, + 0, + ), + Compare::mod_revision( + failure_key, + CompareOp::Equal, + failure.map_or(0, |kv| kv.mod_revision()), + ), + ], + vec![ + TxnOp::put(key, encode(execution), None), + TxnOp::put( + target_lock_key(&self.prefix, &execution.target), + execution_owner(execution), + None, + ), + TxnOp::put( + self.request_key(&execution.target), + execution.target.as_str(), + None, + ), + ], + ) + .await + } + + /// Release only after local work and all commit leaves have drained, or a + /// storage barrier has completed. A stale worker cannot remove a successor. + /// The ordinary master release method can also reconcile these records. + pub async fn release_background(&self, terminal: &Execution) -> Result { + if !terminal.background + || terminal.protocol != 2 + || !matches!(terminal.phase, Phase::Finished | Phase::Recovered) + { + return Err("background merge is not durably terminal".into()); + } + let (mut compares, mut operations) = self.completion_changes(terminal).await?; + compares.extend([ + Compare::value( + execution_key(&self.prefix, &terminal.target), + CompareOp::Equal, + encode(terminal), + ), + Compare::value( + target_lock_key(&self.prefix, &terminal.target), + CompareOp::Equal, + execution_owner(terminal), + ), + ]); + operations.extend([ + self.remove_permits(terminal), + self.remove_progress(terminal), + self.remove_work_progress(terminal), + TxnOp::delete(execution_key(&self.prefix, &terminal.target), None), + TxnOp::delete(target_lock_key(&self.prefix, &terminal.target), None), + ]); + self.transact(compares, operations).await + } + + /// The canonical scheduler claim serializes this with background reserve. + /// This prevents NEW background work; it does not authorize native writes + /// or certify coverage of historical untracked writers. + pub async fn close_background_admission( + &self, + proof: &ClaimProof, + target: &str, + ) -> Result { + self.transact( + vec![ + Compare::value(proof.key.as_str(), CompareOp::Equal, proof.token.as_bytes()), + Compare::value( + target_lock_key(&self.prefix, target), + CompareOp::Equal, + proof.token.as_bytes(), + ), + Compare::version(execution_key(&self.prefix, target), CompareOp::Equal, 0), + ], + vec![TxnOp::put( + admission_key(&self.prefix, target), + "closed", + None, + )], + ) + .await + } +} + +#[cfg(test)] +mod tests { + use super::*; + use etcd_client::{Client, Txn}; + + async fn fixture() -> (Coordinator, Client, String) { + let endpoint = std::env::var("ETCD_TEST_ENDPOINTS").expect("isolated test etcd required"); + let client = Client::connect([endpoint], None).await.unwrap(); + let prefix = format!("/background-admission-test/{}", uuid::Uuid::new_v4()); + (Coordinator::new(client.clone(), &prefix), client, prefix) + } + + fn execution() -> Execution { + let mut execution = Execution::new("hot", "http://worker", "worker-boot", 60); + execution.background = true; + execution + } + + #[test] + fn old_readers_cannot_cas_background_records() { + let background = execution(); + let bytes = encode(&background); + assert_eq!( + serde_json::from_slice::(&bytes).unwrap(), + background + ); + let mut old_reader = background.clone(); + old_reader.background = false; + assert_ne!(encode(&old_reader), bytes); + assert!(!String::from_utf8(encode(&old_reader)) + .unwrap() + .contains("background")); + } + + #[tokio::test] + #[ignore = "requires isolated ETCD_TEST_ENDPOINTS"] + async fn concurrent_triggers_admit_one_durable_writer_and_request() { + let (coordinator, _, _) = fixture().await; + let mut tasks = Vec::new(); + for _ in 0..20 { + let c = coordinator.clone(); + tasks.push(tokio::spawn(async move { + let execution = execution(); + c.reserve_background(&execution) + .await + .unwrap() + .then_some(execution) + })); + } + let mut admitted = Vec::new(); + for task in tasks { + admitted.extend(task.await.unwrap()); + } + assert_eq!(admitted.len(), 1); + assert_eq!( + coordinator.get("hot").await.unwrap(), + Some(admitted[0].clone()) + ); + let (requests, next) = coordinator.request_page(None).await.unwrap(); + assert_eq!(requests.len(), 1); + assert_eq!(requests[0].target, "hot"); + assert!(next.is_none()); + } + + #[tokio::test] + #[ignore = "requires isolated ETCD_TEST_ENDPOINTS"] + async fn closing_race_never_forgets_a_writer_and_survives_controller_loss() { + let (coordinator, mut client, prefix) = fixture().await; + let proof = ClaimProof { + key: format!("{prefix}/claim"), + token: "master".into(), + lease_id: 0, + }; + client + .put(proof.key.as_str(), proof.token.as_str(), None) + .await + .unwrap(); + let background = execution(); + let master_claim = client.txn( + Txn::new() + .when([Compare::version( + target_lock_key(&prefix, "hot"), + CompareOp::Equal, + 0, + )]) + .and_then([TxnOp::put( + target_lock_key(&prefix, "hot"), + proof.token.as_str(), + None, + )]), + ); + let (worker, master) = + tokio::join!(coordinator.reserve_background(&background), master_claim); + let worker = worker.unwrap(); + assert_ne!(worker, master.unwrap().succeeded()); + if worker { + assert!(!coordinator + .close_background_admission(&proof, "hot") + .await + .unwrap()); + assert!(coordinator.release_background(&background).await.is_err()); + let running = coordinator.start(&background).await.unwrap().unwrap(); + assert!(coordinator.finish(&running, Ok(3)).await.unwrap()); + assert!(coordinator + .release_background(&running.finished(Ok(3))) + .await + .unwrap()); + client + .put(target_lock_key(&prefix, "hot"), proof.token.as_str(), None) + .await + .unwrap(); + } + assert!(coordinator + .close_background_admission(&proof, "hot") + .await + .unwrap()); + // Model expired controller/task ownership; only its ephemeral state + // disappears. The separate gate remains closed across a fresh client. + client.delete(proof.key, None).await.unwrap(); + client + .delete(target_lock_key(&prefix, "hot"), None) + .await + .unwrap(); + let restarted = Coordinator::new(client, prefix); + assert!(!restarted.reserve_background(&execution()).await.unwrap()); + assert!(restarted.get("hot").await.unwrap().is_none()); + } + + #[tokio::test] + #[ignore = "requires isolated ETCD_TEST_ENDPOINTS"] + async fn uncertain_writer_requires_fencing_and_cannot_release_a_successor() { + let (coordinator, mut client, prefix) = fixture().await; + let background = execution(); + assert!(coordinator.reserve_background(&background).await.unwrap()); + let running = coordinator.start(&background).await.unwrap().unwrap(); + coordinator + .authorize_commit(&running, "memory://dataset", "base", 7) + .await + .unwrap(); + assert!(coordinator + .report_uncertain(&running, "lost PUT response".into()) + .await + .unwrap()); + let uncertain = coordinator.get("hot").await.unwrap().unwrap(); + assert!(coordinator.release_background(&uncertain).await.is_err()); + assert!(!coordinator.reserve_background(&execution()).await.unwrap()); + let proof = ClaimProof { + key: format!("{prefix}/claim"), + token: "master".into(), + lease_id: 0, + }; + client + .put(proof.key.as_str(), proof.token.as_str(), None) + .await + .unwrap(); + let frozen = coordinator + .freeze(&proof, &uncertain) + .await + .unwrap() + .unwrap(); + assert_eq!( + coordinator.watermarks(&frozen).await.unwrap().versions["base"], + 7 + ); + assert!(coordinator + .authorize_commit(&running, "memory://dataset", "base", 8) + .await + .is_err()); + // Only exercise metadata acknowledgement here; real object-store + // barrier behavior is covered by the core/master storage tests. + assert!(coordinator.finish_recovery(&proof, &frozen).await.unwrap()); + let recovered = coordinator.get("hot").await.unwrap().unwrap(); + assert!(coordinator.release(&proof, &recovered).await.unwrap()); + let successor = Execution::new("hot", "http://successor", "next-boot", 60); + assert!(coordinator.reserve(&proof, &successor).await.unwrap()); + assert!(!coordinator.release_background(&recovered).await.unwrap()); + assert_eq!(coordinator.get("hot").await.unwrap(), Some(successor)); + } + + #[tokio::test] + #[ignore = "requires isolated ETCD_TEST_ENDPOINTS"] + async fn background_failures_preserve_retry_budget_and_external_admission() { + let (coordinator, mut client, prefix) = fixture().await; + let key = execution_key(&prefix, "hot"); + for reserved in [ + target_lock_key(&prefix, "hot"), + key.replace("/merge-executions/", "/catchup-active/"), + key.replace("/merge-executions/", "/merge-claims/"), + key.replace("/merge-executions/", "/compact-commit-ready/"), + ] { + client.put(reserved.as_str(), "other", None).await.unwrap(); + assert!(!coordinator.reserve_background(&execution()).await.unwrap()); + client.delete(reserved, None).await.unwrap(); + } + let background = execution(); + assert!(coordinator.reserve_background(&background).await.unwrap()); + let running = coordinator.start(&background).await.unwrap().unwrap(); + let error = Err("storage timeout".into()); + assert!(coordinator.finish(&running, error.clone()).await.unwrap()); + assert!(coordinator + .release_background(&running.finished(error)) + .await + .unwrap()); + assert!(!coordinator.reserve_background(&execution()).await.unwrap()); + assert_eq!( + coordinator + .failure("hot", "http://worker") + .await + .unwrap() + .unwrap() + .consecutive_attempts, + 1 + ); + } +} diff --git a/crates/lance-context-merge/src/failure.rs b/crates/lance-context-merge/src/failure.rs index 7a196c08..58e539c2 100644 --- a/crates/lance-context-merge/src/failure.rs +++ b/crates/lance-context-merge/src/failure.rs @@ -183,7 +183,7 @@ impl ShardFailure { // Read old records with corrected semantics without resetting attempts, // moving the failure timestamp, or mutating ownership/durable records. // The next fenced completion persists the normal retry result. - fn corrected(mut self) -> Self { + pub(crate) fn corrected(mut self) -> Self { // One rule set: whatever `classify` says today is the effective // class. Only a DataOrConfiguration record can be softened; a stored // Retryable is never hardened on read. @@ -227,7 +227,7 @@ impl Coordinator { fn failures_prefix(&self) -> String { format!("{}/merge-failures/", self.prefix.trim_end_matches('/')) } - fn failure_key(&self, target: &str, endpoint: &str) -> String { + pub(crate) fn failure_key(&self, target: &str, endpoint: &str) -> String { let hex = |s: &str| { s.as_bytes() .iter() diff --git a/crates/lance-context-merge/src/lib.rs b/crates/lance-context-merge/src/lib.rs index e89ec211..2f8115a5 100644 --- a/crates/lance-context-merge/src/lib.rs +++ b/crates/lance-context-merge/src/lib.rs @@ -4,6 +4,7 @@ //! Only the executor can publish its terminal outcome. Recovery first cancels //! and reconciles the old execution; it never clears a running fence on timeout. +pub mod background; pub mod demand; pub mod failure; pub mod fencing; @@ -66,6 +67,10 @@ impl MaintenanceKind { #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] pub struct Execution { + /// Count/timer-triggered worker execution, admitted without a scheduler + /// claim. Its persistent owner and commit permits are otherwise identical. + #[serde(default, skip_serializing_if = "is_false")] + pub background: bool, #[serde(default, skip_serializing_if = "Option::is_none")] pub maintenance: Option, pub id: String, @@ -88,6 +93,10 @@ fn protocol_absent(protocol: &u32) -> bool { *protocol == 0 } +fn is_false(value: &bool) -> bool { + !value +} + fn default_queue_timeout() -> u64 { 600 } @@ -95,6 +104,7 @@ fn default_queue_timeout() -> u64 { impl Execution { pub fn new(target: &str, endpoint: &str, instance: &str, timeout_secs: u64) -> Self { Self { + background: false, maintenance: None, id: uuid::Uuid::new_v4().to_string(), target: target.into(), @@ -220,6 +230,7 @@ impl Coordinator { /// Atomic claim check and admission close the delayed-request/lease-loss race. pub async fn reserve(&self, claim: &ClaimProof, execution: &Execution) -> Result { if execution.phase != Phase::Reserved + || execution.background || execution.timeout_secs == 0 || execution.protocol != 2 { diff --git a/crates/lance-context-merge/src/rollout.rs b/crates/lance-context-merge/src/rollout.rs index 30d2571d..cf0c710a 100644 --- a/crates/lance-context-merge/src/rollout.rs +++ b/crates/lance-context-merge/src/rollout.rs @@ -56,7 +56,7 @@ pub struct MergeRequest { } impl Coordinator { - fn request_key(&self, target: &str) -> String { + pub(crate) fn request_key(&self, target: &str) -> String { execution_key(&self.prefix, target).replace("/merge-executions/", "/merge-requests/") } From 09fff2cd8a54ce304c6a3035abb10e366d36ff21 Mon Sep 17 00:00:00 2001 From: Beinan Wang Date: Tue, 20 Oct 2026 23:19:21 +0000 Subject: [PATCH 02/10] Preserve terminal ownership for a waiting master claim --- crates/lance-context-merge/src/background.rs | 45 ++++++++++++++++++++ 1 file changed, 45 insertions(+) diff --git a/crates/lance-context-merge/src/background.rs b/crates/lance-context-merge/src/background.rs index a1ed00a3..038e96d7 100644 --- a/crates/lance-context-merge/src/background.rs +++ b/crates/lance-context-merge/src/background.rs @@ -110,6 +110,15 @@ impl Coordinator { } let (mut compares, mut operations) = self.completion_changes(terminal).await?; compares.extend([ + // A scheduler may already be reconciling this execution. Leave + // its terminal record intact so ordinary release can transfer the + // write lock to that claim, rather than deleting its predecessor. + Compare::version( + execution_key(&self.prefix, &terminal.target) + .replace("/merge-executions/", "/merge-claims/"), + CompareOp::Equal, + 0, + ), Compare::value( execution_key(&self.prefix, &terminal.target), CompareOp::Equal, @@ -338,6 +347,42 @@ mod tests { assert_eq!(coordinator.get("hot").await.unwrap(), Some(successor)); } + #[tokio::test] + #[ignore = "requires isolated ETCD_TEST_ENDPOINTS"] + async fn terminal_background_writer_transfers_to_a_waiting_master_claim() { + let (coordinator, mut client, prefix) = fixture().await; + let background = execution(); + assert!(coordinator.reserve_background(&background).await.unwrap()); + let running = coordinator.start(&background).await.unwrap().unwrap(); + let proof = ClaimProof { + key: format!("{prefix}/claim"), + token: "master".into(), + lease_id: 0, + }; + client + .put(proof.key.as_str(), proof.token.as_str(), None) + .await + .unwrap(); + client + .put( + execution_key(&prefix, "hot").replace("/merge-executions/", "/merge-claims/"), + proof.token.as_str(), + None, + ) + .await + .unwrap(); + assert!(coordinator.finish(&running, Ok(4)).await.unwrap()); + let terminal = running.finished(Ok(4)); + assert!(!coordinator.release_background(&terminal).await.unwrap()); + assert_eq!( + coordinator.get("hot").await.unwrap(), + Some(terminal.clone()) + ); + assert!(coordinator.release(&proof, &terminal).await.unwrap()); + let next = Execution::new("hot", "http://next", "next-boot", 60); + assert!(coordinator.reserve(&proof, &next).await.unwrap()); + } + #[tokio::test] #[ignore = "requires isolated ETCD_TEST_ENDPOINTS"] async fn background_failures_preserve_retry_budget_and_external_admission() { From 9e1edfc76353ea2dc4cb3740f849c2f9f4cae42e Mon Sep 17 00:00:00 2001 From: Beinan Wang Date: Tue, 20 Oct 2026 23:43:57 +0000 Subject: [PATCH 03/10] Wire worker merge admission and master recovery with persistent protocol mode --- .../lance-context-master/src/eligibility.rs | 1 + .../src/merge_execution.rs | 46 +- crates/lance-context-master/src/scheduler.rs | 135 ++++- crates/lance-context-master/src/state.rs | 7 +- crates/lance-context-merge/src/lib.rs | 1 + crates/lance-context-merge/src/mode.rs | 79 +++ crates/lance-context-merge/src/rollout.rs | 4 + crates/lance-context-server/src/config.rs | 5 + crates/lance-context-server/src/main.rs | 3 + .../src/merge_execution.rs | 105 +++- .../src/merge_execution/background.rs | 484 ++++++++++++++++++ .../src/routes/generic.rs | 9 + .../src/routes/rollouts.rs | 13 + crates/lance-context-server/src/state.rs | 21 + crates/lance-context-server/src/sweeper.rs | 20 +- docs/worker-merge-admission.md | 69 +++ 16 files changed, 968 insertions(+), 34 deletions(-) create mode 100644 crates/lance-context-merge/src/mode.rs create mode 100644 crates/lance-context-server/src/merge_execution/background.rs create mode 100644 docs/worker-merge-admission.md diff --git a/crates/lance-context-master/src/eligibility.rs b/crates/lance-context-master/src/eligibility.rs index e6019d40..611e98dc 100644 --- a/crates/lance-context-master/src/eligibility.rs +++ b/crates/lance-context-master/src/eligibility.rs @@ -183,6 +183,7 @@ mod tests { fn rollout(owned: &[&str], draining: &[&str]) -> MergeRollout { MergeRollout { + fence_legacy: false, owned_targets: owned.iter().map(|s| s.to_string()).collect(), drain_targets: draining.iter().map(|s| s.to_string()).collect(), } diff --git a/crates/lance-context-master/src/merge_execution.rs b/crates/lance-context-master/src/merge_execution.rs index 730dffc3..f0355cc4 100644 --- a/crates/lance-context-master/src/merge_execution.rs +++ b/crates/lance-context-master/src/merge_execution.rs @@ -25,18 +25,14 @@ struct Capabilities { idle_timeout_secs: u64, progress_protocol: u32, owned_targets: Vec, + #[serde(default)] + fenced_legacy_merges: bool, } pub(crate) async fn run_merge_wal( state: &Arc, claim: &TaskClaim, ) -> Result { - if state.config.worker_endpoints.is_empty() - && !state.config.merge_rollout.draining(&claim.task.target) - && !crate::local_merge::selected(&state.config, &claim.task.target) - { - return Err("no worker endpoints configured".into()); - } let coordinator = state.task_store.merge_coordinator(); let proof = state.task_store.merge_claim(claim); let target = &claim.task.target; @@ -56,12 +52,32 @@ pub(crate) async fn run_merge_wal( } return Err("merge target draining for protocol transition".into()); } - if !state.config.merge_rollout.owned(target) { - if coordinator.get(target).await?.is_some() { - return Err("owned execution still present; refusing legacy downgrade".into()); + if !state.config.merge_rollout.owned(target) && !state.config.merge_rollout.fence_legacy { + if let Some(old) = coordinator.get(target).await? { + // Turning off new admission must never disable recovery of durable + // worker records. Observe useful progress before considering cancel. + if old.maintenance.is_some() || old.protocol != 2 { + return Err("owned execution still present; refusing legacy downgrade".into()); + } + ensure_recovery_due(&coordinator, &old).await?; + if reconcile(&state.http, &coordinator, &proof, old, false) + .await + .is_err() + { + if let Some(current) = coordinator.get(target).await? { + recover_execution(state, &coordinator, &proof, current).await?; + } + } + return Ok("reconciled recorded worker merge; new fenced admission disabled".into()); } return run_legacy(state, target).await; } + if state.config.worker_endpoints.is_empty() + && !crate::local_merge::selected(&state.config, target) + && coordinator.get(target).await?.is_none() + { + return Err("no worker endpoints configured".into()); + } if state.config.append.enabled(target) || crate::local_merge::selected(&state.config, target) { if let Some(old) = coordinator.get(target).await? { ensure_recovery_due(&coordinator, &old).await?; @@ -88,7 +104,8 @@ pub(crate) async fn run_merge_wal( ensure_recovery_due(&coordinator, &old).await?; if recovery_round > 0 || matches!(old.phase, Phase::Recovering | Phase::Uncertain) { recover_execution(state, &coordinator, &proof, old).await?; - } else if let Err(error) = reconcile(&state.http, &coordinator, &proof, old, true).await + } else if let Err(error) = + reconcile(&state.http, &coordinator, &proof, old, false).await { if let Some(current) = coordinator.get(&claim.task.target).await? { recover_execution(state, &coordinator, &proof, current).await?; @@ -118,6 +135,9 @@ pub(crate) async fn run_merge_wal( // Compatibility path is chosen only by explicit configuration, never after a // failed owned RPC. It retains legacy limitations until that table is drained. async fn run_legacy(state: &Arc, target: &str) -> Result { + if state.config.worker_endpoints.is_empty() { + return Err("no worker endpoints configured".into()); + } run_legacy_workers(&state.http, &state.config.worker_endpoints, target).await } @@ -373,6 +393,9 @@ async fn run_workers( target: &str, endpoints: &[String], ) -> Result { + if endpoints.is_empty() { + return Err("no worker endpoints configured after reconciling recorded execution".into()); + } let mut pending = Vec::new(); let mut deferred = Vec::new(); for endpoint in endpoints { @@ -481,7 +504,8 @@ async fn one( .map_err(|e| e.to_string())?; if capabilities.protocol != 2 || capabilities.progress_protocol != 1 - || !capabilities.owned_targets.iter().any(|t| t == target) + || (!capabilities.owned_targets.iter().any(|t| t == target) + && !capabilities.fenced_legacy_merges) || capabilities.timeout_secs == 0 || capabilities.queue_timeout_secs == 0 || capabilities.idle_timeout_secs == 0 diff --git a/crates/lance-context-master/src/scheduler.rs b/crates/lance-context-master/src/scheduler.rs index e787731b..e35a15bf 100644 --- a/crates/lance-context-master/src/scheduler.rs +++ b/crates/lance-context-master/src/scheduler.rs @@ -956,7 +956,13 @@ pub(crate) async fn enqueue_merge_request( coordinator: &lance_context_merge::Coordinator, request: &lance_context_merge::rollout::MergeRequest, ) -> Result<(), String> { - if !state.config.merge_rollout.owned(&request.target) { + if !state.config.merge_rollout.owned(&request.target) + && !state.config.merge_rollout.fence_legacy + && !coordinator + .get(&request.target) + .await? + .is_some_and(|e| e.maintenance.is_none()) + { return Ok(()); } let task = enqueue(state, TaskKind::MergeWal, &request.target) @@ -968,6 +974,23 @@ pub(crate) async fn enqueue_merge_request( Ok(()) } +async fn enqueue_execution_recovery( + state: &Arc, + coordinator: &lance_context_merge::Coordinator, + execution: &lance_context_merge::Execution, +) -> Result<(), String> { + if execution.maintenance.is_some() { + return crate::maintenance_execution::enqueue_recovery(state, coordinator, execution).await; + } + // Durable worker executions survive flags, endpoint list changes and loss + // of their coalesced request. Task claims still serialize reconciliation. + crate::merge_execution::ensure_recovery_due(coordinator, execution).await?; + enqueue(state, TaskKind::MergeWal, &execution.target) + .await + .map_err(|e| e.to_string())?; + Ok(()) +} + /// Spawn the scheduler pollers plus the optional periodic auto-sweep. /// /// Returns the handle of the *general* poller only. When a separate WAL-merge @@ -1012,15 +1035,8 @@ pub fn spawn_scheduler(state: &Arc) -> JoinHandle<()> { Ok((executions, next)) => { execution_cursor = next; for execution in executions { - if execution.maintenance.is_none() { - continue; - } - if let Err(error) = crate::maintenance_execution::enqueue_recovery( - &retry_state, - &coordinator, - &execution, - ) - .await + if let Err(error) = + enqueue_execution_recovery(&retry_state, &coordinator, &execution).await { tracing::warn!(target = %execution.target, %error, "local execution recovery enqueue failed"); } @@ -1298,6 +1314,7 @@ mod tests { planner: Default::default(), maintenance: Default::default(), merge_rollout: lance_context_merge::rollout::MergeRollout { + fence_legacy: false, owned_targets: ["exp", "generic:gs", "broken"] .into_iter() .map(str::to_string) @@ -1875,6 +1892,104 @@ mod tests { assert!(coordinator.request_page(None).await.unwrap().0.is_empty()); } + #[tokio::test] + #[ignore = "requires ETCD_TEST_ENDPOINTS"] + async fn background_inventory_recovers_after_flags_and_requests_disappear() { + use lance_context_merge::Execution; + let dir = TempDir::new().unwrap(); + let mut cfg = config(&dir); + cfg.merge_rollout.owned_targets.clear(); + cfg.merge_rollout.fence_legacy = false; + let state = MasterState::new(cfg).await.unwrap(); + let c = state.task_store.merge_coordinator(); + let mut e = Execution::new("new-hot-table", "http://127.0.0.1:1", "old-worker", 60); + e.background = true; + assert!(c.reserve_background(&e).await.unwrap()); + let e = c.start(&e).await.unwrap().unwrap(); + assert!(c.finish(&e, Ok(19)).await.unwrap()); + let request = c.request_page(None).await.unwrap().0.remove(0); + assert!(c.acknowledge_request(&request).await.unwrap()); + let inventory = c.execution_page(None).await.unwrap().0; + enqueue_execution_recovery(&state, &c, &inventory[0]) + .await + .unwrap(); + let claim = state + .task_store + .claim_next_of_kinds(TaskKinds::MERGE_WAL) + .await + .unwrap() + .unwrap(); + let result = crate::merge_execution::run_merge_wal(&state, &claim) + .await + .unwrap(); + assert!(result.contains("reconciled")); + assert!(c.get("new-hot-table").await.unwrap().is_none()); + state.task_store.finish(claim, Ok(result)).await.unwrap(); + // Ambiguous storage needs a real barrier after initiation is disabled. + let uri = state.rollout_uri("new-hot-table"); + RolloutStore::open(&uri) + .await + .unwrap() + .close() + .await + .unwrap(); + let version = lance::Dataset::open(&uri).await.unwrap().version().version; + let mut orphan = Execution::new("new-hot-table", "http://127.0.0.1:1", "lost-worker", 60); + orphan.background = true; + assert!(c.reserve_background(&orphan).await.unwrap()); + let running = c.start(&orphan).await.unwrap().unwrap(); + c.authorize_commit(&running, &uri, "base", version + 1) + .await + .unwrap(); + assert!(c + .report_uncertain(&running, "lost response".into()) + .await + .unwrap()); + let old = c.get("new-hot-table").await.unwrap().unwrap(); + enqueue_execution_recovery(&state, &c, &old).await.unwrap(); + let claim = state + .task_store + .claim_next_of_kinds(TaskKinds::MERGE_WAL) + .await + .unwrap() + .unwrap(); + let result = crate::merge_execution::run_merge_wal(&state, &claim) + .await + .unwrap(); + assert!(c.get("new-hot-table").await.unwrap().is_none()); + assert!(lance::Dataset::open(&uri).await.unwrap().version().version > version + 1); + assert!(c + .authorize_commit(&running, &uri, "base", version + 3) + .await + .is_err()); + state.task_store.finish(claim, Ok(result)).await.unwrap(); + } + + #[tokio::test] + #[ignore = "requires ETCD_TEST_ENDPOINTS"] + async fn fenced_new_table_demand_enqueues_without_static_ownership() { + let dir = TempDir::new().unwrap(); + let mut cfg = config(&dir); + cfg.merge_rollout.owned_targets.clear(); + cfg.merge_rollout.fence_legacy = true; + let state = MasterState::new(cfg).await.unwrap(); + let c = state.task_store.merge_coordinator(); + c.request_merge("new-hot-table").await.unwrap(); + let request = c.request_page(None).await.unwrap().0.remove(0); + enqueue_merge_request(&state, &c, &request).await.unwrap(); + assert!(state + .task_store + .get_active_id(TaskKind::MergeWal, "new-hot-table") + .await + .unwrap() + .is_some()); + assert!(c.request_page(None).await.unwrap().0.is_empty()); + assert!( + !crate::local_merge::selected(&state.config, "new-hot-table"), + "worker recording does not certify native ownership" + ); + } + /// Wait until the task reaches a terminal state, returning its final record. async fn await_terminal(state: &Arc, id: &str) -> TaskRecord { for _ in 0..100 { diff --git a/crates/lance-context-master/src/state.rs b/crates/lance-context-master/src/state.rs index 1dd98722..0a6f56dd 100644 --- a/crates/lance-context-master/src/state.rs +++ b/crates/lance-context-master/src/state.rs @@ -135,7 +135,7 @@ pub(crate) fn worker_http_client() -> Result { impl MasterState { /// Open the registry, stats dataset, and configured durable task store. - pub async fn new(config: MasterConfig) -> lance::Result> { + pub async fn new(mut config: MasterConfig) -> lance::Result> { config.merge_rollout.validate().map_err(lance::Error::io)?; config.catchup.validate().map_err(lance::Error::io)?; crate::local_merge::validate(&config).map_err(lance::Error::io)?; @@ -152,6 +152,11 @@ impl MasterState { )); } let task_store = TaskStore::open(&config).await?; + config.merge_rollout.fence_legacy = task_store + .merge_coordinator() + .legacy_fencing_mode(config.merge_rollout.fence_legacy) + .await + .map_err(lance::Error::io)?; // Serialize first-time registry/stats creation and legacy backfill in // etcd mode. Followers wait briefly rather than racing Lance creates. let dedicated = config.catchup.target.is_some(); diff --git a/crates/lance-context-merge/src/lib.rs b/crates/lance-context-merge/src/lib.rs index 2f8115a5..db6dad7d 100644 --- a/crates/lance-context-merge/src/lib.rs +++ b/crates/lance-context-merge/src/lib.rs @@ -8,6 +8,7 @@ pub mod background; pub mod demand; pub mod failure; pub mod fencing; +mod mode; pub mod progress; pub mod rollout; diff --git a/crates/lance-context-merge/src/mode.rs b/crates/lance-context-merge/src/mode.rs new file mode 100644 index 00000000..f018f101 --- /dev/null +++ b/crates/lance-context-merge/src/mode.rs @@ -0,0 +1,79 @@ +//! A protocol activation survives configuration rollback. This records the +//! required protocol, not proof that pre-upgrade writers have drained. +use crate::{Coordinator, Result}; +use etcd_client::{Compare, CompareOp, TxnOp}; + +impl Coordinator { + pub async fn legacy_fencing_mode(&self, enable: bool) -> Result { + let key = format!( + "{}/worker-merge-admission-mode", + self.prefix.trim_end_matches('/') + ); + let mut client = self.client.clone(); + if enable { + // No lease and no API to reopen legacy admission. Do not overwrite + // a future protocol marker during a binary/config rollback. + self.transact( + vec![Compare::version(key.as_str(), CompareOp::Equal, 0)], + vec![TxnOp::put(key.as_str(), "fenced-v1", None)], + ) + .await?; + } + let reply = client.get(key, None).await.map_err(|e| e.to_string())?; + match reply.kvs().first().map(|kv| kv.value()) { + None if !enable => Ok(false), + None => Err("worker merge protocol activation was not persisted".into()), + Some(b"fenced-v1") => Ok(true), + Some(_) => { + Err("unknown worker merge admission protocol; refusing legacy fallback".into()) + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + #[ignore = "requires isolated local ETCD_TEST_ENDPOINTS"] + async fn activated_protocol_survives_a_new_client_and_disabled_flag() { + let endpoint = std::env::var("ETCD_TEST_ENDPOINTS").unwrap(); + let prefix = format!("/merge-mode/{}", crate::Execution::new("", "", "", 1).id); + let first = Coordinator::new( + etcd_client::Client::connect([&endpoint], None) + .await + .unwrap(), + prefix.clone(), + ); + assert!(!first.legacy_fencing_mode(false).await.unwrap()); + assert!(first.legacy_fencing_mode(true).await.unwrap()); + let restarted = Coordinator::new( + etcd_client::Client::connect([&endpoint], None) + .await + .unwrap(), + prefix, + ); + assert!(restarted.legacy_fencing_mode(false).await.unwrap()); + } + #[tokio::test] + #[ignore = "requires isolated local ETCD_TEST_ENDPOINTS"] + async fn a_newer_protocol_cannot_be_replaced_by_activation() { + let endpoint = std::env::var("ETCD_TEST_ENDPOINTS").unwrap(); + let prefix = format!("/merge-mode/{}", crate::Execution::new("", "", "", 1).id); + let mut client = etcd_client::Client::connect([endpoint], None) + .await + .unwrap(); + client + .put( + format!("{prefix}/worker-merge-admission-mode"), + "fenced-v2", + None, + ) + .await + .unwrap(); + let c = Coordinator::new(client, prefix); + assert!(c.legacy_fencing_mode(false).await.is_err()); + assert!(c.legacy_fencing_mode(true).await.is_err()); + } +} diff --git a/crates/lance-context-merge/src/rollout.rs b/crates/lance-context-merge/src/rollout.rs index cf0c710a..7ae74ccd 100644 --- a/crates/lance-context-merge/src/rollout.rs +++ b/crates/lance-context-merge/src/rollout.rs @@ -4,6 +4,10 @@ use etcd_client::{Compare, CompareOp, GetOptions, TxnOp}; #[derive(Clone, Debug, Default, clap::Args)] pub struct MergeRollout { + /// Record new legacy-target worker merges with commit fencing. This does + /// not enroll targets for native catchup or fence pre-upgrade writers. + #[arg(long, env = "MERGE_FENCE_LEGACY", default_value_t = false)] + pub fence_legacy: bool, /// Exact scheduler targets using owned merges (generic stores: generic:name). /// Empty by default. Enable only after the target's legacy writers drain. #[arg( diff --git a/crates/lance-context-server/src/config.rs b/crates/lance-context-server/src/config.rs index 91622e87..965dd184 100644 --- a/crates/lance-context-server/src/config.rs +++ b/crates/lance-context-server/src/config.rs @@ -28,6 +28,11 @@ pub struct ServerConfig { )] pub http_shutdown_response_idle_secs: u64, + /// Stable worker URL reachable by the master for execution identity and + /// cancellation. Required when MERGE_FENCE_LEGACY is enabled. + #[arg(long, env = "MERGE_EXECUTOR_ENDPOINT")] + pub merge_executor_endpoint: Option, + /// Legacy wire field; owned merges now use the real-progress idle timeout. #[arg(long, env = "MERGE_EXECUTION_TIMEOUT_SECS", default_value_t = 3600)] pub merge_execution_timeout_secs: u64, diff --git a/crates/lance-context-server/src/main.rs b/crates/lance-context-server/src/main.rs index d83edb0e..751da9bd 100644 --- a/crates/lance-context-server/src/main.rs +++ b/crates/lance-context-server/src/main.rs @@ -50,6 +50,9 @@ async fn main() { } }; + // Resolve persisted merge protocol independently of startup and ingestion. + merge_execution::Executions::resolve_mode_in_background(state.clone()); + // Single process-wide WAL-cleanup sweeper (replaces one timer per store). // The handle is detached: it runs for the server's lifetime and stops when // the last `Arc` is dropped. diff --git a/crates/lance-context-server/src/merge_execution.rs b/crates/lance-context-server/src/merge_execution.rs index fb6c96d3..0ce634a4 100644 --- a/crates/lance-context-server/src/merge_execution.rs +++ b/crates/lance-context-server/src/merge_execution.rs @@ -1,5 +1,8 @@ //! HTTP handlers only admit/cancel work. The owned executor, not the connection, //! holds the durable execution fence until the scoped storage future is gone. +mod background; +pub(crate) use background::spawn_background; + use crate::{ error::AppError, routes::{generic, rollouts}, @@ -49,6 +52,10 @@ pub struct Executions { coordinator: OnceCell, etcd: Option, pub(crate) rollout: lance_context_merge::rollout::MergeRollout, + endpoint: Option, + fencing_mode: OnceCell, + mode_pending: std::sync::atomic::AtomicBool, + background_targets: std::sync::Mutex>, queue_timeout_secs: u64, idle_timeout_secs: u64, instance: String, @@ -64,6 +71,10 @@ impl Executions { coordinator: OnceCell::new_with(coordinator), etcd: None, rollout: Default::default(), + endpoint: None, + fencing_mode: OnceCell::new(), + mode_pending: std::sync::atomic::AtomicBool::new(false), + background_targets: Default::default(), queue_timeout_secs: 600, idle_timeout_secs: 600, instance: uuid::Uuid::new_v4().to_string(), @@ -74,6 +85,7 @@ impl Executions { pub fn configured( etcd: lance_context_merge::EtcdConfig, rollout: lance_context_merge::rollout::MergeRollout, + endpoint: Option, timeout_secs: u64, queue_timeout_secs: u64, idle_timeout_secs: u64, @@ -83,6 +95,10 @@ impl Executions { coordinator: OnceCell::new(), etcd: Some(etcd), rollout, + endpoint, + fencing_mode: OnceCell::new(), + mode_pending: std::sync::atomic::AtomicBool::new(false), + background_targets: Default::default(), timeout_secs, queue_timeout_secs, idle_timeout_secs, @@ -91,6 +107,59 @@ impl Executions { } } + /// Unknown mode defers maintenance, never ingestion. An activated protocol + /// is persisted so a configuration rollback cannot reopen untracked writes. + pub(crate) fn fencing_mode_known(&self) -> Option { + if !self.rollout.fence_legacy + && self.coordinator.get().is_none() + && self + .etcd + .as_ref() + .is_none_or(|c| c.etcd_endpoints.is_empty()) + { + return Some(false); + } + self.fencing_mode.get().copied() + } + + pub(crate) async fn fenced_mode(&self) -> Result { + if let Some(mode) = self.fencing_mode_known() { + return Ok(mode); + } + self.fencing_mode.get_or_try_init(|| async { + let mode = self.coordinator().await?.legacy_fencing_mode(self.rollout.fence_legacy) + .await.map_err(AppError::Internal)?; + if mode && self.endpoint.is_none() { + return Err(AppError::InvalidRequest("persisted merge protocol requires MERGE_EXECUTOR_ENDPOINT; refusing legacy rollback".into())); + } + Ok(mode) + }).await.copied() + } + + pub(crate) fn resolve_mode_in_background(state: Arc) { + use std::sync::atomic::Ordering; + if state.merge_executions.fencing_mode_known().is_some() + || state + .merge_executions + .mode_pending + .swap(true, Ordering::SeqCst) + { + return; + } + tokio::spawn(async move { + if let Err(error) = state.merge_executions.fenced_mode().await { + tracing::warn!( + ?error, + "merge protocol unresolved; ingestion continues, untracked merges deferred" + ); + } + state + .merge_executions + .mode_pending + .store(false, Ordering::SeqCst); + }); + } + pub(crate) fn instance(&self) -> &str { &self.instance } @@ -155,6 +224,7 @@ impl Executions { pub async fn capabilities( State(state): State>, ) -> Result, AppError> { + Executions::resolve_mode_in_background(state.clone()); Ok(Json( serde_json::json!({"protocol": 2, "instance": state.merge_executions.instance, "timeout_secs": state.merge_executions.timeout_secs, @@ -162,6 +232,7 @@ pub async fn capabilities( "idle_timeout_secs": state.merge_executions.idle_timeout_secs, "progress_protocol": 1, "legacy_progress_protocol": 1, + "fenced_legacy_merges": state.merge_executions.fencing_mode_known() == Some(true), "rollout_append_protocol": 1, "shard_name": state.instance_id, "owned_targets": state.merge_executions.rollout.owned_targets, @@ -174,7 +245,10 @@ pub async fn start( Json(execution): Json, ) -> Result { let coordinator = state.merge_executions.coordinator().await?; - if !state.merge_executions.owned(&execution.target) + if (!state.merge_executions.owned(&execution.target) + && !state.merge_executions.fenced_mode().await?) + || state.merge_executions.rollout.draining(&execution.target) + || execution.background || execution.maintenance.is_some() || execution.protocol != 2 || execution.instance != state.merge_executions.instance @@ -205,7 +279,14 @@ pub async fn start( // never leave a Running fence without an executor (or cancel a live write). let owned = state.clone(); tokio::spawn(async move { - let result = run(owned.clone(), coordinator, execution.clone(), cancelled).await; + let result = run( + owned.clone(), + coordinator, + execution.clone(), + cancelled, + None, + ) + .await; if let Err(error) = result { tracing::error!(id = %execution.id, target = %execution.target, %error, "merge execution ownership unresolved"); } @@ -224,6 +305,7 @@ async fn run( coordinator: Coordinator, execution: Execution, cancelled: watch::Receiver, + mut slot: Option, ) -> Result<(), String> { // CAS errors are ambiguous: never retry storage work. Reconciliation sees // the durable fence, so an uncertain admission cannot release ownership. @@ -246,7 +328,6 @@ async fn run( tokio::time::sleep(Duration::from_secs(1)).await; }; let target = execution.target.clone(); - let mut slot = None; let uri = match target.strip_prefix("generic:") { Some(name) => state.generic_uri(name), None => state.rollout_uri(&target), @@ -293,13 +374,15 @@ async fn run( // not exceed the compiler's async layout recursion limit. let execute: futures::future::BoxFuture<'_, Result> = Box::pin(async { // No payload work starts before admission to the process-wide slot. - slot = execute_scoped( - async { Ok(state.acquire_merge_slot().await) }, - Duration::from_secs(execution.queue_timeout_secs), - cancelled.clone(), - ) - .await - .map_err(|e| format!("merge queue wait: {e}"))?; + if slot.is_none() { + slot = execute_scoped( + async { Ok(state.acquire_merge_slot().await) }, + Duration::from_secs(execution.queue_timeout_secs), + cancelled.clone(), + ) + .await + .map_err(|e| format!("merge queue wait: {e}"))?; + } if !coordinator.publish_progress(&running, 0).await? { return Err("merge ownership revoked before execution".into()); } @@ -479,7 +562,7 @@ mod tests { use tokio::io::AsyncWriteExt; #[tokio::test] - async fn unavailable_etcd_does_not_block_startup_or_disable_legacy_self_merge() { + async fn unavailable_etcd_does_not_block_startup_or_flush_configuration() { use clap::Parser; let dir = tempfile::tempdir().unwrap(); let config = crate::config::ServerConfig::parse_from([ diff --git a/crates/lance-context-server/src/merge_execution/background.rs b/crates/lance-context-server/src/merge_execution/background.rs new file mode 100644 index 00000000..bdcf4071 --- /dev/null +++ b/crates/lance-context-server/src/merge_execution/background.rs @@ -0,0 +1,484 @@ +//! Count/timer and direct HTTP merges retain their executor independently of +//! callers. Admission uses the same durable owner and commit guard as fanout. +use super::*; +use tokio::sync::{oneshot, OwnedSemaphorePermit}; + +type Reply = oneshot::Receiver>; + +struct TargetGuard { + state: Arc, + target: String, +} +impl Drop for TargetGuard { + fn drop(&mut self) { + self.state + .merge_executions + .background_targets + .lock() + .unwrap() + .remove(&self.target); + } +} + +/// Never wait for a slot or coordinator IO on the flush sweep. Coalescing and +/// the pre-acquired process slot bound both admission requests and executors. +/// Dropping the returned receiver does not drop an admitted storage write. +pub(crate) async fn spawn_background( + state: Arc, + target: String, +) -> Result { + state.merge_executions.legacy_allowed(&target)?; + if !state.merge_executions.rollout.fence_legacy + && state.merge_executions.fencing_mode_known() != Some(true) + { + return Err(AppError::InvalidRequest( + "background fencing is disabled".into(), + )); + } + let endpoint = state.merge_executions.endpoint.clone().ok_or_else(|| { + AppError::InvalidRequest("background merge executor endpoint is missing".into()) + })?; + let slot = state + .merge_slots + .as_ref() + .ok_or_else(|| AppError::InvalidRequest("background merges require bounded slots".into()))? + .clone() + .try_acquire_owned() + .map_err(|_| AppError::Overloaded("merge slots are occupied".into()))?; + if !state + .merge_executions + .background_targets + .lock() + .unwrap() + .insert(target.clone()) + { + return Err(AppError::Overloaded( + "table already has a local merge admission".into(), + )); + } + let guard = TargetGuard { + state: state.clone(), + target: target.clone(), + }; + let mut execution = Execution::new( + &target, + &endpoint, + state.merge_executions.instance(), + state.merge_executions.timeout_secs, + ); + execution.background = true; + execution.queue_timeout_secs = state.merge_executions.queue_timeout_secs; + execution.idle_timeout_secs = state.merge_executions.idle_timeout_secs; + let (cancel, cancelled) = watch::channel(false); + state + .merge_executions + .running + .lock() + .await + .insert(execution.id.clone(), cancel); + let (reply, result) = oneshot::channel(); + // Retain reconciliation BEFORE the first fallible coordinator operation. + tokio::spawn(async move { + let _guard = guard; + let outcome = std::panic::AssertUnwindSafe(execute_background( + state.clone(), + &execution, + cancelled, + slot, + )) + .catch_unwind() + .await + .unwrap_or_else(|_| Err("background executor panicked; durable fence retained".into())); + if let Err(error) = &outcome { + tracing::warn!(target = %target, id = %execution.id, %error, "background merge did not complete"); + } + state + .merge_executions + .running + .lock() + .await + .remove(&execution.id); + let _ = reply.send(outcome); + }); + Ok(result) +} + +async fn execute_background( + state: Arc, + execution: &Execution, + cancelled: watch::Receiver, + slot: OwnedSemaphorePermit, +) -> Result { + if !state + .merge_executions + .fenced_mode() + .await + .map_err(|e| format!("{e:?}"))? + { + return Err("fenced merge protocol is not active".into()); + } + let coordinator = state + .merge_executions + .coordinator() + .await + .map_err(|e| format!("{e:?}"))?; + if *cancelled.borrow() { + return Err("background admission cancelled before reservation".into()); + } + loop { + match coordinator.reserve_background(execution).await { + Ok(true) => break, + outcome => { + match coordinator.get(&execution.target).await { + Ok(Some(current)) if current.id == execution.id => break, + Ok(_) if matches!(outcome, Ok(false)) => { + return Err("background merge deferred by ownership, admission gate or retry budget".into()); + } + _ => { + // An uncertain transaction may have admitted this exact + // execution. Never start an untracked fallback or forget it. + tokio::time::sleep(Duration::from_secs(1)).await; + } + } + } + } + } + run( + state, + coordinator.clone(), + execution.clone(), + cancelled, + Some(slot), + ) + .await?; + loop { + match coordinator.get(&execution.target).await { + Ok(Some(current)) if current.id == execution.id => { + if !matches!(current.phase, Phase::Finished | Phase::Recovered) { + return Err( + "background merge needs master storage recovery; fence retained".into(), + ); + } + match coordinator.release_background(¤t).await { + // False preserves a terminal record for an existing master + // claim. Only that master may transfer the target lock. + Ok(_) => return current.error.map_or(Ok(current.reclaimed), Err), + Err(error) => { + tracing::warn!(id = %execution.id, %error, "retrying background terminal release") + } + } + } + Ok(_) => return Err("background execution reconciled by another owner".into()), + Err(error) => { + tracing::warn!(id = %execution.id, %error, "reconciling background terminal outcome") + } + } + tokio::time::sleep(Duration::from_secs(1)).await; + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::sweeper; + use lance_context_core::{rollout_record_from_add_request, RolloutStore}; + use tokio::sync::RwLock; + + async fn fixture(slots: usize, idle: u64) -> (Arc, Coordinator, tempfile::TempDir) { + let client = + etcd_client::Client::connect([std::env::var("ETCD_TEST_ENDPOINTS").unwrap()], None) + .await + .unwrap(); + let coordinator = Coordinator::new( + client, + format!("/background-runtime/{}", uuid::Uuid::new_v4()), + ); + let dir = tempfile::tempdir().unwrap(); + let mut state = AppState::new_for_test(dir.path().to_path_buf()).await; + state.merge_slots = Some(Arc::new(tokio::sync::Semaphore::new(slots))); + state.rollout_merge_after_generations = 1; + state.merge_executions = Executions::new(Some(coordinator.clone()), 3600); + state.merge_executions.rollout.fence_legacy = true; + state.merge_executions.endpoint = Some("http://127.0.0.1:1".into()); + state.merge_executions.idle_timeout_secs = idle; + (Arc::new(state), coordinator, dir) + } + + async fn pending(state: &Arc, name: &str) -> Arc> { + let _ = rollouts::create_rollout_store( + State(state.clone()), + Json(lance_context_api::CreateRolloutStoreRequest { + name: name.into(), + storage_options: None, + }), + ) + .await + .unwrap(); + let store = state.get_or_open_rollout_store(name).await.unwrap(); + let row = rollout_record_from_add_request(&lance_context_api::AddRolloutRequest { + id: "row".into(), + rollout_id: "trajectory".into(), + ..Default::default() + }); + store.read().await.add(&[row]).await.unwrap(); + store.read().await.flush().await.unwrap(); + store + } + + async fn running(coordinator: &Coordinator, target: &str) -> Execution { + tokio::time::timeout(Duration::from_secs(10), async { + loop { + if let Some(e) = coordinator.get(target).await.unwrap() { + if e.phase == Phase::Running { + return e; + } + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + }) + .await + .unwrap() + } + + #[tokio::test] + #[ignore = "requires isolated local ETCD_TEST_ENDPOINTS"] + async fn disconnected_http_keeps_admitted_executor_and_reclaims_real_wal() { + let (state, coordinator, _dir) = fixture(1, 60).await; + let store = pending(&state, "hot").await; + let held = store.write().await; + let caller = tokio::spawn(rollouts::merge_wal( + State(state.clone()), + axum::extract::Path("hot".into()), + )); + let e = running(&coordinator, "hot").await; + assert!(e.background); + caller.abort(); + assert!(caller.await.unwrap_err().is_cancelled()); + assert_eq!(state.merge_slots.as_ref().unwrap().available_permits(), 0); + assert_eq!(coordinator.get("hot").await.unwrap().unwrap().id, e.id); + drop(held); + tokio::time::timeout(Duration::from_secs(20), async { + loop { + if !state + .merge_executions + .background_targets + .lock() + .unwrap() + .contains("hot") + { + break; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + }) + .await + .unwrap(); + assert!(coordinator.get("hot").await.unwrap().is_none()); + assert_eq!( + store.read().await.pending_wal_generations().await.unwrap(), + 0 + ); + assert!(store.read().await.get_by_id("row").await.unwrap().is_some()); + assert_eq!(state.merge_slots.as_ref().unwrap().available_permits(), 1); + } + + #[tokio::test] + #[ignore = "requires isolated local ETCD_TEST_ENDPOINTS"] + async fn count_and_timer_triggers_execute_without_holding_flush_sweep() { + let (state, coordinator, _dir) = fixture(1, 60).await; + let store = pending(&state, "hot").await; + // A reader permits the count predicate and preparation, but blocks + // commit. The sweeper must return while the owned executor waits. + let held = store.read().await; + tokio::time::timeout( + Duration::from_secs(2), + sweeper::flush_pass_coordinated( + vec![("hot".into(), store.clone())], + Duration::from_secs(5), + state.merge_slots.clone(), + Some(state.clone()), + ), + ) + .await + .unwrap(); + let e = running(&coordinator, "hot").await; + tokio::time::timeout( + Duration::from_secs(2), + sweeper::merge_pass_coordinated( + vec![("hot".into(), store.clone())], + Duration::from_secs(5), + Some(state.clone()), + ), + ) + .await + .unwrap(); + assert_eq!(coordinator.get("hot").await.unwrap().unwrap().id, e.id); + drop(held); + tokio::time::timeout(Duration::from_secs(20), async { + while store.read().await.pending_wal_generations().await.unwrap() != 0 { + tokio::time::sleep(Duration::from_millis(20)).await; + } + }) + .await + .unwrap(); + state.merge_executions.shutdown().await; + assert!(coordinator.get("hot").await.unwrap().is_none()); + } + + #[tokio::test] + #[ignore = "requires isolated local ETCD_TEST_ENDPOINTS"] + async fn busy_slots_do_not_create_unbounded_admissions_or_hide_other_flushes() { + let (state, coordinator, _dir) = fixture(1, 60).await; + let store = pending(&state, "hot").await; + let held = state.acquire_merge_slot().await; + for _ in 0..32 { + assert!(matches!( + spawn_background(state.clone(), "hot".into()).await, + Err(AppError::Overloaded(_)) + )); + } + tokio::time::timeout( + Duration::from_secs(2), + sweeper::flush_pass_coordinated( + vec![("hot".into(), store.clone())], + Duration::from_secs(5), + state.merge_slots.clone(), + Some(state.clone()), + ), + ) + .await + .unwrap(); + assert!(state.merge_executions.running.lock().await.is_empty()); + assert!(coordinator.get("hot").await.unwrap().is_none()); + assert_eq!( + store.read().await.pending_wal_generations().await.unwrap(), + 1 + ); + drop(held); + assert_eq!( + spawn_background(state.clone(), "hot".into()) + .await + .unwrap() + .await + .unwrap() + .unwrap(), + 1 + ); + } + + #[tokio::test] + #[ignore = "requires isolated local ETCD_TEST_ENDPOINTS"] + async fn another_tables_success_does_not_renew_a_stalled_executor() { + let (state, coordinator, _dir) = fixture(2, 2).await; + let slow = pending(&state, "slow").await; + let fast = pending(&state, "fast").await; + let held = slow.write().await; + let stalled = spawn_background(state.clone(), "slow".into()) + .await + .unwrap(); + running(&coordinator, "slow").await; + let stop = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let commits = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let healthy = tokio::spawn({ + let state = state.clone(); + let stop = stop.clone(); + let commits = commits.clone(); + async move { + while !stop.load(std::sync::atomic::Ordering::SeqCst) { + let reclaimed = spawn_background(state.clone(), "fast".into()) + .await + .unwrap() + .await + .unwrap() + .unwrap(); + assert!(reclaimed > 0); + let id = commits.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + let row = + rollout_record_from_add_request(&lance_context_api::AddRolloutRequest { + id: format!("row-{id}"), + rollout_id: "trajectory".into(), + ..Default::default() + }); + fast.read().await.add(&[row]).await.unwrap(); + fast.read().await.flush().await.unwrap(); + tokio::time::sleep(Duration::from_millis(100)).await; + } + } + }); + let error = tokio::time::timeout(Duration::from_secs(10), stalled) + .await + .unwrap() + .unwrap() + .unwrap_err(); + // Real commits continue throughout the stalled table's idle window. + assert!(!healthy.is_finished()); + stop.store(true, std::sync::atomic::Ordering::SeqCst); + healthy.await.unwrap(); + assert!(commits.load(std::sync::atomic::Ordering::SeqCst) >= 2); + assert!(error.contains("no-progress"), "{error}"); + assert!(coordinator.get("slow").await.unwrap().is_none()); + drop(held); + assert_eq!( + slow.read().await.pending_wal_generations().await.unwrap(), + 1 + ); + assert_eq!(state.merge_slots.as_ref().unwrap().available_permits(), 2); + } + #[tokio::test] + #[ignore = "requires isolated local ETCD_TEST_ENDPOINTS"] + async fn restarted_worker_keeps_fencing_after_flag_rollback() { + let (state, coordinator, dir) = fixture(1, 60).await; + let store = pending(&state, "hot").await; + assert!(state.merge_executions.fenced_mode().await.unwrap()); + let mut restarted = AppState::new_for_test(dir.path().to_path_buf()).await; + restarted.merge_slots = Some(Arc::new(tokio::sync::Semaphore::new(1))); + restarted.merge_executions = Executions::new(Some(coordinator.clone()), 3600); + restarted.merge_executions.endpoint = Some("http://127.0.0.1:1".into()); + assert!(!restarted.merge_executions.rollout.fence_legacy); + assert!(restarted.merge_executions.fenced_mode().await.unwrap()); + let restarted = Arc::new(restarted); + let held = store.write().await; + // The reopened worker has its own handle. Hold its lock to inspect the + // actual HTTP admission before allowing storage work to run. + let reopened = restarted.get_or_open_rollout_store("hot").await.unwrap(); + let blocked = reopened.write().await; + let task = tokio::spawn(rollouts::merge_wal( + State(restarted.clone()), + axum::extract::Path("hot".into()), + )); + assert!(running(&coordinator, "hot").await.background); + drop(blocked); + drop(held); + let Json(result) = task.await.unwrap().unwrap(); + assert_eq!(result.reclaimed, 1); + assert!(coordinator.get("hot").await.unwrap().is_none()); + let missing_endpoint = Executions::new(Some(coordinator), 3600); + assert!( + missing_endpoint.fenced_mode().await.is_err(), + "rollback must not silently restore legacy writes" + ); + } + #[tokio::test] + #[ignore = "requires isolated local ETCD_TEST_ENDPOINTS"] + async fn fenced_mode_blocks_untracked_worker_compaction_before_store_lock() { + let (state, _coordinator, _dir) = fixture(1, 60).await; + let store = pending(&state, "hot").await; + let held = store.read().await; + let result = tokio::time::timeout( + Duration::from_secs(2), + rollouts::compact_rollout( + State(state.clone()), + axum::extract::Path("hot".into()), + Json(serde_json::from_value(serde_json::json!({})).unwrap()), + ), + ) + .await + .unwrap(); + assert!(matches!(result, Err(AppError::Overloaded(_)))); + drop(held); + assert_eq!( + store.read().await.pending_wal_generations().await.unwrap(), + 1 + ); + } +} diff --git a/crates/lance-context-server/src/routes/generic.rs b/crates/lance-context-server/src/routes/generic.rs index 07f3b431..c3b26dfb 100644 --- a/crates/lance-context-server/src/routes/generic.rs +++ b/crates/lance-context-server/src/routes/generic.rs @@ -297,6 +297,15 @@ pub async fn merge_generic_wal( state .merge_executions .legacy_allowed(&format!("generic:{name}"))?; + if state.merge_executions.fenced_mode().await? { + let result = + crate::merge_execution::spawn_background(state, format!("generic:{name}")).await?; + let reclaimed = result + .await + .map_err(|e| AppError::Internal(e.to_string()))? + .map_err(AppError::Internal)?; + return Ok(Json(serde_json::json!({"reclaimed": reclaimed}))); + } let execution_state = state.clone(); let execution_name = name.clone(); state diff --git a/crates/lance-context-server/src/routes/rollouts.rs b/crates/lance-context-server/src/routes/rollouts.rs index 8df95141..4a9b29ef 100644 --- a/crates/lance-context-server/src/routes/rollouts.rs +++ b/crates/lance-context-server/src/routes/rollouts.rs @@ -641,6 +641,11 @@ pub async fn compact_rollout( // Once a table is draining or owned, only the master's fenced maintenance // path may publish a rewrite. Check before opening or locking its handle. state.merge_executions.legacy_allowed(&name)?; + if state.merge_executions.fenced_mode().await? { + return Err(AppError::Overloaded( + "fenced table maintenance must run through the master".into(), + )); + } let store_lock = state.get_or_open_rollout_store(&name).await?; let config = if req.target_rows_per_fragment.is_some() || req.materialize_deletions.is_some() { @@ -715,6 +720,14 @@ pub async fn merge_wal( Path(name): Path, ) -> Result, AppError> { state.merge_executions.legacy_allowed(&name)?; + if state.merge_executions.fenced_mode().await? { + let result = crate::merge_execution::spawn_background(state, name).await?; + let reclaimed = result + .await + .map_err(|e| AppError::Internal(e.to_string()))? + .map_err(AppError::Internal)?; + return Ok(Json(MergeWalResponse { reclaimed })); + } let execution_state = state.clone(); let execution_name = name.clone(); state diff --git a/crates/lance-context-server/src/state.rs b/crates/lance-context-server/src/state.rs index cebd07c5..122080c3 100644 --- a/crates/lance-context-server/src/state.rs +++ b/crates/lance-context-server/src/state.rs @@ -364,10 +364,31 @@ impl AppState { "merge execution, queue and idle timeouts must be positive".into(), )); } + if config.merge_rollout.fence_legacy || config.merge_executor_endpoint.is_some() { + let endpoint = config.merge_executor_endpoint.as_deref().unwrap_or(""); + let valid_endpoint = endpoint.parse::().is_ok_and(|url| { + matches!(url.scheme_str(), Some("http" | "https")) + && url + .authority() + .is_some_and(|a| !a.host().is_empty() && !a.as_str().contains('@')) + && url.query().is_none() + && matches!(url.path(), "" | "/") + && !endpoint.contains('#') + }); + if !valid_endpoint + || config.etcd.etcd_endpoints.is_empty() + || config.rollout_merge_concurrency == 0 + { + return Err(AppError::InvalidRequest( + "fenced legacy merges require ETCD_ENDPOINTS, a stable MERGE_EXECUTOR_ENDPOINT and positive ROLLOUT_MERGE_CONCURRENCY".into(), + )); + } + } Ok(Self { merge_executions: crate::merge_execution::Executions::configured( config.merge_etcd_config(), config.merge_rollout.clone(), + config.merge_executor_endpoint.clone(), config.merge_execution_timeout_secs, config.merge_queue_timeout_secs, config.merge_idle_timeout_secs, diff --git a/crates/lance-context-server/src/sweeper.rs b/crates/lance-context-server/src/sweeper.rs index 66449ee0..d7368a8b 100644 --- a/crates/lance-context-server/src/sweeper.rs +++ b/crates/lance-context-server/src/sweeper.rs @@ -481,7 +481,25 @@ async fn route_merge( // unchanged. spawn_publish_demand(state.clone(), target.clone(), S::clone(store)); if !state.merge_executions.owned(&target) { - return Ok(false); + match state.merge_executions.fencing_mode_known() { + Some(false) => return Ok(false), + None if !state.merge_executions.rollout.fence_legacy => { + crate::merge_execution::Executions::resolve_mode_in_background(state.clone()); + return Ok(true); + } + _ => {} + } + if !count_trigger || store.count_merge_due().await? { + match crate::merge_execution::spawn_background(state.clone(), target).await { + Ok(_reply) => {} // Executor is retained independently of this sweep. + Err(crate::error::AppError::Overloaded(_)) => { + metrics::counter!("rollout_wal_self_merge_skipped_total", "kind" => S::kind()) + .increment(1); + } + Err(error) => return Err(format!("{error:?}")), + } + } + return Ok(true); } if !count_trigger || store.count_merge_due().await? { state.merge_executions.request_merge(&target).await?; diff --git a/docs/worker-merge-admission.md b/docs/worker-merge-admission.md new file mode 100644 index 00000000..906bdfb2 --- /dev/null +++ b/docs/worker-merge-admission.md @@ -0,0 +1,69 @@ +# Durable worker merge admission (integration in progress) + +Worker count-triggered and timer-triggered merges previously wrote without a +persistent execution record. A master could observe a hung request but could +not establish what that writer might still commit. This path records those +executions before storage work and reuses the owned executor's useful-progress +watchdog, conditional commit permits, cancellation and storage fencing. + +## Behavior + +`MERGE_FENCE_LEGACY=true` enables the protocol for new rollout and generic worker +merges, independently of the static native-ownership list. It requires a stable +`MERGE_EXECUTOR_ENDPOINT`, private etcd configuration, and positive +`ROLLOUT_MERGE_CONCURRENCY`. Master fanout remains serial. Ordinary HTTP merges, +count triggers and cleanup timers use the same process slots and executor. + +A background trigger tries a slot before spawning; it never waits for capacity +or etcd on the flush sweep. Per-table admission is coalesced. The detached +executor exists before admission IO, so a disconnected HTTP caller cannot +abandon a writer. Its slot remains occupied until its storage leaves drain or +are fenced. An ambiguous terminal record is retained for master recovery. + +Admission atomically persists the execution, exclusive owner and coalesced +request. It respects canonical claims, catchup ownership, prepared-commit +priority, admission closure and persisted failure budgets. A terminal worker +cannot delete an owner being transferred to a waiting master claim. + +The master inventories worker records as well as maintenance records, including +records whose original worker is no longer configured. It observes actual +execution progress before cancellation, reconciles ambiguous commits through +the existing storage barrier, and never treats an empty endpoint list as a +successful catchup. New-table demand can enqueue fenced worker fanout without +adding a static owned target. + +Worker HTTP compaction is refused in this mode; tracked master maintenance must +handle it. This prevents an untracked worker rewrite from bypassing merge +ownership. + +## Persisted protocol and startup + +The private-etcd `worker-merge-admission-mode` key is persistent and unleased. +Once activated, disabling the initiation flag on a restarted participant does +not restore legacy writes. An unknown newer marker is rejected, including when +activation is requested. The master resolves this mode at startup. Workers +resolve it independently of startup/flush; unresolved metadata defers maintenance +and cannot authorize a legacy fallback. Participants without etcd and without +the opt-in keep standalone legacy behavior. + +A worker that has resolved legacy mode retains that mode for its incarnation. +Consequently, initial activation requires a coordinated transition covering +all old worker/master instances and external helpers. A marker, a Pod restart, +or an empty execution inventory does not prove that historical writers are +quiescent. Removing the shared etcd namespace/configuration or deploying an old +binary is not a supported rollback after activation. + +## Remaining native handoff requirements + +This implementation does not certify the initial transition, dynamically add +native ownership, or authorize native writes solely because admission closed. +Before production activation, finish and validate: + +- Complete old-writer coverage, including helpers and master compact/index/repair. +- Durable native enrollment after recorded writers finish or are storage-fenced. +- Automatic new-hot-table selection and bounded continuous native catchup. +- Combined staging of worker admission, tracked master maintenance and native + execution, followed by measured production handoff. + +Keep this integration disabled until those requirements are implemented. The +current production helpers remain necessary until the handoff is demonstrated. From e777f033d9dfb0efbeb84dcb6baf9b9e07fa5380 Mon Sep 17 00:00:00 2001 From: Beinan Wang Date: Tue, 20 Oct 2026 23:59:43 +0000 Subject: [PATCH 04/10] Route new table demand to resident masters after fenced cutover --- .../lance-context-master/src/eligibility.rs | 48 ++++- .../lance-context-master/src/local_merge.rs | 53 +++++- .../src/merge_execution.rs | 18 +- .../src/resident_recovery.rs | 171 ++++++++++++++++++ crates/lance-context-master/src/scheduler.rs | 5 +- crates/lance-context-master/src/state.rs | 5 +- crates/lance-context-master/src/task_store.rs | 3 + .../src/merge_execution.rs | 8 +- .../src/merge_execution/background.rs | 76 +++++++- .../src/routes/generic.rs | 4 + .../src/routes/rollouts.rs | 1 + crates/lance-context-server/src/sweeper.rs | 1 + docs/worker-merge-admission.md | 42 +++-- 13 files changed, 406 insertions(+), 29 deletions(-) diff --git a/crates/lance-context-master/src/eligibility.rs b/crates/lance-context-master/src/eligibility.rs index 611e98dc..ea623c75 100644 --- a/crates/lance-context-master/src/eligibility.rs +++ b/crates/lance-context-master/src/eligibility.rs @@ -31,6 +31,7 @@ pub enum Skip { /// Static per-target scheduling configuration consulted by the filter. pub struct Policy<'a> { + pub auto_local_merge: bool, pub rollout: &'a MergeRollout, pub compaction_prepare_targets: &'a [String], pub index_prepare_targets: &'a [String], @@ -89,10 +90,15 @@ pub fn pre_dependency_filter( } if let Some(targets) = resident_targets { let eligible = kind == TaskKind::MergeWal - && policy.rollout.owned(target) + && (policy.rollout.owned(target) || policy.auto_local_merge) + && !target.starts_with("datagen:") + && lance_context_core::validate_store_name( + target.strip_prefix("generic:").unwrap_or(target), + ) + .is_ok() && !policy.rollout.draining(target) && (targets.iter().any(|t| t == target) - || (!is_generic(target) && listed(targets, target))); + || ((!is_generic(target) || policy.auto_local_merge) && listed(targets, target))); if !eligible { return Err(Skip::NotResidentEligible); } @@ -210,6 +216,7 @@ mod tests { &["draining"], ); let policy = Policy { + auto_local_merge: false, rollout: &rollout, compaction_prepare_targets: &[], index_prepare_targets: &[], @@ -257,6 +264,7 @@ mod tests { let compact = strings(&["t", "d"]); let index = strings(&["*"]); let policy = Policy { + auto_local_merge: false, rollout: &rollout, compaction_prepare_targets: &compact, index_prepare_targets: &index, @@ -287,6 +295,7 @@ mod tests { let rollout = rollout(&["t"], &[]); let shared = strings(&["t"]); let policy = Policy { + auto_local_merge: false, rollout: &rollout, compaction_prepare_targets: &[], index_prepare_targets: &[], @@ -297,6 +306,7 @@ mod tests { assert!(!claim_shape(TaskKind::MergeWal, "t", &policy, true).shared_owner); let wildcard = strings(&["*"]); let policy = Policy { + auto_local_merge: false, maintenance_catchup_targets: &wildcard, ..policy }; @@ -538,4 +548,38 @@ mod tests { "preparation runs beside a worker execution" ); } + #[test] + fn automatic_resident_filter_admits_new_rollout_and_generic_but_keeps_drains() { + let rollout = rollout(&[], &["draining"]); + let policy = Policy { + auto_local_merge: true, + rollout: &rollout, + compaction_prepare_targets: &[], + index_prepare_targets: &[], + maintenance_catchup_targets: &[], + }; + let targets = strings(&["*"]); + for name in ["new-hot", "generic:new-hot"] { + assert_eq!( + pre_dependency_filter(TaskKind::MergeWal, name, |_| true, Some(&targets), &policy), + Ok(()) + ); + } + for name in ["draining", "datagen:new-hot", "../escape", "generic:"] { + assert_eq!( + pre_dependency_filter(TaskKind::MergeWal, name, |_| true, Some(&targets), &policy), + Err(Skip::NotResidentEligible) + ); + } + assert_eq!( + pre_dependency_filter( + TaskKind::Compact, + "new-hot", + |_| true, + Some(&targets), + &policy + ), + Err(Skip::NotResidentEligible) + ); + } } diff --git a/crates/lance-context-master/src/local_merge.rs b/crates/lance-context-master/src/local_merge.rs index ab1277f9..0d60554b 100644 --- a/crates/lance-context-master/src/local_merge.rs +++ b/crates/lance-context-master/src/local_merge.rs @@ -4,8 +4,15 @@ use std::sync::Arc; #[derive(Clone, Debug, Default, clap::Args)] pub struct LocalMergeConfig { + /// Route new rollout/generic demand to resident masters without a static + /// table allowlist. Enable only after every legacy writer/helper has drained + /// and all participants use persistent fenced admission. This flag is not + /// evidence of that initial transition; retain its deployment evidence. + #[arg(long, env = "MASTER_LOCAL_MERGE_AUTO", default_value_t = false)] + pub master_local_merge_auto: bool, /// Execute sealed WAL merges in the master, including generic upsert tables. - /// '*' selects only explicitly owned targets; it never migrates legacy writers. + /// '*' selects explicitly owned targets unless automatic routing is enabled. + /// Neither setting certifies that historical legacy writers have drained. #[arg(long, env = "MASTER_LOCAL_MERGE_TARGETS", value_delimiter = ',')] pub master_local_merge_targets: Vec, } @@ -23,7 +30,12 @@ pub(crate) fn enabled(config: &MasterConfig) -> bool { } pub(crate) fn selected(config: &MasterConfig, target: &str) -> bool { - config.merge_rollout.owned(target) + let name = target.strip_prefix("generic:").unwrap_or(target); + let dynamic = config.local_merge.master_local_merge_auto + && config.merge_rollout.fence_legacy + && !target.starts_with("datagen:") + && lance_context_core::validate_store_name(name).is_ok(); + (config.merge_rollout.owned(target) || dynamic) && !config.merge_rollout.draining(target) && (config.append.local(target) || config.local_merge.selected(target)) } @@ -31,6 +43,9 @@ pub(crate) fn selected(config: &MasterConfig, target: &str) -> bool { /// 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 { + if config.local_merge.master_local_merge_auto { + return vec!["*".into()]; + } config .merge_rollout .owned_targets @@ -41,6 +56,13 @@ pub(crate) fn targets(config: &MasterConfig) -> Vec { } pub(crate) fn validate(config: &MasterConfig) -> Result<(), String> { + if config.local_merge.master_local_merge_auto + && (!config.merge_rollout.fence_legacy + || config.local_merge.master_local_merge_targets != ["*"] + || !config.maintenance.fence_legacy_maintenance) + { + return Err("automatic local merges require persistent worker fencing, fenced master maintenance and MASTER_LOCAL_MERGE_TARGETS=*; initial legacy writer drain must be verified before enabling".into()); + } if config.local_merge.master_local_merge_targets.is_empty() { return Ok(()); } @@ -58,9 +80,11 @@ pub(crate) fn validate(config: &MasterConfig) -> Result<(), String> { { return Err("master local merges require stable shards, bounded passes and a sufficient shared memory budget".into()); } - if targets(config) + if config + .merge_rollout + .owned_targets .iter() - .any(|target| config.append.enabled(target)) + .any(|target| selected(config, target) && config.append.enabled(target)) { config.append.validate()?; if config.append.rollout_append_local_memory_bytes < config.append.rollout_append_max_bytes @@ -134,4 +158,25 @@ mod tests { config.append.rollout_append_local_memory_bytes + 1; assert!(validate(&config).is_err()); } + #[test] + fn automatic_selection_requires_fenced_protocol_and_bounded_resources() { + let mut cfg = MasterConfig::parse_from(["test"]); + cfg.local_merge.master_local_merge_auto = true; + cfg.local_merge.master_local_merge_targets = vec!["*".into()]; + cfg.catchup.shards = vec!["writer".into()]; + assert!(validate(&cfg).is_err()); + assert!(!selected(&cfg, "new-hot")); + cfg.merge_rollout.fence_legacy = true; + assert!(validate(&cfg).is_err()); + cfg.maintenance.fence_legacy_maintenance = true; + assert!(validate(&cfg).is_ok()); + assert!(selected(&cfg, "new-hot")); + assert!(selected(&cfg, "generic:new-hot")); + assert!(!selected(&cfg, "datagen:new-hot")); + cfg.merge_rollout.drain_targets.push("new-hot".into()); + assert!(!selected(&cfg, "new-hot")); + assert_eq!(targets(&cfg), ["*"]); + cfg.catchup.merge_max_bytes = cfg.append.rollout_append_local_memory_bytes + 1; + assert!(validate(&cfg).is_err()); + } } diff --git a/crates/lance-context-master/src/merge_execution.rs b/crates/lance-context-master/src/merge_execution.rs index f0355cc4..c4d45d3f 100644 --- a/crates/lance-context-master/src/merge_execution.rs +++ b/crates/lance-context-master/src/merge_execution.rs @@ -81,14 +81,28 @@ pub(crate) async fn run_merge_wal( if state.config.append.enabled(target) || crate::local_merge::selected(&state.config, target) { if let Some(old) = coordinator.get(target).await? { ensure_recovery_due(&coordinator, &old).await?; - recover_execution(state, &coordinator, &proof, old).await?; + if old.maintenance.is_none() { + // A newly selected resident target may still have useful + // background work. Wait for its real progress and terminal + // acknowledgement before considering storage recovery. + if reconcile(&state.http, &coordinator, &proof, old, false) + .await + .is_err() + { + if let Some(current) = coordinator.get(target).await? { + recover_execution(state, &coordinator, &proof, current).await?; + } + } + } else { + recover_execution(state, &coordinator, &proof, old).await?; + } } return crate::maintenance_execution::run_as( state, claim, lance_context_merge::MaintenanceKind::Catchup, async { - if state.config.append.enabled(target) { + if state.config.merge_rollout.owned(target) && state.config.append.enabled(target) { crate::rollout_append::run(state, target).await } else { crate::local_merge::run(state, target).await diff --git a/crates/lance-context-master/src/resident_recovery.rs b/crates/lance-context-master/src/resident_recovery.rs index 2916ac87..76cf8a39 100644 --- a/crates/lance-context-master/src/resident_recovery.rs +++ b/crates/lance-context-master/src/resident_recovery.rs @@ -37,6 +37,8 @@ fn scope(config: &MasterConfig) -> u64 { sorted.hash(&mut hash); } config.append.rollout_append_local.hash(&mut hash); + config.local_merge.master_local_merge_auto.hash(&mut hash); + config.merge_rollout.fence_legacy.hash(&mut hash); hash.finish() } @@ -483,6 +485,10 @@ mod tests { } async fn generic_fixture() -> (tempfile::TempDir, Arc) { + generic_fixture_with_auto(false).await + } + + async fn generic_fixture_with_auto(auto: bool) -> (tempfile::TempDir, Arc) { let dir = tempfile::tempdir().unwrap(); let mut config = MasterConfig::parse_from(["test"]); config.data_dir = dir.path().to_string_lossy().into(); @@ -497,6 +503,11 @@ mod tests { ); config.merge_rollout.owned_targets = vec!["generic:a".into(), "generic:b".into()]; config.local_merge.master_local_merge_targets = vec!["*".into()]; + if auto { + config.merge_rollout.owned_targets.clear(); + config.merge_rollout.fence_legacy = true; + config.local_merge.master_local_merge_auto = true; + } config.catchup.shards = vec!["writer".into()]; config.catchup.merge_max_generations = 1; config.catchup.merge_max_bytes = 1024 * 1024; @@ -543,6 +554,166 @@ mod tests { writer.flush().await.unwrap(); } + #[tokio::test] + #[ignore = "requires ETCD_TEST_ENDPOINTS"] + async fn automatic_new_table_waits_for_background_then_catches_up_across_masters() { + use lance_context_merge::{Execution, Phase}; + let (_dir, state) = generic_fixture_with_auto(true).await; + assert!(state.config.merge_rollout.owned_targets.is_empty()); + assert!(state.config.maintenance.fence_legacy_maintenance); + // Created after master startup, with no stats row or static allowlist. + let writer = generic_writer(&state, "new-hot").await; + for n in 0..17 { + generic_write(&writer, "shared", &n.to_string()).await; + } + let target = "generic:new-hot"; + let c = state.task_store.merge_coordinator(); + let mut e = Execution::new(target, "http://127.0.0.1:1", "live-worker", 60); + e.background = true; + assert!(c.reserve_background(&e).await.unwrap()); + let running = c.start(&e).await.unwrap().unwrap(); + assert!(c.publish_progress(&running, 0).await.unwrap()); + let request = c.request_page(None).await.unwrap().0.remove(0); + crate::scheduler::enqueue_merge_request(&state, &c, &request) + .await + .unwrap(); + let claim = state + .task_store + .claim_resident_merge(&crate::local_merge::targets(&state.config)) + .await + .unwrap() + .unwrap(); + let next_master = MasterState::new(state.config.clone()).await.unwrap(); + assert!(next_master + .task_store + .claim_resident_merge(&crate::local_merge::targets(&next_master.config)) + .await + .unwrap() + .is_none()); + let first = tokio::spawn({ + let state = state.clone(); + async move { + let outcome = crate::merge_execution::run_merge_wal(&state, &claim).await; + (claim, outcome) + } + }); + for sequence in 1..=4 { + tokio::time::sleep(Duration::from_millis(100)).await; + let current = c.get(target).await.unwrap().unwrap(); + assert_eq!(current.id, running.id); + assert_eq!( + current.phase, + Phase::Running, + "native selection must preserve useful worker progress" + ); + assert!(c.publish_progress(&running, sequence).await.unwrap()); + assert!(!first.is_finished()); + } + assert!(c.finish(&running, Ok(0)).await.unwrap()); + let (claim, result) = tokio::time::timeout(Duration::from_secs(30), first) + .await + .unwrap() + .unwrap(); + assert!( + result.as_ref().unwrap().contains("16 generations"), + "{result:?}" + ); + let request = c.request_page(None).await.unwrap().0.remove(0); + crate::scheduler::enqueue_merge_request(&next_master, &c, &request) + .await + .unwrap(); + assert_eq!(c.request_page(None).await.unwrap().0.len(), 1); + state.task_store.finish(claim, result).await.unwrap(); + crate::scheduler::enqueue_merge_request(&next_master, &c, &request) + .await + .unwrap(); + let claim = next_master + .task_store + .claim_resident_merge(&crate::local_merge::targets(&next_master.config)) + .await + .unwrap() + .unwrap(); + let result = crate::merge_execution::run_merge_wal(&next_master, &claim).await; + assert!( + result.as_ref().unwrap().contains("1 generations"), + "{result:?}" + ); + next_master.task_store.finish(claim, result).await.unwrap(); + assert_eq!(writer.pending_wal_generations().await.unwrap(), 0); + let rows = writer.list(None, None).await.unwrap(); + assert_eq!(rows.len(), 1); + assert_eq!(rows[0]["text"], "16"); + assert_eq!(state.local_append.as_ref().unwrap().budget.reserved(), 0); + assert_eq!( + next_master.local_append.as_ref().unwrap().budget.reserved(), + 0 + ); + generic_write(&writer, "shared", "after-handoff").await; + assert!(probe(&next_master, target).await.unwrap()); + } + + #[tokio::test] + #[ignore = "requires ETCD_TEST_ENDPOINTS"] + async fn automatic_rollout_uses_native_merge_and_preserves_external_owner() { + let (_dir, initial) = generic_fixture_with_auto(true).await; + let mut cfg = initial.config.clone(); + // Existing append wildcard must not send an unowned new table through + // the explicit-ownership staged-append protocol. + cfg.append.rollout_append_targets = vec!["*".into()]; + let state = MasterState::new(cfg).await.unwrap(); + let target = "new-rollout"; + let writer = write(&state, target, 4).await; + let c = state.task_store.merge_coordinator(); + let owner = crate::catchup::store::active_key(&state.config.etcd.etcd_prefix, target); + let mut client = state.task_store.etcd_client().clone(); + client + .put(owner.as_str(), "fixture-helper", None) + .await + .unwrap(); + assert!(!probe(&state, target).await.unwrap()); + c.request_merge(target).await.unwrap(); + let request = c.request_page(None).await.unwrap().0.remove(0); + crate::scheduler::enqueue_merge_request(&state, &c, &request) + .await + .unwrap(); + assert!(state + .task_store + .claim_resident_merge(&crate::local_merge::targets(&state.config)) + .await + .unwrap() + .is_none()); + assert_eq!(writer.pending_wal_generations().await.unwrap(), 4); + // This fixture has no remote actor or commits. End its known ownership. + client.delete(owner, None).await.unwrap(); + let claim = state + .task_store + .claim_resident_merge(&crate::local_merge::targets(&state.config)) + .await + .unwrap() + .unwrap(); + let result = crate::merge_execution::run_merge_wal(&state, &claim).await; + assert!( + result + .as_ref() + .unwrap() + .contains("local WAL merge reclaimed 4 generations"), + "{result:?}" + ); + state.task_store.finish(claim, result).await.unwrap(); + assert_eq!(writer.pending_wal_generations().await.unwrap(), 0); + assert_eq!( + lance::Dataset::open(&state.rollout_uri(target)) + .await + .unwrap() + .count_rows(None) + .await + .unwrap(), + 4 + ); + assert_eq!(state.local_append.as_ref().unwrap().budget.reserved(), 0); + assert!(!probe(&state, target).await.unwrap()); + } + #[tokio::test] #[ignore = "requires ETCD_TEST_ENDPOINTS"] async fn generic_discovery_and_continuation_preserve_updates_without_workers() { diff --git a/crates/lance-context-master/src/scheduler.rs b/crates/lance-context-master/src/scheduler.rs index e35a15bf..8ccb3bc7 100644 --- a/crates/lance-context-master/src/scheduler.rs +++ b/crates/lance-context-master/src/scheduler.rs @@ -932,7 +932,10 @@ async fn should_probe_failure( state: &Arc, failure: &lance_context_merge::failure::ShardFailure, ) -> bool { - if state.config.merge_rollout.owned(&failure.target) { + if !state.config.merge_rollout.draining(&failure.target) + && (state.config.merge_rollout.owned(&failure.target) + || state.config.merge_rollout.fence_legacy) + { return true; } if state.config.merge_rollout.draining(&failure.target) { diff --git a/crates/lance-context-master/src/state.rs b/crates/lance-context-master/src/state.rs index 0a6f56dd..380a246a 100644 --- a/crates/lance-context-master/src/state.rs +++ b/crates/lance-context-master/src/state.rs @@ -138,7 +138,6 @@ impl MasterState { pub async fn new(mut config: MasterConfig) -> lance::Result> { config.merge_rollout.validate().map_err(lance::Error::io)?; config.catchup.validate().map_err(lance::Error::io)?; - crate::local_merge::validate(&config).map_err(lance::Error::io)?; config .append .validate_local(&config) @@ -157,6 +156,10 @@ impl MasterState { .legacy_fencing_mode(config.merge_rollout.fence_legacy) .await .map_err(lance::Error::io)?; + // Protocol activation covers every master maintenance writer as well + // as workers; a flag rollback must not restore unrecorded rewrites. + config.maintenance.fence_legacy_maintenance |= config.merge_rollout.fence_legacy; + crate::local_merge::validate(&config).map_err(lance::Error::io)?; // Serialize first-time registry/stats creation and legacy backfill in // etcd mode. Followers wait briefly rather than racing Lance creates. let dedicated = config.catchup.target.is_some(); diff --git a/crates/lance-context-master/src/task_store.rs b/crates/lance-context-master/src/task_store.rs index 339eeeed..4a9ed7f8 100644 --- a/crates/lance-context-master/src/task_store.rs +++ b/crates/lance-context-master/src/task_store.rs @@ -123,6 +123,7 @@ struct EtcdTaskStore { index_prepare_targets: Vec, maintenance_catchup_targets: Vec, rollout: lance_context_merge::rollout::MergeRollout, + auto_local_merge: bool, } /// Ownership of one running task. Dropping the claim stops lease renewal; etcd @@ -743,6 +744,7 @@ impl EtcdTaskStore { index_prepare_targets: config.maintenance.index_prepare_targets.clone(), maintenance_catchup_targets: config.maintenance.maintenance_catchup_targets.clone(), rollout: config.merge_rollout.clone(), + auto_local_merge: config.local_merge.master_local_merge_auto, }) } @@ -838,6 +840,7 @@ impl EtcdTaskStore { fn eligibility_policy(&self) -> eligibility::Policy<'_> { eligibility::Policy { + auto_local_merge: self.auto_local_merge, rollout: &self.rollout, compaction_prepare_targets: &self.prepare_targets, index_prepare_targets: &self.index_prepare_targets, diff --git a/crates/lance-context-server/src/merge_execution.rs b/crates/lance-context-server/src/merge_execution.rs index 0ce634a4..3b915644 100644 --- a/crates/lance-context-server/src/merge_execution.rs +++ b/crates/lance-context-server/src/merge_execution.rs @@ -1,7 +1,7 @@ //! HTTP handlers only admit/cancel work. The owned executor, not the connection, //! holds the durable execution fence until the scoped storage future is gone. mod background; -pub(crate) use background::spawn_background; +pub(crate) use background::{publish_background_request, spawn_background}; use crate::{ error::AppError, @@ -56,6 +56,8 @@ pub struct Executions { fencing_mode: OnceCell, mode_pending: std::sync::atomic::AtomicBool, background_targets: std::sync::Mutex>, + request_targets: std::sync::Mutex>, + request_slots: Arc, queue_timeout_secs: u64, idle_timeout_secs: u64, instance: String, @@ -75,6 +77,8 @@ impl Executions { fencing_mode: OnceCell::new(), mode_pending: std::sync::atomic::AtomicBool::new(false), background_targets: Default::default(), + request_targets: Default::default(), + request_slots: Arc::new(tokio::sync::Semaphore::new(4)), queue_timeout_secs: 600, idle_timeout_secs: 600, instance: uuid::Uuid::new_v4().to_string(), @@ -99,6 +103,8 @@ impl Executions { fencing_mode: OnceCell::new(), mode_pending: std::sync::atomic::AtomicBool::new(false), background_targets: Default::default(), + request_targets: Default::default(), + request_slots: Arc::new(tokio::sync::Semaphore::new(4)), timeout_secs, queue_timeout_secs, idle_timeout_secs, diff --git a/crates/lance-context-server/src/merge_execution/background.rs b/crates/lance-context-server/src/merge_execution/background.rs index bdcf4071..eeba3483 100644 --- a/crates/lance-context-server/src/merge_execution/background.rs +++ b/crates/lance-context-server/src/merge_execution/background.rs @@ -8,18 +8,64 @@ type Reply = oneshot::Receiver>; struct TargetGuard { state: Arc, target: String, + request: bool, } impl Drop for TargetGuard { fn drop(&mut self) { - self.state - .merge_executions - .background_targets - .lock() - .unwrap() - .remove(&self.target); + let targets = if self.request { + &self.state.merge_executions.request_targets + } else { + &self.state.merge_executions.background_targets + }; + targets.lock().unwrap().remove(&self.target); } } +/// Demand must reach the master even when every worker execution slot is +/// occupied. Metadata publication has its own small bound and per-table +/// coalescing; no payload slot or flush-pass wait is needed. +pub(crate) fn publish_background_request(state: Arc, target: String) { + let Ok(permit) = state + .merge_executions + .request_slots + .clone() + .try_acquire_owned() + else { + return; + }; + if !state + .merge_executions + .request_targets + .lock() + .unwrap() + .insert(target.clone()) + { + return; + } + let guard = TargetGuard { + state: state.clone(), + target: target.clone(), + request: true, + }; + tokio::spawn(async move { + let _permit = permit; + let _guard = guard; + let result = tokio::time::timeout( + Duration::from_secs(2), + state.merge_executions.request_merge(&target), + ) + .await; + let label = match result { + Ok(Ok(())) => "ok", + outcome => { + tracing::warn!(target, ?outcome, "master merge demand publication deferred"); + "failed" + } + }; + metrics::counter!("rollout_wal_background_requests_total", "result" => label).increment(1); + }); +} + /// Never wait for a slot or coordinator IO on the flush sweep. Coalescing and /// the pre-acquired process slot bound both admission requests and executors. /// Dropping the returned receiver does not drop an admitted storage write. @@ -59,6 +105,7 @@ pub(crate) async fn spawn_background( let guard = TargetGuard { state: state.clone(), target: target.clone(), + request: false, }; let mut execution = Execution::new( &target, @@ -349,6 +396,23 @@ mod tests { .await .unwrap(); assert!(state.merge_executions.running.lock().await.is_empty()); + tokio::time::timeout(Duration::from_secs(3), async { + loop { + if coordinator + .request_page(None) + .await + .unwrap() + .0 + .iter() + .any(|r| r.target == "hot") + { + break; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + }) + .await + .expect("saturated worker must still request master catchup"); assert!(coordinator.get("hot").await.unwrap().is_none()); assert_eq!( store.read().await.pending_wal_generations().await.unwrap(), diff --git a/crates/lance-context-server/src/routes/generic.rs b/crates/lance-context-server/src/routes/generic.rs index c3b26dfb..287ba027 100644 --- a/crates/lance-context-server/src/routes/generic.rs +++ b/crates/lance-context-server/src/routes/generic.rs @@ -298,6 +298,10 @@ pub async fn merge_generic_wal( .merge_executions .legacy_allowed(&format!("generic:{name}"))?; if state.merge_executions.fenced_mode().await? { + crate::merge_execution::publish_background_request( + state.clone(), + format!("generic:{name}"), + ); let result = crate::merge_execution::spawn_background(state, format!("generic:{name}")).await?; let reclaimed = result diff --git a/crates/lance-context-server/src/routes/rollouts.rs b/crates/lance-context-server/src/routes/rollouts.rs index 4a9b29ef..1b00fc18 100644 --- a/crates/lance-context-server/src/routes/rollouts.rs +++ b/crates/lance-context-server/src/routes/rollouts.rs @@ -721,6 +721,7 @@ pub async fn merge_wal( ) -> Result, AppError> { state.merge_executions.legacy_allowed(&name)?; if state.merge_executions.fenced_mode().await? { + crate::merge_execution::publish_background_request(state.clone(), name.clone()); let result = crate::merge_execution::spawn_background(state, name).await?; let reclaimed = result .await diff --git a/crates/lance-context-server/src/sweeper.rs b/crates/lance-context-server/src/sweeper.rs index d7368a8b..783980ea 100644 --- a/crates/lance-context-server/src/sweeper.rs +++ b/crates/lance-context-server/src/sweeper.rs @@ -490,6 +490,7 @@ async fn route_merge( _ => {} } if !count_trigger || store.count_merge_due().await? { + crate::merge_execution::publish_background_request(state.clone(), target.clone()); match crate::merge_execution::spawn_background(state.clone(), target).await { Ok(_reply) => {} // Executor is retained independently of this sweep. Err(crate::error::AppError::Overloaded(_)) => { diff --git a/docs/worker-merge-admission.md b/docs/worker-merge-admission.md index 906bdfb2..c5fb6dcc 100644 --- a/docs/worker-merge-admission.md +++ b/docs/worker-merge-admission.md @@ -15,7 +15,9 @@ merges, independently of the static native-ownership list. It requires a stable count triggers and cleanup timers use the same process slots and executor. A background trigger tries a slot before spawning; it never waits for capacity -or etcd on the flush sweep. Per-table admission is coalesced. The detached +or etcd on the flush sweep. Due demand is also published through a separate +four-request metadata bound, so occupied execution slots do not hide a hot table +from the master. Per-table admission and demand are coalesced. The detached executor exists before admission IO, so a disconnected HTTP caller cannot abandon a writer. Its slot remains occupied until its storage leaves drain or are fenced. An ambiguous terminal record is retained for master recovery. @@ -53,17 +55,33 @@ or an empty execution inventory does not prove that historical writers are quiescent. Removing the shared etcd namespace/configuration or deploying an old binary is not a supported rollback after activation. -## Remaining native handoff requirements +## Automatic native routing (validation in progress) -This implementation does not certify the initial transition, dynamically add -native ownership, or authorize native writes solely because admission closed. -Before production activation, finish and validate: +After the initial legacy-writer transition is verified, explicitly enabling +`MASTER_LOCAL_MERGE_AUTO=true` with `MASTER_LOCAL_MERGE_TARGETS=*` routes new +rollout/generic demand through resident masters without changing a table +allowlist. Worker protocol activation also enables fenced master maintenance. +Automatic routing requires that persisted protocol, stable shard IDs, positive +merge capacity and the existing shared memory budget. It creates no Pods or +second queue. Draining targets and datagen tables remain excluded. -- Complete old-writer coverage, including helpers and master compact/index/repair. -- Durable native enrollment after recorded writers finish or are storage-fenced. -- Automatic new-hot-table selection and bounded continuous native catchup. -- Combined staging of worker admission, tracked master maintenance and native - execution, followed by measured production handoff. +The canonical task claim excludes new background admission. A progressing +background executor is reconciled to completion before native catchup starts; +uncertain storage writes retain their fence until recovery succeeds. Native +catchup uses bounded passes and publishes continuation demand before releasing +the claim, so another master can resume it. Existing explicit append-protocol +targets retain that optimization; newly selected targets use native sealed-WAL +catchup, including generic updates. -Keep this integration disabled until those requirements are implemented. The -current production helpers remain necessary until the handoff is demonstrated. +Neither automatic routing nor the activation marker proves old writers are +quiescent. Before enabling this in production, finish and validate: + +- Complete old-writer coverage, including helpers and pre-upgrade master work. +- Combined staging of worker admission, tracked maintenance and automatic + resident execution with ongoing ingestion, progress stalls and restart. +- Measured production handoff under the actual traffic and memory limits. + +The source paths and regression tests are implemented but final validation is +still running. Keep automatic routing disabled until the initial transition +is proven. Existing production helpers remain necessary until the handoff is +demonstrated. From e0792af12f2c5102fc5cd1eb649f485f09d4c612 Mon Sep 17 00:00:00 2001 From: Beinan Wang Date: Wed, 21 Oct 2026 00:13:18 +0000 Subject: [PATCH 05/10] test: establish concurrent commits before measuring idle isolation --- .../src/merge_execution/background.rs | 26 ++++++++++++++----- 1 file changed, 19 insertions(+), 7 deletions(-) diff --git a/crates/lance-context-server/src/merge_execution/background.rs b/crates/lance-context-server/src/merge_execution/background.rs index eeba3483..543651f8 100644 --- a/crates/lance-context-server/src/merge_execution/background.rs +++ b/crates/lance-context-server/src/merge_execution/background.rs @@ -433,14 +433,10 @@ mod tests { #[tokio::test] #[ignore = "requires isolated local ETCD_TEST_ENDPOINTS"] async fn another_tables_success_does_not_renew_a_stalled_executor() { - let (state, coordinator, _dir) = fixture(2, 2).await; + let (state, coordinator, _dir) = fixture(2, 10).await; let slow = pending(&state, "slow").await; let fast = pending(&state, "fast").await; let held = slow.write().await; - let stalled = spawn_background(state.clone(), "slow".into()) - .await - .unwrap(); - running(&coordinator, "slow").await; let stop = Arc::new(std::sync::atomic::AtomicBool::new(false)); let commits = Arc::new(std::sync::atomic::AtomicUsize::new(0)); let healthy = tokio::spawn({ @@ -469,7 +465,23 @@ mod tests { } } }); - let error = tokio::time::timeout(Duration::from_secs(10), stalled) + // Establish actual commit traffic before the idle window starts. Cold + // index/storage setup can otherwise consume the entire short window on + // a busy test host without ever exercising concurrent progress. + tokio::time::timeout(Duration::from_secs(30), async { + while commits.load(std::sync::atomic::Ordering::SeqCst) < 2 { + assert!(!healthy.is_finished()); + tokio::time::sleep(Duration::from_millis(20)).await; + } + }) + .await + .unwrap(); + let stalled = spawn_background(state.clone(), "slow".into()) + .await + .unwrap(); + running(&coordinator, "slow").await; + let before = commits.load(std::sync::atomic::Ordering::SeqCst); + let error = tokio::time::timeout(Duration::from_secs(30), stalled) .await .unwrap() .unwrap() @@ -478,7 +490,7 @@ mod tests { assert!(!healthy.is_finished()); stop.store(true, std::sync::atomic::Ordering::SeqCst); healthy.await.unwrap(); - assert!(commits.load(std::sync::atomic::Ordering::SeqCst) >= 2); + assert!(commits.load(std::sync::atomic::Ordering::SeqCst) - before >= 2); assert!(error.contains("no-progress"), "{error}"); assert!(coordinator.get("slow").await.unwrap().is_none()); drop(held); From 7ecfbe65b5bcf0a4f5032c4055c10d396a25b756 Mon Sep 17 00:00:00 2001 From: Beinan Wang Date: Wed, 21 Oct 2026 00:35:45 +0000 Subject: [PATCH 06/10] test: exercise periodic demand retry while merge slots stay occupied --- .../src/merge_execution/background.rs | 19 +++++++++++++++++-- 1 file changed, 17 insertions(+), 2 deletions(-) diff --git a/crates/lance-context-server/src/merge_execution/background.rs b/crates/lance-context-server/src/merge_execution/background.rs index 543651f8..b71f1129 100644 --- a/crates/lance-context-server/src/merge_execution/background.rs +++ b/crates/lance-context-server/src/merge_execution/background.rs @@ -396,7 +396,7 @@ mod tests { .await .unwrap(); assert!(state.merge_executions.running.lock().await.is_empty()); - tokio::time::timeout(Duration::from_secs(3), async { + tokio::time::timeout(Duration::from_secs(20), async { loop { if coordinator .request_page(None) @@ -408,7 +408,22 @@ mod tests { { break; } - tokio::time::sleep(Duration::from_millis(20)).await; + // Publication has a bounded metadata timeout. Model the next + // periodic flush after a deferred publish while retaining the + // occupied execution slot; retry must not need merge capacity. + tokio::time::timeout( + Duration::from_secs(2), + sweeper::flush_pass_coordinated( + vec![("hot".into(), store.clone())], + Duration::from_secs(5), + state.merge_slots.clone(), + Some(state.clone()), + ), + ) + .await + .expect("metadata retry must not hold the flush sweep"); + assert!(state.merge_executions.running.lock().await.is_empty()); + tokio::time::sleep(Duration::from_millis(100)).await; } }) .await From 72b2eb3a02f47f24f05797ab6044ccac2cc09e9c Mon Sep 17 00:00:00 2001 From: Beinan Wang Date: Wed, 21 Oct 2026 00:47:56 +0000 Subject: [PATCH 07/10] Retry read snapshots while peer merges make bounded progress --- .../lance-context-core/src/rollout_store.rs | 65 +++++++++++++++++++ crates/lance-context-core/src/store_base.rs | 34 +++++++++- docs/worker-merge-admission.md | 8 +++ 3 files changed, 104 insertions(+), 3 deletions(-) diff --git a/crates/lance-context-core/src/rollout_store.rs b/crates/lance-context-core/src/rollout_store.rs index 40aafdc6..f48680e2 100644 --- a/crates/lance-context-core/src/rollout_store.rs +++ b/crates/lance-context-core/src/rollout_store.rs @@ -2692,6 +2692,71 @@ mod tests { assert_eq!(attempts.load(Ordering::SeqCst), 2); } + #[tokio::test] + async fn list_survives_several_progressing_peer_merges() { + use std::sync::atomic::{AtomicUsize, Ordering}; + + let dir = tempfile::tempdir().unwrap(); + let uri = dir.path().to_str().unwrap(); + let writer = RolloutStore::open_with_options( + uri, + RolloutStoreOptions { + shard_id: Some("continuous-merge-writer".into()), + ..Default::default() + }, + ) + .await + .unwrap(); + writer.add(&[assistant_record("row-0")]).await.unwrap(); + writer.flush().await.unwrap(); + let reader = RolloutStore::open(uri).await.unwrap(); + let writer = tokio::sync::Mutex::new(writer); + let attempts = AtomicUsize::new(0); + + let rows = reader + .base + .read_consistent(ListSource::All, |dataset, snapshots| { + let writer = &writer; + let reader = &reader; + let attempts = &attempts; + async move { + let attempt = attempts.fetch_add(1, Ordering::SeqCst); + if attempt < 4 { + // Each failed snapshot loses a DIFFERENT generation to + // a real successful merge. This is useful progress, not + // a permanently missing file to retry indefinitely. + let mut writer = writer.lock().await; + assert_eq!(writer.cleanup_own_shard().await.unwrap(), 1); + for snapshot in &snapshots { + for generation in &snapshot.sstables { + let path = reader + .flushed_generation_uri(snapshot.shard_id, &generation.path); + tokio::time::timeout(std::time::Duration::from_secs(2), async { + while std::path::Path::new(&path).exists() { + tokio::time::sleep(std::time::Duration::from_millis(1)) + .await; + } + }) + .await + .expect("the actual merge GC must invalidate this snapshot"); + } + } + writer + .add(&[assistant_record(&format!("row-{}", attempt + 1))]) + .await + .unwrap(); + writer.flush().await.unwrap(); + } + reader.list_non_blob_snapshot(dataset, snapshots).await + } + }) + .await + .expect("healthy merge progress must not exhaust a three-attempt read limit"); + assert_eq!(attempts.load(Ordering::SeqCst), 5); + let ids: HashSet<_> = rows.into_iter().map(|row| row.id).collect(); + assert_eq!(ids, (0..5).map(|i| format!("row-{i}")).collect()); + } + #[tokio::test] async fn consistent_read_bounds_missing_generation_retries() { use std::sync::atomic::{AtomicUsize, Ordering}; diff --git a/crates/lance-context-core/src/store_base.rs b/crates/lance-context-core/src/store_base.rs index f3e01f9e..54e164d9 100644 --- a/crates/lance-context-core/src/store_base.rs +++ b/crates/lance-context-core/src/store_base.rs @@ -2299,7 +2299,10 @@ impl StorageBase { /// read of that snapshot. Retry missing generations even when the base /// version is unchanged; never return a partial result after one disappears. /// Ordinary WAL appends do not force retries. Pinned reads use only their - /// selected base version. + /// selected base version. A progressing merge may invalidate several fresh + /// views in a few milliseconds: bound refreshes separately from repeated + /// missing-file failures at the same base version. The retry budget limits + /// starting another attempt, not the duration of an otherwise valid scan. pub async fn read_consistent( &self, source: ListSource, @@ -2309,7 +2312,16 @@ impl StorageBase { F: FnMut(Dataset, Vec) -> Fut, Fut: std::future::Future>, { - for attempt in 0..3 { + const MAX_SNAPSHOT_ATTEMPTS: usize = 16; + const MAX_MISSING_AT_VERSION: usize = 3; + let started = std::time::Instant::now(); + let retry_budget = std::time::Duration::from_secs(2); + let mut last_missing_version = None; + let mut missing_at_version = 0; + for attempt in 0..MAX_SNAPSHOT_ATTEMPTS { + if attempt != 0 && started.elapsed() >= retry_budget { + break; + } let mut dataset = self.dataset.clone(); if !self.is_version_pinned() { dataset.checkout_latest().await?; @@ -2335,7 +2347,23 @@ impl StorageBase { return result; } if let Err(err) = &result { - if !is_not_found_error(err) || attempt == 2 { + if !is_not_found_error(err) { + return result; + } + if last_missing_version == Some(version) { + missing_at_version += 1; + } else { + last_missing_version = Some(version); + missing_at_version = 1; + } + // Actual base commits justify refreshing again, but repeated + // failures without base progress retain the old three-attempt + // bound. Neither continuous commits nor missing data permit + // unlimited repeated payload decoding. + if missing_at_version >= MAX_MISSING_AT_VERSION + || attempt + 1 == MAX_SNAPSHOT_ATTEMPTS + || started.elapsed() >= retry_budget + { return result; } // A merge may already have committed before this read began. diff --git a/docs/worker-merge-admission.md b/docs/worker-merge-admission.md index c5fb6dcc..29546983 100644 --- a/docs/worker-merge-admission.md +++ b/docs/worker-merge-admission.md @@ -38,6 +38,14 @@ Worker HTTP compaction is refused in this mode; tracked master maintenance must handle it. This prevents an untracked worker rewrite from bypassing merge ownership. +Concurrent merge progress can invalidate several read snapshots in well under +a second. Consistent reads now distinguish that progress from repeated missing +data at the same base version: up to 16 snapshot attempts are allowed, while +three missing-file failures without a new base commit still stop the request. +No additional attempt starts after the two-second retry budget is consumed; +that budget does not cancel an otherwise valid long scan. Reads never return +partial results or discard required WAL generations to satisfy this bound. + ## Persisted protocol and startup The private-etcd `worker-merge-admission-mode` key is persistent and unleased. From a08a4b49a5a1bd485fa0cedd6e77b769364b5db9 Mon Sep 17 00:00:00 2001 From: Beinan Wang Date: Wed, 21 Oct 2026 00:53:02 +0000 Subject: [PATCH 08/10] Keep new table fanout working during fenced worker rollout --- .../src/merge_execution.rs | 50 ++++++++++++++++++- docs/worker-merge-admission.md | 5 +- 2 files changed, 53 insertions(+), 2 deletions(-) diff --git a/crates/lance-context-master/src/merge_execution.rs b/crates/lance-context-master/src/merge_execution.rs index c4d45d3f..01597f95 100644 --- a/crates/lance-context-master/src/merge_execution.rs +++ b/crates/lance-context-master/src/merge_execution.rs @@ -78,7 +78,9 @@ pub(crate) async fn run_merge_wal( { return Err("no worker endpoints configured".into()); } - if state.config.append.enabled(target) || crate::local_merge::selected(&state.config, target) { + if (state.config.merge_rollout.owned(target) && state.config.append.enabled(target)) + || crate::local_merge::selected(&state.config, target) + { if let Some(old) = coordinator.get(target).await? { ensure_recovery_due(&coordinator, &old).await?; if old.maintenance.is_none() { @@ -854,6 +856,52 @@ mod tests { (Coordinator::new(client, prefix), proof) } + #[tokio::test] + #[ignore = "requires isolated local ETCD_TEST_ENDPOINTS"] + async fn append_wildcard_does_not_adopt_new_tables_before_native_routing() { + use clap::Parser; + use lance_context_api::TaskKind; + + let dir = tempfile::tempdir().unwrap(); + let mut cfg = crate::config::MasterConfig::parse_from(["test"]); + cfg.data_dir = dir.path().to_string_lossy().into(); + cfg.etcd.etcd_endpoints = vec![std::env::var("ETCD_TEST_ENDPOINTS").unwrap()]; + cfg.etcd.etcd_prefix = format!("/merge-fenced-routing/{}", uuid::Uuid::new_v4()); + cfg.merge_rollout.fence_legacy = true; + cfg.append.rollout_append_targets = vec!["*".into()]; + assert!(cfg.merge_rollout.owned_targets.is_empty()); + assert!(!crate::local_merge::selected(&cfg, "table")); + let mut state = MasterState::new(cfg).await.unwrap(); + let coordinator = state.task_store.merge_coordinator(); + let good = Worker { + coordinator: coordinator.clone(), + calls: Arc::new(AtomicUsize::new(0)), + fail: false, + stall_first: false, + name: "fenced-worker", + events: Arc::new(Mutex::new(Vec::new())), + }; + let (endpoint, server) = worker(good.clone()).await; + Arc::get_mut(&mut state).unwrap().config.worker_endpoints = vec![endpoint]; + state + .task_store + .enqueue(TaskKind::MergeWal, "table", Vec::new()) + .await + .unwrap(); + let claim = state + .task_store + .claim_next_of_kinds(crate::task_store::TaskKinds::MERGE_WAL) + .await + .unwrap() + .unwrap(); + let outcome = run_merge_wal(&state, &claim).await; + server.abort(); + assert!(outcome.as_ref().unwrap().contains("merged 7 generations")); + assert_eq!(good.calls.load(Ordering::SeqCst), 1); + assert!(coordinator.get("table").await.unwrap().is_none()); + state.task_store.finish(claim, outcome).await.unwrap(); + } + #[tokio::test] #[ignore = "requires isolated local ETCD_TEST_ENDPOINTS"] async fn orphaned_merge_is_storage_fenced_and_healthy_shard_progresses() { diff --git a/docs/worker-merge-admission.md b/docs/worker-merge-admission.md index 29546983..51857bb0 100644 --- a/docs/worker-merge-admission.md +++ b/docs/worker-merge-admission.md @@ -32,7 +32,10 @@ records whose original worker is no longer configured. It observes actual execution progress before cancellation, reconciles ambiguous commits through the existing storage barrier, and never treats an empty endpoint list as a successful catchup. New-table demand can enqueue fenced worker fanout without -adding a static owned target. +adding a static owned target. Until automatic native routing is enabled, an +existing append wildcard applies only to explicitly owned tables; new targets +continue through fenced worker fanout instead of entering an ineligible local +execution path. Worker HTTP compaction is refused in this mode; tracked master maintenance must handle it. This prevents an untracked worker rewrite from bypassing merge From 6d3bae40e59bb74f6e03dd4496c67644aab5d872 Mon Sep 17 00:00:00 2001 From: Beinan Wang Date: Wed, 21 Oct 2026 01:01:23 +0000 Subject: [PATCH 09/10] Use existing ID generator in fenced routing regression --- crates/lance-context-master/src/merge_execution.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/crates/lance-context-master/src/merge_execution.rs b/crates/lance-context-master/src/merge_execution.rs index 01597f95..a85fb6ee 100644 --- a/crates/lance-context-master/src/merge_execution.rs +++ b/crates/lance-context-master/src/merge_execution.rs @@ -866,7 +866,10 @@ mod tests { let mut cfg = crate::config::MasterConfig::parse_from(["test"]); cfg.data_dir = dir.path().to_string_lossy().into(); cfg.etcd.etcd_endpoints = vec![std::env::var("ETCD_TEST_ENDPOINTS").unwrap()]; - cfg.etcd.etcd_prefix = format!("/merge-fenced-routing/{}", uuid::Uuid::new_v4()); + cfg.etcd.etcd_prefix = format!( + "/merge-fenced-routing/{}", + lance_context_core::generate_id() + ); cfg.merge_rollout.fence_legacy = true; cfg.append.rollout_append_targets = vec!["*".into()]; assert!(cfg.merge_rollout.owned_targets.is_empty()); From f37522de1a1b824d6476a4318d497fb19dd4abd6 Mon Sep 17 00:00:00 2001 From: Beinan Wang Date: Wed, 21 Oct 2026 01:08:45 +0000 Subject: [PATCH 10/10] Recover point reads racing with peer WAL reclamation --- .../lance-context-core/src/generic_store.rs | 142 +++++++++++++++--- .../lance-context-core/src/rollout_store.rs | 93 ++++++++++-- docs/worker-merge-admission.md | 5 + 3 files changed, 212 insertions(+), 28 deletions(-) diff --git a/crates/lance-context-core/src/generic_store.rs b/crates/lance-context-core/src/generic_store.rs index aaa7a356..4ddff565 100644 --- a/crates/lance-context-core/src/generic_store.rs +++ b/crates/lance-context-core/src/generic_store.rs @@ -35,7 +35,9 @@ use lance_index::mem_wal::ShardManifest; use crate::generic_codec::{batch_to_rows, rows_to_batch, Row}; use crate::merge_budget::MergeMemoryBudget; use crate::store::{CompactionConfig, CompactionStats}; -use crate::store_base::{ListSource, PreparedMerge, StorageBase, StorageBaseOptions}; +use crate::store_base::{ + is_not_found_error, ListSource, PreparedMerge, StorageBase, StorageBaseOptions, +}; use lance_context_api::schema_spec::{SchemaSpec, ID_COLUMN}; use lance_context_api::MAX_BATCH_GET_IDS; @@ -444,26 +446,38 @@ impl GenericStore { if let Some(filter) = filter { // Keep point/filtered reads on this handle's view. The server's // refresh-on-miss path refreshes the cached handle itself; reading - // a newer clone here would bypass that cache update. - let refs: Vec<&str> = columns.iter().map(String::as_str).collect(); - let mut scanner = self - .base - .lsm_scanner() - .await? - .project(&refs)? - .filter(filter)?; - if limit.is_some() || offset.is_some() { - scanner = scanner.limit( - map_i64_bound("limit", limit)?, - map_i64_bound("offset", offset)?, - )?; + // a newer clone on every lookup would bypass that cache update. + let result = async { + let snapshots = self.base.wal_shard_snapshots().await?; + self.scan_filtered_snapshot( + self.base.dataset.clone(), + snapshots, + filter, + limit, + offset, + columns, + ) + .await } - let mut stream = scanner.try_into_stream().await?; - let mut rows = Vec::new(); - while let Some(batch) = stream.try_next().await? { - rows.extend(batch_to_rows(&self.spec, &batch)?); + .await; + if !self.base.is_version_pinned() + && matches!(&result, Err(error) if is_not_found_error(error) + && !matches!(error, LanceError::InvalidInput { .. } | LanceError::SchemaMismatch { .. })) + { + // GC can remove an opened generation during the scan. An + // error is not an absent row: retry a fresh consistent view. + // Successful reads and ordinary misses retain cache semantics. + return self + .base + .read_consistent(ListSource::All, |dataset, snapshots| { + self.scan_filtered_snapshot( + dataset, snapshots, filter, limit, offset, columns, + ) + }) + .boxed() + .await; } - return Ok(rows); + return result; } // Only unfiltered lists use this compatibility recovery. Keep retries @@ -509,6 +523,34 @@ impl GenericStore { .await } + async fn scan_filtered_snapshot( + &self, + dataset: lance::Dataset, + snapshots: Vec, + filter: &str, + limit: Option, + offset: Option, + columns: &[String], + ) -> LanceResult> { + let refs: Vec<&str> = columns.iter().map(String::as_str).collect(); + let mut scanner = + StorageBase::lsm_scanner_for_dataset(&dataset, ID_COLUMN, ListSource::All, snapshots) + .project(&refs)? + .filter(filter)?; + if limit.is_some() || offset.is_some() { + scanner = scanner.limit( + map_i64_bound("limit", limit)?, + map_i64_bound("offset", offset)?, + )?; + } + let mut stream = scanner.try_into_stream().await?; + let mut rows = Vec::new(); + while let Some(batch) = stream.try_next().await? { + rows.extend(batch_to_rows(&self.spec, &batch)?); + } + Ok(rows) + } + /// Narrow compatibility path for Lance 9's broken batched PK sidecars. /// Physical scans supply only candidate keys; the native point planner /// still decides visibility/newest-wins, including replay and tombstones. @@ -850,6 +892,68 @@ mod tests { use serde_json::json; use tempfile::TempDir; + #[tokio::test] + async fn filtered_point_snapshot_recovers_peer_gc_and_preserves_filter() { + use std::sync::atomic::{AtomicUsize, Ordering}; + let dir = TempDir::new().unwrap(); + let uri = dir.path().to_str().unwrap(); + let writer = GenericStore::open(uri, spec(), sealing()).await.unwrap(); + writer + .add(&[ + row(json!({"id":"wanted", "user_id":"value"})), + row(json!({"id":"other", "user_id":"hidden"})), + ]) + .await + .unwrap(); + let reader = GenericStore::open_existing(uri, Default::default()) + .await + .unwrap(); + let writer = tokio::sync::Mutex::new(writer); + let attempts = AtomicUsize::new(0); + let columns = vec!["id".to_string(), "user_id".to_string()]; + let rows = reader + .base + .read_consistent(ListSource::All, |dataset, snapshots| { + let reader = &reader; + let writer = &writer; + let attempts = &attempts; + let columns = &columns; + async move { + if attempts.fetch_add(1, Ordering::SeqCst) == 0 { + writer.lock().await.cleanup_wal().await.unwrap(); + for snapshot in &snapshots { + for generation in &snapshot.sstables { + let path = reader + .base + .flushed_generation_uri(snapshot.shard_id, &generation.path); + tokio::time::timeout(std::time::Duration::from_secs(2), async { + while std::path::Path::new(&path).exists() { + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + } + } + } + reader + .scan_filtered_snapshot( + dataset, + snapshots, + "id = 'wanted'", + Some(1), + None, + columns, + ) + .await + } + }) + .await + .unwrap(); + assert_eq!(attempts.load(Ordering::SeqCst), 2); + assert_eq!(rows, vec![row(json!({"id":"wanted", "user_id":"value"}))]); + } + fn spec() -> SchemaSpec { SchemaSpec::new(vec![ ( diff --git a/crates/lance-context-core/src/rollout_store.rs b/crates/lance-context-core/src/rollout_store.rs index f48680e2..8b0f692c 100644 --- a/crates/lance-context-core/src/rollout_store.rs +++ b/crates/lance-context-core/src/rollout_store.rs @@ -66,7 +66,7 @@ use datafusion::physical_plan::ExecutionPlan; use datafusion::prelude::SessionContext; use datafusion::sql::parser::{DFParser, Statement as DFStatement}; use datafusion::sql::sqlparser::ast::Statement as SqlStatement; -use futures::{stream, StreamExt, TryStreamExt}; +use futures::{stream, FutureExt, StreamExt, TryStreamExt}; use lance::dataset::mem_wal::{LsmScanner, ShardManifestStore, ShardSnapshot}; use lance::dataset::optimize::CompactionMetrics; use lance::dataset::scanner::MaterializationStyle; @@ -1555,21 +1555,37 @@ impl RolloutStore { id: &str, source: ListSource, ) -> LanceResult> { - let shard_snapshots = match source { - ListSource::Fragments => Vec::new(), - ListSource::Wal | ListSource::All => self.wal_shard_snapshots().await?, - }; + if source == ListSource::Fragments { + // Keep the cached base hit cheap. A WAL-only miss still reaches + // the server's existing refresh-on-miss path, updating its cache. + return self + .scan_one_by_id_snapshot(id, source, self.base.dataset.clone(), Vec::new()) + .await; + } + self.base + .read_consistent(source, |dataset, snapshots| { + self.scan_one_by_id_snapshot(id, source, dataset, snapshots) + }) + .boxed() + .await + } + + async fn scan_one_by_id_snapshot( + &self, + id: &str, + source: ListSource, + dataset: Dataset, + shard_snapshots: Vec, + ) -> LanceResult> { let escaped_id = id.replace('\'', "''"); - let columns = self - .non_blob_columns() + let columns = Self::non_blob_columns_for(&dataset) .into_iter() .filter(|column| { source == ListSource::Fragments || !CLAIM_CHECK_COLUMNS.contains(&column.as_str()) }) .collect::>(); let refs: Vec<&str> = columns.iter().map(String::as_str).collect(); - let scanner = self - .lsm_scanner_for_source(source, shard_snapshots) + let scanner = StorageBase::lsm_scanner_for_dataset(&dataset, "id", source, shard_snapshots) .project(&refs)? .filter(&format!("id = '{}'", escaped_id))?; let mut stream = scanner.try_into_stream().await?; @@ -2525,6 +2541,65 @@ mod tests { assert_eq!(attempts.load(Ordering::SeqCst), 2); } + #[tokio::test] + async fn point_snapshot_recovers_peer_gc_without_changing_wal_only_semantics() { + use std::sync::atomic::{AtomicUsize, Ordering}; + for source in [ListSource::All, ListSource::Wal] { + let dir = tempfile::tempdir().unwrap(); + let uri = dir.path().to_str().unwrap(); + let writer = RolloutStore::open(uri).await.unwrap(); + writer.add(&[assistant_record("point")]).await.unwrap(); + writer.flush().await.unwrap(); + let reader = RolloutStore::open(uri).await.unwrap(); + let writer = tokio::sync::Mutex::new(writer); + let attempts = AtomicUsize::new(0); + let record = reader + .base + .read_consistent(source, |dataset, snapshots| { + let reader = &reader; + let writer = &writer; + let attempts = &attempts; + async move { + if attempts.fetch_add(1, Ordering::SeqCst) == 0 { + writer.lock().await.cleanup_own_shard().await.unwrap(); + for snapshot in &snapshots { + for generation in &snapshot.sstables { + let path = reader.flushed_generation_uri( + snapshot.shard_id, + &generation.path, + ); + tokio::time::timeout( + std::time::Duration::from_secs(2), + async { + while std::path::Path::new(&path).exists() { + tokio::task::yield_now().await; + } + }, + ) + .await + .unwrap(); + } + } + } + reader + .scan_one_by_id_snapshot("point", source, dataset, snapshots) + .await + } + }) + .await + .unwrap(); + assert_eq!(attempts.load(Ordering::SeqCst), 2); + if source == ListSource::All { + assert_eq!(record.unwrap().id, "point"); + } else { + assert!( + record.is_none(), + "WAL-only must not return the merged base row" + ); + } + } + } + #[tokio::test] async fn completed_list_snapshot_survives_later_merge_commits() { use std::sync::atomic::{AtomicUsize, Ordering}; diff --git a/docs/worker-merge-admission.md b/docs/worker-merge-admission.md index 51857bb0..cccf4715 100644 --- a/docs/worker-merge-admission.md +++ b/docs/worker-merge-admission.md @@ -48,6 +48,11 @@ three missing-file failures without a new base commit still stop the request. No additional attempt starts after the two-second retry budget is consumed; that budget does not cancel an otherwise valid long scan. Reads never return partial results or discard required WAL generations to satisfy this bound. +Rollout point reads apply that recovery to their WAL fallback while preserving +the cheap cached-base hit and WAL-only semantics. Generic filtered/point reads +keep their cached-view behavior on success or ordinary miss, but recover an +actual missing-file race through a fresh consistent snapshot. Pinned reads do +not advance; invalid filters are not retried as missing storage. ## Persisted protocol and startup