From 8866ae25a64e84ddf903b4826c6e33176e3e09b3 Mon Sep 17 00:00:00 2001 From: dada-yan Date: Mon, 21 Sep 2026 00:34:37 +1000 Subject: [PATCH 1/5] Experiment with durable phrase delivery without registering a production handler Signed-off-by: dada-yan --- crates/utopia-store/src/materialize.rs | 52 +++-- crates/utopia-store/src/phrase_bindings.rs | 13 +- .../tests/phrase_delivery_prototype.rs | 215 ++++++++++++++++++ 3 files changed, 266 insertions(+), 14 deletions(-) create mode 100644 crates/utopia-store/tests/phrase_delivery_prototype.rs diff --git a/crates/utopia-store/src/materialize.rs b/crates/utopia-store/src/materialize.rs index b331f59b3..4006aa76b 100644 --- a/crates/utopia-store/src/materialize.rs +++ b/crates/utopia-store/src/materialize.rs @@ -66,6 +66,33 @@ 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) +} + +/// Isolated delivery prototype: Busy never means materialization completed. +pub 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 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 +115,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 +127,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 +154,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 +171,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 +183,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 +195,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 +212,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 +224,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 +232,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 +243,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 +256,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 +265,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, diff --git a/crates/utopia-store/src/phrase_bindings.rs b/crates/utopia-store/src/phrase_bindings.rs index c3031ef31..6b5f78ce4 100644 --- a/crates/utopia-store/src/phrase_bindings.rs +++ b/crates/utopia-store/src/phrase_bindings.rs @@ -225,6 +225,17 @@ pub async fn decide( kb_id: Uuid, sig: &PhraseSignature, d: Decision<'_>, +) -> AppResult { + let mut connection = pool.acquire().await?; + decide_on(&mut connection, kb_id, sig, d).await +} + +/// Isolated delivery prototype: caller owns the decision/enqueue transaction. +pub async fn decide_on( + connection: &mut sqlx::PgConnection, + kb_id: Uuid, + sig: &PhraseSignature, + d: Decision<'_>, ) -> AppResult { if !matches!(d.status, "bound" | "none" | "undecided") { return Err(AppError::Validation(format!( @@ -287,7 +298,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/phrase_delivery_prototype.rs b/crates/utopia-store/tests/phrase_delivery_prototype.rs new file mode 100644 index 000000000..f435c4478 --- /dev/null +++ b/crates/utopia-store/tests/phrase_delivery_prototype.rs @@ -0,0 +1,215 @@ +//! Isolated protocol experiment; 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, + "prototype_materialize_typed", + 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 experiment'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 materialize::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] +async fn delivery_protocol_rollback_busy_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 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),materialize::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()); + blocker.rollback().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?; + anyhow::ensure!(materialize::try_materialize(&pool,kb).await?.is_some()); + 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::try_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=="prototype_materialize_typed"); + anyhow::ensure!(materialize::try_materialize(&pool,kb).await?.is_some()); + 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? + } 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("prototype_materialize_typed"),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 +} From 636370d8296c8ebceecd0dccd7497392b7b5f760 Mon Sep 17 00:00:00 2001 From: dada-yan Date: Mon, 21 Sep 2026 00:42:43 +1000 Subject: [PATCH 2/5] Kill subprocesses at the phrase decision and acknowledgement boundaries Signed-off-by: dada-yan --- .../tests/phrase_delivery_prototype.rs | 134 ++++++++++++++++++ 1 file changed, 134 insertions(+) diff --git a/crates/utopia-store/tests/phrase_delivery_prototype.rs b/crates/utopia-store/tests/phrase_delivery_prototype.rs index f435c4478..2141947c8 100644 --- a/crates/utopia-store/tests/phrase_delivery_prototype.rs +++ b/crates/utopia-store/tests/phrase_delivery_prototype.rs @@ -213,3 +213,137 @@ async fn delivery_protocol_rollback_busy_late_arrivals_recovery_and_cost() -> an 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, + "prototype_materialize_typed", + 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::try_materialize(&pool, kb) + .await + .unwrap() + .unwrap(); + } + std::fs::write(&ready, id.to_string()).unwrap(); + std::future::pending::<()>().await; + }); +} + +struct ChildGuard(std::process::Child); +impl Drop for ChildGuard { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } +} + +#[tokio::test] +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()?); + 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::try_materialize(&pool,kb).await?==Some(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 +} From d730283fdcceada9b8b5108b22768986b80d011e Mon Sep 17 00:00:00 2001 From: dada-yan Date: Mon, 21 Sep 2026 00:51:47 +1000 Subject: [PATCH 3/5] Record the tested phrase delivery proposal and its production gates Signed-off-by: dada-yan --- ...ology-is-a-view-over-what-documents-say.md | 9 ++ scripts/prototypes/phrase-delivery/README.md | 118 ++++++++++++++++++ 2 files changed, 127 insertions(+) create mode 100644 scripts/prototypes/phrase-delivery/README.md 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..06a6b3679 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 @@ -132,3 +132,12 @@ Today's extractor stays as it is until cut 2 passes its thresholds. - Whether derivation rules recover the implicit facts that write-time binding found (Re-DocRED's country and located-in relations are a fifth of its pairs). - Competency questions for a new knowledge base that has none yet. - How much of identity the deterministic evidence settles before the adjudicator is needed. + +## Proposed delivery follow-up · 2026-09-21 (not accepted or implemented) + +[The phrase-delivery experiment](../../scripts/prototypes/phrase-delivery/README.md) +proposes persisting each human phrase decision with its materialization-only job. +It records real database, process-exit and mutation evidence, and the unresolved +HTTP/UI and recovery contract. This does not alter the accepted projection semantics +above and registers no production handler. Agree and land the delivery record before +wiring the public route; unused prototype helpers alone do not resolve #800. diff --git a/scripts/prototypes/phrase-delivery/README.md b/scripts/prototypes/phrase-delivery/README.md new file mode 100644 index 000000000..f8d70249e --- /dev/null +++ b/scripts/prototypes/phrase-delivery/README.md @@ -0,0 +1,118 @@ +# Proposed delivery contract for human phrase decisions + +Status: **isolated prototype; not a production #800 implementation**. No job kind +is registered and no HTTP/UI response changes. Refs #800. This branch exists to +review the protocol before introducing the public API/job contract required by +CONTRIBUTING. Do not deploy or merge the helpers as a standalone solution. + +## 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. + +## Executable evidence + +On the isolated branch, `phrase_bindings::decide_on` factors the existing single SQL +write so the test can compose it with `enqueue_with_max_attempts_tx`. The existing +pool caller remains. `materialize_in_tx` mechanically shares the original body; +`try_materialize` uses the exact existing lock key. No production caller uses the +new entry points. These are experiment support, pending the contract above. + +Run against a **dedicated, otherwise idle test database** (the actual worker recovery +test consumes jobs; never point it at another session's database): + +```sh +export UTOPIA_DATABASE_URL='postgres://.../dedicated_delivery_experiment' +export UTOPIA_TEST_REQUIRE_DB=1 +cargo test --locked -p utopia-store --test phrase_delivery_prototype -- --test-threads=1 --nocapture +``` + +Linux PostgreSQL 16: **two experiment tests passed**. The ignored `crash_child` entry +is an explicit subprocess probe; the parent invokes it three times with per-fixture +identifiers and kills/waits for those children. It is not a silently skipped crash +case. Evidence includes: + +- The real enqueue helper rejects an invalid budget after the binding write; neither + binding nor job commits. This is fault injection at the helper boundary, not a + simulated disk failure during COMMIT. +- A held real advisory lock does not stall acceptance; the same job is Deferred and + the two-connection pool remains available through repeated busy attempts. +- A decision after an old projection's final read has its own job; reverse-order and + duplicate processing preserve final none and the open statement. +- Actual `run_worker` startup reclaims a running job and idempotently recomputes; no + model path is present in the prototype handler. +- Actual OS subprocess termination before commit rolls back both rows; termination + after accept preserves the queued job; termination after projection commit before + ack preserves a running job and recomputation creates no duplicate typed fact. +- Exhausted deferral/failure budget becomes visible failed and explicit scoped + requeue can recover. No sleep guesses which SQL lock is held. + +Three runtime mutations failed their specific assertions: writing the binding via +an independent pool transaction left an orphan decision; treating Busy as success +marked the job done; adding a model-config prerequisite prevented pure recomputation. +Restoring the prototype passes both experiments again. + +## 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. From e83f015f9a3949e53b1ae849b8d6dad0e2c4546e Mon Sep 17 00:00:00 2001 From: dada-yan Date: Mon, 21 Sep 2026 00:52:27 +1000 Subject: [PATCH 4/5] Keep process and queue experiments opt-in on a dedicated database Signed-off-by: dada-yan --- crates/utopia-store/tests/phrase_delivery_prototype.rs | 2 ++ scripts/prototypes/phrase-delivery/README.md | 4 ++-- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/crates/utopia-store/tests/phrase_delivery_prototype.rs b/crates/utopia-store/tests/phrase_delivery_prototype.rs index 2141947c8..aa2a21171 100644 --- a/crates/utopia-store/tests/phrase_delivery_prototype.rs +++ b/crates/utopia-store/tests/phrase_delivery_prototype.rs @@ -64,6 +64,7 @@ async fn status(pool: &PgPool, id: i64) -> anyhow::Result { } #[tokio::test] +#[ignore = "opt-in protocol experiment; requires a dedicated idle database"] async fn delivery_protocol_rollback_busy_late_arrivals_recovery_and_cost() -> anyhow::Result<()> { let Some(url) = utopia_store::test_db::url() else { return Ok(()); @@ -283,6 +284,7 @@ impl Drop for ChildGuard { } #[tokio::test] +#[ignore = "opt-in subprocess experiment; 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(()); diff --git a/scripts/prototypes/phrase-delivery/README.md b/scripts/prototypes/phrase-delivery/README.md index f8d70249e..ed78f1481 100644 --- a/scripts/prototypes/phrase-delivery/README.md +++ b/scripts/prototypes/phrase-delivery/README.md @@ -52,11 +52,11 @@ test consumes jobs; never point it at another session's database): ```sh export UTOPIA_DATABASE_URL='postgres://.../dedicated_delivery_experiment' export UTOPIA_TEST_REQUIRE_DB=1 -cargo test --locked -p utopia-store --test phrase_delivery_prototype -- --test-threads=1 --nocapture +cargo test --locked -p utopia-store --test phrase_delivery_prototype -- --ignored --skip crash_child --test-threads=1 --nocapture ``` Linux PostgreSQL 16: **two experiment tests passed**. The ignored `crash_child` entry -is an explicit subprocess probe; the parent invokes it three times with per-fixture +is an explicit subprocess probe; both parent tests are also opt-in to avoid consuming unrelated test jobs in a shared test database. The command above runs the parents explicitly; the parent invokes it three times with per-fixture identifiers and kills/waits for those children. It is not a silently skipped crash case. Evidence includes: From 7e49b7fd6d555727f169a77d36ab12367bf97003 Mon Sep 17 00:00:00 2001 From: dada-yan Date: Mon, 21 Sep 2026 11:47:10 +1000 Subject: [PATCH 5/5] Retain real delivery regressions without exporting an unused materializer. Signed-off-by: dada-yan --- CONTRIBUTING.md | 21 ++ crates/utopia-store/src/materialize.rs | 22 +- .../src/materialize_delivery_tests.rs | 189 ++++++++++++++++++ crates/utopia-store/src/phrase_bindings.rs | 20 +- ... human_phrase_materialization_delivery.rs} | 65 +++--- ...ology-is-a-view-over-what-documents-say.md | 12 +- ...cision-carries-its-materialization-work.md | 67 ++----- docs/decisions/README.md | 2 + 8 files changed, 277 insertions(+), 121 deletions(-) create mode 100644 crates/utopia-store/src/materialize_delivery_tests.rs rename crates/utopia-store/tests/{phrase_delivery_prototype.rs => human_phrase_materialization_delivery.rs} (84%) rename scripts/prototypes/phrase-delivery/README.md => docs/decisions/0051-a-human-phrase-decision-carries-its-materialization-work.md (57%) 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 4006aa76b..8079ab366 100644 --- a/crates/utopia-store/src/materialize.rs +++ b/crates/utopia-store/src/materialize.rs @@ -71,24 +71,6 @@ pub async fn materialize(pool: &PgPool, kb_id: Uuid) -> AppResult { Ok(outcome) } -/// Isolated delivery prototype: Busy never means materialization completed. -pub 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 materialize_in_tx( tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, kb_id: Uuid, @@ -286,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 6b5f78ce4..a93899f70 100644 --- a/crates/utopia-store/src/phrase_bindings.rs +++ b/crates/utopia-store/src/phrase_bindings.rs @@ -226,17 +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 } -/// Isolated delivery prototype: caller owns the decision/enqueue transaction. -pub async fn decide_on( - connection: &mut sqlx::PgConnection, - kb_id: Uuid, - sig: &PhraseSignature, - d: Decision<'_>, -) -> AppResult { +fn validate_decision(sig: &PhraseSignature, d: &Decision<'_>) -> AppResult { if !matches!(d.status, "bound" | "none" | "undecided") { return Err(AppError::Validation(format!( "unknown binding status {:?}", @@ -264,6 +259,17 @@ pub async fn decide_on( 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, diff --git a/crates/utopia-store/tests/phrase_delivery_prototype.rs b/crates/utopia-store/tests/human_phrase_materialization_delivery.rs similarity index 84% rename from crates/utopia-store/tests/phrase_delivery_prototype.rs rename to crates/utopia-store/tests/human_phrase_materialization_delivery.rs index aa2a21171..c1059866c 100644 --- a/crates/utopia-store/tests/phrase_delivery_prototype.rs +++ b/crates/utopia-store/tests/human_phrase_materialization_delivery.rs @@ -1,4 +1,4 @@ -//! Isolated protocol experiment; no production job kind or HTTP route is registered. +//! 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}; @@ -30,7 +30,7 @@ async fn accept( ); let id = jobs::enqueue_with_max_attempts_tx( &mut tx, - "prototype_materialize_typed", + "test_human_phrase_materialize", json!({"kb_id":kb}), budget, ) @@ -39,21 +39,16 @@ async fn accept( Ok(id) } async fn claim(pool: &PgPool, id: i64) -> anyhow::Result { - // Restrict the production claim SQL to this experiment's job, never steal work. + // 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 materialize::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?; - } + 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 { @@ -64,8 +59,8 @@ async fn status(pool: &PgPool, id: i64) -> anyhow::Result { } #[tokio::test] -#[ignore = "opt-in protocol experiment; requires a dedicated idle database"] -async fn delivery_protocol_rollback_busy_late_arrivals_recovery_and_cost() -> anyhow::Result<()> { +#[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(()); }; @@ -119,17 +114,7 @@ async fn delivery_protocol_rollback_busy_late_arrivals_recovery_and_cost() -> an anyhow::ensure!(count==0); println!("B-T01/T02 PASS same-transaction enqueue failure rolls back decision"); - 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),materialize::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()); - blocker.rollback().await?; + 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); @@ -137,7 +122,7 @@ async fn delivery_protocol_rollback_busy_late_arrivals_recovery_and_cost() -> an // 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?; - anyhow::ensure!(materialize::try_materialize(&pool,kb).await?.is_some()); + 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?; @@ -151,15 +136,15 @@ async fn delivery_protocol_rollback_busy_late_arrivals_recovery_and_cost() -> an // 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::try_materialize(&pool,kb).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=="prototype_materialize_typed"); - anyhow::ensure!(materialize::try_materialize(&pool,kb).await?.is_some()); + anyhow::ensure!(job.kind=="test_human_phrase_materialize"); + materialize::materialize(&pool,kb).await?; sent.send(job.id)?; Ok(()) } @@ -169,7 +154,7 @@ async fn delivery_protocol_rollback_busy_late_arrivals_recovery_and_cost() -> an tokio::time::timeout(Duration::from_secs(5),async { while status(&pool,recovery).await? != "done" {tokio::task::yield_now().await;} anyhow::Ok(()) - }).await? + }).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); @@ -181,7 +166,7 @@ async fn delivery_protocol_rollback_busy_late_arrivals_recovery_and_cost() -> an 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("prototype_materialize_typed"),failed_since:None}).await?==1); + 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)"); @@ -252,7 +237,7 @@ fn crash_child() { .unwrap(); let id = jobs::enqueue_with_max_attempts_tx( &mut tx, - "prototype_materialize_typed", + "test_human_phrase_materialize", json!({"kb_id":kb}), 3, ) @@ -265,26 +250,24 @@ fn crash_child() { tx.commit().await.unwrap(); if phase == "unacked" { claim(&pool, id).await.unwrap(); - materialize::try_materialize(&pool, kb) - .await - .unwrap() - .unwrap(); + materialize::materialize(&pool, kb).await.unwrap(); } std::fs::write(&ready, id.to_string()).unwrap(); std::future::pending::<()>().await; }); } -struct ChildGuard(std::process::Child); +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 experiment; requires a dedicated idle database"] +#[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(()); @@ -312,7 +295,7 @@ async fn process_exit_preserves_the_committed_delivery_boundary() -> anyhow::Res .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()?); + .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::();} @@ -333,7 +316,7 @@ async fn process_exit_preserves_the_committed_delivery_boundary() -> anyhow::Res 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::try_materialize(&pool,kb).await?==Some(materialize::Outcome::default())); + anyhow::ensure!(materialize::materialize(&pool,kb).await?==materialize::Outcome::default()); } println!("OS process kill phase={phase}: PASS"); } 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 06a6b3679..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. @@ -132,12 +135,3 @@ Today's extractor stays as it is until cut 2 passes its thresholds. - Whether derivation rules recover the implicit facts that write-time binding found (Re-DocRED's country and located-in relations are a fifth of its pairs). - Competency questions for a new knowledge base that has none yet. - How much of identity the deterministic evidence settles before the adjudicator is needed. - -## Proposed delivery follow-up · 2026-09-21 (not accepted or implemented) - -[The phrase-delivery experiment](../../scripts/prototypes/phrase-delivery/README.md) -proposes persisting each human phrase decision with its materialization-only job. -It records real database, process-exit and mutation evidence, and the unresolved -HTTP/UI and recovery contract. This does not alter the accepted projection semantics -above and registers no production handler. Agree and land the delivery record before -wiring the public route; unused prototype helpers alone do not resolve #800. diff --git a/scripts/prototypes/phrase-delivery/README.md b/docs/decisions/0051-a-human-phrase-decision-carries-its-materialization-work.md similarity index 57% rename from scripts/prototypes/phrase-delivery/README.md rename to docs/decisions/0051-a-human-phrase-decision-carries-its-materialization-work.md index ed78f1481..84505a655 100644 --- a/scripts/prototypes/phrase-delivery/README.md +++ b/docs/decisions/0051-a-human-phrase-decision-carries-its-materialization-work.md @@ -1,9 +1,12 @@ -# Proposed delivery contract for human phrase decisions +# 0051 · A human phrase decision carries its materialization work -Status: **isolated prototype; not a production #800 implementation**. No job kind -is registered and no HTTP/UI response changes. Refs #800. This branch exists to -review the protocol before introducing the public API/job contract required by -CONTRIBUTING. Do not deploy or merge the helpers as a standalone solution. +- **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 @@ -38,47 +41,19 @@ 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. -## Executable evidence - -On the isolated branch, `phrase_bindings::decide_on` factors the existing single SQL -write so the test can compose it with `enqueue_with_max_attempts_tx`. The existing -pool caller remains. `materialize_in_tx` mechanically shares the original body; -`try_materialize` uses the exact existing lock key. No production caller uses the -new entry points. These are experiment support, pending the contract above. - -Run against a **dedicated, otherwise idle test database** (the actual worker recovery -test consumes jobs; never point it at another session's database): - -```sh -export UTOPIA_DATABASE_URL='postgres://.../dedicated_delivery_experiment' -export UTOPIA_TEST_REQUIRE_DB=1 -cargo test --locked -p utopia-store --test phrase_delivery_prototype -- --ignored --skip crash_child --test-threads=1 --nocapture -``` - -Linux PostgreSQL 16: **two experiment tests passed**. The ignored `crash_child` entry -is an explicit subprocess probe; both parent tests are also opt-in to avoid consuming unrelated test jobs in a shared test database. The command above runs the parents explicitly; the parent invokes it three times with per-fixture -identifiers and kills/waits for those children. It is not a silently skipped crash -case. Evidence includes: - -- The real enqueue helper rejects an invalid budget after the binding write; neither - binding nor job commits. This is fault injection at the helper boundary, not a - simulated disk failure during COMMIT. -- A held real advisory lock does not stall acceptance; the same job is Deferred and - the two-connection pool remains available through repeated busy attempts. -- A decision after an old projection's final read has its own job; reverse-order and - duplicate processing preserve final none and the open statement. -- Actual `run_worker` startup reclaims a running job and idempotently recomputes; no - model path is present in the prototype handler. -- Actual OS subprocess termination before commit rolls back both rows; termination - after accept preserves the queued job; termination after projection commit before - ack preserves a running job and recomputation creates no duplicate typed fact. -- Exhausted deferral/failure budget becomes visible failed and explicit scoped - requeue can recover. No sleep guesses which SQL lock is held. - -Three runtime mutations failed their specific assertions: writing the binding via -an independent pool transaction left an orphan decision; treating Busy as success -marked the job done; adding a model-config prerequisite prevented pure recomputation. -Restoring the prototype passes both experiments again. +## 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 diff --git a/docs/decisions/README.md b/docs/decisions/README.md index 6fc359926..46c78ab72 100644 --- a/docs/decisions/README.md +++ b/docs/decisions/README.md @@ -73,6 +73,7 @@ The test for writing one: if someone (including us) looks at a piece of code in | 0045 | [A time mention is resolved against its document](0045-a-time-mention-is-resolved-against-its-document.md) | Accepted · cuts 1 and 2 built (#740): a document is dated from its own text, each mention is interpreted by the model and computed by code, upload time is used nowhere · cuts 3 and 4 (grades replace the confidence gate, re-resolution and the anchor queue) not built · a time expression is a mention with its words and place; the model returns shape, anchor, offset and granularity and code computes the interval; a document carries its own date, calendars and anchors across chunks, never its upload time; unresolved mentions wait for an anchor; timelines close on resolution grade instead of confidence | | 0046 | [The app surface is MCP](0046-the-app-surface-is-mcp.md) | Decided, with the refused design kept. Asked for an app center: applications built on this knowledge, mounted, run in a sandbox, handed to a team. The answer is that the surface already exists — a personal token carries identity and scope, ten read tools serve chat and MCP from one place, `as_of` reaches every graph read, and a read returns `structuredContent` with stable ledger identities — so a coding agent builds on this base today in its own platform, its own language and its own sandbox. Refused here because the layer an app would read is being replaced under it (typed facts now come only from alignment), because 0016 closes open seams before cutting new ones, and because a catalog, an execution boundary and quotas are three other products. The shape is kept with the four gates it would have to hold (runs as the caller, egress only through a declared action, a declared clock, the existing queue) and the dead ends: a container runtime (withdrawn the day it was written — WeKnora's skills are human-written and assume a shell, and they pay for it), a Wasm component runtime (better on every axis including the determinism re-parse needs, still not built because the reason is priority), a service identity per app, an app as a saved conversation. Reopened by a named customer who needs a button inside the product, by the type layer settling, or after 0034 | | 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 | +| 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 | |---|---|---|---| @@ -123,6 +124,7 @@ The test for writing one: if someone (including us) looks at a piece of code in | 0045 | [A time mention is resolved against its document](0045-a-time-mention-is-resolved-against-its-document.md) | time | current | | 0046 | [The app surface is MCP](0046-the-app-surface-is-mcp.md) | chat-and-mcp | current | | 0047 | [A rule may conclude a relation](0047-a-rule-may-conclude-a-relation.md) | rules | current | +| 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.