A human phrase decision materializes ahead of queued model work

crates/utopia-cli/src/main.rs

@@ -82,7 +82,7 @@ struct ManifestDataDir {
 /// not a side effect of a code change.
 // 是迁移文件的**个数**，不是最大的编号（守卫 `schema_version_policy_compares_against_current`
 // 按个数比）：编号有空缺时两者不同——0071 由一个开放 PR 占着，0072 先落，个数是 71
-const CURRENT_SCHEMA_VERSION: u32 = 104;
+const CURRENT_SCHEMA_VERSION: u32 = 105;
 
 fn main() -> anyhow::Result<()> {
     dotenvy::dotenv().ok();

crates/utopia-store/src/jobs_claim_tests.rs

@@ -1,30 +1,100 @@
-//! 认领顺序（#1300）：不调对话模型的任务排在模型任务前面，哪怕它来得晚。
+//! 认领顺序（#1300、#1351）：不调对话模型的任务排在模型任务前面，哪怕它来得晚。
 //!
 //! `claim_one` 认领的是全库的任务，所以这两条放在一百年前，排在库里别的一切前面。
 
 use super::claim_one;
-use sqlx::PgPool;
+use sqlx::{Connection, PgConnection, PgPool};
 
 #[tokio::test]
-async fn a_later_sync_is_claimed_before_an_earlier_extraction() -> anyhow::Result<()> {
+async fn a_later_intake_job_is_claimed_before_an_earlier_extraction() -> anyhow::Result<()> {
     let Some(url) = crate::test_db::url() else {
         return Ok(());
     };
     let pool = PgPool::connect(&url).await?;
-    let ids: Vec<i64> = sqlx::query_scalar(
-        r#"INSERT INTO jobs (kind, payload, run_at) VALUES
+    for kind in ["sync_source", crate::phrase_bindings::MATERIALIZE_KIND] {
+        let ids: Vec<i64> = sqlx::query_scalar(
+            r#"INSERT INTO jobs (kind, payload, run_at) VALUES
                ('extract_document', '{"test": "claim-order"}', now() - interval '101 years'),
-               ('sync_source',      '{"test": "claim-order"}', now() - interval '100 years')
+               ($1,                 '{"test": "claim-order"}', now() - interval '100 years')
            RETURNING id"#,
-    )
-    .fetch_all(&pool)
-    .await?;
+        )
+        .bind(kind)
+        .fetch_all(&pool)
+        .await?;
+
+        let claimed = claim_one(&pool, false).await?.map(|job| job.kind);
+        sqlx::query("DELETE FROM jobs WHERE id = ANY($1)")
+            .bind(&ids)
+            .execute(&pool)
+            .await?;
+        assert_eq!(claimed.as_deref(), Some(kind));
+    }
+    Ok(())
+}
+
+#[tokio::test]
+async fn an_upgrade_reindexes_already_queued_materialization() -> anyhow::Result<()> {
+    let Some(url) = crate::test_db::url() else {
+        return Ok(());
+    };
+    let pool = PgPool::connect(&url).await?;
+    let schema = format!("claim_upgrade_{}", uuid::Uuid::now_v7().simple());
+    let result: anyhow::Result<Vec<String>> = async {
+        let mut tx = pool.begin().await?;
+        // 旧函数和旧索引只装进这个 schema，不改并发测试正在用的队列。
+        sqlx::raw_sql(&format!(
+            "CREATE SCHEMA {schema};
+             SET LOCAL search_path TO {schema};
+             CREATE TABLE jobs (kind text, run_at timestamptz, status text DEFAULT 'queued');"
+        ))
+        .execute(&mut *tx)
+        .await?;
+        sqlx::raw_sql(include_str!(
+            "../../../migrations/0116_intake_jobs_are_claimed_before_model_work.sql"
+        ))
+        .execute(&mut *tx)
+        .await?;
+        sqlx::query(
+            "INSERT INTO jobs (kind, run_at) VALUES
+               ('extract_document', now() - interval '1 day'),
+               ($1, now())",
+        )
+        .bind(crate::phrase_bindings::MATERIALIZE_KIND)
+        .execute(&mut *tx)
+        .await?;
+        // 旧数据在升级前已经提交；新进程迁移时没有旧函数的内联计划缓存。
+        tx.commit().await?;
+        let mut upgrade = PgConnection::connect(&url).await?;
+        sqlx::raw_sql(&format!(
+            "SET search_path TO {schema}; {}",
+            include_str!("../../../migrations/0118_human_phrase_materialization_is_intake.sql")
+        ))
+        .execute(&mut upgrade)
+        .await?;
+        upgrade.close().await?;
+
+        // 认领由升级后的新 worker 连接执行，直接验证已有队列的索引键。
+        let mut fresh = PgConnection::connect(&url).await?;
+        sqlx::raw_sql(&format!(
+            "SET search_path TO {schema};
+             SET enable_seqscan = off; SET enable_bitmapscan = off;"
+        ))
+        .execute(&mut fresh)
+        .await?;
+        let eligible = sqlx::query_scalar(
+            "SELECT kind FROM jobs WHERE status = 'queued' AND job_priority(kind) = 0
+             ORDER BY job_priority(kind), run_at",
+        )
+        .fetch_all(&mut fresh)
+        .await?;
+        fresh.close().await?;
+        Ok(eligible)
+    }
+    .await;
 
-    let claimed = claim_one(&pool, false).await?.map(|job| job.kind);
-    sqlx::query("DELETE FROM jobs WHERE id = ANY($1)")
-        .bind(&ids)
+    sqlx::query(&format!("DROP SCHEMA IF EXISTS {schema} CASCADE"))
         .execute(&pool)
         .await?;
-    assert_eq!(claimed.as_deref(), Some("sync_source"));
+    assert_eq!(result?, vec![crate::phrase_bindings::MATERIALIZE_KIND]);
     Ok(())
 }

crates/utopia-store/tests/a_sync_runs_while_model_work_holds_every_slot.rs

@@ -1,15 +1,15 @@
-//! #1300：常规名额全被调模型的任务占着时，后来的同步照样有名额跑（`jobs::NO_CHAT_LANE`）。
+//! #1300、#1351：常规名额全被调模型的任务占着时，同步和人判定后的物化照样有名额跑。
 //!
 //! 起真实的 `run_worker`，它会认领库里任何排队任务，所以自成一个二进制（见
-//! `tests/store/main.rs`）。处理器只认带本测试标记的两条，别人的任务一概报错，不替人干活。
+//! `tests/store/main.rs`）。处理器只认带本测试标记的任务，别人的任务一概报错，不替人干活。
 
 use std::sync::atomic::AtomicUsize;
 use std::sync::Arc;
 use std::time::Duration;
 use utopia_store::jobs;
 
 #[tokio::test]
-async fn a_sync_runs_while_model_work_holds_every_slot() -> anyhow::Result<()> {
+async fn intake_runs_while_model_work_holds_every_slot() -> anyhow::Result<()> {
     let Some(url) = utopia_store::test_db::url() else {
         return Ok(());
     };
@@ -48,12 +48,17 @@ async fn a_sync_runs_while_model_work_holds_every_slot() -> anyhow::Result<()> {
     let result = async {
         queue("extract_document").await?;
         tokio::time::timeout(Duration::from_secs(10), extracting.recv()).await?;
-        queue("sync_source").await?;
-        let synced = tokio::time::timeout(Duration::from_secs(10), ran.recv()).await;
-        anyhow::ensure!(
-            matches!(&synced, Ok(Some(kind)) if kind == "sync_source"),
-            "the sync waited behind the extraction: {synced:?}"
-        );
+        for kind in [
+            "sync_source",
+            utopia_store::phrase_bindings::MATERIALIZE_KIND,
+        ] {
+            queue(kind).await?;
+            let claimed = tokio::time::timeout(Duration::from_secs(10), ran.recv()).await;
+            anyhow::ensure!(
+                matches!(&claimed, Ok(Some(claimed_kind)) if claimed_kind == kind),
+                "{kind} waited behind the extraction: {claimed:?}"
+            );
+        }
         anyhow::Ok(())
     }
     .await;

migrations/0118_human_phrase_materialization_is_intake.sql

@@ -0,0 +1,12 @@
+-- 人判定短语后的物化只读写库，不等模型，不能排在抽取积压后面（#1351）。
+-- 优先级 0 也让它能用 #1349 留给非对话任务的两个名额。
+CREATE OR REPLACE FUNCTION job_priority(kind text) RETURNS smallint
+LANGUAGE sql IMMUTABLE PARALLEL SAFE AS $$
+    SELECT CASE WHEN kind IN ('sync_source', 'process_document', 'hydrate_rss_entry',
+                              'embed_ontology', 'memory_ingest', 'build_vector_index',
+                              'materialize_inferences', 'materialize_typed', 'noop')
+                THEN 0::smallint ELSE 1::smallint END
+$$;
+
+-- IMMUTABLE 函数改了不会更新已经存下的索引键；旧的物化任务也必须按新优先级被找到。
+REINDEX INDEX jobs_priority_claim_idx;