From be91a24e3fbc2056d1fe86eff01f3606bf113a6f Mon Sep 17 00:00:00 2001 From: Beinan Date: Thu, 8 Oct 2026 21:25:12 +0000 Subject: [PATCH] feat(master): stats scan publishes scan-sourced demand for every shard MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P1.4 (docs/design/scheduler-p1-data-model.md §1.3). The scan is the reconciliation source for demand: it covers shards whose writer never ran the server flush path (retired writers, pre-upgrade shards) and corrects any writer event that was dropped. - core: all_shard_watermarks() reads every shard's latest manifest for (shard_id, writer_epoch, max flushed generation, pending count). Unlike wal_shard_snapshots it does not enforce the read cap: the scheduler must see an over-cap shard, not be refused by it. - master: demand_publish::spawn_scan_demand runs after each successful observation, detached and bounded (15 s timeout, 4 slots, drop and count), publishing one scan-sourced event per shard through the same per-dimension join as writers and executors, so a scan can only raise a watermark and never lowers a merged mark it does not know about. Metric master_scan_demand_events_total{result}. The scan's existing enqueue and retirement behaviour is untouched. Validation: etcd test with two writer shards and no server flush path: one scan event per shard, fold gives six pending; after an executor reports merged=2 on one shard, a second scan leaves merged intact and does not rewrite the record. 15 scanner unit tests pass; clippy -D warnings, fmt, typos clean. Refs #333. Stacked on #352. Co-Authored-By: Claude Fable 5 --- .../lance-context-core/src/generic_store.rs | 7 ++ .../lance-context-core/src/rollout_store.rs | 7 ++ crates/lance-context-core/src/store_base.rs | 43 +++++++ .../src/demand_publish.rs | 86 +++++++++++++- crates/lance-context-master/src/scanner.rs | 27 ++++- crates/lance-context-master/src/scheduler.rs | 108 ++++++++++++++++++ 6 files changed, 272 insertions(+), 6 deletions(-) diff --git a/crates/lance-context-core/src/generic_store.rs b/crates/lance-context-core/src/generic_store.rs index 3cbc3ccc..8798d075 100644 --- a/crates/lance-context-core/src/generic_store.rs +++ b/crates/lance-context-core/src/generic_store.rs @@ -641,6 +641,13 @@ impl GenericStore { self.base.own_shard_watermark().await } + /// Every shard's watermark from its latest manifest. Metadata only. + pub async fn all_shard_watermarks( + &self, + ) -> LanceResult> { + self.base.all_shard_watermarks().await + } + /// The shared-lock half of [`Self::cleanup_wal`]: seal, then read a /// budgeted prefix of flushed generations into memory. Callers holding a /// read lock run this while appends continue, then take the write lock diff --git a/crates/lance-context-core/src/rollout_store.rs b/crates/lance-context-core/src/rollout_store.rs index 4ef8893a..5564b3c2 100644 --- a/crates/lance-context-core/src/rollout_store.rs +++ b/crates/lance-context-core/src/rollout_store.rs @@ -673,6 +673,13 @@ impl RolloutStore { self.base.own_shard_watermark().await } + /// Every shard's watermark from its latest manifest. Metadata only. + pub async fn all_shard_watermarks( + &self, + ) -> LanceResult> { + self.base.all_shard_watermarks().await + } + /// [`Self::prepare_merge_if_ready`], but seals the active memtable first — /// the time-triggered behavior of [`Self::cleanup_own_shard`]. pub async fn prepare_cleanup_merge( diff --git a/crates/lance-context-core/src/store_base.rs b/crates/lance-context-core/src/store_base.rs index 3d71bef6..c7dfb9c2 100644 --- a/crates/lance-context-core/src/store_base.rs +++ b/crates/lance-context-core/src/store_base.rs @@ -2083,6 +2083,49 @@ impl StorageBase { builder.load().await } + /// Every shard's watermark (highest flushed generation and writer epoch) + /// from its latest manifest. Metadata only; bounded-concurrent like + /// [`Self::wal_shard_snapshots`], but does not enforce the read cap: + /// the scheduler must see an over-cap shard, not be refused by it. + pub async fn all_shard_watermarks(&self) -> LanceResult> { + if self.is_version_pinned() { + return Ok(Vec::new()); + } + 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?; + let marks: Vec> = stream::iter(shard_ids) + .map(|shard_id| { + let object_store = object_store.clone(); + let branch_path = branch_path.clone(); + async move { + let manifest_store = ShardManifestStore::new( + object_store, + &branch_path, + shard_id, + DEFAULT_MANIFEST_SCAN_BATCH_SIZE, + ); + Ok::<_, LanceError>(manifest_store.read_latest().await?.map(|manifest| { + ShardWatermark { + shard_id: manifest.shard_id, + writer_epoch: manifest.writer_epoch, + sealed_through: manifest + .flushed_generations + .iter() + .map(|g| g.generation) + .max() + .unwrap_or(0), + pending_generations: manifest.flushed_generations.len(), + } + })) + } + }) + .buffer_unordered(8) + .try_collect() + .await?; + Ok(marks.into_iter().flatten().collect()) + } + /// Read the latest manifest for every MemWAL shard. Manifest reads are /// bounded-concurrent so stores with many writer instances do not pay one /// object-store round trip per shard serially. diff --git a/crates/lance-context-master/src/demand_publish.rs b/crates/lance-context-master/src/demand_publish.rs index 65911037..5377cef3 100644 --- a/crates/lance-context-master/src/demand_publish.rs +++ b/crates/lance-context-master/src/demand_publish.rs @@ -1,5 +1,7 @@ //! Detached, bounded publication of demand watermarks from master-side -//! sources (`docs/design/scheduler-p1-data-model.md` §1.3). +//! sources (`docs/design/scheduler-p1-data-model.md` §1.3): the stats scan +//! (source `scan`) and executors after a successful WAL catch-up release +//! (source `executor`). //! //! Same discipline as the writer-side publish: never awaited by the caller, //! one hard timeout, a process-wide concurrency cap, drop-and-count when the @@ -9,10 +11,90 @@ use std::sync::Arc; use std::time::Duration; -use lance_context_merge::demand::ShardMerged; +use lance_context_core::{GenericStore, GenericStoreOptions, RolloutStore, RolloutStoreOptions}; +use lance_context_merge::demand::{DemandEvent, EventSource, ShardMerged, SCHEMA_VERSION}; use lance_context_merge::Coordinator; use tokio::sync::Semaphore; +use crate::scanner::ScanKind; +use crate::state::MasterState; + +/// Manifest reads for every shard plus one etcd put per shard. +const SCAN_PUBLISH_TIMEOUT: Duration = Duration::from_secs(15); + +static SCAN_PUBLISH_SLOTS: std::sync::LazyLock> = + std::sync::LazyLock::new(|| Arc::new(Semaphore::new(4))); + +pub(crate) fn spawn_scan_demand( + state: Arc, + target: String, + uri: String, + kind: ScanKind, + rollout_options: RolloutStoreOptions, + generic_options: GenericStoreOptions, +) { + let Ok(permit) = SCAN_PUBLISH_SLOTS.clone().try_acquire_owned() else { + metrics::counter!("master_scan_demand_events_total", "result" => "dropped").increment(1); + return; + }; + tokio::spawn(async move { + let _permit = permit; + let result = tokio::time::timeout(SCAN_PUBLISH_TIMEOUT, async { + let marks = match kind { + ScanKind::Rollout => { + RolloutStore::open_existing_with_options(&uri, rollout_options) + .await + .map_err(|e| e.to_string())? + .all_shard_watermarks() + .await + .map_err(|e| e.to_string())? + } + ScanKind::Generic => GenericStore::open_existing(&uri, generic_options) + .await + .map_err(|e| e.to_string())? + .all_shard_watermarks() + .await + .map_err(|e| e.to_string())?, + }; + let coordinator = state.task_store.merge_coordinator(); + let now_ms = chrono::Utc::now().timestamp_millis(); + let mut written = 0usize; + for mark in marks { + let event = DemandEvent { + v: SCHEMA_VERSION, + shard: mark.shard_id.to_string(), + sealed_through: mark.sealed_through, + sealed_bytes_through: 0, + merged_through: None, + flushed_at_ms: now_ms, + writer_epoch: mark.writer_epoch, + source: EventSource::Scan, + merged_epoch: None, + sealed_times: Default::default(), + }; + if coordinator.publish_demand_event(&target, &event).await? { + written += 1; + } + } + Ok::(written) + }) + .await; + let label = match result { + Ok(Ok(n)) if n > 0 => "written", + Ok(Ok(_)) => "unchanged", + Ok(Err(error)) => { + tracing::debug!(target = %target, %error, "scan demand publish failed"); + "failed" + } + Err(_) => { + tracing::debug!(target = %target, "scan demand publish timed out"); + "timeout" + } + }; + metrics::counter!("master_scan_demand_events_total", "result" => label).increment(1); + }); +} + /// Base-table index read for the merged watermarks plus one etcd put per shard. const MERGED_PUBLISH_TIMEOUT: Duration = Duration::from_secs(15); diff --git a/crates/lance-context-master/src/scanner.rs b/crates/lance-context-master/src/scanner.rs index d1d5b5f0..bca92bbf 100644 --- a/crates/lance-context-master/src/scanner.rs +++ b/crates/lance-context-master/src/scanner.rs @@ -330,18 +330,37 @@ async fn scan_once_inner(state: &Arc) -> lance::Result { let previous = previous.clone(); let rollout_options = rollout_options.clone(); let generic_options = generic_options.clone(); + let state = state.clone(); async move { let prev = previous.get(&entry.name); let observed = match entry.kind { ScanKind::Rollout => { - observe_one(&entry.name, &entry.uri, prev, rollout_options).await + observe_one(&entry.name, &entry.uri, prev, rollout_options.clone()).await } ScanKind::Generic => { - observe_generic(&entry.name, &entry.uri, prev, generic_options).await + observe_generic(&entry.name, &entry.uri, prev, generic_options.clone()) + .await } }; match observed { - Ok(result) => (entry.name, Some(result)), + Ok(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), + }; + crate::demand_publish::spawn_scan_demand( + state.clone(), + target, + entry.uri.clone(), + entry.kind, + rollout_options, + generic_options, + ); + (entry.name, Some(result)) + } Err(e) => { tracing::warn!(store = %entry.name, error = %e, "scan: observe failed"); (entry.name, None) @@ -517,7 +536,7 @@ async fn scan_once_inner(state: &Arc) -> lance::Result { /// that finishes inside its interval and one that never does. /// Which registry a scan entry came from, which decides how it is opened. #[derive(Debug, Clone, Copy)] -enum ScanKind { +pub(crate) enum ScanKind { Rollout, Generic, } diff --git a/crates/lance-context-master/src/scheduler.rs b/crates/lance-context-master/src/scheduler.rs index 6d9b1a05..d6d75ba6 100644 --- a/crates/lance-context-master/src/scheduler.rs +++ b/crates/lance-context-master/src/scheduler.rs @@ -1631,6 +1631,114 @@ mod tests { } } + /// The stats scan publishes a scan-sourced demand event for every shard + /// it observes, including shards whose writer never ran the server + /// flush path, and a later scan cannot lower a watermark a writer or + /// executor set at the same epoch. + #[tokio::test] + #[ignore = "requires ETCD_TEST_ENDPOINTS"] + async fn scan_publishes_every_shard_and_never_lowers_a_watermark() { + use lance_context_core::RolloutStoreOptions; + use lance_context_merge::demand::{DemandEvent, EventSource, TableDemand}; + let dir = TempDir::new().unwrap(); + let cfg = config(&dir); + let state = MasterState::new(cfg).await.unwrap(); + let uri = state.rollout_uri("exp"); + for shard in ["a", "b"] { + let writer = RolloutStore::open_with_options( + &uri, + RolloutStoreOptions { + shard_id: Some(shard.into()), + merge_after_generations: Some(0), + ..Default::default() + }, + ) + .await + .unwrap(); + for i in 0..3 { + writer + .add(&[rollout_record(&format!("{shard}-{i}"))]) + .await + .unwrap(); + writer.flush().await.unwrap(); + } + } + state.registry.upsert("exp", &uri).await.unwrap(); + crate::scanner::scan_once(&state).await.unwrap(); + + let prefix = state + .config + .etcd + .etcd_prefix + .trim_end_matches('/') + .to_string(); + let encoded: String = "exp".bytes().map(|b| format!("{b:02x}")).collect(); + let events_prefix = format!("{prefix}/demand-events/{encoded}/"); + let client = state.task_store.etcd_client().clone(); + let deadline = std::time::Instant::now() + Duration::from_secs(10); + let events = loop { + let kvs = client + .clone() + .get( + events_prefix.as_str(), + Some(etcd_client::GetOptions::new().with_prefix()), + ) + .await + .unwrap(); + if kvs.kvs().len() >= 2 { + break kvs; + } + assert!( + std::time::Instant::now() < deadline, + "scan demand never landed" + ); + tokio::time::sleep(Duration::from_millis(50)).await; + }; + assert_eq!(events.kvs().len(), 2, "one scan event per shard"); + let mut table = TableDemand::default(); + for kv in events.kvs() { + let event: DemandEvent = serde_json::from_slice(kv.value()).unwrap(); + assert_eq!(event.source, EventSource::Scan); + assert!(event.sealed_through >= 3, "{event:?}"); + table.fold(&event, kv.mod_revision(), 0).unwrap(); + } + assert_eq!(table.pending_generations(), 6); + + // An executor reports merged=2 on shard "a" at the same epoch. A + // second scan (which knows nothing about merged) must not lower it. + let coordinator = state.task_store.merge_coordinator(); + let (key, mut a): (String, DemandEvent) = events + .kvs() + .iter() + .map(|kv| { + ( + String::from_utf8_lossy(kv.key()).to_string(), + serde_json::from_slice::(kv.value()).unwrap(), + ) + }) + .next() + .unwrap(); + a.merged_through = Some(2); + a.source = EventSource::Executor; + assert!(coordinator.publish_demand_event("exp", &a).await.unwrap()); + let before = client.clone().get(key.as_str(), None).await.unwrap().kvs()[0].mod_revision(); + crate::scanner::scan_once(&state).await.unwrap(); + tokio::time::sleep(Duration::from_millis(500)).await; + let after = client.clone().get(key.as_str(), None).await.unwrap(); + let stored: DemandEvent = serde_json::from_slice(after.kvs()[0].value()).unwrap(); + assert_eq!( + stored.merged_through, + Some(2), + "scan lowered merged: {stored:?}" + ); + assert_eq!(stored.sealed_through, a.sealed_through); + assert_eq!( + after.kvs()[0].mod_revision(), + before, + "an unchanged scan must not rewrite the record" + ); + } + #[tokio::test] #[ignore = "requires ETCD_TEST_ENDPOINTS"] async fn compact_noop_survives_restart_and_changes_invalidate_it() {