design: validate durable delivery of human phrase decisions

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.

crates/utopia-store/src/materialize.rs

@@ -66,6 +66,15 @@ pub async fn materialize(pool: &PgPool, kb_id: Uuid) -> AppResult<Outcome> {
         .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<Outcome> {
     // 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<Outcome> {
                                     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<Outcome> {
             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<Outcome> {
           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<Outcome> {
         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<Outcome> {
             }
             (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<Outcome> {
             }
             (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<Outcome> {
             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,15 +206,15 @@ pub async fn materialize(pool: &PgPool, kb_id: Uuid) -> AppResult<Outcome> {
                  ON CONFLICT DO NOTHING",
             )
             .bind(fact)
-            .execute(&mut *tx)
+            .execute(&mut **tx)
             .await?;
             sqlx::query(
                 "DELETE FROM typed_fact_sources src
                   USING facts n
                   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<Outcome> {
         )
         .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<Outcome> {
         )
         .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<Outcome> {
         )
         .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<i64> {
     .fetch_one(pool)
     .await?)
 }
+
+#[cfg(test)]
+#[path = "materialize_delivery_tests.rs"]
+mod delivery_tests;

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<T>(tokio::task::JoinHandle<T>);
+impl<T> Drop for AbortOnDrop<T> {
+    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<Option<Outcome>> {
+    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<Uuid>,
+    budget: i32,
+) -> anyhow::Result<i64> {
+    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<jobs::Job> {
+    // 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<String> {
+    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
+}

crates/utopia-store/src/phrase_bindings.rs

@@ -226,6 +226,12 @@ pub async fn decide(
     sig: &PhraseSignature,
     d: Decision<'_>,
 ) -> AppResult<bool> {
+    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<String> {
     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<bool> {
+    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)
 }