A write does not queue a vector index build while one for the same table and dims is running

crates/utopia-store/src/vector_index.rs

@@ -200,7 +200,7 @@ fn is_known(name: &str) -> bool {
 /// 返回 `Some(job id)` = 这一次排上了。
 ///
 /// 索引在的话记在进程里，往后只是一次 HashSet 查找；不在的时候每次写入一次目录
-/// 查询加一次「没排着才排」的插入——建索引那一两分钟里写入照常，代价可忽略。
+/// 查询加一次「没排着、也没在建才排」的插入——建索引的那段时间里写入照常，代价可忽略。
 /// 超过 [`MAX_DIMS`] 的维度什么都不排：建不了，查询走精确路径
 pub async fn request(pool: &PgPool, target: Target, dims: usize) -> AppResult<Option<i64>> {
     if dims == 0 || dims > MAX_DIMS {
@@ -214,10 +214,11 @@ pub async fn request(pool: &PgPool, target: Target, dims: usize) -> AppResult<Op
         remember(&name);
         return Ok(None);
     }
-    crate::jobs::enqueue_unless_queued(
+    crate::jobs::enqueue_unless_pending(
         pool,
         JOB_KIND,
         serde_json::json!({ "table": target.key(), "dims": dims }),
+        std::time::Duration::ZERO,
     )
     .await
 }

crates/utopia-store/tests/a_vector_index_is_built_by_a_job.rs

@@ -79,6 +79,16 @@ async fn the_first_write_of_a_dimension_queues_one_build() -> anyhow::Result<()>
         let second = vector_index::request(&pool, Target::Chunks, DIMS).await?;
         assert_eq!(second, None, "队列里已经有一条，不再排");
         assert_eq!(queued(&pool, DIMS).await?, 1);
+        // worker 认领了它、正在建：再写也不排。排上的每一条都会占一个 worker 名额，
+        // 在建索引的咨询锁上空等到前一条建完（3072 维的库上一建一个多小时）
+        sqlx::query("UPDATE jobs SET status = 'running' WHERE kind = $1 AND payload->>'dims' = $2")
+            .bind(JOB_KIND)
+            .bind(DIMS.to_string())
+            .execute(&pool)
+            .await?;
+        let while_building = vector_index::request(&pool, Target::Chunks, DIMS).await?;
+        assert_eq!(while_building, None, "正在建的时候不再排");
+        assert_eq!(queued(&pool, DIMS).await?, 0);
 
         let built = vector_index::build(&pool, Target::Chunks, DIMS).await?;
         assert!(built.created);