From 04a1872a1b2204686729891c3667055c109103cb Mon Sep 17 00:00:00 2001 From: Beinan Date: Fri, 9 Oct 2026 03:08:09 +0000 Subject: [PATCH] fix: demand pipeline at production scale Review round 5 brought production numbers: 13,405 tables, 354 with pending WAL, 318 generic. Six findings, all confirmed and fixed here. 1. Full prefix reads fail past the etcd client's 4 MiB cap (~27k shard records). Every planner read (demand-events, demand, executors, assignments, queue, running) is now paged at 512 keys at one revision. Reconcile runs detached from the leader loop so a slow pass cannot starve the lease keepalive; one pass in flight, skipped ticks counted. 2. The scan read only sealed, so a lost merge-completion event left a fully merged shard pending forever and an old table's first contact read generation 1001 as 1001 pending. all_shard_watermarks now also reads the base table's merged watermark per shard and the scan event carries it. 3. Scan events for generic stores were published under generic:generic:; entry.name already carries the prefix. 4. Shadow placement reserved 1 byte when pending_bytes was unknown (every real publisher reports 0). An unknown now reserves the full per-pass cost (2 x ROLLOUT_APPEND_MAX_BYTES) and is counted. 5. One "disagreement" counter mixed discovery, capacity and ownership effects. Replaced by planner_shadow_comparison{outcome} with discovery_miss, placement_gap, phantom_task and agree, each a precise question, documented as observations and not a takeover gate; the old counter remains as their sum. 6. Age after merging past the kept entries: 128 pending, merge 64 gave "unknown" in memory and generation 128's time after rebuild. The times map is now bounded by cap only (pruning by merged made the kept set order dependent), keeps the lowest 63 plus the newest, and is filtered by merged on read. The answer carries a bound - Exact when the oldest pending generation's time is kept, Upper otherwise - and only an Exact age drives scoring and promotion. Executor events no longer seed flush times (their timestamp is observation time). Tests: a production-shape test seeds 13,405 tables x 2 shards, asserts the unpaged read fails, the paged read returns all 26,810 records at one revision, the fold counts them, a leader completes a reconcile while its lease survives twelve seconds past the pass, and cleans up after itself (gated on PLANNER_SCALE_TEST=1; run alone). Demand tests cover the 128/64 case as a flagged Upper bound identical after rebuild and in reverse order, scan-with-merged folding, and the randomized rebuild-equals-fold property still holds with the new bookkeeping. Scoring asserts a bound never promotes or boosts. 16 demand, 5 scoring, 17 planner/executor/scan/resident etcd tests and 9 sweeper tests pass on a clean etcd. Clippy -D warnings, fmt, typos clean. Spec updated. Refs #333. Review round 5. Co-Authored-By: Claude Fable 5 --- crates/lance-context-core/src/store_base.rs | 16 + .../src/demand_publish.rs | 2 +- crates/lance-context-master/src/executors.rs | 27 +- crates/lance-context-master/src/planner.rs | 395 ++++++++++++++++-- crates/lance-context-master/src/scanner.rs | 7 +- crates/lance-context-master/src/scoring.rs | 44 +- crates/lance-context-master/src/task_store.rs | 21 + crates/lance-context-merge/src/demand.rs | 200 +++++++-- docs/design/scheduler-p1-data-model.md | 61 ++- 9 files changed, 657 insertions(+), 116 deletions(-) diff --git a/crates/lance-context-core/src/store_base.rs b/crates/lance-context-core/src/store_base.rs index c7dfb9c2..06bfaf52 100644 --- a/crates/lance-context-core/src/store_base.rs +++ b/crates/lance-context-core/src/store_base.rs @@ -224,6 +224,10 @@ pub struct ShardWatermark { pub sealed_through: u64, /// Flushed generations not yet merged into the base table. pub pending_generations: usize, + /// Highest generation the base table's MemWAL index records as merged + /// for this shard, when the caller read the base index. `None` means + /// "not read", not "nothing merged". + pub merged_through: Option, } pub fn is_pending_generations_exceeded(err: &LanceError) -> bool { @@ -948,6 +952,7 @@ impl StorageBase { .max() .unwrap_or(0), pending_generations: manifest.flushed_generations.len(), + merged_through: None, })) } @@ -2091,6 +2096,15 @@ impl StorageBase { if self.is_version_pinned() { return Ok(Vec::new()); } + // The base table's MemWAL index is the authority on what has been + // merged. A scan must report it, or a lost completion event leaves + // a fully merged shard looking pending forever, and an old table's + // first contact reads generation 1001 as 1001 pending. + let merged = Arc::new( + crate::rollout_append::watermarks(&self.dataset) + .await + .unwrap_or_default(), + ); let object_store = self.dataset.object_store(None).await?; let branch_path = self.dataset.branch_location().path.clone(); let shard_ids = self.dataset.list_mem_wal_latest_shard_ids().await?; @@ -2098,6 +2112,7 @@ impl StorageBase { .map(|shard_id| { let object_store = object_store.clone(); let branch_path = branch_path.clone(); + let merged = merged.clone(); async move { let manifest_store = ShardManifestStore::new( object_store, @@ -2116,6 +2131,7 @@ impl StorageBase { .max() .unwrap_or(0), pending_generations: manifest.flushed_generations.len(), + merged_through: merged.get(&manifest.shard_id).copied(), } })) } diff --git a/crates/lance-context-master/src/demand_publish.rs b/crates/lance-context-master/src/demand_publish.rs index 5377cef3..dd560515 100644 --- a/crates/lance-context-master/src/demand_publish.rs +++ b/crates/lance-context-master/src/demand_publish.rs @@ -65,7 +65,7 @@ pub(crate) fn spawn_scan_demand( shard: mark.shard_id.to_string(), sealed_through: mark.sealed_through, sealed_bytes_through: 0, - merged_through: None, + merged_through: mark.merged_through, flushed_at_ms: now_ms, writer_epoch: mark.writer_epoch, source: EventSource::Scan, diff --git a/crates/lance-context-master/src/executors.rs b/crates/lance-context-master/src/executors.rs index 49bceca1..4a849c3b 100644 --- a/crates/lance-context-master/src/executors.rs +++ b/crates/lance-context-master/src/executors.rs @@ -16,7 +16,7 @@ use std::collections::BTreeMap; use std::sync::Arc; use std::time::Duration; -use etcd_client::{GetOptions, PutOptions}; +use etcd_client::PutOptions; use lance_context_api::TaskKind; use serde::{Deserialize, Serialize}; @@ -322,38 +322,23 @@ pub(crate) async fn load_headroom( client: &etcd_client::Client, keys: &Keys, ) -> Result, String> { - let mut client = client.clone(); - let hbs = client - .get( - keys.executors_prefix(), - Some(GetOptions::new().with_prefix()), - ) - .await - .map_err(|e| e.to_string())?; + let (hbs, _) = crate::planner::read_prefix_paged(client, &keys.executors_prefix()).await?; let prefix = keys.executors_prefix(); let heartbeats: Vec<(String, Heartbeat)> = hbs - .kvs() .iter() .filter_map(|kv| { - let id = std::str::from_utf8(kv.key()) + let id = std::str::from_utf8(&kv.key) .ok()? .strip_prefix(prefix.as_str())? .to_string(); - let hb: Heartbeat = serde_json::from_slice(kv.value()).ok()?; + let hb: Heartbeat = serde_json::from_slice(&kv.value).ok()?; (hb.v == SCHEMA_VERSION).then_some((id, hb)) }) .collect(); - let asg = client - .get( - keys.assignments_prefix(), - Some(GetOptions::new().with_prefix()), - ) - .await - .map_err(|e| e.to_string())?; + let (asg, _) = crate::planner::read_prefix_paged(client, &keys.assignments_prefix()).await?; let assignments: Vec = asg - .kvs() .iter() - .filter_map(|kv| serde_json::from_slice::(kv.value()).ok()) + .filter_map(|kv| serde_json::from_slice::(&kv.value).ok()) .filter(|a| a.v == SCHEMA_VERSION) .collect(); Ok(headroom( diff --git a/crates/lance-context-master/src/planner.rs b/crates/lance-context-master/src/planner.rs index b6bf1a7c..c3160800 100644 --- a/crates/lance-context-master/src/planner.rs +++ b/crates/lance-context-master/src/planner.rs @@ -69,6 +69,71 @@ pub(crate) fn unhex(encoded: &str) -> Option { String::from_utf8(bytes?).ok() } +/// A key/value/revision triple from a paged prefix read. +pub(crate) struct Kv { + pub key: Vec, + pub value: Vec, + pub mod_revision: i64, +} + +/// Page size for prefix reads. 13k tables x a few shards each must never go +/// through one response: the etcd client caps a message at 4 MiB. +const PAGE: i64 = 512; + +/// Read every key under `prefix` in pages, all at one revision so the +/// result is a consistent snapshot. Returns the kvs and that revision. +pub(crate) async fn read_prefix_paged( + client: &etcd_client::Client, + prefix: &str, +) -> Result<(Vec, i64), String> { + let mut client = client.clone(); + let range_end = { + let mut end = prefix.as_bytes().to_vec(); + for i in (0..end.len()).rev() { + if end[i] != u8::MAX { + end[i] += 1; + end.truncate(i + 1); + break; + } + } + end + }; + let mut out = Vec::new(); + let mut start = prefix.as_bytes().to_vec(); + let mut revision: i64 = 0; + loop { + let mut opts = GetOptions::new() + .with_range(range_end.clone()) + .with_limit(PAGE); + if revision > 0 { + opts = opts.with_revision(revision); + } + let response = client + .get(start.clone(), Some(opts)) + .await + .map_err(|e| e.to_string())?; + if revision == 0 { + revision = response.header().map_or(0, |h| h.revision()); + } + let more = response.more(); + let last = response.kvs().last().map(|kv| kv.key().to_vec()); + for kv in response.kvs() { + out.push(Kv { + key: kv.key().to_vec(), + value: kv.value().to_vec(), + mod_revision: kv.mod_revision(), + }); + } + let Some(last) = last else { break }; + if !more { + break; + } + start = last; + start.push(0); + } + Ok((out, revision)) +} + pub(crate) struct Keys { prefix: String, } @@ -213,6 +278,7 @@ async fn lead(state: &Arc) -> Result { let reconcile_every = Duration::from_secs(state.config.planner.planner_reconcile_secs.max(5)); let mut reconcile = tokio::time::interval(reconcile_every); let mut keepalive = tokio::time::interval(keepalive_every); + let mut inflight: Option> = None; let result = loop { tokio::select! { _ = keepalive.tick() => { @@ -228,14 +294,32 @@ async fn lead(state: &Arc) -> Result { if !state.admission.status().accepting { break Ok(true); } - if let Err(error) = reconcile_once(state, &keys, &token).await { - tracing::warn!(%error, "planner reconcile failed"); - metrics::counter!("planner_reconcile_errors_total").increment(1); + // Reconcile runs detached so a slow pass (13k tables, many + // pages) can never starve this loop's lease keepalive. At + // most one pass in flight; a tick that finds one running is + // skipped and counted. + if let Some(h) = &inflight { + if !h.is_finished() { + metrics::counter!("planner_reconcile_skipped_total").increment(1); + continue; + } } + let st = state.clone(); + let k = Keys::new(&state.config.etcd.etcd_prefix); + let t = token.clone(); + inflight = Some(tokio::spawn(async move { + if let Err(error) = reconcile_once(&st, &k, &t).await { + tracing::warn!(%error, "planner reconcile failed"); + metrics::counter!("planner_reconcile_errors_total").increment(1); + } + })); } } }; metrics::gauge!("planner_is_leader").set(0.0); + if let Some(h) = inflight.take() { + h.abort(); + } drop(keeper); let _ = client.lease_revoke(lease).await; tracing::info!(token = %token, ?result, "planner leadership released"); @@ -251,31 +335,27 @@ pub(crate) async fn reconcile_once( ) -> Result { let started = std::time::Instant::now(); let mut client = state.task_store.etcd_client().clone(); - let events = client - .get(keys.events_prefix(), Some(GetOptions::new().with_prefix())) - .await - .map_err(|e| e.to_string())?; - let header_revision = events.header().map_or(0, |h| h.revision()); + let (events, header_revision) = + read_prefix_paged(state.task_store.etcd_client(), &keys.events_prefix()).await?; let now = now_ms(); let (tables, skipped) = fold_events( keys, events - .kvs() .iter() - .map(|kv| (kv.key(), kv.value(), kv.mod_revision())), + .map(|kv| (kv.key.as_slice(), kv.value.as_slice(), kv.mod_revision)), now, ); + metrics::gauge!("planner_events").set(events.len() as f64); if skipped > 0 { metrics::counter!("planner_events_skipped_total").increment(skipped as u64); } - let existing = client - .get(keys.demand_prefix(), Some(GetOptions::new().with_prefix())) - .await - .map_err(|e| e.to_string())?; + let (existing, _) = + read_prefix_paged(state.task_store.etcd_client(), &keys.demand_prefix()).await?; + let existing_count = existing.len(); let mut existing_by_key: BTreeMap, TableDemand> = BTreeMap::new(); - for kv in existing.kvs() { - if let Ok(record) = serde_json::from_slice::(kv.value()) { - existing_by_key.insert(kv.key().to_vec(), record); + for kv in &existing { + if let Ok(record) = serde_json::from_slice::(&kv.value) { + existing_by_key.insert(kv.key.clone(), record); } } let mut ops = Vec::new(); @@ -364,7 +444,7 @@ pub(crate) async fn reconcile_once( tracing::warn!(%error, "planner could not load executor headroom"); } } - metrics::gauge!("planner_tables").set(existing.kvs().len() as f64); + metrics::gauge!("planner_tables").set(existing_count as f64); metrics::gauge!("planner_pending_generations_total").set(total_pending as f64); metrics::histogram!("planner_reconcile_seconds").record(started.elapsed().as_secs_f64()); metrics::gauge!("planner_last_revision").set(header_revision as f64); @@ -419,21 +499,18 @@ async fn shadow_pass( // Only Shadow records from previous passes are replaced; nothing else // under assignments/ is touched. let mut client = state.task_store.etcd_client().clone(); - let existing = client - .get( - executors_keys.assignments_prefix(), - Some(GetOptions::new().with_prefix()), - ) - .await - .map_err(|e| e.to_string())?; + let (existing, _) = read_prefix_paged( + state.task_store.etcd_client(), + &executors_keys.assignments_prefix(), + ) + .await?; let mut ops: Vec = existing - .kvs() .iter() .filter(|kv| { - serde_json::from_slice::(kv.value()) + serde_json::from_slice::(&kv.value) .is_ok_and(|a| a.state == AssignmentState::Shadow) }) - .map(|kv| TxnOp::delete(kv.key().to_vec(), None)) + .map(|kv| TxnOp::delete(kv.key.clone(), None)) .collect(); // Headroom already counts previous Shadow records; release them in the // arithmetic too, since they are about to be replaced. @@ -442,8 +519,8 @@ async fn shadow_pass( h.bytes_reserved = 0; h.assignments = 0; } - for kv in existing.kvs() { - if let Ok(a) = serde_json::from_slice::(kv.value()) { + for kv in &existing { + if let Ok(a) = serde_json::from_slice::(&kv.value) { if a.state != AssignmentState::Shadow && a.holds_capacity() { if let Some(h) = headroom.get_mut(&a.executor) { for (k, n) in &a.reserved_slots { @@ -462,11 +539,22 @@ async fn shadow_pass( if s.class == Class::Tail { continue; // the real sweeps do not touch tails either } - // Estimated cost: staging buffers are ~2x decoded bytes, capped at - // the per-pass byte limit the append path enforces. - let expected_bytes = (s.pending_bytes.saturating_mul(2)) - .min(state.config.append.rollout_append_max_bytes as u64 * 2) - .max(1); + // Estimated cost. A merge pass stages at most + // ROLLOUT_APPEND_MAX_BYTES of decoded data and the append path + // reserves ~2x that for buffers, so one unit costs at most + // 2 * max_bytes regardless of backlog. When pending bytes are + // unknown (writers do not measure; only the scan does) the estimate + // is that full cost, never a token: an unknown must reserve the + // worst case or the capacity check validates nothing. + let per_pass = state.config.append.rollout_append_max_bytes as u64 * 2; + let expected_bytes = if s.pending_bytes == 0 { + per_pass + } else { + s.pending_bytes.saturating_mul(2).min(per_pass).max(1) + }; + if s.pending_bytes == 0 { + metrics::counter!("planner_shadow_bytes_unknown_total").increment(1); + } let pick = headroom .values_mut() .filter(|h| h.fits(TaskKind::MergeWal, expected_bytes)) @@ -521,12 +609,37 @@ async fn shadow_pass( } } - // Disagreement: planner says place vs. reality has no task; or planner - // says nothing (no demand scored, or tail) vs. reality is running one. + // Comparison with the live system. These are **observations, not a + // takeover gate**: the shadow planner has only demand and capacity, + // while the real queue also reflects ownership, eligibility, cooldowns + // and executors this planner cannot see. Each metric names one precise + // question so the answer can be interpreted: + // + // discovery_miss planner scored demand on a table that has no + // MergeWal task queued or running at all - the real + // discovery loops have not noticed it yet (or it is a + // tail they ignore by design). + // placement_gap planner found no executor headroom for a due table + // that reality is in fact running - the planner's + // capacity model is tighter than the real pools. + // phantom_task reality is running/queued a MergeWal task on a table + // the planner sees no demand for - either a lost + // demand event, or work the planner would not have + // scheduled. + // + // The old single "disagreement" counter is kept for continuity but is + // the sum of these and should not be read on its own. let placed: std::collections::BTreeSet<&str> = would_place.iter().map(|(t, _)| t.as_str()).collect(); - let mut disagreements = 0u64; + let mut discovery_miss = 0u64; + let mut placement_gap = 0u64; + let mut agree = 0u64; + let mut scored_targets: std::collections::BTreeSet<&str> = std::collections::BTreeSet::new(); for s in &scored { + scored_targets.insert(s.target.as_str()); + if s.class == Class::Tail { + continue; + } let active = state .task_store .get_active_id(TaskKind::MergeWal, &s.target) @@ -534,15 +647,45 @@ async fn shadow_pass( .map_err(|e| e.to_string())? .is_some(); let planner_wants = placed.contains(s.target.as_str()); - if planner_wants != active { - disagreements += 1; - tracing::debug!( - target = %s.target, class = ?s.class, score = s.score, - planner_wants, reality_active = active, - "planner shadow disagreement" - ); + match (planner_wants, active) { + (true, true) | (false, false) if planner_wants => agree += 1, + (true, false) => { + discovery_miss += 1; + tracing::debug!(target = %s.target, class = ?s.class, score = s.score, + "shadow: planner would place, reality has no task"); + } + (false, true) => { + placement_gap += 1; + tracing::debug!(target = %s.target, class = ?s.class, score = s.score, + "shadow: due but unplaceable, reality is running it"); + } + _ => {} } } + // Phantom tasks: live MergeWal tasks on tables with no scored demand. + let mut phantom_task = 0u64; + if let Ok(running) = state + .task_store + .list_active_targets(TaskKind::MergeWal) + .await + { + for target in running { + if !scored_targets.contains(target.as_str()) { + phantom_task += 1; + tracing::debug!(target = %target, + "shadow: reality runs a MergeWal task with no scored demand"); + } + } + } + for (label, n) in [ + ("discovery_miss", discovery_miss), + ("placement_gap", placement_gap), + ("phantom_task", phantom_task), + ("agree", agree), + ] { + metrics::gauge!("planner_shadow_comparison", "outcome" => label).set(n as f64); + } + let disagreements = discovery_miss + placement_gap + phantom_task; metrics::counter!("scheduler_shadow_disagreements_total", "kind" => "merge_wal") .increment(disagreements); metrics::gauge!("planner_shadow_disagreements_last_pass").set(disagreements as f64); @@ -613,6 +756,168 @@ mod tests { assert_eq!(unhex("zz"), None); } + /// 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 + /// return every record at one revision, the fold must count them all, + /// and a reconcile must complete while the leader keeps renewing its + /// lease (the slow-reconcile-starves-keepalive finding). + /// Seeds ~27k keys and reads them back several times; on a shared + /// single-node etcd this starves concurrently running tests with + /// `request timed out`. Run alone: `cargo test -p lance-context-master + /// --lib reconcile_at_production_scale -- --ignored --test-threads=1`. + /// Gated on `PLANNER_SCALE_TEST=1` so the ordinary `--ignored` sweep in + /// CI does not include it. + #[tokio::test] + #[ignore = "requires ETCD_TEST_ENDPOINTS and PLANNER_SCALE_TEST=1; run alone"] + async fn reconcile_at_production_scale_pages_reads_and_keeps_the_lease() { + if std::env::var("PLANNER_SCALE_TEST").is_err() { + eprintln!("PLANNER_SCALE_TEST unset; skipping"); + return; + } + use crate::state::MasterState; + use clap::Parser; + const TABLES: usize = 13_405; + 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-scale/{}", generate_id()); + cfg.planner.planner_enabled = true; + cfg.planner.planner_reconcile_secs = 5; + cfg.etcd_lease_ttl_secs = 5; + cfg.stats_scan_interval_secs = 0; + cfg.compaction_interval_secs = 0; + cfg.merge_wal_interval_secs = 0; + let state = MasterState::new(cfg.clone()).await.unwrap(); + let keys = Keys::new(&cfg.etcd.etcd_prefix); + let mut client = state.task_store.etcd_client().clone(); + + // Seed events directly, in batched txns, with realistic bodies. + let mut ops = Vec::with_capacity(128); + for i in 0..TABLES { + let target = format!("exp-{i:05}"); + let pending = (i % 50) as u64; // 354-ish tables with real backlog is + // fewer than this; worst case is fine + for shard in [ + "aaaaaaaa-0000-0000-0000-000000000001", + "aaaaaaaa-0000-0000-0000-000000000002", + ] { + let event = DemandEvent { + v: SCHEMA_VERSION, + shard: shard.into(), + sealed_through: 1000 + pending, + sealed_bytes_through: 64 << 20, + merged_through: Some(1000), + flushed_at_ms: 1_700_000_000_000, + writer_epoch: 3, + source: EventSource::Scan, + merged_epoch: None, + sealed_times: (1001..=1000 + pending) + .map(|g| (g, 1_700_000_000_000 + g as i64)) + .collect(), + }; + ops.push(TxnOp::put( + format!("{}{}/{shard}", keys.events_prefix(), hex(&target)), + serde_json::to_vec(&event).unwrap(), + None, + )); + if ops.len() == 100 { + client + .txn(Txn::new().and_then(std::mem::take(&mut ops))) + .await + .unwrap(); + } + } + } + if !ops.is_empty() { + client.txn(Txn::new().and_then(ops)).await.unwrap(); + } + + // A single unpaged read of this prefix exceeds the client limit. + let single = client + .get(keys.events_prefix(), Some(GetOptions::new().with_prefix())) + .await; + assert!( + single.is_err(), + "expected the unpaged read to exceed the message cap; got {} kvs", + single.map(|r| r.kvs().len()).unwrap_or(0) + ); + + // The paged read returns everything at one revision. + let (kvs, revision) = read_prefix_paged(&client, &keys.events_prefix()) + .await + .unwrap(); + assert_eq!(kvs.len(), TABLES * 2); + assert!(revision > 0); + let (tables, skipped) = fold_events( + &keys, + kvs.iter() + .map(|kv| (kv.key.as_slice(), kv.value.as_slice(), kv.mod_revision)), + 0, + ); + assert_eq!(skipped, 0); + assert_eq!(tables.len(), TABLES); + let expected_pending: u64 = (0..TABLES as u64).map(|i| (i % 50) * 2).sum(); + assert_eq!( + tables + .values() + .map(|t| t.pending_generations()) + .sum::(), + expected_pending + ); + + // Lead for real: the reconcile runs detached, so the keepalive must + // keep firing and the leader key must survive a full pass. + let leader = tokio::spawn({ + let state = state.clone(); + async move { lead(&state).await } + }); + let deadline = std::time::Instant::now() + Duration::from_secs(120); + loop { + let (cache, _) = read_prefix_paged(&client, &keys.demand_prefix()) + .await + .unwrap(); + if cache.len() == TABLES { + break; + } + assert!( + std::time::Instant::now() < deadline, + "cache incomplete: {}", + cache.len() + ); + assert!( + !leader.is_finished(), + "leader loop ended during reconcile: {:?}", + leader.await + ); + tokio::time::sleep(Duration::from_millis(500)).await; + } + // Lease is alive well past several TTLs. + tokio::time::sleep(Duration::from_secs(12)).await; + let lk = client.get(keys.leader(), None).await.unwrap(); + assert_eq!( + lk.kvs().len(), + 1, + "leader lost its lease during/after the pass" + ); + assert!(!leader.is_finished()); + leader.abort(); + let _ = client + .delete( + cfg.etcd.etcd_prefix.as_str(), + Some(etcd_client::DeleteOptions::new().with_prefix()), + ) + .await; + } + #[tokio::test] #[ignore = "requires ETCD_TEST_ENDPOINTS"] async fn single_leader_builds_a_cache_equal_to_a_direct_fold_and_deposed_leader_cannot_write() { diff --git a/crates/lance-context-master/src/scanner.rs b/crates/lance-context-master/src/scanner.rs index bca92bbf..f950905e 100644 --- a/crates/lance-context-master/src/scanner.rs +++ b/crates/lance-context-master/src/scanner.rs @@ -347,10 +347,9 @@ async fn scan_once_inner(state: &Arc) -> lance::Result { // Demand reconciliation rides the scan: publish every // shard's watermark as a scan-sourced event. Detached // and bounded; the scan never waits on it. - let target = match entry.kind { - ScanKind::Rollout => entry.name.clone(), - ScanKind::Generic => format!("generic:{}", entry.name), - }; + // `entry.name` is already the task target: bare for + // rollout, `generic:` for generic stores. + let target = entry.name.clone(); crate::demand_publish::spawn_scan_demand( state.clone(), target, diff --git a/crates/lance-context-master/src/scoring.rs b/crates/lance-context-master/src/scoring.rs index c69a7bd8..1f876c81 100644 --- a/crates/lance-context-master/src/scoring.rs +++ b/crates/lance-context-master/src/scoring.rs @@ -86,7 +86,12 @@ pub fn score_merge( if pending == 0 { return None; } - let wait_ms = demand.oldest_pending_ms().map(|t| (now_ms - t).max(0)); + // Only an exact oldest-pending time may drive aging and promotion. A + // shard whose age is known only as an upper bound (merged past the kept + // flush times) contributes no wait; its count still scores. + let wait_ms = demand + .oldest_pending_exact_ms() + .map(|t| (now_ms - t).max(0)); let by_gens = pending as f64 / policy.min_generations.max(1) as f64; let by_bytes = if policy.min_bytes > 0 { bytes as f64 / policy.min_bytes as f64 @@ -236,6 +241,43 @@ mod tests { assert_eq!(v[0].target, "b", "class before score"); } + /// Round-5 finding 6: when the oldest pending time is only an upper + /// bound (merged past the kept entries), it must not promote for age + /// and must not inflate the score. + #[test] + fn an_age_bound_never_promotes() { + let p = MergePolicy { + max_turn_wait_secs: 10, + max_age_secs: 10, + ..Default::default() + }; + let now = 1_000_000_000; + let mut t = demand(128, now - 500_000); + let _ = t.fold( + &DemandEvent { + v: SCHEMA_VERSION, + shard: "s".into(), + sealed_through: 128, + sealed_bytes_through: 0, + merged_through: Some(64), + flushed_at_ms: 0, + writer_epoch: 1, + source: EventSource::Executor, + merged_epoch: None, + sealed_times: Default::default(), + }, + 999, + 0, + ); + assert_eq!(t.pending_generations(), 64); + assert!(t.oldest_pending_ms().is_some(), "a bound is shown"); + assert_eq!(t.oldest_pending_exact_ms(), None, "but is not exact"); + let s = score_merge("x", &t, &p, now).unwrap(); + assert_eq!(s.wait_ms, None); + assert_eq!(s.class, Class::Normal, "64/8 = 8 >= 1, not promoted"); + assert!((s.score - 8.0).abs() < 1e-9, "no age boost: {}", s.score); + } + #[test] fn ordering_is_total_and_stable() { let p = MergePolicy::default(); diff --git a/crates/lance-context-master/src/task_store.rs b/crates/lance-context-master/src/task_store.rs index 935b2ebd..fe1f9cd7 100644 --- a/crates/lance-context-master/src/task_store.rs +++ b/crates/lance-context-master/src/task_store.rs @@ -545,6 +545,27 @@ impl TaskStore { /// Whether the sweeps should skip this target for now because it has /// failed repeatedly. Manual enqueues are not gated by this. + /// Targets with a `kind` task queued or running right now, read in + /// pages. For the shadow planner's comparison against reality. + pub async fn list_active_targets(&self, kind: TaskKind) -> lance::Result> { + let mut out = Vec::new(); + for prefix in [self.inner.queue_prefix(), self.inner.running_prefix()] { + let (kvs, _) = crate::planner::read_prefix_paged(&self.inner.client, &prefix) + .await + .map_err(lance::Error::io)?; + for kv in kvs { + if let Ok(task) = decode_task(&kv.value, &String::from_utf8_lossy(&kv.key)) { + if task.kind == kind { + out.push(task.target); + } + } + } + } + out.sort(); + out.dedup(); + Ok(out) + } + /// Id of the queued or running task that `enqueue(kind, target)` would /// dedupe into, if any. Lets a sweep tell a new task from a no-op. pub async fn get_active_id( diff --git a/crates/lance-context-merge/src/demand.rs b/crates/lance-context-merge/src/demand.rs index 31f034b1..8e7a3c87 100644 --- a/crates/lance-context-merge/src/demand.rs +++ b/crates/lance-context-merge/src/demand.rs @@ -120,6 +120,15 @@ pub struct ShardDemand { pub sealed_times: BTreeMap, } +/// How much to trust an oldest-pending time. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum AgeBound { + Exact, + /// The true time is at or before this; never promote on it. + Upper, +} + impl ShardDemand { pub fn pending_generations(&self) -> u64 { self.sealed @@ -127,22 +136,56 @@ impl ShardDemand { .saturating_sub(self.merged.map_or(0, |m| m.value)) } - /// Flush time of the oldest generation still unmerged, if known. - pub fn oldest_pending_ms(&self) -> Option { + /// Flush time of the oldest generation still unmerged, with how much + /// to trust it. + /// + /// The record keeps flush times for at most `SEALED_TIMES_CAP` of the + /// lowest pending generations. While the oldest pending generation's + /// own time is kept the answer is **exact**. Once merging has consumed + /// every kept entry (a shard that had more than the cap pending), only + /// the newest sealed time is known; that is an **upper** bound on the + /// oldest pending flush time, so it must never drive age-based + /// promotion. Never `None` while something is pending. + pub fn oldest_pending(&self) -> Option<(i64, AgeBound)> { if self.pending_generations() == 0 { return None; } - self.sealed_times.values().next().copied() + let oldest = self.merged.map_or(0, |m| m.value) + 1; + if let Some(t) = self.sealed_times.get(&oldest) { + return Some((*t, AgeBound::Exact)); + } + // Entries at or below merged are stale and ignored. The lowest kept + // entry above `oldest` is an upper bound on its flush time; the + // newest sealed generation is always kept (see `prune_times`), so + // there is always one to give. + self.sealed_times + .range(oldest..) + .next() + .map(|(_, t)| (*t, AgeBound::Upper)) } - fn prune(&mut self) { - if let Some(m) = self.merged { - self.sealed_times.retain(|g, _| *g > m.value); - } - while self.sealed_times.len() > SEALED_TIMES_CAP { - self.sealed_times.pop_last(); + /// Convenience: the time only, whatever its bound. + pub fn oldest_pending_ms(&self) -> Option { + self.oldest_pending().map(|(t, _)| t) + } + + /// Exact oldest-pending time, or `None` when only a bound is known. + /// What the scheduler uses for promotion: a bound never promotes. + pub fn oldest_pending_exact_ms(&self) -> Option { + match self.oldest_pending() { + Some((t, AgeBound::Exact)) => Some(t), + _ => None, } } + + /// Bound the map by cap **only**. Dropping entries at or below `merged` + /// here would make the kept set depend on whether the merge or the + /// flushes arrived first (merge-first keeps 65..128, flush-first keeps + /// 1..63 and 128), which breaks order independence. Entries at or below + /// merged are ignored at read time instead. + fn prune(&mut self) { + prune_times(&mut self.sealed_times); + } } /// The per-table record under `P/demand/`. @@ -192,6 +235,23 @@ pub enum Ignored { UnknownVersion(u32), } +/// Bound a times map: the lowest `SEALED_TIMES_CAP - 1` generations plus +/// the single highest. Both parts are functions of the set, so the result +/// is order independent. Keeping the highest means an upper bound on every +/// pending age survives even when merging consumes all the low entries. +fn prune_times(times: &mut BTreeMap) { + if times.len() <= SEALED_TIMES_CAP { + return; + } + let newest = times.pop_last(); + while times.len() > SEALED_TIMES_CAP - 1 { + times.pop_last(); + } + if let Some((g, t)) = newest { + times.insert(g, t); + } +} + /// Insert `time` for `generation`, keeping the earliest. Returns whether the /// map changed. fn note_time(times: &mut BTreeMap, generation: u64, time: i64) -> bool { @@ -238,7 +298,8 @@ impl TableDemand { .sum() } - /// Flush time of the oldest generation still unmerged on any shard. + /// Flush time of the oldest generation still unmerged on any shard, + /// whatever its bound. For display. pub fn oldest_pending_ms(&self) -> Option { self.shards .values() @@ -246,6 +307,15 @@ impl TableDemand { .min() } + /// Oldest **exact** pending time across shards. What promotion uses; + /// a shard whose age is only an upper bound contributes nothing. + pub fn oldest_pending_exact_ms(&self) -> Option { + self.shards + .values() + .filter_map(ShardDemand::oldest_pending_exact_ms) + .min() + } + /// Fold one event. Returns `Err(reason)` when the record is unchanged. pub fn fold( &mut self, @@ -282,8 +352,11 @@ impl TableDemand { // Age bookkeeping: this event's own flush time plus any times it // carries from a stored record, min per generation, only for // generations still above merged. - let floor = shard.merged.map_or(0, |m| m.value); - if event.sealed_through > floor + // An executor's flushed_at_ms is its own observation time, not when + // the generation was flushed; only writer and scan events seed the + // age clock from their own timestamp. + if event.source != EventSource::Executor + && event.sealed_through > 0 && note_time( &mut shard.sealed_times, event.sealed_through, @@ -293,7 +366,7 @@ impl TableDemand { changed = true; } for (g, t) in &event.sealed_times { - if *g > floor && note_time(&mut shard.sealed_times, *g, *t) { + if note_time(&mut shard.sealed_times, *g, *t) { changed = true; } } @@ -326,18 +399,14 @@ pub fn join_events(stored: Option<&DemandEvent>, event: &DemandEvent) -> Option< if fresh.merged_epoch == Some(fresh.writer_epoch) { fresh.merged_epoch = None; } - let floor = fresh.merged_through.unwrap_or(0); - if fresh.sealed_through > floor { + if fresh.source != EventSource::Executor && fresh.sealed_through > 0 { note_time( &mut fresh.sealed_times, fresh.sealed_through, fresh.flushed_at_ms, ); } - fresh.sealed_times.retain(|g, _| *g > floor); - while fresh.sealed_times.len() > SEALED_TIMES_CAP { - fresh.sealed_times.pop_last(); - } + prune_times(&mut fresh.sealed_times); return Some(fresh); }; let s_sealed = Mark { @@ -369,23 +438,19 @@ pub fn join_events(stored: Option<&DemandEvent>, event: &DemandEvent) -> Option< }; // Times: union of everything known, min per generation. - let floor = best_merged.map_or(0, |m| m.value); let mut times = stored.sealed_times.clone(); for (g, t) in &event.sealed_times { note_time(&mut times, *g, *t); } - for (g, t) in [ - (stored.sealed_through, stored.flushed_at_ms), - (event.sealed_through, event.flushed_at_ms), + for (src, g, t) in [ + (stored.source, stored.sealed_through, stored.flushed_at_ms), + (event.source, event.sealed_through, event.flushed_at_ms), ] { - if g > floor { + if src != EventSource::Executor && g > 0 { note_time(&mut times, g, t); } } - times.retain(|g, _| *g > floor); - while times.len() > SEALED_TIMES_CAP { - times.pop_last(); - } + prune_times(&mut times); let times_changed = times != stored.sealed_times; if !sealed_wins && !merged_wins && !times_changed { @@ -401,9 +466,9 @@ pub fn join_events(stored: Option<&DemandEvent>, event: &DemandEvent) -> Option< next.merged_epoch = best_merged .map(|m| m.epoch) .filter(|e| *e != next.writer_epoch); - if merged_wins && !sealed_wins { - next.source = event.source; - } + // `source` and `flushed_at_ms` describe who set `sealed_through` and + // when it was flushed; they stay with the sealed mark so a rebuild + // re-derives exactly the time entries the fold did. next.sealed_times = times; next.v = SCHEMA_VERSION; Some(next) @@ -701,11 +766,21 @@ mod tests { Err(Ignored::NotNewer) ); // A "newer epoch" with a lower generation cannot occur in Lance; if - // reported it loses the max. + // reported it loses the max. Its flush time is recorded once (the + // map is bounded by cap only; see ShardDemand::prune) and is inert: + // generation 2 is below merged and never read. + table + .fold(&ev("a", 2, None, 4, EventSource::Scan), 5, 0) + .unwrap(); assert_eq!( - table.fold(&ev("a", 2, None, 4, EventSource::Scan), 5, 0), + table.fold(&ev("a", 2, None, 4, EventSource::Scan), 6, 0), Err(Ignored::NotNewer) ); + assert_eq!( + table.shards["a"].oldest_pending_ms(), + Some(150), + "gen 15's time, not gen 2's" + ); assert_eq!( table.shards["a"].sealed, Mark { @@ -776,12 +851,69 @@ mod tests { table.fold(&merged, 4, 0).unwrap(); assert_eq!(table.oldest_pending_ms(), Some(2000)); let stored = join_events(Some(&stored), &merged).unwrap(); - assert_eq!(stored.sealed_times.values().next(), Some(&2000)); + let mut from_stored = TableDemand::default(); + from_stored.fold(&stored, 0, 0).unwrap(); + assert_eq!(from_stored.oldest_pending_ms(), Some(2000)); let all = [flush(1, 1000), flush(2, 2000), flush(3, 3000), merged]; assert_eq!(rebuild(&all).oldest_pending_ms(), Some(2000)); assert_order_independent(&all); } + /// Round-5 finding 6: 128 pending, merge the first 64. The oldest + /// pending is generation 65, whose time the cap never kept. The answer + /// must not be unknown and must not drive promotion as if exact: it is + /// reported as an Upper bound (the newest kept or sealed time), excluded + /// from `oldest_pending_exact_ms`, and identical after a rebuild. + #[test] + fn age_after_merging_past_the_cap_is_a_flagged_bound() { + let flush = |g: u64| DemandEvent { + flushed_at_ms: g as i64 * 1000, + ..ev("s", g, None, 1, EventSource::Writer) + }; + let mut events: Vec<_> = (1..=128).map(flush).collect(); + let table = fold_all(&events); + assert_eq!( + table.shards["s"].oldest_pending(), + Some((1000, AgeBound::Exact)) + ); + assert_eq!(table.oldest_pending_exact_ms(), Some(1000)); + events.push(ev("s", 128, Some(64), 1, EventSource::Executor)); + let table = fold_all(&events); + let (t, bound) = table.shards["s"] + .oldest_pending() + .expect("64 pending: not unknown"); + assert_eq!(bound, AgeBound::Upper, "gen 65's time was never kept"); + assert!( + t >= 65_000, + "an upper bound is at or after the true time: {t}" + ); + assert_eq!( + table.oldest_pending_exact_ms(), + None, + "a bound never promotes" + ); + assert_eq!(table.oldest_pending_ms(), Some(t), "but is shown"); + assert_eq!( + rebuild(&events).shards["s"].oldest_pending(), + Some((t, AgeBound::Upper)) + ); + let mut reversed = events.clone(); + reversed.reverse(); + assert_eq!( + fold_all(&reversed).shards["s"].oldest_pending(), + Some((t, AgeBound::Upper)) + ); + // Below the cap the answer is exact again: 10 pending, merge 3 -> gen 4. + let small: Vec<_> = (1..=10) + .map(flush) + .chain([ev("s", 10, Some(3), 1, EventSource::Executor)]) + .collect(); + assert_eq!( + fold_all(&small).shards["s"].oldest_pending(), + Some((4000, AgeBound::Exact)) + ); + } + #[test] fn beyond_cap_counts_stay_exact_and_age_is_a_bound() { let n = (SEALED_TIMES_CAP as u64) + 20; diff --git a/docs/design/scheduler-p1-data-model.md b/docs/design/scheduler-p1-data-model.md index 43c19cf6..865cca5f 100644 --- a/docs/design/scheduler-p1-data-model.md +++ b/docs/design/scheduler-p1-data-model.md @@ -56,11 +56,17 @@ Rules: epoch a merged mark was reported under when it differs from the record's `writer_epoch`; provenance and tiebreak only. * `oldest_pending_ms` is the flush time of the lowest unmerged generation. The - stored record carries `sealed_times` (bounded to `SEALED_TIMES_CAP = 64` - lowest unmerged generations, min time per generation, pruned as `merged` - advances) so the age clock **survives a rebuild from etcd**. A newer flush - adds an entry and never resets the clock. Counts and bytes are exact in every - order; the age field is exact up to the cap and a lower bound beyond it. + stored record carries `sealed_times`: the lowest `SEALED_TIMES_CAP − 1 = 63` + recorded generations plus the newest, min time per generation, bounded by + cap **only** (pruning by `merged` would make the kept set depend on arrival + order). Entries at or below `merged` are ignored on read. The age clock + therefore **survives a rebuild from etcd** and a newer flush never resets + it. The answer carries a bound: **Exact** when the oldest pending + generation's own time is kept; **Upper** (the lowest kept time above it, + at worst the newest sealed) once merging has consumed every kept entry on a + shard that had more than the cap pending. Only an Exact age drives aging + and promotion; a bound is displayed but never promotes. Counts and bytes + are exact in every order. * Records carry `v`. A reader rejects an event it does not understand (`Ignored::UnknownVersion`) and a publisher **refuses to overwrite a stored record from a newer schema**; `v == 0` is read as version 1. @@ -88,13 +94,23 @@ source of truth, and may be rebuilt from events + scan at any time. Folding is a pure function `fold(table, event) -> table` and is tested for commutativity and idempotency (design invariant 5a). +### 1.2a Reads at scale + +Production has ~13,400 tables and ~27,000 shard records. A single prefix +read of `demand-events/` exceeds the etcd client's 4 MiB message cap. Every +planner prefix read (`demand-events/`, `demand/`, `executors/`, +`assignments/`, `queue/`, `running/`) is paged (512 keys) **at one +revision**, so each pass sees a consistent snapshot. The reconcile runs +detached from the leader loop so a slow pass can never starve the lease +keepalive; at most one pass is in flight and skipped ticks are counted. + ### 1.3 What writes what in P1 | Source | Writes | Change from today | |---|---|---| | Rollout server flush (`sweeper.rs::route_merge`) | `demand-events//` **in addition to** `merge-requests/` | additive | | Master append merge release (`rollout_append.rs`, `merge_execution.rs`) | `merged_through` per shard in the release txn | additive | -| Stats scanner (`scanner.rs`) | `demand-events` with `source: scan` for every shard it observes, plus `fragment_count`/`index_stale` into `demand/` | additive; existing enqueue/retire paths untouched | +| Stats scanner (`scanner.rs`) | `demand-events` with `source: scan` for every shard it observes, carrying **both** the shard manifest's `sealed_through` and the base table's merged watermark for that shard, so a lost completion event is corrected on the next scan and an old table's first contact does not read generation 1001 as 1001 pending | additive; existing enqueue/retire paths untouched | | Planner (new, leader only) | `demand/` | new | No existing key is removed or re-purposed in P1. @@ -164,10 +180,35 @@ leadership). The planner scores `MergeWal` demand only (design §4.2: classes Critical / Normal / Tail; class 1 oldest‑first), places against headroom, and writes assignments **only in shadow mode**: state `shadow`, replaced wholesale every pass, never read by any executor (a source‑level test enforces this), no -binding. Tails are not placed, matching the real sweeps. Each pass then -compares its placements with whether a real `MergeWal` task is active per -table and increments `scheduler_shadow_disagreements_total{kind}`; the -per‑pass count is `planner_shadow_disagreements_last_pass`. Compaction and +binding. Tails are not placed, matching the real sweeps. + +Each pass then compares itself with the live system. **These are +observations, not a takeover gate.** The shadow planner sees demand and +capacity; the real queue also reflects ownership, eligibility, cooldowns, +and executors the planner cannot see (legacy workers, catch‑up Jobs). The +comparison therefore answers three precise questions, each its own gauge +`planner_shadow_comparison{outcome}`: + +| outcome | meaning | +|---|---| +| `discovery_miss` | planner scored due demand on a table with no `MergeWal` task queued or running — the real discovery loops have not noticed it | +| `placement_gap` | planner found no headroom for a due table that reality is running — the planner's capacity model is tighter than the real pools | +| `phantom_task` | reality has a `MergeWal` task on a table the planner sees no demand for — a lost demand event, or work the planner would not schedule | +| `agree` | both would run it | + +`scheduler_shadow_disagreements_total{kind}` is the sum of the first three, +kept for continuity; read the components. What "approach zero" means per +component: `discovery_miss` falling to the tail population says the demand +feed is complete; `phantom_task` at zero says no demand events are lost; +`placement_gap` is expected to stay non‑zero until legacy workers and +catch‑up Jobs heartbeat as executors, and is not a defect of the planner. + +Byte estimates: writers do not measure generation sizes, so `pending_bytes` +is zero for most tables. An unknown reserves the **full per‑pass cost** +(`2 × ROLLOUT_APPEND_MAX_BYTES`), never a token — a capacity model that +reserves 1 byte validates nothing. `planner_shadow_bytes_unknown_total` +counts these. Measured sizes come with the scan reading generation data +statistics, a later change. Compaction and index scoring, commit‑ready class 2 and `max_consecutive_turns` come with P2's preparation integration; catch‑up Job heartbeats come when Jobs become an executor kind.