feat(master): control-plane observability crate (stats scan + list API)

Cargo.toml

@@ -4,6 +4,7 @@ members = [
     "crates/lance-context",
     "crates/lance-context-api",
     "crates/lance-context-server",
+    "crates/lance-context-master",
     "crates/lance-context-client",
     "python",
 ]

crates/lance-context-api/src/lib.rs

@@ -833,6 +833,71 @@ where
     }
 }
 
+// ---------------------------------------------------------------------------
+// Master control-plane DTOs
+// ---------------------------------------------------------------------------
+
+/// One row of the control-plane experiment listing, backed by the master's
+/// periodically-refreshed stats table.
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct ExperimentSummary {
+    /// Logical experiment / rollout-store name (unique key).
+    pub name: String,
+    /// Physical dataset URI.
+    pub uri: String,
+    /// Base-table logical row count as of the last scan.
+    pub row_count: i64,
+    /// Base-table fragment count as of the last scan.
+    pub fragment_count: i64,
+    /// Base-table manifest timestamp, Unix milliseconds.
+    pub last_updated: i64,
+    /// Flushed MemWAL generations pending merge as of the last scan.
+    pub pending_wal_generations: i64,
+    /// Timestamp of the last successful compaction, Unix ms, if any.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub last_compaction: Option<i64>,
+    /// Total compactions the master has driven for this experiment.
+    pub total_compactions: i64,
+    /// When the master last scanned this experiment, Unix milliseconds.
+    pub scanned_at: i64,
+}
+
+/// Detailed view of a single experiment. Currently the same shape as
+/// [`ExperimentSummary`]; a distinct type leaves room for detail-only fields.
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct ExperimentDetail {
+    #[serde(flatten)]
+    pub summary: ExperimentSummary,
+}
+
+/// Paginated experiment listing.
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct ExperimentListResponse {
+    pub experiments: Vec<ExperimentSummary>,
+    /// Total number of experiments matching the (optional) search filter,
+    /// ignoring pagination.
+    pub total: i64,
+}
+
+/// State of a manual or automatic compaction job for one experiment.
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "snake_case", tag = "state")]
+pub enum CompactJobStatus {
+    /// Queued but not yet started.
+    Queued,
+    /// Currently running.
+    Running,
+    /// Finished successfully.
+    Done {
+        fragments_removed: usize,
+        fragments_added: usize,
+    },
+    /// Failed with an error message.
+    Failed { error: String },
+    /// No compaction has been requested for this experiment.
+    None,
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

crates/lance-context-core/src/lib.rs

@@ -34,7 +34,7 @@ pub use record::{
 };
 pub use registry::{RegistryEntry, RolloutRegistry};
 pub use rollout::{RolloutRecord, ROLE_ARTIFACT, ROLE_ASSISTANT, ROLE_GRADE, ROLE_TOOL};
-pub use rollout_store::{rollout_schema, RolloutStore, RolloutStoreOptions};
+pub use rollout_store::{rollout_schema, RolloutObservation, RolloutStore, RolloutStoreOptions};
 pub use store::{
     CompactionConfig, CompactionStats, ContextStore, ContextStoreOptions, DistanceMetric,
     IdIndexType, ReadProjection,

crates/lance-context-core/src/rollout_store.rs

@@ -85,6 +85,24 @@ use crate::store::{
 /// shard state (mirrors the constant used by `ContextStore`).
 const DEFAULT_MANIFEST_SCAN_BATCH_SIZE: usize = 16;
 
+/// Cheap, read-only observability snapshot of a rollout store's base table.
+///
+/// Produced by [`RolloutStore::observe`] from dataset metadata (no full scan).
+/// Consumed by the control-plane stats scanner.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct RolloutObservation {
+    /// Logical row count of the base table (fragment metadata, cheap).
+    pub row_count: i64,
+    /// Number of fragments in the base table.
+    pub fragment_count: i64,
+    /// Current base dataset manifest version.
+    pub version: u64,
+    /// Manifest timestamp, Unix milliseconds — when the base table last changed.
+    pub last_updated: i64,
+    /// Flushed MemWAL generations pending merge for this instance's shard.
+    pub pending_wal_generations: i64,
+}
+
 /// Configuration for opening a [`RolloutStore`].
 #[derive(Debug, Clone, Default)]
 pub struct RolloutStoreOptions {
@@ -632,6 +650,48 @@ impl RolloutStore {
         true
     }
 
+    /// Number of flushed MemWAL generations pending merge into the base table
+    /// for this instance's own shard. `0` when the shard has no manifest yet.
+    ///
+    /// Read-only: unlike [`Self::cleanup_own_shard`] it never merges. Used by the
+    /// control-plane stats scanner to surface read-amplification pressure.
+    pub async fn pending_wal_generations(&self) -> LanceResult<usize> {
+        let object_store = self.dataset.object_store(None).await?;
+        let branch_location = self.dataset.branch_location();
+        let manifest_store = ShardManifestStore::new(
+            object_store,
+            &branch_location.path,
+            self.write_shard,
+            DEFAULT_MANIFEST_SCAN_BATCH_SIZE,
+        );
+        match manifest_store.read_latest().await? {
+            Some(manifest) => Ok(manifest.flushed_generations.len()),
+            None => Ok(0),
+        }
+    }
+
+    /// Snapshot cheap, read-only observability metrics for the base table.
+    ///
+    /// All fields come from dataset metadata (manifest + fragment counts), not a
+    /// full table scan: `row_count` and `fragment_count` read fragment metadata,
+    /// `version`/`last_updated` come from the manifest. `pending_wal_generations`
+    /// reads this instance's shard manifest. Intended for the control-plane stats
+    /// scanner, which opens each store read-only and records the result.
+    pub async fn observe(&self) -> LanceResult<RolloutObservation> {
+        let row_count = self.dataset.count_rows(None).await? as i64;
+        let fragment_count = self.dataset.count_fragments() as i64;
+        let version = self.dataset.manifest.version;
+        let last_updated = self.dataset.manifest.timestamp().timestamp_millis();
+        let pending_wal_generations = self.pending_wal_generations().await? as i64;
+        Ok(RolloutObservation {
+            row_count,
+            fragment_count,
+            version,
+            last_updated,
+            pending_wal_generations,
+        })
+    }
+
     /// Current compaction statistics for the base table.
     ///
     /// `is_compacting` is always `false`: compaction here runs synchronously
@@ -1628,6 +1688,33 @@ mod tests {
         });
     }
 
+    #[test]
+    fn observe_reports_cheap_metrics() {
+        let dir = TempDir::new().unwrap();
+        let uri = dir.path().to_string_lossy().to_string();
+        let runtime = tokio::runtime::Runtime::new().unwrap();
+        runtime.block_on(async {
+            let mut store = RolloutStore::open(&uri).await.unwrap();
+            // Empty store: no rows, no pending generations.
+            let obs = store.observe().await.unwrap();
+            assert_eq!(obs.row_count, 0);
+            assert_eq!(obs.pending_wal_generations, 0);
+
+            store
+                .add(&[assistant_record("a"), assistant_record("b")])
+                .await
+                .unwrap();
+
+            // Two MemWAL-appended rows are visible via the LSM read path; the
+            // base table itself has not been merged, so they surface as a
+            // pending flushed generation rather than base rows.
+            let obs = store.observe().await.unwrap();
+            assert!(obs.fragment_count >= 0);
+            assert!(obs.pending_wal_generations >= 1);
+            assert!(obs.last_updated > 0);
+        });
+    }
+
     #[test]
     fn appends_accumulate_and_are_immutable() {
         // MemWAL appends are append-only: successive `add`s accumulate, and a

crates/lance-context-master/Cargo.toml

@@ -0,0 +1,33 @@
+[package]
+name = "lance-context-master"
+version = "0.6.3"
+edition = "2021"
+license = "Apache-2.0"
+authors = ["Lance Devs <dev@lancedb.com>"]
+repository = "https://github.com/lancedb/lance-context"
+description = "Control-plane (master) process for lance-context rollout stores"
+keywords = ["context", "lance", "control-plane", "compaction"]
+
+[[bin]]
+name = "lance-context-master"
+path = "src/main.rs"
+
+[dependencies]
+lance-context-core = { version = "0.6.3", path = "../lance-context-core" }
+lance-context-api = { version = "0.6.3", path = "../lance-context-api" }
+arrow-array = "58"
+arrow-schema = "58"
+axum = { version = "0.8", features = ["json"] }
+chrono = { version = "0.4", default-features = false, features = ["clock"] }
+clap = { version = "4", features = ["derive", "env"] }
+futures = "0.3"
+lance = "7.0.0"
+serde = { version = "1", features = ["derive"] }
+serde_json = "1"
+tokio = { version = "1", features = ["rt-multi-thread", "macros", "signal", "sync"] }
+tower-http = { version = "0.6", features = ["cors", "fs", "trace"] }
+tracing = "0.1"
+tracing-subscriber = { version = "0.3", features = ["env-filter"] }
+
+[dev-dependencies]
+tempfile = "3"

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

@@ -0,0 +1,47 @@
+//! Command-line / environment configuration for the master control-plane.
+
+use clap::Parser;
+
+/// Control-plane (master) process for lance-context rollout stores.
+#[derive(Debug, Clone, Parser)]
+#[command(name = "lance-context-master", version)]
+pub struct MasterConfig {
+    /// Data directory / object-store prefix shared with the data-plane server.
+    #[arg(long, env = "DATA_DIR", default_value = "./data")]
+    pub data_dir: String,
+
+    /// Host to bind the admin API + UI to.
+    #[arg(long, env = "MASTER_HOST", default_value = "0.0.0.0")]
+    pub host: String,
+
+    /// Port to bind the admin API + UI to.
+    #[arg(long, env = "MASTER_PORT", default_value_t = 8090)]
+    pub port: u16,
+
+    /// Interval, in seconds, between full stats scans of every experiment.
+    /// `0` disables the background scanner.
+    #[arg(long, env = "STATS_SCAN_INTERVAL_SECS", default_value_t = 300)]
+    pub stats_scan_interval_secs: u64,
+
+    /// Bounded concurrency when scanning experiments each round.
+    #[arg(long, env = "SCAN_CONCURRENCY", default_value_t = 8)]
+    pub scan_concurrency: usize,
+
+    /// Interval, in seconds, between automatic compaction sweeps. `0` disables
+    /// automatic compaction (manual triggers still work).
+    #[arg(long, env = "COMPACTION_INTERVAL_SECS", default_value_t = 600)]
+    pub compaction_interval_secs: u64,
+
+    /// Minimum fragment count before an experiment is auto-compacted.
+    #[arg(long, env = "MIN_FRAGMENTS", default_value_t = 16)]
+    pub min_fragments: usize,
+
+    /// Target rows per fragment passed to compaction.
+    #[arg(long, env = "TARGET_ROWS_PER_FRAGMENT", default_value_t = 1_048_576)]
+    pub target_rows_per_fragment: usize,
+
+    /// Directory of built UI assets to serve. When unset, only the JSON API is
+    /// exposed.
+    #[arg(long, env = "UI_DIR")]
+    pub ui_dir: Option<String>,
+}

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

@@ -0,0 +1,30 @@
+//! Admin HTTP error type.
+
+use axum::http::StatusCode;
+use axum::response::{IntoResponse, Response};
+use axum::Json;
+use serde_json::json;
+
+/// Errors surfaced by the master admin API.
+#[derive(Debug)]
+pub enum MasterError {
+    NotFound(String),
+    Internal(String),
+}
+
+impl MasterError {
+    /// Map a Lance error to an internal server error.
+    pub fn from_lance(err: lance::Error) -> Self {
+        MasterError::Internal(err.to_string())
+    }
+}
+
+impl IntoResponse for MasterError {
+    fn into_response(self) -> Response {
+        let (status, msg) = match self {
+            MasterError::NotFound(m) => (StatusCode::NOT_FOUND, m),
+            MasterError::Internal(m) => (StatusCode::INTERNAL_SERVER_ERROR, m),
+        };
+        (status, Json(json!({ "error": msg }))).into_response()
+    }
+}

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

@@ -0,0 +1,8 @@
+//! Control-plane (master) library surface.
+
+pub mod config;
+pub mod error;
+pub mod routes;
+pub mod scanner;
+pub mod state;
+pub mod stats_store;