diff --git a/src/main.rs b/src/main.rs index e00aface..123ed4dc 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1175,7 +1175,7 @@ fn run_flush_tick( let workers = state.embed_parallel(); tokio::task::block_in_place(|| { parallel_map(flushed, workers, |name| { - match state.refresh_embeddings(&name, Deadline::unbounded()) { + match state.auto_refresh_embeddings(&name, Deadline::unbounded()) { None | Some(Ok((0, _))) => {} Some(Ok((embedded, _))) => { info!(context = %name, embedded, "auto-embedded glosses"); @@ -1199,7 +1199,7 @@ fn run_flush_tick( let workers = state.embed_parallel(); tokio::task::block_in_place(|| { parallel_map(stale, workers, |name| { - match state.refresh_passage_embeddings(&name, Deadline::unbounded()) { + match state.auto_refresh_passage_embeddings(&name, Deadline::unbounded()) { None => {} Some(Ok(outcome)) if outcome.embedded == 0 => {} Some(Ok(outcome)) => { diff --git a/src/registry.rs b/src/registry.rs index 92ba04fc..0ca558fe 100644 --- a/src/registry.rs +++ b/src/registry.rs @@ -2300,6 +2300,14 @@ struct StateInner { /// fallback path. Valid for the whole process because the provider /// (and so the model) is fixed at boot. cue_cache: Mutex, + /// The output width the provider most recently produced, and when + /// — shared across every context because the width is a property + /// of the PROVIDER, not of any one context's data. Lets a refresh + /// skip its width probe (`embeddings.rs`) when a recent-enough + /// observation already agrees with what is carried, instead of + /// paying a provider round trip on every no-op pass. See + /// [`AppState::provider_width_recently_confirmed`]. + observed_embed_width: Mutex>, /// The exact-match retrieval cache (issue #150): an identical /// request against an identical corpus state answers from the /// stored response. Unlike `cue_cache`, validity is @@ -2738,6 +2746,14 @@ impl AppState { match self.timed_embed(embedder, texts, EmbedPurpose::Index, deadline) { Ok(vectors) => { self.0.metrics.record_embed_refresh(true); + // Every refresh embed call — probe included — lands + // here, so this is where the process-wide width + // observation `provider_width_recently_confirmed` + // reads gets refreshed, regardless of which context or + // table asked. + if let Some(width) = vectors.first().map(Vec::len) { + *self.0.observed_embed_width.lock() = Some((std::time::Instant::now(), width)); + } Ok(vectors) } Err(error) => { @@ -2747,6 +2763,36 @@ impl AppState { } } + /// Whether a refresh may skip its own width probe: the provider's + /// most recently observed output width, from ANY context's embed + /// call, is both fresh (`WIDTH_OBSERVATION_TRUST`) and equal to + /// `carried` — the width this refresh's own sidecar already holds. + /// Deliberately one-sided: a stale-or-absent observation, or one + /// that disagrees with `carried`, answers `false` and lets the + /// caller fall back to spending an actual provider round trip — + /// an unconfirmed skip could hide a real width change, but an + /// unconfirmed probe only ever costs one extra call. + fn provider_width_recently_confirmed(&self, carried: usize) -> bool { + self.0 + .observed_embed_width + .lock() + .is_some_and(|(observed_at, width)| { + width == carried && width_observation_fresh(&observed_at) + }) + } + + /// Test-only: rewinds the remembered provider width observation so + /// `WIDTH_OBSERVATION_TRUST` can elapse without the test sleeping + /// through it — the same shape as `age_load_failures` below. + #[cfg(test)] + pub fn age_width_observation(&self, by: std::time::Duration) { + if let Some((observed_at, _)) = &mut *self.0.observed_embed_width.lock() { + *observed_at = observed_at + .checked_sub(by) + .expect("test ages within the Instant range"); + } + } + /// The query side of every embedding lookup: process cache first, /// provider (as [`EmbedPurpose::Query`]) on a miss. No lock is held /// across the provider call. @@ -3182,6 +3228,24 @@ fn still_quarantined(failed_at: &std::time::Instant) -> bool { failed_at.elapsed() < LOAD_FAILURE_RETRY } +/// How long a provider's most recently observed output width is +/// trusted for skipping a refresh's own width probe (issue #677 item +/// 2). Long enough that a busy, gloss-stable context stops paying a +/// provider round trip on every 5s flush tick; short enough that a +/// genuine backend swap is still caught well within a minute even for +/// a context whose own writes never trigger a real embed. +const WIDTH_OBSERVATION_TRUST: std::time::Duration = std::time::Duration::from_secs(60); + +/// Whether a remembered width observation is still within its trust +/// window. Same shape as `still_quarantined` above — the `<` boundary +/// is not a gap a behavioral test can close (see that function's doc) +/// — kept as its own predicate so both share the one place the +/// boundary is decided. +#[mutants::skip] +fn width_observation_fresh(observed_at: &std::time::Instant) -> bool { + observed_at.elapsed() < WIDTH_OBSERVATION_TRUST +} + /// Loads the image behind a cold slot and replays whatever the WAL /// holds above the image's watermark; hot slots pass through. On /// success the slot is hot, the stats are fresh, and `wal_seq` diff --git a/src/registry/boot.rs b/src/registry/boot.rs index 271f1876..3ffea49f 100644 --- a/src/registry/boot.rs +++ b/src/registry/boot.rs @@ -204,6 +204,7 @@ impl AppState { .unwrap_or(DEFAULT_SEMANTIC_FLOOR) .clamp(0.0, 1.0), cue_cache: Mutex::new(CueCache::default()), + observed_embed_width: Mutex::new(None), retrieval_cache: Mutex::new(retrieval_cache::RetrievalCache::new( retrieval_cache_bytes, )), diff --git a/src/registry/embeddings.rs b/src/registry/embeddings.rs index 244350ca..9b8c7c75 100644 --- a/src/registry/embeddings.rs +++ b/src/registry/embeddings.rs @@ -209,10 +209,43 @@ impl AppState { /// operator calls this after ingesting, so embedding spend stays /// intentional. Returns (newly embedded, total vectors), or `None` /// for an unknown context. + /// + /// Always pays for its own width probe when one is needed — see + /// [`AppState::auto_refresh_embeddings`] for the throttled variant + /// the auto-embed ticker uses instead. An explicit call (this one) + /// is rare and deliberate, often an operator diagnosing a stale + /// result, so it must reliably detect and heal a width change in + /// ONE call — the contract `tests/http_api/width_probe.rs` pins. pub fn refresh_embeddings( &self, name: &str, deadline: Deadline, + ) -> Option> { + self.refresh_embeddings_inner(name, deadline, false) + } + + /// The auto-embed ticker's variant of [`AppState::refresh_embeddings`] + /// (issue #677 item 2): identical, except its width probe is + /// skipped when a recent embed from ANY context already confirmed + /// the provider's current width (`provider_width_recently_confirmed` + /// — the width is the provider's property, not this context's). A + /// busy, gloss-stable context would otherwise pay one provider + /// round trip per flush tick forever; an explicit caller still gets + /// the unthrottled [`AppState::refresh_embeddings`] instead, so a + /// deliberate refresh always heals a width change in one call. + pub(crate) fn auto_refresh_embeddings( + &self, + name: &str, + deadline: Deadline, + ) -> Option> { + self.refresh_embeddings_inner(name, deadline, true) + } + + fn refresh_embeddings_inner( + &self, + name: &str, + deadline: Deadline, + throttle_probe: bool, ) -> Option> { let Some(embedder) = self.0.embedder.clone() else { return Some(Err( @@ -338,12 +371,28 @@ impl AppState { let mut fresh_width = width(&embedded_concepts).or_else(|| width(&embedded_labels)); // Unchanged hashes embed nothing, which would leave the width // change of exactly this scenario — backend swap, no gloss - // edits — undetectable forever. One probe embedding per no-op - // refresh keeps that from hiding. + // edits — undetectable forever. A probe embedding keeps that + // from hiding, UNLESS this is the throttled ticker path AND a + // recent embed from any context (this one or another) already + // confirmed the provider is still producing `carried`'s width + // — the width is the provider's property, not this context's, + // so that confirmation is just as good as one bought here. An + // explicit caller (`throttle_probe: false`) always probes, so + // it reliably heals a width change in this one call. Both + // tables must be confirmed (whichever carry a width), matching + // the mismatch check just below: a probe confined to labels + // alone would miss a change confined to concepts, so a skip + // confined to labels alone must not happen either. + let width_confirmed = |carried: Option| { + carried.is_none_or(|w| self.provider_width_recently_confirmed(w)) + }; if failure.is_none() && !fresh_model && (carried_concepts_width.is_some() || carried_labels_width.is_some()) && fresh_width.is_none() + && !(throttle_probe + && width_confirmed(carried_concepts_width) + && width_confirmed(carried_labels_width)) && let Some((_, gloss)) = concepts.first().or_else(|| labels.first()) { match self.timed_embed_for_refresh(embedder.as_ref(), &[gloss.as_str()], deadline) { @@ -631,10 +680,36 @@ impl AppState { /// across the whole backfill. A provider failure partway persists /// what did land and reports the error — the next refresh continues /// from there instead of re-buying the same vectors. + /// + /// Always pays for its own width probe when one is needed — see + /// [`AppState::auto_refresh_passage_embeddings`] for the throttled + /// variant the auto-embed ticker uses instead, and + /// [`AppState::refresh_embeddings`]'s doc for why the split exists. pub fn refresh_passage_embeddings( &self, name: &str, deadline: Deadline, + ) -> Option> { + self.refresh_passage_embeddings_inner(name, deadline, false) + } + + /// The auto-embed ticker's variant of + /// [`AppState::refresh_passage_embeddings`] — see + /// [`AppState::auto_refresh_embeddings`]'s doc for the throttle + /// this mirrors (issue #677 item 2). + pub(crate) fn auto_refresh_passage_embeddings( + &self, + name: &str, + deadline: Deadline, + ) -> Option> { + self.refresh_passage_embeddings_inner(name, deadline, true) + } + + fn refresh_passage_embeddings_inner( + &self, + name: &str, + deadline: Deadline, + throttle_probe: bool, ) -> Option> { let Some(embedder) = self.0.embedder.clone() else { return Some(Err( @@ -827,11 +902,17 @@ impl AppState { } // Unchanged hashes embed nothing, which would leave the width // change of exactly this scenario — backend swap, no passage - // edits — undetectable. One probe embedding per no-op refresh - // keeps it from hiding, matching the concept refresh. + // edits — undetectable. A probe embedding keeps it from + // hiding, matching the concept refresh — including the same + // ticker-only skip when a recent embed from any context + // already confirmed the provider is still producing + // `carried_width` (see `provider_width_recently_confirmed`'s + // doc: the width is the provider's property, not this + // context's). An explicit caller always probes. if failure.is_none() && !fresh_model - && carried_width.is_some() + && carried_width + .is_some_and(|w| !(throttle_probe && self.provider_width_recently_confirmed(w))) && fresh_width.is_none() && let Some(probe) = records .iter() diff --git a/src/registry/embeddings/gloss_tests.rs b/src/registry/embeddings/gloss_tests.rs index 9f0376c5..ed96d77f 100644 --- a/src/registry/embeddings/gloss_tests.rs +++ b/src/registry/embeddings/gloss_tests.rs @@ -603,6 +603,271 @@ mod tests { let _ = fs::remove_dir_all(dir); } + /// Issue #677 item 2: a busy, gloss-stable context used to pay one + /// provider round trip (the width probe) on every single no-op + /// refresh, forever — expensive under a write-driven auto-embed + /// ticker. The real embed a pass just paid for already answers + /// "what width does the provider speak right now" just as well as + /// a dedicated probe would, so the very next no-op TICKER refresh + /// must not pay for one of its own. This throttle is scoped to + /// `auto_refresh_embeddings` only — the public `refresh_embeddings` + /// an explicit caller uses always probes, so it reliably heals a + /// width change in one call (see that function's own doc, and + /// `tests/http_api/width_probe.rs`, for why that contract must + /// hold). + #[test] + fn a_no_op_auto_refresh_right_after_a_real_one_makes_no_provider_call_at_all() { + struct CountingEmbeddings(usize, Arc); + impl EmbeddingProvider for CountingEmbeddings { + fn model(&self) -> &str { + "stable-name" + } + fn embed( + &self, + texts: &[&str], + _purpose: EmbedPurpose, + _deadline: Deadline, + ) -> Result>, String> { + self.1.fetch_add(1, Ordering::Relaxed); + Ok(texts + .iter() + .map(|_| { + let mut vector = vec![0.0; self.0]; + vector[0] = 1.0; + vector + }) + .collect()) + } + } + + let dir = scratch_dir("no-double-charge"); + let calls = Arc::new(AtomicUsize::new(0)); + let embedder = + Some(Arc::new(CountingEmbeddings(2, Arc::clone(&calls))) as Arc); + let state = AppState::boot(dir.clone(), usize::MAX, embedder).unwrap(); + state + .create("w", ContextMeta::default()) + .map_err(|_| "create") + .unwrap(); + state + .write_context("w", |context| { + context.associate("a", "l", "b", 1.0).unwrap(); + }) + .map_err(|_| "write") + .unwrap(); + + let (embedded, _) = state + .auto_refresh_embeddings("w", Deadline::unbounded()) + .unwrap() + .unwrap(); + assert!( + embedded > 0, + "the first pass must genuinely embed something" + ); + let after_real = calls.load(Ordering::Relaxed); + + let (embedded_again, _) = state + .auto_refresh_embeddings("w", Deadline::unbounded()) + .unwrap() + .unwrap(); + assert_eq!(embedded_again, 0); + assert_eq!( + calls.load(Ordering::Relaxed), + after_real, + "the no-op ticker refresh's own width probe must be skipped \ + entirely, not just cheap: the pass just above already \ + confirmed this width" + ); + + let _ = fs::remove_dir_all(dir); + } + + /// The throttle only ever defers detection, never cancels it: once + /// a confirmed observation ages past `WIDTH_OBSERVATION_TRUST`, the + /// ticker's next no-op refresh probes again and still catches a + /// genuine backend swap — the same swap an explicit refresh would + /// have caught immediately, just later. + #[test] + fn a_ticker_probe_re_arms_once_its_observation_ages_past_the_trust_window() { + struct SwappableWidthEmbeddings(Arc, Arc); + impl EmbeddingProvider for SwappableWidthEmbeddings { + fn model(&self) -> &str { + "stable-name" + } + fn embed( + &self, + texts: &[&str], + _purpose: EmbedPurpose, + _deadline: Deadline, + ) -> Result>, String> { + self.1.fetch_add(1, Ordering::Relaxed); + let width = self.0.load(Ordering::Relaxed); + Ok(texts + .iter() + .map(|_| { + let mut vector = vec![0.0; width]; + vector[0] = 1.0; + vector + }) + .collect()) + } + } + + let dir = scratch_dir("ticker-probe-re-arms"); + let width = Arc::new(AtomicUsize::new(2)); + let calls = Arc::new(AtomicUsize::new(0)); + let embedder = Some(Arc::new(SwappableWidthEmbeddings( + Arc::clone(&width), + Arc::clone(&calls), + )) as Arc); + let state = AppState::boot(dir.clone(), usize::MAX, embedder).unwrap(); + state + .create("w", ContextMeta::default()) + .map_err(|_| "create") + .unwrap(); + state + .write_context("w", |context| { + context.associate("a", "l", "b", 1.0).unwrap(); + }) + .map_err(|_| "write") + .unwrap(); + let (embedded, _) = state + .auto_refresh_embeddings("w", Deadline::unbounded()) + .unwrap() + .unwrap(); + assert!(embedded > 0); + + // The provider now speaks width 3 behind the same model name. + // Within the trust window, the ticker's probe stays skipped — + // this is the cost saving item 2 exists for. + width.store(3, Ordering::Relaxed); + let calls_before_stale_no_op = calls.load(Ordering::Relaxed); + let (embedded, total) = state + .auto_refresh_embeddings("w", Deadline::unbounded()) + .unwrap() + .unwrap(); + assert_eq!((embedded, total), (0, 3), "still within the trust window"); + assert_eq!( + calls.load(Ordering::Relaxed), + calls_before_stale_no_op, + "no probe yet — the observation is still trusted" + ); + + // Once the observation ages out, the very next ticker refresh + // probes again and this time catches the swap. + state.age_width_observation(crate::registry::WIDTH_OBSERVATION_TRUST); + let (embedded, total) = state + .auto_refresh_embeddings("w", Deadline::unbounded()) + .unwrap() + .unwrap(); + assert_eq!( + (embedded, total), + (3, 3), + "the aged-out observation must not suppress detection forever" + ); + let store = VectorStore::load(&vectors_path(&dir, &file_stem("w"))); + assert!( + store + .concepts + .values() + .chain(store.labels.values()) + .all(|(_, vector)| vector.len() == 3), + "the swap must actually heal, not just get noticed" + ); + + let _ = fs::remove_dir_all(dir); + } + + /// The width observation is shared by the whole process, not scoped + /// to one context — because the width is the PROVIDER's property. + /// A context that has never itself embedded anything this boot + /// still skips its own ticker probe once some OTHER context's + /// ticker probe has already confirmed the width. + #[test] + fn a_probe_confirmed_by_one_context_lets_a_sibling_skip_its_own() { + struct CountingWidthEmbeddings(usize, Arc); + impl EmbeddingProvider for CountingWidthEmbeddings { + fn model(&self) -> &str { + "stable-name" + } + fn embed( + &self, + texts: &[&str], + _purpose: EmbedPurpose, + _deadline: Deadline, + ) -> Result>, String> { + self.1.fetch_add(1, Ordering::Relaxed); + Ok(texts + .iter() + .map(|_| { + let mut vector = vec![0.0; self.0]; + vector[0] = 1.0; + vector + }) + .collect()) + } + } + + let dir = scratch_dir("cross-context-width-confirm"); + { + let embedder = Some( + Arc::new(CountingWidthEmbeddings(2, Arc::new(AtomicUsize::new(0)))) + as Arc, + ); + let state = AppState::boot(dir.clone(), usize::MAX, embedder).unwrap(); + for name in ["a", "b"] { + state + .create(name, ContextMeta::default()) + .map_err(|_| "create") + .unwrap(); + state + .write_context(name, |context| { + context.associate("x", "l", "y", 1.0).unwrap(); + }) + .map_err(|_| "write") + .unwrap(); + state + .refresh_embeddings(name, Deadline::unbounded()) + .unwrap() + .unwrap(); + } + state.flush_dirty(); + } + + // Fresh boot: no observation yet. Both contexts already carry + // width 2 on disk and neither has new content this pass. + let calls = Arc::new(AtomicUsize::new(0)); + let embedder = + Some(Arc::new(CountingWidthEmbeddings(2, Arc::clone(&calls))) + as Arc); + let state = AppState::boot(dir.clone(), usize::MAX, embedder).unwrap(); + + // "a"'s own no-op ticker refresh has no observation to lean on + // yet, so it pays for its own probe. + let (embedded_a, _) = state + .auto_refresh_embeddings("a", Deadline::unbounded()) + .unwrap() + .unwrap(); + assert_eq!(embedded_a, 0); + assert_eq!(calls.load(Ordering::Relaxed), 1, "a's own probe"); + + // "b" was never touched this boot — but the width is the + // provider's, not "a"'s, so "a"'s probe already answers for + // "b" too: no second provider call. + let (embedded_b, _) = state + .auto_refresh_embeddings("b", Deadline::unbounded()) + .unwrap() + .unwrap(); + assert_eq!(embedded_b, 0); + assert_eq!( + calls.load(Ordering::Relaxed), + 1, + "b's probe is skipped: a's already confirmed the width this boot" + ); + + let _ = fs::remove_dir_all(dir); + } + /// A width change that rides alongside genuinely new content (rather /// than being caught only by the no-op probe) is noticed from the /// freshly embedded rows directly — but the redo it triggers must diff --git a/src/registry/embeddings/passage_tests.rs b/src/registry/embeddings/passage_tests.rs index 9d415e16..aaba76f7 100644 --- a/src/registry/embeddings/passage_tests.rs +++ b/src/registry/embeddings/passage_tests.rs @@ -72,6 +72,50 @@ mod tests { let _ = fs::remove_dir_all(dir); } + /// Issue #677 item 2, passage side: the auto-embed ticker's variant + /// skips its width probe when a recent embed (this pass or any + /// other context's) already confirmed the width — unlike an + /// explicit `refresh_passage_embeddings` call, which always probes + /// (see the test just above, and that function's own doc). + #[test] + fn a_no_op_auto_passage_refresh_right_after_a_real_one_makes_no_provider_call_at_all() { + let dir = scratch_dir("pvec-no-double-charge"); + let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let state = + boot_for_passage_embedding(&dir, Arc::new(MockEmbeddings::fruity(&calls)), 20_000); + state + .create("sake", ContextMeta::default()) + .map_err(|_| "create") + .unwrap(); + let mut passages = BTreeMap::new(); + passages.insert("doc-a".to_string(), "最初の段落。".to_string()); + state + .store_passages("sake", plain(passages)) + .unwrap() + .unwrap(); + + let outcome = state + .auto_refresh_passage_embeddings("sake", Deadline::unbounded()) + .unwrap() + .unwrap(); + assert_eq!(outcome.embedded, 1); + let after_real = calls.load(Ordering::Relaxed); + + let again = state + .auto_refresh_passage_embeddings("sake", Deadline::unbounded()) + .unwrap() + .unwrap(); + assert_eq!(again.embedded, 0); + assert_eq!( + calls.load(Ordering::Relaxed), + after_real, + "the no-op ticker refresh's own width probe must be skipped \ + entirely: the pass just above already confirmed this width" + ); + + let _ = fs::remove_dir_all(dir); + } + #[test] fn refresh_passage_embeddings_re_embeds_only_the_changed_paragraph() { let dir = scratch_dir("pvec-diff");