embed_names runs on the fast lane, so name vectors keep up with extraction

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 = 107;
+const CURRENT_SCHEMA_VERSION: u32 = 108;
 
 fn main() -> anyhow::Result<()> {
     dotenvy::dotenv().ok();

crates/utopia-store/tests/a_name_embedding_runs_while_model_work_holds_every_slot.rs

@@ -0,0 +1,68 @@
+//! re:cinq fork：`embed_names` 只调嵌入模型，常规名额被抽取占满时保留通道照样认领它（迁移 9003）。
+//!
+//! 起真实的 `run_worker`，所以自成一个二进制。处理器只认带本测试标记的任务。
+
+use std::sync::atomic::AtomicUsize;
+use std::sync::Arc;
+use std::time::Duration;
+use utopia_store::jobs;
+
+#[tokio::test]
+async fn a_name_embedding_runs_while_model_work_holds_every_slot() -> anyhow::Result<()> {
+    let Some(url) = utopia_store::test_db::url() else {
+        return Ok(());
+    };
+    let pool = sqlx::PgPool::connect(&url).await?;
+    let priority: i16 = sqlx::query_scalar("SELECT job_priority('embed_names')")
+        .fetch_one(&pool)
+        .await?;
+    assert_eq!(priority, 0);
+    let mark = uuid::Uuid::now_v7().to_string();
+    let (started, mut extracting) = tokio::sync::mpsc::unbounded_channel();
+    let (finished, mut ran) = tokio::sync::mpsc::unbounded_channel();
+    let ours = mark.clone();
+    let worker = tokio::spawn(jobs::run_worker(
+        pool.clone(),
+        Arc::new(AtomicUsize::new(1)),
+        move |job| {
+            let (ours, started, finished) = (ours.clone(), started.clone(), finished.clone());
+            async move {
+                anyhow::ensure!(
+                    job.payload["lane_test"] == ours.as_str(),
+                    "not this test's job"
+                );
+                if job.kind == "extract_document" {
+                    started.send(())?;
+                    std::future::pending::<()>().await;
+                }
+                finished.send(job.kind)?;
+                Ok(())
+            }
+        },
+    ));
+    let queue = |kind: &'static str| {
+        sqlx::query("INSERT INTO jobs (kind, payload, run_at) VALUES ($1, $2, now() - interval '100 years')")
+            .bind(kind)
+            .bind(serde_json::json!({ "lane_test": mark }))
+            .execute(&pool)
+    };
+    let result = async {
+        queue("extract_document").await?;
+        tokio::time::timeout(Duration::from_secs(10), extracting.recv()).await?;
+        queue("embed_names").await?;
+        let embedded = tokio::time::timeout(Duration::from_secs(10), ran.recv()).await;
+        anyhow::ensure!(
+            matches!(&embedded, Ok(Some(kind)) if kind == "embed_names"),
+            "the name embedding waited behind the extraction: {embedded:?}"
+        );
+        anyhow::Ok(())
+    }
+    .await;
+    worker.abort();
+    let _ = worker.await;
+    sqlx::query("DELETE FROM jobs WHERE payload->>'lane_test' = $1")
+        .bind(&mark)
+        .execute(&pool)
+        .await?;
+    result
+}

crates/utopia-store/tests/store/main.rs

@@ -27,6 +27,7 @@
 //!   （比如 lib 单元测试里的 `phrase_bindings_delivery_tests`）就会多数出一个。
 //! - `a_sync_runs_while_model_work_holds_every_slot`：同样起真实的 `run_worker`，同样会认领
 //!   库里任何排队任务，不是自己的就报错。
+//! - `a_name_embedding_runs_while_model_work_holds_every_slot`：同上，换成 `embed_names`。
 //!
 //! 新加测试默认放这个目录；只有碰上面那类全局状态时才放顶层，并在这里补一行为什么。
 mod a_batch_decides_like_a_person;

migrations/9003_embed_names_runs_on_the_fast_lane.sql

@@ -0,0 +1,24 @@
+-- re:cinq fork only (RE-CINQ.md: fork migrations are numbered from 9000 and idempotent).
+--
+-- embed_names（每个库一份、补名字向量）只调嵌入模型、不调对话模型，和 embed_ontology 一样
+-- 属于不等对话模型的任务。0116 的 job_priority 没列它，它按 1 排在抽取后面，保留通道也不认领。
+--
+-- jobs_priority_claim_idx 是 job_priority(kind) 的表达式索引，IMMUTABLE 函数的结果在建索引时
+-- 就存下了，改函数体不会让已有索引条目重算；所以函数一变就必须重建索引。
+-- 只在函数还没列 embed_names 时才改并重建，第二次运行什么都不动。没有存储列，已排队的行不用补。
+DO $$
+BEGIN
+    IF (SELECT prosrc FROM pg_proc WHERE proname = 'job_priority') LIKE '%embed_names%' THEN
+        RETURN;
+    END IF;
+    CREATE OR REPLACE FUNCTION job_priority(kind text) RETURNS smallint
+    LANGUAGE sql IMMUTABLE PARALLEL SAFE AS $f$
+        SELECT CASE WHEN kind IN ('sync_source', 'process_document', 'hydrate_rss_entry',
+                                  'embed_ontology', 'memory_ingest', 'build_vector_index',
+                                  'materialize_inferences', 'noop', 'embed_names')
+                    THEN 0::smallint ELSE 1::smallint END
+    $f$;
+    DROP INDEX IF EXISTS jobs_priority_claim_idx;
+    CREATE INDEX jobs_priority_claim_idx ON jobs (job_priority(kind), run_at) WHERE status = 'queued';
+END
+$$;