From 2b10741e47612de5a8711eac6d9a590104edc858 Mon Sep 17 00:00:00 2001 From: Beinan Date: Thu, 8 Oct 2026 18:27:13 +0000 Subject: [PATCH] refactor(master): extract claim eligibility into a pure module `claim_next` decided per candidate whether this poller may attempt the claim transaction through a chain of inlined `continue`s: kind filter, resident-only filter, preparation shape, commit-ready yield, seven-key ownership predicate and the worker-execution rule. None of it was unit testable without etcd and no skip had a name. Move those decisions to `eligibility.rs` as functions of plain inputs (`pre_dependency_filter`, `claim_shape`, `admission_filter`) returning `Result<(), Skip>`. `claim_next` keeps the same etcd reads in the same order, maps `YieldToPreparedCommit` and `OwnershipBlocked` to the existing counters, and traces every other skip with its reason. No behaviour change. `requires_target_lock` moves unchanged. The `preparing` predicate, which was computed twice with slightly different spelling, now exists once. Validation: 7 new etcd-free unit tests cover resident filtering, preparation opt-in, catch-up sharing, the commit-ready yield including the recovery bypass, every ownership predicate, and the worker-execution rule. Clippy -D warnings, fmt and typos clean. task_store (23) and scheduler (31) etcd-backed suites pass locally. Refs #333. Co-Authored-By: Claude Fable 5 --- .../lance-context-master/src/eligibility.rs | 544 ++++++++++++++++++ crates/lance-context-master/src/lib.rs | 1 + crates/lance-context-master/src/task_store.rs | 162 +++--- 3 files changed, 619 insertions(+), 88 deletions(-) create mode 100644 crates/lance-context-master/src/eligibility.rs diff --git a/crates/lance-context-master/src/eligibility.rs b/crates/lance-context-master/src/eligibility.rs new file mode 100644 index 0000000..8a5a293 --- /dev/null +++ b/crates/lance-context-master/src/eligibility.rs @@ -0,0 +1,544 @@ +//! Pure admission decisions for queued maintenance tasks. +//! +//! `claim_next` in `task_store.rs` scans the etcd queue and, for each +//! candidate, decides whether this poller may attempt the claim transaction. +//! Those decisions used to be inlined as a chain of `continue`s. They are +//! collected here as functions of plain inputs so they can be unit tested +//! without etcd and so every skip has a nameable reason. +//! +//! Nothing in this module reads or writes etcd. The final claim transaction +//! still re-checks every ownership predicate; this is the cheap pre-filter. + +use lance_context_api::TaskKind; +use lance_context_merge::rollout::MergeRollout; + +/// Why a queued task was not attempted by this poller. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Skip { + /// The poller's kind set does not include this task kind. + KindNotRunnable, + /// A resident-only poller saw a task that is not a local append merge + /// on an owned, non-draining, opted-in rollout target. + NotResidentEligible, + /// A prepared compaction or index is waiting for its commit turn on + /// this table; merge yields one admission turn. + YieldToPreparedCommit, + /// Another claim, preparation, catch-up owner or write owner already + /// holds what this task needs. + OwnershipBlocked, + /// Only MergeWal reconciles a worker (non-maintenance) execution. + WorkerExecutionNeedsMerge, +} + +/// Static per-target scheduling configuration consulted by the filter. +pub struct Policy<'a> { + pub rollout: &'a MergeRollout, + pub compaction_prepare_targets: &'a [String], + pub index_prepare_targets: &'a [String], + pub maintenance_catchup_targets: &'a [String], +} + +/// The seven ownership keys `claim_next` snapshots for one candidate. +/// Each is the raw etcd value or `None` when the key is absent. +pub struct Snapshot<'a> { + pub merge_execution: Option<&'a [u8]>, + pub target_lock: Option<&'a [u8]>, + pub catchup_active: Option<&'a [u8]>, + pub merge_claim: Option<&'a [u8]>, + pub task_claim: Option<&'a [u8]>, + pub compact_preparation: Option<&'a [u8]>, + pub compact_commit_ready: Option<&'a [u8]>, +} + +/// How the candidate would be claimed if admitted. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct ClaimShape { + /// Compact/IndexId preparing outside the table write lock. + pub preparing: bool, + /// Whether the claim takes the table write lock. + pub write_claim: bool, + /// Catch-up/maintenance sharing lets this maintenance task run while an + /// external catch-up owner is active on the table. + pub shared_owner: bool, +} + +fn listed(targets: &[String], target: &str) -> bool { + targets.iter().any(|t| t == "*" || t == target) +} + +fn is_generic(target: &str) -> bool { + target.starts_with("generic:") +} + +pub(crate) fn requires_target_lock(kind: TaskKind) -> bool { + matches!( + kind, + TaskKind::Compact | TaskKind::IndexId | TaskKind::Repair | TaskKind::MergeWal + ) +} + +/// Filter applied before any etcd read beyond the queue page itself. +pub fn pre_dependency_filter( + kind: TaskKind, + target: &str, + runnable: impl Fn(TaskKind) -> bool, + resident_targets: Option<&[String]>, + policy: &Policy<'_>, +) -> Result<(), Skip> { + if !runnable(kind) { + return Err(Skip::KindNotRunnable); + } + if let Some(targets) = resident_targets { + let eligible = kind == TaskKind::MergeWal + && !is_generic(target) + && policy.rollout.owned(target) + && !policy.rollout.draining(target) + && listed(targets, target); + if !eligible { + return Err(Skip::NotResidentEligible); + } + } + Ok(()) +} + +/// Decide the claim shape from kind, target and policy alone. +pub fn claim_shape( + kind: TaskKind, + target: &str, + policy: &Policy<'_>, + catchup_active: bool, +) -> ClaimShape { + let preparation_targets: &[String] = match kind { + TaskKind::Compact => policy.compaction_prepare_targets, + TaskKind::IndexId => policy.index_prepare_targets, + _ => &[], + }; + let maintenance = matches!(kind, TaskKind::Compact | TaskKind::IndexId); + let preparing = maintenance + && !is_generic(target) + && !policy.rollout.draining(target) + && listed(preparation_targets, target); + let shared_owner = catchup_active + && maintenance + && policy.rollout.owned(target) + && !policy.rollout.draining(target) + && !is_generic(target) + && policy + .maintenance_catchup_targets + .iter() + .any(|t| t == target); + ClaimShape { + preparing, + write_claim: requires_target_lock(kind) && !preparing, + shared_owner, + } +} + +/// Filter applied after the seven-key ownership snapshot. +/// +/// `expected_owner` is the `merge-execution:{id}` owner string when a merge +/// execution record exists; `execution_is_worker` says that record has no +/// maintenance kind. `expected_job` is the catch-up job a dedicated +/// executor claims under, if any. +pub fn admission_filter( + kind: TaskKind, + shape: ClaimShape, + snapshot: &Snapshot<'_>, + expected_owner: Option<&str>, + execution_is_worker: bool, + expected_job: Option<&[u8]>, +) -> Result<(), Skip> { + let has_execution = snapshot.merge_execution.is_some(); + if kind == TaskKind::MergeWal + && !has_execution + && snapshot.compact_commit_ready.is_some() + && snapshot.compact_commit_ready == snapshot.compact_preparation + { + return Err(Skip::YieldToPreparedCommit); + } + let expected_job = expected_job.or(if shape.shared_owner { + snapshot.catchup_active + } else { + None + }); + if snapshot.task_claim.is_some() + || (shape.preparing + && (snapshot.compact_preparation.is_some() + || (snapshot.catchup_active.is_some() && !shape.shared_owner))) + || (shape.write_claim + && (snapshot.catchup_active != expected_job + || snapshot.merge_claim.is_some() + || snapshot.target_lock != expected_owner.map(str::as_bytes))) + { + return Err(Skip::OwnershipBlocked); + } + if !shape.preparing && kind != TaskKind::MergeWal && has_execution && execution_is_worker { + return Err(Skip::WorkerExecutionNeedsMerge); + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn rollout(owned: &[&str], draining: &[&str]) -> MergeRollout { + MergeRollout { + owned_targets: owned.iter().map(|s| s.to_string()).collect(), + drain_targets: draining.iter().map(|s| s.to_string()).collect(), + } + } + + fn strings(items: &[&str]) -> Vec { + items.iter().map(|s| s.to_string()).collect() + } + + const EMPTY: Snapshot<'static> = Snapshot { + merge_execution: None, + target_lock: None, + catchup_active: None, + merge_claim: None, + task_claim: None, + compact_preparation: None, + compact_commit_ready: None, + }; + + #[test] + fn resident_filter_only_admits_owned_opted_in_non_draining_merges() { + let rollout = rollout( + &["hot", "draining", "excluded", "generic:gs"], + &["draining"], + ); + let policy = Policy { + rollout: &rollout, + compaction_prepare_targets: &[], + index_prepare_targets: &[], + maintenance_catchup_targets: &[], + }; + let targets = strings(&["hot", "draining", "generic:gs"]); + let run = |k| k == TaskKind::MergeWal; + let check = + |kind, target| pre_dependency_filter(kind, target, run, Some(&targets), &policy); + assert_eq!(check(TaskKind::MergeWal, "hot"), Ok(())); + assert_eq!( + check(TaskKind::MergeWal, "legacy"), + Err(Skip::NotResidentEligible) + ); + assert_eq!( + check(TaskKind::MergeWal, "draining"), + Err(Skip::NotResidentEligible) + ); + assert_eq!( + check(TaskKind::MergeWal, "excluded"), + Err(Skip::NotResidentEligible) + ); + assert_eq!( + check(TaskKind::MergeWal, "generic:gs"), + Err(Skip::NotResidentEligible) + ); + assert_eq!(check(TaskKind::Compact, "hot"), Err(Skip::KindNotRunnable)); + let wildcard = strings(&["*"]); + assert_eq!( + pre_dependency_filter( + TaskKind::MergeWal, + "excluded", + run, + Some(&wildcard), + &policy + ), + Ok(()) + ); + assert_eq!( + pre_dependency_filter(TaskKind::MergeWal, "legacy", run, None, &policy), + Ok(()) + ); + } + + #[test] + fn claim_shape_prepares_only_opted_in_maintenance_on_live_rollouts() { + let rollout = rollout(&["t", "d"], &["d"]); + let compact = strings(&["t", "d"]); + let index = strings(&["*"]); + let policy = Policy { + rollout: &rollout, + compaction_prepare_targets: &compact, + index_prepare_targets: &index, + maintenance_catchup_targets: &[], + }; + let shape = claim_shape(TaskKind::Compact, "t", &policy, false); + assert!(shape.preparing && !shape.write_claim); + let shape = claim_shape(TaskKind::Compact, "d", &policy, false); + assert!( + !shape.preparing && shape.write_claim, + "draining never prepares" + ); + let shape = claim_shape(TaskKind::IndexId, "generic:x", &policy, false); + assert!( + !shape.preparing && shape.write_claim, + "generic never prepares" + ); + let shape = claim_shape(TaskKind::IndexId, "anything", &policy, false); + assert!(shape.preparing, "wildcard applies"); + let shape = claim_shape(TaskKind::MergeWal, "t", &policy, false); + assert!(!shape.preparing && shape.write_claim); + let shape = claim_shape(TaskKind::Repair, "t", &policy, false); + assert!(!shape.preparing && shape.write_claim); + } + + #[test] + fn shared_owner_requires_catchup_active_and_exact_listing() { + let rollout = rollout(&["t"], &[]); + let shared = strings(&["t"]); + let policy = Policy { + rollout: &rollout, + compaction_prepare_targets: &[], + index_prepare_targets: &[], + maintenance_catchup_targets: &shared, + }; + assert!(claim_shape(TaskKind::Compact, "t", &policy, true).shared_owner); + assert!(!claim_shape(TaskKind::Compact, "t", &policy, false).shared_owner); + assert!(!claim_shape(TaskKind::MergeWal, "t", &policy, true).shared_owner); + let wildcard = strings(&["*"]); + let policy = Policy { + maintenance_catchup_targets: &wildcard, + ..policy + }; + assert!( + !claim_shape(TaskKind::Compact, "t", &policy, true).shared_owner, + "catch-up sharing is exact-match only" + ); + } + + #[test] + fn merge_yields_to_matching_prepared_commit_only_without_execution() { + let shape = ClaimShape { + preparing: false, + write_claim: true, + shared_owner: false, + }; + let ready = Snapshot { + compact_preparation: Some(b"tok"), + compact_commit_ready: Some(b"tok"), + ..EMPTY + }; + assert_eq!( + admission_filter(TaskKind::MergeWal, shape, &ready, None, false, None), + Err(Skip::YieldToPreparedCommit) + ); + let stale = Snapshot { + compact_preparation: Some(b"new"), + compact_commit_ready: Some(b"old"), + ..EMPTY + }; + assert_eq!( + admission_filter(TaskKind::MergeWal, shape, &stale, None, false, None), + Ok(()), + "a hint from a dead preparer is ignored" + ); + let recovering = Snapshot { + merge_execution: Some(b"{}"), + target_lock: Some(b"merge-execution:x"), + ..ready + }; + assert_eq!( + admission_filter( + TaskKind::MergeWal, + shape, + &recovering, + Some("merge-execution:x"), + true, + None + ), + Ok(()), + "recovery is never blocked by a fairness hint" + ); + let compact = ClaimShape { + preparing: true, + write_claim: false, + shared_owner: false, + }; + assert_eq!( + admission_filter(TaskKind::Compact, compact, &EMPTY, None, false, None), + Ok(()), + "the hint only affects merge" + ); + } + + #[test] + fn ownership_predicates_block_as_before() { + let write = ClaimShape { + preparing: false, + write_claim: true, + shared_owner: false, + }; + assert_eq!( + admission_filter(TaskKind::MergeWal, write, &EMPTY, None, false, None), + Ok(()) + ); + let claimed = Snapshot { + task_claim: Some(b"x"), + ..EMPTY + }; + assert_eq!( + admission_filter(TaskKind::MergeWal, write, &claimed, None, false, None), + Err(Skip::OwnershipBlocked) + ); + let locked = Snapshot { + target_lock: Some(b"other"), + ..EMPTY + }; + assert_eq!( + admission_filter(TaskKind::MergeWal, write, &locked, None, false, None), + Err(Skip::OwnershipBlocked) + ); + let merge_claimed = Snapshot { + merge_claim: Some(b"m"), + ..EMPTY + }; + assert_eq!( + admission_filter(TaskKind::MergeWal, write, &merge_claimed, None, false, None), + Err(Skip::OwnershipBlocked) + ); + let catchup = Snapshot { + catchup_active: Some(b"job-1"), + ..EMPTY + }; + assert_eq!( + admission_filter(TaskKind::MergeWal, write, &catchup, None, false, None), + Err(Skip::OwnershipBlocked) + ); + assert_eq!( + admission_filter( + TaskKind::MergeWal, + write, + &catchup, + None, + false, + Some(b"job-1") + ), + Ok(()), + "the dedicated executor claims under its own job" + ); + let owned = Snapshot { + merge_execution: Some(b"{}"), + target_lock: Some(b"merge-execution:e1"), + ..EMPTY + }; + assert_eq!( + admission_filter( + TaskKind::MergeWal, + write, + &owned, + Some("merge-execution:e1"), + true, + None + ), + Ok(()), + "merge reconciles its own execution's lock" + ); + } + + #[test] + fn preparation_shares_only_with_catchup_when_configured() { + let prepare = ClaimShape { + preparing: true, + write_claim: false, + shared_owner: false, + }; + let catchup = Snapshot { + catchup_active: Some(b"job"), + ..EMPTY + }; + assert_eq!( + admission_filter(TaskKind::Compact, prepare, &catchup, None, false, None), + Err(Skip::OwnershipBlocked) + ); + let shared = ClaimShape { + shared_owner: true, + ..prepare + }; + assert_eq!( + admission_filter(TaskKind::Compact, shared, &catchup, None, false, None), + Ok(()) + ); + let other_prep = Snapshot { + compact_preparation: Some(b"p"), + ..EMPTY + }; + assert_eq!( + admission_filter(TaskKind::Compact, prepare, &other_prep, None, false, None), + Err(Skip::OwnershipBlocked), + "one preparation slot per table" + ); + let write_lock_held = Snapshot { + target_lock: Some(b"someone"), + ..EMPTY + }; + assert_eq!( + admission_filter( + TaskKind::Compact, + prepare, + &write_lock_held, + None, + false, + None + ), + Ok(()), + "preparation does not need the write lock" + ); + } + + #[test] + fn worker_execution_is_only_reconciled_by_merge() { + let write = ClaimShape { + preparing: false, + write_claim: true, + shared_owner: false, + }; + let worker = Snapshot { + merge_execution: Some(b"{}"), + target_lock: Some(b"merge-execution:w"), + ..EMPTY + }; + assert_eq!( + admission_filter( + TaskKind::Compact, + write, + &worker, + Some("merge-execution:w"), + true, + None + ), + Err(Skip::WorkerExecutionNeedsMerge) + ); + assert_eq!( + admission_filter( + TaskKind::Compact, + write, + &worker, + Some("merge-execution:w"), + false, + None + ), + Ok(()), + "a local maintenance execution may be recovered by any table writer" + ); + let prepare = ClaimShape { + preparing: true, + write_claim: false, + shared_owner: false, + }; + assert_eq!( + admission_filter( + TaskKind::IndexId, + prepare, + &worker, + Some("merge-execution:w"), + true, + None + ), + Ok(()), + "preparation runs beside a worker execution" + ); + } +} diff --git a/crates/lance-context-master/src/lib.rs b/crates/lance-context-master/src/lib.rs index 5bac303..892d638 100644 --- a/crates/lance-context-master/src/lib.rs +++ b/crates/lance-context-master/src/lib.rs @@ -4,6 +4,7 @@ pub mod admission; pub mod catchup; pub mod config; pub mod discovery; +pub mod eligibility; pub mod error; mod maintenance_execution; mod merge_execution; diff --git a/crates/lance-context-master/src/task_store.rs b/crates/lance-context-master/src/task_store.rs index 4ca0207..b301943 100644 --- a/crates/lance-context-master/src/task_store.rs +++ b/crates/lance-context-master/src/task_store.rs @@ -19,6 +19,7 @@ use tokio::sync::oneshot; use tokio::task::JoinHandle; use crate::config::MasterConfig; +use crate::eligibility; const TASK_POLL_BATCH: usize = 256; @@ -805,6 +806,15 @@ impl EtcdTaskStore { .map_err(|_| lance::Error::io("etcd returned an invalid queue count")) } + fn eligibility_policy(&self) -> eligibility::Policy<'_> { + eligibility::Policy { + rollout: &self.rollout, + compaction_prepare_targets: &self.prepare_targets, + index_prepare_targets: &self.index_prepare_targets, + maintenance_catchup_targets: &self.maintenance_catchup_targets, + } + } + async fn claim_next( &self, kinds: TaskKinds, @@ -848,19 +858,18 @@ impl EtcdTaskStore { .collect::>>()?; for mut task in queued { - // Skip kinds this caller cannot run *before* the dependency + // Cheap kind/resident filtering happens before the dependency // probe: an unrunnable kind should cost nothing, and this is // what lets a scarce kind be found behind a dominant one. - if !kinds.contains(task.kind) { - continue; - } - if resident_targets.is_some_and(|targets| { - task.kind != TaskKind::MergeWal - || task.target.starts_with("generic:") - || !self.rollout.owned(&task.target) - || self.rollout.draining(&task.target) - || !targets.iter().any(|t| t == "*" || t == &task.target) - }) { + let policy = self.eligibility_policy(); + if let Err(skip) = eligibility::pre_dependency_filter( + task.kind, + &task.target, + |kind| kinds.contains(kind), + resident_targets, + &policy, + ) { + tracing::trace!(task = %task.id, target = %task.target, ?skip, "task skipped"); continue; } match self.dependency_status(&task).await? { @@ -874,28 +883,14 @@ impl EtcdTaskStore { } } - let preparation_targets: &[String] = match task.kind { - TaskKind::Compact => &self.prepare_targets, - TaskKind::IndexId => &self.index_prepare_targets, - _ => &[], - }; - let preparing = matches!(task.kind, TaskKind::Compact | TaskKind::IndexId) - && !task.target.starts_with("generic:") - && !self.rollout.draining(&task.target) - && preparation_targets - .iter() - .any(|t| t == "*" || t == &task.target); - let write_claim = requires_target_lock(task.kind) && !preparing; // Keep the existing wire keys so older masters also yield to // prepared index commits. Compact/IndexId share one preparation // slot per table; append merges need neither preparation slot. - let preparation_key = preparing.then(|| { - format!( - "{}/compact-preparations/{}", - self.prefix, - encode_segment(&task.target) - ) - }); + let preparation_key = format!( + "{}/compact-preparations/{}", + self.prefix, + encode_segment(&task.target) + ); let execution_key = lance_context_merge::execution_key(&self.prefix, &task.target); let snapshot = self .read_values(&[ @@ -904,13 +899,7 @@ impl EtcdTaskStore { crate::catchup::store::active_key(&self.prefix, &task.target), execution_key.replace("/merge-executions/", "/merge-claims/"), self.claim_key(&task.id), - preparation_key.clone().unwrap_or_else(|| { - format!( - "{}/compact-preparations/{}", - self.prefix, - encode_segment(&task.target) - ) - }), + preparation_key.clone(), self.compaction_commit_key(&task.target), ]) .await?; @@ -919,56 +908,60 @@ 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; - } + let shape = eligibility::claim_shape( + task.kind, + &task.target, + &policy, + snapshot[2].is_some(), + ); + let preparing = shape.preparing; + let write_claim = shape.write_claim; + let preparation_key = preparing.then_some(preparation_key); // Avoid lease grant/revoke for a clearly blocked candidate. // The final transaction still checks every ownership predicate. let expected_owner = merge_execution .as_ref() .map(lance_context_merge::execution_owner); - 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() && !shared_owner))) - || (write_claim - && (snapshot[2].as_deref() != expected_job - || snapshot[3].is_some() - || snapshot[1].as_deref() - != expected_owner.as_deref().map(str::as_bytes))) - { - metrics::counter!("master_task_admission_blocked_total").increment(1); - continue; - } - // 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 !preparing - && task.kind != TaskKind::MergeWal - && merge_execution + let expected_job = + only_id + .map(|(_, job)| job.as_bytes()) + .or(if shape.shared_owner { + snapshot[2].as_deref() + } else { + None + }); + let view = eligibility::Snapshot { + merge_execution: snapshot[0].as_deref(), + target_lock: snapshot[1].as_deref(), + catchup_active: snapshot[2].as_deref(), + merge_claim: snapshot[3].as_deref(), + task_claim: snapshot[4].as_deref(), + compact_preparation: snapshot[5].as_deref(), + compact_commit_ready: snapshot[6].as_deref(), + }; + match eligibility::admission_filter( + task.kind, + shape, + &view, + expected_owner.as_deref(), + merge_execution .as_ref() - .is_some_and(|e| e.maintenance.is_none()) - { - continue; + .is_some_and(|e| e.maintenance.is_none()), + expected_job, + ) { + Ok(()) => {} + Err(eligibility::Skip::YieldToPreparedCommit) => { + metrics::counter!("master_merge_yield_to_compaction_total").increment(1); + continue; + } + Err(eligibility::Skip::OwnershipBlocked) => { + metrics::counter!("master_task_admission_blocked_total").increment(1); + continue; + } + Err(skip) => { + tracing::trace!(task = %task.id, target = %task.target, ?skip, "task skipped"); + continue; + } } let token = generate_id(); let lease_id = self.grant_lease().await?; @@ -1862,13 +1855,6 @@ fn should_dedupe(kind: TaskKind, depends_on: &[String]) -> bool { ) } -fn requires_target_lock(kind: TaskKind) -> bool { - matches!( - kind, - TaskKind::Compact | TaskKind::IndexId | TaskKind::Repair | TaskKind::MergeWal - ) -} - fn kind_label(kind: TaskKind) -> &'static str { match kind { TaskKind::Compact => "compact",