diff --git a/crates/lance-context-api/src/lib.rs b/crates/lance-context-api/src/lib.rs index 4fd716f..bfb3684 100644 --- a/crates/lance-context-api/src/lib.rs +++ b/crates/lance-context-api/src/lib.rs @@ -1504,6 +1504,21 @@ pub enum TaskState { } /// One unit of scheduled work plus its lifecycle, as surfaced to the queue UI. +/// A task target the master's sweeps are skipping because it has failed +/// repeatedly. Reported by `GET /api/v1/scheduler/cooldowns`. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct TaskCooldown { + pub kind: TaskKind, + pub target: String, + /// Consecutive failures observed. + pub failures: u32, + /// When the cooldown lapses (ms since epoch). `None` while still below + /// the failure threshold (the record only tracks the count). + pub until_ms: Option, + /// The most recent failure's error, truncated. + pub last_error: String, +} + #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] pub struct TaskRecord { /// Time-ordered unique id (UUIDv7). diff --git a/crates/lance-context-master/src/config.rs b/crates/lance-context-master/src/config.rs index 316c4f3..753c684 100644 --- a/crates/lance-context-master/src/config.rs +++ b/crates/lance-context-master/src/config.rs @@ -215,6 +215,22 @@ pub struct MasterConfig { #[arg(long, env = "TASK_HISTORY_TTL_SECS", default_value_t = 86_400)] pub task_history_ttl_secs: u64, + /// Consecutive failures of the same (kind, target) after which the + /// auto-sweeps stop re-enqueueing it for a while. A store whose manifest + /// names a fragment that no longer exists fails every merge and compaction + /// forever; without this, each sweep re-enqueued it and each attempt held a + /// task slot through a full worker fan-out before failing. `0` disables. + #[arg(long, env = "TASK_COOLDOWN_AFTER_FAILURES", default_value_t = 3)] + pub task_cooldown_after_failures: u32, + + /// Cooldown after the threshold is reached; doubles per further failure. + #[arg(long, env = "TASK_COOLDOWN_BASE_SECS", default_value_t = 600)] + pub task_cooldown_base_secs: u64, + + /// Longest cooldown. A success resets the count. + #[arg(long, env = "TASK_COOLDOWN_MAX_SECS", default_value_t = 21_600)] + pub task_cooldown_max_secs: u64, + /// Directory of built UI assets to serve. When unset, only the JSON API is /// exposed. #[arg(long, env = "UI_DIR")] diff --git a/crates/lance-context-master/src/routes.rs b/crates/lance-context-master/src/routes.rs index 8d18db4..d5a0a25 100644 --- a/crates/lance-context-master/src/routes.rs +++ b/crates/lance-context-master/src/routes.rs @@ -13,8 +13,8 @@ use serde::Deserialize; use lance_context_api::{ CompactJobStatus, EnqueueTaskRequest, ExperimentDetail, ExperimentListResponse, - ExperimentRecordsResponse, ExperimentSummary, SqlQueryRequest, SqlQueryResponse, TaskKind, - TaskListResponse, TaskRecord, TaskState, + ExperimentRecordsResponse, ExperimentSummary, SqlQueryRequest, SqlQueryResponse, TaskCooldown, + TaskKind, TaskListResponse, TaskRecord, TaskState, }; use lance_context_core::{rollout_record_to_dto, ListSource, RolloutFilters, RolloutStore}; use tokio::sync::RwLock; @@ -532,6 +532,22 @@ pub async fn enqueue_task( Ok((StatusCode::ACCEPTED, Json(task))) } +/// `GET /api/v1/scheduler/cooldowns` — targets the auto-sweeps are skipping +/// because they failed repeatedly, with the failure count, when the cooldown +/// lapses, and the last error. A store in this list is broken in a way that +/// retrying will not fix (a manifest naming a missing fragment, for example); +/// a manual `POST /tasks` still enqueues it. +pub async fn list_cooldowns( + State(state): State>, +) -> Result>, MasterError> { + state + .task_store + .list_cooldowns() + .await + .map(Json) + .map_err(MasterError::from_lance) +} + /// `GET /api/v1/tasks` — paginated tasks (queue + recent history), newest first. pub async fn list_tasks( State(state): State>, @@ -584,6 +600,7 @@ pub fn api_router() -> Router> { .route("/experiments/{name}/compact", post(compact_experiment)) .route("/experiments/{name}/compact/status", get(compact_status)) .route("/tasks", post(enqueue_task).get(list_tasks)) + .route("/scheduler/cooldowns", get(list_cooldowns)) .route("/tasks/{id}", get(get_task)) .route("/rescan", post(rescan)) } @@ -678,6 +695,9 @@ mod tests { etcd_lease_ttl_secs: 5, task_history_limit: 1_000, task_history_ttl_secs: 86_400, + task_cooldown_after_failures: 3, + task_cooldown_base_secs: 600, + task_cooldown_max_secs: 21_600, ui_dir: None, } } diff --git a/crates/lance-context-master/src/scheduler.rs b/crates/lance-context-master/src/scheduler.rs index 67a58c3..2742ebf 100644 --- a/crates/lance-context-master/src/scheduler.rs +++ b/crates/lance-context-master/src/scheduler.rs @@ -453,6 +453,13 @@ async fn sweep_candidates_inner(state: &Arc) -> lance::Result) -> lance::Result .await?; let mut queued = 0; for row in rows { + if state + .task_store + .is_cooling_down(TaskKind::MergeWal, &row.name) + .await? + { + continue; + } enqueue(state, TaskKind::MergeWal, &row.name).await?; queued += 1; } @@ -692,6 +706,9 @@ mod tests { worker_endpoints: vec![], task_concurrency: 4, merge_wal_concurrency: 4, + task_cooldown_after_failures: 3, + task_cooldown_base_secs: 600, + task_cooldown_max_secs: 21_600, etcd_endpoints: std::env::var("ETCD_TEST_ENDPOINTS") .map(|value| value.split(',').map(str::to_string).collect()) .unwrap_or_default(), @@ -1190,6 +1207,104 @@ mod tests { /// The WAL-merge sweep enqueues a `MergeWal` only for experiments whose /// pending generation count is at or above the threshold, and de-dupes so a + /// A target that keeps failing is skipped by the sweep after the cooldown + /// threshold, so a permanently broken store stops consuming task slots + /// and blocking healthy stores behind it. Manual enqueue bypasses the + /// cooldown, and a later success clears it. + #[tokio::test] + #[ignore = "requires ETCD_TEST_ENDPOINTS"] + async fn repeated_failures_cool_the_target_down_and_sweeps_skip_it() { + use crate::stats_store::StatRow; + + let dir = TempDir::new().unwrap(); + let mut cfg = config(&dir); + cfg.merge_wal_min_generations = 1; + cfg.task_cooldown_after_failures = 2; + cfg.task_cooldown_base_secs = 3600; + // No worker endpoints: every MergeWal fails immediately. + cfg.worker_endpoints = vec![]; + let state = MasterState::new(cfg).await.unwrap(); + let worker = spawn_scheduler(&state); + + { + let mut stats = state.stats.lock().await; + stats + .upsert(&StatRow { + version: StatRow::UNKNOWN_VERSION, + name: "broken".to_string(), + uri: state.rollout_uri("broken"), + row_count: 0, + fragment_count: 0, + last_updated: 0, + pending_wal_generations: 50, + last_compaction: StatRow::NO_COMPACTION, + total_compactions: 0, + scanned_at: 0, + }) + .await + .unwrap(); + } + + // Two sweeps, two failures: the target crosses the threshold. + for round in 0..2 { + assert!( + !state + .task_store + .is_cooling_down(TaskKind::MergeWal, "broken") + .await + .unwrap(), + "round {round}: must not cool down below the failure threshold" + ); + assert_eq!(sweep_merge_wal_candidates(&state).await.unwrap(), 1); + let id = state + .task_store + .list() + .await + .unwrap() + .into_iter() + .filter(|t| t.kind == TaskKind::MergeWal && t.target == "broken") + .max_by_key(|t| t.enqueued_at) + .unwrap() + .id; + assert_eq!(await_terminal(&state, &id).await.state, TaskState::Failed); + } + // `finish` commits the task's terminal state first and then records the + // failure; `await_terminal` returns on the former, so give the latter a + // moment. + let mut cooling = false; + for _ in 0..40 { + if state + .task_store + .is_cooling_down(TaskKind::MergeWal, "broken") + .await + .unwrap() + { + cooling = true; + break; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + assert!(cooling, "target must be cooling down after 2 failures"); + let cooldowns = state.task_store.list_cooldowns().await.unwrap(); + assert_eq!(cooldowns.len(), 1); + assert_eq!(cooldowns[0].target, "broken"); + assert_eq!(cooldowns[0].failures, 2); + assert!(cooldowns[0].until_ms.is_some()); + + // The sweep now skips it. + assert_eq!( + sweep_merge_wal_candidates(&state).await.unwrap(), + 0, + "a cooled-down target must not be re-enqueued by the sweep" + ); + + // A manual enqueue is not gated. + let manual = enqueue(&state, TaskKind::MergeWal, "broken").await.unwrap(); + assert_eq!(manual.target, "broken"); + + worker.abort(); + } + /// second sweep does not pile up a duplicate for the same target. #[tokio::test] #[ignore = "requires ETCD_TEST_ENDPOINTS"] diff --git a/crates/lance-context-master/src/state.rs b/crates/lance-context-master/src/state.rs index 722256e..6fe54d5 100644 --- a/crates/lance-context-master/src/state.rs +++ b/crates/lance-context-master/src/state.rs @@ -278,6 +278,9 @@ mod tests { etcd_lease_ttl_secs: 5, task_history_limit: 1_000, task_history_ttl_secs: 86_400, + task_cooldown_after_failures: 3, + task_cooldown_base_secs: 600, + task_cooldown_max_secs: 21_600, ui_dir: None, } } diff --git a/crates/lance-context-master/src/task_store.rs b/crates/lance-context-master/src/task_store.rs index 96411b4..a67b844 100644 --- a/crates/lance-context-master/src/task_store.rs +++ b/crates/lance-context-master/src/task_store.rs @@ -16,7 +16,7 @@ use etcd_client::{ Certificate, Client, Compare, CompareOp, ConnectOptions, GetOptions, Identity, PutOptions, TlsOptions, Txn, TxnOp, }; -use lance_context_api::{TaskKind, TaskRecord, TaskState}; +use lance_context_api::{TaskCooldown, TaskKind, TaskRecord, TaskState}; use lance_context_core::generate_id; use tokio::sync::oneshot; use tokio::task::JoinHandle; @@ -72,6 +72,36 @@ pub struct TaskStore { inner: Arc, history_limit: usize, history_ttl_secs: u64, + cooldown: CooldownPolicy, +} + +/// When to stop re-enqueueing a target that keeps failing, and for how long. +/// +/// A store whose base-table manifest names a fragment that no longer exists +/// fails every merge and every compaction at the same point, forever. Without +/// memory of that, each sweep re-enqueued it, each attempt fanned out to +/// every worker before failing, and five such stores consumed roughly half +/// the fleet's task slots while healthy stores queued behind them. +#[derive(Debug, Clone, Copy)] +pub struct CooldownPolicy { + /// Consecutive failures before a target is cooled down; `0` disables. + pub after_failures: u32, + /// Cooldown after the threshold is first reached; doubles per further failure. + pub base: Duration, + /// Longest cooldown. + pub max: Duration, +} + +impl CooldownPolicy { + fn duration_for(&self, failures: u32) -> Duration { + let over = failures.saturating_sub(self.after_failures); + let secs = self + .base + .as_secs() + .saturating_mul(1u64 << over.min(20)) + .min(self.max.as_secs()); + Duration::from_secs(secs) + } } struct EtcdTaskStore { @@ -124,6 +154,11 @@ impl TaskStore { inner: Arc::new(EtcdTaskStore::connect(config).await?), history_limit: config.task_history_limit.max(1), history_ttl_secs: config.task_history_ttl_secs, + cooldown: CooldownPolicy { + after_failures: config.task_cooldown_after_failures, + base: Duration::from_secs(config.task_cooldown_base_secs), + max: Duration::from_secs(config.task_cooldown_max_secs), + }, }; store.recover_orphaned().await?; store.prune_terminal_history().await?; @@ -191,10 +226,48 @@ impl TaskStore { claim: TaskClaim, outcome: Result, ) -> lance::Result<()> { + let kind = claim.task.kind; + let target = claim.task.target.clone(); + let failed = outcome.as_ref().err().cloned(); self.inner.finish(claim, outcome).await?; + // Cooldown bookkeeping is best-effort and never fails the completion: + // the task's terminal state is already committed above. + if self.cooldown.after_failures > 0 { + let result = match failed { + Some(error) => { + self.inner + .record_failure(kind, &target, &error, self.cooldown) + .await + } + None => self.inner.clear_cooldown(kind, &target).await, + }; + if let Err(error) = result { + tracing::warn!(kind = ?kind, target = %target, %error, "cooldown bookkeeping failed"); + } + } self.prune_terminal_history().await.map(|_| ()) } + /// Whether the sweeps should skip this target for now because it has + /// failed repeatedly. Manual enqueues are not gated by this. + pub async fn is_cooling_down(&self, kind: TaskKind, target: &str) -> lance::Result { + if self.cooldown.after_failures == 0 { + return Ok(false); + } + // Below the threshold the record only carries the failure count; it is + // a cooldown only once `until_ms` is set. + Ok(self + .inner + .get_cooldown(kind, target) + .await? + .is_some_and(|c| c.until_ms.is_some())) + } + + /// Every target currently in cooldown, for operators. + pub async fn list_cooldowns(&self) -> lance::Result> { + self.inner.list_cooldowns().await + } + /// Try to acquire a named coordination lock without waiting. This is used /// around the shared Lance stats table, whose mutations must have one writer /// across all master replicas. @@ -852,6 +925,135 @@ impl EtcdTaskStore { format!("{}/target-locks/{}", self.prefix, encode_segment(target)) } + fn cooldown_prefix(&self) -> String { + format!("{}/cooldown/", self.prefix) + } + + fn cooldown_key(&self, kind: TaskKind, target: &str) -> String { + format!( + "{}{}/{}", + self.cooldown_prefix(), + kind_label(kind), + encode_segment(target) + ) + } + + async fn get_cooldown( + &self, + kind: TaskKind, + target: &str, + ) -> lance::Result> { + let mut client = self.client.clone(); + let response = client + .get(self.cooldown_key(kind, target), None) + .await + .map_err(etcd_error("get cooldown"))?; + response + .kvs() + .first() + .map(|kv| { + serde_json::from_slice::(kv.value()) + .map_err(|e| lance::Error::io(format!("decode cooldown: {e}"))) + }) + .transpose() + } + + /// Bump the consecutive-failure count and, past the threshold, write the + /// cooldown key under a lease that expires when the cooldown does. The + /// count itself lives in the same key, so a target that keeps failing + /// after its cooldown lapses starts again from the threshold rather than + /// from zero -- the lease is the only thing that ages the record out. + async fn record_failure( + &self, + kind: TaskKind, + target: &str, + error: &str, + policy: CooldownPolicy, + ) -> lance::Result<()> { + let failures = self + .get_cooldown(kind, target) + .await? + .map_or(0, |c| c.failures) + .saturating_add(1); + let now = now_ms(); + let cooling = failures >= policy.after_failures; + let duration = if cooling { + policy.duration_for(failures) + } else { + // Below the threshold, remember the count for a bounded time so a + // slow trickle of unrelated failures does not accumulate forever. + policy.max + }; + let record = TaskCooldown { + kind, + target: target.to_string(), + failures, + until_ms: if cooling { + Some(now + duration.as_millis() as i64) + } else { + None + }, + last_error: error.chars().take(512).collect(), + }; + let mut client = self.client.clone(); + let lease = client + .lease_grant(duration.as_secs().max(1) as i64, None) + .await + .map_err(etcd_error("grant cooldown lease"))? + .id(); + client + .put( + self.cooldown_key(kind, target), + serde_json::to_vec(&record) + .map_err(|e| lance::Error::io(format!("encode cooldown: {e}")))?, + Some(PutOptions::new().with_lease(lease)), + ) + .await + .map_err(etcd_error("put cooldown"))?; + if cooling { + metrics::counter!("master_task_cooldowns_total", "kind" => kind_label(kind)) + .increment(1); + tracing::warn!( + kind = ?kind, + target = %target, + failures, + cooldown_secs = duration.as_secs(), + last_error = %record.last_error, + "target failed repeatedly; sweeps will skip it until the cooldown lapses" + ); + } + Ok(()) + } + + async fn clear_cooldown(&self, kind: TaskKind, target: &str) -> lance::Result<()> { + let mut client = self.client.clone(); + client + .delete(self.cooldown_key(kind, target), None) + .await + .map_err(etcd_error("clear cooldown"))?; + Ok(()) + } + + async fn list_cooldowns(&self) -> lance::Result> { + let mut client = self.client.clone(); + let response = client + .get( + self.cooldown_prefix(), + Some(GetOptions::new().with_prefix()), + ) + .await + .map_err(etcd_error("list cooldowns"))?; + response + .kvs() + .iter() + .map(|kv| { + serde_json::from_slice::(kv.value()) + .map_err(|e| lance::Error::io(format!("decode cooldown: {e}"))) + }) + .filter(|r| !matches!(r, Ok(c) if c.until_ms.is_none())) + .collect() + } + fn dedupe_key(&self, kind: TaskKind, target: &str, depends_on: &[String]) -> Option { should_dedupe(kind, depends_on).then(|| { format!( @@ -1017,6 +1219,9 @@ mod tests { etcd_lease_ttl_secs: 30, task_history_limit: 1_000, task_history_ttl_secs: 86_400, + task_cooldown_after_failures: 3, + task_cooldown_base_secs: 600, + task_cooldown_max_secs: 21_600, ui_dir: None, } }