From 0b03cbb3bf17f1a5e37754385d5d9a8d7f2e1112 Mon Sep 17 00:00:00 2001 From: Beinan Date: Fri, 7 Aug 2026 22:32:25 +0000 Subject: [PATCH] fix(master): unwedge /experiments and drain compaction backlogs faster Four production failures, all in the master, all reinforcing each other during a backlog. 1. `/experiments` hung behind the stats scan. `state.stats` is an exclusive `Mutex` (every `StatsStore` read takes `&mut self`), and `list_experiments` used it as its only data source with a bare `lock().await` -- no timeout, no fallback, and no `TimeoutLayer` on the router. Any slow scan round wedged the endpoint outright. The round was slow by construction: `scan_once_inner` took the lock and then called `retire_cold_experiments`, a serial loop that per cold experiment opens the store, runs `cleanup_own_shard` (600s bound), *waits on the process-wide compaction semaphore*, and compacts (another 600s). That put up to `N_cold x ~20min` of object-store IO inside the mutex, and blocked on a semaphore while holding it, so an unrelated slow compaction extended the hold further. Retirement needs nothing from the lock -- it reads a local and returns a name set -- so it now runs before the lock is taken. The critical section is one commit again. Independently, the handler now `try_lock`s and falls back to an in-memory snapshot published after each round, flagging the response `stale`. Slightly-old data beats an unbounded hang, and the staleness is visible rather than silent. 2. The sweeps ignored backlog size. `list_above` pushed `limit` into the scan *before* sorting, then sorted by name -- so which rows came back was scan order. An experiment with 15k fragments had no better chance of being swept than one with 17, and the per-sweep cap could be spent entirely on nearly-clean experiments while the worst offenders sat untouched. Now ordered by the threshold column descending, so each sweep spends its budget worst-first. The predicate is still pushed down; only the arbitrary truncation is gone. 3. WAL merge starved compaction. `MergeWal` is an HTTP fan-out to every worker: slow in wall-clock, near free in local resources. Drawing from the single `task_concurrency` pool let a backlog of fan-outs occupy every slot -- exactly when both backlogs are growing and both need to drain. It now has its own budget (`MERGE_WAL_CONCURRENCY`, default 4; `0` restores the shared pool). 4. `task_concurrency` was documented as if it were global. It is per-process, so N replicas run up to N*task_concurrency tasks. Correctness never depended on it (etcd claims and per-target locks already prevent two replicas touching one experiment), so this is a doc fix -- but the old wording invited tuning it as a cluster-wide bound. The `/experiments` test holds the stats lock for the duration of the call, which reproduces the production wedge exactly; it times out against the previous blocking read. Co-Authored-By: Claude Opus 5 --- crates/lance-context-api/src/lib.rs | 9 + crates/lance-context-master/src/config.rs | 21 ++ crates/lance-context-master/src/routes.rs | 209 ++++++++++++++---- crates/lance-context-master/src/scanner.rs | 31 ++- crates/lance-context-master/src/scheduler.rs | 24 +- crates/lance-context-master/src/state.rs | 19 +- .../lance-context-master/src/stats_store.rs | 100 ++++++++- crates/lance-context-master/src/task_store.rs | 1 + crates/lance-context-master/ui/src/App.tsx | 10 + crates/lance-context-master/ui/src/api.ts | 6 + crates/lance-context-master/ui/src/styles.css | 7 + 11 files changed, 384 insertions(+), 53 deletions(-) diff --git a/crates/lance-context-api/src/lib.rs b/crates/lance-context-api/src/lib.rs index 1a17d92..368f029 100644 --- a/crates/lance-context-api/src/lib.rs +++ b/crates/lance-context-api/src/lib.rs @@ -1388,6 +1388,15 @@ pub struct ExperimentListResponse { /// Total number of experiments matching the (optional) search filter, /// ignoring pagination. pub total: i64, + /// Whether this response was served from the master's in-memory snapshot + /// because the stats table was busy. + /// + /// The rows are then at most one scan interval old, and a `search` may omit + /// retired experiments that an uncontended request would have found. + /// Defaults to `false` so older clients and stored payloads deserialize + /// unchanged. + #[serde(default)] + pub stale: bool, } /// Paginated rollout records for one experiment in the master data browser. diff --git a/crates/lance-context-master/src/config.rs b/crates/lance-context-master/src/config.rs index 07d7e74..f2b9ccc 100644 --- a/crates/lance-context-master/src/config.rs +++ b/crates/lance-context-master/src/config.rs @@ -133,9 +133,30 @@ pub struct MasterConfig { /// the *same* experiment is always serialized regardless of this value /// (two `Rewrite`s on one dataset conflict); this only bounds how many /// *distinct* experiments/tasks run at once. + /// + /// This is a per-process limit, not a cluster-wide one: N master replicas + /// run up to `N * task_concurrency` tasks between them. Correctness does + /// not depend on the value -- etcd claims and per-target locks already + /// prevent two replicas touching one experiment -- so it is purely a + /// throughput/resource knob. #[arg(long, env = "TASK_CONCURRENCY", default_value_t = 4)] pub task_concurrency: usize, + /// Maximum `MergeWal` tasks executing concurrently in this process. + /// + /// WAL merge shares the scheduler's dispatch loop with compaction but is a + /// different kind of work: it is an HTTP fan-out to every worker endpoint, + /// so it is slow in wall-clock terms while costing this process almost + /// nothing. Drawing from the single `task_concurrency` pool let a backlog + /// of slow fan-outs occupy every slot and starve compaction -- exactly when + /// both backlogs are growing and both need to drain. Giving merge its own + /// budget decouples them. + /// + /// `0` disables the separate budget and falls back to sharing the general + /// `task_concurrency` pool. + #[arg(long, env = "MERGE_WAL_CONCURRENCY", default_value_t = 4)] + pub merge_wal_concurrency: usize, + /// Comma-separated etcd v3 endpoints. Scheduler state (task queue, /// lease-based claims, per-experiment write locks) lives in etcd so several /// stateless master replicas can share one queue. Required. diff --git a/crates/lance-context-master/src/routes.rs b/crates/lance-context-master/src/routes.rs index 6d23ac7..8d18db4 100644 --- a/crates/lance-context-master/src/routes.rs +++ b/crates/lance-context-master/src/routes.rs @@ -23,6 +23,7 @@ use crate::error::MasterError; use crate::scanner; use crate::scheduler; use crate::state::MasterState; +use crate::stats_store::StatRow; /// Query params for the experiment listing. #[derive(Debug, Deserialize)] @@ -130,56 +131,101 @@ pub async fn list_experiments( ) -> Result, MasterError> { let search = params.search.as_deref().filter(|s| !s.is_empty()); - let (mut experiments, mut total) = { - let mut stats = state.stats.lock().await; - let total = stats.count(search).await.map_err(MasterError::from_lance)?; - let rows = stats - .list(search, params.limit, params.offset) - .await - .map_err(MasterError::from_lance)?; - let experiments: Vec = - rows.into_iter().map(|r| r.into_summary()).collect(); - (experiments, total) + // Try the authoritative table, but never block on it. + // + // `state.stats` is an exclusive mutex held for the whole write phase of a + // scan round. This handler is the only source for the default view, so a + // bare `lock().await` here -- with no timeout, no fallback, and no + // `TimeoutLayer` on the router -- meant a slow round wedged the UI + // outright. Serving the previous snapshot with `stale: true` is strictly + // better: the data is at most one scan interval old, and the caller can + // see that it is. + let (mut experiments, mut total, stale) = match state.stats.try_lock() { + Ok(mut stats) => { + let total = stats.count(search).await.map_err(MasterError::from_lance)?; + let rows = stats + .list(search, params.limit, params.offset) + .await + .map_err(MasterError::from_lance)?; + let experiments: Vec = + rows.into_iter().map(|r| r.into_summary()).collect(); + (experiments, total, false) + } + Err(_) => { + metrics::counter!("master_experiments_served_stale_total").increment(1); + let cached = state.stats_cache.read().await.clone(); + let matching: Vec<&StatRow> = cached + .iter() + .filter(|row| search.is_none_or(|q| row.name.contains(q))) + .collect(); + let total = matching.len() as i64; + let experiments: Vec = matching + .into_iter() + .skip(params.offset) + .take(params.limit) + .map(|row| row.clone().into_summary()) + .collect(); + tracing::debug!( + returned = experiments.len(), + "stats lock busy; serving cached experiment list" + ); + (experiments, total, true) + } }; if let Some(query) = search { // Retired experiments have no stats row, so a search that only consulted // the stats table would silently omit them. The registry is the // authoritative list of what exists. - let known: HashSet = experiments.iter().map(|e| e.name.clone()).collect(); - let matches: Vec<_> = state - .registry - .write() - .await - .list() - .await - .map_err(MasterError::from_lance)? - .into_iter() - .filter(|entry| entry.name.contains(query) && !known.contains(&entry.name)) - .collect(); - - total += matches.len() as i64; - - // Observe the retired matches this page needs. Bounded by the page - // size: a search matching thousands of cold experiments must not open - // thousands of datasets to render one page. - let want = params.limit.saturating_sub(experiments.len()); - for entry in matches.into_iter().take(want) { - match scanner::observe_cold(&state, &entry.name, &entry.uri).await { - Ok(summary) => experiments.push(summary), - Err(e) => { - tracing::warn!( - store = %entry.name, - error = %e, - "search: failed to observe retired experiment" - ); + // + // Skipped when serving stale: this branch takes the registry write lock + // and then opens datasets on demand via `observe_cold`. We are already + // here because the process is busy, so doing the most expensive work in + // the handler is exactly wrong -- it would turn lock contention into a + // second, slower stall. Retired experiments are simply omitted from a + // degraded search; the next uncontended request finds them. + if stale { + tracing::debug!("stats lock busy; skipping registry search fallback"); + } else { + let known: HashSet = experiments.iter().map(|e| e.name.clone()).collect(); + let matches: Vec<_> = state + .registry + .write() + .await + .list() + .await + .map_err(MasterError::from_lance)? + .into_iter() + .filter(|entry| entry.name.contains(query) && !known.contains(&entry.name)) + .collect(); + + total += matches.len() as i64; + + // Observe the retired matches this page needs. Bounded by the page + // size: a search matching thousands of cold experiments must not open + // thousands of datasets to render one page. + let want = params.limit.saturating_sub(experiments.len()); + for entry in matches.into_iter().take(want) { + match scanner::observe_cold(&state, &entry.name, &entry.uri).await { + Ok(summary) => experiments.push(summary), + Err(e) => { + tracing::warn!( + store = %entry.name, + error = %e, + "search: failed to observe retired experiment" + ); + } } } } experiments.sort_by(|a, b| a.name.cmp(&b.name)); } - Ok(Json(ExperimentListResponse { experiments, total })) + Ok(Json(ExperimentListResponse { + experiments, + total, + stale, + })) } /// `GET /api/v1/experiments/{name}` @@ -621,6 +667,7 @@ mod tests { merge_wal_min_generations: 8, worker_endpoints: vec![], task_concurrency: 4, + merge_wal_concurrency: 4, etcd_endpoints: test_etcd_endpoints(), etcd_prefix: format!("/lance-context/test/{}", generate_id()), etcd_username: None, @@ -737,6 +784,92 @@ mod tests { assert_eq!(one.experiments[0].name, "exp-1"); } + /// `/experiments` must answer even while a scan round holds the stats lock. + /// + /// This is the production wedge: the handler's only data source is an + /// exclusive `Mutex`, so a slow scan round (retirement, first-pass + /// maintenance) used to make the endpoint hang with no timeout and no + /// fallback. Holding the lock for the duration of the call reproduces that + /// exactly; the assertion is that we get a bounded, flagged response + /// instead of blocking. + #[tokio::test] + #[ignore = "requires ETCD_TEST_ENDPOINTS"] + async fn list_experiments_serves_cache_when_stats_lock_is_held() { + let dir = TempDir::new().unwrap(); + let state = MasterState::new(test_config(&dir)).await.unwrap(); + + for i in 0..2 { + let name = format!("exp-{i}"); + let uri = state.rollout_uri(&name); + RolloutStore::open(&uri).await.unwrap(); + state + .registry + .write() + .await + .upsert(&name, &uri) + .await + .unwrap(); + } + // Populates both the stats table and the in-memory cache. + scanner::scan_once(&state).await.unwrap(); + + // Simulate a scan round in its write phase. + let held = state.stats.lock().await; + + let listed = tokio::time::timeout( + std::time::Duration::from_secs(5), + list_experiments( + State(state.clone()), + Query(ListParams { + search: None, + limit: 50, + offset: 0, + }), + ), + ) + .await + .expect("must not block on the stats lock"); + + let Json(resp) = listed.unwrap(); + assert!(resp.stale, "a cache-served response must be flagged stale"); + assert_eq!(resp.total, 2); + assert_eq!(resp.experiments.len(), 2); + + // Pagination and search still apply to the cached rows. + let Json(page) = list_experiments( + State(state.clone()), + Query(ListParams { + search: Some("exp-1".to_string()), + limit: 50, + offset: 0, + }), + ) + .await + .unwrap(); + assert!(page.stale); + assert_eq!(page.experiments.len(), 1); + assert_eq!(page.experiments[0].name, "exp-1"); + + drop(held); + + // Once the lock frees, the authoritative path is used again. + let Json(fresh) = list_experiments( + State(state.clone()), + Query(ListParams { + search: None, + limit: 50, + offset: 0, + }), + ) + .await + .unwrap(); + assert!( + !fresh.stale, + "an uncontended request must read the stats table" + ); + assert_eq!(fresh.total, 2); + } + #[tokio::test] #[ignore = "requires ETCD_TEST_ENDPOINTS"] async fn scan_sees_registry_commits_from_another_handle() { diff --git a/crates/lance-context-master/src/scanner.rs b/crates/lance-context-master/src/scanner.rs index ad68268..e29afdf 100644 --- a/crates/lance-context-master/src/scanner.rs +++ b/crates/lance-context-master/src/scanner.rs @@ -226,8 +226,6 @@ async fn scan_once_inner(state: &Arc) -> lance::Result { .filter(|(_, row)| matches!(row, Some((_, true)))) .count(); - let mut stats = state.stats.lock().await; - // Build the round's snapshot. This replaces the whole table in one commit // instead of two commits per experiment, and subsumes the old // reconcile-remove pass: an experiment absent from the registry is simply @@ -256,9 +254,23 @@ async fn scan_once_inner(state: &Arc) -> lance::Result { // dropped, because a row absent from this table is invisible to both // auto-sweeps forever. See `retire_cold_experiments`. // - // Done before the snapshot is written so a retirement takes effect in the - // same commit rather than leaving a round where the row is stale. + // # This must not hold the stats lock + // + // Retirement genuinely rewrites data: per experiment it opens the store, + // runs `cleanup_own_shard` (up to `RETIRE_TIMEOUT`, 600s), waits on the + // process-wide compaction semaphore, and compacts (another 600s). It is a + // serial loop over every cold row. Running that while holding + // `state.stats` put up to `N_cold x ~20min` of object-store IO inside the + // mutex that `GET /experiments` needs for its only data source -- and + // blocked on a semaphore *inside* a mutex, so a slow compaction elsewhere + // in the process extended the hold further. That is the wedge behind + // `/experiments` hanging in production. + // + // Nothing here needs the lock: the loop reads `snapshot` (a local) and + // returns a set of names. Taking the lock afterwards, for the write alone, + // is equivalent and keeps the critical section to one commit. let retire_after = Duration::from_secs(state.config.stats_cold_retire_secs); + let retire_start = std::time::Instant::now(); let retired = retire_cold_experiments( &snapshot, retire_after, @@ -271,8 +283,12 @@ async fn scan_once_inner(state: &Arc) -> lance::Result { if !retired.is_empty() { snapshot.retain(|row| !retired.contains(&row.name)); } + metrics::histogram!("master_stats_retire_duration_seconds") + .record(retire_start.elapsed().as_secs_f64()); metrics::gauge!("master_stats_hot_experiments").set(snapshot.len() as f64); + let mut stats = state.stats.lock().await; + let mut total_rows: i64 = 0; let mut total_fragments: i64 = 0; let mut live_count: usize = 0; @@ -302,6 +318,13 @@ async fn scan_once_inner(state: &Arc) -> lance::Result { tracing::warn!(error = %e, "stats snapshot write failed"); } + // Publish the round's snapshot for the read path. Done after the write so + // the cache never advertises rows that failed to commit, and while still + // holding the stats lock so a concurrent round cannot interleave and + // publish an older snapshot over a newer one. + *state.stats_cache.write().await = Arc::new(snapshot); + drop(stats); + metrics::histogram!("master_scan_duration_seconds").record(scan_start.elapsed().as_secs_f64()); // How much of the round was avoided by the version check. A ratio near 1 is // the healthy state at scale: most experiments are cold and cost only an diff --git a/crates/lance-context-master/src/scheduler.rs b/crates/lance-context-master/src/scheduler.rs index 07c7d39..34783b7 100644 --- a/crates/lance-context-master/src/scheduler.rs +++ b/crates/lance-context-master/src/scheduler.rs @@ -489,13 +489,24 @@ pub fn spawn_scheduler(state: &Arc) -> JoinHandle<()> { let concurrency = state.config.task_concurrency.max(1); let sem = Arc::new(Semaphore::new(concurrency)); + // WAL merge draws from its own budget so a backlog of slow HTTP fan-outs + // cannot occupy every general slot and starve compaction. `0` opts back + // into the shared pool. + let merge_sem = (state.config.merge_wal_concurrency > 0) + .then(|| Arc::new(Semaphore::new(state.config.merge_wal_concurrency))); let dispatch_state = state.clone(); tokio::spawn(async move { loop { if let Ok(queued) = dispatch_state.task_store.queue_depth().await { metrics::gauge!("master_task_queue_depth").set(queued as f64); } - while sem.available_permits() > 0 { + // Poll while *either* pool can accept work; the claimed task's kind + // decides which one it draws from below. + while sem.available_permits() > 0 + || merge_sem + .as_ref() + .is_some_and(|s| s.available_permits() > 0) + { let claim_start = std::time::Instant::now(); match dispatch_state.task_store.claim_next().await { Ok(Some(claim)) => { @@ -505,11 +516,11 @@ pub fn spawn_scheduler(state: &Arc) -> JoinHandle<()> { // spent here is a claimed-but-idle task holding its // per-experiment lock — worth seeing separately. let permit_start = std::time::Instant::now(); - let permit = sem - .clone() - .acquire_owned() - .await - .expect("semaphore never closed"); + let pool = match (claim.task.kind, merge_sem.as_ref()) { + (TaskKind::MergeWal, Some(merge)) => merge.clone(), + _ => sem.clone(), + }; + let permit = pool.acquire_owned().await.expect("semaphore never closed"); let timing = TaskClaimTiming { claim: claim_elapsed, permit_wait: permit_start.elapsed(), @@ -564,6 +575,7 @@ mod tests { merge_wal_min_generations: 2, worker_endpoints: vec![], task_concurrency: 4, + merge_wal_concurrency: 4, etcd_endpoints: std::env::var("ETCD_TEST_ENDPOINTS") .map(|value| value.split(',').map(str::to_string).collect()) .unwrap_or_default(), diff --git a/crates/lance-context-master/src/state.rs b/crates/lance-context-master/src/state.rs index 94928ce..c21330f 100644 --- a/crates/lance-context-master/src/state.rs +++ b/crates/lance-context-master/src/state.rs @@ -11,7 +11,7 @@ use tokio::sync::{Mutex, RwLock, Semaphore}; use crate::config::MasterConfig; use crate::discovery; -use crate::stats_store::StatsStore; +use crate::stats_store::{StatRow, StatsStore}; use crate::task_store::TaskStore; /// Bounded number of rollout handles retained by the master data browser. @@ -62,6 +62,21 @@ pub struct MasterState { pub registry: RwLock, /// Periodically-refreshed per-experiment metrics (master-owned). pub stats: Mutex, + /// Last snapshot written to the stats table, kept in memory so + /// `GET /experiments` can answer without touching `stats`. + /// + /// `stats` is an exclusive `Mutex` because every `StatsStore` read method + /// takes `&mut self`, so the read path has no way to run concurrently with + /// a scan round. When a round is slow, requests queued on that mutex with + /// no timeout and no fallback -- the handler simply hung, which is how a + /// wedged scan turned into a wedged UI. + /// + /// This cache is the fallback. The handler tries the lock, and on + /// contention serves the last snapshot and marks the response `stale`. A + /// slightly-old list beats an unbounded hang, and the staleness is visible + /// to the caller rather than silent. + pub stats_cache: RwLock>>, + /// Read handles used by the records browser, retained to reuse Lance /// session and fragment metadata caches across requests. record_stores: Mutex>>>, @@ -129,6 +144,7 @@ impl MasterState { let state = Arc::new(Self { registry: RwLock::new(registry), stats: Mutex::new(stats), + stats_cache: RwLock::new(Arc::new(Vec::new())), record_stores: Mutex::new(LruCache::new( NonZeroUsize::new(RECORD_STORE_CACHE_CAPACITY).unwrap(), )), @@ -221,6 +237,7 @@ mod tests { merge_wal_min_generations: 8, worker_endpoints: vec![], task_concurrency: 4, + merge_wal_concurrency: 4, etcd_endpoints: std::env::var("ETCD_TEST_ENDPOINTS") .map(|value| value.split(',').map(str::to_string).collect()) .unwrap_or_default(), diff --git a/crates/lance-context-master/src/stats_store.rs b/crates/lance-context-master/src/stats_store.rs index e06d586..a493408 100644 --- a/crates/lance-context-master/src/stats_store.rs +++ b/crates/lance-context-master/src/stats_store.rs @@ -411,6 +411,23 @@ impl StatsStore { /// /// `column` is a fixed identifier chosen by the two callers above, never /// caller input, so it is safe to interpolate; `threshold` is an `i64`. + /// + /// # Why the limit is not pushed into the scan + /// + /// It used to be: `scanner.limit(limit)` ran *before* any ordering, and the + /// rows were then sorted by `name`. Which `limit` rows came back was + /// therefore scan order — effectively arbitrary. During a backlog that is + /// actively harmful: an experiment with 15k fragments had no better chance + /// of being swept than one with 17, so the worst offenders could go + /// untouched for many sweeps while the cap was spent on nearly-clean + /// experiments. + /// + /// The predicate is still pushed down, so the scan reads only rows above + /// the threshold — the property that pushdown was added for. Ordering by + /// the threshold column descending means each sweep spends its budget on + /// the largest backlogs first. The materialised set is bounded by how many + /// experiments are genuinely above the threshold, and each row is a handful + /// of scalars. async fn list_above( &mut self, column: &str, @@ -420,13 +437,18 @@ impl StatsStore { self.dataset.checkout_latest().await?; let mut scanner = self.dataset.scan(); scanner.filter(&format!("{column} >= {threshold}"))?; - scanner.limit(Some(limit as i64), None)?; let mut stream = scanner.try_into_stream().await?; let mut rows = Vec::new(); while let Some(batch) = stream.try_next().await? { rows.extend(Self::batch_to_rows(&batch)?); } - rows.sort_by(|a, b| a.name.cmp(&b.name)); + // Worst backlog first; `name` only breaks ties, so pagination-style + // stability is preserved among equal counts. + let key = |row: &StatRow| match column { + "pending_wal_generations" => row.pending_wal_generations, + _ => row.fragment_count, + }; + rows.sort_by(|a, b| key(b).cmp(&key(a)).then_with(|| a.name.cmp(&b.name))); rows.truncate(limit); Ok(rows) } @@ -681,6 +703,75 @@ mod tests { with.last_compaction = 123; assert_eq!(with.into_summary().last_compaction, Some(123)); } + + /// The sweeps' budget must be spent on the worst backlogs. + /// + /// The limit used to be pushed into the scan and applied *before* the sort, + /// so which rows came back was scan order. During a real backlog that meant + /// a 15k-fragment experiment could lose its slot to a 17-fragment one for + /// many sweeps in a row. + #[tokio::test] + async fn list_above_fragment_count_returns_worst_backlog_first() { + let dir = TempDir::new().unwrap(); + let mut s = new_store(&dir).await; + + // Inserted in ascending order so a scan-order result would return the + // smallest first, and name order would return "exp-a" (the smallest). + for (name, fragments) in [("exp-a", 20), ("exp-b", 5_000), ("exp-c", 15_000)] { + let mut row = sample(name, 100); + row.fragment_count = fragments; + s.upsert(&row).await.unwrap(); + } + + let rows = s.list_above_fragment_count(16, 2).await.unwrap(); + assert_eq!(rows.len(), 2, "the cap must still be honored"); + assert_eq!( + rows.iter().map(|r| r.name.as_str()).collect::>(), + vec!["exp-c", "exp-b"], + "the two largest backlogs must be swept first, not the first two scanned" + ); + } + + /// Rows below the threshold must stay excluded — ordering by backlog must + /// not accidentally widen what the sweep acts on. + #[tokio::test] + async fn list_above_fragment_count_still_applies_threshold() { + let dir = TempDir::new().unwrap(); + let mut s = new_store(&dir).await; + for (name, fragments) in [("small", 2), ("big", 900)] { + let mut row = sample(name, 10); + row.fragment_count = fragments; + s.upsert(&row).await.unwrap(); + } + let rows = s.list_above_fragment_count(16, 10).await.unwrap(); + assert_eq!(rows.len(), 1); + assert_eq!(rows[0].name, "big"); + } + + /// The WAL sweep orders by its own column, not by `fragment_count`. + #[tokio::test] + async fn list_above_pending_wal_orders_by_generations() { + let dir = TempDir::new().unwrap(); + let mut s = new_store(&dir).await; + + // `worst` has the most pending generations but the fewest fragments, so + // ordering by the wrong column puts it last. + let mut worst = sample("worst", 10); + worst.pending_wal_generations = 98; + worst.fragment_count = 1; + let mut mild = sample("mild", 10); + mild.pending_wal_generations = 9; + mild.fragment_count = 9_000; + s.upsert(&worst).await.unwrap(); + s.upsert(&mild).await.unwrap(); + + let rows = s.list_above_pending_wal(8, 1).await.unwrap(); + assert_eq!(rows.len(), 1); + assert_eq!( + rows[0].name, "worst", + "the WAL sweep must rank by pending generations, not fragments" + ); + } } /// Tests pinning the write-amplification and pagination properties that make @@ -854,8 +945,9 @@ mod scale_tests { .iter() .map(|r| r.fragment_count) .collect::>(), - vec![16, 17, 18, 19], - "threshold query must return exactly the rows over the threshold" + vec![19, 18, 17, 16], + "threshold query must return exactly the rows over the threshold, \ + worst backlog first" ); let mergeable = s.list_above_pending_wal(30, 100).await.unwrap(); diff --git a/crates/lance-context-master/src/task_store.rs b/crates/lance-context-master/src/task_store.rs index 5bc7090..203de90 100644 --- a/crates/lance-context-master/src/task_store.rs +++ b/crates/lance-context-master/src/task_store.rs @@ -941,6 +941,7 @@ mod tests { merge_wal_min_generations: 8, worker_endpoints: vec![], task_concurrency: 4, + merge_wal_concurrency: 4, etcd_endpoints: vec![], etcd_prefix: "/test".to_string(), etcd_username: None, diff --git a/crates/lance-context-master/ui/src/App.tsx b/crates/lance-context-master/ui/src/App.tsx index 7b7a3af..a173175 100644 --- a/crates/lance-context-master/ui/src/App.tsx +++ b/crates/lance-context-master/ui/src/App.tsx @@ -1133,6 +1133,7 @@ function ExperimentsList() { const total = list.data?.total ?? 0; const maxPage = Math.max(0, Math.ceil(total / pageSize) - 1); const rows = list.data?.experiments ?? []; + const stale = list.data?.stale ?? false; // Aggregate the visible page for the stat strip. const pageRows = rows.reduce((a, e) => a + e.row_count, 0); @@ -1158,6 +1159,15 @@ function ExperimentsList() { />
+ {stale && ( + + + cached + + )} {fmtInt(total)} total
diff --git a/crates/lance-context-master/ui/src/api.ts b/crates/lance-context-master/ui/src/api.ts index 6378f42..8ee1ff4 100644 --- a/crates/lance-context-master/ui/src/api.ts +++ b/crates/lance-context-master/ui/src/api.ts @@ -16,6 +16,12 @@ export interface ExperimentSummary { export interface ExperimentListResponse { experiments: ExperimentSummary[]; total: number; + /** + * Set when the master served this list from its in-memory snapshot because + * the stats table was busy. Rows are then at most one scan interval old, and + * a search may omit retired experiments. + */ + stale?: boolean; } export interface Relationship { diff --git a/crates/lance-context-master/ui/src/styles.css b/crates/lance-context-master/ui/src/styles.css index c6c7b18..931fe95 100644 --- a/crates/lance-context-master/ui/src/styles.css +++ b/crates/lance-context-master/ui/src/styles.css @@ -454,6 +454,13 @@ td .muted { background: var(--err-dim); border-color: rgba(248, 81, 73, 0.3); } +/* Served from the master's in-memory snapshot because the stats table was + busy. Not an error -- the data is just up to one scan interval old. */ +.pill--stale { + color: var(--warn, #d29922); + background: var(--warn-dim, rgba(210, 153, 34, 0.12)); + border-color: rgba(210, 153, 34, 0.3); +} /* ---- Pagination --------------------------------------------------------- */