From 45e0ff5cb0460b90abc7387b960c59389eac7a79 Mon Sep 17 00:00:00 2001 From: Beinan Wang <> Date: Tue, 6 Oct 2026 06:37:07 +0000 Subject: [PATCH] fix: yield new merge claims to compaction ready to commit --- crates/lance-context-master/src/scheduler.rs | 8 + crates/lance-context-master/src/task_store.rs | 236 ++++++++++++++++++ docs/compaction-commit-fairness.md | 30 +++ 3 files changed, 274 insertions(+) create mode 100644 docs/compaction-commit-fairness.md diff --git a/crates/lance-context-master/src/scheduler.rs b/crates/lance-context-master/src/scheduler.rs index fa59a29..f35d52e 100644 --- a/crates/lance-context-master/src/scheduler.rs +++ b/crates/lance-context-master/src/scheduler.rs @@ -395,6 +395,14 @@ async fn wait_for_compaction_commit( claim: &mut TaskClaim, ) -> Result<(), String> { let waiting = std::time::Instant::now(); + if !state + .task_store + .request_compaction_commit(claim) + .await + .map_err(|e| e.to_string())? + { + return Err("compaction preparation claim lost before requesting commit".into()); + } loop { if !state .task_store diff --git a/crates/lance-context-master/src/task_store.rs b/crates/lance-context-master/src/task_store.rs index ddeb320..f1e4974 100644 --- a/crates/lance-context-master/src/task_store.rs +++ b/crates/lance-context-master/src/task_store.rs @@ -183,6 +183,34 @@ impl TaskStore { .all(|v| v.as_deref() == Some(claim.backend.token.as_bytes()))) } + /// Ask subsequent merge admissions to yield after files are ready. This + /// marker shares the preparation lease: crash/finish removes the request. + /// It grants no write ownership and never cancels an admitted merge. + pub(crate) async fn request_compaction_commit(&self, claim: &TaskClaim) -> lance::Result { + let Some(preparation) = &claim.backend.preparation_key else { + return Ok(false); + }; + let b = &claim.backend; + self.inner + .client + .clone() + .txn( + Txn::new() + .when([ + Compare::value(b.claim_key.as_str(), CompareOp::Equal, b.token.as_bytes()), + Compare::value(preparation.as_str(), CompareOp::Equal, b.token.as_bytes()), + ]) + .and_then([TxnOp::put( + self.inner.compaction_commit_key(&claim.task.target), + b.token.as_bytes(), + Some(PutOptions::new().with_lease(b.lease_id)), + )]), + ) + .await + .map(|r| r.succeeded()) + .map_err(etcd_error("request compaction commit turn")) + } + /// Acquire the existing table writer protocol only after immutable files /// are ready. An ambiguous response can adopt only this claim's own token. pub(crate) async fn promote_compaction(&self, claim: &mut TaskClaim) -> lance::Result { @@ -842,6 +870,7 @@ impl EtcdTaskStore { encode_segment(&task.target) ) }), + self.compaction_commit_key(&task.target), ]) .await?; let merge_execution: Option = snapshot[0] @@ -849,6 +878,18 @@ impl EtcdTaskStore { .map(serde_json::from_slice) .transpose() .map_err(|e| lance::Error::io(format!("invalid merge execution: {e}")))?; + // Preparation itself stays concurrent. Once its output is + // ready, yield the next merge turn so the short commit can win. + // An unresolved execution still needs merge recovery; never + // let a scheduling hint block that storage safety path. + if task.kind == TaskKind::MergeWal + && merge_execution.is_none() + && snapshot[6].is_some() + && snapshot[6] == snapshot[5] + { + metrics::counter!("master_merge_yield_to_compaction_total").increment(1); + continue; + } // Avoid lease grant/revoke for a clearly blocked candidate. // The final transaction still checks every ownership predicate. let expected_owner = merge_execution @@ -920,6 +961,22 @@ impl EtcdTaskStore { None => Compare::version(active_key, CompareOp::Equal, 0), }); } + if task.kind == TaskKind::MergeWal && merge_execution.is_none() { + // A commit request arriving between the read and claim + // must win over this stale merge admission attempt. + compares.push(match &snapshot[6] { + Some(value) => Compare::value( + self.compaction_commit_key(&task.target), + CompareOp::Equal, + value.clone(), + ), + None => Compare::version( + self.compaction_commit_key(&task.target), + CompareOp::Equal, + 0, + ), + }); + } let catchup_key = crate::catchup::store::active_key(&self.prefix, &task.target); if write_claim { compares.push(match expected_job { @@ -1442,6 +1499,14 @@ impl EtcdTaskStore { format!("{}/claims/{id}", self.prefix) } + fn compaction_commit_key(&self, target: &str) -> String { + format!( + "{}/compact-commit-ready/{}", + self.prefix, + encode_segment(target) + ) + } + fn target_lock_key(&self, target: &str) -> String { format!("{}/target-locks/{}", self.prefix, encode_segment(target)) } @@ -2476,6 +2541,177 @@ mod tests { store.finish(next, Ok("continued".into())).await.unwrap(); } + #[tokio::test] + #[ignore = "requires ETCD_TEST_ENDPOINTS"] + async fn ready_compaction_gets_the_next_native_merge_turn_without_blocking_preparation() { + 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, "native", None) + .await + .unwrap(); + store + .enqueue(TaskKind::Compact, "hot", vec![]) + .await + .unwrap(); + let mut prep = store + .claim_next_of_kinds(TaskKinds::COMPACT) + .await + .unwrap() + .unwrap(); + store + .enqueue(TaskKind::MergeWal, "hot", vec![]) + .await + .unwrap(); + let native = store + .claim_merge_target("hot", "native") + .await + .unwrap() + .unwrap(); + assert!( + prep.preparing_compaction(), + "preparation must not block native merge admission" + ); + assert!(store.request_compaction_commit(&prep).await.unwrap()); + assert!( + !store.promote_compaction(&mut prep).await.unwrap(), + "running merge must finish cooperatively" + ); + store + .finish(native, Ok("joined pass".into())) + .await + .unwrap(); + store + .enqueue(TaskKind::MergeWal, "hot", vec![]) + .await + .unwrap(); + assert!( + store + .claim_merge_target("hot", "native") + .await + .unwrap() + .is_none(), + "new pass must yield to ready compact" + ); + assert!(store.promote_compaction(&mut prep).await.unwrap()); + store + .finish(prep, Ok("compact committed".into())) + .await + .unwrap(); + assert!(store + .inner + .get_text(&store.inner.compaction_commit_key("hot")) + .await + .unwrap() + .is_none()); + let next = store + .claim_merge_target("hot", "native") + .await + .unwrap() + .unwrap(); + store + .finish(next, Ok("merge resumes".into())) + .await + .unwrap(); + } + + #[tokio::test] + #[ignore = "requires ETCD_TEST_ENDPOINTS"] + async fn ready_compaction_lease_loss_unblocks_merges_and_cannot_publish_a_stale_request() { + let (_dir, _cfg, store) = preparation_test_store().await; + store + .enqueue(TaskKind::Compact, "hot", vec![]) + .await + .unwrap(); + let prep = store + .claim_next_of_kinds(TaskKinds::COMPACT) + .await + .unwrap() + .unwrap(); + assert!(store.request_compaction_commit(&prep).await.unwrap()); + store + .inner + .revoke_lease(prep.backend.lease_id) + .await + .unwrap(); + assert!(!store.request_compaction_commit(&prep).await.unwrap()); + assert!(store + .inner + .get_text(&store.inner.compaction_commit_key("hot")) + .await + .unwrap() + .is_none()); + store + .enqueue(TaskKind::MergeWal, "hot", vec![]) + .await + .unwrap(); + let merge = store + .claim_next_of_kinds(TaskKinds::MERGE_WAL) + .await + .unwrap() + .unwrap(); + store + .finish(merge, Ok("merge remains available".into())) + .await + .unwrap(); + drop(prep); + } + + #[tokio::test] + #[ignore = "requires ETCD_TEST_ENDPOINTS"] + async fn ready_compaction_never_blocks_recovery_of_an_unresolved_merge_execution() { + let (_dir, _cfg, store) = preparation_test_store().await; + store + .enqueue(TaskKind::Compact, "hot", vec![]) + .await + .unwrap(); + let prep = store + .claim_next_of_kinds(TaskKinds::COMPACT) + .await + .unwrap() + .unwrap(); + store + .enqueue(TaskKind::MergeWal, "hot", vec![]) + .await + .unwrap(); + let merge = store + .claim_next_of_kinds(TaskKinds::MERGE_WAL) + .await + .unwrap() + .unwrap(); + let coordinator = store.merge_coordinator(); + let execution = lance_context_merge::Execution::new("hot", "worker", "boot", 600); + assert!(coordinator + .reserve(&store.merge_claim(&merge), &execution) + .await + .unwrap()); + assert!(store.request_compaction_commit(&prep).await.unwrap()); + store.abandon_claim_for_test(merge).await.unwrap(); + let recovery = store + .claim_next_of_kinds(TaskKinds::MERGE_WAL) + .await + .unwrap() + .unwrap(); + assert_eq!(coordinator.get("hot").await.unwrap(), Some(execution)); + store + .finish( + recovery, + Err("fixture retains unresolved storage ownership".into()), + ) + .await + .unwrap(); + store + .finish(prep, Err("fixture preparation ends".into())) + .await + .unwrap(); + } + #[tokio::test] #[ignore = "requires ETCD_TEST_ENDPOINTS"] async fn opted_in_native_owner_allows_preparation_and_exclusive_maintenance() { diff --git a/docs/compaction-commit-fairness.md b/docs/compaction-commit-fairness.md new file mode 100644 index 0000000..d4923e5 --- /dev/null +++ b/docs/compaction-commit-fairness.md @@ -0,0 +1,30 @@ +# Give prepared compaction the next commit turn + +A prepared compaction can repeatedly exhaust its ownership wait budget when a +persistent publisher starts another merge pass as soon as the previous one ends. +The prepared files are discarded and rebuilt while the table keeps accumulating +fragments. Extending a timeout does not give the compactor a turn. + +Once nonempty compaction output is ready, the master publishes a commit-ready +marker under the existing preparation claim lease. Subsequent MergeWal admissions +for that table yield to the matching live preparation. This includes native +catchup target claims. The claim transaction checks the marker again, so a +request arriving after the admission read cannot be missed by a stale claim. +Preparation alone creates no marker: merge can continue throughout file encoding. +An already admitted merge is never cancelled by this scheduling hint. + +The compactor still obtains the normal exclusive write claim and storage fence +before commit. Its existing bounded commit wait applies. Finishing, abandoning +or losing the preparation lease removes the marker; it is not an independent +persistent table lock. The marker is ignored if its token does not match the live +preparation. Other tables continue normally. An unresolved merge execution can +still be claimed for recovery even when compaction is waiting, so the hint cannot +block the storage recovery needed to make the table writable. + +No new per-file or per-generation polling is added: ready state is published once +and read with the existing admission snapshot transaction. The counter +`master_merge_yield_to_compaction_total` counts admission deferrals, not failures. +Deploy compatible scheduling/catchup binaries for every contender to honor the +hint. It does not forcibly interrupt old binaries or guarantee completion before +a progressing current merge finishes. Legacy worker self-merges outside master +admission retain their existing coordination limitations.