diff --git a/crates/lance-context-api/src/lib.rs b/crates/lance-context-api/src/lib.rs index 78d13ea..f59670c 100644 --- a/crates/lance-context-api/src/lib.rs +++ b/crates/lance-context-api/src/lib.rs @@ -980,6 +980,11 @@ pub struct TaskRecord { /// When the task reached a terminal state, Unix milliseconds, if finished. #[serde(default, skip_serializing_if = "Option::is_none")] pub finished_at: Option, + /// Task ids that must reach `Done` before this task may run. The scheduler + /// defers a task while any dependency is still `Queued`/`Running`, and marks + /// it `Failed` if any dependency `Failed`. Empty for standalone tasks. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub depends_on: Vec, } /// Request body for `POST /api/v1/tasks`. @@ -987,6 +992,9 @@ pub struct TaskRecord { pub struct EnqueueTaskRequest { pub kind: TaskKind, pub target: String, + /// Optional task ids this task must wait for (see [`TaskRecord::depends_on`]). + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub depends_on: Vec, } /// Response for `GET /api/v1/tasks`. @@ -1096,6 +1104,7 @@ mod tests { enqueued_at: 1_700_000_000_000, started_at: None, finished_at: None, + depends_on: Vec::new(), }; let json = serde_json::to_string(&queued).unwrap(); // snake_case tags for the enums. @@ -1105,6 +1114,7 @@ mod tests { assert!(!json.contains("error")); assert!(!json.contains("started_at")); assert!(!json.contains("finished_at")); + assert!(!json.contains("depends_on")); let back: TaskRecord = serde_json::from_str(&json).unwrap(); assert_eq!(back, queued); } @@ -1121,6 +1131,7 @@ mod tests { enqueued_at: 1, started_at: Some(2), finished_at: Some(3), + depends_on: Vec::new(), }; let back: TaskRecord = serde_json::from_str(&serde_json::to_string(&done).unwrap()).unwrap(); diff --git a/crates/lance-context-master/src/routes.rs b/crates/lance-context-master/src/routes.rs index 338f8cb..931dc90 100644 --- a/crates/lance-context-master/src/routes.rs +++ b/crates/lance-context-master/src/routes.rs @@ -314,7 +314,7 @@ pub async fn enqueue_task( req.target ))); } - let task = scheduler::enqueue(&state, req.kind, &req.target) + let task = scheduler::enqueue_with_deps(&state, req.kind, &req.target, req.depends_on) .await .map_err(MasterError::from_lance)?; Ok((StatusCode::ACCEPTED, Json(task))) diff --git a/crates/lance-context-master/src/scheduler.rs b/crates/lance-context-master/src/scheduler.rs index f155a40..a4f7024 100644 --- a/crates/lance-context-master/src/scheduler.rs +++ b/crates/lance-context-master/src/scheduler.rs @@ -68,9 +68,22 @@ pub async fn enqueue( state: &Arc, kind: TaskKind, target: &str, +) -> lance::Result { + enqueue_with_deps(state, kind, target, Vec::new()).await +} + +/// Like [`enqueue`] but the task waits for `depends_on` (task ids) to reach +/// `Done` before it runs. De-dup is skipped when dependencies are present: a +/// dependent task is part of an ordered chain and must not collapse into an +/// unrelated in-flight task for the same target. +pub async fn enqueue_with_deps( + state: &Arc, + kind: TaskKind, + target: &str, + depends_on: Vec, ) -> lance::Result { let mut tasks = state.tasks.lock().await; - if matches!(kind, TaskKind::Compact | TaskKind::IndexId) { + if depends_on.is_empty() && matches!(kind, TaskKind::Compact | TaskKind::IndexId) { if let Some(existing) = tasks.values().find(|t| { t.kind == kind && t.target == target @@ -90,6 +103,7 @@ pub async fn enqueue( enqueued_at: now_ms(), started_at: None, finished_at: None, + depends_on, }; state.task_store.put(&record)?; tasks.insert(record.id.clone(), record.clone()); @@ -126,6 +140,38 @@ async fn update_task( Ok(()) } +/// Readiness of a task with respect to its declared dependencies. +enum DepStatus { + /// All dependencies reached `Done` (or the task has none): ok to run. + Ready, + /// At least one dependency is still `Queued`/`Running`: defer and re-check. + Waiting, + /// At least one dependency `Failed` (or vanished): the dependent cannot run. + /// Carries the id of the offending dependency for the error message. + DepFailed(String), +} + +/// Inspect a task's `depends_on` against the current task map. A missing +/// dependency is treated as failed (its record was never created or was +/// dropped), so a dependent never waits forever on a ghost id. +async fn dep_status(state: &Arc, id: &str) -> DepStatus { + let tasks = state.tasks.lock().await; + let Some(task) = tasks.get(id) else { + return DepStatus::Ready; // vanished; run_task will no-op + }; + for dep in &task.depends_on { + match tasks.get(dep) { + Some(d) => match d.state { + TaskState::Done => {} + TaskState::Failed => return DepStatus::DepFailed(dep.clone()), + TaskState::Queued | TaskState::Running => return DepStatus::Waiting, + }, + None => return DepStatus::DepFailed(dep.clone()), + } + } + DepStatus::Ready +} + /// Execute one task by id: dispatch on its kind, recording lifecycle timestamps /// and the terminal state. async fn run_task(state: &Arc, id: String) { @@ -435,6 +481,36 @@ pub fn spawn_scheduler(state: &Arc) -> JoinHandle<()> { let dispatch_state = state.clone(); tokio::spawn(async move { while let Some(id) = rx.recv().await { + // Resolve dependencies before taking a slot, so a task waiting on a + // chain never occupies a permit that its own dependency needs to + // finish (which would deadlock at concurrency == 1). + match dep_status(&dispatch_state, &id).await { + DepStatus::Ready => {} + DepStatus::DepFailed(dep) => { + // Skip: mark the dependent Failed and move on. + update_task(&dispatch_state, &id, |t| { + t.state = TaskState::Failed; + t.finished_at = Some(now_ms()); + t.error = Some(format!("dependency {dep} did not complete")); + }) + .await + .unwrap_or_else(|error| { + tracing::error!(task = %id, error = %error, "failed to persist skipped task"); + }); + metrics::gauge!("master_task_queue_depth").decrement(1.0); + continue; + } + DepStatus::Waiting => { + // Re-check later without busy-looping. Re-send the id after a + // short delay; the queue is unbounded so this always succeeds. + let tx = dispatch_state.task_tx.clone(); + tokio::spawn(async move { + tokio::time::sleep(Duration::from_millis(200)).await; + let _ = tx.send(id); + }); + continue; + } + } // Acquire a slot before spawning so at most `concurrency` run at once. let permit = sem .clone() @@ -645,8 +721,82 @@ mod tests { worker.abort(); } - /// Minimal rollout record builder for tests (the core struct has no - /// `Default`). + /// A dependent task runs only after its dependency reaches `Done`: an + /// `index_id` depending on a `compact` must start after compaction finishes, + /// so the two never contend for the shared per-experiment base-table gate. + #[tokio::test] + async fn dependent_task_runs_after_dependency_done() { + let dir = TempDir::new().unwrap(); + let state = MasterState::new(config(&dir)).await.unwrap(); + let worker = spawn_scheduler(&state); + + let name = "exp"; + let uri = state.rollout_uri(name); + { + let mut store = RolloutStore::open(&uri).await.unwrap(); + for i in 0..4 { + let rec = rollout_record(&format!("r{i}")); + store.add(&[rec]).await.unwrap(); + store.cleanup_own_shard().await.unwrap(); + } + } + state + .registry + .write() + .await + .upsert(name, &uri) + .await + .unwrap(); + crate::scanner::scan_once(&state).await.unwrap(); + + let compact = enqueue(&state, TaskKind::Compact, name).await.unwrap(); + let index = enqueue_with_deps(&state, TaskKind::IndexId, name, vec![compact.id.clone()]) + .await + .unwrap(); + + let compact_final = await_terminal(&state, &compact.id).await; + assert_eq!( + compact_final.state, + TaskState::Done, + "got {compact_final:?}" + ); + let index_final = await_terminal(&state, &index.id).await; + assert_eq!(index_final.state, TaskState::Done, "got {index_final:?}"); + // The dependent could not have started before the dependency finished. + assert!( + index_final.started_at.unwrap() >= compact_final.finished_at.unwrap(), + "index started {:?} before compact finished {:?}", + index_final.started_at, + compact_final.finished_at + ); + + worker.abort(); + } + + /// A dependent whose dependency `Failed` is skipped (marked `Failed`) rather + /// than run. + #[tokio::test] + async fn dependent_skipped_when_dependency_fails() { + let dir = TempDir::new().unwrap(); + let state = MasterState::new(config(&dir)).await.unwrap(); + let worker = spawn_scheduler(&state); + + // MergeWal with no worker endpoints fails; its dependent must be skipped. + let merge = enqueue(&state, TaskKind::MergeWal, "exp").await.unwrap(); + let dependent = enqueue_with_deps(&state, TaskKind::Compact, "exp", vec![merge.id.clone()]) + .await + .unwrap(); + + let merge_final = await_terminal(&state, &merge.id).await; + assert_eq!(merge_final.state, TaskState::Failed); + let dep_final = await_terminal(&state, &dependent.id).await; + assert_eq!(dep_final.state, TaskState::Failed, "got {dep_final:?}"); + assert!(dep_final.error.unwrap().contains("dependency")); + + worker.abort(); + } + + /// Minimal rollout record builder for tests (the core struct has no /// `Default`). pub fn rollout_record(id: &str) -> lance_context_core::RolloutRecord { use chrono::TimeZone; lance_context_core::RolloutRecord { diff --git a/crates/lance-context-master/src/state.rs b/crates/lance-context-master/src/state.rs index af96a97..689d2c6 100644 --- a/crates/lance-context-master/src/state.rs +++ b/crates/lance-context-master/src/state.rs @@ -222,6 +222,7 @@ mod tests { enqueued_at: 1, started_at: (task_state == TaskState::Running).then_some(2), finished_at: (task_state == TaskState::Done).then_some(3), + depends_on: Vec::new(), }; state.task_store.put(&task).unwrap(); } @@ -265,6 +266,7 @@ mod tests { started_at: None, finished_at: matches!(state, TaskState::Done | TaskState::Failed) .then_some(timestamp), + depends_on: Vec::new(), }; store.put(&task).unwrap(); tasks.insert(id.to_string(), task); diff --git a/crates/lance-context-master/src/task_store.rs b/crates/lance-context-master/src/task_store.rs index b60926b..2e47e91 100644 --- a/crates/lance-context-master/src/task_store.rs +++ b/crates/lance-context-master/src/task_store.rs @@ -127,6 +127,7 @@ mod tests { enqueued_at, started_at: None, finished_at: None, + depends_on: Vec::new(), } } diff --git a/crates/lance-context-master/ui/package-lock.json b/crates/lance-context-master/ui/package-lock.json index 4d382d9..77073db 100644 --- a/crates/lance-context-master/ui/package-lock.json +++ b/crates/lance-context-master/ui/package-lock.json @@ -11,6 +11,7 @@ "@tanstack/react-query": "^5.59.0", "react": "^18.3.1", "react-dom": "^18.3.1", + "react-router-dom": "^6.30.4", "zustand": "^4.5.5" }, "devDependencies": { @@ -744,6 +745,15 @@ "@jridgewell/sourcemap-codec": "^1.4.14" } }, + "node_modules/@remix-run/router": { + "version": "1.23.3", + "resolved": "https://ms-feed-12.pkgs.visualstudio.com/1es-public/_packaging/npm-public/npm/registry/@remix-run/router/-/router-1.23.3.tgz", + "integrity": "sha1-lXwJjUOT0wGoqn3M8+8o6lQw42o=", + "license": "MIT", + "engines": { + "node": ">=14.0.0" + } + }, "node_modules/@rolldown/pluginutils": { "version": "1.0.0-beta.27", "resolved": "https://ms-feed-12.pkgs.visualstudio.com/1es-public/_packaging/npm-public/npm/registry/@rolldown/pluginutils/-/pluginutils-1.0.0-beta.27.tgz", @@ -1569,6 +1579,38 @@ "node": ">=0.10.0" } }, + "node_modules/react-router": { + "version": "6.30.4", + "resolved": "https://ms-feed-12.pkgs.visualstudio.com/1es-public/_packaging/npm-public/npm/registry/react-router/-/react-router-6.30.4.tgz", + "integrity": "sha1-Y481F2UnvSQ9ltgdNdM7dXutRsI=", + "license": "MIT", + "dependencies": { + "@remix-run/router": "1.23.3" + }, + "engines": { + "node": ">=14.0.0" + }, + "peerDependencies": { + "react": ">=16.8" + } + }, + "node_modules/react-router-dom": { + "version": "6.30.4", + "resolved": "https://ms-feed-12.pkgs.visualstudio.com/1es-public/_packaging/npm-public/npm/registry/react-router-dom/-/react-router-dom-6.30.4.tgz", + "integrity": "sha1-9xZ789psfZEyEw6phd0G3vJehNU=", + "license": "MIT", + "dependencies": { + "@remix-run/router": "1.23.3", + "react-router": "6.30.4" + }, + "engines": { + "node": ">=14.0.0" + }, + "peerDependencies": { + "react": ">=16.8", + "react-dom": ">=16.8" + } + }, "node_modules/rollup": { "version": "4.62.2", "resolved": "https://ms-feed-12.pkgs.visualstudio.com/1es-public/_packaging/npm-public/npm/registry/rollup/-/rollup-4.62.2.tgz", diff --git a/crates/lance-context-master/ui/package.json b/crates/lance-context-master/ui/package.json index bd4a747..c3090c1 100644 --- a/crates/lance-context-master/ui/package.json +++ b/crates/lance-context-master/ui/package.json @@ -12,6 +12,7 @@ "@tanstack/react-query": "^5.59.0", "react": "^18.3.1", "react-dom": "^18.3.1", + "react-router-dom": "^6.30.4", "zustand": "^4.5.5" }, "devDependencies": { diff --git a/crates/lance-context-master/ui/src/App.tsx b/crates/lance-context-master/ui/src/App.tsx index 8d25ffa..7c19863 100644 --- a/crates/lance-context-master/ui/src/App.tsx +++ b/crates/lance-context-master/ui/src/App.tsx @@ -1,5 +1,13 @@ -import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query"; +import { useMutation, useQuery, useQueryClient, useIsFetching } from "@tanstack/react-query"; import { Fragment, useEffect, useState, type FormEvent } from "react"; +import { + NavLink, + Navigate, + Route, + Routes, + useNavigate, + useParams, +} from "react-router-dom"; import { compactionStatus, enqueueTask, @@ -80,8 +88,31 @@ function DownloadIcon() { ); } -function ChevronIcon({ open }: { open: boolean }) { +function LinkIcon() { return ( + + + + + ); +} + +/** Copy an absolute URL (origin + given path) to the clipboard. Best-effort: + * silently ignores clipboard failures (e.g. non-secure contexts). */ +function copyLink(path: string) { + const url = `${window.location.origin}${path}`; + void navigator.clipboard?.writeText(url).catch(() => {}); +} + +function ChevronIcon({ open }: { open: boolean }) { return ( { + // Chain the three tasks so they run in order: compact waits for the WAL + // merge, and the id-index build waits for compaction. compact and + // index_id share a per-experiment base-table write gate, so running them + // concurrently would make one fail; the dependency chain serializes them. + const merge = await enqueueTask({ kind: "merge_wal", target: name }); + const compact = await enqueueTask({ + kind: "compact", + target: name, + depends_on: [merge.id], + }); + await enqueueTask({ + kind: "index_id", + target: name, + depends_on: [compact.id], + }); + }, + onSuccess: () => { + qc.invalidateQueries({ queryKey: ["tasks"] }); + navigate("/tasks"); + }, + }); + return ( + + ); +} + function CompactButton({ name, variant }: { name: string; variant?: "accent" }) { const { status, trigger, busy } = useCompaction(name); return ( @@ -180,12 +253,15 @@ function CompactButton({ name, variant }: { name: string; variant?: "accent" }) /* ---- table --------------------------------------------------------------- */ function Row({ exp }: { exp: ExperimentSummary }) { - const select = useUiStore((s) => s.select); + const navigate = useNavigate(); const hot = exp.fragment_count >= 16; return (
- @@ -195,7 +271,7 @@ function Row({ exp }: { exp: ExperimentSummary }) { {relTime(exp.last_updated)} {relTime(exp.last_compaction)} - +
); @@ -241,6 +317,19 @@ function JsonBlock({ value }: { value: unknown }) { function RecordDetails({ record, name }: { record: RolloutRecord; name: string }) { return (
+
+ +
ID @@ -347,11 +436,18 @@ function RecordDetails({ record, name }: { record: RolloutRecord; name: string } ); } -function RecordsView({ name }: { name: string }) { +function RecordsView({ + name, + expandedId, + onToggle, +}: { + name: string; + expandedId: string | null; + onToggle: (id: string | null) => void; +}) { const [draft, setDraft] = useState({ ...EMPTY_RECORD_FILTERS }); const [filters, setFilters] = useState({ ...EMPTY_RECORD_FILTERS }); const [page, setPage] = useState(0); - const [expanded, setExpanded] = useState(null); const pageSize = 25; const records = useQuery({ queryKey: ["experiment-records", name, filters, page], @@ -368,13 +464,13 @@ function RecordsView({ name }: { name: string }) { event.preventDefault(); setFilters({ ...draft }); setPage(0); - setExpanded(null); + onToggle(null); }; const clear = () => { setDraft({ ...EMPTY_RECORD_FILTERS }); setFilters({ ...EMPTY_RECORD_FILTERS }); setPage(0); - setExpanded(null); + onToggle(null); }; const hasFilters = Object.values(filters).some(Boolean); @@ -493,7 +589,7 @@ function RecordsView({ name }: { name: string }) { ))} {!records.isLoading && rows.map((record) => { - const open = expanded === record.id; + const open = expandedId === record.id; const signal = record.score ?? record.reward; return ( @@ -501,14 +597,14 @@ function RecordsView({ name }: { name: string }) { - @@ -575,9 +671,18 @@ function RecordsView({ name }: { name: string }) { ); } -function Drawer({ name }: { name: string }) { - const select = useUiStore((s) => s.select); - const [tab, setTab] = useState<"records" | "overview">("records"); +function Drawer({ + name, + tab, + expandedId, +}: { + name: string; + tab: "records" | "overview"; + expandedId: string | null; +}) { + const navigate = useNavigate(); + const encName = encodeURIComponent(name); + const close = () => navigate("/experiments"); const detail = useQuery({ queryKey: ["experiment", name], queryFn: () => getExperiment(name), @@ -585,24 +690,38 @@ function Drawer({ name }: { name: string }) { const d = detail.data; useEffect(() => { - const onKey = (e: KeyboardEvent) => e.key === "Escape" && select(null); + const onKey = (e: KeyboardEvent) => e.key === "Escape" && navigate("/experiments"); window.addEventListener("keydown", onKey); return () => window.removeEventListener("keydown", onKey); - }, [select]); + }, [navigate]); + + // Toggling a record maps to a URL: expanded → /records/:id, collapsed → base. + const onToggle = (id: string | null) => { + if (id) navigate(`/experiments/${encName}/records/${encodeURIComponent(id)}`); + else navigate(`/experiments/${encName}`); + }; return ( <> -
select(null)} /> +