From 8e33fe768f45083662548ba8b459b844095e5286 Mon Sep 17 00:00:00 2001 From: Beinan Date: Fri, 9 Oct 2026 19:40:28 +0000 Subject: [PATCH] test(master): leader failover rebuild and sustained watchdog isolation The two test gaps left open across review rounds 5 and 7. Leader failover (spec test 4): master A leads, builds the demand cache and shadow placements over three tables, and is aborted without releasing so its lease expires naturally. Its token is then refused ("leadership lost"). Master B campaigns with only etcd to go on and must rebuild an identical demand cache, identical shadow placements (target, executor, bytes), identical executor totals, and must keep a durable Bound reservation written by the old leader. Mutations checked: a second leader that places differently fails on the placements assertion; a rebuild that drops Bound reservations fails on the durable-reservation assertion. Sustained concurrent isolation: four real index preparations at once for several idle windows - two throttled-but-alive at different rates, one frozen from the start, one frozen partway after real progress. Alive builds must finish past the idle window with real IO and the slower one must take longer; frozen-start is cancelled in [IDLE, IDLE+4) from the start; frozen-mid is cancelled in that window measured from its own freeze, not from the start and not from a neighbour's progress; alive builds outlive both cancellations. The whole scenario is bounded so a never-cancelled build fails rather than hangs. Mutation checked: a shared progress counter across watchdogs fails the test. Both pass against real etcd (9 s and 16 s); planner, watchdog and isolation suites green. Clippy -D warnings, fmt, typos clean. Refs #333. Co-Authored-By: Claude Fable 5 --- crates/lance-context-master/src/planner.rs | 245 +++++++++++++++++++ crates/lance-context-master/src/scheduler.rs | 199 +++++++++++++++ 2 files changed, 444 insertions(+) diff --git a/crates/lance-context-master/src/planner.rs b/crates/lance-context-master/src/planner.rs index 04a4cd7..7080b9e 100644 --- a/crates/lance-context-master/src/planner.rs +++ b/crates/lance-context-master/src/planner.rs @@ -848,6 +848,251 @@ mod tests { assert_eq!(unhex("zz"), None); } + /// Spec test 4 / review rounds 5 and 7: leader failover. A first leader + /// builds the cache and shadow placements, then loses its lease. A + /// second master, with nothing but etcd, must (a) win, (b) rebuild an + /// identical cache from the same events, (c) rebuild identical headroom + /// from heartbeats and durable assignments, and (d) reach the same + /// shadow placement decisions. The deposed leader must be refused on + /// its next write. Nothing lives only in a leader's memory. + #[tokio::test] + #[ignore = "requires ETCD_TEST_ENDPOINTS"] + async fn leader_failover_rebuilds_identical_state_and_fences_the_old_leader() { + use crate::executors::{Assignment, AssignmentState}; + use crate::state::MasterState; + use clap::Parser; + use lance_context_api::TaskKind; + let dir = tempfile::tempdir().unwrap(); + let mut cfg = crate::config::MasterConfig::parse_from([ + "test", + "--data-dir", + dir.path().to_str().unwrap(), + ]); + cfg.etcd.etcd_endpoints = std::env::var("ETCD_TEST_ENDPOINTS") + .unwrap() + .split(',') + .map(str::to_owned) + .collect(); + cfg.etcd.etcd_prefix = format!("/planner-failover/{}", generate_id()); + cfg.planner.planner_enabled = true; + cfg.planner.planner_reconcile_secs = 1; + cfg.etcd_lease_ttl_secs = 5; + cfg.merge_wal_concurrency = 2; + cfg.stats_scan_interval_secs = 0; + cfg.compaction_interval_secs = 0; + cfg.merge_wal_interval_secs = 0; + let a = MasterState::new(cfg.clone()).await.unwrap(); + let b = MasterState::new(cfg.clone()).await.unwrap(); + let keys = Keys::new(&cfg.etcd.etcd_prefix); + let ekeys = crate::executors::Keys::new(&cfg.etcd.etcd_prefix); + let client = a.task_store.etcd_client().clone(); + + // Demand: three tables, mixed shapes. + let coordinator = a.task_store.merge_coordinator(); + for (target, shard, sealed, merged) in [ + ("hot", "s1", 400u64, Some(40u64)), + ("hot", "s2", 30, None), + ("warm", "s1", 20, Some(4)), + ("idle", "s1", 7, Some(7)), + ] { + coordinator + .publish_demand_event( + target, + &DemandEvent { + v: SCHEMA_VERSION, + shard: shard.into(), + sealed_through: sealed, + sealed_bytes_through: sealed * 1_000_000, + merged_through: merged, + flushed_at_ms: 1_700_000_000_000, + writer_epoch: 1, + source: EventSource::Writer, + merged_epoch: None, + sealed_times: Default::default(), + }, + ) + .await + .unwrap(); + } + // Both masters heartbeat as executors throughout. + let hb_a = tokio::spawn({ + let a = a.clone(); + async move { crate::executors::heartbeat_loop_for_test(&a).await } + }); + let hb_b = tokio::spawn({ + let b = b.clone(); + async move { crate::executors::heartbeat_loop_for_test(&b).await } + }); + // A durable non-shadow assignment, as a bound unit would leave. + let bound = Assignment { + v: crate::executors::SCHEMA_VERSION, + unit_id: "bound-1".into(), + kind: TaskKind::MergeWal, + target: "elsewhere".into(), + needs_write_turn: true, + executor: a.admission.status().executor_id.clone(), + planner_token: "old".into(), + reserved_slots: [(TaskKind::MergeWal, 1)].into_iter().collect(), + reserved_bytes: 1 << 20, + state: AssignmentState::Bound, + created_ms: 0, + bind_deadline_ms: 0, + }; + client + .clone() + .put( + ekeys.assignment("elsewhere", &bound.unit_id), + serde_json::to_vec(&bound).unwrap(), + None, + ) + .await + .unwrap(); + + // Leader 1 (master a) campaigns alone and completes a pass. + let lead_a = tokio::spawn({ + let a = a.clone(); + async move { lead(&a).await } + }); + let snapshot = |client: etcd_client::Client, keys: Keys, ekeys: crate::executors::Keys| async move { + let (cache, _) = read_prefix_paged(&client, &keys.demand_prefix()) + .await + .unwrap(); + let mut demand: BTreeMap = BTreeMap::new(); + for kv in cache { + let mut t: TableDemand = serde_json::from_slice(&kv.value).unwrap(); + t.updated_ms = 0; + demand.insert(String::from_utf8_lossy(&kv.key).to_string(), t); + } + let (asg, _) = read_prefix_paged(&client, &ekeys.assignments_prefix()) + .await + .unwrap(); + let mut shadows: Vec<(String, String, u64)> = asg + .iter() + .filter_map(|kv| serde_json::from_slice::(&kv.value).ok()) + .filter(|a| a.state == AssignmentState::Shadow) + .map(|a| (a.target, a.executor, a.reserved_bytes)) + .collect(); + shadows.sort(); + let headroom = crate::executors::load_headroom(&client, &ekeys) + .await + .unwrap(); + (demand, shadows, headroom) + }; + let deadline = std::time::Instant::now() + Duration::from_secs(20); + let first = loop { + let snap = snapshot( + client.clone(), + Keys::new(&cfg.etcd.etcd_prefix), + crate::executors::Keys::new(&cfg.etcd.etcd_prefix), + ) + .await; + if snap.0.len() == 3 && !snap.1.is_empty() && snap.2.len() == 2 { + break snap; + } + assert!( + std::time::Instant::now() < deadline, + "leader 1 never completed a pass: {:?}", + (snap.0.len(), snap.1.len(), snap.2.len()) + ); + tokio::time::sleep(Duration::from_millis(200)).await; + }; + let token_1: String = String::from_utf8( + client + .clone() + .get(keys.leader_token(), None) + .await + .unwrap() + .kvs()[0] + .value() + .to_vec(), + ) + .unwrap(); + assert_eq!( + first.1.len(), + 2, + "hot and warm are due; idle is not: {:?}", + first.1 + ); + + // Leader 1 dies without releasing: abort the task so no revoke runs, + // then let the lease expire naturally. + lead_a.abort(); + let deadline = std::time::Instant::now() + Duration::from_secs(15); + loop { + if client + .clone() + .get(keys.leader(), None) + .await + .unwrap() + .kvs() + .is_empty() + { + break; + } + assert!( + std::time::Instant::now() < deadline, + "leader 1 lease never expired" + ); + tokio::time::sleep(Duration::from_millis(250)).await; + } + // The deposed leader's token can no longer write. + let err = reconcile_once(&a, &keys, &token_1).await.unwrap_err(); + assert!(err.contains("leadership lost"), "{err}"); + + // Leader 2 (master b) campaigns with only etcd to go on. + let lead_b = tokio::spawn({ + let b = b.clone(); + async move { lead(&b).await } + }); + let deadline = std::time::Instant::now() + Duration::from_secs(20); + loop { + let lk = client.clone().get(keys.leader(), None).await.unwrap(); + if let Some(kv) = lk.kvs().first() { + let rec: LeaderRecord = serde_json::from_slice(kv.value()).unwrap(); + if rec.token != token_1 { + break; + } + } + assert!(std::time::Instant::now() < deadline, "leader 2 never won"); + tokio::time::sleep(Duration::from_millis(200)).await; + } + // Wait for leader 2's first pass to replace the shadow set, then compare. + tokio::time::sleep(Duration::from_millis(2500)).await; + let second = snapshot( + client.clone(), + Keys::new(&cfg.etcd.etcd_prefix), + crate::executors::Keys::new(&cfg.etcd.etcd_prefix), + ) + .await; + assert_eq!(second.0, first.0, "demand cache differs after failover"); + assert_eq!(second.1, first.1, "shadow placements differ after failover"); + // Headroom: same executors, same durable (non-shadow) reservations. + for (id, h1) in &first.2 { + let h2 = &second.2[id]; + assert_eq!(h2.slots_total, h1.slots_total, "{id}"); + assert_eq!(h2.bytes_total, h1.bytes_total, "{id}"); + } + let durable = |h: &BTreeMap| -> u64 { + // The Bound assignment's bytes survive on master a's row in both. + h[&a.admission.status().executor_id].bytes_reserved + }; + assert!(durable(&first.2) >= 1 << 20); + assert!( + durable(&second.2) >= 1 << 20, + "durable reservation lost on failover" + ); + lead_b.abort(); + hb_a.abort(); + hb_b.abort(); + let _ = client + .clone() + .delete( + cfg.etcd.etcd_prefix.as_str(), + Some(etcd_client::DeleteOptions::new().with_prefix()), + ) + .await; + } + /// Round-5 finding 1 at production shape: 13,405 tables, two shards /// each (about 12 MB of events, three times the etcd client's 4 MiB /// message cap). A single-shot prefix read fails; the paged read must diff --git a/crates/lance-context-master/src/scheduler.rs b/crates/lance-context-master/src/scheduler.rs index b6bb954..6b1c0f0 100644 --- a/crates/lance-context-master/src/scheduler.rs +++ b/crates/lance-context-master/src/scheduler.rs @@ -2397,6 +2397,205 @@ mod tests { .is_some_and(|d| d.contains("prepared outside write lock"))); } + /// Review rounds 5 and 7: sustained concurrent progress isolation. + /// Four real index preparations run at once for several idle timeouts: + /// two throttled-but-alive at different rates, one frozen from the + /// start, one frozen partway through. Each watchdog must judge only its + /// own scope's completed IO. The alive pair must finish regardless of + /// the frozen pair's silence; the frozen pair must be cancelled within + /// the idle window regardless of the alive pair's progress; and the + /// partway freeze must be detected from its own last progress, not the + /// test start. Runs long enough (> 4 idle windows) that a shared or + /// leaked progress counter would show up as a wrong verdict. + #[tokio::test] + #[ignore = "requires ETCD_TEST_ENDPOINTS; ~20 s"] + async fn concurrent_preparations_are_judged_only_by_their_own_progress() { + use lance_context_core::merge_write_scope::MergeWriteScope; + use lance_context_core::preparation_io::test_support::{Throttle, ThrottleWrapper}; + use std::sync::atomic::Ordering; + const IDLE: u64 = 3; + let dir = TempDir::new().unwrap(); + let mut cfg = config(&dir); + cfg.data_dir = format!("file-object-store://{}", dir.path().display()); + cfg.maintenance.index_prepare_targets = vec!["*".into()]; + cfg.maintenance.maintenance_idle_timeout_secs = IDLE; + cfg.merge_rollout.owned_targets = ["slow-a", "slow-b", "frozen-start", "frozen-mid"] + .map(String::from) + .to_vec(); + cfg.task_concurrency = 4; + let state = MasterState::new(cfg).await.unwrap(); + for target in ["slow-a", "slow-b", "frozen-start", "frozen-mid"] { + let mut writer = RolloutStore::open(&state.rollout_uri(target)) + .await + .unwrap(); + let records: Vec<_> = (0..64) + .map(|i| rollout_record(&format!("{target}-{i}"))) + .collect(); + writer.add(&records).await.unwrap(); + writer.cleanup_own_shard().await.unwrap(); + } + let mut claims = std::collections::HashMap::new(); + for _ in 0..4 { + let claim = state + .task_store + .claim_next_of_kinds(TaskKinds::GENERAL.without_compact()) + .await + .unwrap(); + let claim = match claim { + Some(c) => c, + None => { + // Enqueue lazily so claims come out one per table. + for target in ["slow-a", "slow-b", "frozen-start", "frozen-mid"] { + if !claims.contains_key(target) { + enqueue(&state, TaskKind::IndexId, target).await.unwrap(); + } + } + state + .task_store + .claim_next_of_kinds(TaskKinds::GENERAL.without_compact()) + .await + .unwrap() + .unwrap() + } + }; + assert!(claim.preparing_maintenance()); + claims.insert(claim.task.target.clone(), claim); + } + assert_eq!(claims.len(), 4); + + let run = + |target: &'static str, claim: crate::task_store::TaskClaim, throttle: Throttle| { + let state = state.clone(); + let scope = MergeWriteScope::with_preparation_authorizer_and_io_shim( + Arc::new(PreparationOnly), + Arc::new(ThrottleWrapper(throttle.clone())), + ); + let uri = state.rollout_uri(target); + async move { + let began = std::time::Instant::now(); + let work = scope.run(async { + let store = RolloutStore::open_existing_with_options( + &uri, + state.rollout_store_options(), + ) + .await + .map_err(|e| e.to_string())?; + store + .prepare_id_key_index() + .await + .map(|_| "prepared") + .map_err(|e| e.to_string()) + }); + let outcome = tokio::select! { + done = work => done, + error = watch_maintenance_preparation(&state, &claim, &scope) => Err(error), + }; + ( + outcome, + began.elapsed(), + throttle.ops.load(Ordering::Relaxed), + claim, + ) + } + }; + let slow_a = Throttle::new(Duration::from_millis(1200)); // 2.5x per op below IDLE + let slow_b = Throttle::new(Duration::from_millis(2500)); // just under IDLE per op + let frozen_start = Throttle::new(Duration::ZERO); + frozen_start.frozen.store(true, Ordering::Relaxed); + // Slow enough that the build is still mid-flight when frozen: at + // 1.2 s/op a 64-row build runs well past 2 x IDLE. + let frozen_mid = Throttle::new(Duration::from_millis(1200)); + let freezer = { + let f = frozen_mid.clone(); + tokio::spawn(async move { + // Let it make real progress for ~1.5 idle windows, then stop. + tokio::time::sleep(Duration::from_millis(IDLE * 1500)).await; + f.frozen.store(true, Ordering::Relaxed); + std::time::Instant::now() + }) + }; + let started = std::time::Instant::now(); + let c_a = claims.remove("slow-a").unwrap(); + let c_b = claims.remove("slow-b").unwrap(); + let c_fs = claims.remove("frozen-start").unwrap(); + let c_fm = claims.remove("frozen-mid").unwrap(); + // A frozen build that is never cancelled would hang here; bound the + // whole scenario so a regression fails fast instead. + let (a, b, fs, fm) = tokio::time::timeout(Duration::from_secs(60), async { + tokio::join!( + run("slow-a", c_a, slow_a.clone()), + run("slow-b", c_b, slow_b.clone()), + run("frozen-start", c_fs, frozen_start.clone()), + run("frozen-mid", c_fm, frozen_mid.clone()), + ) + }) + .await + .expect("a frozen preparation was never cancelled: watchdogs are not isolated"); + let froze_at = freezer.await.unwrap(); + let idle = Duration::from_secs(IDLE); + + // Alive pair: finished, each past the idle window, with real IO. + for (name, (outcome, elapsed, ops, _)) in [("slow-a", &a), ("slow-b", &b)] { + assert_eq!( + *outcome, + Ok("prepared"), + "{name} was cancelled: {outcome:?}" + ); + assert!( + *elapsed > idle, + "{name} finished in {elapsed:?}, inside one idle window; vacuous" + ); + assert!(*ops >= 3, "{name} saw {ops} ops; shim not intercepting"); + } + assert!( + b.1 > a.1, + "the slower build must take longer (a={:?}, b={:?}); else rates are not real", + a.1, + b.1 + ); + // Frozen from the start: cancelled in [IDLE, IDLE+4) from test start. + assert_eq!( + fs.0, + Err("maintenance preparation made no progress".to_string()) + ); + assert!( + fs.1 >= idle && fs.1 < idle + Duration::from_secs(4), + "frozen-start at {:?}", + fs.1 + ); + // Frozen midway: cancelled in [IDLE, IDLE+4) from *its own* freeze, + // not from the start and not from a neighbour's progress. + assert_eq!( + fm.0, + Err("maintenance preparation made no progress".to_string()) + ); + let since_freeze = (started + fm.1).saturating_duration_since(froze_at); + assert!( + since_freeze >= idle - Duration::from_millis(500) + && since_freeze < idle + Duration::from_secs(4), + "frozen-mid cancelled {since_freeze:?} after its freeze (total {:?})", + fm.1 + ); + assert!( + fm.2 >= 2, + "frozen-mid should have made progress before freezing: {} ops", + fm.2 + ); + // And the alive builds outlived both cancellations. + assert!( + a.1 > fs.1 && b.1 > fs.1, + "alive builds ended before the frozen one was cancelled" + ); + for (claim, ok) in [(a.3, true), (b.3, true), (fs.3, false), (fm.3, false)] { + let result = if ok { + Ok("prepared".to_string()) + } else { + Err("stalled".to_string()) + }; + state.task_store.finish(claim, result).await.unwrap(); + } + } + /// Review finding 6 on #340: prove the watchdog on a *real* index build /// with real storage behaviour, not hand-placed checkpoints. ///