From b9e1b2c93005fb36e87d8108824cee527f032cf4 Mon Sep 17 00:00:00 2001 From: Beinan Wang <> Date: Sat, 26 Sep 2026 09:51:51 +0000 Subject: [PATCH 1/2] fix(master): keep the failure count across cooldown windows so they double The cooldown record was leased for the window's own duration, so when the window lapsed the count went with it and the next failure started again from the threshold. On staging every master's sweep re-probed the broken store the instant its 10-minute window closed -- a burst of one task per master, three failures, and a fresh 10-minute window, forever. Lease the record for the maximum window instead. The count now survives the window closing, so the first failure after it is N+1 and the next window doubles as the policy intends. is_cooling_down and list_cooldowns check until_ms against now rather than treating the record's presence as the window. A success still clears the record. Co-Authored-By: Claude Fable 5 --- crates/lance-context-master/src/scheduler.rs | 26 +++++++++++++++ crates/lance-context-master/src/task_store.rs | 32 ++++++++++--------- 2 files changed, 43 insertions(+), 15 deletions(-) diff --git a/crates/lance-context-master/src/scheduler.rs b/crates/lance-context-master/src/scheduler.rs index ed009375..19904b54 100644 --- a/crates/lance-context-master/src/scheduler.rs +++ b/crates/lance-context-master/src/scheduler.rs @@ -1360,6 +1360,32 @@ mod tests { // A manual enqueue is not gated. let manual = enqueue(&state, TaskKind::MergeWal, "broken").await.unwrap(); assert_eq!(manual.target, "broken"); + assert_eq!( + await_terminal(&state, &manual.id).await.state, + TaskState::Failed + ); + + // The count keeps climbing across windows: that third failure (the + // manual one) is recorded as failure 3, not a fresh 1, so the window + // doubles. This is what stops every master re-probing the target the + // instant a window closes. + let mut cd = None; + for _ in 0..40 { + let cs = state.task_store.list_cooldowns().await.unwrap(); + if let Some(c) = cs.iter().find(|c| c.target == "broken" && c.failures >= 3) { + cd = Some(c.clone()); + break; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + let cd = cd.expect("cooldown record must carry the cumulative count"); + assert_eq!(cd.failures, 3); + let window_ms = cd.until_ms.unwrap() - chrono::Utc::now().timestamp_millis(); + assert!( + window_ms > 3600 * 1000 * 3 / 2, + "third failure must get a doubled (2h) window, got {}s", + window_ms / 1000 + ); worker.abort(); } diff --git a/crates/lance-context-master/src/task_store.rs b/crates/lance-context-master/src/task_store.rs index a67b8446..2aa28787 100644 --- a/crates/lance-context-master/src/task_store.rs +++ b/crates/lance-context-master/src/task_store.rs @@ -256,11 +256,14 @@ impl TaskStore { } // Below the threshold the record only carries the failure count; it is // a cooldown only once `until_ms` is set. + // The record outlives the cooldown window so the failure count keeps + // climbing across windows; only `until_ms` says whether we are inside + // one right now. Ok(self .inner .get_cooldown(kind, target) .await? - .is_some_and(|c| c.until_ms.is_some())) + .is_some_and(|c| c.until_ms.is_some_and(|until| until > now_ms()))) } /// Every target currently in cooldown, for operators. @@ -958,11 +961,14 @@ impl EtcdTaskStore { .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. + /// Bump the consecutive-failure count and, past the threshold, set the + /// cooldown window. The record is leased for `policy.max` regardless of + /// the window, so the count survives the window lapsing: a target that + /// fails again right after its cooldown ends is on failure N+1 and gets + /// a doubled window, not a fresh threshold. Without that, every master's + /// sweep re-probed the target the instant the window closed (a burst of + /// one task per master), it failed the threshold again, and the window + /// never grew. Only a success clears the record. async fn record_failure( &self, kind: TaskKind, @@ -977,13 +983,7 @@ impl EtcdTaskStore { .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 duration = policy.duration_for(failures); let record = TaskCooldown { kind, target: target.to_string(), @@ -995,9 +995,11 @@ impl EtcdTaskStore { }, last_error: error.chars().take(512).collect(), }; + // Lease for the longest window so the count outlives any single + // cooldown; a slow trickle of unrelated failures still ages out. let mut client = self.client.clone(); let lease = client - .lease_grant(duration.as_secs().max(1) as i64, None) + .lease_grant(policy.max.as_secs().max(1) as i64, None) .await .map_err(etcd_error("grant cooldown lease"))? .id(); @@ -1050,7 +1052,7 @@ impl EtcdTaskStore { 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())) + .filter(|r| !matches!(r, Ok(c) if !c.until_ms.is_some_and(|until| until > now_ms()))) .collect() } From fb909e5de32d5e3be254346a48ee0c7d4df5495e Mon Sep 17 00:00:00 2001 From: Beinan Wang <> Date: Sat, 26 Sep 2026 09:55:37 +0000 Subject: [PATCH 2/2] style: clippy nonminimal_bool Co-Authored-By: Claude Fable 5 --- crates/lance-context-master/src/task_store.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/lance-context-master/src/task_store.rs b/crates/lance-context-master/src/task_store.rs index 2aa28787..d103df7e 100644 --- a/crates/lance-context-master/src/task_store.rs +++ b/crates/lance-context-master/src/task_store.rs @@ -1052,7 +1052,7 @@ impl EtcdTaskStore { 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_some_and(|until| until > now_ms()))) + .filter(|r| !matches!(r, Ok(c) if c.until_ms.is_none_or(|until| until <= now_ms()))) .collect() }