diff --git a/crates/utopia-cli/src/main.rs b/crates/utopia-cli/src/main.rs index 1d93a525b..2c85aaded 100644 --- a/crates/utopia-cli/src/main.rs +++ b/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(); diff --git a/crates/utopia-store/tests/a_name_embedding_runs_while_model_work_holds_every_slot.rs b/crates/utopia-store/tests/a_name_embedding_runs_while_model_work_holds_every_slot.rs new file mode 100644 index 000000000..bc8409842 --- /dev/null +++ b/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 +} diff --git a/crates/utopia-store/tests/store/main.rs b/crates/utopia-store/tests/store/main.rs index 47874c03a..923e33415 100644 --- a/crates/utopia-store/tests/store/main.rs +++ b/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; diff --git a/migrations/9003_embed_names_runs_on_the_fast_lane.sql b/migrations/9003_embed_names_runs_on_the_fast_lane.sql new file mode 100644 index 000000000..bc37ceb55 --- /dev/null +++ b/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 +$$;