From 1892b389bb9d9a02de60522df061f8948468629b Mon Sep 17 00:00:00 2001 From: Beinan Wang <> Date: Fri, 2 Oct 2026 07:04:09 +0000 Subject: [PATCH 1/3] Recover stalled local maintenance with durable execution fencing --- .../src/merge_write_scope.rs | 108 ++-- .../lance-context-core/src/rollout_store.rs | 90 +++- crates/lance-context-core/src/store_base.rs | 9 +- crates/lance-context-master/src/config.rs | 25 + crates/lance-context-master/src/lib.rs | 1 + .../src/maintenance_execution.rs | 504 ++++++++++++++++++ .../src/merge_execution.rs | 7 +- crates/lance-context-master/src/routes.rs | 1 + crates/lance-context-master/src/scheduler.rs | 76 ++- crates/lance-context-master/src/state.rs | 9 + crates/lance-context-master/src/task_store.rs | 35 +- crates/lance-context-merge/src/lib.rs | 132 +++++ .../src/merge_execution.rs | 1 + docs/owned-maintenance-recovery.md | 58 ++ 14 files changed, 978 insertions(+), 78 deletions(-) create mode 100644 crates/lance-context-master/src/maintenance_execution.rs create mode 100644 docs/owned-maintenance-recovery.md diff --git a/crates/lance-context-core/src/merge_write_scope.rs b/crates/lance-context-core/src/merge_write_scope.rs index 5b569d7..1308af8 100644 --- a/crates/lance-context-core/src/merge_write_scope.rs +++ b/crates/lance-context-core/src/merge_write_scope.rs @@ -43,6 +43,7 @@ pub struct MergeWriteScope { progress: Mutex, changed: Notify, authorizer: Option>, + pin_opened_handles: bool, leaves: Mutex>, } @@ -61,6 +62,17 @@ impl MergeWriteScope { }) } + /// Bind newly opened dataset handles to this execution, including commits + /// from child tasks. Use only for ephemeral maintenance handles: worker + /// caches must keep resolving the current execution at each call instead. + pub fn with_pinned_authorizer(authorizer: Arc) -> Arc { + Arc::new(Self { + authorizer: Some(authorizer), + pin_opened_handles: true, + ..Self::default() + }) + } + /// Only after dropping the merge future and after a durable storage /// barrier fenced every admitted version. Aborting before that evidence /// would abandon an uncertain storage write. @@ -336,7 +348,21 @@ pub async fn fence_manifest_versions( } #[derive(Debug)] -pub(crate) struct GuardedCommit(pub Arc); +pub(crate) struct GuardedCommit { + inner: Arc, + scope: Option>, +} +impl GuardedCommit { + pub(crate) fn new(inner: Arc) -> Self { + Self { + inner, + scope: CURRENT + .try_with(Arc::clone) + .ok() + .filter(|scope| scope.pin_opened_handles), + } + } +} #[async_trait::async_trait] #[allow(clippy::too_many_arguments)] @@ -351,37 +377,43 @@ impl CommitHandler for GuardedCommit { naming_scheme: ManifestNamingScheme, transaction: Option, ) -> std::result::Result { - authorize("base", manifest.version).await?; - let inner = self.0.clone(); - let mut owned_manifest = manifest.clone(); - let path = base_path.clone(); - let store = object_store.clone(); - let scope = CURRENT.try_with(Arc::clone).ok(); - let (location, committed) = shield(async move { - let location = inner - .commit( - &mut owned_manifest, - indices, - &path, - &store, - manifest_writer, - naming_scheme, - transaction, - ) - .await; - // Only a definite conflict proves this conditional commit did - // not happen. A transport/storage error can arrive after the - // remote service accepted it, even though the Rust future ended. - if matches!(&location, Err(CommitError::OtherError(_))) { - if let Some(scope) = scope { - scope.mark_uncertain(); + let work = async { + authorize("base", manifest.version).await?; + let inner = self.inner.clone(); + let mut owned_manifest = manifest.clone(); + let path = base_path.clone(); + let store = object_store.clone(); + let scope = CURRENT.try_with(Arc::clone).ok(); + let (location, committed) = shield(async move { + let location = inner + .commit( + &mut owned_manifest, + indices, + &path, + &store, + manifest_writer, + naming_scheme, + transaction, + ) + .await; + // Only a definite conflict proves this conditional commit did + // not happen. A transport/storage error can arrive after the + // remote service accepted it, even though the Rust future ended. + if matches!(&location, Err(CommitError::OtherError(_))) { + if let Some(scope) = scope { + scope.mark_uncertain(); + } } - } - Ok::<_, CommitError>((location, owned_manifest)) - }) - .await?; - *manifest = committed; - location + Ok::<_, CommitError>((location, owned_manifest)) + }) + .await?; + *manifest = committed; + location + }; + match &self.scope { + Some(scope) => scope.run(work).await, + None => work.await, + } } async fn resolve_latest_location( @@ -389,7 +421,7 @@ impl CommitHandler for GuardedCommit { base_path: &Path, object_store: &ObjectStore, ) -> Result { - self.0 + self.inner .resolve_latest_location(base_path, object_store) .await } @@ -399,7 +431,7 @@ impl CommitHandler for GuardedCommit { version: u64, object_store: &dyn object_store::ObjectStore, ) -> Result { - self.0 + self.inner .resolve_version_location(base_path, version, object_store) .await } @@ -410,7 +442,7 @@ impl CommitHandler for GuardedCommit { object_store: &dyn object_store::ObjectStore, naming: ManifestNamingScheme, ) -> Result { - self.0 + self.inner .version_exists(base_path, version, object_store, naming) .await } @@ -419,7 +451,7 @@ impl CommitHandler for GuardedCommit { base_path: &Path, object_store: &'a ObjectStore, ) -> BoxStream<'a, Result> { - self.0 + self.inner .list_detached_manifest_locations(base_path, object_store) } fn list_manifest_locations<'a>( @@ -428,7 +460,7 @@ impl CommitHandler for GuardedCommit { object_store: &'a ObjectStore, sorted: bool, ) -> BoxStream<'a, Result> { - self.0 + self.inner .list_manifest_locations(base_path, object_store, sorted) } fn list_manifest_locations_since<'a>( @@ -437,11 +469,11 @@ impl CommitHandler for GuardedCommit { object_store: &'a ObjectStore, since: u64, ) -> BoxStream<'a, Result> { - self.0 + self.inner .list_manifest_locations_since(base_path, object_store, since) } async fn delete(&self, base_path: &Path) -> Result<()> { - self.0.delete(base_path).await + self.inner.delete(base_path).await } } diff --git a/crates/lance-context-core/src/rollout_store.rs b/crates/lance-context-core/src/rollout_store.rs index f1c0de4..f78384d 100644 --- a/crates/lance-context-core/src/rollout_store.rs +++ b/crates/lance-context-core/src/rollout_store.rs @@ -2372,6 +2372,94 @@ mod tests { assert_eq!(attempts.load(Ordering::SeqCst), 1); } + #[tokio::test] + async fn owned_store_commit_guard_survives_transfer_to_a_child_task() { + use crate::merge_write_scope::{CommitAuthorizer, MergeWriteScope}; + #[derive(Debug)] + struct Deny; + impl CommitAuthorizer for Deny { + fn authorize<'a>( + &'a self, + _: &'a str, + _: u64, + ) -> std::pin::Pin> + Send + 'a>> + { + Box::pin(async { Err(lance::Error::io("old execution fenced")) }) + } + } + let dir = tempfile::tempdir().unwrap(); + let uri = dir.path().to_str().unwrap(); + let store = RolloutStore::open(uri).await.unwrap(); + let version = store.version(); + let scope = MergeWriteScope::with_pinned_authorizer(Arc::new(Deny)); + let mut dataset = scope + .run(StorageBase::load_with_options(uri, None, None)) + .await + .unwrap(); + let error = + tokio::spawn(async move { dataset.update_metadata([("attempt", "late-child")]).await }) + .await + .unwrap() + .unwrap_err(); + assert!(error.to_string().contains("old execution fenced")); + scope.drain().await; + assert_eq!( + RolloutStore::open_existing_with_options(uri, Default::default()) + .await + .unwrap() + .version(), + version + ); + } + + #[tokio::test] + async fn cached_store_commit_guard_uses_each_new_execution() { + use crate::merge_write_scope::{CommitAuthorizer, MergeWriteScope}; + #[derive(Debug)] + struct Deny; + impl CommitAuthorizer for Deny { + fn authorize<'a>( + &'a self, + _: &'a str, + _: u64, + ) -> std::pin::Pin> + Send + 'a>> + { + Box::pin(async { Err(lance::Error::io("current execution fenced")) }) + } + } + let dir = tempfile::tempdir().unwrap(); + let uri = dir.path().to_str().unwrap(); + RolloutStore::open(uri).await.unwrap(); + let old = MergeWriteScope::with_authorizer(Arc::new(Deny)); + let mut dataset = old + .run(StorageBase::load_with_options(uri, None, None)) + .await + .unwrap(); + old.drain().await; + // A cached handle must not retain the old, now closed execution. + let next = MergeWriteScope::new(); + next.run(async { + dataset + .update_metadata([("attempt", "next-execution")]) + .await + }) + .await + .unwrap(); + next.drain().await; + // The same handle must still enforce a later execution's authorization. + let denied = MergeWriteScope::with_authorizer(Arc::new(Deny)); + let error = denied + .run(async { + dataset + .update_metadata([("attempt", "denied-execution")]) + .await + }) + .await + .unwrap_err(); + assert!(error.to_string().contains("current execution fenced")); + denied.drain().await; + } + #[test] fn rollout_filters_parse_supported_fields() { let filters = RolloutFilters::from_json_value(json!({ @@ -2509,7 +2597,7 @@ mod tests { store.flush().await.unwrap(); let entered = Arc::new(tokio::sync::Notify::new()); let release = Arc::new(tokio::sync::Notify::new()); - let handler = Arc::new(GuardedCommit(Arc::new(PausedCommit { + let handler = Arc::new(GuardedCommit::new(Arc::new(PausedCommit { inner: lance_table::io::commit::commit_handler_from_url(uri, &None) .await .unwrap(), diff --git a/crates/lance-context-core/src/store_base.rs b/crates/lance-context-core/src/store_base.rs index b4fa0b5..44736a7 100644 --- a/crates/lance-context-core/src/store_base.rs +++ b/crates/lance-context-core/src/store_base.rs @@ -2039,8 +2039,9 @@ impl StorageBase { ..Default::default() }); let handler = lance_table::io::commit::commit_handler_from_url(uri, &store_params).await?; - let mut builder = DatasetBuilder::from_uri(uri) - .with_commit_handler(Arc::new(crate::merge_write_scope::GuardedCommit(handler))); + let mut builder = DatasetBuilder::from_uri(uri).with_commit_handler(Arc::new( + crate::merge_write_scope::GuardedCommit::new(handler), + )); if let Some(options) = storage_options { builder = builder.with_storage_options(options); } @@ -2078,7 +2079,9 @@ impl StorageBase { params.session = session; let handler = lance_table::io::commit::commit_handler_from_url(uri, ¶ms.store_params).await?; - params.commit_handler = Some(Arc::new(crate::merge_write_scope::GuardedCommit(handler))); + params.commit_handler = Some(Arc::new(crate::merge_write_scope::GuardedCommit::new( + handler, + ))); Dataset::write(batches, uri, Some(params)).await } diff --git a/crates/lance-context-master/src/config.rs b/crates/lance-context-master/src/config.rs index fe7ed96..7db3e87 100644 --- a/crates/lance-context-master/src/config.rs +++ b/crates/lance-context-master/src/config.rs @@ -8,6 +8,8 @@ use clap::Parser; pub struct MasterConfig { #[command(flatten)] pub merge_rollout: lance_context_merge::rollout::MergeRollout, + #[command(flatten)] + pub maintenance: MaintenanceConfig, /// Data directory / object-store prefix shared with the data-plane server. #[arg(long, env = "DATA_DIR", default_value = "./data")] pub data_dir: String, @@ -308,3 +310,26 @@ mod tests { assert_eq!(config.compaction_max_bytes_per_file, 1024 * 1024 * 1024); } } + +/// Deadlines for local table mutations on explicitly owned targets. +#[derive(Debug, Clone, clap::Args)] +pub struct MaintenanceConfig { + /// Maximum local index/compact/repair execution time on owned targets. + #[arg(long, env = "MAINTENANCE_TIMEOUT_SECS", default_value_t = 3600)] + pub maintenance_timeout_secs: u64, + /// Maximum time without a completed storage/processing step. + #[arg(long, env = "MAINTENANCE_IDLE_TIMEOUT_SECS", default_value_t = 600)] + pub maintenance_idle_timeout_secs: u64, + /// Grace for in-flight manifest commits before storage fencing. + #[arg(long, env = "MAINTENANCE_DRAIN_TIMEOUT_SECS", default_value_t = 30)] + pub maintenance_drain_timeout_secs: u64, +} +impl Default for MaintenanceConfig { + fn default() -> Self { + Self { + maintenance_timeout_secs: 3600, + maintenance_idle_timeout_secs: 600, + maintenance_drain_timeout_secs: 30, + } + } +} diff --git a/crates/lance-context-master/src/lib.rs b/crates/lance-context-master/src/lib.rs index 982179a..f1d23a0 100644 --- a/crates/lance-context-master/src/lib.rs +++ b/crates/lance-context-master/src/lib.rs @@ -3,6 +3,7 @@ pub mod config; pub mod discovery; pub mod error; +mod maintenance_execution; mod merge_execution; pub mod routes; pub mod scanner; diff --git a/crates/lance-context-master/src/maintenance_execution.rs b/crates/lance-context-master/src/maintenance_execution.rs new file mode 100644 index 0000000..81db07d --- /dev/null +++ b/crates/lance-context-master/src/maintenance_execution.rs @@ -0,0 +1,504 @@ +//! Bounded local table mutations sharing WAL merge's durable storage fence. +use crate::{state::MasterState, task_store::TaskClaim}; +use futures::FutureExt; +use lance_context_api::TaskKind; +use lance_context_core::merge_write_scope::{CommitAuthorizer, MergeWriteScope}; +use lance_context_merge::{Coordinator, Execution, MaintenanceKind, Phase}; +use std::{future::Future, pin::Pin, sync::Arc, time::Duration}; + +pub(crate) fn kind(task: TaskKind) -> Option { + match task { + TaskKind::Compact => Some(MaintenanceKind::Compact), + TaskKind::IndexId => Some(MaintenanceKind::IndexId), + TaskKind::Repair => Some(MaintenanceKind::Repair), + TaskKind::MergeWal => None, + } +} + +pub(crate) fn retry_kind(endpoint: &str) -> Option { + match MaintenanceKind::from_endpoint(endpoint)? { + MaintenanceKind::Compact => Some(TaskKind::Compact), + MaintenanceKind::IndexId => Some(TaskKind::IndexId), + MaintenanceKind::Repair => Some(TaskKind::Repair), + } +} + +pub(crate) async fn enqueue_recovery( + state: &Arc, + coordinator: &Coordinator, + execution: &Execution, +) -> Result<(), String> { + let Some(task_kind) = retry_kind(&execution.endpoint) else { + return Ok(()); + }; + if !state.config.merge_rollout.owned(&execution.target) + && !state.config.merge_rollout.draining(&execution.target) + { + return Ok(()); + } + if coordinator + .failure(&execution.target, &execution.endpoint) + .await? + .is_some_and(|f| f.next_retry_ms > lance_context_merge::failure::now_ms()) + { + return Ok(()); + } + crate::scheduler::enqueue(state, task_kind, &execution.target) + .await + .map_err(|e| e.to_string())?; + Ok(()) +} + +/// Claim admission excludes every other reconciler while this runs. A lost +/// process lease is permission to fence its admitted versions, not to delete +/// its persistent target lock or assume that its last PUT failed. +pub(crate) async fn reconcile_previous( + state: &Arc, + claim: &TaskClaim, +) -> Result<(), String> { + let coordinator = state.task_store.merge_coordinator(); + let Some(old) = coordinator.get(&claim.task.target).await? else { + return Ok(()); + }; + if old.maintenance.is_none() { + return Ok(()); + } + let proof = state.task_store.merge_claim(claim); + crate::merge_execution::ensure_recovery_due(&coordinator, &old).await?; + match old.phase { + Phase::Reserved => { + if !coordinator.cancel_reserved(&old).await? { + return Err("maintenance reservation changed".into()); + } + let terminal = coordinator + .get(&old.target) + .await? + .ok_or("maintenance reservation disappeared")?; + if !coordinator.release(&proof, &terminal).await? { + return Err("claim lost releasing maintenance".into()); + } + } + Phase::Finished | Phase::Recovered => { + if !coordinator.release(&proof, &old).await? { + return Err("claim lost releasing maintenance".into()); + } + } + _ => crate::merge_execution::recover_execution(state, &coordinator, &proof, old).await?, + } + Ok(()) +} + +struct Guard { + coordinator: Coordinator, + execution: Execution, + uri: String, +} +impl std::fmt::Debug for Guard { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("MaintenanceGuard") + .field("id", &self.execution.id) + .finish() + } +} +impl CommitAuthorizer for Guard { + fn authorize<'a>( + &'a self, + resource: &'a str, + version: u64, + ) -> Pin> + Send + 'a>> { + Box::pin(async move { + self.coordinator + .authorize_commit(&self.execution, &self.uri, resource, version) + .await + .map_err(lance::Error::io) + }) + } +} + +async fn watch( + coordinator: &Coordinator, + execution: &Execution, + scope: &MergeWriteScope, +) -> String { + let mut sequence = scope.completed_steps(); + let mut changed = tokio::time::Instant::now(); + loop { + tokio::time::sleep(Duration::from_secs(1)).await; + let current = scope.completed_steps(); + if current != sequence { + changed = tokio::time::Instant::now(); + sequence = current; + } + if changed.elapsed() >= Duration::from_secs(execution.idle_timeout_secs) { + return "maintenance no-progress deadline exceeded".into(); + } + // Publishing an unchanged sequence checks ownership but does not count + // as progress. A healthy lease never resets the idle deadline. + match coordinator.publish_progress(execution, sequence).await { + Ok(true) => {} + Ok(false) => return "maintenance ownership revoked".into(), + Err(error) => return format!("maintenance progress publication failed: {error}"), + } + } +} + +pub(crate) async fn run( + state: &Arc, + claim: &TaskClaim, + work: F, +) -> Result +where + F: Future>, +{ + if !state.config.merge_rollout.owned(&claim.task.target) { + return work.await; + } + let maintenance = kind(claim.task.kind).ok_or("invalid local maintenance kind")?; + let uri = state.rollout_uri(&claim.task.target); + if !lance_context_core::merge_write_scope::supports_version_fencing(&uri) { + return Err("invalid storage backend for maintenance fencing".into()); + } + let coordinator = state.task_store.merge_coordinator(); + let proof = state.task_store.merge_claim(claim); + if let Some(failure) = coordinator + .failure(&claim.task.target, maintenance.endpoint()) + .await? + { + if failure.next_retry_ms > lance_context_merge::failure::now_ms() { + return Err(format!( + "maintenance retry at {}: {}", + failure.next_retry_ms, failure.last_error + )); + } + } + let config = &state.config.maintenance; + let mut reserved = Execution::new( + &claim.task.target, + maintenance.endpoint(), + &claim.task.id, + config.maintenance_timeout_secs, + ); + reserved.maintenance = Some(maintenance); + reserved.idle_timeout_secs = config.maintenance_idle_timeout_secs; + if !coordinator.reserve(&proof, &reserved).await? { + return Err("maintenance admission lost".into()); + } + let running = coordinator + .start(&reserved) + .await? + .ok_or("maintenance start lost")?; + let scope = MergeWriteScope::with_pinned_authorizer(Arc::new(Guard { + coordinator: coordinator.clone(), + execution: running.clone(), + uri, + })); + let outcome = std::panic::AssertUnwindSafe(async { + tokio::time::timeout(Duration::from_secs(running.timeout_secs), async { + tokio::select! { + result = scope.run(work) => result, + error = watch(&coordinator, &running, &scope) => Err(error), + } + }) + .await + .unwrap_or_else(|_| Err("maintenance execution deadline exceeded".into())) + }) + .catch_unwind() + .await + .unwrap_or_else(|_| Err("maintenance executor panicked".into())); + // The work future has been dropped, so no preparation may issue another + // commit. The shielded manifest leaves still need acknowledgement or fencing. + let drained = tokio::time::timeout( + Duration::from_secs(config.maintenance_drain_timeout_secs), + scope.drain(), + ) + .await + .is_ok(); + if !drained { + retain_until_fenced(coordinator.clone(), running.clone(), scope.clone()); + } + if !drained || scope.has_uncertain_commit() { + let error = outcome + .as_ref() + .err() + .cloned() + .unwrap_or_else(|| "maintenance manifest result unknown".into()); + let _ = coordinator.report_uncertain(&running, error.clone()).await; + let current = coordinator + .get(&running.target) + .await? + .ok_or("maintenance ownership disappeared")?; + match crate::merge_execution::recover_execution(state, &coordinator, &proof, current).await + { + Ok(()) => { + scope.abort_fenced_leaves().await; + } + Err(recovery) => { + return Err(recovery); + } + } + return Err(format!( + "maintenance storage fenced after ambiguous commit: {error}" + )); + } + let terminal = running.finished(outcome.as_ref().map(|_| 0).map_err(Clone::clone)); + if !coordinator + .finish(&running, outcome.as_ref().map(|_| 0).map_err(Clone::clone)) + .await? + { + return Err("maintenance ownership changed before completion".into()); + } + if !coordinator.release(&proof, &terminal).await? { + return Err("maintenance completion claim lost".into()); + } + outcome +} + +// If storage is unavailable, preserve ownership and return the scheduler slot. +// Only these metadata commit leaves survive cancellation; payload work was +// dropped. A later successful barrier authorizes aborting the old leaves. +fn retain_until_fenced( + coordinator: Coordinator, + execution: Execution, + scope: Arc, +) { + tokio::spawn(async move { + loop { + tokio::select! { + _ = scope.drain() => return, + _ = tokio::time::sleep(Duration::from_secs(5)) => {} + } + match coordinator.get(&execution.target).await { + Ok(None) => { + scope.abort_fenced_leaves().await; + return; + } + Ok(Some(current)) + if current.id != execution.id || current.phase == Phase::Recovered => + { + scope.abort_fenced_leaves().await; + return; + } + _ => {} + } + } + }); +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::{config::MasterConfig, task_store::TaskKinds}; + use clap::Parser; + use std::sync::atomic::{AtomicBool, Ordering}; + + async fn fixture() -> (tempfile::TempDir, Arc) { + let dir = tempfile::tempdir().unwrap(); + let mut config = + MasterConfig::parse_from(["master", "--data-dir", dir.path().to_str().unwrap()]); + config.etcd_endpoints = std::env::var("ETCD_TEST_ENDPOINTS") + .unwrap() + .split(',') + .map(str::to_owned) + .collect(); + config.etcd_prefix = format!("/maintenance-test/{}", lance_context_core::generate_id()); + config.merge_rollout.owned_targets = vec!["table".into()]; + config.stats_scan_interval_secs = 0; + config.compaction_interval_secs = 0; + config.merge_wal_interval_secs = 0; + config.maintenance.maintenance_timeout_secs = 60; + config.maintenance.maintenance_idle_timeout_secs = 1; + config.maintenance.maintenance_drain_timeout_secs = 1; + (dir, MasterState::new(config).await.unwrap()) + } + + async fn claim(state: &Arc, kind: TaskKind) -> TaskClaim { + crate::scheduler::enqueue(state, kind, "table") + .await + .unwrap(); + state.task_store.claim_next().await.unwrap().unwrap() + } + + #[tokio::test] + #[ignore = "requires ETCD_TEST_ENDPOINTS"] + async fn stalled_local_kinds_cancel_and_release_without_stopping_other_tables() { + struct Active(Arc); + impl Drop for Active { + fn drop(&mut self) { + self.0.store(false, Ordering::SeqCst); + } + } + let (_dir, state) = fixture().await; + for kind in [TaskKind::IndexId, TaskKind::Compact, TaskKind::Repair] { + let owned = claim(&state, kind).await; + let active = Arc::new(AtomicBool::new(true)); + let guard = Active(active.clone()); + let result = run(&state, &owned, async move { + let _guard = guard; + std::future::pending::>().await + }); + let other = async { + let task = crate::scheduler::enqueue(&state, TaskKind::Compact, "healthy") + .await + .unwrap(); + let healthy = state.task_store.claim_next().await.unwrap().unwrap(); + assert_eq!(healthy.task.id, task.id); + state + .task_store + .finish(healthy, Ok("healthy continued".into())) + .await + .unwrap(); + }; + let (result, ()) = tokio::join!(result, other); + assert!(result.as_ref().unwrap_err().contains("no-progress")); + assert!(!active.load(Ordering::SeqCst)); + assert!(state + .task_store + .merge_coordinator() + .get("table") + .await + .unwrap() + .is_none()); + state.task_store.finish(owned, result).await.unwrap(); + let failure = state + .task_store + .merge_coordinator() + .failure("table", super::kind(kind).unwrap().endpoint()) + .await + .unwrap() + .unwrap(); + assert_eq!(failure.consecutive_attempts, 1); + assert!(failure.next_retry_ms > failure.last_failure_ms); + } + } + + #[tokio::test] + #[ignore = "requires ETCD_TEST_ENDPOINTS"] + async fn lost_master_index_is_fenced_before_merge_takes_over() { + let (_dir, state) = fixture().await; + let uri = state.rollout_uri("table"); + let store = lance_context_core::RolloutStore::open(&uri).await.unwrap(); + let high = store.version() + 1; + let old_claim = claim(&state, TaskKind::IndexId).await; + let coordinator = state.task_store.merge_coordinator(); + let proof = state.task_store.merge_claim(&old_claim); + let mut old = Execution::new("table", "master:index_id", "old-master", 60); + old.maintenance = Some(MaintenanceKind::IndexId); + assert!(coordinator.reserve(&proof, &old).await.unwrap()); + let old = coordinator.start(&old).await.unwrap().unwrap(); + coordinator + .authorize_commit(&old, &uri, "base", high) + .await + .unwrap(); + crate::scheduler::enqueue(&state, TaskKind::MergeWal, "table") + .await + .unwrap(); + assert!( + state + .task_store + .claim_next_of_kinds(TaskKinds::MERGE_WAL) + .await + .unwrap() + .is_none(), + "live reconciler must remain exclusive" + ); + state + .task_store + .abandon_claim_for_test(old_claim) + .await + .unwrap(); + let next = state + .task_store + .claim_next_of_kinds(TaskKinds::MERGE_WAL) + .await + .unwrap() + .unwrap(); + reconcile_previous(&state, &next).await.unwrap(); + assert!(coordinator.get("table").await.unwrap().is_none()); + let fresh = + lance_context_core::RolloutStore::open_existing_with_options(&uri, Default::default()) + .await + .unwrap(); + assert!( + fresh.version() > high, + "all old admitted immutable versions must be occupied" + ); + assert!(coordinator + .authorize_commit(&old, &uri, "base", high + 2) + .await + .is_err()); + state + .task_store + .finish(next, Ok("safe to continue".into())) + .await + .unwrap(); + } + + #[tokio::test] + #[ignore = "requires ETCD_TEST_ENDPOINTS"] + async fn failed_storage_barrier_keeps_ownership_and_retries_do_not_reset_budget() { + let (_dir, state) = fixture().await; + let old_claim = claim(&state, TaskKind::Repair).await; + let coordinator = state.task_store.merge_coordinator(); + let mut old = Execution::new("table", "master:repair", "old-master", 60); + old.maintenance = Some(MaintenanceKind::Repair); + assert!(coordinator + .reserve(&state.task_store.merge_claim(&old_claim), &old) + .await + .unwrap()); + let old = coordinator.start(&old).await.unwrap().unwrap(); + // No dataset exists: the metadata recovery barrier must fail closed. + coordinator + .authorize_commit(&old, &state.rollout_uri("table"), "base", 1) + .await + .unwrap(); + state + .task_store + .finish(old_claim, Err("lost ambiguous execution".into())) + .await + .unwrap(); + let (rows, _) = coordinator.execution_page(None).await.unwrap(); + assert_eq!(rows.len(), 1); + // There is deliberately no failure ledger yet: execution inventory + // alone must schedule recovery even with all stats sweeps disabled. + enqueue_recovery(&state, &coordinator, &rows[0]) + .await + .unwrap(); + let retry = state.task_store.claim_next().await.unwrap().unwrap(); + let error = reconcile_previous(&state, &retry).await.unwrap_err(); + assert_eq!( + coordinator.get("table").await.unwrap().unwrap().phase, + Phase::Recovering + ); + assert!(coordinator + .authorize_commit(&old, &state.rollout_uri("table"), "base", 2) + .await + .is_err()); + state.task_store.finish(retry, Err(error)).await.unwrap(); + let failure = coordinator + .failure("table", "master:repair") + .await + .unwrap() + .unwrap(); + enqueue_recovery(&state, &coordinator, &rows[0]) + .await + .unwrap(); + assert!( + state + .task_store + .get_active_id(TaskKind::Repair, "table") + .await + .unwrap() + .is_none(), + "backoff must not recreate a fresh retry budget" + ); + assert_eq!( + coordinator + .failure("table", "master:repair") + .await + .unwrap() + .unwrap() + .consecutive_attempts, + failure.consecutive_attempts + ); + } +} diff --git a/crates/lance-context-master/src/merge_execution.rs b/crates/lance-context-master/src/merge_execution.rs index 33c5807..15aec17 100644 --- a/crates/lance-context-master/src/merge_execution.rs +++ b/crates/lance-context-master/src/merge_execution.rs @@ -168,7 +168,10 @@ impl Deadlines { } } -async fn ensure_recovery_due(coordinator: &Coordinator, old: &Execution) -> Result<(), String> { +pub(crate) async fn ensure_recovery_due( + coordinator: &Coordinator, + old: &Execution, +) -> Result<(), String> { if old.phase == Phase::Recovering { if let Some(failure) = coordinator.failure(&old.target, &old.endpoint).await? { if failure.last_error.contains("recovery barrier failed") @@ -184,7 +187,7 @@ async fn ensure_recovery_due(coordinator: &Coordinator, old: &Execution) -> Resu Ok(()) } -async fn recover_execution( +pub(crate) async fn recover_execution( state: &Arc, coordinator: &Coordinator, proof: &ClaimProof, diff --git a/crates/lance-context-master/src/routes.rs b/crates/lance-context-master/src/routes.rs index c9d7cd8..ae97cb9 100644 --- a/crates/lance-context-master/src/routes.rs +++ b/crates/lance-context-master/src/routes.rs @@ -715,6 +715,7 @@ mod tests { fn test_config(dir: &TempDir) -> MasterConfig { MasterConfig { + maintenance: Default::default(), merge_rollout: Default::default(), data_dir: dir.path().to_string_lossy().to_string(), host: "127.0.0.1".to_string(), diff --git a/crates/lance-context-master/src/scheduler.rs b/crates/lance-context-master/src/scheduler.rs index 0fed321..3a9ceec 100644 --- a/crates/lance-context-master/src/scheduler.rs +++ b/crates/lance-context-master/src/scheduler.rs @@ -147,17 +147,29 @@ async fn run_task(state: &Arc, claim: TaskClaim, timing: TaskClaimT .record(timing.permit_wait.as_secs_f64()); let started = std::time::Instant::now(); - let outcome = - if state.config.merge_rollout.draining(&task.target) && task.kind != TaskKind::MergeWal { + let outcome = match crate::maintenance_execution::reconcile_previous(state, &claim).await { + Err(error) => Err(error), + Ok(()) + if state.config.merge_rollout.draining(&task.target) + && task.kind != TaskKind::MergeWal => + { Err("target draining legacy writers for merge protocol transition".into()) - } else { - match task.kind { - TaskKind::Compact => run_compaction(state, &task).await, - TaskKind::MergeWal => crate::merge_execution::run_merge_wal(state, &claim).await, - TaskKind::IndexId => run_index_id(state, &task.target).await, - TaskKind::Repair => run_repair(state, &task).await, + } + Ok(()) => match task.kind { + TaskKind::MergeWal => crate::merge_execution::run_merge_wal(state, &claim).await, + _ => { + crate::maintenance_execution::run(state, &claim, async { + match task.kind { + TaskKind::Compact => run_compaction(state, &task).await, + TaskKind::IndexId => run_index_id(state, &task.target).await, + TaskKind::Repair => run_repair(state, &task).await, + TaskKind::MergeWal => unreachable!(), + } + }) + .await } - }; + }, + }; let work_elapsed = started.elapsed(); let result = if outcome.is_ok() { "success" } else { "failed" }; @@ -674,6 +686,7 @@ pub fn spawn_scheduler(state: &Arc) -> JoinHandle<()> { let coordinator = retry_state.task_store.merge_coordinator(); let mut cursor = None; let mut request_cursor = None; + let mut execution_cursor = None; let mut ticker = tokio::time::interval(Duration::from_secs(15)); loop { ticker.tick().await; @@ -695,23 +708,53 @@ pub fn spawn_scheduler(state: &Arc) -> JoinHandle<()> { } Err(error) => tracing::warn!(%error, "worker merge demand scan failed"), } + match coordinator + .execution_page(execution_cursor.as_deref()) + .await + { + 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 + { + tracing::warn!(target = %execution.target, %error, "local execution recovery enqueue failed"); + } + } + } + Err(error) => tracing::warn!(%error, "local execution recovery scan failed"), + } match coordinator.failure_page(cursor.as_deref(), 256).await { Ok((rows, next)) => { cursor = next; let mut targets = std::collections::HashSet::new(); for failure in rows { + let task_kind = crate::maintenance_execution::retry_kind(&failure.endpoint) + .or_else(|| { + retry_state + .config + .worker_endpoints + .contains(&failure.endpoint) + .then_some(TaskKind::MergeWal) + }); + let Some(task_kind) = task_kind else { + continue; + }; if failure.next_retry_ms <= lance_context_merge::failure::now_ms() - && retry_state - .config - .worker_endpoints - .contains(&failure.endpoint) - && targets.insert(failure.target.clone()) + && targets.insert((failure.target.clone(), kind_label(task_kind))) && should_probe_failure(&retry_state, &failure).await { if let Err(error) = - enqueue(&retry_state, TaskKind::MergeWal, &failure.target).await + enqueue(&retry_state, task_kind, &failure.target).await { - tracing::warn!(target = %failure.target, %error, "merge recovery enqueue failed"); + tracing::warn!(target = %failure.target, %error, "maintenance recovery enqueue failed"); } } } @@ -871,6 +914,7 @@ mod tests { fn config(dir: &TempDir) -> MasterConfig { MasterConfig { + maintenance: Default::default(), merge_rollout: lance_context_merge::rollout::MergeRollout { owned_targets: ["exp", "generic:gs", "broken"] .into_iter() diff --git a/crates/lance-context-master/src/state.rs b/crates/lance-context-master/src/state.rs index 703a40e..8155665 100644 --- a/crates/lance-context-master/src/state.rs +++ b/crates/lance-context-master/src/state.rs @@ -121,6 +121,14 @@ impl MasterState { /// Open the registry, stats dataset, and configured durable task store. pub async fn new(config: MasterConfig) -> lance::Result> { config.merge_rollout.validate().map_err(lance::Error::io)?; + if config.maintenance.maintenance_timeout_secs == 0 + || config.maintenance.maintenance_idle_timeout_secs == 0 + || config.maintenance.maintenance_drain_timeout_secs == 0 + { + return Err(lance::Error::invalid_input( + "maintenance deadlines must be positive", + )); + } let task_store = TaskStore::open(&config).await?; // Serialize first-time registry/stats creation and legacy backfill in // etcd mode. Followers wait briefly rather than racing Lance creates. @@ -240,6 +248,7 @@ mod tests { fn test_config(dir: &TempDir) -> MasterConfig { MasterConfig { + maintenance: Default::default(), merge_rollout: Default::default(), data_dir: dir.path().to_string_lossy().to_string(), host: "127.0.0.1".to_string(), diff --git a/crates/lance-context-master/src/task_store.rs b/crates/lance-context-master/src/task_store.rs index aecea41..403e504 100644 --- a/crates/lance-context-master/src/task_store.rs +++ b/crates/lance-context-master/src/task_store.rs @@ -679,14 +679,21 @@ impl EtcdTaskStore { } } - let merge_execution = if task.kind == TaskKind::MergeWal { + let merge_execution = lance_context_merge::Coordinator::new(self.client.clone(), self.prefix.clone()) .get(&task.target) .await - .map_err(lance::Error::io)? - } else { - None - }; + .map_err(lance::Error::io)?; + // Only MergeWal reconciles a worker execution. A local fenced + // mutation may be recovered by any table writer after the + // exclusive reconciler lease has expired. + if task.kind != TaskKind::MergeWal + && merge_execution + .as_ref() + .is_some_and(|e| e.maintenance.is_none()) + { + continue; + } let token = generate_id(); let lease_id = self.grant_lease().await?; let claim_key = self.claim_key(&task.id); @@ -714,22 +721,13 @@ impl EtcdTaskStore { compares.push(Compare::version(key.as_str(), CompareOp::Equal, 0)); } } - // A lost scheduler lease does not terminate remote storage work. - // MergeWal may claim to reconcile it; every other writer waits. - if task.kind != TaskKind::MergeWal && target_key.is_some() { - compares.push(Compare::version( - lance_context_merge::execution_key(&self.prefix, &task.target), - CompareOp::Equal, - 0, - )); - } // Remote execution ownership persists, but its reconciler is // still exclusive and leased. Dependency-chain merge tasks // must not cancel another live scheduler's execution. let merge_claim_key = lance_context_merge::execution_key(&self.prefix, &task.target) .replace("/merge-executions/", "/merge-claims/"); - if task.kind == TaskKind::MergeWal { + if requires_target_lock(task.kind) { compares.push(Compare::version( merge_claim_key.as_str(), CompareOp::Equal, @@ -743,7 +741,7 @@ impl EtcdTaskStore { TxnOp::put(running_key, running_value, None), TxnOp::put(claim_key.as_str(), token.as_bytes(), lease_options.clone()), ]; - if task.kind == TaskKind::MergeWal { + if requires_target_lock(task.kind) { operations.push(TxnOp::put( merge_claim_key, token.as_bytes(), @@ -803,7 +801,7 @@ impl EtcdTaskStore { keepalive, } = claim.backend; let mut task = claim.task; - let unresolved = if task.kind == TaskKind::MergeWal && outcome.is_err() { + let unresolved = if requires_target_lock(task.kind) { lance_context_merge::Coordinator::new(self.client.clone(), self.prefix.clone()) .get(&task.target) .await @@ -817,7 +815,7 @@ impl EtcdTaskStore { TxnOp::delete(claim_key.as_str(), None), TxnOp::delete(self.running_key(&task.id), None), ]; - if task.kind == TaskKind::MergeWal { + if requires_target_lock(task.kind) { operations.push(TxnOp::delete( lance_context_merge::execution_key(&self.prefix, &task.target) .replace("/merge-executions/", "/merge-claims/"), @@ -1516,6 +1514,7 @@ mod tests { fn config(dir: &TempDir) -> MasterConfig { MasterConfig { + maintenance: Default::default(), merge_rollout: Default::default(), data_dir: dir.path().to_string_lossy().to_string(), host: "127.0.0.1".to_string(), diff --git a/crates/lance-context-merge/src/lib.rs b/crates/lance-context-merge/src/lib.rs index 87b193d..1f53395 100644 --- a/crates/lance-context-merge/src/lib.rs +++ b/crates/lance-context-merge/src/lib.rs @@ -32,8 +32,38 @@ pub enum Phase { Finished, } +/// Local master mutations use the same storage fence as WAL merging. Older +/// readers cannot CAS these records because they do not preserve this field. +#[derive(Clone, Copy, Debug, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum MaintenanceKind { + Compact, + IndexId, + Repair, +} + +impl MaintenanceKind { + pub fn endpoint(self) -> &'static str { + match self { + Self::Compact => "master:compact", + Self::IndexId => "master:index_id", + Self::Repair => "master:repair", + } + } + pub fn from_endpoint(endpoint: &str) -> Option { + match endpoint { + "master:compact" => Some(Self::Compact), + "master:index_id" => Some(Self::IndexId), + "master:repair" => Some(Self::Repair), + _ => None, + } + } +} + #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] pub struct Execution { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub maintenance: Option, pub id: String, pub target: String, pub endpoint: String, @@ -61,6 +91,7 @@ fn default_queue_timeout() -> u64 { impl Execution { pub fn new(target: &str, endpoint: &str, instance: &str, timeout_secs: u64) -> Self { Self { + maintenance: None, id: uuid::Uuid::new_v4().to_string(), target: target.into(), endpoint: endpoint.into(), @@ -140,6 +171,48 @@ impl Coordinator { .transpose() } + /// Bounded recovery inventory; execution records themselves are durable + /// demand even if a process died before writing its failure ledger. + pub async fn execution_page( + &self, + after: Option<&str>, + ) -> Result<(Vec, Option)> { + let prefix = format!("{}/merge-executions/", self.prefix.trim_end_matches('/')); + let (start, options) = match after { + None => (prefix.clone(), etcd_client::GetOptions::new().with_prefix()), + Some(key) if key.starts_with(&prefix) => { + let mut end = prefix.as_bytes().to_vec(); + *end.last_mut().unwrap() += 1; + ( + format!("{key}\0"), + etcd_client::GetOptions::new().with_range(end), + ) + } + Some(_) => return Err("invalid execution cursor".into()), + }; + let response = self + .client + .clone() + .get(start, Some(options.with_limit(256))) + .await + .map_err(|e| e.to_string())?; + let rows = response + .kvs() + .iter() + .map(|kv| serde_json::from_slice(kv.value()).map_err(|e| e.to_string())) + .collect::>>()?; + let next = response + .more() + .then(|| { + response + .kvs() + .last() + .map(|kv| String::from_utf8_lossy(kv.key()).into_owned()) + }) + .flatten(); + Ok((rows, next)) + } + /// 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 @@ -448,6 +521,65 @@ mod tests { assert!(result.is_err()); } + #[test] + fn maintenance_records_preserve_legacy_worker_encoding_and_require_new_readers() { + let legacy = Execution::new("table", "http://worker", "instance", 600); + let encoded = encode(&legacy); + assert!(!String::from_utf8_lossy(&encoded).contains("maintenance")); + assert_eq!( + serde_json::from_slice::(&encoded).unwrap(), + legacy + ); + for kind in [ + MaintenanceKind::Compact, + MaintenanceKind::IndexId, + MaintenanceKind::Repair, + ] { + let mut local = legacy.clone(); + local.maintenance = Some(kind); + local.endpoint = kind.endpoint().into(); + let encoded = encode(&local); + assert_eq!( + serde_json::from_slice::(&encoded).unwrap(), + local + ); + assert_eq!(MaintenanceKind::from_endpoint(&local.endpoint), Some(kind)); + // An old reader omits the new field on re-encoding: its value CAS + // cannot start, finish, freeze or release this local execution. + let mut old_reader = local.clone(); + old_reader.maintenance = None; + assert_ne!(encode(&old_reader), encoded); + } + assert_eq!(MaintenanceKind::from_endpoint("http://worker"), None); + } + + #[tokio::test] + #[ignore = "requires isolated local ETCD_TEST_ENDPOINTS"] + async fn execution_inventory_pages_without_crossing_namespace() { + let (coordinator, mut client, _, _) = fixture().await; + for n in 0..257 { + let execution = Execution::new(&format!("table-{n:03}"), "master:repair", "lost", 60); + client + .put( + execution_key(&coordinator.prefix, &execution.target), + encode(&execution), + None, + ) + .await + .unwrap(); + } + let (first, cursor) = coordinator.execution_page(None).await.unwrap(); + assert_eq!(first.len(), 256); + let (last, done) = coordinator.execution_page(cursor.as_deref()).await.unwrap(); + assert_eq!(last.len(), 1); + assert!(done.is_none()); + assert!(!first.iter().any(|e| e.id == last[0].id)); + assert!(coordinator + .execution_page(Some("/another-prefix/key")) + .await + .is_err()); + } + // Isolated prefix on a LOCAL test etcd. These tests never use production. async fn fixture() -> (Coordinator, Client, ClaimProof, i64) { let endpoint = std::env::var("ETCD_TEST_ENDPOINTS") diff --git a/crates/lance-context-server/src/merge_execution.rs b/crates/lance-context-server/src/merge_execution.rs index e5c2577..f84ba5e 100644 --- a/crates/lance-context-server/src/merge_execution.rs +++ b/crates/lance-context-server/src/merge_execution.rs @@ -153,6 +153,7 @@ pub async fn start( ) -> Result { let coordinator = state.merge_executions.coordinator().await?; if !state.merge_executions.owned(&execution.target) + || execution.maintenance.is_some() || execution.protocol != 2 || execution.instance != state.merge_executions.instance || execution.phase != Phase::Reserved diff --git a/docs/owned-maintenance-recovery.md b/docs/owned-maintenance-recovery.md new file mode 100644 index 0000000..f371740 --- /dev/null +++ b/docs/owned-maintenance-recovery.md @@ -0,0 +1,58 @@ +# Recovering stalled local table maintenance + +Index, compaction and repair tasks can hold the same table lock needed by WAL +merging. On explicitly owned merge targets, these local mutations now reserve a +durable execution using the existing version-fencing protocol. Legacy targets +retain their previous execution path. Worker merge fan-out remains serial; +compaction concurrency and all merge memory/byte budgets are unchanged. + +## Cancellation and takeover + +The default total deadline is 3600 seconds (`MAINTENANCE_TIMEOUT_SECS`), with a +600-second no-progress deadline (`MAINTENANCE_IDLE_TIMEOUT_SECS`) and 30-second +commit drain grace (`MAINTENANCE_DRAIN_TIMEOUT_SECS`). All must be positive. +Progress counts completed scoped processing steps or manifest commits, never +heartbeats. Lance index building and individual compaction rewrites do not +expose fine-grained scan progress here: a healthy long operation can reach the +idle deadline. Size that deadline above measured large-table operation times; +it is an execution bound, not proof of an internal deadlock. + +Cancellation drops the local work future and closes commit admission. If every +manifest commit has a definite result, the task can complete and release its +lock. Ambiguous or still-running commits require durable admission revocation +and metadata-only conditional manifest writes beyond every admitted version. +Only a successful barrier permits another writer to take over. Newly opened +maintenance handles retain their execution guard even when passed to a child +task. Cached worker handles continue using the current execution dynamically. + +If storage fencing fails, keep the durable table ownership, return the scheduler +slot, and retry recovery with persistent backoff. A small supervisor retains any +unfinished manifest leaves until they drain or a later reconciler proves the +storage fence. Other tables keep running. A master crash or an uncertain +admission response is recoverable from the durable execution inventory, scanned +in pages of 256 every 15 seconds independently of stats sweeps. + +## Repeated failures and diagnostics + +Failures use the existing `/merge-failures` API and durable per-target ledger, +with endpoints `master:compact`, `master:index_id`, and `master:repair`. A new +task or process does not reset the attempt count. Retryable failures back off; +persistent failures get `needs_attention` and hourly probes. An unresolved +storage barrier retains ownership and probes at up to five-minute intervals. +An unavailable dataset, bad configuration or repeatable corruption requires +repair of that cause; retrying cannot manufacture a successful storage fence. + +## Activation and compatibility + +This uses the existing owned-target configuration, not a new rollout switch. +Deploy compatible masters before activating more owned targets. Include every +auxiliary maintenance master in that inventory. Old readers cannot CAS records +containing the maintenance discriminator, so they fail closed rather than +acting as a compatible recovery controller. Worker execution admission rejects +local maintenance records. Existing worker execution JSON remains unchanged. + +The existing protocol transition still requires legacy writes to be drained: +this change cannot retroactively fence an old unguarded write. It does not enable +owned targets, alter production pods, or complete registry migration. Recovery +continues on draining targets while an owned execution remains. Validate large +table timings and fault recovery before selecting tighter deadlines. From 3de3604d31b1bf07e1a1520df55434c0a9e06937 Mon Sep 17 00:00:00 2001 From: Beinan Wang <> Date: Fri, 2 Oct 2026 07:09:45 +0000 Subject: [PATCH 2/3] Resume repaired work without resetting unrelated retry budgets --- .../src/maintenance_execution.rs | 35 +++++++++++- crates/lance-context-master/src/scheduler.rs | 54 ++++++++++++++++++- docs/owned-maintenance-recovery.md | 5 +- 3 files changed, 89 insertions(+), 5 deletions(-) diff --git a/crates/lance-context-master/src/maintenance_execution.rs b/crates/lance-context-master/src/maintenance_execution.rs index 81db07d..309398e 100644 --- a/crates/lance-context-master/src/maintenance_execution.rs +++ b/crates/lance-context-master/src/maintenance_execution.rs @@ -166,8 +166,8 @@ where { if failure.next_retry_ms > lance_context_merge::failure::now_ms() { return Err(format!( - "maintenance retry at {}: {}", - failure.next_retry_ms, failure.last_error + "maintenance retry at {} after {:?} failure", + failure.next_retry_ms, failure.class )); } } @@ -371,6 +371,37 @@ mod tests { } } + #[tokio::test] + #[ignore = "requires ETCD_TEST_ENDPOINTS"] + async fn a_new_task_cannot_bypass_local_failure_backoff() { + let (_dir, state) = fixture().await; + let first = claim(&state, TaskKind::IndexId).await; + let outcome = run(&state, &first, async { Err("invalid index schema".into()) }).await; + state.task_store.finish(first, outcome).await.unwrap(); + let coordinator = state.task_store.merge_coordinator(); + let before = coordinator + .failure("table", "master:index_id") + .await + .unwrap() + .unwrap(); + assert!(before.needs_attention); + let fresh = claim(&state, TaskKind::IndexId).await; + let result = run(&state, &fresh, async { + panic!("backoff must prevent new payload work") + }) + .await; + assert!(result.as_ref().unwrap_err().contains("retry at")); + state.task_store.finish(fresh, result).await.unwrap(); + let after = coordinator + .failure("table", "master:index_id") + .await + .unwrap() + .unwrap(); + assert_eq!(before.consecutive_attempts, after.consecutive_attempts); + assert_eq!(before.next_retry_ms, after.next_retry_ms); + assert!(coordinator.get("table").await.unwrap().is_none()); + } + #[tokio::test] #[ignore = "requires ETCD_TEST_ENDPOINTS"] async fn lost_master_index_is_fenced_before_merge_takes_over() { diff --git a/crates/lance-context-master/src/scheduler.rs b/crates/lance-context-master/src/scheduler.rs index 3a9ceec..4e6cdb7 100644 --- a/crates/lance-context-master/src/scheduler.rs +++ b/crates/lance-context-master/src/scheduler.rs @@ -162,7 +162,7 @@ async fn run_task(state: &Arc, claim: TaskClaim, timing: TaskClaimT match task.kind { TaskKind::Compact => run_compaction(state, &task).await, TaskKind::IndexId => run_index_id(state, &task.target).await, - TaskKind::Repair => run_repair(state, &task).await, + TaskKind::Repair => run_repair(state, &claim).await, TaskKind::MergeWal => unreachable!(), } }) @@ -238,7 +238,8 @@ async fn schedule_repair(state: &Arc, failed: &TaskRecord) { /// Drop base-table fragments whose files are missing (see /// `StorageBase::repair_missing_fragments`) and record what was dropped. -async fn run_repair(state: &Arc, task: &TaskRecord) -> Result { +async fn run_repair(state: &Arc, claim: &TaskClaim) -> Result { + let task = &claim.task; let (kind, name) = parse_target(&task.target); if kind != StoreKind::Rollout { return Err(format!("repair is not implemented for {kind:?} stores")); @@ -254,6 +255,21 @@ async fn run_repair(state: &Arc, task: &TaskRecord) -> Result Date: Fri, 2 Oct 2026 07:28:47 +0000 Subject: [PATCH 3/3] Configure maintenance fault fixtures through stable CLI arguments --- .../src/maintenance_execution.rs | 19 +++++++++++-------- 1 file changed, 11 insertions(+), 8 deletions(-) diff --git a/crates/lance-context-master/src/maintenance_execution.rs b/crates/lance-context-master/src/maintenance_execution.rs index 309398e..4b159e3 100644 --- a/crates/lance-context-master/src/maintenance_execution.rs +++ b/crates/lance-context-master/src/maintenance_execution.rs @@ -293,14 +293,17 @@ mod tests { async fn fixture() -> (tempfile::TempDir, Arc) { let dir = tempfile::tempdir().unwrap(); - let mut config = - MasterConfig::parse_from(["master", "--data-dir", dir.path().to_str().unwrap()]); - config.etcd_endpoints = std::env::var("ETCD_TEST_ENDPOINTS") - .unwrap() - .split(',') - .map(str::to_owned) - .collect(); - config.etcd_prefix = format!("/maintenance-test/{}", lance_context_core::generate_id()); + let endpoints = std::env::var("ETCD_TEST_ENDPOINTS").unwrap(); + let prefix = format!("/maintenance-test/{}", lance_context_core::generate_id()); + let mut config = MasterConfig::parse_from([ + "master", + "--data-dir", + dir.path().to_str().unwrap(), + "--etcd-endpoints", + &endpoints, + "--etcd-prefix", + &prefix, + ]); config.merge_rollout.owned_targets = vec!["table".into()]; config.stats_scan_interval_secs = 0; config.compaction_interval_secs = 0;