Intake jobs are claimed before model work, so a sync no longer waits behind the extraction backlog

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

crates/utopia-store/src/jobs.rs

@@ -371,14 +371,17 @@ pub async fn status_in_kb(pool: &PgPool, kb_id: Uuid, id: i64) -> AppResult<Opti
 }
 
 /// 认领一个到期任务；没有则返回 None。
+///
+/// 不调对话模型的任务排在前面（`job_priority`，迁移 0116，#1300）：同步、文档入库这些几秒
+/// 的活，不该等在几百个更早的抽取后面
 async fn claim_one(pool: &PgPool) -> AppResult<Option<Job>> {
     let job = sqlx::query_as(
         "UPDATE jobs SET status = 'running', locked_at = now(),
                 attempts = attempts + 1, updated_at = now()
          WHERE id = (
              SELECT id FROM jobs
              WHERE status = 'queued' AND run_at <= now()
-             ORDER BY run_at
+             ORDER BY job_priority(kind), run_at
              FOR UPDATE SKIP LOCKED
              LIMIT 1
          )
@@ -675,6 +678,10 @@ where
 #[path = "jobs_writeback_tests.rs"]
 mod writeback_tests;
 
+#[cfg(test)]
+#[path = "jobs_claim_tests.rs"]
+mod claim_tests;
+
 #[cfg(test)]
 mod tests {
     use super::{deferred_retry_secs, retry_delay};

crates/utopia-store/src/jobs_claim_tests.rs

@@ -0,0 +1,30 @@
+//! 认领顺序（#1300）：不调对话模型的任务排在模型任务前面，哪怕它来得晚。
+//!
+//! `claim_one` 认领的是全库的任务，所以这两条放在一百年前，排在库里别的一切前面。
+
+use super::claim_one;
+use sqlx::PgPool;
+
+#[tokio::test]
+async fn a_later_sync_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
+               ('extract_document', '{"test": "claim-order"}', now() - interval '101 years'),
+               ('sync_source',      '{"test": "claim-order"}', now() - interval '100 years')
+           RETURNING id"#,
+    )
+    .fetch_all(&pool)
+    .await?;
+
+    let claimed = claim_one(&pool).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("sync_source"));
+    Ok(())
+}

migrations/0116_intake_jobs_are_claimed_before_model_work.sql

@@ -0,0 +1,20 @@
+-- 不调对话模型的任务先认领（#1300）。
+--
+-- 认领原来只按 run_at，所有种类排一条队。几百篇文档等抽取的时候，整点的 sync_source、
+-- 新同步进来那批文档的 process_document 都要等前面每一个更早的 extract_document 跑完；
+-- 本地模型上那是好几天——来源十个小时没同步，423 篇新文档一直 pending、搜不到。
+--
+-- 优先级从 kind 算出来，排任务的地方一处不用改：不调对话模型的是 0，其余是 1。
+-- embed_ontology、memory_ingest 等的是嵌入闸门，process_document 可能调 OCR 或转写，
+-- 都不等对话模型。名单只在这个函数里，认领的 ORDER BY 和下面的索引都读它；要改名单，
+-- 在新的迁移里 CREATE OR REPLACE 它并重建索引。没有加一列：jobs 留着全部跑完的任务，
+-- 加存储列要在启动迁移时重写整张表
+CREATE 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', 'noop')
+                THEN 0::smallint ELSE 1::smallint END
+$$;
+
+CREATE INDEX jobs_priority_claim_idx ON jobs (job_priority(kind), run_at) WHERE status = 'queued';