diff --git a/crates/utopia-server/src/api/jobs_routes.rs b/crates/utopia-server/src/api/jobs_routes.rs new file mode 100644 index 00000000..a48fad2d --- /dev/null +++ b/crates/utopia-server/src/api/jobs_routes.rs @@ -0,0 +1,98 @@ +//! 失败任务回队列（#216）。 +//! +//! 一个失败任务原本只能通过它属的对象再跑：文档能重抽、来源能重同步；`bootstrap_ontology`、 +//! `adjudicate_entities` 这些没有对象可点。余额耗尽（#201 让它第一次就 failed）一批文档 +//! 全停，充值之后要逐个点。这里给两个入口：库内（Editor）与全局（管理员），范围可按 +//! 种类与失败时间收窄——告警上的「再跑一遍」传的就是那次故障的时间窗。 + +use axum::extract::{Path, State}; +use axum::Json; +use serde::Deserialize; +use serde_json::json; +use utopia_core::models::Role; +use utopia_store::jobs::RequeueScope; +use uuid::Uuid; + +use super::graph_routes::require_kb; +use crate::auth::AuthUser; +use crate::error::ApiResult; +use crate::state::AppState; + +#[derive(Deserialize, Default)] +pub struct RequeueBody { + #[serde(default)] + pub kind: Option, + /// 只排这个时刻之后失败的 + #[serde(default)] + pub failed_since: Option>, +} + +pub async fn failed_in_kb( + State(state): State, + AuthUser(user): AuthUser, + Path(kb_id): Path, +) -> ApiResult> { + require_kb(&state, &user, kb_id, Role::Viewer).await?; + let failed = utopia_store::jobs::failed_count(&state.pool, Some(kb_id)).await?; + Ok(Json(json!({ "failed": failed }))) +} + +pub async fn requeue_in_kb( + State(state): State, + AuthUser(user): AuthUser, + Path(kb_id): Path, + Json(body): Json, +) -> ApiResult> { + require_kb(&state, &user, kb_id, Role::Editor).await?; + let requeued = utopia_store::jobs::requeue_failed( + &state.pool, + RequeueScope { + kb_id: Some(kb_id), + kind: body.kind.as_deref(), + failed_since: body.failed_since, + }, + ) + .await?; + let _ = utopia_store::audit::record( + &state.pool, + Some(kb_id), + user.id, + "jobs.requeued", + "kb", + Some(kb_id), + json!({ "requeued": requeued, "kind": body.kind, "failed_since": body.failed_since }), + ) + .await; + Ok(Json(json!({ "requeued": requeued }))) +} + +/// 全局重排：系统级告警（没有库的）从这里走。只给管理员——它碰的是所有库的任务 +pub async fn requeue_all( + State(state): State, + AuthUser(user): AuthUser, + Json(body): Json, +) -> ApiResult> { + if !user.is_admin { + return Err(utopia_core::AppError::Forbidden.into()); + } + let requeued = utopia_store::jobs::requeue_failed( + &state.pool, + RequeueScope { + kb_id: None, + kind: body.kind.as_deref(), + failed_since: body.failed_since, + }, + ) + .await?; + let _ = utopia_store::audit::record( + &state.pool, + None, + user.id, + "jobs.requeued", + "system", + None, + json!({ "requeued": requeued, "kind": body.kind, "failed_since": body.failed_since }), + ) + .await; + Ok(Json(json!({ "requeued": requeued }))) +} diff --git a/crates/utopia-server/src/api/mod.rs b/crates/utopia-server/src/api/mod.rs index aa5ff762..2b0ee313 100644 --- a/crates/utopia-server/src/api/mod.rs +++ b/crates/utopia-server/src/api/mod.rs @@ -6,6 +6,7 @@ mod datasource_routes; mod documents_routes; mod events_routes; mod graph_routes; +mod jobs_routes; mod kbs; mod mapping_routes; mod mcp; @@ -86,6 +87,10 @@ pub fn router(state: AppState, cfg: &AppConfig) -> Router { ) .route("/kbs/{id}/members", get(kbs::members)) .route("/kbs/{id}/audit", get(kbs::audit_log)) + // 失败任务回队列（#216）：库内给 Editor，全局给管理员 + .route("/kbs/{id}/jobs/failed", get(jobs_routes::failed_in_kb)) + .route("/kbs/{id}/jobs/requeue", post(jobs_routes::requeue_in_kb)) + .route("/jobs/requeue", post(jobs_routes::requeue_all)) .route( "/kbs/{id}/members/{user_id}", axum::routing::put(kbs::set_member).delete(kbs::remove_member), diff --git a/crates/utopia-store/src/jobs.rs b/crates/utopia-store/src/jobs.rs index 0a9d689f..c86a7cbd 100644 --- a/crates/utopia-store/src/jobs.rs +++ b/crates/utopia-store/src/jobs.rs @@ -4,11 +4,13 @@ //! 并发消费：调度循环按"运行中 < 目标数"续派，任务在独立 task 执行； //! 目标数经 AtomicUsize 热读——系统设置里改并发即时生效，无需重启。 +use chrono::{DateTime, Utc}; use sqlx::PgPool; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; use std::time::Duration; use utopia_core::AppResult; +use uuid::Uuid; #[derive(Debug, Clone, sqlx::FromRow)] pub struct Job { @@ -29,6 +31,61 @@ pub async fn enqueue(pool: &PgPool, kind: &str, payload: serde_json::Value) -> A Ok(id) } +/// 重排失败任务的范围（#216）。三个条件都可空，空 = 不限。 +/// +/// **按库圈要解 payload**：任务表没有 kb 列，payload 只带 `document_id` / +/// `source_id` / `kb_id` 三种之一，各自解到库。没有库的系统任务只在不限库时才动 +#[derive(Debug, Default, Clone, Copy)] +pub struct RequeueScope<'a> { + pub kb_id: Option, + pub kind: Option<&'a str>, + /// 只排这个时刻之后失败的——告警上的「再跑一遍」圈的正是那次故障窗口 + pub failed_since: Option>, +} + +/// 库范围的 SQL 谓词，`$N` 是库 id；`requeue_failed` 与 `failed_count` 共用 +const KB_SCOPE: &str = "( + (j.payload ? 'kb_id' AND j.payload->>'kb_id' = $KB::text) + OR (j.payload ? 'document_id' AND EXISTS ( + SELECT 1 FROM documents d + WHERE d.id::text = j.payload->>'document_id' AND d.kb_id = $KB)) + OR (j.payload ? 'source_id' AND EXISTS ( + SELECT 1 FROM sources s + WHERE s.id::text = j.payload->>'source_id' AND s.kb_id = $KB)))"; + +/// 把范围内的 failed 任务放回队列：`attempts` 归零、立即到期。 +/// +/// 处理器都是幂等的（启动时回收孤儿就靠这一点），所以重排永远安全； +/// 此前 `failed` 是终点，余额耗尽一批文档全失败，充值之后只能逐个点或整源重抽 +pub async fn requeue_failed(pool: &PgPool, scope: RequeueScope<'_>) -> AppResult { + let sql = format!( + "UPDATE jobs j + SET status = 'queued', attempts = 0, run_at = now(), updated_at = now() + WHERE j.status = 'failed' + AND ($1::text IS NULL OR j.kind = $1) + AND ($2::timestamptz IS NULL OR j.updated_at >= $2) + AND ($3::uuid IS NULL OR {})", + KB_SCOPE.replace("$KB", "$3") + ); + let res = sqlx::query(&sql) + .bind(scope.kind) + .bind(scope.failed_since) + .bind(scope.kb_id) + .execute(pool) + .await?; + Ok(res.rows_affected()) +} + +/// 范围内 failed 的条数——设置页那一行「N 个失败任务」 +pub async fn failed_count(pool: &PgPool, kb_id: Option) -> AppResult { + let sql = format!( + "SELECT count(*) FROM jobs j + WHERE j.status = 'failed' AND ($1::uuid IS NULL OR {})", + KB_SCOPE.replace("$KB", "$1") + ); + Ok(sqlx::query_scalar(&sql).bind(kb_id).fetch_one(pool).await?) +} + /// 认领一个到期任务；没有则返回 None。 async fn claim_one(pool: &PgPool) -> AppResult> { let job = sqlx::query_as( diff --git a/crates/utopia-store/tests/a_failed_job_finds_its_way_back.rs b/crates/utopia-store/tests/a_failed_job_finds_its_way_back.rs new file mode 100644 index 00000000..47fa9c01 --- /dev/null +++ b/crates/utopia-store/tests/a_failed_job_finds_its_way_back.rs @@ -0,0 +1,255 @@ +//! #216：失败的任务能回到队列。 +//! +//! 此前 `failed` 就是终点——余额耗尽一批文档全失败，充值之后只能逐个点，或整源重抽 +//! （已成功的也重跑一遍模型）。`jobs::requeue_failed` 按范围重排：库、种类、失败时间 +//! 三个条件都可空。这里守三件事： +//! +//! 1. **按库圈得准。** 任务的 payload 只带 `document_id` / `source_id` / `kb_id` 三种 +//! 之一，按库重排要解到库——别的库的一条都不碰，没有库的（系统任务）只有不限库时才动。 +//! 2. **只动 failed 的**，`done` 与 `queued` 不动；重排后 `attempts` 归零、立即到期。 +//! 3. **种类与时间窗**：`kind` 只排那一种；`failed_since` 只排那之后失败的—— +//! 告警上的「再跑一遍」圈的正是那次故障窗口里的任务。 +//! +//! 没有 `UTOPIA_DATABASE_URL` 时跳过而不是失败。自建自拆，绝不碰已有的库。 + +use chrono::{Duration, Utc}; +use sqlx::PgPool; +use utopia_store::jobs::{self, RequeueScope}; +use uuid::Uuid; + +struct Fx { + org: Uuid, + kb1: Uuid, + kb2: Uuid, + doc1: Uuid, + src2: Uuid, +} + +async fn seed(pool: &PgPool) -> anyhow::Result { + let (org, ws, kb1, kb2) = ( + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + ); + let (src1, doc1, src2) = (Uuid::now_v7(), Uuid::now_v7(), Uuid::now_v7()); + sqlx::query("INSERT INTO organizations (id, name) VALUES ($1, 'requeue-test')") + .bind(org) + .execute(pool) + .await?; + sqlx::query("INSERT INTO workspaces (id, org_id, name) VALUES ($1, $2, 'requeue-test')") + .bind(ws) + .bind(org) + .execute(pool) + .await?; + for kb in [kb1, kb2] { + sqlx::query( + "INSERT INTO knowledge_bases (id, workspace_id, name) VALUES ($1, $2, 'requeue-test')", + ) + .bind(kb) + .bind(ws) + .execute(pool) + .await?; + } + sqlx::query("INSERT INTO sources (id, kb_id, kind, name) VALUES ($1, $2, 'folder', 'f')") + .bind(src1) + .bind(kb1) + .execute(pool) + .await?; + sqlx::query( + "INSERT INTO documents (id, kb_id, source_id, filename, sha256, status) + VALUES ($1, $2, $3, 'a.md', 'requeue', 'ready')", + ) + .bind(doc1) + .bind(kb1) + .bind(src1) + .execute(pool) + .await?; + sqlx::query("INSERT INTO sources (id, kb_id, kind, name) VALUES ($1, $2, 'url', 'u')") + .bind(src2) + .bind(kb2) + .execute(pool) + .await?; + Ok(Fx { + org, + kb1, + kb2, + doc1, + src2, + }) +} + +async fn job( + pool: &PgPool, + kind: &str, + payload: serde_json::Value, + status: &str, + failed_ago: Duration, +) -> anyhow::Result { + let (id,): (i64,) = sqlx::query_as( + "INSERT INTO jobs (kind, payload, status, attempts, last_error, updated_at) + VALUES ($1, $2, $3, 3, 'out of credit', $4) RETURNING id", + ) + .bind(kind) + .bind(payload) + .bind(status) + .bind(Utc::now() - failed_ago) + .fetch_one(pool) + .await?; + Ok(id) +} + +async fn state(pool: &PgPool, id: i64) -> anyhow::Result<(String, i32)> { + Ok( + sqlx::query_as("SELECT status, attempts FROM jobs WHERE id = $1") + .bind(id) + .fetch_one(pool) + .await?, + ) +} + +#[tokio::test] +async fn a_failed_job_finds_its_way_back() -> anyhow::Result<()> { + let Some(url) = utopia_store::test_db::url() else { + return Ok(()); + }; + let pool = PgPool::connect(&url).await?; + let f = seed(&pool).await?; + let mut ids: Vec = Vec::new(); + + let run = async { + let m = Duration::minutes(1); + // kb1：文档任务、库级任务、一条很早就失败的、一条已经成功的 + let d1 = job( + &pool, + "process_document", + serde_json::json!({ "document_id": f.doc1 }), + "failed", + m, + ) + .await?; + let b1 = job( + &pool, + "bootstrap_ontology", + serde_json::json!({ "kb_id": f.kb1 }), + "failed", + m, + ) + .await?; + let old1 = job( + &pool, + "process_document", + serde_json::json!({ "document_id": f.doc1 }), + "failed", + Duration::days(3), + ) + .await?; + let done1 = job( + &pool, + "process_document", + serde_json::json!({ "document_id": f.doc1 }), + "done", + m, + ) + .await?; + // kb2：来源任务 + let s2 = job( + &pool, + "sync_source", + serde_json::json!({ "source_id": f.src2 }), + "failed", + m, + ) + .await?; + // 没有库的系统任务 + let sys = job(&pool, "noop", serde_json::json!({}), "failed", m).await?; + ids.extend([d1, b1, old1, done1, s2, sys]); + + assert_eq!(jobs::failed_count(&pool, Some(f.kb1)).await?, 3); + assert_eq!(jobs::failed_count(&pool, Some(f.kb2)).await?, 1); + + // 3. 时间窗：只排最近一小时里失败的 kb1 任务 + let n = jobs::requeue_failed( + &pool, + RequeueScope { + kb_id: Some(f.kb1), + kind: None, + failed_since: Some(Utc::now() - Duration::hours(1)), + }, + ) + .await?; + assert_eq!( + n, 2, + "the document job and the base-level job, not the old one" + ); + assert_eq!(state(&pool, d1).await?, ("queued".into(), 0)); + assert_eq!(state(&pool, b1).await?, ("queued".into(), 0)); + assert_eq!(state(&pool, old1).await?.0, "failed"); + assert_eq!(state(&pool, done1).await?.0, "done", "done stays done"); + assert_eq!( + state(&pool, s2).await?.0, + "failed", + "another base is untouched" + ); + assert_eq!( + state(&pool, sys).await?.0, + "failed", + "a system job has no base" + ); + + // 3. 种类：kb1 里只排 process_document，老的那条这次回来 + let n = jobs::requeue_failed( + &pool, + RequeueScope { + kb_id: Some(f.kb1), + kind: Some("process_document"), + failed_since: None, + }, + ) + .await?; + assert_eq!(n, 1); + assert_eq!(state(&pool, old1).await?, ("queued".into(), 0)); + assert_eq!(jobs::failed_count(&pool, Some(f.kb1)).await?, 0); + + // 1. 按库：kb2 的来源任务 + let n = jobs::requeue_failed( + &pool, + RequeueScope { + kb_id: Some(f.kb2), + kind: None, + failed_since: None, + }, + ) + .await?; + assert_eq!(n, 1); + assert_eq!(state(&pool, s2).await?, ("queued".into(), 0)); + + // 不限库才动系统任务 + let before = jobs::failed_count(&pool, None).await?; + let n = jobs::requeue_failed( + &pool, + RequeueScope { + kb_id: None, + kind: Some("noop"), + failed_since: Some(Utc::now() - Duration::hours(1)), + }, + ) + .await?; + assert!(n >= 1); + assert_eq!(state(&pool, sys).await?.0, "queued"); + assert!(jobs::failed_count(&pool, None).await? < before); + Ok::<_, anyhow::Error>(()) + } + .await; + + // 自拆：任务表不随组织级联，手动清 + let _ = sqlx::query("DELETE FROM jobs WHERE id = ANY($1)") + .bind(&ids) + .execute(&pool) + .await; + let _ = sqlx::query("DELETE FROM organizations WHERE id = $1") + .bind(f.org) + .execute(&pool) + .await; + run +} diff --git a/docs/decisions/0005-alert-center.md b/docs/decisions/0005-alert-center.md index b55c0162..9fed687c 100644 --- a/docs/decisions/0005-alert-center.md +++ b/docs/decisions/0005-alert-center.md @@ -119,6 +119,7 @@ channel: something in this document did not land. None of it goes to Review. too narrow. - `llm.unreachable` first matched only transport failures, so the commonest fault (a wrong URL, a proxy answering HTML) produced no alert at all. +- 2026-09-03: an alert can carry the action that closes its loop (#216). `llm.out_of_credit` and `llm.unreachable` groups offer "Run those again", which puts the jobs that failed in that group's time window back in the queue (`jobs::requeue_failed`, scoped to the group's knowledge base, or to everything for a system alert — admins only). The KB settings page shows the failed count with the same action for every other kind of failure. A queue page was not built: the one failure where the operator does something specific and then wants the work to resume is the one that reports through an alert. ## Open questions diff --git a/web/src/api.ts b/web/src/api.ts index 8161099a..71263e1e 100644 --- a/web/src/api.ts +++ b/web/src/api.ts @@ -983,6 +983,17 @@ export const api = { ); }, alertsUnread: () => request<{ unread: number }>("/api/v1/alerts/unread"), + /** 失败任务回队列（#216）。库内一条、全局一条（管理员）；范围可按种类与失败时间收窄 */ + failedJobs: (kbId: string) => + request<{ failed: number }>(`/api/v1/kbs/${kbId}/jobs/failed`), + requeueJobs: ( + kbId: string | null, + body: { kind?: string; failed_since?: string } = {}, + ) => + request<{ requeued: number }>( + kbId ? `/api/v1/kbs/${kbId}/jobs/requeue` : "/api/v1/jobs/requeue", + { method: "POST", body: JSON.stringify(body) }, + ), alertReadGroup: (g: { kb_id: string | null; kind: string; diff --git a/web/src/i18n/en.ts b/web/src/i18n/en.ts index c430f17e..d7093d62 100644 --- a/web/src/i18n/en.ts +++ b/web/src/i18n/en.ts @@ -220,6 +220,10 @@ export const en = { } as Record, // 没见过的 kind 也要能显示：新告警源上线时前端可能还没更新 unknownKind: (kind: string) => kind, + /** 修好之后接着跑（#216） */ + runAgain: "Run those again", + requeued: (n: number) => + n === 1 ? "1 job back in the queue" : `${n} jobs back in the queue`, }, kbScope: { @@ -1466,6 +1470,10 @@ export const en = { inferEvery: "Re-derive every", minutes: "minutes", lastInference: (when: string) => `last run ${when}`, + failedJobs: (n: number) => (n === 1 ? "1 failed job" : `${n} failed jobs`), + requeue: "Run again", + requeued: (n: number) => + n === 1 ? "1 job back in the queue" : `${n} jobs back in the queue`, /* 语料语言。措辞要把"这不是界面语言"讲清楚，否则一定有人当成界面开关 */ ontologyLang: "Language of this ontology", ontologyLangNote: diff --git a/web/src/i18n/zh.ts b/web/src/i18n/zh.ts index afa26cd4..bcfc4eea 100644 --- a/web/src/i18n/zh.ts +++ b/web/src/i18n/zh.ts @@ -198,6 +198,8 @@ export const zh: Strings = { }, } as Record, unknownKind: (kind: string) => kind, + runAgain: "再跑一遍", + requeued: (n: number) => `${n} 个任务回到队列`, }, kbScope: { @@ -1319,6 +1321,9 @@ export const zh: Strings = { inferEvery: "每隔", minutes: "分钟重推", lastInference: (when: string) => `上次 ${when}`, + failedJobs: (n: number) => `${n} 个失败的任务`, + requeue: "再跑一遍", + requeued: (n: number) => `${n} 个任务回到队列`, ontologyLang: "本体的语言", ontologyLangNote: "类与关系的描述用哪种语言写。它们会被原样送进抽取提示词，" + diff --git a/web/src/pages/AlertBell.tsx b/web/src/pages/AlertBell.tsx index a8428551..eceaeb93 100644 --- a/web/src/pages/AlertBell.tsx +++ b/web/src/pages/AlertBell.tsx @@ -11,6 +11,7 @@ import { Bell, Search, X } from "lucide-react"; import { api, type AlertGroup } from "../api"; import { S } from "../i18n"; +import { toast } from "../toast"; import { Chip, Pager, cn } from "../ui"; import { usePopoverFlip } from "../ui/popoverFlip"; @@ -22,12 +23,19 @@ function line(d: AlertGroup["lines"][number]): string | null { return parts.length ? parts.join(" — ") : null; } +/** 哪些告警带「再跑一遍」：故障修好之后（充值、改端点）任务不会自己回来的那几种 */ +const REQUEUE_KINDS = new Set(["llm.out_of_credit", "llm.unreachable"]); + function AlertRow({ g, onRead, + onRequeue, + requeuing, }: { g: AlertGroup; onRead: (g: AlertGroup) => void; + onRequeue: (g: AlertGroup) => void; + requeuing: boolean; }) { // 没见过的 kind 也得显示得出来：新告警源上线时前端可能还没跟上， // 而"有条告警但我不认识它"远好过"什么都不显示" @@ -36,14 +44,19 @@ function AlertRow({ // count 数的是整组，lines 只带回前几条——差额是"还有 N 条" const rest = g.count - lines.length; return ( - + )} - + ); } @@ -126,6 +155,16 @@ function Panel({ panelRef }: { panelRef: Ref }) { mutationFn: () => api.alertsReadAll(), onSuccess: () => qc.invalidateQueries({ queryKey: ["alerts"] }), }); + // 时间窗从这组最早那次故障起——之前失败的不是这次的事 + const requeue = useMutation({ + mutationFn: (g: AlertGroup) => + api.requeueJobs(g.kb_id, { failed_since: g.earliest_at }), + onSuccess: (r) => { + toast.success(S.alerts.requeued(r.requeued)); + qc.invalidateQueries({ queryKey: ["jobs"] }); + }, + onError: (e) => toast.error(String(e)), + }); const groups = list.data?.items ?? []; const total = list.data?.total ?? 0; @@ -185,6 +224,8 @@ function Panel({ panelRef }: { panelRef: Ref }) { key={`${g.kb_id ?? "system"}|${g.kind}|${g.latest_at}`} g={g} onRead={(x) => read.mutate(x)} + onRequeue={(x) => requeue.mutate(x)} + requeuing={requeue.isPending} /> )) )} diff --git a/web/src/pages/KbSettings.tsx b/web/src/pages/KbSettings.tsx index d6eccbbf..ab658223 100644 --- a/web/src/pages/KbSettings.tsx +++ b/web/src/pages/KbSettings.tsx @@ -14,6 +14,7 @@ import { } from "lucide-react"; import { api, type AuditEvent } from "../api"; import { LANG_NAMES, S } from "../i18n"; +import { toast } from "../toast"; import { DangerConfirm, Dropdown, @@ -52,6 +53,20 @@ export function KbSettings() { 带着库走的地方，现在整片都在 /kb/$kbId 之下，它就不必自成一格了 */ const { kbId } = useParams({ from: "/app/kb/$kbId/settings" }); + // 失败任务数与重排（#216）。查询键带库 id，重排后失效重取 + const failedJobs = useQuery({ + queryKey: ["jobs", "failed", kbId], + queryFn: () => api.failedJobs(kbId!), + enabled: !!kbId, + }); + const requeue = useMutation({ + mutationFn: () => api.requeueJobs(kbId!), + onSuccess: (r) => { + toast.success(S.kbset.requeued(r.requeued)); + queryClient.invalidateQueries({ queryKey: ["jobs", "failed", kbId] }); + }, + onError: (e) => toast.error(String(e)), + }); const kb = useQuery({ queryKey: ["kbOne", kbId], queryFn: () => api.kbDetail(kbId!), @@ -296,6 +311,21 @@ export function KbSettings() { )} )} + {/* 失败的任务（#216）：有才露出来。「再跑一遍」把这个库里全部 failed 放回队列 */} + {failedJobs.data && failedJobs.data.failed > 0 && ( +
+ + {S.kbset.failedJobs(failedJobs.data.failed)} + + +
+ )} {/* 语料语言。**不是界面语言**——类描述逐字进抽取提示词， 读者是正在读这些文档的模型，所以它跟文档走不跟读者走 */}