diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index fd970a14c..eeba68679 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -83,6 +83,27 @@ export UTOPIA_DATABASE_URL=postgres://utopia:utopia@localhost:5432/utopia cargo test --workspace ``` +### Human phrase delivery regressions + +`human_phrase_materialization_delivery` exercises the real store and kills child +processes at three commit boundaries. It starts the actual queue worker, so run +it **only against a dedicated, otherwise idle test database**, separately from the +workspace suite. Busy-lock coverage calls the private materialization body from +`cfg(test)` and observes the existing production entry point waiting on the lock. +These tests do not register an asynchronous production handler. + +```bash +export UTOPIA_DATABASE_URL=postgres://.../dedicated_delivery_tests +export UTOPIA_TEST_REQUIRE_DB=1 +cargo test --locked -p utopia-store --test human_phrase_materialization_delivery -- --ignored --skip crash_child --test-threads=1 --nocapture +cargo test --locked -p utopia-store --lib materialize::delivery_tests::busy_defers_without_retaining_connections -- --ignored --test-threads=1 --nocapture +``` + +The first command explicitly runs both parents; the process-exit parent invokes +`crash_child` itself and kills and waits for each child. Do not run that child by +hand. See [0051](docs/decisions/0051-a-human-phrase-decision-carries-its-materialization-work.md) +for the proposed delivery contract and remaining production acceptance. + ## Things review will send back **Don't collide migration numbers.** `migrations/` rolls forward by number. Check the latest number on `main` before opening a PR — two branches each writing an `0011_` has happened, and after the merge neither one runs. diff --git a/crates/utopia-store/src/materialize.rs b/crates/utopia-store/src/materialize.rs index b331f59b3..8079ab366 100644 --- a/crates/utopia-store/src/materialize.rs +++ b/crates/utopia-store/src/materialize.rs @@ -66,6 +66,15 @@ pub async fn materialize(pool: &PgPool, kb_id: Uuid) -> AppResult { .bind(kb_id.to_string()) .execute(&mut *tx) .await?; + let outcome = materialize_in_tx(&mut tx, kb_id).await?; + tx.commit().await?; + Ok(outcome) +} + +async fn materialize_in_tx( + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + kb_id: Uuid, +) -> AppResult { // 1. 删不再成立的来源：陈述死了、行死了、签名没绑着、属性或方向变了、陈述带了 mood sqlx::query(&format!( "DELETE FROM typed_fact_sources src @@ -88,7 +97,7 @@ pub async fn materialize(pool: &PgPool, kb_id: Uuid) -> AppResult { WHERE q.fact_id = s.id AND q.role = 'mood'))" )) .bind(kb_id) - .execute(&mut *tx) + .execute(&mut **tx) .await?; // 2. 作废来源全空的类型化行：只动算出来的行（带 from_statement_id 的），人写的不碰 @@ -100,7 +109,7 @@ pub async fn materialize(pool: &PgPool, kb_id: Uuid) -> AppResult { AND NOT EXISTS (SELECT 1 FROM typed_fact_sources src WHERE src.fact_id = t.id)", ) .bind(kb_id) - .execute(&mut *tx) + .execute(&mut **tx) .await? .rows_affected(); @@ -127,7 +136,7 @@ pub async fn materialize(pool: &PgPool, kb_id: Uuid) -> AppResult { ORDER BY s.id" )) .bind(kb_id) - .fetch_all(&mut *tx) + .fetch_all(&mut **tx) .await?; let (mut added, mut merged) = (0u64, 0u64); @@ -144,7 +153,7 @@ pub async fn materialize(pool: &PgPool, kb_id: Uuid) -> AppResult { let (fact, new) = match (reverse, d.object_id, &d.object_value) { (true, Some(object), _) => { insert_fact_on( - &mut tx, + tx, kb_id, object, Some(d.property), @@ -156,7 +165,7 @@ pub async fn materialize(pool: &PgPool, kb_id: Uuid) -> AppResult { } (false, Some(object), _) => { insert_fact_on( - &mut tx, + tx, kb_id, d.subject_id, Some(d.property), @@ -168,7 +177,7 @@ pub async fn materialize(pool: &PgPool, kb_id: Uuid) -> AppResult { } (false, None, Some(value)) => { insert_fact_on( - &mut tx, + tx, kb_id, d.subject_id, Some(d.property), @@ -185,7 +194,7 @@ pub async fn materialize(pool: &PgPool, kb_id: Uuid) -> AppResult { sqlx::query("UPDATE facts SET from_statement_id = $2 WHERE id = $1 AND from_statement_id IS NULL") .bind(fact) .bind(d.statement) - .execute(&mut *tx) + .execute(&mut **tx) .await?; // 新行取代了一条裸行（时间精化，supersedes 链上）：被取代那行的来源跟着搬过来， // 这一轮就收敛，不等下一轮把旧来源当「不成立」删掉再补 @@ -197,7 +206,7 @@ pub async fn materialize(pool: &PgPool, kb_id: Uuid) -> AppResult { ON CONFLICT DO NOTHING", ) .bind(fact) - .execute(&mut *tx) + .execute(&mut **tx) .await?; sqlx::query( "DELETE FROM typed_fact_sources src @@ -205,7 +214,7 @@ pub async fn materialize(pool: &PgPool, kb_id: Uuid) -> AppResult { WHERE n.id = $1 AND src.fact_id = n.supersedes", ) .bind(fact) - .execute(&mut *tx) + .execute(&mut **tx) .await?; } else { merged += 1; @@ -216,7 +225,7 @@ pub async fn materialize(pool: &PgPool, kb_id: Uuid) -> AppResult { ) .bind(fact) .bind(d.statement) - .execute(&mut *tx) + .execute(&mut **tx) .await?; // 证据与限定各抄一份：证据是同一段原文的同一处引文；限定照角色词原样带过去 sqlx::query( @@ -229,7 +238,7 @@ pub async fn materialize(pool: &PgPool, kb_id: Uuid) -> AppResult { ) .bind(fact) .bind(d.statement) - .execute(&mut *tx) + .execute(&mut **tx) .await?; sqlx::query( "INSERT INTO statement_qualifiers (fact_id, role, value, entity_id) @@ -238,10 +247,9 @@ pub async fn materialize(pool: &PgPool, kb_id: Uuid) -> AppResult { ) .bind(fact) .bind(d.statement) - .execute(&mut *tx) + .execute(&mut **tx) .await?; } - tx.commit().await?; Ok(Outcome { retired, added, @@ -260,3 +268,7 @@ pub async fn count(pool: &PgPool, kb_id: Uuid) -> AppResult { .fetch_one(pool) .await?) } + +#[cfg(test)] +#[path = "materialize_delivery_tests.rs"] +mod delivery_tests; diff --git a/crates/utopia-store/src/materialize_delivery_tests.rs b/crates/utopia-store/src/materialize_delivery_tests.rs new file mode 100644 index 000000000..16f964e9f --- /dev/null +++ b/crates/utopia-store/src/materialize_delivery_tests.rs @@ -0,0 +1,189 @@ +//! Opt-in: requires a dedicated, otherwise idle migrated database. +//! UTOPIA_TEST_REQUIRE_DB=1 cargo test -p utopia-store --lib materialize::delivery_tests::busy_defers_without_retaining_connections -- --ignored --test-threads=1 --nocapture +use super::{materialize_in_tx, Outcome}; +use crate::{jobs, materialize, phrase_bindings}; +use serde_json::json; +use sqlx::{postgres::PgPoolOptions, PgPool}; +use std::time::{Duration, Instant}; +use utopia_core::AppResult; +use uuid::Uuid; + +struct AbortOnDrop(tokio::task::JoinHandle); +impl Drop for AbortOnDrop { + fn drop(&mut self) { + self.0.abort(); + } +} +// Test-only adapter: use the real body after acquiring the production lock. +async fn try_materialize(pool: &PgPool, kb_id: Uuid) -> AppResult> { + let mut tx = pool.begin().await?; + 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 accept( + pool: &PgPool, + kb: Uuid, + sig: &phrase_bindings::PhraseSignature, + property: Option, + budget: i32, +) -> anyhow::Result { + let mut tx = pool.begin().await?; + anyhow::ensure!( + phrase_bindings::decide_on( + &mut tx, + kb, + sig, + phrase_bindings::Decision { + relation_type_id: property, + direction: property.map(|_| "forward"), + status: if property.is_some() { "bound" } else { "none" }, + votes: &json!({}), + decided_by: "person", + } + ) + .await? + ); + let id = jobs::enqueue_with_max_attempts_tx( + &mut tx, + "test_human_phrase_materialize", + json!({"kb_id":kb}), + budget, + ) + .await?; + tx.commit().await?; + Ok(id) +} +async fn claim(pool: &PgPool, id: i64) -> anyhow::Result { + // Restrict the production claim SQL to this test's job, never steal work. + Ok(sqlx::query_as("UPDATE jobs SET status='running', attempts=attempts+1, locked_at=now() WHERE id=$1 AND status='queued' RETURNING id,kind,payload,attempts,max_attempts") + .bind(id).fetch_one(pool).await?) +} +async fn handle(pool: &PgPool, kb: Uuid, job: &jobs::Job) -> anyhow::Result<()> { + if try_materialize(pool, kb).await?.is_some() { + sqlx::query("UPDATE jobs SET status='done',last_error=NULL WHERE id=$1") + .bind(job.id) + .execute(pool) + .await?; + } else { + let e = anyhow::anyhow!("typed projection busy") + .context(utopia_core::Deferred::new(Duration::from_secs(1))); + jobs::mark_failed(pool, job, &e).await?; + } + Ok(()) +} +async fn status(pool: &PgPool, id: i64) -> anyhow::Result { + Ok(sqlx::query_scalar("SELECT status FROM jobs WHERE id=$1") + .bind(id) + .fetch_one(pool) + .await?) +} + +#[tokio::test] +#[ignore = "requires a dedicated idle database; observes a blocked production call"] +async fn busy_defers_without_retaining_connections() -> anyhow::Result<()> { + let Some(url) = crate::test_db::url() else { + return Ok(()); + }; + let control = PgPool::connect(&url).await?; + crate::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,'materialize-race')") + .bind(org) + .execute(&pool) + .await?; + sqlx::query("INSERT INTO workspaces(id,org_id,name) VALUES($1,$2,'materialize-race')") + .bind(ws) + .bind(org) + .execute(&pool) + .await?; + sqlx::query( + "INSERT INTO knowledge_bases(id,workspace_id,name) VALUES($1,$2,'materialize-race')", + ) + .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 signature = phrase_bindings::signatures(&pool, kb).await?.remove(0); + let run=async { + let mut blocker=control.begin().await?; + sqlx::query("SELECT pg_advisory_xact_lock(hashtext('typed_materialize'),hashtext($1))").bind(kb.to_string()).execute(&mut *blocker).await?; + let start=Instant::now(); + let id=tokio::time::timeout(Duration::from_secs(2),accept(&pool,kb,&signature,Some(property),3)).await??; + let job=claim(&pool,id).await?; + tokio::time::timeout(Duration::from_secs(2),handle(&pool,kb,&job)).await??; + anyhow::ensure!(status(&pool,id).await?=="queued"); + for _ in 0..10 { anyhow::ensure!(tokio::time::timeout(Duration::from_secs(1),try_materialize(&pool,kb)).await??.is_none()); } + anyhow::ensure!(tokio::time::timeout(Duration::from_secs(1),pool.acquire()).await?.is_ok()); + println!("B-T04/T07/T21 PASS busy deferred same job; two-connection pool available; elapsed_ms={}",start.elapsed().as_millis()); + let blocker_pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()").fetch_one(&mut *blocker).await?; + let production_pool = pool.clone(); + let production = tokio::spawn(async move { super::materialize(&production_pool, kb).await }); + let mut production = AbortOnDrop(production); + let observed = tokio::time::timeout(Duration::from_secs(5), async { + loop { + let waiting: bool = sqlx::query_scalar("SELECT EXISTS (SELECT 1 FROM pg_stat_activity WHERE $1 = ANY(pg_blocking_pids(pid)) AND query LIKE '%pg_advisory_xact_lock%' AND wait_event_type='Lock')") + .bind(blocker_pid).fetch_one(&control).await?; + if waiting { break anyhow::Ok(()); } + tokio::task::yield_now().await; + } + }).await; + if !matches!(observed, Ok(Ok(()))) { + blocker.rollback().await?; + production.0.abort(); let _ = (&mut production.0).await; + anyhow::bail!("production materialize did not wait on the test lock: {observed:?}"); + } + println!("production materialize observed blocked by pg_blocking_pids"); + blocker.rollback().await?; + tokio::time::timeout(Duration::from_secs(5), &mut production.0).await???; + + handle(&pool,kb,&claim(&pool,id).await?).await?; + anyhow::ensure!(materialize::count(&pool,kb).await?==1); + + anyhow::Ok(()) + }.await; + sqlx::query("DELETE FROM jobs WHERE payload->>'kb_id'=$1") + .bind(kb.to_string()) + .execute(&pool) + .await?; + sqlx::query("DELETE FROM organizations WHERE id=$1") + .bind(org) + .execute(&pool) + .await?; + pool.close().await; + control.close().await; + run +} diff --git a/crates/utopia-store/src/phrase_bindings.rs b/crates/utopia-store/src/phrase_bindings.rs index c3031ef31..a93899f70 100644 --- a/crates/utopia-store/src/phrase_bindings.rs +++ b/crates/utopia-store/src/phrase_bindings.rs @@ -226,6 +226,12 @@ pub async fn decide( sig: &PhraseSignature, d: Decision<'_>, ) -> AppResult { + validate_decision(sig, &d)?; + let mut connection = pool.acquire().await?; + decide_on(&mut connection, kb_id, sig, d).await +} + +fn validate_decision(sig: &PhraseSignature, d: &Decision<'_>) -> AppResult { if !matches!(d.status, "bound" | "none" | "undecided") { return Err(AppError::Validation(format!( "unknown binding status {:?}", @@ -253,6 +259,17 @@ pub async fn decide( if phrase.is_empty() { return Err(AppError::Validation("an empty phrase binds nothing".into())); } + Ok(phrase) +} + +/// Write on the caller's connection, so related durable work can share its transaction. +pub async fn decide_on( + connection: &mut sqlx::PgConnection, + kb_id: Uuid, + sig: &PhraseSignature, + d: Decision<'_>, +) -> AppResult { + let phrase = validate_decision(sig, &d)?; let res = sqlx::query( "INSERT INTO phrase_bindings (id, kb_id, phrase, subject_type_id, object_type_id, object_is_value, @@ -287,7 +304,7 @@ pub async fn decide( .bind(i32::try_from(sig.count).unwrap_or(i32::MAX)) .bind(&sig.examples) .bind(d.decided_by) - .execute(pool) + .execute(connection) .await?; Ok(res.rows_affected() > 0) } diff --git a/crates/utopia-store/tests/human_phrase_materialization_delivery.rs b/crates/utopia-store/tests/human_phrase_materialization_delivery.rs new file mode 100644 index 000000000..c1059866c --- /dev/null +++ b/crates/utopia-store/tests/human_phrase_materialization_delivery.rs @@ -0,0 +1,334 @@ +//! Isolated delivery regression; no production job kind or HTTP route is registered. +use serde_json::json; +use sqlx::{postgres::PgPoolOptions, PgPool}; +use std::time::{Duration, Instant}; +use utopia_store::{jobs, materialize, phrase_bindings}; +use uuid::Uuid; + +async fn accept( + pool: &PgPool, + kb: Uuid, + sig: &phrase_bindings::PhraseSignature, + property: Option, + budget: i32, +) -> anyhow::Result { + let mut tx = pool.begin().await?; + anyhow::ensure!( + phrase_bindings::decide_on( + &mut tx, + kb, + sig, + phrase_bindings::Decision { + relation_type_id: property, + direction: property.map(|_| "forward"), + status: if property.is_some() { "bound" } else { "none" }, + votes: &json!({}), + decided_by: "person", + } + ) + .await? + ); + let id = jobs::enqueue_with_max_attempts_tx( + &mut tx, + "test_human_phrase_materialize", + json!({"kb_id":kb}), + budget, + ) + .await?; + tx.commit().await?; + Ok(id) +} +async fn claim(pool: &PgPool, id: i64) -> anyhow::Result { + // Restrict the production claim SQL to this test's job, never steal work. + Ok(sqlx::query_as("UPDATE jobs SET status='running', attempts=attempts+1, locked_at=now() WHERE id=$1 AND status='queued' RETURNING id,kind,payload,attempts,max_attempts") + .bind(id).fetch_one(pool).await?) +} +async fn handle(pool: &PgPool, kb: Uuid, job: &jobs::Job) -> anyhow::Result<()> { + materialize::materialize(pool, kb).await?; + sqlx::query("UPDATE jobs SET status='done',last_error=NULL WHERE id=$1") + .bind(job.id) + .execute(pool) + .await?; + Ok(()) +} +async fn status(pool: &PgPool, id: i64) -> anyhow::Result { + Ok(sqlx::query_scalar("SELECT status FROM jobs WHERE id=$1") + .bind(id) + .fetch_one(pool) + .await?) +} + +#[tokio::test] +#[ignore = "opt-in delivery regression; requires a dedicated idle database"] +async fn delivery_rollback_late_arrivals_recovery_and_cost() -> 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,'materialize-race')") + .bind(org) + .execute(&pool) + .await?; + sqlx::query("INSERT INTO workspaces(id,org_id,name) VALUES($1,$2,'materialize-race')") + .bind(ws) + .bind(org) + .execute(&pool) + .await?; + sqlx::query( + "INSERT INTO knowledge_bases(id,workspace_id,name) VALUES($1,$2,'materialize-race')", + ) + .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 signature = phrase_bindings::signatures(&pool, kb).await?.remove(0); + let run=async { + // Invalid enqueue budget is a real helper failure after the decision write. + anyhow::ensure!(accept(&pool,kb,&signature,Some(property),0).await.is_err()); + anyhow::ensure!(phrase_bindings::bindings(&pool,kb).await?.is_empty()); + let count:i64=sqlx::query_scalar("SELECT count(*) FROM jobs WHERE payload->>'kb_id'=$1").bind(kb.to_string()).fetch_one(&pool).await?; + anyhow::ensure!(count==0); + println!("B-T01/T02 PASS same-transaction enqueue failure rolls back decision"); + + let id=accept(&pool,kb,&signature,Some(property),3).await?; + handle(&pool,kb,&claim(&pool,id).await?).await?; + anyhow::ensure!(materialize::count(&pool,kb).await?==1); + + // A decision after the older materializer's last read has its own job. + // Pause before ack by not acking the first completed projection yet. + let old=accept(&pool,kb,&signature,Some(property),3).await?; + let old_job=claim(&pool,old).await?; + materialize::materialize(&pool,kb).await?; + let newer=accept(&pool,kb,&signature,None,3).await?; + anyhow::ensure!(status(&pool,newer).await?=="queued"); + handle(&pool,kb,&claim(&pool,newer).await?).await?; + handle(&pool,kb,&old_job).await?; + anyhow::ensure!(materialize::count(&pool,kb).await?==0); + anyhow::ensure!(phrase_bindings::bindings(&pool,kb).await?[0].status=="none"); + let open:i64=sqlx::query_scalar("SELECT count(*) FROM facts WHERE id=$1 AND invalidated_at IS NULL").bind(statement).fetch_one(&pool).await?; + anyhow::ensure!(open==1); + println!("B-T05/T06/T09/T11/T14/T15 PASS late decision, reverse order and duplicate processing converge without replay"); + + // Actual queue recovery (task restart, not an OS process crash). + let recovery=accept(&pool,kb,&signature,Some(property),3).await?; + let _unacked=claim(&pool,recovery).await?; + materialize::materialize(&pool,kb).await?; + let (sent,mut received)=tokio::sync::mpsc::unbounded_channel(); + let worker_pool=pool.clone(); + let run_pool=pool.clone(); + let worker=tokio::spawn(jobs::run_worker(worker_pool,std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(1)),move |job| { + let pool=run_pool.clone(); let sent=sent.clone(); + async move { + anyhow::ensure!(job.kind=="test_human_phrase_materialize"); + materialize::materialize(&pool,kb).await?; + sent.send(job.id)?; + Ok(()) + } + })); + let result=tokio::time::timeout(Duration::from_secs(10),received.recv()).await; + let acknowledged=if matches!(result,Ok(Some(x)) if x==recovery) { + tokio::time::timeout(Duration::from_secs(5),async { + while status(&pool,recovery).await? != "done" {tokio::task::yield_now().await;} + anyhow::Ok(()) + }).await.unwrap_or_else(|e| Err(e.into())) + } else { Err(anyhow::anyhow!("worker did not recover expected job: {result:?}")) }; + worker.abort(); let _=worker.await; acknowledged?; + anyhow::ensure!(materialize::count(&pool,kb).await?==1); + println!("B-T09/T12/T20 PASS actual run_worker startup recovery and idempotent recompute; no model path linked to handler; task-level restart only"); + + // Deferred is bounded, then the normal finite failure budget takes over. + let failure=accept(&pool,kb,&signature,None,1).await?; + sqlx::query("UPDATE jobs SET payload=payload || jsonb_build_object('deferred_since',(now()-interval '1 day')::text) WHERE id=$1").bind(failure).execute(&pool).await?; + let job=claim(&pool,failure).await?; + jobs::mark_failed(&pool,&job,&anyhow::anyhow!("still busy").context(utopia_core::Deferred::new(Duration::from_secs(1)))).await?; + anyhow::ensure!(status(&pool,failure).await?=="failed"); + anyhow::ensure!(jobs::requeue_failed(&pool,jobs::RequeueScope{kb_id:Some(kb),kind:Some("test_human_phrase_materialize"),failed_since:None}).await?==1); + handle(&pool,kb,&claim(&pool,failure).await?).await?; + anyhow::ensure!(status(&pool,failure).await?=="done"); + println!("B-T08 PASS finite deferral exhaustion stays visible and can be explicitly requeued (not a process-crash test)"); + + // Cost on a 100-statement graph. No listener or model is needed to find durable rows. + for i in 1..100 { + let subject=Uuid::now_v7(); + sqlx::query("INSERT INTO entities(id,kb_id,canonical_name) VALUES($1,$2,$3)").bind(subject).bind(kb).bind(format!("Entity {i}")).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(Uuid::now_v7()).bind(kb).bind(subject).bind(object).execute(&pool).await?; + } + for n in [1,10,100] { + let start=Instant::now(); let mut ids=Vec::new(); let mut max_ms=0; + for _ in 0..n { let tick=Instant::now(); ids.push(accept(&pool,kb,&signature,Some(property),3).await?); max_ms=max_ms.max(tick.elapsed().as_micros()); } + let accept_ms=start.elapsed().as_millis(); + for id in ids { handle(&pool,kb,&claim(&pool,id).await?).await?; } + anyhow::ensure!(materialize::count(&pool,kb).await?==100); + println!("B-T22 decisions={n} statements=100 jobs={n} recomputations={n} accept_total_ms={accept_ms} max_accept_us={max_ms} convergence_ms={}",start.elapsed().as_millis()); + } + anyhow::Ok(()) + }.await; + sqlx::query("DELETE FROM jobs WHERE payload->>'kb_id'=$1") + .bind(kb.to_string()) + .execute(&pool) + .await?; + sqlx::query("DELETE FROM organizations WHERE id=$1") + .bind(org) + .execute(&pool) + .await?; + pool.close().await; + control.close().await; + run +} + +// The integration-test executable is also a subprocess probe. This ignored entry +// runs only with explicit per-fixture environment from the parent, never in CI. +#[test] +#[ignore = "spawned only by the isolated crash-window experiment"] +fn crash_child() { + let phase = std::env::var("UTOPIA_PROBE_PHASE").expect("explicit probe phase"); + let kb: Uuid = std::env::var("UTOPIA_PROBE_KB").unwrap().parse().unwrap(); + let property: Uuid = std::env::var("UTOPIA_PROBE_PROPERTY") + .unwrap() + .parse() + .unwrap(); + let ready = std::env::var("UTOPIA_PROBE_READY").unwrap(); + tokio::runtime::Runtime::new().unwrap().block_on(async { + let pool = PgPool::connect(&utopia_store::test_db::url().unwrap()) + .await + .unwrap(); + let signature = phrase_bindings::signatures(&pool, kb) + .await + .unwrap() + .remove(0); + let mut tx = pool.begin().await.unwrap(); + phrase_bindings::decide_on( + &mut tx, + kb, + &signature, + phrase_bindings::Decision { + relation_type_id: Some(property), + direction: Some("forward"), + status: "bound", + votes: &json!({}), + decided_by: "person", + }, + ) + .await + .unwrap(); + let id = jobs::enqueue_with_max_attempts_tx( + &mut tx, + "test_human_phrase_materialize", + json!({"kb_id":kb}), + 3, + ) + .await + .unwrap(); + if phase == "uncommitted" { + std::fs::write(&ready, id.to_string()).unwrap(); + std::future::pending::<()>().await; + } + tx.commit().await.unwrap(); + if phase == "unacked" { + claim(&pool, id).await.unwrap(); + materialize::materialize(&pool, kb).await.unwrap(); + } + std::fs::write(&ready, id.to_string()).unwrap(); + std::future::pending::<()>().await; + }); +} + +struct ChildGuard(std::process::Child, std::path::PathBuf); +impl Drop for ChildGuard { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + let _ = std::fs::remove_file(&self.1); + } +} + +#[tokio::test] +#[ignore = "opt-in subprocess regression; requires a dedicated idle database"] +async fn process_exit_preserves_the_committed_delivery_boundary() -> anyhow::Result<()> { + let Some(url) = utopia_store::test_db::url() else { + return Ok(()); + }; + let pool = PgPool::connect(&url).await?; + utopia_store::db::migrate(&pool).await?; + let (org, ws, kb, subject, object, property) = ( + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + ); + sqlx::raw_sql(&format!("INSERT INTO organizations(id,name) VALUES('{org}','crash-probe'); + INSERT INTO workspaces(id,org_id,name) VALUES('{ws}','{org}','crash-probe'); + INSERT INTO knowledge_bases(id,workspace_id,name) VALUES('{kb}','{ws}','crash-probe'); + INSERT INTO entities(id,kb_id,canonical_name) VALUES('{subject}','{kb}','S'),('{object}','{kb}','O'); + INSERT INTO relation_types(id,kb_id,key,label) VALUES('{property}','{kb}','rel','Rel'); + INSERT INTO facts(id,kb_id,subject_id,object_id,layer,phrase) VALUES('{}','{kb}','{subject}','{object}','open','rel');",Uuid::now_v7())).execute(&pool).await?; + let run=async { + for phase in ["uncommitted","accepted","unacked"] { + let ready=std::env::temp_dir().join(format!("utopia-crash-{}",Uuid::now_v7())); + let mut child=ChildGuard(std::process::Command::new(std::env::current_exe()?) + .args(["--exact","crash_child","--ignored","--nocapture"]) + .env("UTOPIA_PROBE_PHASE",phase).env("UTOPIA_PROBE_KB",kb.to_string()) + .env("UTOPIA_PROBE_PROPERTY",property.to_string()).env("UTOPIA_PROBE_READY",&ready) + .spawn()?, ready.clone()); + let id=tokio::time::timeout(Duration::from_secs(15),async { + loop { + if let Ok(value)=std::fs::read_to_string(&ready) {break value.parse::();} + tokio::time::sleep(Duration::from_millis(10)).await; + } + }).await??; + child.0.kill()?;child.0.wait()?; + let _=std::fs::remove_file(&ready); + if phase=="uncommitted" { + anyhow::ensure!(phrase_bindings::bindings(&pool,kb).await?.is_empty()); + let exists:bool=sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM jobs WHERE id=$1)").bind(id).fetch_one(&pool).await?; + anyhow::ensure!(!exists); + } else { + anyhow::ensure!(phrase_bindings::bindings(&pool,kb).await?[0].status=="bound"); + anyhow::ensure!(status(&pool,id).await?==if phase=="accepted" {"queued"} else {"running"}); + // Exercise the same single-instance recovery update, scoped to + // our job. Actual run_worker startup is tested separately above. + sqlx::query("UPDATE jobs SET status='queued',locked_at=NULL WHERE id=$1 AND status='running'").bind(id).execute(&pool).await?; + handle(&pool,kb,&claim(&pool,id).await?).await?; + anyhow::ensure!(materialize::count(&pool,kb).await?==1); + anyhow::ensure!(materialize::materialize(&pool,kb).await?==materialize::Outcome::default()); + } + println!("OS process kill phase={phase}: PASS"); + } + anyhow::Ok(()) + }.await; + sqlx::query("DELETE FROM jobs WHERE payload->>'kb_id'=$1") + .bind(kb.to_string()) + .execute(&pool) + .await?; + sqlx::query("DELETE FROM organizations WHERE id=$1") + .bind(org) + .execute(&pool) + .await?; + run +} diff --git a/docs/decisions/0044-the-ontology-is-a-view-over-what-documents-say.md b/docs/decisions/0044-the-ontology-is-a-view-over-what-documents-say.md index 9bd47def3..baf99da98 100644 --- a/docs/decisions/0044-the-ontology-is-a-view-over-what-documents-say.md +++ b/docs/decisions/0044-the-ontology-is-a-view-over-what-documents-say.md @@ -68,6 +68,9 @@ This is how the facts a reader draws without the text stating them (a place's co When the ontology changes, only facts under changed signatures and rules are recomputed. A signature with no property stays in the open graph, loses nothing, and counts toward the workbench's suggestions. +**Revision proposed 2026-09-21:** [0051](0051-a-human-phrase-decision-carries-its-materialization-work.md) addresses delivery after a human binding commits beyond an older materializer’s final read. It proposes a decision and its own durable job in one transaction, while retaining the current projection semantics. The asynchronous HTTP/job/UI contract remains unimplemented. + + ### 4. The ontology is built on a workbench from three sources The ontology page becomes a workbench. Its elements come from three sources: **suggestions from the open graph** (the most frequent unbound signatures, the type words in use, and an ontology agent that reads them against competency questions and proposes object types, link types, properties and rules with definitions, examples and the signatures they would bind); **an imported file** (a pack of 0008, schema.org, an OWL or JSON-LD file); and **online editing**. People approve through actions; every approved element carries regression cases drawn from the open graph, and changing a definition reruns them. The ontology is judged by whether the competency questions can be answered correctly. Structure the slice and the rules depend on (class hierarchy, equivalences, domains, ranges) is part of approval, and duplicate properties are merged as part of governance. 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 new file mode 100644 index 000000000..84505a655 --- /dev/null +++ b/docs/decisions/0051-a-human-phrase-decision-carries-its-materialization-work.md @@ -0,0 +1,93 @@ +# 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. +- **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). + +## Problem + +A human binding may commit after a running aligner's final read. Depending on that aligner's late recheck can therefore leave the accepted decision without a projection. The proposed delivery unit is the accepted decision and its own durable work, rather than a guess that another running job will cover it. + +## Decision requested + +Prefer one durable materialization-only job for every accepted human phrase +binding, committed in the **same transaction** as that binding. The job reads the +current bindings; it never replays the old decision. Its payload needs KB identity, +not an old property/status. Do not suppress delivery because another job is running. + +Return an honest saved/accepted response with job ID (prefer HTTP 202), rather than +inventing `typed: {added:0,...}`. Before production wiring, agree a minimal authorized +KB/job-kind status read and UI completion/failure behavior. Returning a job ID alone +does not provide those surfaces. The present route still materializes synchronously. + +A job acquires the existing `typed_materialize` transaction advisory lock using +try-lock, then executes the original materialization body on that same connection. +Busy rolls back and enters the existing Deferred path; it is not a successful job. +Commit projection before acknowledging the job. Use normal finite failure budgets +and the existing bounded deferral window/requeue surface; no infinite hidden retry. + +## Why a late decision is not lost + +For every committed decision D there is a durable J_D committed with it. J_D is only +visible after D commits. Reading after acquiring the materialization lock observes +previously committed bindings under READ COMMITTED. If an older worker already did +its final read, J_D remains independently queued. If jobs run out of order, both +read the latest bindings instead of restoring old payload values. Once a finite +sequence of decisions stops, a successfully processed follow-up converges to the +last binding, assuming database/worker availability and eventual lock acquisition. + +This does not promise a separate historical projection for every intermediate +click, a single snapshot across the whole multi-statement recompute, or progress +through permanent failures. Existing human priority and statement/evidence/temporal +semantics belong to the reused materializer, not the queue. + +## Reusable code and regression evidence + +The production `phrase_bindings::decide` calls `decide_on`; production `materialize` calls private `materialize_in_tx`. They share the existing SQL and transaction body with tests. No unused public try-lock entry point is exported. Busy orchestration belongs in module-local `cfg(test)` code, while normal recomputation uses the existing public materializer. + +The retained integration target is [`human_phrase_materialization_delivery`](../../crates/utopia-store/tests/human_phrase_materialization_delivery.rs). Its module header documents opt-in execution on a dedicated, otherwise idle database, including real worker startup and OS subprocess termination. Busy coverage is in `materialize`'s module-local tests. These test adapters do not register a production handler. + +Historical evidence at `e83f015f9a3949e53b1ae849b8d6dad0e2c4546e` on Linux / PostgreSQL 16.15 comprised two explicit parent tests and three actual killed subprocesses: before decision/job commit, after acceptance commit, and after projection commit before ack. Enqueue-helper failure rolled back both rows; Busy deferred the same job and released a two-connection pool; late arrivals, reverse processing and duplicates converged; actual worker startup reclaimed running work; exhausted deferral became visible failed and scoped requeue recovered it. The enqueue failure is helper-boundary injection, not a disk failure at COMMIT. + +Three historical runtime mutations were rejected: splitting decision and enqueue transactions left an orphan; acknowledging Busy marked unfinished work done; adding a model prerequisite stopped pure recomputation. These results concern real store behavior but do not establish an asynchronous production route. The renamed tests preserve those assertions; new validation must be reported against its own head rather than reusing these counts. + +## Alternatives + +A late recheck cannot cover a decision committed after that check. Skipping enqueue when a worker is running loses this independent delivery obligation. Replaying a decision's old property payload can overwrite a later human choice. Blocking on the materialization lock retains scarce connections; a bounded Deferred outcome preserves work without claiming completion. A global lease/recovery redesign would expand the present single-process queue contract and is outside this proposal. + +## Measured cost, not a throughput claim + +One Linux run on a 100-open-statement graph, two request-pool connections: + +| Decisions | Jobs/recomputations | Total accept ms | Max accept µs | Total convergence ms | +|---|---|---|---|---| +| 1 | 1 | 0 | 985 | 142 | +| 10 | 10 | 10 | 1770 | 58 | +| 100 | 100 | 75 | 1036 | 593 | + +The first pass creates projections; later passes are largely no-ops. Times are +observations, not a percentile benchmark or a maximum latency guarantee. Production +large graphs and concurrent ingestion were not measured. The cost can be N full +recomputations for N decisions. Optimize only with a separately tested finite set of +covered job IDs, never with “one is running, so skip enqueue.” + +## Remaining production acceptance and recovery limits + +The current queue assumes **one server process**: startup requeues all running jobs. +This experiment does not establish safe multi-instance ownership. A mark_done write +failure can leave running until restart; this is not live lease recovery. An actual +kill inside a partly written materialization transaction, failed ack persistence, +notification-loss polling, old-aligner/new-handler overlap, and production route +permissions/status reads/UI E2E remain explicit acceptance work. Existing temporal +and evidence tests must pass after any extraction; the experiment is not a substitute. + +After contract approval, wire the route's existing authorization and binding lookup +to a same-transaction decision+job function; register the pure handler in main; +add job status authorization and completion/failure events; update Review and both +languages. Keep #828's kind-word lock timeout isolated. Do not reuse align_phrases, +whose model dependency, busy guard and late recheck are a different contract. + +Rollback first stops accepting new jobs of this kind, drains or explicitly retains +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. diff --git a/docs/decisions/README.md b/docs/decisions/README.md index 944018a8d..989758f9f 100644 --- a/docs/decisions/README.md +++ b/docs/decisions/README.md @@ -75,6 +75,7 @@ The test for writing one: if someone (including us) looks at a piece of code in | 0047 | [A rule may conclude a relation](0047-a-rule-may-conclude-a-relation.md) | Proposed 2026-09-20 · nothing built · A rule reads one entity and concludes about that same entity, so a threshold over a chain — a holding above 50% in a company that itself holds above 50% in another — cannot be written at all, and the query-time path walk that answers it produces no interval, no premises and nothing a queue can see. The conclusion becomes a **relation** between the subject and one entity reached across one declared relation, valid on the intersection of every premise interval including the join edge's. The concluded edge rejoins the pool `derive()` reads and the axiom pass runs once per round, coupling the two reasoners for the first time: 0021's cycle objection is answered with the **finiteness** argument [0030](0030-a-rule-may-read-what-a-rule-concluded.md) already put in place of acyclicity, rather than with a fixed ordering that would let a legitimate rule silently never fire. Reading a value across a hop is [0032](0032-a-rule-computes-what-it-concludes.md)'s decision, reused rather than re-decided. Negation, aggregation, a second hop and user-defined recursion stay out; the three caps in play are set by measurement in the PR that changes them | | 0048 | [Provenance references stay inside the knowledge base](0048-provenance-references-stay-inside-the-knowledge-base.md) | Proposed 2026-09-20 · implemented in PR #832 (migration 0070), pending review · a column foreign key proves the target exists, not that it is the same KB's — every reference an export can resolve gets a schema-level same-KB invariant: composite `(kb_id, ref)` foreign keys on the 26 edges whose row carries its own `kb_id` (same-table self-references deferred to commit), row triggers on the 13 whose kb authority is a parent row, `kb_id` immutability on every owned table, and a precondition scan that fails the migration closed on an already-cross-KB ledger · measured populate cost within noise; mechanism question open as issue #842 | | 0050 | [An action attempt keeps its identity and uncertain outcome](0050-an-action-attempt-keeps-its-identity-and-uncertain-outcome.md) | Proposed · durable execution identity and uncertain outcomes; no sender | +| 0051 | [A human phrase decision carries its materialization work](0051-a-human-phrase-decision-carries-its-materialization-work.md) | Proposed · decision and materialization delivery; shared refactors and real regressions only | | | Record | Domain | Status | |---|---|---|---| @@ -127,6 +128,7 @@ The test for writing one: if someone (including us) looks at a piece of code in | 0047 | [A rule may conclude a relation](0047-a-rule-may-conclude-a-relation.md) | rules | current | | 0048 | [Provenance references stay inside the knowledge base](0048-provenance-references-stay-inside-the-knowledge-base.md) | ledger | current | | 0050 | [An action attempt keeps its identity and uncertain outcome](0050-an-action-attempt-keeps-its-identity-and-uncertain-outcome.md) | lakehouse-and-actions | proposed | +| 0051 | [A human phrase decision carries its materialization work](0051-a-human-phrase-decision-carries-its-materialization-work.md) | ontology | proposed | The status word is whether a later record has overtaken this one; what is built is in the record's own status line. Domains are the files of [../design/](../design/README.md), where every record is dated and the status words are defined.