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

src/bm25.rs

@@ -180,7 +180,12 @@ impl Bm25Index {
     /// on a deployment with no embedding provider at all.
     pub(crate) fn upsert_source(&mut self, source: &str, record: &PassageRecord) {
         let source_id = self.intern(source);
-        self.tombstone(source_id);
+        // The store's own record replaces the source unconditionally —
+        // whether or not anything was actually live to tombstone, at
+        // least one paragraph is about to be indexed below, so this
+        // call always changes the index. `tombstone`'s return is only
+        // meaningful to `remove_source`, which has no such guarantee.
+        let _ = 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 +
@@ -224,15 +229,24 @@ impl Bm25Index {
         self.reclaim_if_due();
     }
 
-    /// 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 +255,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 +1029,31 @@ 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"
+        );
+    }
+
     #[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,

src/registry.rs

@@ -50,7 +50,7 @@ use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU64, Ordering};
 use parking_lot::{Mutex, RwLock};
 use serde::{Deserialize, Serialize};
 use taguru::context::{AliasError, CompactionError, Context, LabelUsage, dead_ratio_of};
-use taguru::deadline::Deadline;
+use taguru::deadline::{Deadline, DeadlineExceeded as SlotDeadlineExceeded};
 
 use crate::api::evidence::rerank::{EvidenceReranker, RerankOutcome};
 use crate::embedding::{EmbedPurpose, EmbeddingProvider, PassageVectorStore, VectorStore};
@@ -1788,8 +1788,13 @@ impl BootConfig {
             // Worker threads dispatching each 128-item embedding chunk to
             // the provider concurrently; 1 keeps the old strictly-
             // sequential behavior. Raise to match the provider's rate
-            // limit, not the machine's core count.
-            embed_parallel: crate::env::env_number("TAGURU_EMBED_PARALLEL", 1),
+            // limit, not the machine's core count. `=0` is rejected
+            // loudly (issue #563 item 5) rather than silently
+            // rewritten — see `resolve_embed_parallel`'s doc.
+            embed_parallel: crate::env::resolve_embed_parallel(crate::env::env_number(
+                "TAGURU_EMBED_PARALLEL",
+                1,
+            )),
             // The right semantic floor is a property of the embedding
             // model (cosine bands differ per model), so its
             // recalibration lives beside TAGURU_EMBED_MODEL rather
@@ -2143,6 +2148,12 @@ struct StateInner {
 struct CueCache {
     vectors: HashMap<String, (Arc<Vec<f32>>, u64)>,
     tick: u64,
+    /// The dimension every resident vector was inserted at, `None`
+    /// while empty. Never an EMPTY vector's width — `cue_vector`
+    /// (issue #563 item 1) refuses to cache a provider's `Ok(vec![])`
+    /// answer at all, so `insert` never sees a zero-length vector to
+    /// begin with.
+    width: Option<usize>,
 }
 
 impl CueCache {
@@ -2156,10 +2167,34 @@ impl CueCache {
         Some(Arc::clone(&entry.0))
     }
 
+    /// A read AND a recency event either way: an existing key gets its
+    /// tick bumped in place (issue #563 item 3 — without this, a cue
+    /// repeatedly re-resolved after its provider call still reads as
+    /// "not recently touched" to the eviction below, and can be
+    /// evicted while genuinely hot); a new key is admitted, evicting
+    /// the least-recently-touched entry first if the cache is full.
     fn insert(&mut self, cue: String, vector: Arc<Vec<f32>>) {
-        if self.vectors.contains_key(&cue) {
+        self.tick += 1;
+        let tick = self.tick;
+        if let Some(entry) = self.vectors.get_mut(&cue) {
+            entry.1 = tick;
             return;
         }
+        // A stable width is what makes every resident vector
+        // comparable to the same gloss/paragraph table it scores
+        // against. A backend swap behind an unchanged model name (the
+        // same hazard `embeddings.rs`'s gloss refresh already guards
+        // against on the index side) would otherwise leave whichever
+        // cues embedded before the swap silently stuck at
+        // `similarity`'s width-mismatch 0.0 forever — invisible until
+        // every one happens to be re-queried. Clearing on drift makes
+        // the cache self-heal instead: every resident cue is forced to
+        // re-embed behind its next query, all agreeing with the new
+        // width from then on.
+        if self.width.is_some_and(|width| width != vector.len()) {
+            self.vectors.clear();
+        }
+        self.width = Some(vector.len());
         if self.vectors.len() >= Self::CAP
             && let Some(oldest) = self
                 .vectors
@@ -2169,8 +2204,7 @@ impl CueCache {
         {
             self.vectors.remove(&oldest);
         }
-        self.tick += 1;
-        self.vectors.insert(cue, (vector, self.tick));
+        self.vectors.insert(cue, (vector, tick));
     }
 }
 
@@ -2331,7 +2365,19 @@ impl AppState {
         // which context or which dispatch layer it came from, so it is
         // where the outer and inner worker pools' ceilings actually
         // become one global ceiling instead of two that multiply.
-        let _permit = self.0.embed_provider_slots.acquire();
+        let acquisition = self.0.embed_provider_slots.acquire_until(deadline);
+        if acquisition.queued {
+            self.0.metrics.record_embed_slot_wait();
+        }
+        let Some(_permit) = acquisition.permit else {
+            // The permit never came, so the provider was never called
+            // — this is a slot-queue timeout, not a provider failure,
+            // but a refresh caller has no other outcome to report it
+            // under than the deadline expiring (issue #563 item 4).
+            self.0.metrics.record_embed_slot_timeout();
+            self.0.metrics.record_embed_refresh(false);
+            return Err(SlotDeadlineExceeded.to_string());
+        };
         match self.timed_embed(embedder, texts, EmbedPurpose::Index, deadline) {
             Ok(vectors) => {
                 self.0.metrics.record_embed_refresh(true);
@@ -2364,8 +2410,26 @@ impl AppState {
         }
         match self.timed_embed(embedder, &[cue], EmbedPurpose::Query, deadline) {
             Ok(mut vectors) => {
+                let vector = vectors.pop().unwrap_or_default();
+                // `Ok` but zero-length (a malformed provider response,
+                // or an `unwrap_or_default` masking a shorter-than-
+                // requested batch) is not a resolved embedding — caching
+                // it here would be a PROCESS-LIFETIME poison: every
+                // future search for this exact cue would silently score
+                // 0.0 forever (`similarity`'s width-mismatch sentinel),
+                // with no way to invalidate it short of a restart
+                // (issue #563 item 1). Treat it as the resolve failure
+                // it actually is instead.
+                if vector.is_empty() {
+                    self.0.metrics.record_embed_resolve(false);
+                    return Err(format!(
+                        "embedding provider returned an empty vector for the query cue \
+                         (model {:?})",
+                        embedder.model()
+                    ));
+                }
                 self.0.metrics.record_embed_resolve(true);
-                let vector = Arc::new(vectors.pop().unwrap_or_default());
+                let vector = Arc::new(vector);
                 self.0
                     .cue_cache
                     .lock()
@@ -2410,13 +2474,26 @@ impl AppState {
         let Some(index) = guard.as_mut() else {
             return;
         };
+        // `!sources.is_empty()` used to be the dirty gate — wrong
+        // whenever every `store.get` in this batch came back `None`
+        // for a source `remove_source` had nothing live to tombstone
+        // for (issue #563 item 2): the batch touched nothing, but the
+        // sidecar still got rewritten on the next tick. `changed`
+        // tracks whether anything in the index actually moved instead.
+        let mut changed = false;
         for source in sources {
             match store.get(source) {
-                Some(record) => index.upsert_source(source, &record),
-                None => index.remove_source(source),
+                Some(record) => {
+                    // A stored record always changes the index — it
+                    // just replaced whatever paragraphs (zero or more)
+                    // this source held before.
+                    index.upsert_source(source, &record);
+                    changed = true;
+                }
+                None => changed |= index.remove_source(source),
             }
         }
-        if !sources.is_empty() {
+        if changed {
             entry.bm25_dirty.store(true, Ordering::Relaxed);
         }
     }