From 4a4d8b029ee62ff97c0cc0246e6f86aade3d54c3 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 29 Sep 2026 19:42:03 +0200 Subject: [PATCH] Bound the whole search path by the search's own time; lag is not a rebuild Soul pass 2 follow-ups F1, F2, F4, F5 on the semantic index. F1: every call a search makes (identity reads, label read, embed, query) is bounded by what remains of one Budget, and the queued ticket carries the moment SEARCH_DEADLINE after it was asked. The ports take the bound per call (model_identity, describe, search); QUERY_EMBED_TIMEOUT is gone, since the ticket's deadline is the one number. A hung identity read frees the worker within the search's time. F2: one refusal predicate. Behind is lag and no longer refuses; only Reconciling, Failing and Refused do, for the sink and the projector alike. The worker absorbs the inbox before taking a search and leaves the search queued while the writes ahead of it are unfinished, so a client that admits and then searches reads its own writes. An index being rebuilt is still refused. F4: the model-changed refusal and the worker's failure answer to queued searches name the state and the pending count. F5: a search that fails for its own reasons no longer forgets the reconciliation; only a changed model or label does. Co-Authored-By: Claude Opus 5.5 --- crates/huginn-daemon/src/index/fakes.rs | 31 +- crates/huginn-daemon/src/index/mod.rs | 226 +++++++----- crates/huginn-daemon/src/index/ollama.rs | 16 +- crates/huginn-daemon/src/index/qdrant.rs | 21 +- crates/huginn-daemon/src/index/tests.rs | 324 +++++++++++++----- crates/huginn-daemon/src/semantic_tests.rs | 26 +- .../huginn-daemon/tests/live_index_smoke.rs | 8 +- 7 files changed, 436 insertions(+), 216 deletions(-) diff --git a/crates/huginn-daemon/src/index/fakes.rs b/crates/huginn-daemon/src/index/fakes.rs index 47502b8..61c98a2 100644 --- a/crates/huginn-daemon/src/index/fakes.rs +++ b/crates/huginn-daemon/src/index/fakes.rs @@ -66,6 +66,11 @@ pub(crate) struct EmbedderState { /// The length of the vectors it answers with, where that is not the /// identity's. pub(crate) vector_length: Option, + /// The bound each identity read was given, in order. + pub(crate) identity_bounds: Vec, + /// Whether an identity read is never answered: it fails when its bound + /// has passed, as an adapter that honours its bound does. + pub(crate) hang_identity: bool, } #[derive(Clone)] @@ -76,7 +81,7 @@ pub(crate) struct FakeEmbedder { impl FakeEmbedder { pub(crate) fn new(digest: &str) -> Self { - let state = EmbedderState { identity: identity(digest), down: false, embedded: Vec::new(), bounds: Vec::new(), vector_length: None }; + let state = EmbedderState { identity: identity(digest), down: false, embedded: Vec::new(), bounds: Vec::new(), vector_length: None, identity_bounds: Vec::new(), hang_identity: false }; Self { state: Arc::new(Mutex::new(state)), gate: None } } @@ -91,8 +96,14 @@ impl FakeEmbedder { } impl Embedder for FakeEmbedder { - fn model_identity(&mut self) -> Result { - let state = self.state.lock().unwrap(); + fn model_identity(&mut self, within: Duration) -> Result { + let mut state = self.state.lock().unwrap(); + state.identity_bounds.push(within); + if state.hang_identity { + drop(state); + std::thread::sleep(within); + bail!("the embedder did not answer within {within:?}"); + } if state.down { bail!("the embedder is down"); } @@ -135,6 +146,10 @@ pub(crate) struct IndexState { pub(crate) listed: u32, /// Whether a search fails at the vector store, the rest of it working. pub(crate) fail_search: bool, + /// The bound each describe and each search was given, in order, by call. + pub(crate) bounds: Vec<(&'static str, Duration)>, + /// How many points the collection held when each search reached it. + pub(crate) held_when_searched: Vec, } #[derive(Clone, Default)] @@ -155,8 +170,9 @@ impl FakeIndex { } impl VectorIndex for FakeIndex { - fn describe(&mut self, collection: &str) -> Result { - let state = self.state.lock().unwrap(); + fn describe(&mut self, collection: &str, within: Duration) -> Result { + let mut state = self.state.lock().unwrap(); + state.bounds.push(("describe", within)); if state.down { bail!("the vector store is down"); } @@ -197,14 +213,17 @@ impl VectorIndex for FakeIndex { Ok(()) } - fn search(&mut self, collection: &str, vector: &[f32], limit: u32) -> Result> { + fn search(&mut self, collection: &str, vector: &[f32], limit: u32, within: Duration) -> Result> { let mut state = self.state.lock().unwrap(); + state.bounds.push(("search", within)); if state.down { bail!("the vector store is down"); } state.searches.push((collection.into(), vector.len(), limit)); ensure!(!state.fail_search, "the vector store failed the search"); ensure!(state.collections.contains_key(collection), "no collection {collection}"); + let held = state.collections[collection].points.len(); + state.held_when_searched.push(held); Ok(state.hits.clone()) } } diff --git a/crates/huginn-daemon/src/index/mod.rs b/crates/huginn-daemon/src/index/mod.rs index 3f9102c..367e675 100644 --- a/crates/huginn-daemon/src/index/mod.rs +++ b/crates/huginn-daemon/src/index/mod.rs @@ -65,15 +65,24 @@ //! (`ServeOptions::search_timeout` bounds the wait), and takes back a search //! it has stopped waiting for. At most `SEARCHES_QUEUED_MAX` wait; one more is //! refused. A search whose embed is already running cannot be recalled, so it -//! can hold the writes off for as long as its bound, `QUERY_EMBED_TIMEOUT`, -//! which is `SEARCH_DEADLINE`, the deadline the serve loop gives the search. -//! - Only a current index answers a search. While the worker is `Reconciling` -//! or `Behind` (a re-pulled model, a lost collection, entries not yet -//! written), `Failing` or `Refused`, a search is refused at once with an -//! `Unavailable` that names the state and how many documents are pending; it -//! is never answered from a half-filled collection. A search that fails after -//! it was accepted forgets the reconciliation, so `whoami` stops reporting -//! `Current` and the worker verifies the model and the collection again. +//! can hold the writes off for as long as it has left: the ticket carries the +//! moment `SEARCH_DEADLINE` after it was asked, and every call the search +//! makes (the model's identity, the collection's label, the embed, the +//! query) is bounded by what remains of it. +//! - A search reads its own writes. Admissions and searches meet under one +//! lock, admissions first: a search waits in the queue while the worker +//! absorbs and writes what was admitted before it, and is answered once the +//! index has caught up. Lag is not a rebuild and refuses nothing. Under a +//! steady stream of admissions a search may never see the index caught up; +//! its deadline then answers it. +//! - A search is refused only while the index is being rebuilt or cannot be +//! asked: `Reconciling` (a re-pulled model, a lost collection, a mind's first +//! admission), `Failing` or `Refused`. The refusal is an `Unavailable` that +//! names the state and how many documents are pending; it is never answered +//! from a collection being rebuilt. A search that fails for its own reasons +//! (a timeout, a store error) fails that search alone and the +//! reconciliation stands. Only what the check-then-embed finds changed (the +//! model, the collection's label) forgets it. pub mod ollama; pub mod qdrant; @@ -101,21 +110,21 @@ pub const MANAGED_BY: &str = "huginn"; pub const BATCH: usize = 32; /// How long a search may take, queue wait included: the serve loop's default -/// `ServeOptions::search_timeout`. It is the one number the query embed's -/// bound derives from, so the two cannot drift apart. +/// `ServeOptions::search_timeout`. Every bound on the search path derives from +/// the moment this long after the search was asked, so the two cannot drift +/// apart. pub const SEARCH_DEADLINE: Duration = Duration::from_secs(30); -/// The bound on one query embed. A running embed cannot be recalled, so it -/// holds the index's writes off for as long as it lasts; it is held to the -/// search's own deadline, after which nobody is waiting for its vector. -pub const QUERY_EMBED_TIMEOUT: Duration = SEARCH_DEADLINE; - -/// The bound on a bulk index-flush embed, and on the embedder's identity -/// reads. A cold model takes seconds to load. It does not block serving: the -/// serve loop never waits on the worker, and a search queued behind a flush is -/// answered by its own deadline. +/// The bound on a bulk index-flush embed, the identity and label reads around +/// it, and the reconciliation's identity read. A cold model takes seconds to +/// load. It does not block serving: the serve loop never waits on the worker, +/// and a search queued behind a flush is answered by its own deadline. pub const BULK_EMBED_TIMEOUT: Duration = Duration::from_secs(300); +/// The bound on a call to the vector store that is not a search: the +/// adapter's default. +pub const INDEX_CALL_TIMEOUT: Duration = Duration::from_secs(60); + /// The most candidates one search asks the index for. pub const OVERSAMPLE_MAX: u32 = 200; @@ -179,13 +188,16 @@ pub struct Hit { } pub trait Embedder { - fn model_identity(&mut self) -> Result; + /// The model's identity, or an error once `within` has passed. + fn model_identity(&mut self, within: Duration) -> Result; /// One vector per text, in order, or an error once `within` has passed. fn embed(&mut self, texts: &[String], within: Duration) -> Result>>; } pub trait VectorIndex { - fn describe(&mut self, collection: &str) -> Result; + /// What the collection says about itself, or an error once `within` has + /// passed. + fn describe(&mut self, collection: &str, within: Duration) -> Result; /// Drops the collection if it exists and creates it empty with this /// metadata. fn recreate(&mut self, collection: &str, meta: &CollectionMeta) -> Result<()>; @@ -193,8 +205,9 @@ pub trait VectorIndex { fn ids(&mut self, collection: &str) -> Result>; fn upsert(&mut self, collection: &str, points: &[Point]) -> Result<()>; /// The `limit` nearest points to `vector`, best first, with no filter: which - /// documents may be shown is the mind's to decide. - fn search(&mut self, collection: &str, vector: &[f32], limit: u32) -> Result>; + /// documents may be shown is the mind's to decide. An error once `within` + /// has passed. + fn search(&mut self, collection: &str, vector: &[f32], limit: u32, within: Duration) -> Result>; } /// The mind's collection: one per instance. @@ -245,6 +258,25 @@ pub enum Advance { Idle, } +/// The time one operation has left. Every call the operation makes is bounded +/// by what remains, not by a bound of its own, so their sum cannot outlast it. +struct Budget { + end: Instant, + total: Duration, +} + +impl Budget { + fn new(total: Duration) -> Self { + Self { end: Instant::now() + total, total } + } + + fn left(&self) -> Result { + let left = self.end.saturating_duration_since(Instant::now()); + ensure!(!left.is_zero(), "the time allowed ({:?}) has run out", self.total); + Ok(left) + } +} + /// The projection's state: the entries not yet known to be in the collection, /// and, once reconciled, the length of vector it accepts. pub struct Projector { @@ -344,8 +376,8 @@ impl Projector { /// collection was reconciled under. A different model, digest or length /// forgets the reconciliation, so the next step reconciles (and rebuilds) /// instead of writing or searching with vectors the collection cannot mix. - fn model_unchanged(&mut self) -> Result { - let now = self.embedder.model_identity().context("re-reading the embedding model's identity")?; + fn model_unchanged(&mut self, budget: &Budget) -> Result { + let now = self.embedder.model_identity(budget.left()?).context("re-reading the embedding model's identity")?; if self.reconciled.as_ref() == Some(&now) { return Ok(true); } @@ -362,9 +394,10 @@ impl Projector { let Some(mind) = self.mind.clone() else { return Err(StepError::Unavailable(anyhow::anyhow!("the mind has no identity yet"))); }; - let identity = self.embedder.model_identity().context("reading the embedding model's identity")?; + let identity = + self.embedder.model_identity(BULK_EMBED_TIMEOUT).context("reading the embedding model's identity")?; let wanted = self.collection_meta(&mind, &identity); - match self.index.describe(&self.collection).context("describing the collection")? { + match self.index.describe(&self.collection, INDEX_CALL_TIMEOUT).context("describing the collection")? { Described::Absent => self.recreate(&wanted)?, Described::Unlabelled => { return Err(StepError::Refused(format!( @@ -411,10 +444,10 @@ impl Projector { /// the first `BATCH` wanted entries are embedded, written, and only then /// no longer wanted. pub fn flush_batch(&mut self) -> Result<(), StepError> { - self.flush().map_err(StepError::Unavailable) + self.flush(&Budget::new(BULK_EMBED_TIMEOUT)).map_err(StepError::Unavailable) } - fn flush(&mut self) -> Result<()> { + fn flush(&mut self, budget: &Budget) -> Result<()> { let dimensions = self.reconciled.as_ref().context("flush before reconcile")?.dimensions as usize; let batch: Vec = self.wanted.iter().take(BATCH).filter_map(|document_id| self.known.get(document_id)).cloned().collect(); @@ -424,7 +457,7 @@ impl Projector { let texts: Vec = batch.iter().map(|entry| entry.text.clone()).collect(); // A model that changed leaves nothing written and the reconciliation // forgotten: the next step reconciles and rebuilds. - let Ok(vectors) = self.embed_verified(&texts, BULK_EMBED_TIMEOUT)? else { return Ok(()) }; + let Ok(vectors) = self.embed_verified(&texts, budget)? else { return Ok(()) }; ensure!(vectors.len() == batch.len(), "the embedder returned {} vectors for {} texts", vectors.len(), batch.len()); let mut points = Vec::with_capacity(batch.len()); for (entry, vector) in batch.iter().zip(vectors) { @@ -453,28 +486,29 @@ impl Projector { Ok(()) } - /// The one check-then-embed: reads the model's identity and the - /// collection's label, embeds, reads the identity again, and gives the - /// vectors back only if the model is still the one the collection was - /// reconciled under. A model that changed forgets the reconciliation and - /// answers `Err(when)`; a collection that is no longer this mind's, built - /// as reconciled, is an error and forgets it too. Search and flush both - /// embed through here, so neither compares or writes vectors made by a - /// model the collection cannot mix. - fn embed_verified(&mut self, texts: &[String], within: Duration) -> Result>, &'static str>> { + /// The one check-then-embed, every call bounded by what `budget` has left: + /// reads the model's identity and the collection's label, embeds, reads the + /// identity again, and gives the vectors back only if the model is still + /// the one the collection was reconciled under. A model that changed + /// forgets the reconciliation and answers `Err(when)`; a collection that + /// is no longer this mind's, built as reconciled, is an error and forgets + /// it too. A call that fails or times out forgets nothing. Search and + /// flush both embed through here, so neither compares or writes vectors + /// made by a model the collection cannot mix. + fn embed_verified(&mut self, texts: &[String], budget: &Budget) -> Result>, &'static str>> { let (Some(mind), Some(reconciled)) = (self.mind.clone(), self.reconciled.clone()) else { bail!("the mind has no identity yet, so it has no index"); }; - if !self.model_unchanged().map_err(|error| anyhow::anyhow!("{error}"))? { + if !self.model_unchanged(budget)? { return Ok(Err("since the collection was built")); } - let described = self.index.describe(&self.collection).context("describing the collection")?; + let described = self.index.describe(&self.collection, budget.left()?).context("describing the collection")?; if described != Described::Labelled(self.collection_meta(&mind, &reconciled)) { self.reconciled = None; bail!("collection {} is no longer the one this mind's index was built as: {described:?}", self.collection); } - let vectors = self.embedder.embed(texts, within).context("embedding")?; - if !self.model_unchanged().map_err(|error| anyhow::anyhow!("{error}"))? { + let vectors = self.embedder.embed(texts, budget.left()?).context("embedding")?; + if !self.model_unchanged(budget)? { return Ok(Err("while the texts were embedded")); } Ok(Ok(vectors)) @@ -486,41 +520,38 @@ impl Projector { /// what the index cannot know is not in force. A hit naming a kind this /// organ has no name for is logged and left out. /// - /// A search is answered only from an index that is current: while it is - /// being reconciled or written to, it refuses, naming why and how many - /// documents are pending, rather than answer from a half-filled - /// collection. It is also answered only from a collection that is still - /// this mind's, built by the model that is still the embedder: the model's - /// identity is read before and after the query is embedded and the - /// collection's label before it, and the query is refused if either - /// disagrees with what the collection was reconciled under. Any failure - /// after that forgets the reconciliation, so `progress_status` stops - /// reporting `Current` and the worker verifies again before it answers. - pub fn search(&mut self, text: &str, top_k: u32) -> Result { + /// A search is refused while the index is being rebuilt or cannot be asked + /// (`unsearchable`), naming the state and how many documents are pending. + /// Catching up on lag is the worker's: it serves a search only once the + /// documents admitted before it are written. The answer is also only from + /// a collection that is still this mind's, built by the model that is + /// still the embedder: the identity is read before and after the query is + /// embedded and the label before it, and the query is refused, with the + /// reconciliation forgotten, if either disagrees with what the collection + /// was reconciled under. Every call is bounded by what remains of + /// `within`. A search that fails for any other reason (a timeout, a store + /// error) fails alone and leaves the reconciliation standing. + pub fn search(&mut self, text: &str, top_k: u32, within: Duration) -> Result { if let Some(why) = unsearchable(&self.progress_status()) { bail!("{why}"); } - let found = self.nearest(text, top_k); - if found.is_err() { - self.reconciled = None; - } - found - } - - fn nearest(&mut self, text: &str, top_k: u32) -> Result { + let budget = Budget::new(within); let Some(reconciled) = self.reconciled.clone() else { bail!("the mind has no identity yet, so it has no index"); }; let dimensions = reconciled.dimensions as usize; let query = format!("Instruct: {QUERY_INSTRUCTION}\nQuery: {text}"); - let mut vectors = match self.embed_verified(&[query], QUERY_EMBED_TIMEOUT)? { + let mut vectors = match self.embed_verified(&[query], &budget)? { Ok(vectors) => vectors, - Err(when) => bail!("the embedding model changed {when}; the index is being rebuilt for it"), + Err(when) => { + bail!("the embedding model changed {when}: {}", unsearchable(&self.progress_status()).unwrap_or_default()) + } }; ensure!(vectors.len() == 1, "the embedder returned {} vectors for one query", vectors.len()); let vector = vectors.remove(0); ensure!(vector.len() == dimensions, "the embedder returned a query vector of {} where the collection holds {dimensions}", vector.len()); - let found = self.index.search(&self.collection, &vector, top_k.saturating_mul(4).min(OVERSAMPLE_MAX))?; + let limit = top_k.saturating_mul(4).min(OVERSAMPLE_MAX); + let found = self.index.search(&self.collection, &vector, limit, budget.left()?)?; let mut hits = Vec::with_capacity(found.len()); for hit in found { match PipelineKind::ALL.into_iter().find(|kind| kind.name() == hit.kind) { @@ -579,6 +610,16 @@ impl Backoff { /// told, not by the worker's patience. pub const SEARCHES_QUEUED_MAX: usize = 16; +/// A search waiting for the worker: what was asked, and the moment nobody is +/// waiting for the answer any more. Every call the search makes is bounded by +/// what remains until `due`. +struct Queued { + ticket: SearchTicket, + text: String, + top_k: u32, + due: Instant, +} + struct Shared { inbox: Vec, /// The mind's identity, once a commit has revealed it to a mind that had none. @@ -588,7 +629,7 @@ struct Shared { /// Searches started and not yet taken by the worker, at most /// `SEARCHES_QUEUED_MAX`; the serve loop takes back one it has stopped /// waiting for. - searches: VecDeque<(SearchTicket, String, u32)>, + searches: VecDeque, /// Searches the worker finished and the serve loop has not collected. found: Vec<(SearchTicket, Result)>, next_ticket: u64, @@ -655,18 +696,19 @@ fn including(status: &IndexStatus, queued: u32) -> IndexStatus { } } -/// Why a status cannot answer a search, or `None` when it can. Only a current -/// index answers: one that is being reconciled or written to holds a -/// half-filled collection, and one that is failing or refused cannot be asked. -/// The reason names the state and how many documents are pending. +/// Why a status cannot answer a search, or `None` when it can: the one +/// predicate for asking and for answering. An index being rebuilt +/// (`Reconciling`) holds a collection that is not the model's yet, and one +/// that is failing or refused cannot be asked. `Behind` is lag, not a +/// rebuild: the worker writes what was admitted before a search and then +/// answers it. The reason names the state and how many documents are pending. fn unsearchable(status: &IndexStatus) -> Option { use IndexStatus as S; match status { - S::Current => None, + S::Current | S::Behind { .. } => None, S::Reconciling { pending } => Some(format!( "the semantic index is being rebuilt: reconciling with the model and the collection (pending: {pending})" )), - S::Behind { pending } => Some(format!("the semantic index is being rebuilt: behind the mind (pending: {pending})")), S::Failing { pending, error, .. } => Some(format!("the semantic index is failing: {error} (pending: {pending})")), S::Refused { pending, reason, .. } => Some(format!("the semantic index is refused: {reason} (pending: {pending})")), } @@ -676,9 +718,18 @@ fn unsearchable(status: &IndexStatus) -> Option { /// for the loop to collect. Between batches, never inside one, and only one: /// the worker returns to its writes after each, so a queue of searches cannot /// hold the index's writes off for the sum of their embeddings. +/// +/// Admissions come first: the inbox is absorbed before the search is taken, so +/// every document committed before the search was asked is wanted, and a search +/// behind lag stays queued until the worker's writes have caught up. An index +/// being rebuilt does not wait: the search is taken and refused. The search's +/// calls are bounded by what remains of the time it was given when it was asked. fn serve_search(projector: &mut Projector, shared: &Lock) { - let Some((ticket, text, top_k)) = lock(shared).searches.pop_front() else { return }; - let found = projector.search(&text, top_k).map_err(|error| format!("{error:#}")); + if !absorb(projector, shared, None) || (projector.pending() > 0 && !projector.stale()) { + return; + } + let Some(Queued { ticket, text, top_k, due }) = lock(shared).searches.pop_front() else { return }; + let found = projector.search(&text, top_k, due.saturating_duration_since(Instant::now())).map_err(|error| format!("{error:#}")); lock(shared).found.push((ticket, found)); } @@ -686,8 +737,8 @@ fn serve_search(projector: &mut Projector, sh /// while it was healthy and it has failed since. fn refuse_searches(shared: &Lock, why: &str) { let mut guard = lock(shared); - while let Some((ticket, ..)) = guard.searches.pop_front() { - guard.found.push((ticket, Err(why.to_owned()))); + while let Some(queued) = guard.searches.pop_front() { + guard.found.push((queued.ticket, Err(why.to_owned()))); } } @@ -748,8 +799,8 @@ fn work(mut projector: Projector, shared: Loc } publish(&shared, projector.progress_status()); serve_search(&mut projector, &shared); - // A search that failed has left the reconciliation to redo: - // do it now, not at the next wake. + // A model or collection the search found changed has left the + // reconciliation to redo: do it now, not at the next wake. if advance == Advance::Idle && !projector.stale() { wait_for_work(&shared, Wait::Work { until: recheck_at }); } @@ -762,7 +813,7 @@ fn work(mut projector: Projector, shared: Loc delay ); publish(&shared, projector.failure_status(&error, attempts)); - refuse_searches(&shared, &format!("the index could not advance: {error}")); + refuse_searches(&shared, &unsearchable(&projector.failure_status(&error, attempts)).unwrap_or_default()); failing = Some((error, attempts)); wait_for_work(&shared, Wait::Backoff(delay)); delay = backoff.after(delay); @@ -875,8 +926,10 @@ impl IndexSink for WorkerSink { } /// Queues the search for the worker and returns its ticket. An index that - /// is not current is not asked, and neither is a full queue: the refusal - /// is the answer, naming why and how many documents are pending. + /// is being rebuilt or cannot be asked is not asked, and neither is a full + /// queue: the refusal is the answer, naming why and how many documents are + /// pending. Documents admitted and not yet written are not a refusal: the + /// search waits behind them. fn search(&mut self, text: &str, top_k: u32) -> Result { let mut guard = lock(&self.shared); let status = including(&guard.status, u32::try_from(guard.inbox.len()).unwrap_or(u32::MAX)); @@ -891,7 +944,7 @@ impl IndexSink for WorkerSink { } let ticket = SearchTicket(guard.next_ticket); guard.next_ticket += 1; - guard.searches.push_back((ticket, text.to_owned(), top_k)); + guard.searches.push_back(Queued { ticket, text: text.to_owned(), top_k, due: Instant::now() + SEARCH_DEADLINE }); drop(guard); self.shared.1.notify_all(); Ok(ticket) @@ -905,7 +958,7 @@ impl IndexSink for WorkerSink { /// the worker already holds cannot be recalled; its answer is collected /// and dropped by the loop, which no longer waits for it. fn abandon(&mut self, ticket: SearchTicket) { - lock(&self.shared).searches.retain(|(queued, ..)| *queued != ticket); + lock(&self.shared).searches.retain(|queued| queued.ticket != ticket); } } @@ -932,6 +985,11 @@ impl Http { Self { agent, base: base_url.trim_end_matches('/').to_owned() } } + /// The same address under a bound of its own. + pub(crate) fn bounded(&self, timeout: Duration) -> Self { + Self::new(&self.base, timeout) + } + /// The status and the body, whatever the status is. pub(crate) fn request(&self, method: Method, path: &str, body: Option<&str>) -> Result<(u16, String)> { let url = format!("{}{path}", self.base); diff --git a/crates/huginn-daemon/src/index/ollama.rs b/crates/huginn-daemon/src/index/ollama.rs index 603bd08..0461729 100644 --- a/crates/huginn-daemon/src/index/ollama.rs +++ b/crates/huginn-daemon/src/index/ollama.rs @@ -6,7 +6,7 @@ use std::time::Duration; use anyhow::{Context, Result, bail, ensure}; use serde_json::{Value, json}; -use super::{BULK_EMBED_TIMEOUT, Embedder, Http, Method, ModelIdentity}; +use super::{Embedder, Http, Method, ModelIdentity}; /// The instruction Qwen3 embedding models take in front of a query (never in /// front of a document). Cut 11b's semantic read prepends it as @@ -14,18 +14,15 @@ use super::{BULK_EMBED_TIMEOUT, Embedder, Http, Method, ModelIdentity}; pub const QUERY_INSTRUCTION: &str = "Given a question about a project's pipeline state, retrieve the documents that answer it"; -/// One model on one Ollama. A cold model takes seconds to load, so the -/// identity reads get `BULK_EMBED_TIMEOUT`; the worker owns the schedule -/// around it. Each embed call carries its own bound. +/// One model on one Ollama. Every call carries its own bound. pub struct OllamaEmbedder { - http: Http, base_url: String, model: String, } impl OllamaEmbedder { pub fn new(base_url: &str, model: &str) -> Self { - Self { http: Http::new(base_url, BULK_EMBED_TIMEOUT), base_url: base_url.to_owned(), model: model.to_owned() } + Self { base_url: base_url.to_owned(), model: model.to_owned() } } fn ask(&self, http: &Http, method: Method, path: &str, body: Option<&Value>) -> Result { @@ -39,8 +36,9 @@ impl OllamaEmbedder { impl Embedder for OllamaEmbedder { /// The configured name, the digest Ollama lists for it, and the model's /// embedding length. The address is not part of the identity. - fn model_identity(&mut self) -> Result { - let tags = self.ask(&self.http, Method::Get, "/api/tags", None)?; + fn model_identity(&mut self, within: Duration) -> Result { + let http = Http::new(&self.base_url, within); + let tags = self.ask(&http, Method::Get, "/api/tags", None)?; let listed = tags["models"].as_array().context("Ollama's tag list has no models")?; let digest = listed .iter() @@ -48,7 +46,7 @@ impl Embedder for OllamaEmbedder { .and_then(|entry| entry["digest"].as_str()) .with_context(|| format!("Ollama does not list the model {}", self.model))? .to_owned(); - let shown = self.ask(&self.http, Method::Post, "/api/show", Some(&json!({ "model": self.model })))?; + let shown = self.ask(&http, Method::Post, "/api/show", Some(&json!({ "model": self.model })))?; let info = shown["model_info"].as_object().context("Ollama's model description has no model_info")?; let Some(length) = info.iter().find(|(key, _)| key.ends_with(".embedding_length")).and_then(|(_, value)| value.as_u64()) else { diff --git a/crates/huginn-daemon/src/index/qdrant.rs b/crates/huginn-daemon/src/index/qdrant.rs index 2999220..f4a1c85 100644 --- a/crates/huginn-daemon/src/index/qdrant.rs +++ b/crates/huginn-daemon/src/index/qdrant.rs @@ -7,7 +7,7 @@ use std::time::Duration; use anyhow::{Context, Result, ensure}; use serde_json::{Value, json}; -use super::{CollectionMeta, Described, Hit, Http, Method, Point, VectorIndex}; +use super::{CollectionMeta, INDEX_CALL_TIMEOUT, Described, Hit, Http, Method, Point, VectorIndex}; /// One Qdrant. Every path names a collection this organ was told to own; it /// never lists or touches another. @@ -20,12 +20,16 @@ const SCROLL_PAGE: u64 = 1000; impl QdrantIndex { pub fn new(base_url: &str) -> Self { - Self { http: Http::new(base_url, Duration::from_secs(60)) } + Self { http: Http::new(base_url, INDEX_CALL_TIMEOUT) } } fn ask(&self, method: Method, path: &str, body: Option<&Value>, ok: &[u16]) -> Result<(u16, Value)> { + self.ask_with(&self.http, method, path, body, ok) + } + + fn ask_with(&self, http: &Http, method: Method, path: &str, body: Option<&Value>, ok: &[u16]) -> Result<(u16, Value)> { let text = body.map(Value::to_string); - let (status, answer) = self.http.request(method, path, text.as_deref())?; + let (status, answer) = http.request(method, path, text.as_deref())?; ensure!(ok.contains(&status), "Qdrant answered {status} to {path}: {}", answer.chars().take(300).collect::()); let parsed = if answer.is_empty() { Value::Null } else { serde_json::from_str(&answer).unwrap_or(Value::Null) }; Ok((status, parsed)) @@ -67,8 +71,9 @@ fn described(metadata: &Value) -> Described { } impl VectorIndex for QdrantIndex { - fn describe(&mut self, collection: &str) -> Result { - let (status, answer) = self.ask(Method::Get, &format!("/collections/{collection}"), None, &[200, 404])?; + fn describe(&mut self, collection: &str, within: Duration) -> Result { + let http = self.http.bounded(within); + let (status, answer) = self.ask_with(&http, Method::Get, &format!("/collections/{collection}"), None, &[200, 404])?; if status == 404 { return Ok(Described::Absent); } @@ -133,9 +138,11 @@ impl VectorIndex for QdrantIndex { Ok(()) } - fn search(&mut self, collection: &str, vector: &[f32], limit: u32) -> Result> { + fn search(&mut self, collection: &str, vector: &[f32], limit: u32, within: Duration) -> Result> { + let http = self.http.bounded(within); let body = json!({ "vector": vector, "limit": limit, "with_payload": ["doc_id", "kind"], "with_vector": false }); - let (_, answer) = self.ask(Method::Post, &format!("/collections/{collection}/points/search"), Some(&body), &[200])?; + let (_, answer) = + self.ask_with(&http, Method::Post, &format!("/collections/{collection}/points/search"), Some(&body), &[200])?; let found = answer["result"].as_array().context("Qdrant's search has no result")?; found .iter() diff --git a/crates/huginn-daemon/src/index/tests.rs b/crates/huginn-daemon/src/index/tests.rs index c36699b..37d64f3 100644 --- a/crates/huginn-daemon/src/index/tests.rs +++ b/crates/huginn-daemon/src/index/tests.rs @@ -23,6 +23,9 @@ fn collection() -> String { collection_name(&slug(INSTANCE)) } +/// The bound the adapter tests give a call. +const TIME: Duration = Duration::from_secs(30); + /// A mind identity for tests that project entries without a mind behind them. const MIND: &str = "mind-commit-first"; @@ -308,25 +311,92 @@ fn change_label(index: &FakeIndex, change: impl FnOnce(&mut CollectionMeta)) { let Described::Labelled(meta) = &mut state.collections.get_mut(&collection()).unwrap().label else { panic!() }; change(meta); } - -/// The query embed is bounded by the search's deadline, one constant behind -/// both, so a query embed that cannot be recalled cannot hold the writes off -/// past the moment nobody is waiting for it. A bulk flush keeps its own bound. +/// Every call on the search path is bounded by what remains of the search's own +/// time: the identity reads, the label read, the embed and the query. A search +/// whose embed cannot be recalled therefore cannot hold the writes off past the +/// moment nobody is waiting for it. A bulk flush keeps its own, longer, bound. #[test] -fn the_query_embed_is_bounded_by_the_search_deadline_and_the_flush_by_its_own() { +fn every_call_on_the_search_path_is_bounded_by_the_searchs_time_and_the_flush_by_its_own() { let (_root, mind) = mind_with_documents(); let entries = mind.index_entries(None).unwrap(); let (embedder, index) = pair(); let mut projector = projected(&embedder, &index, &entries); - assert!(!embedder.state.lock().unwrap().bounds.is_empty()); - assert!(embedder.state.lock().unwrap().bounds.iter().all(|bound| *bound == BULK_EMBED_TIMEOUT), "the flushes"); - projector.search("anything", 3).unwrap(); - let bounds = embedder.state.lock().unwrap().bounds.clone(); - assert_eq!(bounds.last(), Some(&QUERY_EMBED_TIMEOUT), "the query"); + { + let state = embedder.state.lock().unwrap(); + assert!(!state.bounds.is_empty()); + let near = |bound: &Duration| *bound <= BULK_EMBED_TIMEOUT && *bound > BULK_EMBED_TIMEOUT - Duration::from_secs(60); + assert!(state.bounds.iter().all(near), "the flushes' embeds: {:?}", state.bounds); + assert!(state.identity_bounds.iter().all(near), "the reconciliation's and the flushes' identity reads: {:?}", state.identity_bounds); + } + embedder.state.lock().unwrap().bounds.clear(); + embedder.state.lock().unwrap().identity_bounds.clear(); + index.state.lock().unwrap().bounds.clear(); + + let within = Duration::from_secs(20); + projector.search("anything", 3, within).unwrap(); + let embedder = embedder.state.lock().unwrap(); + let index = index.state.lock().unwrap(); + assert_eq!(embedder.identity_bounds.len(), 2, "the model is read before and after the embed"); + assert_eq!(embedder.bounds.len(), 1); + assert_eq!(index.bounds.iter().map(|(call, _)| *call).collect::>(), ["describe", "search"]); + let bounds: Vec = embedder + .identity_bounds + .iter() + .chain(&embedder.bounds) + .copied() + .chain(index.bounds.iter().map(|(_, bound)| *bound)) + .collect(); + assert!(bounds.iter().all(|bound| !bound.is_zero() && *bound <= within), "{bounds:?}"); let served = crate::serve::ServeOptions::default().search_timeout; assert_eq!(served, SEARCH_DEADLINE); - assert!(QUERY_EMBED_TIMEOUT <= served); - assert!(QUERY_EMBED_TIMEOUT < BULK_EMBED_TIMEOUT); + assert!(SEARCH_DEADLINE < BULK_EMBED_TIMEOUT); +} + +/// A hung Ollama does not hold the worker past the search that asked: the +/// identity read never answers, and the search is answered, and the worker +/// free, when the search's own time has passed. A search that waited past its +/// time in the queue is answered without asking the embedder anything. A +/// ticket's time is `SEARCH_DEADLINE` from the moment it was asked. +#[test] +fn a_hung_embedder_holds_the_worker_no_longer_than_the_search_that_asked() { + let (_root, mind) = mind_with_documents(); + let entries = mind.index_entries(None).unwrap(); + let (embedder, index) = pair(); + let mut sink = WorkerSink::spawn(embedder.clone(), index, &slug(INSTANCE), Some(MIND.into()), entries, quick()); + sink.wait_status(|status| *status == IndexStatus::Current); + + let asked = Instant::now(); + ask_sink(&mut sink, "the ticket's time", 3).unwrap(); + let due = lock(&sink.shared).searches.back().expect("the search is queued").due; + assert!(due >= asked + SEARCH_DEADLINE && due <= Instant::now() + SEARCH_DEADLINE, "due is SEARCH_DEADLINE after it was asked"); + answers(&mut sink, 1); + + embedder.state.lock().unwrap().hang_identity = true; + let started = Instant::now(); + let due = started + Duration::from_millis(400); + lock(&sink.shared).searches.push_back(Queued { ticket: SearchTicket(70), text: "hung".into(), top_k: 3, due }); + sink.shared.1.notify_all(); + let answered = answers(&mut sink, 1); + assert_eq!(answered[0].0, SearchTicket(70)); + assert!(answered[0].1.is_err(), "{answered:?}"); + assert!(started.elapsed() < Duration::from_secs(5), "the worker was held for {:?}", started.elapsed()); + + embedder.state.lock().unwrap().hang_identity = false; + let reads = embedder.state.lock().unwrap().identity_bounds.len(); + lock(&sink.shared).searches.push_back(Queued { + ticket: SearchTicket(71), + text: "late".into(), + top_k: 3, + due: Instant::now().checked_sub(Duration::from_secs(1)).unwrap_or_else(Instant::now), + }); + sink.shared.1.notify_all(); + let late = answers(&mut sink, 1); + let Err(why) = &late[0].1 else { panic!("{late:?}") }; + assert!(why.contains("has run out"), "{why}"); + assert_eq!(embedder.state.lock().unwrap().identity_bounds.len(), reads, "nothing was asked for a search nobody waits for"); + + let ticket = ask_sink(&mut sink, "the worker is free", 3).expect("the worker is still current"); + assert_eq!(answers(&mut sink, 1)[0].0, ticket); } /// The Ollama adapter enforces the bound it is given: an embed that is never @@ -543,7 +613,7 @@ const EMBED: &str = r#"{"embeddings":[[0.5,1.0,2.0,3.0],[0.5,1.0,2.0,3.0],[0.5,1 fn ollama_identity_is_the_listed_digest_and_the_models_embedding_length() { let ollama = stub(&[("GET /api/tags", 200, TAGS), ("POST /api/show", 200, SHOW), ("POST /api/embed", 200, EMBED)]); let mut embedder = OllamaEmbedder::new(&ollama.url, "m:1"); - assert_eq!(embedder.model_identity().unwrap(), ModelIdentity { name: "m:1".into(), digest: "ac6d".into(), dimensions: 4 }); + assert_eq!(embedder.model_identity(TIME).unwrap(), ModelIdentity { name: "m:1".into(), digest: "ac6d".into(), dimensions: 4 }); let vectors = embedder.embed(&["a".into(), "b".into(), "c".into(), "d".into()], Duration::from_secs(5)).unwrap(); assert_eq!(vectors.len(), 4); assert_eq!(vectors[0], vec![0.5, 1.0, 2.0, 3.0]); @@ -557,10 +627,10 @@ fn ollama_identity_is_the_listed_digest_and_the_models_embedding_length() { ("GET /api/tags", 200, r#"{"models":[{"name":"m:1","digest":"ac6d"}]}"#), ("POST /api/show", 200, SHOW), ]); - assert_eq!(OllamaEmbedder::new(&bare.url, "m:1").model_identity().unwrap().digest, "ac6d"); + assert_eq!(OllamaEmbedder::new(&bare.url, "m:1").model_identity(TIME).unwrap().digest, "ac6d"); let mut absent = OllamaEmbedder::new(&ollama.url, "not-pulled:1"); - assert!(format!("{:#}", absent.model_identity().unwrap_err()).contains("does not list the model")); + assert!(format!("{:#}", absent.model_identity(TIME).unwrap_err()).contains("does not list the model")); let refusing = stub(&[("POST /api/embed", 500, r#"{"error":"boom"}"#)]); assert!(OllamaEmbedder::new(&refusing.url, "m:1").embed(&["a".into()], Duration::from_secs(5)).is_err()); } @@ -600,12 +670,12 @@ fn the_embedders_address_is_not_part_of_the_collections_compatibility() { fn qdrant_describes_creates_lists_and_writes_what_the_projector_asks() { let name = collection(); let absent = stub(&[("GET /collections/", 404, r#"{"status":{"error":"not found"}}"#)]); - assert_eq!(QdrantIndex::new(&absent.url).describe(&name).unwrap(), Described::Absent); + assert_eq!(QdrantIndex::new(&absent.url).describe(&name, TIME).unwrap(), Described::Absent); let unlabelled = stub(&[("GET /collections/", 200, r#"{"result":{"config":{"metadata":null}}}"#)]); - assert_eq!(QdrantIndex::new(&unlabelled.url).describe(&name).unwrap(), Described::Unlabelled); + assert_eq!(QdrantIndex::new(&unlabelled.url).describe(&name, TIME).unwrap(), Described::Unlabelled); let voidbot = stub(&[("GET /collections/", 200, r#"{"result":{"config":{"metadata":{"managedBy":"voidbot"}}}}"#)]); - assert_eq!(QdrantIndex::new(&voidbot.url).describe(&name).unwrap(), Described::Unlabelled); + assert_eq!(QdrantIndex::new(&voidbot.url).describe(&name, TIME).unwrap(), Described::Unlabelled); let labelled = stub(&[( "GET /collections/", @@ -613,7 +683,7 @@ fn qdrant_describes_creates_lists_and_writes_what_the_projector_asks() { r#"{"result":{"config":{"metadata":{"managed_by":"huginn","instance":"yggdrasil","mind":"mind-commit-first","model":"m:1","model_digest":"ac6d","dimensions":4,"index_text_version":1}}}}"#, )]); assert_eq!( - QdrantIndex::new(&labelled.url).describe(&name).unwrap(), + QdrantIndex::new(&labelled.url).describe(&name, TIME).unwrap(), Described::Labelled(CollectionMeta { managed_by: "huginn".into(), instance: "yggdrasil".into(), @@ -943,14 +1013,14 @@ fn a_search_embeds_the_query_with_its_instruction_and_oversamples() { Hit { doc_id: "eureka-state:mystery:X".into(), kind: "mystery".into(), score: 0.8 }, ]; - let hits = projector.search("who owns the state", 3).unwrap(); + let hits = projector.search("who owns the state", 3, SEARCH_DEADLINE).unwrap(); assert_eq!(hits.len(), 1, "the unknown kind is left out"); assert_eq!(hits[0].0.id.0, "eureka-state:question:Q1"); assert_eq!(hits[0].0.kind, PipelineKind::Question); assert_eq!(hits[0].1, 0.9); let embedded = embedder.texts(); assert_eq!(embedded.last().unwrap(), &format!("Instruct: {QUERY_INSTRUCTION}\nQuery: who owns the state")); - projector.search("again", 100).unwrap(); + projector.search("again", 100, SEARCH_DEADLINE).unwrap(); let searches = index.state.lock().unwrap().searches.clone(); assert_eq!(searches.iter().map(|(name, length, limit)| (name.as_str(), *length, *limit)).collect::>(), [ (collection().as_str(), 4, 12), @@ -967,13 +1037,13 @@ fn a_search_refuses_when_the_index_is_not_reconciled_with_the_model() { let entries = mind.index_entries(None).unwrap(); let (embedder, index) = pair(); let mut fresh = Projector::new(embedder.clone(), index.clone(), &slug(INSTANCE), Some(MIND.into())); - assert!(format!("{:#}", fresh.search("anything", 3).unwrap_err()).contains("being rebuilt")); + assert!(format!("{:#}", fresh.search("anything", 3, SEARCH_DEADLINE).unwrap_err()).contains("being rebuilt")); assert!(embedder.texts().is_empty() && index.state.lock().unwrap().searches.is_empty()); let mut projector = projected(&embedder, &index, &entries); embedder.state.lock().unwrap().identity.digest = "d2".into(); let before = embedder.texts().len(); - assert!(format!("{:#}", projector.search("anything", 3).unwrap_err()).contains("model changed")); + assert!(format!("{:#}", projector.search("anything", 3, SEARCH_DEADLINE).unwrap_err()).contains("model changed")); assert_eq!(embedder.texts().len(), before, "the query was not embedded under the new model"); assert!(index.state.lock().unwrap().searches.is_empty()); assert!(projector.stale(), "and the reconciliation is to be redone"); @@ -991,7 +1061,7 @@ fn qdrant_search_sends_no_filter_and_reads_ids_kinds_and_scores() { 200, r#"{"result":[{"id":"p1","score":0.75,"payload":{"doc_id":"eureka-state:question:Q1","kind":"question"}},{"id":"p2","score":0.5,"payload":{"doc_id":"eureka-state:ruling:R1","kind":"ruling"}}]}"#, )]); - let hits = QdrantIndex::new(&answering.url).search(&name, &[0.5, 1.0, 2.0, 3.0], 12).unwrap(); + let hits = QdrantIndex::new(&answering.url).search(&name, &[0.5, 1.0, 2.0, 3.0], 12, TIME).unwrap(); assert_eq!( hits, [ @@ -1008,15 +1078,19 @@ fn qdrant_search_sends_no_filter_and_reads_ids_kinds_and_scores() { drop(seen); let nameless = stub(&[("POST /collections/", 200, r#"{"result":[{"id":"p","score":0.5,"payload":{"kind":"question"}}]}"#)]); - assert!(QdrantIndex::new(&nameless.url).search(&name, &[1.0], 1).is_err()); + assert!(QdrantIndex::new(&nameless.url).search(&name, &[1.0], 1, TIME).is_err()); let failing = stub(&[("POST /collections/", 404, r#"{"status":{"error":"not found"}}"#)]); - assert!(QdrantIndex::new(&failing.url).search(&name, &[1.0], 1).is_err()); + assert!(QdrantIndex::new(&failing.url).search(&name, &[1.0], 1, TIME).is_err()); +} + +fn queued(ticket: u64, text: &str) -> Queued { + Queued { ticket: SearchTicket(ticket), text: text.into(), top_k: 3, due: Instant::now() + SEARCH_DEADLINE } } fn shared_with( inbox: Vec, mind: Option, - searches: Vec<(SearchTicket, String, u32)>, + searches: Vec, closed: bool, ) -> Lock { Arc::new(( @@ -1048,7 +1122,7 @@ fn an_idle_wait_ends_for_work_and_otherwise_at_its_deadline() { }; assert!(waits(&shared_with(vec![entry], None, Vec::new(), false)), "an entry"); assert!(waits(&shared_with(Vec::new(), Some(MIND.into()), Vec::new(), false)), "the mind's identity"); - assert!(waits(&shared_with(Vec::new(), None, vec![(SearchTicket(0), "q".into(), 3)], false)), "a search"); + assert!(waits(&shared_with(Vec::new(), None, vec![queued(0, "q")], false)), "a search"); assert!(waits(&shared_with(Vec::new(), None, Vec::new(), true)), "the sink closing"); let quiet = shared_with(Vec::new(), None, Vec::new(), false); @@ -1066,7 +1140,7 @@ fn an_idle_wait_ends_for_work_and_otherwise_at_its_deadline() { fn a_backoff_wait_is_the_whole_delay_whatever_arrives() { let (_root, mind) = mind_with_documents(); let entry = mind.index_entries(None).unwrap().remove(0); - let busy = shared_with(vec![entry], Some(MIND.into()), vec![(SearchTicket(0), "q".into(), 3)], false); + let busy = shared_with(vec![entry], Some(MIND.into()), vec![queued(0, "q")], false); let started = Instant::now(); wait_for_work(&busy, Wait::Backoff(Duration::from_millis(80))); assert!(started.elapsed() >= Duration::from_millis(80), "the delay was waited out although work was queued"); @@ -1080,8 +1154,7 @@ fn a_backoff_wait_is_the_whole_delay_whatever_arrives() { /// reason, when the worker fails: nothing is left to wait for its deadline. #[test] fn queued_searches_are_answered_with_why_when_the_worker_gives_up() { - let queued = vec![(SearchTicket(3), "a".into(), 1), (SearchTicket(4), "b".into(), 2)]; - let shared = shared_with(Vec::new(), None, queued, false); + let shared = shared_with(Vec::new(), None, vec![queued(3, "a"), queued(4, "b")], false); refuse_searches(&shared, "the index could not advance: down"); let guard = lock(&shared); assert!(guard.searches.is_empty()); @@ -1141,13 +1214,15 @@ fn queries(embedder: &FakeEmbedder) -> Vec { embedder.texts().into_iter().filter(|text| text.starts_with("Instruct: ")).collect() } -/// Only a current index answers. Being reconciled, being written to, failing -/// and being refused each refuse by name, and the count of documents pending -/// is in the words. +/// A search is refused only while the index is being rebuilt or cannot be +/// asked, and the words name the state and how many documents are pending. Lag +/// is not a rebuild: the sink and the projector ask the one predicate and both +/// let it through. #[test] -fn only_a_current_index_answers_a_search() { +fn a_search_is_refused_only_for_a_rebuild_or_a_failure() { use IndexStatus as S; assert_eq!(unsearchable(&S::Current), None); + assert_eq!(unsearchable(&S::Behind { pending: 5 }), None, "lag is waited out, not refused"); let says = |status: S, wanted: &[&str]| { let why = unsearchable(&status).unwrap_or_else(|| panic!("{status:?} answered")); for word in wanted { @@ -1155,65 +1230,68 @@ fn only_a_current_index_answers_a_search() { } }; says(S::Reconciling { pending: 7 }, &["being rebuilt", "reconciling", "pending: 7"]); - says(S::Behind { pending: 5 }, &["being rebuilt", "behind", "pending: 5"]); says(S::Failing { pending: 3, attempts: 2, error: "boom".into() }, &["failing", "boom", "pending: 3"]); says(S::Refused { pending: 4, attempts: 2, reason: "not ours".into() }, &["refused", "not ours", "pending: 4"]); - // The projector never answers from a half-filled collection, and never - // touches the embedder or the vector store to find out. + // The sink: what the worker last published, and the documents admitted + // since, which are lag whatever the published state says of a current index. + let sink_in = |status: S, inbox: usize| { + let shared = shared_with(entries_of(inbox), None, Vec::new(), false); + lock(&shared).status = status; + WorkerSink { shared, identified: true } + }; + assert!(ask_sink(&mut sink_in(S::Current, 2), "anything", 3).is_ok(), "admitted a moment ago and not yet absorbed"); + assert!(ask_sink(&mut sink_in(S::Behind { pending: 4 }, 0), "anything", 3).is_ok(), "the worker is writing"); + let refusal = ask_sink(&mut sink_in(S::Reconciling { pending: 4 }, 2), "anything", 3).unwrap_err(); + assert!(refusal.contains("being rebuilt") && refusal.contains("pending: 6"), "{refusal}"); + let refusal = ask_sink(&mut sink_in(S::Failing { pending: 1, attempts: 1, error: "boom".into() }, 0), "anything", 3).unwrap_err(); + assert!(refusal.contains("failing") && refusal.contains("boom"), "{refusal}"); + + // The projector: a rebuild refuses without touching the ports to find out. let (_root, mind) = mind_with_documents(); let entries = mind.index_entries(None).unwrap(); let (embedder, index) = pair(); - let mut projector = projected(&embedder, &index, &entries[..3]); - projector.want(entries[3..].iter().cloned()); - let embedded = embedder.texts().len(); - let refusal = format!("{:#}", projector.search("anything", 3).unwrap_err()); - assert!(refusal.contains("being rebuilt") && refusal.contains("pending: 1"), "{refusal}"); - assert_eq!(embedder.texts().len(), embedded); - assert!(index.state.lock().unwrap().searches.is_empty()); - drain(&mut projector); - assert!(projector.search("anything", 3).is_ok(), "once it has caught up it answers"); - - // The sink refuses before the worker is asked, while the worker is inside a batch. - let gate = Gate::shut(); - let mut sink = WorkerSink::spawn( - FakeEmbedder::new("d1").behind(&gate), - FakeIndex::default(), - &slug(INSTANCE), - Some(MIND.into()), - entries.clone(), - quick(), - ); - gate.wait_arrived(1); - let refusal = ask_sink(&mut sink, "anything", 3).unwrap_err(); + let mut fresh = Projector::new(embedder.clone(), index.clone(), &slug(INSTANCE), Some(MIND.into())); + fresh.want(entries.iter().cloned()); + let refusal = format!("{:#}", fresh.search("anything", 3, SEARCH_DEADLINE).unwrap_err()); assert!(refusal.contains("being rebuilt") && refusal.contains("pending: 4"), "{refusal}"); - gate.open(); - sink.wait_status(|status| *status == IndexStatus::Current); - assert!(ask_sink(&mut sink, "anything", 3).is_ok()); + assert!(embedder.texts().is_empty() && index.state.lock().unwrap().searches.is_empty()); + + // Lag it lets through: catching up is the worker's, before the search is taken. + let mut lagging = projected(&embedder, &index, &entries[..3]); + lagging.want(entries[3..].iter().cloned()); + assert_eq!(lagging.progress_status(), IndexStatus::Behind { pending: 1 }); + assert!(lagging.search("anything", 3, SEARCH_DEADLINE).is_ok()); } -/// A search that fails after it was accepted has found the index not what it -/// claimed: the reconciliation is forgotten, so the status stops reading -/// `Current` and the worker verifies the model and the collection again. +/// A search that fails for its own reasons (the vector store errs, the +/// embedder is down, an identity read times out) fails that search alone: the +/// reconciliation stands, the status stays `Current`, the collection is not +/// listed again, and the next search is asked and answered. #[test] -fn a_search_that_fails_forgets_the_reconciliation() { +fn a_failed_search_fails_that_search_alone() { let (_root, mind) = mind_with_documents(); let entries = mind.index_entries(None).unwrap(); - let breaks: [(&str, Box); 2] = [ - ("the vector store fails the search", Box::new(|_, index| index.state.lock().unwrap().fail_search = true)), - ("the embedder is down", Box::new(|embedder, _| embedder.state.lock().unwrap().down = true)), + type Fault = Box; + let faults: [(&str, Fault); 3] = [ + ("the vector store fails the search", Box::new(|_, index, on| index.state.lock().unwrap().fail_search = on)), + ("the embedder is down", Box::new(|embedder, _, on| embedder.state.lock().unwrap().down = on)), + ("the identity read never answers", Box::new(|embedder, _, on| embedder.state.lock().unwrap().hang_identity = on)), ]; - for (what, breaks) in breaks { + for (what, fault) in faults { let (embedder, index) = pair(); let mut projector = projected(&embedder, &index, &entries); - assert_eq!(projector.progress_status(), IndexStatus::Current); - breaks(&embedder, &index); - assert!(projector.search("anything", 3).is_err(), "{what}"); - assert_eq!(projector.progress_status(), IndexStatus::Reconciling { pending: 0 }, "{what}"); - assert!(projector.stale(), "{what}"); + let listed = index.state.lock().unwrap().listed; + fault(&embedder, &index, true); + assert!(projector.search("anything", 3, Duration::from_millis(50)).is_err(), "{what}"); + assert_eq!(projector.progress_status(), IndexStatus::Current, "{what}"); + assert!(!projector.stale(), "{what}"); + fault(&embedder, &index, false); + assert!(projector.search("anything", 3, SEARCH_DEADLINE).is_ok(), "{what}: the next search is answered"); + assert_eq!(index.state.lock().unwrap().listed, listed, "{what}: the collection was not listed again"); } - // In the worker the next step is the verification, with nothing to prompt it. + // In the worker the failed search is answered and nothing else moves. let (embedder, index) = pair(); let mut sink = WorkerSink::spawn(embedder, index.clone(), &slug(INSTANCE), Some(MIND.into()), entries, quick()); sink.wait_status(|status| *status == IndexStatus::Current); @@ -1223,7 +1301,24 @@ fn a_search_that_fails_forgets_the_reconciliation() { let failed = answers(&mut sink, 1); assert_eq!(failed[0].0, ticket); assert!(failed[0].1.is_err()); - eventually("the worker listing the collection again", || index.state.lock().unwrap().listed > listed); + index.state.lock().unwrap().fail_search = false; + let next = ask_sink(&mut sink, "anything", 3).expect("the index is still current, not being rebuilt"); + let answered = answers(&mut sink, 1); + assert_eq!((answered[0].0, answered[0].1.is_ok()), (next, true)); + assert_eq!(index.state.lock().unwrap().listed, listed, "the worker did not list the collection again"); +} + +/// A model the search finds changed is a rebuild, and its refusal names the +/// state and how many documents are pending, as every other refusal does. +#[test] +fn a_model_changed_refusal_names_the_state_and_the_pending_count() { + let (_root, mind) = mind_with_documents(); + let entries = mind.index_entries(None).unwrap(); + let (embedder, index) = pair(); + let mut projector = projected(&embedder, &index, &entries); + embedder.state.lock().unwrap().identity.digest = "d2".into(); + let refusal = format!("{:#}", projector.search("anything", 3, SEARCH_DEADLINE).unwrap_err()); + assert!(refusal.contains("model changed") && refusal.contains("being rebuilt") && refusal.contains("pending: 4"), "{refusal}"); } /// A search asks the store what the collection says about itself: a label for @@ -1252,7 +1347,7 @@ fn a_search_re_reads_the_collections_label() { let mut projector = projected(&embedder, &index, &entries); change(index.state.lock().unwrap().collections.get_mut(&collection()).unwrap()); let embedded = embedder.texts().len(); - let refusal = format!("{:#}", projector.search("anything", 3).unwrap_err()); + let refusal = format!("{:#}", projector.search("anything", 3, SEARCH_DEADLINE).unwrap_err()); assert!(refusal.contains("no longer the one this mind's index was built as"), "{what}: {refusal}"); assert_eq!(embedder.texts().len(), embedded, "{what}: nothing was embedded"); assert!(index.state.lock().unwrap().searches.is_empty(), "{what}"); @@ -1261,7 +1356,7 @@ fn a_search_re_reads_the_collections_label() { let (embedder, index) = pair(); let mut projector = projected(&embedder, &index, &entries); index.state.lock().unwrap().collections.clear(); - assert!(projector.search("anything", 3).is_err(), "no collection"); + assert!(projector.search("anything", 3, SEARCH_DEADLINE).is_err(), "no collection"); assert!(projector.stale()); } @@ -1278,7 +1373,7 @@ fn a_model_that_changes_while_the_query_is_embedded_is_refused() { let mut projector = projected(&embedder, &index, &entries); gate.shut_again(); let searching = std::thread::spawn(move || { - let found = projector.search("anything", 3); + let found = projector.search("anything", 3, SEARCH_DEADLINE); (projector, found) }); gate.wait_arrived(1); @@ -1372,33 +1467,72 @@ fn one_queued_search_is_served_per_turn() { let entries = mind.index_entries(None).unwrap(); let (embedder, index) = pair(); let mut projector = projected(&embedder, &index, &entries); - let queued = vec![(SearchTicket(1), "a".into(), 3), (SearchTicket(2), "b".into(), 3), (SearchTicket(3), "c".into(), 3)]; - let shared = shared_with(Vec::new(), None, queued, false); + let shared = shared_with(Vec::new(), None, vec![queued(1, "a"), queued(2, "b"), queued(3, "c")], false); serve_search(&mut projector, &shared); let guard = lock(&shared); - assert_eq!(guard.searches.iter().map(|(ticket, ..)| *ticket).collect::>(), [SearchTicket(2), SearchTicket(3)]); + assert_eq!(guard.searches.iter().map(|queued| queued.ticket).collect::>(), [SearchTicket(2), SearchTicket(3)]); assert_eq!(guard.found.iter().map(|(ticket, _)| *ticket).collect::>(), [SearchTicket(1)]); } -/// Searches are answered between batches, not after the last: a search waiting -/// while the first of two batches is written is answered, refused for the -/// documents still pending, before the second batch is embedded. +/// A search reads its own writes. Asked while a batch is in the embedder and +/// eight more documents wait, it is accepted, waits behind them, and is +/// answered only once every one is in the collection. #[test] -fn a_waiting_search_is_answered_between_batches() { +fn a_search_waits_for_the_writes_ahead_of_it_and_reads_them() { let gate = Gate::shut(); let (_, index) = pair(); let embedder = FakeEmbedder::new("d1").behind(&gate); - let mut sink = WorkerSink::spawn(embedder, index, &slug(INSTANCE), Some(MIND.into()), entries_of(BATCH + 8), quick()); + let mut sink = WorkerSink::spawn(embedder, index.clone(), &slug(INSTANCE), Some(MIND.into()), entries_of(BATCH + 8), quick()); gate.wait_arrived(1); - lock(&sink.shared).searches.push_back((SearchTicket(9), "anything".into(), 3)); + let ticket = ask_sink(&mut sink, "anything", 3).expect("lag is not a rebuild"); gate.open(); let answered = answers(&mut sink, 1); - assert_eq!(answered[0].0, SearchTicket(9)); - let Err(refusal) = &answered[0].1 else { panic!("a search was answered from a half-filled collection: {answered:?}") }; - assert!(refusal.contains("being rebuilt") && refusal.contains("pending: 8"), "{refusal}"); + assert_eq!(answered[0].0, ticket); + assert!(answered[0].1.is_ok(), "{answered:?}"); + assert_eq!(index.state.lock().unwrap().held_when_searched, [BATCH + 8], "the search saw every document admitted before it"); sink.wait_status(|status| *status == IndexStatus::Current); } +/// The worker takes admissions before the search, and leaves the search queued +/// while the writes ahead of it are unfinished: a search asked with a document +/// still in the inbox absorbs it and waits, and once the index has caught up +/// it is answered. An index being rebuilt is not waited on: the search is taken +/// and refused, naming the state and the pending count. +#[test] +fn the_worker_absorbs_admissions_before_a_search_and_waits_for_lag_but_not_for_a_rebuild() { + let (_root, mind) = mind_with_documents(); + let entries = mind.index_entries(None).unwrap(); + let (embedder, index) = pair(); + let mut projector = projected(&embedder, &index, &entries[..3]); + let shared = shared_with(vec![entries[3].clone()], None, vec![queued(1, "a")], false); + + serve_search(&mut projector, &shared); + { + let guard = lock(&shared); + assert!(guard.inbox.is_empty(), "the admission ahead of the search was absorbed"); + assert_eq!(guard.searches.len(), 1, "the search waits for it to be written"); + assert!(guard.found.is_empty()); + } + assert_eq!(projector.pending(), 1); + + drain(&mut projector); + serve_search(&mut projector, &shared); + let guard = lock(&shared); + assert!(guard.searches.is_empty()); + assert!(matches!(guard.found.as_slice(), [(SearchTicket(1), Ok(_))]), "{:?}", guard.found); + assert_eq!(index.state.lock().unwrap().held_when_searched, [4]); + drop(guard); + + // A rebuild: the projector has forgotten its reconciliation with the entries owed. + let mut rebuilding = Projector::new(embedder.clone(), index.clone(), &slug(INSTANCE), Some(MIND.into())); + rebuilding.want(entries.iter().cloned()); + let shared = shared_with(Vec::new(), None, vec![queued(2, "b")], false); + serve_search(&mut rebuilding, &shared); + let guard = lock(&shared); + let [(SearchTicket(2), Err(refusal))] = guard.found.as_slice() else { panic!("{:?}", guard.found) }; + assert!(refusal.contains("being rebuilt") && refusal.contains("pending: 4"), "{refusal}"); +} + /// A search that was queued when the worker failed is answered with why, by /// the worker's own failure path. #[test] @@ -1408,13 +1542,13 @@ fn a_search_queued_before_the_worker_fails_is_answered_with_why() { let embedder = FakeEmbedder::new("d1").behind(&gate); let mut sink = WorkerSink::spawn(embedder.clone(), index, &slug(INSTANCE), Some(MIND.into()), entries_of(1), quick()); gate.wait_arrived(1); - lock(&sink.shared).searches.push_back((SearchTicket(5), "anything".into(), 3)); + lock(&sink.shared).searches.push_back(queued(5, "anything")); embedder.state.lock().unwrap().down = true; gate.open(); let answered = answers(&mut sink, 1); assert_eq!(answered[0].0, SearchTicket(5)); let Err(why) = &answered[0].1 else { panic!("{answered:?}") }; - assert!(why.contains("the index could not advance") && why.contains("the embedder is down"), "{why}"); + assert!(why.contains("the semantic index is failing") && why.contains("pending: 1") && why.contains("the embedder is down"), "{why}"); } /// While a failure is being reported, the report's pending count is the one diff --git a/crates/huginn-daemon/src/semantic_tests.rs b/crates/huginn-daemon/src/semantic_tests.rs index 6880009..9a5646b 100644 --- a/crates/huginn-daemon/src/semantic_tests.rs +++ b/crates/huginn-daemon/src/semantic_tests.rs @@ -461,12 +461,13 @@ fn a_search_that_timed_out_while_queued_is_never_embedded() { assert!(embedded[0].ends_with("Query: first") && embedded[1].ends_with("Query: third"), "{embedded:?}"); } -/// While the index is rebuilding, a semantic query is refused with an -/// `Unavailable` that names the state and how many documents are pending, and -/// the embedder is not asked for it. When the index has caught up the same -/// query is answered. +/// A search asked while the index lags the mind is not refused: it waits behind +/// the writes ahead of it, through the real serve loop, and reads them. The +/// worker is inside its first batch with the rest of the mind still to write; +/// the reply does not come until the gate opens, and it comes with every +/// document already in the collection. #[test] -fn a_search_while_the_index_is_being_rebuilt_is_refused_and_asks_the_embedder_nothing() { +fn a_search_asked_while_the_index_lags_waits_for_the_writes_ahead_of_it_and_reads_them() { let gate = Gate::shut(); let embedder = FakeEmbedder::new("d1").behind(&gate); let harness = Harness::start(shelf(), embedder, FakeIndex::default(), Duration::from_secs(30)); @@ -474,14 +475,17 @@ fn a_search_while_the_index_is_being_rebuilt_is_refused_and_asks_the_embedder_no let mut client = harness.client(); let IndexStatus::Behind { pending } = index_status(&mut client, "m-status") else { panic!("the worker is inside its first batch") }; assert!(pending > 0); + harness.index.state.lock().unwrap().hits = vec![hit("ruling", id("ruling", "R4"), 0.5)]; - let refused = ask(&mut client, "m-behind", &semantic(standing(), 5)); - let HuginnMindResponse::Refused(MindRefusal::Unavailable { detail }) = &refused else { panic!("{refused:?}") }; - assert!(detail.contains("being rebuilt") && detail.contains(&format!("pending: {pending}")), "{detail}"); - assert!(harness.embedder.texts().is_empty(), "nothing was embedded, not even the documents, for a refused search"); + send(&mut client, "m-behind", &semantic(standing(), 5)); + assert!(receive(&mut client, "m-behind", 100).is_none(), "the search waits: it is neither refused nor answered from a half-filled collection"); gate.open(); + let answered = receive(&mut client, "m-behind", 3000).expect("the search is answered once the index has caught up"); + assert_eq!(ids(&answered), vec![id("ruling", "R4")]); until_index(&mut client, "Current", |status| *status == IndexStatus::Current); - harness.index.state.lock().unwrap().hits = vec![hit("ruling", id("ruling", "R4"), 0.5)]; - assert_eq!(ids(&ask(&mut client, "m-current", &semantic(standing(), 5))), vec![id("ruling", "R4")]); + let state = harness.index.state.lock().unwrap(); + let written = state.collections[&collection_name(&slug(INSTANCE))].points.len(); + assert_eq!(state.held_when_searched, [written], "the search saw every document"); + assert!(written >= pending as usize); } diff --git a/crates/huginn-daemon/tests/live_index_smoke.rs b/crates/huginn-daemon/tests/live_index_smoke.rs index 7f6b4d4..85fabc2 100644 --- a/crates/huginn-daemon/tests/live_index_smoke.rs +++ b/crates/huginn-daemon/tests/live_index_smoke.rs @@ -44,7 +44,7 @@ fn the_adapters_embed_write_query_and_clean_up_a_scratch_collection() { let dotted = format!("{plain}.dotted"); let mut embedder = OllamaEmbedder::new(&ollama, &model); - let identity = embedder.model_identity().expect("the model is listed"); + let identity = embedder.model_identity(Duration::from_secs(60)).expect("the model is listed"); println!("identity: {identity:?}"); let texts = ["The organ owns the mind.".to_string(), "A recipe for lentil soup.".to_string()]; let vectors = embedder.embed(&texts, Duration::from_secs(60)).expect("two texts embed"); @@ -63,9 +63,9 @@ fn the_adapters_embed_write_query_and_clean_up_a_scratch_collection() { let mut store = QdrantIndex::new(&qdrant); let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { for name in [&plain, &dotted] { - assert_eq!(store.describe(name).unwrap(), Described::Absent); + assert_eq!(store.describe(name, Duration::from_secs(60)).unwrap(), Described::Absent); store.recreate(name, &meta).unwrap(); - assert_eq!(store.describe(name).unwrap(), Described::Labelled(meta.clone())); + assert_eq!(store.describe(name, Duration::from_secs(60)).unwrap(), Described::Labelled(meta.clone())); assert!(store.ids(name).unwrap().is_empty()); let points: Vec = texts .iter() @@ -97,7 +97,7 @@ fn the_adapters_embed_write_query_and_clean_up_a_scratch_collection() { let asked = format!("Instruct: {QUERY_INSTRUCTION} Query: Who is responsible for keeping the memory?"); let query = embedder.embed(&[asked], Duration::from_secs(60)).expect("the query embeds").remove(0); - let hits = store.search(name, &query, 2).unwrap(); + let hits = store.search(name, &query, 2, Duration::from_secs(60)).unwrap(); assert_eq!(hits.len(), 2, "{hits:?}"); assert_eq!((hits[0].doc_id.as_str(), hits[0].kind.as_str()), ("smoke:doc:0", "question"), "{hits:?}"); assert!(hits[0].score > hits[1].score, "{hits:?}");