fix(queue): rank reembed_backfill right after seal in claim_next

gitbooks/job-queue.md

@@ -146,14 +146,17 @@ Eligibility is `status='ready' AND available_at_ms <= now AND kind NOT IN (retir
 
 ```text
 ORDER BY CASE kind
-           WHEN 'seal'          THEN 1
-           WHEN 'flush_stale'   THEN 2
-           WHEN 'append_buffer' THEN 3
-           ELSE 4                 -- extract_chunk, reembed_backfill, seal_document
+           WHEN 'seal'             THEN 1
+           WHEN 'reembed_backfill' THEN 2
+           WHEN 'flush_stale'      THEN 3
+           WHEN 'append_buffer'    THEN 4
+           ELSE 5                    -- extract_chunk, seal_document
          END ASC,
          available_at_ms ASC
 ```
 
+`reembed_backfill` sits right behind `seal` because it is the only path that writes chunk vectors (`extract_chunk` no longer embeds inline) and it holds the same single LLM permit as every `extract_chunk`. Ranked with the extraction backlog, the gate-busy defer would round-robin it behind that whole backlog and vectors would trail extraction by the backlog's length. Each backfill step embeds one bounded batch, defers `REEMBED_BACKFILL_REVISIT_MS` (750 ms), and settles `Done` once the space is covered, so extraction is never starved in return.
+
 `DEFAULT_LOCK_DURATION_MS = 5 * 60 * 1000` (5 min) — comfortably larger than any expected single-job runtime, so a crashed worker's row is recovered after the window without leaving real failures stuck for hours. Retry backoff is exponential: `backoff_ms(attempts)` = `min(60s * 2^(attempts-1), 1h)` (`RETRY_BASE_MS = 60s`, `RETRY_CAP_MS = 1h`), so the first retry waits 60s, then 120s, 240s, … capped at one hour.
 
 ## Worker loop

src/memory/queue/store.rs

@@ -107,6 +107,10 @@ pub(crate) fn enqueue_conn(
 /// Sets `status=running`, bumps `attempts`, stamps `started_at_ms` and
 /// `locked_until_ms`. Returns `None` when the queue is empty / not yet due.
 ///
+/// Due rows are ranked by kind — `seal`, then `reembed_backfill`, then
+/// `flush_stale`, then `append_buffer`, then everything else — and by
+/// `available_at_ms` within a rank (the query below explains why).
+///
 /// Retired kinds (`topic_route`, `digest_daily`) are excluded from the claim so
 /// a leftover old-queue row never reaches `row_to_job` (which would fail to
 /// parse it). [`purge_retired_jobs`] removes such rows.
@@ -120,6 +124,18 @@ pub fn claim_next(config: &MemoryConfig, lock_duration_ms: i64) -> Result<Option
                 // Drain forward, don't widen. Most-downstream kinds run first so
                 // a slow LLM-bound `extract_chunk` can't starve the seal pipeline
                 // behind it.
+                //
+                // `reembed_backfill` ranks right after `seal`: it is the only
+                // path that writes chunk vectors (extract no longer embeds
+                // inline) and it shares the single-permit LLM gate with every
+                // `extract_chunk`. Left in the ELSE bucket, the gate-busy defer
+                // round-robins it behind the whole extraction backlog, so
+                // vectors trail extraction by the length of that backlog
+                // (tinyhumansai/tinycortex#168). Ranked here, each freed permit
+                // goes to the backfill while it has work; one step embeds one
+                // bounded batch, defers `REEMBED_BACKFILL_REVISIT_MS`, and
+                // settles `Done` once covered, so extraction is never starved
+                // in return.
                 "UPDATE mem_tree_jobs
                     SET status = 'running',
                         attempts = attempts + 1,
@@ -133,10 +149,11 @@ pub fn claim_next(config: &MemoryConfig, lock_duration_ms: i64) -> Result<Option
                          AND kind NOT IN ('topic_route', 'digest_daily')
                        ORDER BY
                          CASE kind
-                           WHEN 'seal'          THEN 1
-                           WHEN 'flush_stale'   THEN 2
-                           WHEN 'append_buffer' THEN 3
-                           ELSE 4
+                           WHEN 'seal'             THEN 1
+                           WHEN 'reembed_backfill' THEN 2
+                           WHEN 'flush_stale'      THEN 3
+                           WHEN 'append_buffer'    THEN 4
+                           ELSE 5
                          END ASC,
                          available_at_ms ASC
                        LIMIT 1

src/memory/queue/store_tests.rs

@@ -1,7 +1,10 @@
 use super::*;
 use crate::memory::chunks::with_connection;
 use crate::memory::config::MemoryConfig;
-use crate::memory::queue::types::ExtractChunkPayload;
+use crate::memory::queue::types::{
+    AppendBufferPayload, AppendTarget, ExtractChunkPayload, FlushStalePayload, NodeRef,
+    ReembedBackfillPayload, SealDocumentPayload, SealPayload,
+};
 use tempfile::TempDir;
 
 fn test_config() -> (TempDir, MemoryConfig) {
@@ -164,3 +167,124 @@ fn is_retired_kind_recognises_legacy_strings() {
     assert!(!is_retired_kind("extract_chunk"));
     assert!(!is_retired_kind("seal"));
 }
+
+/// tinyhumansai/tinycortex#168: the deduped `reembed_backfill` row is the only
+/// writer of chunk vectors and shares the LLM gate with every `extract_chunk`.
+/// It must be claimed ahead of an *older* due `extract_chunk`, otherwise the
+/// gate-busy defer round-robins it behind the whole extraction backlog.
+#[test]
+fn claim_next_prefers_reembed_backfill_over_older_extract_chunk() {
+    let (_tmp, cfg) = test_config();
+    let now_ms = Utc::now().timestamp_millis();
+
+    let mut extract = NewJob::extract_chunk(&ExtractChunkPayload {
+        chunk_id: "c-older".into(),
+    })
+    .unwrap();
+    // Due well before the backfill row, so an age-ordered claim would pick it.
+    extract.available_at_ms = Some(now_ms - 5_000);
+    let extract_id = enqueue(&cfg, &extract).unwrap().expect("inserted");
+
+    let backfill = NewJob::reembed_backfill(&ReembedBackfillPayload {
+        signature: "provider=test;model=x;dims=3".into(),
+    })
+    .unwrap();
+    let backfill_id = enqueue(&cfg, &backfill).unwrap().expect("inserted");
+
+    let first = claim_next(&cfg, DEFAULT_LOCK_DURATION_MS).unwrap().unwrap();
+    assert_eq!(first.id, backfill_id);
+    assert_eq!(first.kind, JobKind::ReembedBackfill);
+
+    let second = claim_next(&cfg, DEFAULT_LOCK_DURATION_MS).unwrap().unwrap();
+    assert_eq!(second.id, extract_id);
+    assert_eq!(second.kind, JobKind::ExtractChunk);
+
+    assert!(claim_next(&cfg, DEFAULT_LOCK_DURATION_MS)
+        .unwrap()
+        .is_none());
+}
+
+/// Pins the whole claim ladder: `seal` > `reembed_backfill` > `flush_stale` >
+/// `append_buffer` > everything else, and `available_at_ms` (oldest first)
+/// only inside a rank. Rows are enqueued in the reverse of the expected claim
+/// order with strictly *older* due times, so a FIFO / age-ordered claim would
+/// return them in enqueue order and fail.
+#[test]
+fn claim_next_ranks_seal_then_backfill_then_flush_then_append_then_age() {
+    let (_tmp, cfg) = test_config();
+    let now_ms = Utc::now().timestamp_millis();
+
+    // (job, age_ms): a larger age is an older, earlier-due row.
+    let mut jobs = [
+        (
+            NewJob::extract_chunk(&ExtractChunkPayload {
+                chunk_id: "c1".into(),
+            })
+            .unwrap(),
+            60_000,
+        ),
+        (
+            NewJob::seal_document(&SealDocumentPayload {
+                tree_scope: "gmail:acct".into(),
+                doc_id: "doc-1".into(),
+                version_ms: None,
+                chunk_ids: vec!["c1".into()],
+            })
+            .unwrap(),
+            50_000,
+        ),
+        (
+            NewJob::append_buffer(&AppendBufferPayload {
+                node: NodeRef::Leaf {
+                    chunk_id: "c1".into(),
+                },
+                target: AppendTarget::Source {
+                    source_id: "src-1".into(),
+                },
+            })
+            .unwrap(),
+            40_000,
+        ),
+        (
+            NewJob::flush_stale(&FlushStalePayload::default(), "2026-09-07", 4).unwrap(),
+            30_000,
+        ),
+        (
+            NewJob::reembed_backfill(&ReembedBackfillPayload {
+                signature: "provider=test;model=x;dims=3".into(),
+            })
+            .unwrap(),
+            20_000,
+        ),
+        (
+            NewJob::seal(&SealPayload {
+                tree_id: "tree:1".into(),
+                level: 0,
+                force_now_ms: None,
+            })
+            .unwrap(),
+            10_000,
+        ),
+    ];
+    for (job, age_ms) in jobs.iter_mut() {
+        job.available_at_ms = Some(now_ms - *age_ms);
+        enqueue(&cfg, job).unwrap().expect("inserted");
+    }
+
+    let mut claimed = Vec::new();
+    while let Some(job) = claim_next(&cfg, DEFAULT_LOCK_DURATION_MS).unwrap() {
+        claimed.push(job.kind);
+    }
+    assert_eq!(
+        claimed,
+        [
+            JobKind::Seal,
+            JobKind::ReembedBackfill,
+            JobKind::FlushStale,
+            JobKind::AppendBuffer,
+            // ELSE bucket: oldest `available_at_ms` first.
+            JobKind::ExtractChunk,
+            JobKind::SealDocument,
+        ]
+    );
+}