registry: fix five silent cue/BM25 cache decay paths (#563)

.github/workflows/mutants-diff.yml

@@ -47,11 +47,17 @@ permissions:
 
 env:
   CARGO_TERM_COLOR: always
-  # Starting value, to be recalibrated once a few PRs have run. The
-  # issue's sizing: a normal PR is a handful to a few dozen mutants;
-  # anything bigger is new-module territory that a dispatched sweep
-  # covers in shards.
-  MUTANT_BUDGET: 60
+  # Recalibrated from the original 60 after PR #574: at this job's
+  # actual rate (non-incremental, --jobs 3 — see the mutants profile
+  # note in Cargo.toml), 21-22 mutants took 32min, but 56 mutants blew
+  # past the 60min job ceiling and got cancelled mid-run with 13 still
+  # untested (an infrastructure timeout, not a missed-mutant finding —
+  # see the header note on what turns this job red). ~1.6min/mutant
+  # plus ~5min of fixed baseline/setup/report overhead puts 25 mutants
+  # at ~45min, comfortably inside the 60min ceiling with headroom for
+  # a slower/noisier runner. Anything bigger is new-module territory
+  # that a dispatched sweep covers in shards.
+  MUTANT_BUDGET: 25
 
 jobs:
   mutants:

CLAUDE.md

@@ -6,11 +6,11 @@
 
 ## Pre-PR Mutation Gate
 
-- Before `gh pr create` — and before pushing fix commits to an existing PR — run the same diff-scoped mutation check CI runs (mutants-diff.yml), so missed mutants are resolved locally instead of surfacing as a CI round-trip. Fetch first so the diff matches the PR's actual base, and count before running — the full run only happens within CI's per-PR budget (60):
+- Before `gh pr create` — and before pushing fix commits to an existing PR — run the same diff-scoped mutation check CI runs (mutants-diff.yml), so missed mutants are resolved locally instead of surfacing as a CI round-trip. Fetch first so the diff matches the PR's actual base, and count before running — the full run only happens within CI's per-PR budget (25, recalibrated on #574 — see mutants-diff.yml's `MUTANT_BUDGET` comment for the measured per-mutant rate behind the number):
   ```sh
   git fetch origin main &&
     diff=$(mktemp) && git diff origin/main...HEAD > "$diff" &&
-    if [ "$(cargo mutants --in-diff "$diff" --list | wc -l)" -le 60 ]; then
+    if [ "$(cargo mutants --in-diff "$diff" --list | wc -l)" -le 25 ]; then
       CARGO_INCREMENTAL=1 cargo mutants --profile=mutants --in-diff "$diff" --jobs 4
     else
       echo "over budget: dispatch a module sweep (mutants-sweep.yml) instead"

src/bm25.rs

@@ -178,16 +178,26 @@ impl Bm25Index {
     /// lexical mirror of the vector lane's question rows, so a
     /// question-shaped query lands on its answer-shaped paragraph even
     /// on a deployment with no embedding provider at all.
-    pub(crate) fn upsert_source(&mut self, source: &str, record: &PassageRecord) {
+    ///
+    /// Returns whether this call actually changed the index — a real
+    /// tombstone, a paragraph appended, or both. `false` only for the
+    /// otherwise-inert case CodeRabbit caught on #574 (issue #563 item
+    /// 2's own review): an empty or whitespace-only `record` upserted
+    /// for a source with nothing live to tombstone either — `intern`
+    /// still registers the source name, but nothing search-observable
+    /// moved, so callers deciding whether to mark the index dirty must
+    /// not read this as a change.
+    pub(crate) fn upsert_source(&mut self, source: &str, record: &PassageRecord) -> bool {
         let source_id = self.intern(source);
-        self.tombstone(source_id);
+        let mut changed = self.tombstone(source_id);
         let slot_list = self.by_source.entry(source_id).or_default();
         // The record's questions are sorted by paragraph, so one cursor
         // walks them in lockstep with the paragraphs — O(paragraphs +
         // questions), and the terms and the question hash come out of
         // the same pass.
         let mut questions = record.questions.iter().peekable();
         for (span, text) in record.paragraph_texts() {
+            changed = true;
             let slot = self.slots.len() as u32;
             let mut frequencies: HashMap<u64, f32> = HashMap::new();
             let mut length = 0f32;
@@ -222,17 +232,27 @@ impl Bm25Index {
             self.live_total_length += f64::from(length);
         }
         self.reclaim_if_due();
+        changed
     }
 
-    /// Tombstones one source's paragraphs (a retraction).
-    pub(crate) fn remove_source(&mut self, source: &str) {
-        if let Some(&source_id) = self.source_ids.get(source) {
-            self.tombstone(source_id);
-            self.reclaim_if_due();
-        }
+    /// Tombstones one source's paragraphs (a retraction). Returns
+    /// whether any slot was actually live to tombstone — false for a
+    /// never-interned source or one already fully dead, which callers
+    /// (`AppState::refresh_bm25`, issue #563 item 2) need to tell apart
+    /// from a real change: retracting a source this index never held
+    /// anything for must not mark the index dirty.
+    pub(crate) fn remove_source(&mut self, source: &str) -> bool {
+        let Some(&source_id) = self.source_ids.get(source) else {
+            return false;
+        };
+        let changed = self.tombstone(source_id);
+        self.reclaim_if_due();
+        changed
     }
 
-    fn tombstone(&mut self, source_id: u32) {
+    /// Returns whether any slot flipped from alive to dead.
+    fn tombstone(&mut self, source_id: u32) -> bool {
+        let mut changed = false;
         if let Some(slot_list) = self.by_source.get_mut(&source_id) {
             for &slot in slot_list.iter() {
                 let slot = &mut self.slots[slot as usize];
@@ -241,10 +261,12 @@ impl Bm25Index {
                     self.live_count -= 1;
                     self.live_total_length -= f64::from(slot.length);
                     self.dead_count += 1;
+                    changed = true;
                 }
             }
             slot_list.clear();
         }
+        changed
     }
 
     /// In-place tombstone reclamation: rebuild the whole structure from
@@ -1013,6 +1035,61 @@ mod tests {
         );
     }
 
+    /// Issue #563 item 2: `AppState::refresh_bm25` uses this return to
+    /// decide whether a retraction actually changed the resident index
+    /// — wrong here means the sidecar gets rewritten on every flush
+    /// tick even when nothing moved. Three shapes: a source never
+    /// interned, a source already fully tombstoned, and a source with
+    /// live paragraphs still to kill.
+    #[test]
+    fn remove_source_reports_whether_it_actually_tombstoned_anything() {
+        let records = vec![("a".to_string(), record("霧沢町の湧き水。"))];
+        let mut index = Bm25Index::build(&records);
+
+        assert!(
+            !index.remove_source("never-interned"),
+            "a source this index never saw must report no change"
+        );
+        assert!(
+            index.remove_source("a"),
+            "a source with live paragraphs must report a change"
+        );
+        assert!(
+            !index.remove_source("a"),
+            "retracting an already-tombstoned source a second time must report no change"
+        );
+    }
+
+    /// Caught in review on #574 (issue #563 item 2 itself): an empty
+    /// or whitespace-only `PassageRecord` — a legitimate submission,
+    /// `PassageStore` accepts one — has zero paragraphs, so upserting
+    /// it for a source with nothing live to tombstone either leaves
+    /// the index untouched. `AppState::refresh_bm25`'s dirty gate
+    /// trusts this return now instead of assuming every `Some(record)`
+    /// arm is a change.
+    #[test]
+    fn upsert_source_reports_no_change_for_an_empty_record_with_nothing_to_tombstone() {
+        let mut index = Bm25Index::empty();
+
+        assert!(
+            !index.upsert_source("empty", &record("")),
+            "a brand-new source with zero paragraphs and nothing to tombstone \
+             must report no change"
+        );
+        assert!(
+            index.upsert_source("real", &record("霧沢町の湧き水。")),
+            "a record with actual paragraphs is a real change"
+        );
+        assert!(
+            index.upsert_source("real", &record("")),
+            "replacing it with an empty record still tombstones what was live"
+        );
+        assert!(
+            !index.upsert_source("real", &record("")),
+            "and once nothing is left live, upserting empty again is inert"
+        );
+    }
+
     #[test]
     fn tombstoned_postings_do_not_inflate_document_frequency() {
         // Two paragraphs share a term; kill one. If df still counted

src/env.rs

@@ -124,6 +124,26 @@ pub(crate) fn resolve_flush_secs(requested: usize) -> usize {
     }
 }
 
+/// `TAGURU_EMBED_PARALLEL=0` would zero-size both the outer
+/// per-context worker pool AND `embed_provider_slots`
+/// ([`crate::registry::concurrency::Semaphore`]) — the refresh loop
+/// spins up no workers, and any earlier `Semaphore::new` construction
+/// would have needed to `.max(1)` its own way out of a permanently
+/// starved semaphore. Floor to 1 here instead, loudly, so the
+/// constructor never has to guess an operator's zero was a typo for
+/// "off" (there is no "off"; unset already means strictly sequential).
+pub(crate) fn resolve_embed_parallel(requested: usize) -> usize {
+    if requested == 0 {
+        warn!(
+            "TAGURU_EMBED_PARALLEL=0 would starve the embedding refresh workers; using 1 \
+             (the same strictly-sequential behavior as leaving it unset)"
+        );
+        1
+    } else {
+        requested
+    }
+}
+
 /// The limiter holds its budget in a u32; a bigger env value would be
 /// silently clamped inside the constructor while the boot line logged
 /// the raw number — the logged limit and the enforced limit must be
@@ -245,6 +265,12 @@ mod tests {
         assert_eq!(resolve_flush_secs(5), 5);
     }
 
+    #[test]
+    fn embed_parallel_zero_is_floored_to_one_instead_of_starving_the_semaphore() {
+        assert_eq!(resolve_embed_parallel(0), 1);
+        assert_eq!(resolve_embed_parallel(3), 3);
+    }
+
     /// The knob's three shapes — and the deliberate reading of `1` as
     /// the boolean "all", never top-1 (see the parser's doc).
     #[test]

src/metrics.rs

@@ -186,6 +186,16 @@ pub struct Metrics {
     /// signal that a context's disk gauges and quota accounting are
     /// running on stale data.
     disk_stat_failures: AtomicU64,
+    /// `embed_provider_slots` (the process-wide cap on concurrent
+    /// embedding-provider round trips, issue #563 item 4) acquires
+    /// that had to queue behind a full semaphore — a rising rate says
+    /// the provider is the bottleneck, not disk or lock contention.
+    /// `_timeouts` is its alertable half: an acquire that queued past
+    /// its request deadline and gave up, which surfaces as a refresh
+    /// failure the operator otherwise has no way to distinguish from a
+    /// provider error.
+    embed_slot_waits: AtomicU64,
+    embed_slot_timeouts: AtomicU64,
     /// Keyring hot reloads (issue #134): applied swaps (unchanged
     /// no-ops included — the reload RAN) and refusals that kept the
     /// previous table armed. The refusal counter is the alertable
@@ -260,6 +270,7 @@ mod tests {
             retrieval_cache_entries: 0,
             retrieval_cache_bytes: 0,
             semantic_cache_entries: 0,
+            embed_slot_waiters: 0,
             per_context: Vec::new(),
         }
     }
@@ -1057,6 +1068,7 @@ mod tests {
             retrieval_cache_entries: 3,
             retrieval_cache_bytes: 4096,
             semantic_cache_entries: 5,
+            embed_slot_waiters: 2,
             // One row so the per-context families render — their
             // HELP/TYPE discipline is checked here like everyone
             // else's.

src/metrics/prometheus.rs

@@ -471,6 +471,13 @@ impl Metrics {
             "Equivalence claims resident in the semantic cache (slots; payloads live in the exact-match cache).",
             gauges.semantic_cache_entries,
         );
+        push_value(
+            &mut out,
+            "taguru_embed_slot_waiters",
+            "gauge",
+            "Threads currently queued for a permit on the process-wide embedding-provider concurrency cap (TAGURU_EMBED_PARALLEL).",
+            gauges.embed_slot_waiters,
+        );
         push_value(
             &mut out,
             "taguru_wal_bytes",
@@ -707,6 +714,20 @@ impl Metrics {
             "Per-context disk-usage stats that failed for a reason other than the file being absent — the entry's disk gauges and storage-quota accounting stay on their last known snapshot until this heals.",
             self.disk_stat_failures.load(Ordering::Relaxed),
         );
+        push_value(
+            &mut out,
+            "taguru_embed_slot_waits_total",
+            "counter",
+            "Acquires of the process-wide embedding-provider concurrency permit that found none free and had to queue.",
+            self.embed_slot_waits.load(Ordering::Relaxed),
+        );
+        push_value(
+            &mut out,
+            "taguru_embed_slot_timeouts_total",
+            "counter",
+            "Acquires of the process-wide embedding-provider concurrency permit abandoned after the request deadline passed while still queued.",
+            self.embed_slot_timeouts.load(Ordering::Relaxed),
+        );
         push_value(
             &mut out,
             "taguru_keyring_reloads_total",

src/metrics/record.rs

@@ -239,6 +239,18 @@ impl Metrics {
         self.disk_stat_failures.fetch_add(1, Ordering::Relaxed);
     }
 
+    /// Count one `embed_provider_slots` acquire that found every
+    /// permit taken and had to queue (issue #563 item 4).
+    pub fn record_embed_slot_wait(&self) {
+        self.embed_slot_waits.fetch_add(1, Ordering::Relaxed);
+    }
+
+    /// Count one `embed_provider_slots` acquire that gave up after its
+    /// deadline passed while still queued.
+    pub fn record_embed_slot_timeout(&self) {
+        self.embed_slot_timeouts.fetch_add(1, Ordering::Relaxed);
+    }
+
     /// Count one keyring reload attempt (issue #134) by whether a
     /// table (possibly identical) was armed or the previous one kept.
     pub fn record_keyring_reload(&self, applied: bool) {

src/metrics/taxonomy.rs

@@ -444,6 +444,11 @@ pub struct GaugeSnapshot {
     /// Equivalence claims resident in the semantic cache (slots, not
     /// bytes — payloads live in the exact tier).
     pub semantic_cache_entries: u64,
+    /// Threads currently queued on `embed_provider_slots` waiting for
+    /// a permit (issue #563 item 4) — read live at scrape time, unlike
+    /// the wait/timeout counters in [`crate::metrics::Metrics`], which
+    /// accumulate across scrapes.
+    pub embed_slot_waiters: u64,
     /// Per-context rows, empty unless `TAGURU_METRICS_PER_CONTEXT`
     /// asked for them — the one other sanctioned exception (after the
     /// replication lag maps) to this file's no-context-labels rule,