Skip to content

fix: allow fenced maintenance beside persistent native publishers #314

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
beinan merged 1 commit into from
Oct 5, 2026
Merged

fix: allow fenced maintenance beside persistent native publishers #314

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions crates/lance-context-master/src/config.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<String>,
/// 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<String>,
/// 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)]
Expand All @@ -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,
Expand Down
221 changes: 210 additions & 11 deletions crates/lance-context-master/src/task_store.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -119,6 +119,7 @@ struct EtcdTaskStore {
prefix: String,
lease_ttl: i64,
prepare_targets: Vec<String>,
maintenance_catchup_targets: Vec<String>,
rollout: lance_context_merge::rollout::MergeRollout,
}

Expand All @@ -141,6 +142,7 @@ struct ClaimBackend {
claim_key: String,
target_key: Option<String>,
preparation_key: Option<String>,
preparation_owner: Option<Vec<u8>>,
keepalive: LeaseKeepalive,
}

Expand Down Expand Up @@ -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);
Expand All @@ -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]),
Expand Down Expand Up @@ -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(),
})
}
Expand Down Expand Up @@ -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()
Expand Down Expand Up @@ -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),
});
}
Expand Down Expand Up @@ -987,6 +1004,7 @@ impl EtcdTaskStore {
claim_key,
target_key,
preparation_key,
preparation_owner: snapshot[2].clone(),
keepalive,
},
}),
Expand Down Expand Up @@ -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();
Expand Down Expand Up @@ -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() {
Expand Down
37 changes: 37 additions & 0 deletions docs/operations/native-maintenance.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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.
Loading