feat(master): cool down targets that fail repeatedly so sweeps stop re-enqueueing them

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<i64>,
+    /// 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).

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")]

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<Arc<MasterState>>,
+) -> Result<Json<Vec<TaskCooldown>>, 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<Arc<MasterState>>,
@@ -584,6 +600,7 @@ pub fn api_router() -> Router<Arc<MasterState>> {
         .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,
         }
     }

crates/lance-context-master/src/scheduler.rs

@@ -453,6 +453,13 @@ async fn sweep_candidates_inner(state: &Arc<MasterState>) -> lance::Result<usize
         if parse_target(&row.name).0 != StoreKind::Rollout {
             continue;
         }
+        if state
+            .task_store
+            .is_cooling_down(TaskKind::Compact, &row.name)
+            .await?
+        {
+            continue;
+        }
         enqueue(state, TaskKind::Compact, &row.name).await?;
         queued += 1;
     }
@@ -504,6 +511,13 @@ async fn sweep_merge_wal_inner(state: &Arc<MasterState>) -> lance::Result<usize>
         .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"]

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,
         }
     }