feat(master): centralized serial compaction scheduler + compact routes

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

@@ -4,5 +4,6 @@ pub mod config;
 pub mod error;
 pub mod routes;
 pub mod scanner;
+pub mod scheduler;
 pub mod state;
 pub mod stats_store;

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

@@ -9,6 +9,7 @@ use tracing_subscriber::EnvFilter;
 use lance_context_master::config::MasterConfig;
 use lance_context_master::routes;
 use lance_context_master::scanner;
+use lance_context_master::scheduler;
 use lance_context_master::state::MasterState;
 
 #[tokio::main]
@@ -42,6 +43,9 @@ async fn main() {
     // Background stats scanner (detached; stops when the last Arc drops).
     let _scanner = scanner::spawn_scanner(&state);
 
+    // Single serial compaction worker (+ optional periodic auto-sweep).
+    let _scheduler = scheduler::spawn_scheduler(&state);
+
     let mut app = Router::new().nest("/api/v1", routes::api_router());
 
     // Serve the built UI (SPA) as a fallback when a `--ui-dir` is configured.

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

@@ -3,14 +3,18 @@
 use std::sync::Arc;
 
 use axum::extract::{Path, Query, State};
+use axum::http::StatusCode;
 use axum::routing::{get, post};
 use axum::{Json, Router};
 use serde::Deserialize;
 
-use lance_context_api::{ExperimentDetail, ExperimentListResponse, ExperimentSummary};
+use lance_context_api::{
+    CompactJobStatus, ExperimentDetail, ExperimentListResponse, ExperimentSummary,
+};
 
 use crate::error::MasterError;
 use crate::scanner;
+use crate::scheduler;
 use crate::state::MasterState;
 
 /// Query params for the experiment listing.
@@ -101,11 +105,53 @@ pub async fn rescan(
     Ok(Json(serde_json::json!({ "scanned": n })))
 }
 
+/// `POST /api/v1/experiments/{name}/compact` — enqueue a manual compaction.
+/// Returns 202 Accepted with the (possibly de-duped) job status.
+pub async fn compact_experiment(
+    State(state): State<Arc<MasterState>>,
+    Path(name): Path<String>,
+) -> Result<(StatusCode, Json<CompactJobStatus>), MasterError> {
+    // Only enqueue known experiments.
+    let exists = state
+        .registry
+        .read()
+        .await
+        .contains(&name)
+        .await
+        .map_err(MasterError::from_lance)?;
+    if !exists {
+        return Err(MasterError::NotFound(format!(
+            "experiment '{}' does not exist",
+            name
+        )));
+    }
+    let status = scheduler::enqueue(&state, &name).await;
+    Ok((StatusCode::ACCEPTED, Json(status)))
+}
+
+/// `GET /api/v1/experiments/{name}/compact/status` — latest compaction job
+/// state for an experiment (`None` when never requested).
+pub async fn compact_status(
+    State(state): State<Arc<MasterState>>,
+    Path(name): Path<String>,
+) -> Json<CompactJobStatus> {
+    let status = state
+        .jobs
+        .lock()
+        .await
+        .get(&name)
+        .cloned()
+        .unwrap_or(CompactJobStatus::None);
+    Json(status)
+}
+
 /// Build the admin API router (mounted under `/api/v1`).
 pub fn api_router() -> Router<Arc<MasterState>> {
     Router::new()
         .route("/experiments", get(list_experiments))
         .route("/experiments/{name}", get(get_experiment))
+        .route("/experiments/{name}/compact", post(compact_experiment))
+        .route("/experiments/{name}/compact/status", get(compact_status))
         .route("/rescan", post(rescan))
 }