From 7f8e5f17848d187227c76378dbd86db6136ee054 Mon Sep 17 00:00:00 2001 From: Beinan Date: Fri, 24 Jul 2026 20:04:45 +0000 Subject: [PATCH] feat(master): paginate task view and add TTL-based history cleanup MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Task Queue view returned every task and rendered them all, and terminal task history was only bounded by a count cap (TASK_HISTORY_LIMIT), so old finished tasks under the cap lived forever. - TTL cleanup: new TASK_HISTORY_TTL_SECS (default 24h, 0 disables) prunes terminal Done/Failed tasks older than the TTL. It composes with the existing count cap — whichever removes a task first wins — and reuses the existing prune trigger points (open/finish/claim), so no new background loop. Queued/Running tasks and terminal tasks a live task still depends on are never pruned. Prune selection is extracted to a pure, unit-tested `prunable_terminal_ids`. - Server-side pagination: GET /api/v1/tasks now takes limit/offset (default 50, clamped 1..=200) and returns total/limit/offset alongside the page. The TaskQueue view gains Prev/Next paging (reusing the records pager) and shows the true total in the stat strip. Co-Authored-By: Claude --- crates/lance-context-api/src/lib.rs | 6 +- crates/lance-context-master/src/config.rs | 8 ++ crates/lance-context-master/src/routes.rs | 23 ++- crates/lance-context-master/src/scheduler.rs | 1 + crates/lance-context-master/src/state.rs | 1 + crates/lance-context-master/src/task_store.rs | 132 +++++++++++++++--- crates/lance-context-master/ui/src/App.tsx | 24 +++- crates/lance-context-master/ui/src/api.ts | 13 +- 8 files changed, 182 insertions(+), 26 deletions(-) diff --git a/crates/lance-context-api/src/lib.rs b/crates/lance-context-api/src/lib.rs index 8dbc682..b696982 100644 --- a/crates/lance-context-api/src/lib.rs +++ b/crates/lance-context-api/src/lib.rs @@ -1070,10 +1070,14 @@ pub struct EnqueueTaskRequest { pub depends_on: Vec, } -/// Response for `GET /api/v1/tasks`. +/// Response for `GET /api/v1/tasks`. Paginated newest-first. #[derive(Debug, Clone, Serialize, Deserialize)] pub struct TaskListResponse { pub tasks: Vec, + /// Total number of tasks (queue + retained history) before paging. + pub total: usize, + pub limit: usize, + pub offset: usize, } #[cfg(test)] diff --git a/crates/lance-context-master/src/config.rs b/crates/lance-context-master/src/config.rs index 27c4e00..eb62a8a 100644 --- a/crates/lance-context-master/src/config.rs +++ b/crates/lance-context-master/src/config.rs @@ -110,6 +110,14 @@ pub struct MasterConfig { #[arg(long, env = "TASK_HISTORY_LIMIT", default_value_t = 1_000)] pub task_history_limit: usize, + /// Maximum age of terminal (Done/Failed) scheduler tasks before they are + /// pruned, in seconds. `0` disables age-based pruning (the count cap in + /// `TASK_HISTORY_LIMIT` still applies). Whichever of the two removes a task + /// first wins. Queued/running tasks and terminal tasks that a live task + /// still depends on are never pruned regardless of age. + #[arg(long, env = "TASK_HISTORY_TTL_SECS", default_value_t = 86_400)] + pub task_history_ttl_secs: u64, + /// Directory of built UI assets to serve. When unset, only the JSON API is /// exposed. #[arg(long, env = "UI_DIR")] diff --git a/crates/lance-context-master/src/routes.rs b/crates/lance-context-master/src/routes.rs index 6db326b..2ae436e 100644 --- a/crates/lance-context-master/src/routes.rs +++ b/crates/lance-context-master/src/routes.rs @@ -42,6 +42,15 @@ fn default_records_limit() -> usize { 25 } +/// Query params for the paginated task list. +#[derive(Debug, Deserialize)] +pub struct TaskListParams { + #[serde(default = "default_limit")] + pub limit: usize, + #[serde(default)] + pub offset: usize, +} + /// Query params for the detail endpoint. #[derive(Debug, Deserialize)] pub struct DetailParams { @@ -410,9 +419,10 @@ pub async fn enqueue_task( Ok((StatusCode::ACCEPTED, Json(task))) } -/// `GET /api/v1/tasks` — all tasks (queue + recent history), newest first. +/// `GET /api/v1/tasks` — paginated tasks (queue + recent history), newest first. pub async fn list_tasks( State(state): State>, + Query(params): Query, ) -> Result, MasterError> { let mut tasks = state .task_store @@ -420,7 +430,15 @@ pub async fn list_tasks( .await .map_err(MasterError::from_lance)?; tasks.sort_by_key(|t| std::cmp::Reverse(t.enqueued_at)); - Ok(Json(TaskListResponse { tasks })) + let total = tasks.len(); + let limit = params.limit.clamp(1, 200); + let page = tasks.into_iter().skip(params.offset).take(limit).collect(); + Ok(Json(TaskListResponse { + tasks: page, + total, + limit, + offset: params.offset, + })) } /// `GET /api/v1/tasks/{id}` — a single task by id. @@ -536,6 +554,7 @@ mod tests { etcd_client_key: None, etcd_lease_ttl_secs: 5, task_history_limit: 1_000, + task_history_ttl_secs: 86_400, ui_dir: None, } } diff --git a/crates/lance-context-master/src/scheduler.rs b/crates/lance-context-master/src/scheduler.rs index fbc5508..28044e7 100644 --- a/crates/lance-context-master/src/scheduler.rs +++ b/crates/lance-context-master/src/scheduler.rs @@ -449,6 +449,7 @@ mod tests { etcd_client_key: None, etcd_lease_ttl_secs: 5, task_history_limit: 1_000, + task_history_ttl_secs: 86_400, ui_dir: None, } } diff --git a/crates/lance-context-master/src/state.rs b/crates/lance-context-master/src/state.rs index c56a45d..f3b6ac3 100644 --- a/crates/lance-context-master/src/state.rs +++ b/crates/lance-context-master/src/state.rs @@ -149,6 +149,7 @@ mod tests { etcd_client_key: None, etcd_lease_ttl_secs: 5, task_history_limit: 1_000, + task_history_ttl_secs: 86_400, ui_dir: None, } } diff --git a/crates/lance-context-master/src/task_store.rs b/crates/lance-context-master/src/task_store.rs index cfeeb29..09680a0 100644 --- a/crates/lance-context-master/src/task_store.rs +++ b/crates/lance-context-master/src/task_store.rs @@ -29,6 +29,7 @@ const TASK_POLL_BATCH: usize = 256; pub struct TaskStore { inner: Arc, history_limit: usize, + history_ttl_secs: u64, } struct EtcdTaskStore { @@ -80,6 +81,7 @@ impl TaskStore { let store = Self { inner: Arc::new(EtcdTaskStore::connect(config).await?), history_limit: config.task_history_limit.max(1), + history_ttl_secs: config.task_history_ttl_secs, }; store.recover_orphaned().await?; store.prune_terminal_history().await?; @@ -171,23 +173,12 @@ impl TaskStore { async fn prune_terminal_history(&self) -> lance::Result { let tasks = self.list().await?; - let protected = tasks - .iter() - .filter(|task| matches!(task.state, TaskState::Queued | TaskState::Running)) - .flat_map(|task| task.depends_on.iter().cloned()) - .collect::>(); - let mut terminal = tasks - .into_iter() - .filter(|task| matches!(task.state, TaskState::Done | TaskState::Failed)) - .filter(|task| !protected.contains(&task.id)) - .collect::>(); - terminal - .sort_by_key(|task| std::cmp::Reverse(task.finished_at.unwrap_or(task.enqueued_at))); - let ids = terminal - .into_iter() - .skip(self.history_limit) - .map(|task| task.id) - .collect::>(); + let ttl_cutoff = if self.history_ttl_secs > 0 { + Some(now_ms() - (self.history_ttl_secs as i64) * 1_000) + } else { + None + }; + let ids = prunable_terminal_ids(tasks, self.history_limit, ttl_cutoff); if ids.is_empty() { return Ok(0); } @@ -196,6 +187,45 @@ impl TaskStore { } } +/// Select terminal (Done/Failed) task ids to prune under two independent +/// policies, whichever removes a task first: +/// - **count**: keep only the newest `history_limit` terminal tasks; +/// - **age (TTL)**: when `ttl_cutoff` is `Some`, drop terminal tasks whose +/// `finished_at` (fallback `enqueued_at`) is at or before the cutoff. +/// +/// Queued/Running tasks, and any terminal task a live (Queued/Running) task +/// still lists in `depends_on`, are never selected regardless of age or count. +/// Pure and etcd-free so the policy is unit-testable. +fn prunable_terminal_ids( + tasks: Vec, + history_limit: usize, + ttl_cutoff: Option, +) -> Vec { + let protected = tasks + .iter() + .filter(|task| matches!(task.state, TaskState::Queued | TaskState::Running)) + .flat_map(|task| task.depends_on.iter().cloned()) + .collect::>(); + let mut terminal = tasks + .into_iter() + .filter(|task| matches!(task.state, TaskState::Done | TaskState::Failed)) + .filter(|task| !protected.contains(&task.id)) + .collect::>(); + // Newest first, so rank >= history_limit are the ones over the count cap. + terminal.sort_by_key(|task| std::cmp::Reverse(task.finished_at.unwrap_or(task.enqueued_at))); + terminal + .into_iter() + .enumerate() + .filter(|(rank, task)| { + let over_count = *rank >= history_limit; + let expired = ttl_cutoff + .is_some_and(|cutoff| task.finished_at.unwrap_or(task.enqueued_at) <= cutoff); + over_count || expired + }) + .map(|(_, task)| task.id) + .collect() +} + impl EtcdTaskStore { async fn connect(config: &MasterConfig) -> lance::Result { if config.etcd_endpoints.is_empty() { @@ -896,6 +926,7 @@ mod tests { etcd_client_key: None, etcd_lease_ttl_secs: 30, task_history_limit: 1_000, + task_history_ttl_secs: 86_400, ui_dir: None, } } @@ -910,6 +941,73 @@ mod tests { assert!(error.to_string().contains("ETCD_ENDPOINTS is required")); } + fn terminal_task(id: &str, state: TaskState, finished_at: i64) -> TaskRecord { + TaskRecord { + id: id.to_string(), + kind: TaskKind::Compact, + target: "exp".to_string(), + state, + error: None, + detail: None, + enqueued_at: finished_at, + started_at: Some(finished_at), + finished_at: Some(finished_at), + depends_on: Vec::new(), + } + } + + #[test] + fn prune_keeps_newest_over_count_cap() { + // 5 terminal tasks, cap 2, TTL disabled → oldest 3 pruned. + let tasks = (0..5) + .map(|i| terminal_task(&format!("t{i}"), TaskState::Done, 1_000 + i * 10)) + .collect::>(); + let mut pruned = prunable_terminal_ids(tasks, 2, None); + pruned.sort(); + assert_eq!(pruned, vec!["t0", "t1", "t2"]); + } + + #[test] + fn prune_expires_by_ttl_cutoff() { + // High count cap so only the TTL policy fires. + let tasks = vec![ + terminal_task("old-1", TaskState::Done, 100), + terminal_task("old-2", TaskState::Failed, 200), + terminal_task("fresh", TaskState::Done, 5_000), + ]; + let mut pruned = prunable_terminal_ids(tasks, 1_000, Some(1_000)); + pruned.sort(); + assert_eq!(pruned, vec!["old-1", "old-2"]); + } + + #[test] + fn prune_never_touches_active_or_depended_on() { + // A queued task depends on a terminal one that is otherwise TTL-expired; + // that dependency must be protected. Queued/Running are never terminal + // candidates in the first place. + let mut dep = terminal_task("dep", TaskState::Done, 100); + dep.id = "dep".to_string(); + let mut queued = terminal_task("live", TaskState::Queued, 100); + queued.depends_on = vec!["dep".to_string()]; + let expired = terminal_task("expired", TaskState::Done, 100); + let tasks = vec![dep, queued, expired]; + + let pruned = prunable_terminal_ids(tasks, 0, Some(1_000)); + // `dep` protected by the live task; `live` is Queued (not terminal); + // only the unreferenced expired terminal task is pruned. + assert_eq!(pruned, vec!["expired"]); + } + + #[test] + fn prune_ttl_disabled_uses_count_only() { + let tasks = vec![ + terminal_task("a", TaskState::Done, 1), + terminal_task("b", TaskState::Done, 2), + ]; + // TTL None + generous cap → nothing pruned even though timestamps are old. + assert!(prunable_terminal_ids(tasks, 10, None).is_empty()); + } + #[tokio::test] #[ignore = "requires ETCD_TEST_ENDPOINTS"] async fn etcd_coordinates_dedupe_claims_and_target_locks() { diff --git a/crates/lance-context-master/ui/src/App.tsx b/crates/lance-context-master/ui/src/App.tsx index 4e0a073..7b7a3af 100644 --- a/crates/lance-context-master/ui/src/App.tsx +++ b/crates/lance-context-master/ui/src/App.tsx @@ -931,19 +931,24 @@ function IndexIdButton({ name }: { name: string }) { } function TaskQueue() { + const [page, setPage] = useState(0); + const pageSize = 50; const tasks = useQuery({ - queryKey: ["tasks"], - queryFn: () => listTasks(), + queryKey: ["tasks", page], + queryFn: () => listTasks(pageSize, page * pageSize), refetchInterval: 1000, + placeholderData: (prev) => prev, }); const rows = tasks.data?.tasks ?? []; + const total = tasks.data?.total ?? 0; const active = rows.filter((t) => t.state === "queued" || t.state === "running").length; + const hasMore = (page + 1) * pageSize < total; return ( <>
- - + +
{tasks.isError &&
{String(tasks.error)}
} @@ -979,6 +984,17 @@ function TaskQueue() {
No tasks in the queue.
)}
+ {(page > 0 || hasMore) && ( +
+ page {page + 1} + + +
+ )} ); } diff --git a/crates/lance-context-master/ui/src/api.ts b/crates/lance-context-master/ui/src/api.ts index 1fb8268..6378f42 100644 --- a/crates/lance-context-master/ui/src/api.ts +++ b/crates/lance-context-master/ui/src/api.ts @@ -104,6 +104,9 @@ export interface TaskRecord { export interface TaskListResponse { tasks: TaskRecord[]; + total: number; + limit: number; + offset: number; } export interface EnqueueTaskRequest { @@ -181,8 +184,14 @@ export async function compactionStatus(name: string): Promise ); } -export async function listTasks(): Promise { - return json(await fetch(`${API}/tasks`)); +export async function listTasks( + limit = 50, + offset = 0, +): Promise { + const params = new URLSearchParams(); + params.set("limit", String(limit)); + params.set("offset", String(offset)); + return json(await fetch(`${API}/tasks?${params.toString()}`)); } export async function getTask(id: string): Promise {