From a76a089336dfd143757bcd10f7fd76c1fd30f579 Mon Sep 17 00:00:00 2001 From: Beinan Wang <> Date: Mon, 5 Oct 2026 20:49:48 +0000 Subject: [PATCH] fix: admit fenced maintenance beside opted-in native publishers --- crates/lance-context-master/src/config.rs | 7 + crates/lance-context-master/src/task_store.rs | 221 +++++++++++++++++- docs/operations/native-maintenance.md | 37 +++ 3 files changed, 254 insertions(+), 11 deletions(-) create mode 100644 docs/operations/native-maintenance.md diff --git a/crates/lance-context-master/src/config.rs b/crates/lance-context-master/src/config.rs index c1315a3d..84ccf2a2 100644 --- a/crates/lance-context-master/src/config.rs +++ b/crates/lance-context-master/src/config.rs @@ -325,6 +325,12 @@ pub struct MaintenanceConfig { /// source fragments invalidate preparation instead of being overwritten. #[arg(long, env = "COMPACTION_PREPARE_TARGETS", value_delimiter = ',')] pub compaction_prepare_targets: Vec, + /// Exact owned rollout targets whose dedicated publishers all use the + /// common fenced task/target/merge-claim protocol. Allows Compact/IndexId + /// beside persistent catch-up identity; empty retains legacy exclusion. + /// Enable only after verifying publisher compatibility. No wildcard. + #[arg(long, env = "MAINTENANCE_CATCHUP_TARGETS", value_delimiter = ',')] + pub maintenance_catchup_targets: Vec, /// Yield prepared output after this long waiting for another table writer. /// This limits local capacity retention; it never cancels the current writer. #[arg(long, env = "COMPACTION_COMMIT_WAIT_SECS", default_value_t = 120)] @@ -343,6 +349,7 @@ impl Default for MaintenanceConfig { fn default() -> Self { Self { compaction_prepare_targets: Vec::new(), + maintenance_catchup_targets: Vec::new(), compaction_commit_wait_secs: 120, maintenance_timeout_secs: 3600, maintenance_idle_timeout_secs: 600, diff --git a/crates/lance-context-master/src/task_store.rs b/crates/lance-context-master/src/task_store.rs index fe63b778..2b0997a7 100644 --- a/crates/lance-context-master/src/task_store.rs +++ b/crates/lance-context-master/src/task_store.rs @@ -119,6 +119,7 @@ struct EtcdTaskStore { prefix: String, lease_ttl: i64, prepare_targets: Vec, + maintenance_catchup_targets: Vec, rollout: lance_context_merge::rollout::MergeRollout, } @@ -141,6 +142,7 @@ struct ClaimBackend { claim_key: String, target_key: Option, preparation_key: Option, + preparation_owner: Option>, keepalive: LeaseKeepalive, } @@ -206,7 +208,7 @@ impl TaskStore { .transpose() .map_err(|e| lance::Error::io(e.to_string()))?; if values[2].is_some() - || values[3].is_some() + || values[3] != b.preparation_owner || old.as_ref().is_some_and(|e| e.maintenance.is_none()) { return Ok(false); @@ -223,7 +225,10 @@ impl TaskStore { b.token.as_bytes(), ), Compare::version(merge_key.as_str(), CompareOp::Equal, 0), - Compare::version(active_key, CompareOp::Equal, 0), + match &b.preparation_owner { + Some(owner) => Compare::value(active_key, CompareOp::Equal, owner.clone()), + None => Compare::version(active_key, CompareOp::Equal, 0), + }, ]; for (key, value) in [ (execution_key, &values[0]), @@ -651,6 +656,7 @@ impl EtcdTaskStore { prefix: config.etcd.prefix().to_string(), lease_ttl: config.etcd_lease_ttl_secs, prepare_targets: config.maintenance.compaction_prepare_targets.clone(), + maintenance_catchup_targets: config.maintenance.maintenance_catchup_targets.clone(), rollout: config.merge_rollout.clone(), }) } @@ -846,9 +852,20 @@ impl EtcdTaskStore { let expected_owner = merge_execution .as_ref() .map(lance_context_merge::execution_owner); - let expected_job = only_id.map(|(_, job)| job.as_bytes()); + let shared_owner = snapshot[2].is_some() + && matches!(task.kind, TaskKind::Compact | TaskKind::IndexId) + && self.rollout.owned(&task.target) + && !self.rollout.draining(&task.target) + && !task.target.starts_with("generic:") + && self.maintenance_catchup_targets.contains(&task.target); + let expected_job = only_id.map(|(_, job)| job.as_bytes()).or(if shared_owner { + snapshot[2].as_deref() + } else { + None + }); if snapshot[4].is_some() - || (preparing && (snapshot[5].is_some() || snapshot[2].is_some())) + || (preparing + && (snapshot[5].is_some() || (snapshot[2].is_some() && !shared_owner))) || (write_claim && (snapshot[2].as_deref() != expected_job || snapshot[3].is_some() @@ -895,16 +912,16 @@ impl EtcdTaskStore { ]; if let Some(key) = &preparation_key { compares.push(Compare::version(key.as_str(), CompareOp::Equal, 0)); - compares.push(Compare::version( - crate::catchup::store::active_key(&self.prefix, &task.target), - CompareOp::Equal, - 0, - )); + let active_key = crate::catchup::store::active_key(&self.prefix, &task.target); + compares.push(match &snapshot[2] { + Some(owner) => Compare::value(active_key, CompareOp::Equal, owner.clone()), + None => Compare::version(active_key, CompareOp::Equal, 0), + }); } let catchup_key = crate::catchup::store::active_key(&self.prefix, &task.target); if write_claim { - compares.push(match only_id { - Some((_, job)) => Compare::value(catchup_key, CompareOp::Equal, job), + compares.push(match expected_job { + Some(job) => Compare::value(catchup_key, CompareOp::Equal, job), None => Compare::version(catchup_key, CompareOp::Equal, 0), }); } @@ -987,6 +1004,7 @@ impl EtcdTaskStore { claim_key, target_key, preparation_key, + preparation_owner: snapshot[2].clone(), keepalive, }, }), @@ -1015,6 +1033,7 @@ impl EtcdTaskStore { target_key, preparation_key, keepalive, + .. } = claim.backend; let mut task = claim.task; let owns_table = target_key.is_some(); @@ -2340,6 +2359,186 @@ mod tests { store.finish(next, Ok("continued".into())).await.unwrap(); } + #[tokio::test] + #[ignore = "requires ETCD_TEST_ENDPOINTS"] + async fn opted_in_native_owner_allows_preparation_and_exclusive_maintenance() { + let (_dir, mut cfg, _) = preparation_test_store().await; + cfg.maintenance.maintenance_catchup_targets = vec!["hot".into()]; + cfg.merge_rollout.owned_targets = vec!["hot".into()]; + let store = TaskStore::open(&cfg).await.unwrap(); + let active = crate::catchup::store::active_key(&cfg.etcd.etcd_prefix, "hot"); + store + .inner + .client + .clone() + .put(active.clone(), "native", None) + .await + .unwrap(); + store + .enqueue(TaskKind::MergeWal, "hot", Vec::new()) + .await + .unwrap(); + assert!(store + .claim_next_of_kinds(TaskKinds::MERGE_WAL) + .await + .unwrap() + .is_none()); + let native = store + .claim_merge_target("hot", "native") + .await + .unwrap() + .unwrap(); + store + .enqueue(TaskKind::Compact, "hot", Vec::new()) + .await + .unwrap(); + let mut prep = store + .claim_next_of_kinds(TaskKinds::COMPACT) + .await + .unwrap() + .unwrap(); + assert!(prep.preparing_compaction()); + assert!(!store.promote_compaction(&mut prep).await.unwrap()); + store + .finish(native, Ok("native pass joined".into())) + .await + .unwrap(); + assert!(store.promote_compaction(&mut prep).await.unwrap()); + store + .enqueue(TaskKind::MergeWal, "hot", Vec::new()) + .await + .unwrap(); + assert!(store + .claim_merge_target("hot", "native") + .await + .unwrap() + .is_none()); + store + .finish(prep, Ok("compact published".into())) + .await + .unwrap(); + store + .enqueue(TaskKind::IndexId, "hot", Vec::new()) + .await + .unwrap(); + let index = store + .claim_next_of_kinds(TaskKinds::GENERAL.without_compact()) + .await + .unwrap() + .unwrap(); + assert_eq!(index.task.kind, TaskKind::IndexId); + assert!(store + .claim_merge_target("hot", "native") + .await + .unwrap() + .is_none()); + store + .finish(index, Ok("index published".into())) + .await + .unwrap(); + let native = store + .claim_merge_target("hot", "native") + .await + .unwrap() + .unwrap(); + store + .finish(native, Ok("native drainage continues".into())) + .await + .unwrap(); + assert_eq!( + store.inner.get_text(&active).await.unwrap().as_deref(), + Some("native") + ); + } + + #[tokio::test] + #[ignore = "requires ETCD_TEST_ENDPOINTS"] + async fn native_maintenance_opt_in_cannot_bypass_legacy_generic_or_drain_guards() { + for (target, owned, draining, opt_in) in [ + ("hot", false, false, "hot"), + ("hot", true, true, "hot"), + ("generic:hot", true, false, "generic:hot"), + ("hot", true, false, "*"), + ] { + let (_dir, mut cfg, _) = preparation_test_store().await; + cfg.maintenance.maintenance_catchup_targets = vec![opt_in.into()]; + if owned { + cfg.merge_rollout.owned_targets = vec![target.into()]; + } + if draining { + cfg.merge_rollout.drain_targets = vec![target.into()]; + } + let store = TaskStore::open(&cfg).await.unwrap(); + let active = crate::catchup::store::active_key(&cfg.etcd.etcd_prefix, target); + store + .inner + .client + .clone() + .put(active.clone(), "native", None) + .await + .unwrap(); + store + .enqueue(TaskKind::Compact, target, Vec::new()) + .await + .unwrap(); + store + .enqueue(TaskKind::IndexId, target, Vec::new()) + .await + .unwrap(); + assert!(store + .claim_next_of_kinds(TaskKinds::GENERAL) + .await + .unwrap() + .is_none()); + assert_eq!( + store.inner.get_text(&active).await.unwrap().as_deref(), + Some("native") + ); + } + } + + #[tokio::test] + #[ignore = "requires ETCD_TEST_ENDPOINTS"] + async fn native_owner_change_invalidates_prepared_admission() { + let (_dir, mut cfg, _) = preparation_test_store().await; + cfg.maintenance.maintenance_catchup_targets = vec!["hot".into()]; + cfg.merge_rollout.owned_targets = vec!["hot".into()]; + let store = TaskStore::open(&cfg).await.unwrap(); + let active = crate::catchup::store::active_key(&cfg.etcd.etcd_prefix, "hot"); + store + .inner + .client + .clone() + .put(active.clone(), "native", None) + .await + .unwrap(); + store + .enqueue(TaskKind::Compact, "hot", Vec::new()) + .await + .unwrap(); + let mut prep = store + .claim_next_of_kinds(TaskKinds::COMPACT) + .await + .unwrap() + .unwrap(); + store + .inner + .client + .clone() + .put(active.clone(), "replacement", None) + .await + .unwrap(); + assert!(!store.promote_compaction(&mut prep).await.unwrap()); + store + .finish(prep, Err("native identity changed".into())) + .await + .unwrap(); + assert_eq!( + store.inner.get_text(&active).await.unwrap().as_deref(), + Some("replacement") + ); + } + #[tokio::test] #[ignore = "requires ETCD_TEST_ENDPOINTS"] async fn preparation_and_promotion_respect_dedicated_owner() { diff --git a/docs/operations/native-maintenance.md b/docs/operations/native-maintenance.md new file mode 100644 index 00000000..d235916a --- /dev/null +++ b/docs/operations/native-maintenance.md @@ -0,0 +1,37 @@ +# Maintenance beside persistent native catch-up owners + +`MAINTENANCE_CATCHUP_TARGETS` is an explicit comma-separated list of rollout +targets. It defaults to empty and does not support `*`. A target must also be +owned by the merge protocol, non-generic, and not draining. Enable it only +when every dedicated publisher for that target uses the common task claim, +merge claim, target lock, and fenced Catchup execution protocol. + +For opted-in targets, a persistent `catchup-active` identity no longer +permanently excludes Compact and IndexId. Immutable compaction preparation +may overlap an active native merge; publication and index changes still take +the same exclusive writer ownership. Each admission CAS compares the exact +observed dedicated identity. Compaction promotion also requires the identity +recorded at preparation admission; a replacement owner invalidates promotion. +Neither maintenance completion nor cleanup removes the publisher identity. +Ordinary MergeWal and Repair do not receive this exception. + +With `COMPACTION_PREPARE_TARGETS` enabled, compact file preparation can overlap +native progress, then wait for the existing pass to release its claim. The +commit wait remains bounded by `COMPACTION_COMMIT_WAIT_SECS`; it does not cancel +the native writer. Without preparation, Compact takes exclusive ownership +before rewriting. IndexId always takes exclusive ownership. Old/unverified +publishers, generic targets and draining targets retain the original guards. + +Validate against a persistent native supervisor before production enablement: +keep ingestion active; enqueue Compact and IndexId; observe successful +maintenance publications and subsequent native drainage without clearing the +active identity. Verify old/new WAL watermarks, all test payloads, index query +results and memory bounds. A test with ordinary merges alone does not cover +this ownership lifecycle. Gate a canary on current native executable/image and +supervisor identity, then verify a maintained table continues draining. + +This enables eligibility, not a guarantee that maintenance wins every writer +race. A continuously reclaiming publisher may still exhaust a compact's bounded +publication wait. Measure that contention and keep the existing merge/drain +recovery running; do not reset claims or stop progressing writers to force a +maintenance slot.