From 0f90734089e52a5b016f26384486e14747f795c1 Mon Sep 17 00:00:00 2001 From: Aschenbath Date: Wed, 21 Oct 2026 14:45:13 +0000 Subject: [PATCH] fix: let human phrase decisions materialize ahead of model work Signed-off-by: Aschenbath --- crates/utopia-cli/src/main.rs | 2 +- crates/utopia-store/src/jobs_claim_tests.rs | 96 ++++++++++++++++--- ..._runs_while_model_work_holds_every_slot.rs | 23 +++-- ...human_phrase_materialization_is_intake.sql | 12 +++ 4 files changed, 110 insertions(+), 23 deletions(-) create mode 100644 migrations/0118_human_phrase_materialization_is_intake.sql diff --git a/crates/utopia-cli/src/main.rs b/crates/utopia-cli/src/main.rs index 0d41dcf34..8f0fe0364 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 = 104; +const CURRENT_SCHEMA_VERSION: u32 = 105; fn main() -> anyhow::Result<()> { dotenvy::dotenv().ok(); diff --git a/crates/utopia-store/src/jobs_claim_tests.rs b/crates/utopia-store/src/jobs_claim_tests.rs index e1822a3fe..c7aa60010 100644 --- a/crates/utopia-store/src/jobs_claim_tests.rs +++ b/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 = 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 = 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> = 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(()) } diff --git a/crates/utopia-store/tests/a_sync_runs_while_model_work_holds_every_slot.rs b/crates/utopia-store/tests/a_sync_runs_while_model_work_holds_every_slot.rs index 51df79ec3..f236278ac 100644 --- a/crates/utopia-store/tests/a_sync_runs_while_model_work_holds_every_slot.rs +++ b/crates/utopia-store/tests/a_sync_runs_while_model_work_holds_every_slot.rs @@ -1,7 +1,7 @@ -//! #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; @@ -9,7 +9,7 @@ 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; diff --git a/migrations/0118_human_phrase_materialization_is_intake.sql b/migrations/0118_human_phrase_materialization_is_intake.sql new file mode 100644 index 000000000..f5c2e239b --- /dev/null +++ b/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;