feat(master): stats scan publishes scan-sourced demand for every shard

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<Vec<crate::store_base::ShardWatermark>> {
+        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

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<Vec<crate::store_base::ShardWatermark>> {
+        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(

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<Vec<ShardWatermark>> {
+        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<Option<ShardWatermark>> = 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.

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<Arc<Semaphore>> =
+    std::sync::LazyLock::new(|| Arc::new(Semaphore::new(4)));
+
+pub(crate) fn spawn_scan_demand(
+    state: Arc<MasterState>,
+    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::<usize, String>(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);
 

crates/lance-context-master/src/scanner.rs

@@ -330,18 +330,37 @@ async fn scan_once_inner(state: &Arc<MasterState>) -> lance::Result<usize> {
             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<MasterState>) -> lance::Result<usize> {
 /// 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,
 }

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::<DemandEvent>(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() {