diff --git a/crates/utopia-server/src/api/jobs_routes.rs b/crates/utopia-server/src/api/jobs_routes.rs index 6e459b5a5..8a8cf4eda 100644 --- a/crates/utopia-server/src/api/jobs_routes.rs +++ b/crates/utopia-server/src/api/jobs_routes.rs @@ -51,6 +51,21 @@ pub async fn failed_in_kb( Ok(Json(json!({ "failed": failed }))) } +/// 一个任务跑完了没（0051）。人定一条短语签名时拿到的是 job id 而不是结果， +/// 结果要么从 `review` / `graph` 事件里等到，要么来这里问。Viewer 就能问： +/// 任务属于这个库才答，否则 404，与库里看不见的东西一个口径 +pub async fn job_in_kb( + State(state): State, + AuthUser(user): AuthUser, + Path((kb_id, job_id)): Path<(Uuid, i64)>, +) -> ApiResult> { + require_kb(&state, &user, kb_id, Role::Viewer).await?; + let job = utopia_store::jobs::status_in_kb(&state.pool, kb_id, job_id) + .await? + .ok_or(utopia_core::AppError::NotFound)?; + Ok(Json(json!({ "job": job }))) +} + pub async fn requeue_in_kb( State(state): State, AuthUser(user): AuthUser, diff --git a/crates/utopia-server/src/api/mod.rs b/crates/utopia-server/src/api/mod.rs index 53e7c1b38..c7536ae97 100644 --- a/crates/utopia-server/src/api/mod.rs +++ b/crates/utopia-server/src/api/mod.rs @@ -174,6 +174,8 @@ pub fn router(state: AppState, cfg: &AppConfig) -> Router { .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)) + // 一个任务的状态（0051）：人定完短语签名拿到 job id 后来这里问跑完没 + .route("/kbs/{id}/jobs/{job_id}", get(jobs_routes::job_in_kb)) .route( "/kbs/{id}/members/{user_id}", axum::routing::put(kbs::set_member).delete(kbs::remove_member), diff --git a/crates/utopia-server/src/api/review_routes.rs b/crates/utopia-server/src/api/review_routes.rs index 9a62f3bb5..5de8adb18 100644 --- a/crates/utopia-server/src/api/review_routes.rs +++ b/crates/utopia-server/src/api/review_routes.rs @@ -12,6 +12,10 @@ use crate::auth::AuthUser; use crate::error::ApiResult; use crate::state::AppState; +#[cfg(test)] +#[path = "review_routes_phrase_tests.rs"] +mod phrase_tests; + /// 一页多少条。**服务端的默认，不是上限**——前端可以要更少，多则被 clamp 挡住 const REVIEW_PAGE: i64 = 10; @@ -1228,7 +1232,7 @@ pub async fn decide_alignment_phrase( AuthUser(user): AuthUser, Path((kb_id, binding_id)): Path<(Uuid, Uuid)>, Json(req): Json, -) -> ApiResult> { +) -> ApiResult<(axum::http::StatusCode, Json)> { require_kb(&state, &user, kb_id, Role::Editor).await?; let sig = utopia_store::phrase_bindings::signature_of(&state.pool, kb_id, binding_id) .await? @@ -1264,7 +1268,10 @@ pub async fn decide_alignment_phrase( } }; let votes = json!({ "person": { "property": req.property, "direction": direction } }); - utopia_store::phrase_bindings::decide( + // 判定和它的重算任务一次提交（0051）。这里**不再**同步重算：等物化锁占的是池里的 + // 连接，而正在跑的那次对齐可能已经读完最后一遍，谁也不替这条判定投影。一个 job + // 只在判定提交后可见，worker 读的是当前绑定；屏幕上等的是 `review` / `graph` 事件 + let job_id = utopia_store::phrase_bindings::decide_with_delivery( &state.pool, kb_id, &sig, @@ -1276,8 +1283,8 @@ pub async fn decide_alignment_phrase( decided_by: "person", }, ) - .await?; - let typed = utopia_store::materialize::materialize_human(&state.pool, kb_id).await?; + .await? + .ok_or_else(|| utopia_core::AppError::Conflict("the decision was not written".into()))?; let _ = utopia_store::audit::record( &state.pool, Some(kb_id), @@ -1286,13 +1293,14 @@ pub async fn decide_alignment_phrase( "phrase_binding", Some(binding_id), json!({ "phrase": sig.phrase, "property": req.property, "direction": direction, - "typed_added": typed.added, "typed_retired": typed.retired }), + "job_id": job_id }), ) .await; state.emit_review(kb_id); - state.emit_graph(kb_id); - Ok(Json( - json!({ "ok": true, "typed": { "added": typed.added, "merged": typed.merged, "retired": typed.retired } }), + // 202：收下了，投影在路上。不编一个 typed: {added: 0} 出来——那不是这次请求知道的事 + Ok(( + axum::http::StatusCode::ACCEPTED, + Json(json!({ "ok": true, "job_id": job_id, "status": "accepted" })), )) } diff --git a/crates/utopia-server/src/api/review_routes_phrase_tests.rs b/crates/utopia-server/src/api/review_routes_phrase_tests.rs new file mode 100644 index 000000000..a6cfb1a88 --- /dev/null +++ b/crates/utopia-server/src/api/review_routes_phrase_tests.rs @@ -0,0 +1,246 @@ +//! 人定一条短语签名：判定和它的重算任务一次提交，请求答 202 和 job id，不编数字（0051）。 +//! +//! 三件事：路由答 202、job 排着、绑定写成人判的；`GET /kbs/{id}/jobs/{job_id}` 在本库能读、 +//! 换个库答 404；Viewer 能读状态但不能判。没有 `UTOPIA_DATABASE_URL` 时跳过。 +use axum::body::{to_bytes, Body}; +use axum::http::{Request, StatusCode}; +use serde_json::{json, Value}; +use std::sync::Arc; +use tower::ServiceExt; +use utopia_store::phrase_bindings; +use uuid::Uuid; + +struct Fx { + pool: sqlx::PgPool, + app: axum::Router, + org: Uuid, + kb: Uuid, + other_kb: Uuid, + editor: String, + viewer: String, + binding: Uuid, + _dir: tempfile::TempDir, +} + +impl Fx { + async fn new() -> anyhow::Result> { + let Some(url) = utopia_store::test_db::url() else { + return Ok(None); + }; + let pool = sqlx::PgPool::connect(&url).await?; + utopia_store::db::migrate(&pool).await?; + let (org, ws, kb, other_kb, editor, viewer, subject, object, property, statement) = ( + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + ); + // Only locally generated UUIDs are interpolated into fixture SQL. + sqlx::raw_sql(&format!( + "INSERT INTO organizations(id,name) VALUES ('{org}','phrase-route-test'); + INSERT INTO workspaces(id,org_id,name) VALUES ('{ws}','{org}','phrase-route-test'); + INSERT INTO users(id,org_id,email,display_name,password_hash) VALUES + ('{editor}','{org}','{editor}@phrase.test','editor','unused'), + ('{viewer}','{org}','{viewer}@phrase.test','viewer','unused'); + INSERT INTO knowledge_bases(id,workspace_id,name) VALUES + ('{kb}','{ws}','phrases'), ('{other_kb}','{ws}','other'); + INSERT INTO kb_members(kb_id,user_id,role) VALUES + ('{kb}','{editor}','editor'), ('{kb}','{viewer}','viewer'), + ('{other_kb}','{editor}','editor'); + INSERT INTO entities(id,kb_id,canonical_name) VALUES + ('{subject}','{kb}','Acme'), ('{object}','{kb}','London'); + INSERT INTO relation_types(id,kb_id,key,label,temporal) VALUES + ('{property}','{kb}','based_in','based in','state'); + INSERT INTO facts(id,kb_id,subject_id,object_id,layer,phrase) VALUES + ('{statement}','{kb}','{subject}','{object}','open','based in');" + )) + .execute(&pool) + .await?; + // 队列里的一条：代理判成 undecided，人来定 + let sig = phrase_bindings::signatures(&pool, kb).await?.remove(0); + phrase_bindings::decide( + &pool, + kb, + &sig, + phrase_bindings::Decision { + relation_type_id: None, + direction: None, + status: "undecided", + votes: &json!({}), + decided_by: "agent", + }, + ) + .await?; + let binding: Uuid = sqlx::query_scalar("SELECT id FROM phrase_bindings WHERE kb_id=$1") + .bind(kb) + .fetch_one(&pool) + .await?; + let dir = tempfile::tempdir()?; + let cfg = utopia_core::config::AppConfig { + data_dir: dir.path().to_string_lossy().into_owned(), + ..Default::default() + }; + let search = Arc::new(utopia_search::SearchIndex::open( + &dir.path().join("search"), + )?); + let state = crate::state::AppState::new(pool.clone(), &cfg, search, "test-only".into()); + let editor_token = crate::auth::issue_token(&state, editor)?; + let viewer_token = crate::auth::issue_token(&state, viewer)?; + let app = super::super::router(state, &cfg); + Ok(Some(Self { + pool, + app, + org, + kb, + other_kb, + editor: editor_token, + viewer: viewer_token, + binding, + _dir: dir, + })) + } + + async fn call( + &self, + token: &str, + method: &str, + path: &str, + body: Option, + ) -> anyhow::Result<(StatusCode, Value)> { + let request = Request::builder() + .method(method) + .uri(path) + .header("Authorization", format!("Bearer {token}")); + let request = match body { + Some(b) => request + .header("Content-Type", "application/json") + .body(Body::from(b.to_string()))?, + None => request.body(Body::empty())?, + }; + let response = self.app.clone().oneshot(request).await?; + let status = response.status(); + let bytes = to_bytes(response.into_body(), usize::MAX).await?; + let value = if bytes.is_empty() { + Value::Null + } else { + serde_json::from_slice(&bytes)? + }; + Ok((status, value)) + } + + async fn cleanup(self) -> anyhow::Result<()> { + sqlx::query("DELETE FROM jobs WHERE payload->>'kb_id'=$1") + .bind(self.kb.to_string()) + .execute(&self.pool) + .await?; + sqlx::query("DELETE FROM organizations WHERE id=$1") + .bind(self.org) + .execute(&self.pool) + .await?; + Ok(()) + } +} + +#[tokio::test] +async fn a_phrase_decision_is_accepted_with_its_job() -> anyhow::Result<()> { + let Some(f) = Fx::new().await? else { + return Ok(()); + }; + let run = async { + let path = format!( + "/api/v1/kbs/{}/review/alignment/phrases/{}", + f.kb, f.binding + ); + let (status, body) = f + .call( + &f.editor, + "POST", + &path, + Some(json!({ "property": "based_in", "direction": "forward" })), + ) + .await?; + assert_eq!(status, StatusCode::ACCEPTED, "{body}"); + assert_eq!(body["ok"], json!(true)); + assert_eq!(body["status"], json!("accepted")); + assert!(body.get("typed").is_none(), "no invented counts: {body}"); + let job_id = body["job_id"].as_i64().expect("job id"); + + // job 排着，载荷是这个库；绑定是人判的、bound + let (kind, job_status, payload): (String, String, Value) = + sqlx::query_as("SELECT kind, status, payload FROM jobs WHERE id=$1") + .bind(job_id) + .fetch_one(&f.pool) + .await?; + assert_eq!(kind, phrase_bindings::MATERIALIZE_KIND); + assert_eq!(job_status, "queued"); + assert_eq!(payload["kb_id"], json!(f.kb)); + let b = phrase_bindings::bindings(&f.pool, f.kb).await?.remove(0); + assert_eq!( + (b.status.as_str(), b.decided_by.as_str()), + ("bound", "person") + ); + // 这次请求没有重算：投影要等 job + assert_eq!(utopia_store::materialize::count(&f.pool, f.kb).await?, 0); + + // 状态读：本库 200，换库 404，Viewer 也能读 + let (status, body) = f + .call( + &f.editor, + "GET", + &format!("/api/v1/kbs/{}/jobs/{job_id}", f.kb), + None, + ) + .await?; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(body["job"]["status"], json!("queued")); + assert_eq!( + body["job"]["kind"], + json!(phrase_bindings::MATERIALIZE_KIND) + ); + let (status, _) = f + .call( + &f.editor, + "GET", + &format!("/api/v1/kbs/{}/jobs/{job_id}", f.other_kb), + None, + ) + .await?; + assert_eq!(status, StatusCode::NOT_FOUND); + let (status, _) = f + .call( + &f.viewer, + "GET", + &format!("/api/v1/kbs/{}/jobs/{job_id}", f.kb), + None, + ) + .await?; + assert_eq!(status, StatusCode::OK); + + // Viewer 不能判 + let (status, _) = f + .call( + &f.viewer, + "POST", + &path, + Some(json!({ "property": null, "direction": "forward" })), + ) + .await?; + assert_eq!(status, StatusCode::FORBIDDEN); + + // 让 job 该做的事在这里做一遍：读当前绑定，算出一条类型化行 + let outcome = utopia_store::materialize::try_materialize(&f.pool, f.kb) + .await? + .expect("nobody holds the lock"); + assert_eq!(outcome.added, 1); + anyhow::Ok(()) + } + .await; + f.cleanup().await?; + run +} diff --git a/crates/utopia-server/src/main.rs b/crates/utopia-server/src/main.rs index 6c7e585fd..f02c69ce2 100644 --- a/crates/utopia-server/src/main.rs +++ b/crates/utopia-server/src/main.rs @@ -498,6 +498,45 @@ async fn dispatch(st: &state::AppState, job: &utopia_store::jobs::Job) -> anyhow } // 时间提及按文档解析（0045）：抽完一篇排一个，重排一次就是重新解析 // 类别词绑到类（0044 对齐的第一片）：库级任务，抽完一篇排一个，本体改了再排 + // 人定了一条短语签名，随判定同事务排下的重算（0051）。试锁不等：拿不到就 + // 挂 `Deferred` 十秒后再来，占着连接排队的是别人的池子；拿到了就是一次完整 + // 的按绑定重算，读的是当前绑定而不是判定时的载荷 + utopia_store::phrase_bindings::MATERIALIZE_KIND => { + let kb_id: Uuid = job + .payload + .get("kb_id") + .and_then(|v| v.as_str()) + .and_then(|s| s.parse().ok()) + .ok_or_else(|| anyhow::anyhow!("payload 缺少 kb_id"))?; + match utopia_store::materialize::try_materialize(&st.pool, kb_id).await? { + Some(typed) => { + if typed.added > 0 || typed.merged > 0 || typed.retired > 0 { + let _ = utopia_store::audit::record( + &st.pool, + Some(kb_id), + Uuid::nil(), + "alignment.materialized", + "knowledge_base", + Some(kb_id), + serde_json::json!({ + "job_id": job.id, + "added": typed.added, + "merged": typed.merged, + "retired": typed.retired, + }), + ) + .await; + st.emit_graph(kb_id); + } + // 队列卡片按绑定的状态显示，重算完了才算这条判定「落地」 + st.emit_review(kb_id); + Ok(()) + } + None => Err(anyhow::anyhow!("typed projection busy").context( + utopia_core::Deferred::new(std::time::Duration::from_secs(10)), + )), + } + } "align_types" => { let kb_id: Uuid = job .payload diff --git a/crates/utopia-store/src/jobs.rs b/crates/utopia-store/src/jobs.rs index 653abab69..1cb9db598 100644 --- a/crates/utopia-store/src/jobs.rs +++ b/crates/utopia-store/src/jobs.rs @@ -259,6 +259,34 @@ pub async fn failed_count(pool: &PgPool, kb_id: Option) -> AppResult Ok(sqlx::query_scalar(&sql).bind(kb_id).fetch_one(pool).await?) } +/// 一个任务此刻的样子，给「我刚排下去的那件事跑完了没」这个问题用（0051）。 +#[derive(Debug, Clone, serde::Serialize, sqlx::FromRow)] +pub struct JobStatus { + pub id: i64, + pub kind: String, + pub status: String, + pub attempts: i32, + pub max_attempts: i32, + pub last_error: Option, + pub run_at: chrono::DateTime, + pub updated_at: chrono::DateTime, +} + +/// 按 id 读一个任务，**且只在它属于这个库时**。授权跟着库走：能看这个库的人能看 +/// 它的任务；别的库的任务 id 猜对了也只得到 None，与看不见的文档一样答 404。 +pub async fn status_in_kb(pool: &PgPool, kb_id: Uuid, id: i64) -> AppResult> { + let sql = format!( + "SELECT j.id, j.kind, j.status, j.attempts, j.max_attempts, j.last_error, j.run_at, j.updated_at + FROM jobs j WHERE j.id = $1 AND {}", + KB_SCOPE.replace("$KB", "$2") + ); + Ok(sqlx::query_as(&sql) + .bind(id) + .bind(kb_id) + .fetch_optional(pool) + .await?) +} + /// 认领一个到期任务；没有则返回 None。 async fn claim_one(pool: &PgPool) -> AppResult> { let job = sqlx::query_as( diff --git a/crates/utopia-store/src/materialize.rs b/crates/utopia-store/src/materialize.rs index f8044a257..08337774e 100644 --- a/crates/utopia-store/src/materialize.rs +++ b/crates/utopia-store/src/materialize.rs @@ -14,7 +14,7 @@ //! 没有」的（陈述, 绑定）对补上——有同断言的行就并进去，没有才新建。没有模型调用。 use sqlx::PgPool; -use utopia_core::{AppError, AppResult}; +use utopia_core::AppResult; use uuid::Uuid; use crate::graph::{insert_fact_on, FactObject, Validity}; @@ -74,55 +74,29 @@ pub async fn materialize(pool: &PgPool, kb_id: Uuid) -> AppResult { Ok(outcome) } -/// 人的入口（#798 跟进、#800 复盘）：与 `materialize` 同样的活，外加一个 -/// 事务级的 `lock_timeout` 预算，超时映射到表「语言可读的重试提示」。 +/// 任务里的入口（0051）：**试锁，不等**。拿到 `typed_materialize` 就在这条连接上 +/// 跑完整的重算并提交；拿不到就回滚、返回 `None`，由调用方挂成 `Deferred` 稍后再来。 /// -/// 设计要点（沿用 #828 的同一条理由）： -/// - `lock_timeout` 是**等锁的预算**，不是请求总预算，也不是查询执行预算 -/// - worker 不动也不动：背景不会因为 5xx 醒不了，没人在屏幕前面等 -/// - advisory 锁本身在事务提交时释放（pg_advisory_xact_lock），所以这里的 -/// 等待只在等 worker 的 1 次 materialize 完成；不会等下一个做多次锁 -/// - `55P03` 是 `lock_not_available`，PostgreSQL 唯一的「这个锁等不到」错 -pub async fn materialize_human(pool: &PgPool, kb_id: Uuid) -> AppResult { +/// 为什么不像 `materialize` 那样等锁：等锁的是一条池里的连接，几个决定连着点下来， +/// 每个 job 都抱着一条连接排队，池就空了（0051 §Alternatives）。也为什么不像旧的 +/// 人工入口那样给等待设 2 秒预算：job 没人在屏幕前面等，超时只是把同一次重算推到 +/// 下一次重试，不如一开始就不等。人的那一次点击只提交决定和这个 job（同一事务， +/// `phrase_bindings::decide_with_delivery`），屏幕上等的是事件，不是锁。 +pub async fn try_materialize(pool: &PgPool, kb_id: Uuid) -> AppResult> { let mut tx = pool.begin().await?; - let result = async { - // 2 秒是 0033 的「人点一下」的预算：再长就显成 spinner 了。 - // worker 的等待策略不在这一格里——它走 `materialize`，没预算 - sqlx::query("SET LOCAL lock_timeout = '2s'") - .execute(&mut *tx) - .await?; - sqlx::query("SELECT pg_advisory_xact_lock(hashtext('typed_materialize'), hashtext($1))") - .bind(kb_id.to_string()) - .execute(&mut *tx) - .await?; - materialize_in_tx(&mut tx, kb_id).await - } - .await; - match result { - Ok(outcome) => { - tx.commit().await?; - Ok(outcome) - } - Err(error) => { - // 关掉失败路径的回滚要先把连接释放掉，再走错误映射。 - // 双重错误（回滚也炸了）保留两者，别只报回滚 - if let Err(rollback) = tx.rollback().await { - return Err(AppError::Other(anyhow::Error::new(error).context(format!( - "rolling back human materialization: {rollback}" - )))); - } - if matches!(&error, AppError::Db(sqlx::Error::Database(e)) - if e.code().as_deref() == Some("55P03")) - { - return Err(AppError::CodedConflict { - code: "alignment_busy", - message: "This base is being recomputed by another operation. Please try again shortly." - .into(), - }); - } - Err(error) - } + let acquired: bool = sqlx::query_scalar( + "SELECT pg_try_advisory_xact_lock(hashtext('typed_materialize'), hashtext($1))", + ) + .bind(kb_id.to_string()) + .fetch_one(&mut *tx) + .await?; + if !acquired { + tx.rollback().await?; + return Ok(None); } + let outcome = materialize_in_tx(&mut tx, kb_id).await?; + tx.commit().await?; + Ok(Some(outcome)) } async fn materialize_in_tx( diff --git a/crates/utopia-store/src/phrase_bindings.rs b/crates/utopia-store/src/phrase_bindings.rs index a93899f70..5e06a7060 100644 --- a/crates/utopia-store/src/phrase_bindings.rs +++ b/crates/utopia-store/src/phrase_bindings.rs @@ -262,6 +262,53 @@ fn validate_decision(sig: &PhraseSignature, d: &Decision<'_>) -> AppResult, +) -> AppResult> { + decide_with_delivery_budget(pool, kb_id, sig, d, 3).await +} + +/// 预算单独成参只为了测「排队失败要连判定一起回滚」：0 会被 `enqueue` 拒掉， +/// 那正是一次发生在判定写入之后的真实失败。生产入口固定给 3。 +async fn decide_with_delivery_budget( + pool: &PgPool, + kb_id: Uuid, + sig: &PhraseSignature, + d: Decision<'_>, + max_attempts: i32, +) -> AppResult> { + let mut tx = pool.begin().await?; + if !decide_on(&mut tx, kb_id, sig, d).await? { + tx.rollback().await?; + return Ok(None); + } + let id = crate::jobs::enqueue_with_max_attempts_tx( + &mut tx, + MATERIALIZE_KIND, + serde_json::json!({ "kb_id": kb_id }), + max_attempts, + ) + .await?; + tx.commit().await?; + Ok(Some(id)) +} + /// Write on the caller's connection, so related durable work can share its transaction. pub async fn decide_on( connection: &mut sqlx::PgConnection, @@ -309,6 +356,10 @@ pub async fn decide_on( Ok(res.rows_affected() > 0) } +#[cfg(test)] +#[path = "phrase_bindings_delivery_tests.rs"] +mod delivery_tests; + #[cfg(test)] mod tests { use super::normalize; diff --git a/crates/utopia-store/src/phrase_bindings_delivery_tests.rs b/crates/utopia-store/src/phrase_bindings_delivery_tests.rs new file mode 100644 index 000000000..0aabc0153 --- /dev/null +++ b/crates/utopia-store/src/phrase_bindings_delivery_tests.rs @@ -0,0 +1,226 @@ +//! 人的判定与它的重算任务是一次提交（0051）。 +//! +//! 四件事：判定和 job 一起落库、载荷指着这个库；排队失败时判定也没了；别人握着 +//! 物化锁时 `try_materialize` 立刻让开而不是等；两次判定各自的 job 不管先后处理， +//! 类型化图都收敛到最后一次判定。没有 `UTOPIA_DATABASE_URL` 时跳过。 + +use super::{decide_with_delivery, decide_with_delivery_budget, Decision, MATERIALIZE_KIND}; +use crate::{materialize, phrase_bindings}; +use serde_json::json; +use sqlx::PgPool; +use uuid::Uuid; + +struct Fx { + org: Uuid, + kb: Uuid, + property: Uuid, + sig: phrase_bindings::PhraseSignature, +} + +async fn seed(pool: &PgPool) -> anyhow::Result { + let (org, ws, kb, subject, object, property, statement) = ( + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + ); + sqlx::query("INSERT INTO organizations(id,name) VALUES($1,'phrase-delivery')") + .bind(org) + .execute(pool) + .await?; + sqlx::query("INSERT INTO workspaces(id,org_id,name) VALUES($1,$2,'phrase-delivery')") + .bind(ws) + .bind(org) + .execute(pool) + .await?; + sqlx::query( + "INSERT INTO knowledge_bases(id,workspace_id,name) VALUES($1,$2,'phrase-delivery')", + ) + .bind(kb) + .bind(ws) + .execute(pool) + .await?; + for (id, name) in [(subject, "Acme"), (object, "London")] { + sqlx::query("INSERT INTO entities(id,kb_id,canonical_name) VALUES($1,$2,$3)") + .bind(id) + .bind(kb) + .bind(name) + .execute(pool) + .await?; + } + sqlx::query("INSERT INTO relation_types(id,kb_id,key,label,temporal) VALUES($1,$2,'based_in','based in','state')") + .bind(property).bind(kb).execute(pool).await?; + sqlx::query("INSERT INTO facts(id,kb_id,subject_id,object_id,layer,phrase) VALUES($1,$2,$3,$4,'open','based in')") + .bind(statement).bind(kb).bind(subject).bind(object).execute(pool).await?; + let sig = phrase_bindings::signatures(pool, kb).await?.remove(0); + Ok(Fx { + org, + kb, + property, + sig, + }) +} + +async fn cleanup(pool: &PgPool, f: &Fx) -> anyhow::Result<()> { + sqlx::query("DELETE FROM jobs WHERE payload->>'kb_id'=$1") + .bind(f.kb.to_string()) + .execute(pool) + .await?; + sqlx::query("DELETE FROM organizations WHERE id=$1") + .bind(f.org) + .execute(pool) + .await?; + Ok(()) +} + +fn bound(property: Uuid) -> Decision<'static> { + Decision { + relation_type_id: Some(property), + direction: Some("forward"), + status: "bound", + votes: &serde_json::Value::Null, + decided_by: "person", + } +} + +fn none() -> Decision<'static> { + Decision { + relation_type_id: None, + direction: None, + status: "none", + votes: &serde_json::Value::Null, + decided_by: "person", + } +} + +async fn jobs_for(pool: &PgPool, kb: Uuid) -> anyhow::Result> { + Ok( + sqlx::query_as("SELECT id, kind, status FROM jobs WHERE payload->>'kb_id'=$1 ORDER BY id") + .bind(kb.to_string()) + .fetch_all(pool) + .await?, + ) +} + +#[tokio::test] +async fn a_decision_and_its_job_commit_together() -> anyhow::Result<()> { + let Some(url) = crate::test_db::url() else { + return Ok(()); + }; + let pool = PgPool::connect(&url).await?; + crate::db::migrate(&pool).await?; + let f = seed(&pool).await?; + let run = async { + let id = decide_with_delivery(&pool, f.kb, &f.sig, bound(f.property)) + .await? + .expect("a person's decision is always written"); + let jobs = jobs_for(&pool, f.kb).await?; + assert_eq!(jobs.len(), 1); + assert_eq!(jobs[0].0, id); + assert_eq!(jobs[0].1, MATERIALIZE_KIND); + assert_eq!(jobs[0].2, "queued"); + let bindings = phrase_bindings::bindings(&pool, f.kb).await?; + assert_eq!(bindings.len(), 1); + assert_eq!(bindings[0].status, "bound"); + // 载荷只带库：job 读当前绑定，不回放这次判定的属性 + let payload: serde_json::Value = sqlx::query_scalar("SELECT payload FROM jobs WHERE id=$1") + .bind(id) + .fetch_one(&pool) + .await?; + assert_eq!(payload, json!({ "kb_id": f.kb })); + anyhow::Ok(()) + } + .await; + cleanup(&pool, &f).await?; + run +} + +#[tokio::test] +async fn an_enqueue_failure_takes_the_decision_with_it() -> anyhow::Result<()> { + let Some(url) = crate::test_db::url() else { + return Ok(()); + }; + let pool = PgPool::connect(&url).await?; + crate::db::migrate(&pool).await?; + let f = seed(&pool).await?; + let run = async { + // 预算 0 被 enqueue 拒掉：一次发生在判定写入之后的真实失败 + assert!( + decide_with_delivery_budget(&pool, f.kb, &f.sig, bound(f.property), 0) + .await + .is_err() + ); + assert!(phrase_bindings::bindings(&pool, f.kb).await?.is_empty()); + assert!(jobs_for(&pool, f.kb).await?.is_empty()); + anyhow::Ok(()) + } + .await; + cleanup(&pool, &f).await?; + run +} + +#[tokio::test] +async fn a_busy_projection_is_declined_not_waited_for() -> anyhow::Result<()> { + let Some(url) = crate::test_db::url() else { + return Ok(()); + }; + let pool = PgPool::connect(&url).await?; + crate::db::migrate(&pool).await?; + let f = seed(&pool).await?; + let run = async { + decide_with_delivery(&pool, f.kb, &f.sig, bound(f.property)).await?; + // 另一条连接抱着锁：try 版本立刻回 None，不占第二条连接排队 + let mut gate = pool.begin().await?; + sqlx::query("SELECT pg_advisory_xact_lock(hashtext('typed_materialize'), hashtext($1))") + .bind(f.kb.to_string()) + .execute(&mut *gate) + .await?; + let started = std::time::Instant::now(); + assert!(materialize::try_materialize(&pool, f.kb).await?.is_none()); + assert!(started.elapsed() < std::time::Duration::from_secs(1)); + gate.commit().await?; + let outcome = materialize::try_materialize(&pool, f.kb) + .await? + .expect("lock released"); + assert_eq!(outcome.added, 1); + anyhow::Ok(()) + } + .await; + cleanup(&pool, &f).await?; + run +} + +#[tokio::test] +async fn jobs_processed_in_any_order_converge_on_the_last_decision() -> anyhow::Result<()> { + let Some(url) = crate::test_db::url() else { + return Ok(()); + }; + let pool = PgPool::connect(&url).await?; + crate::db::migrate(&pool).await?; + let f = seed(&pool).await?; + let run = async { + let first = decide_with_delivery(&pool, f.kb, &f.sig, bound(f.property)).await?; + let second = decide_with_delivery(&pool, f.kb, &f.sig, none()).await?; + assert!(first < second); + // 两个 job 都排着；先处理后来的，再处理先来的——每次都读当前绑定 + assert_eq!(jobs_for(&pool, f.kb).await?.len(), 2); + materialize::try_materialize(&pool, f.kb).await?; + materialize::try_materialize(&pool, f.kb).await?; + assert_eq!( + materialize::count(&pool, f.kb).await?, + 0, + "the last decision was none" + ); + // 反过来：先绑、再解绑、先处理老的 + decide_with_delivery(&pool, f.kb, &f.sig, bound(f.property)).await?; + materialize::try_materialize(&pool, f.kb).await?; + assert_eq!(materialize::count(&pool, f.kb).await?, 1); + anyhow::Ok(()) + } + .await; + cleanup(&pool, &f).await?; + run +} diff --git a/crates/utopia-store/tests/human_materialization_has_a_lock_budget.rs b/crates/utopia-store/tests/human_materialization_has_a_lock_budget.rs deleted file mode 100644 index 428618e94..000000000 --- a/crates/utopia-store/tests/human_materialization_has_a_lock_budget.rs +++ /dev/null @@ -1,242 +0,0 @@ -//! A person clicking Review should never wait indefinitely behind a background -//! materialization. The human entry point (`materialize_human`) sets a 2-second -//! `lock_timeout` and maps 55P03 to `AppError::CodedConflict { code: "alignment_busy" }`. -//! -//! The worker entry point (`materialize`) keeps its existing unbounded wait — -//! workers don't have a spinner to look at. This test pins the *human* half only. -//! -//! `#798` survey follow-up; same pattern as `#828`'s `decide_and_apply_human`. - -use sqlx::{postgres::PgPoolOptions, PgPool}; -use utopia_store::{materialize, phrase_bindings}; -use uuid::Uuid; - -async fn wait_for_blocked_descendants( - pool: &PgPool, - blocker: i32, - count: i64, -) -> anyhow::Result<()> { - tokio::time::timeout(std::time::Duration::from_secs(10), async { - loop { - let blocked: i64 = sqlx::query_scalar( - "WITH RECURSIVE blocked(pid) AS ( - SELECT pid FROM pg_stat_activity WHERE $1=ANY(pg_blocking_pids(pid)) - UNION SELECT p.pid FROM pg_stat_activity p JOIN blocked b ON b.pid=ANY(pg_blocking_pids(p.pid)) - ) SELECT count(*) FROM blocked", - ) - .bind(blocker) - .fetch_one(pool) - .await?; - if blocked >= count { - return anyhow::Ok(()); - } - tokio::task::yield_now().await; - } - }) - .await??; - Ok(()) -} - -/// A human materialization that hits a worker-held advisory lock returns -/// CodedConflict("alignment_busy") within a couple of seconds, not after the -/// worker releases naturally. -#[tokio::test] -async fn human_materialization_against_a_held_worker_lock_returns_alignment_busy( -) -> anyhow::Result<()> { - let Some(url) = utopia_store::test_db::url() else { - return Ok(()); - }; - let control = PgPool::connect(&url).await?; - utopia_store::db::migrate(&control).await?; - let pool = PgPoolOptions::new() - .max_connections(2) - .connect(&url) - .await?; - let (org, ws, kb, subject, object, property, statement) = ( - Uuid::now_v7(), - Uuid::now_v7(), - Uuid::now_v7(), - Uuid::now_v7(), - Uuid::now_v7(), - Uuid::now_v7(), - Uuid::now_v7(), - ); - sqlx::query("INSERT INTO organizations(id,name) VALUES($1,'human-materialize-budget')") - .bind(org) - .execute(&pool) - .await?; - sqlx::query("INSERT INTO workspaces(id,org_id,name) VALUES($1,$2,'human-materialize-budget')") - .bind(ws) - .bind(org) - .execute(&pool) - .await?; - sqlx::query("INSERT INTO knowledge_bases(id,workspace_id,name) VALUES($1,$2,'human-materialize-budget')") - .bind(kb).bind(ws).execute(&pool).await?; - sqlx::query( - "INSERT INTO entities(id,kb_id,canonical_name) VALUES($1,$2,'Acme'),($3,$2,'London')", - ) - .bind(subject) - .bind(kb) - .bind(object) - .execute(&pool) - .await?; - sqlx::query("INSERT INTO relation_types(id,kb_id,key,label,temporal) VALUES($1,$2,'based_in','based in','state')") - .bind(property).bind(kb).execute(&pool).await?; - sqlx::query("INSERT INTO facts(id,kb_id,subject_id,object_id,layer,phrase) VALUES($1,$2,$3,$4,'open','based in')") - .bind(statement).bind(kb).bind(subject).bind(object).execute(&pool).await?; - let signature = phrase_bindings::signatures(&pool, kb).await?.remove(0); - phrase_bindings::decide( - &pool, - kb, - &signature, - phrase_bindings::Decision { - relation_type_id: Some(property), - direction: Some("forward"), - status: "bound", - votes: &serde_json::json!({}), - decided_by: "agent", - }, - ) - .await?; - - // Hold the typed_materialize advisory lock on a separate connection so the - // human materialization has to wait for it. We DO NOT commit until the test - // asserts the human request has already returned 409 with alignment_busy. - let mut gate = pool.begin().await?; - let blocker: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") - .fetch_one(&mut *gate) - .await?; - sqlx::query("SELECT pg_advisory_xact_lock(hashtext('typed_materialize'), hashtext($1))") - .bind(kb.to_string()) - .execute(&mut *gate) - .await?; - let human_pool = pool.clone(); - let human_kb = kb; - let human = tokio::spawn(async move { - // Allow up to 10s wall-clock. Without the fix this would still be - // running at 10s and the spawned task would not have returned. - tokio::time::timeout( - std::time::Duration::from_secs(10), - materialize::materialize_human(&human_pool, human_kb), - ) - .await - }); - // The human materialization is on a different connection but the same - // pool — pool size is 2 so it should queue on `gate` and reach the lock - // wait within a few hundred ms. - wait_for_blocked_descendants(&control, blocker, 1).await?; - let result = human.await??; - let err = result.expect_err("the human materialization must have hit lock_timeout"); - let utopia_core::AppError::CodedConflict { code, message } = err else { - panic!("expected CodedConflict, got: {err:?}"); - }; - assert_eq!(code, "alignment_busy"); - assert!(message.contains("try again"), "message: {message}"); - - // Now release the gate and confirm a *worker* materialize (no budget) - // completes normally on this base. - gate.commit().await?; - let outcome = materialize::materialize(&pool, kb).await?; - assert_eq!( - outcome.added, 1, - "the bound statement should produce one typed fact" - ); - sqlx::query("DELETE FROM organizations WHERE id=$1") - .bind(org) - .execute(&pool) - .await?; - pool.close().await; - control.close().await; - Ok(()) -} - -/// When the typed_materialize lock is free, materialize_human completes -/// normally (no regression on the worker side). -#[tokio::test] -async fn human_materialization_with_no_contention_matches_succeeds() -> anyhow::Result<()> { - let Some(url) = utopia_store::test_db::url() else { - return Ok(()); - }; - let pool = PgPoolOptions::new() - .max_connections(2) - .connect(&url) - .await?; - utopia_store::db::migrate(&pool).await?; - let (org, ws, kb, subject, object, property, statement) = ( - Uuid::now_v7(), - Uuid::now_v7(), - Uuid::now_v7(), - Uuid::now_v7(), - Uuid::now_v7(), - Uuid::now_v7(), - Uuid::now_v7(), - ); - sqlx::query("INSERT INTO organizations(id,name) VALUES($1,'human-materialize-no-contention')") - .bind(org) - .execute(&pool) - .await?; - sqlx::query( - "INSERT INTO workspaces(id,org_id,name) VALUES($1,$2,'human-materialize-no-contention')", - ) - .bind(ws) - .bind(org) - .execute(&pool) - .await?; - sqlx::query("INSERT INTO knowledge_bases(id,workspace_id,name) VALUES($1,$2,'human-materialize-no-contention')") - .bind(kb).bind(ws).execute(&pool).await?; - sqlx::query( - "INSERT INTO entities(id,kb_id,canonical_name) VALUES($1,$2,'Acme'),($3,$2,'London')", - ) - .bind(subject) - .bind(kb) - .bind(object) - .execute(&pool) - .await?; - sqlx::query("INSERT INTO relation_types(id,kb_id,key,label,temporal) VALUES($1,$2,'based_in','based in','state')") - .bind(property).bind(kb).execute(&pool).await?; - sqlx::query("INSERT INTO facts(id,kb_id,subject_id,object_id,layer,phrase) VALUES($1,$2,$3,$4,'open','based in')") - .bind(statement).bind(kb).bind(subject).bind(object).execute(&pool).await?; - let signature = phrase_bindings::signatures(&pool, kb).await?.remove(0); - phrase_bindings::decide( - &pool, - kb, - &signature, - phrase_bindings::Decision { - relation_type_id: Some(property), - direction: Some("forward"), - status: "bound", - votes: &serde_json::json!({}), - decided_by: "agent", - }, - ) - .await?; - - let outcome = materialize::materialize_human(&pool, kb).await?; - assert_eq!(outcome.added, 1, "one typed fact should be added"); - - // The SET LOCAL inside materialize_human must not leak to the session — - // otherwise the next caller on this connection inherits our 2-second - // budget, which would be wrong for a worker. Capture the value before and - // after, and assert they're equal. Postgres' session default is `0` - // (wait forever) on a clean connection; we don't depend on that. - let before: Option = sqlx::query_scalar("SHOW lock_timeout") - .fetch_optional(&pool) - .await?; - // Run materialize_human once more — if SET LOCAL had leaked, this second - // call would still see `2s` (which we never set at session level). - let _ = materialize::materialize_human(&pool, kb).await?; - let after: Option = sqlx::query_scalar("SHOW lock_timeout") - .fetch_optional(&pool) - .await?; - assert_eq!( - before, after, - "lock_timeout should not leak out of materialize_human's transaction; before={before:?} after={after:?}" - ); - - sqlx::query("DELETE FROM organizations WHERE id=$1") - .bind(org) - .execute(&pool) - .await?; - pool.close().await; - Ok(()) -} diff --git a/docs/decisions/0051-a-human-phrase-decision-carries-its-materialization-work.md b/docs/decisions/0051-a-human-phrase-decision-carries-its-materialization-work.md index 84505a655..aa3f0a5b7 100644 --- a/docs/decisions/0051-a-human-phrase-decision-carries-its-materialization-work.md +++ b/docs/decisions/0051-a-human-phrase-decision-carries-its-materialization-work.md @@ -1,6 +1,6 @@ # 0051 · A human phrase decision carries its materialization work -- **Status**: proposed; domain contract pending review. Shared store refactors and real regressions only; no production job kind or HTTP/UI change. +- **Status**: implemented 2026-09-23 on the production path (PR #876; the store refactors and regressions landed first in #841) · job kind `materialize_typed` registered in `main`, `phrase_bindings::decide_with_delivery` commits the decision and the job in one transaction, the phrase route answers `202` with the job id and no invented `typed` counts, `GET /kbs/{id}/jobs/{job_id}` is the authorized status read, completion reaches the page as the existing `review` / `graph` events, Review copy in both languages · the synchronous human entry point with a 2-second lock budget (#864, same day) is retired on this route as a consequence: the route no longer waits for the lock at all, so there is nothing left to bound; the kind-word route's #828 timeout is untouched - **Written**: 2026-09-21 - **Related**: [0044](0044-the-ontology-is-a-view-over-what-documents-say.md); [PR #841](https://github.com/deeplethe/utopia/pull/841). @@ -91,3 +91,5 @@ Rollback first stops accepting new jobs of this kind, drains or explicitly retai outstanding jobs, then returns to the old binary. Old workers cannot silently drop an unknown kind. Keep failed work visible; never mark outstanding jobs done just to make rollback clean. External actions (#530) must not use this retry/recovery path. + +**Revision 2026-09-23 (implementation).** The "asynchronous HTTP/job/UI contract" this record left open is now the shape above. Two choices the record left to the implementation: the job's Busy outcome retries after 10 seconds through the existing `Deferred` path and its bounded window, and the status read is scoped by the job payload's `kb_id` so a job of another base answers 404 like an invisible document. The materialization job also writes an `alignment.materialized` audit row when the projection changed, with the job id, so a person can tie a click to what it did once the event has passed. diff --git a/docs/design/ontology.md b/docs/design/ontology.md index 89b2bb779..4ca77e424 100644 --- a/docs/design/ontology.md +++ b/docs/design/ontology.md @@ -74,7 +74,10 @@ reevaluation. Even one edit can reopen all older automatic negative bindings on base, requiring two votes per eligible item through batched model requests; debouncing reduces the number of runs, not the items reconsidered. A burst of ontology edits debounces into one run rather than one run each [#757]; -a person's decision is never overwritten by the agent. On +a person's decision is never overwritten by the agent. A person's phrase decision commits together +with its own recomputation job and the request answers `202` with the job id; the typed graph is +recomputed by that job, never by the request, and the page learns of it through the `review` and +`graph` events or `GET /kbs/{id}/jobs/{job_id}` [0051]. On the 25-document batch with a hand-written ontology of 14 classes and 28 properties, and 60 of 400 kind words bound, 423 signatures cover 861 statements: 36 bind (184 statements), 110 bind to nothing, 1 splits the votes and 276 have no admissible property because an end is unbound; a diff --git a/web/src/api.ts b/web/src/api.ts index 1f139e9cf..7a7577b33 100644 --- a/web/src/api.ts +++ b/web/src/api.ts @@ -2308,14 +2308,16 @@ export const api = { defects_found: number; defects_new: number; }>(`/api/v1/kbs/${kbId}/consistency/check`, { method: "POST" }), - /** 人定一条短语签名：属性与方向，或没有（陈述留在开放图谱）。类型化图谱立刻重算 */ + /** 人定一条短语签名：属性与方向，或没有（陈述留在开放图谱）。判定和它的重算任务 + * 一次提交，答 202 和 job id（0051）；类型化图谱在后台重算，`review` / `graph` + * 事件到了就是算完了，也可以拿 job id 去 `/kbs/{id}/jobs/{job_id}` 问 */ decideAlignmentPhrase: ( kbId: string, bindingId: string, property: string | null, direction: "forward" | "reverse", ) => - request<{ ok: boolean; typed: { added: number; merged: number; retired: number } }>( + request<{ ok: boolean; job_id: number; status: "accepted" }>( `/api/v1/kbs/${kbId}/review/alignment/phrases/${bindingId}`, { method: "POST", body: JSON.stringify({ property, direction }) }, ), diff --git a/web/src/i18n/en.ts b/web/src/i18n/en.ts index 714df96b2..94c43df5d 100644 --- a/web/src/i18n/en.ts +++ b/web/src/i18n/en.ts @@ -1955,8 +1955,7 @@ export const en = { alignmentVotes: (first: string, second: string) => `Votes: ${first} · ${second}`, alignmentConflict: "This decision conflicts with the current state. Refresh and review it before trying again.", alignmentKindWordBusy: "This kind word is being updated by another operation. Please try again shortly.", - alignmentTyped: (kept: number, retired: number) => - `Typed graph recomputed: ${kept} statements typed, ${retired} rows retired`, + alignmentAccepted: "Decision saved. The typed graph is being recomputed and will refresh here when it is done.", defects: "Ontology contradicts itself", defectsHint: "Problems in the definitions themselves — no facts involved. These come first: while a definition contradicts itself, every fact-level finding that rests on it is suspect.", diff --git a/web/src/i18n/zh.ts b/web/src/i18n/zh.ts index ccde8ac0c..4c6403b19 100644 --- a/web/src/i18n/zh.ts +++ b/web/src/i18n/zh.ts @@ -1714,7 +1714,7 @@ export const zh: Strings = { alignmentVotes: (first: string, second: string) => `两票：${first} · ${second}`, alignmentConflict: "此决定与当前状态冲突。请刷新并核对后再试。", alignmentKindWordBusy: "这个类别词正在被其他操作更新，请稍后重试。", - alignmentTyped: (kept: number, retired: number) => `类型化图谱已重算：${kept} 条成了类型化事实，${retired} 行作废`, + alignmentAccepted: "已保存。类型化图谱正在后台重算，算完会在这里自动刷新。", defects: "本体自相矛盾", defectsHint: "定义本身的问题，没有牵涉任何事实。这一档排在前面：定义站不住的时候，据它报出来的每一条事实级结论都可疑。", diff --git a/web/src/pages/Review.tsx b/web/src/pages/Review.tsx index 79b5b50bf..c90e43f5e 100644 --- a/web/src/pages/Review.tsx +++ b/web/src/pages/Review.tsx @@ -1354,11 +1354,9 @@ export function Review() { property: string | null; direction: "forward" | "reverse"; }) => api.decideAlignmentPhrase(kb!.id, id, property, direction), - onSuccess: (r) => { - if (r.typed.added + r.typed.merged + r.typed.retired > 0) { - toast.success(S.review.alignmentTyped(r.typed.added + r.typed.merged, r.typed.retired)); - } - }, + // 202：判定收下了，类型化图谱在后台重算；算完 `review` / `graph` 事件会把 + // 队列和图刷一遍，这里只告诉人「已保存」，不编一个数字出来 + onSuccess: () => toast.success(S.review.alignmentAccepted), onSettled: invalidate, }); const alignmentKindWordAction = useMutation({