feat(master): unified task scheduler for compaction and WAL merge

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

@@ -912,6 +912,78 @@ pub enum CompactJobStatus {
     None,
 }
 
+// ---------------------------------------------------------------------------
+// Unified task scheduler (master control-plane)
+// ---------------------------------------------------------------------------
+
+/// The kind of work a scheduled task performs.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum TaskKind {
+    /// Compact an experiment's base table (rewrites fragments; runs on the
+    /// master, serialized per experiment because two `Rewrite`s conflict).
+    Compact,
+    /// Fold flushed MemWAL generations back into the base table. The master
+    /// cannot do this directly without fencing the live shard writer, so this
+    /// task fans out to the configured worker endpoints and each worker merges
+    /// its own shard.
+    MergeWal,
+}
+
+/// Lifecycle state of a scheduled task, generalized from [`CompactJobStatus`]
+/// so it applies uniformly to every [`TaskKind`].
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum TaskState {
+    /// Accepted and waiting for a scheduler slot.
+    Queued,
+    /// Currently executing.
+    Running,
+    /// Finished successfully.
+    Done,
+    /// Finished with an error (see [`TaskRecord::error`]).
+    Failed,
+}
+
+/// One unit of scheduled work plus its lifecycle, as surfaced to the queue UI.
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct TaskRecord {
+    /// Time-ordered unique id (UUIDv7).
+    pub id: String,
+    pub kind: TaskKind,
+    /// Experiment / rollout-store name this task acts on.
+    pub target: String,
+    pub state: TaskState,
+    /// Error message when `state == Failed`.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub error: Option<String>,
+    /// Human-readable outcome summary when `state == Done`
+    /// (e.g. `"removed 3 / added 1 fragments"` or `"merged 4 gens on 2/3 workers"`).
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub detail: Option<String>,
+    /// When the task was enqueued, Unix milliseconds.
+    pub enqueued_at: i64,
+    /// When execution began, Unix milliseconds, if started.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub started_at: Option<i64>,
+    /// When the task reached a terminal state, Unix milliseconds, if finished.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub finished_at: Option<i64>,
+}
+
+/// Request body for `POST /api/v1/tasks`.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct EnqueueTaskRequest {
+    pub kind: TaskKind,
+    pub target: String,
+}
+
+/// Response for `GET /api/v1/tasks`.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct TaskListResponse {
+    pub tasks: Vec<TaskRecord>,
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -1000,4 +1072,55 @@ mod tests {
         .unwrap();
         assert_eq!(legacy.payload_uri, None);
     }
+
+    #[test]
+    fn task_record_roundtrips_and_omits_empty_optionals() {
+        let queued = TaskRecord {
+            id: "0190-abc".to_string(),
+            kind: TaskKind::MergeWal,
+            target: "exp-1".to_string(),
+            state: TaskState::Queued,
+            error: None,
+            detail: None,
+            enqueued_at: 1_700_000_000_000,
+            started_at: None,
+            finished_at: None,
+        };
+        let json = serde_json::to_string(&queued).unwrap();
+        // snake_case tags for the enums.
+        assert!(json.contains(r#""kind":"merge_wal""#));
+        assert!(json.contains(r#""state":"queued""#));
+        // Absent optionals are not serialized.
+        assert!(!json.contains("error"));
+        assert!(!json.contains("started_at"));
+        assert!(!json.contains("finished_at"));
+        let back: TaskRecord = serde_json::from_str(&json).unwrap();
+        assert_eq!(back, queued);
+    }
+
+    #[test]
+    fn task_record_done_and_failed_carry_details() {
+        let done = TaskRecord {
+            id: "id".to_string(),
+            kind: TaskKind::Compact,
+            target: "exp".to_string(),
+            state: TaskState::Done,
+            error: None,
+            detail: Some("removed 3 / added 1 fragments".to_string()),
+            enqueued_at: 1,
+            started_at: Some(2),
+            finished_at: Some(3),
+        };
+        let back: TaskRecord =
+            serde_json::from_str(&serde_json::to_string(&done).unwrap()).unwrap();
+        assert_eq!(back, done);
+    }
+
+    #[test]
+    fn enqueue_task_request_parses_snake_case_kind() {
+        let req: EnqueueTaskRequest =
+            serde_json::from_str(r#"{"kind":"compact","target":"exp-7"}"#).unwrap();
+        assert_eq!(req.kind, TaskKind::Compact);
+        assert_eq!(req.target, "exp-7");
+    }
 }

crates/lance-context-master/Cargo.toml

@@ -24,6 +24,7 @@ clap = { version = "4", features = ["derive", "env"] }
 futures = "0.3"
 lance = "7.0.0"
 metrics = "0.24"
+reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] }
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 tokio = { version = "1", features = ["rt-multi-thread", "macros", "signal", "sync"] }

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

@@ -40,6 +40,22 @@ pub struct MasterConfig {
     #[arg(long, env = "TARGET_ROWS_PER_FRAGMENT", default_value_t = 1_048_576)]
     pub target_rows_per_fragment: usize,
 
+    /// Data-plane worker base URLs (comma-separated), e.g.
+    /// `http://rollout-0:3000,http://rollout-1:3000`. A `MergeWal` task fans out
+    /// to every endpoint so each worker merges its own MemWAL shard (the master
+    /// cannot merge a shard it does not own without fencing the live writer).
+    /// Empty (the default) means WAL-merge tasks have nowhere to go and fail
+    /// fast with a clear message.
+    #[arg(long, env = "WORKER_ENDPOINTS", value_delimiter = ',')]
+    pub worker_endpoints: Vec<String>,
+
+    /// Maximum number of scheduler tasks executing concurrently. Compaction of
+    /// the *same* experiment is always serialized regardless of this value
+    /// (two `Rewrite`s on one dataset conflict); this only bounds how many
+    /// *distinct* experiments/tasks run at once.
+    #[arg(long, env = "TASK_CONCURRENCY", default_value_t = 4)]
+    pub task_concurrency: usize,
+
     /// Directory of built UI assets to serve. When unset, only the JSON API is
     /// exposed.
     #[arg(long, env = "UI_DIR")]

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

@@ -11,8 +11,9 @@ use axum::{Json, Router};
 use serde::Deserialize;
 
 use lance_context_api::{
-    CompactJobStatus, ExperimentDetail, ExperimentListResponse, ExperimentRecordsResponse,
-    ExperimentSummary,
+    CompactJobStatus, EnqueueTaskRequest, ExperimentDetail, ExperimentListResponse,
+    ExperimentRecordsResponse, ExperimentSummary, TaskKind, TaskListResponse, TaskRecord,
+    TaskState,
 };
 use lance_context_core::{
     rollout_record_to_dto, RolloutFilters, RolloutStore, RolloutStoreOptions,
@@ -205,8 +206,59 @@ pub async fn rescan(
     Ok(Json(serde_json::json!({ "scanned": n })))
 }
 
+/// Map a task's terminal `detail` string back into the fragment counts the
+/// legacy `CompactJobStatus::Done` shape carries. Best-effort: unparsable
+/// details yield zeros (the UI only needs the state to stop polling).
+fn parse_fragment_detail(detail: Option<&str>) -> (usize, usize) {
+    // Detail format: "removed {r} / added {a} fragments".
+    let parse = |s: &str, key: &str| -> usize {
+        s.split_whitespace()
+            .skip_while(|w| *w != key)
+            .nth(1)
+            .and_then(|w| w.parse().ok())
+            .unwrap_or(0)
+    };
+    match detail {
+        Some(d) => (parse(d, "removed"), parse(d, "added")),
+        None => (0, 0),
+    }
+}
+
+/// Project a [`TaskRecord`] into the legacy [`CompactJobStatus`] shape so the
+/// existing compaction UI keeps working while the queue view rolls out.
+fn task_to_compact_status(task: &TaskRecord) -> CompactJobStatus {
+    match task.state {
+        TaskState::Queued => CompactJobStatus::Queued,
+        TaskState::Running => CompactJobStatus::Running,
+        TaskState::Done => {
+            let (fragments_removed, fragments_added) =
+                parse_fragment_detail(task.detail.as_deref());
+            CompactJobStatus::Done {
+                fragments_removed,
+                fragments_added,
+            }
+        }
+        TaskState::Failed => CompactJobStatus::Failed {
+            error: task.error.clone().unwrap_or_default(),
+        },
+    }
+}
+
+/// The most recent `Compact` task for `name`, if any (by `enqueued_at`).
+async fn latest_compact_task(state: &Arc<MasterState>, name: &str) -> Option<TaskRecord> {
+    state
+        .tasks
+        .lock()
+        .await
+        .values()
+        .filter(|t| t.kind == TaskKind::Compact && t.target == name)
+        .max_by_key(|t| t.enqueued_at)
+        .cloned()
+}
+
 /// `POST /api/v1/experiments/{name}/compact` — enqueue a manual compaction.
-/// Returns 202 Accepted with the (possibly de-duped) job status.
+/// Returns 202 Accepted with the (possibly de-duped) job status. Retained for
+/// backward compatibility; internally this is just a `Compact` task.
 pub async fn compact_experiment(
     State(state): State<Arc<MasterState>>,
     Path(name): Path<String>,
@@ -225,8 +277,8 @@ pub async fn compact_experiment(
             name
         )));
     }
-    let status = scheduler::enqueue(&state, &name).await;
-    Ok((StatusCode::ACCEPTED, Json(status)))
+    let task = scheduler::enqueue(&state, TaskKind::Compact, &name).await;
+    Ok((StatusCode::ACCEPTED, Json(task_to_compact_status(&task))))
 }
 
 /// `GET /api/v1/experiments/{name}/compact/status` — latest compaction job
@@ -235,14 +287,51 @@ pub async fn compact_status(
     State(state): State<Arc<MasterState>>,
     Path(name): Path<String>,
 ) -> Json<CompactJobStatus> {
-    let status = state
-        .jobs
-        .lock()
+    match latest_compact_task(&state, &name).await {
+        Some(task) => Json(task_to_compact_status(&task)),
+        None => Json(CompactJobStatus::None),
+    }
+}
+
+/// `POST /api/v1/tasks` — enqueue a task of any kind. Returns 202 Accepted with
+/// the (possibly de-duped) task record.
+pub async fn enqueue_task(
+    State(state): State<Arc<MasterState>>,
+    Json(req): Json<EnqueueTaskRequest>,
+) -> Result<(StatusCode, Json<TaskRecord>), MasterError> {
+    let exists = state
+        .registry
+        .write()
         .await
-        .get(&name)
-        .cloned()
-        .unwrap_or(CompactJobStatus::None);
-    Json(status)
+        .contains(&req.target)
+        .await
+        .map_err(MasterError::from_lance)?;
+    if !exists {
+        return Err(MasterError::NotFound(format!(
+            "experiment '{}' does not exist",
+            req.target
+        )));
+    }
+    let task = scheduler::enqueue(&state, req.kind, &req.target).await;
+    Ok((StatusCode::ACCEPTED, Json(task)))
+}
+
+/// `GET /api/v1/tasks` — all tasks (queue + recent history), newest first.
+pub async fn list_tasks(State(state): State<Arc<MasterState>>) -> Json<TaskListResponse> {
+    let mut tasks: Vec<TaskRecord> = state.tasks.lock().await.values().cloned().collect();
+    tasks.sort_by_key(|t| std::cmp::Reverse(t.enqueued_at));
+    Json(TaskListResponse { tasks })
+}
+
+/// `GET /api/v1/tasks/{id}` — a single task by id.
+pub async fn get_task(
+    State(state): State<Arc<MasterState>>,
+    Path(id): Path<String>,
+) -> Result<Json<TaskRecord>, MasterError> {
+    match state.tasks.lock().await.get(&id).cloned() {
+        Some(task) => Ok(Json(task)),
+        None => Err(MasterError::NotFound(format!("task '{}' not found", id))),
+    }
 }
 
 /// Build the admin API router (mounted under `/api/v1`).
@@ -257,6 +346,8 @@ pub fn api_router() -> Router<Arc<MasterState>> {
         )
         .route("/experiments/{name}/compact", post(compact_experiment))
         .route("/experiments/{name}/compact/status", get(compact_status))
+        .route("/tasks", post(enqueue_task).get(list_tasks))
+        .route("/tasks/{id}", get(get_task))
         .route("/rescan", post(rescan))
 }
 
@@ -325,6 +416,8 @@ mod tests {
             compaction_interval_secs: 0,
             min_fragments: 16,
             target_rows_per_fragment: 1_048_576,
+            worker_endpoints: vec![],
+            task_concurrency: 4,
             ui_dir: None,
         }
     }