diff --git a/crates/huginn-daemon/src/daemon.rs b/crates/huginn-daemon/src/daemon.rs index 51a9fe6..c42d644 100644 --- a/crates/huginn-daemon/src/daemon.rs +++ b/crates/huginn-daemon/src/daemon.rs @@ -28,6 +28,10 @@ pub trait IndexSink { /// Every search that finished since the last call, by ticket: its /// candidates and scores, or why it failed. fn searched(&mut self) -> Vec<(SearchTicket, Result)>; + /// The asker has stopped waiting for `ticket`. A search not yet started is + /// dropped, so a queue of searches nobody wants cannot hold the index's + /// writes off. + fn abandon(&mut self, ticket: SearchTicket); } /// Names one search the index was asked, so the loop can match the answer to @@ -161,6 +165,11 @@ impl> Daemon { self.index.searched() } + /// The client has stopped waiting for `ticket`. + pub fn abandon(&mut self, ticket: SearchTicket) { + self.index.abandon(ticket); + } + /// The answer to a search: the index's candidates ranked through the mind /// (`Mind::rank` drops what the mind does not hold and lets the selection /// decide the rest), or, when the index could not answer, a typed @@ -438,6 +447,8 @@ pub(crate) mod tests { fn searched(&mut self) -> Vec<(SearchTicket, Result)> { Vec::new() } + + fn abandon(&mut self, _ticket: SearchTicket) {} } /// A daemon over the instance's mind under `state_root`, with no index. @@ -472,6 +483,8 @@ pub(crate) mod tests { fn searched(&mut self) -> Vec<(SearchTicket, Result)> { Vec::new() } + + fn abandon(&mut self, _ticket: SearchTicket) {} } struct FailingIndex; @@ -492,6 +505,8 @@ pub(crate) mod tests { fn searched(&mut self) -> Vec<(SearchTicket, Result)> { Vec::new() } + + fn abandon(&mut self, _ticket: SearchTicket) {} } fn status(daemon: &mut Daemon>) -> MindStatus { diff --git a/crates/huginn-daemon/src/index/fakes.rs b/crates/huginn-daemon/src/index/fakes.rs index d7e4eee..3ed7560 100644 --- a/crates/huginn-daemon/src/index/fakes.rs +++ b/crates/huginn-daemon/src/index/fakes.rs @@ -24,9 +24,12 @@ impl Gate { /// Blocks until `count` calls to `embed` have arrived at the gate. pub(crate) fn wait_arrived(&self, count: u32) { + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30); let mut state = self.0.0.lock().unwrap(); while state.0 < count { - state = self.0.1.wait(state).unwrap(); + let left = deadline.saturating_duration_since(std::time::Instant::now()); + assert!(!left.is_zero(), "{count} calls never arrived at the gate; {} did", state.0); + state = self.0.1.wait_timeout(state, left).unwrap().0; } } @@ -121,6 +124,10 @@ pub(crate) struct IndexState { pub(crate) hits: Vec, /// The collection, vector length and limit of every search, in order. pub(crate) searches: Vec<(String, usize, u32)>, + /// How many times every point id was listed: only a reconciliation does. + pub(crate) listed: u32, + /// Whether a search fails at the vector store, the rest of it working. + pub(crate) fail_search: bool, } #[derive(Clone, Default)] @@ -162,10 +169,11 @@ impl VectorIndex for FakeIndex { } fn ids(&mut self, collection: &str) -> Result> { - let state = self.state.lock().unwrap(); + let mut state = self.state.lock().unwrap(); if state.down { bail!("the vector store is down"); } + state.listed += 1; Ok(state.collections.get(collection).map(|stored| stored.points.keys().cloned().collect()).unwrap_or_default()) } @@ -188,6 +196,7 @@ impl VectorIndex for FakeIndex { 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}"); Ok(state.hits.clone()) } diff --git a/crates/huginn-daemon/src/index/mod.rs b/crates/huginn-daemon/src/index/mod.rs index 9a943ba..1d7aaa6 100644 --- a/crates/huginn-daemon/src/index/mod.rs +++ b/crates/huginn-daemon/src/index/mod.rs @@ -26,10 +26,10 @@ //! re-reads the model's identity (name, digest, dimensions; about 0.3 s against //! Ollama) and, if it is not the one the collection was reconciled under, //! forgets the reconciliation instead of writing: a model re-pulled under the -//! same dimensions never mixes its vectors with the old model's. An idle -//! worker forgets the reconciliation every `Backoff::recheck` (60 s by -//! default), which re-reads the model, describes the collection and lists its -//! points: a collection lost while nothing was being written is found within +//! same dimensions never mixes its vectors with the old model's. The worker +//! forgets the reconciliation every `Backoff::recheck` (60 s by default) on a +//! schedule that neither admissions nor searches move, which re-reads the model, +//! describes the collection and lists its points: a collection lost while nothing was being written is found within //! that interval, and until then `Current` is what the worker last verified. //! There is one judge either way: `reconcile`. Refilling needs //! the text, so the projector keeps every entry it was given (`known`) beside @@ -60,9 +60,19 @@ //! them back through `Mind::rank`, which drops ids the mind does not hold and //! lets the selection decide the rest, in-force derivation included. //! - A search runs on the worker between batches, so it waits for the batch in -//! flight. The serve loop never waits: it holds the reply and answers when -//! the worker has (`ServeOptions::search_timeout` bounds the wait). A search -//! is refused at once while the worker is `Failing` or `Refused`. +//! flight, and at most one runs between two steps of the writes. The serve +//! loop never waits: it holds the reply and answers when the worker has +//! (`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 Ollama's own timeout. +//! - 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. pub mod ollama; pub mod qdrant; @@ -339,15 +349,7 @@ impl Projector { 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 wanted = CollectionMeta { - managed_by: MANAGED_BY.into(), - instance: self.instance.clone(), - mind: mind.clone(), - model: identity.name.clone(), - model_digest: identity.digest.clone(), - dimensions: identity.dimensions, - index_text_version: INDEX_TEXT_VERSION, - }; + let wanted = self.collection_meta(&mind, &identity); match self.index.describe(&self.collection).context("describing the collection")? { Described::Absent => self.recreate(&wanted)?, Described::Unlabelled => { @@ -373,6 +375,20 @@ impl Projector { Ok(()) } + /// What the collection must say about itself to be this mind's, under + /// this model. + fn collection_meta(&self, mind: &str, identity: &ModelIdentity) -> CollectionMeta { + CollectionMeta { + managed_by: MANAGED_BY.into(), + instance: self.instance.clone(), + mind: mind.to_owned(), + model: identity.name.clone(), + model_digest: identity.digest.clone(), + dimensions: identity.dimensions, + index_text_version: INDEX_TEXT_VERSION, + } + } + fn recreate(&mut self, meta: &CollectionMeta) -> Result<()> { self.index.recreate(&self.collection, meta).context("creating the collection") } @@ -424,22 +440,56 @@ impl Projector { /// The candidates a query text finds: the text is embedded with the model's /// query instruction and the collection is searched, with no filter and an /// oversample of `min(top_k * 4, 200)`, because the mind will drop most of - /// what the index cannot know is not in force. A model that has changed - /// since the reconciliation refuses the search rather than compare vectors - /// of two models. A hit naming a kind this organ has no name for is logged - /// and left out. + /// 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 { - let Some(dimensions) = self.reconciled.as_ref().map(|model| model.dimensions as usize) else { - bail!("the index has not been reconciled with the model and the collection"); + 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 (Some(mind), Some(reconciled)) = (self.mind.clone(), self.reconciled.clone()) else { + bail!("the mind has no identity yet, so it has no index"); }; + let dimensions = reconciled.dimensions as usize; ensure!( self.model_unchanged().map_err(|error| anyhow::anyhow!("{error}"))?, "the embedding model changed since the collection was built; the index is being rebuilt for it" ); + let described = self.index.describe(&self.collection).context("describing the collection")?; + ensure!( + described == Described::Labelled(self.collection_meta(&mind, &reconciled)), + "collection {} is no longer the one this mind's index was built as: {described:?}", + self.collection + ); let query = format!("Instruct: {QUERY_INSTRUCTION}\nQuery: {text}"); let mut vectors = self.embedder.embed(&[query]).context("embedding the query")?; ensure!(vectors.len() == 1, "the embedder returned {} vectors for one query", vectors.len()); let vector = vectors.remove(0); + // The model is read again after the query was embedded: only a vector + // made between two reads of the same model is compared with the + // collection. + ensure!( + self.model_unchanged().map_err(|error| anyhow::anyhow!("{error}"))?, + "the embedding model changed while the query was embedded; the index is being rebuilt for it" + ); 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 mut hits = Vec::with_capacity(found.len()); @@ -495,13 +545,20 @@ impl Backoff { } } +/// The most searches the worker holds waiting. A search asked when this many +/// wait is refused, by name, at once: the queue is bounded by the asker being +/// told, not by the worker's patience. +pub const SEARCHES_QUEUED_MAX: usize = 16; + struct Shared { inbox: Vec, /// The mind's identity, once a commit has revealed it to a mind that had none. mind: Option, status: IndexStatus, closed: bool, - /// Searches started and not yet taken by the worker. + /// 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 the worker finished and the serve loop has not collected. found: Vec<(SearchTicket, Result)>, @@ -517,31 +574,30 @@ fn lock(shared: &Lock) -> MutexGuard<'_, Shared> { /// How long the worker waits, and what ends the wait early. enum Wait { /// Idle: until something arrives (an entry, the mind's identity, a search) - /// or `recheck` passes. - Work { recheck: Duration }, + /// or `until`. The deadline is the caller's, fixed when it was set: work + /// that wakes the worker does not move it. + Work { until: Instant }, /// Backing off from a failure: the whole delay, whatever arrives. A failing /// index is retried on its schedule, not once per admission. Backoff(Duration), } -/// Whether the wait ended because the recheck interval passed. -fn wait_for_work(shared: &Lock, wait: Wait) -> bool { - let (timeout, wakes_on_work) = match wait { - Wait::Work { recheck } => (recheck, true), - Wait::Backoff(delay) => (delay, false), +fn wait_for_work(shared: &Lock, wait: Wait) { + let (deadline, wakes_on_work) = match wait { + Wait::Work { until } => (until, true), + Wait::Backoff(delay) => (Instant::now() + delay, false), }; - let deadline = Instant::now() + timeout; let mut guard = lock(shared); loop { if guard.closed { - return false; + return; } if wakes_on_work && (!guard.inbox.is_empty() || guard.mind.is_some() || !guard.searches.is_empty()) { - return false; + return; } let left = deadline.saturating_duration_since(Instant::now()); if left.is_zero() { - return wakes_on_work; + return; } guard = shared.1.wait_timeout(guard, left).unwrap_or_else(|poisoned| poisoned.into_inner()).0; } @@ -570,16 +626,33 @@ fn including(status: &IndexStatus, queued: u32) -> IndexStatus { } } -/// Runs every search the serve loop has queued, one at a time, and leaves each -/// answer for the loop to collect. Between batches, never inside one. -fn serve_searches(projector: &mut Projector, shared: &Lock) { - loop { - 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:#}")); - lock(shared).found.push((ticket, found)); +/// 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. +fn unsearchable(status: &IndexStatus) -> Option { + use IndexStatus as S; + match status { + S::Current => 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})")), } } +/// Runs the search at the front of the queue, if any, and leaves its answer +/// 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. +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:#}")); + lock(shared).found.push((ticket, found)); +} + /// Answers every queued search with why the index cannot: they were accepted /// while it was healthy and it has failed since. fn refuse_searches(shared: &Lock, why: &str) { @@ -589,63 +662,77 @@ fn refuse_searches(shared: &Lock, why: &str) { } } +/// The worker's take of what arrived since it last looked: the inbox is taken, +/// absorbed and its effect published under one lock. Between taking entries +/// out of the inbox and publishing them as wanted, `status` would otherwise +/// read a stale count with nothing queued, and a reader would see the mind +/// caught up (or a failure short of what it is failing to write) while +/// entries were in the worker's hands. `failing` is the failure being +/// reported and how many attempts it has had: the report stands, with the new +/// pending count, until an attempt succeeds. Returns false once the sink has +/// closed. +fn absorb( + projector: &mut Projector, + shared: &Lock, + failing: Option<(&StepError, u32)>, +) -> bool { + let mut guard = lock(shared); + if guard.closed { + return false; + } + let mind = guard.mind.take(); + let inbox = std::mem::take(&mut guard.inbox); + if let Some(mind) = mind { + projector.identify(mind); + } + projector.want(inbox); + guard.status = match failing { + None => projector.progress_status(), + Some((error, attempts)) => projector.failure_status(error, attempts), + }; + true +} + fn work(mut projector: Projector, shared: Lock, backoff: Backoff) { - let mut failures = 0_u32; + let mut failing: Option<(StepError, u32)> = None; let mut delay = backoff.initial; - let mut recheck = false; + // When the worker next verifies the model and the collection whether or + // not anything has woken it: an absolute time, moved only by a + // verification, so a steady stream of searches cannot keep pushing it out. + let mut recheck_at = Instant::now() + backoff.recheck; loop { - // The inbox is taken, absorbed and its effect published under one - // lock: between taking entries out of the inbox and publishing them - // as wanted, `status` would otherwise read a stale `Current` with - // nothing queued, and a reader would see the mind caught up while - // entries were in the worker's hands. - { - let mut guard = lock(&shared); - if guard.closed { - return; - } - let mind = guard.mind.take(); - let inbox = std::mem::take(&mut guard.inbox); - if let Some(mind) = mind { - projector.identify(mind); - } - projector.want(inbox); - // While a failure is being reported, the report stands until an - // attempt succeeds: publishing progress between retries would - // flicker. - if failures == 0 { - guard.status = projector.progress_status(); - } + if !absorb(&mut projector, &shared, failing.as_ref().map(|(error, attempts)| (error, *attempts))) { + return; } shared.1.notify_all(); - let stepped = if std::mem::take(&mut recheck) { projector.recheck() } else { projector.advance() }; + let due = Instant::now() >= recheck_at; + let verifies = due || projector.stale(); + let stepped = if due { projector.recheck() } else { projector.advance() }; match stepped { - Ok(Advance::Progressed) => { - failures = 0; + Ok(advance) => { + failing = None; delay = backoff.initial; + if verifies { + recheck_at = Instant::now() + backoff.recheck; + } publish(&shared, projector.progress_status()); - serve_searches(&mut projector, &shared); - } - Ok(Advance::Idle) => { - failures = 0; - delay = backoff.initial; - publish(&shared, projector.progress_status()); - serve_searches(&mut projector, &shared); - // A search that found the model changed has left the - // reconciliation to redo: do it now, not at the next wake. - if !projector.stale() { - recheck = wait_for_work(&shared, Wait::Work { recheck: backoff.recheck }); + serve_search(&mut projector, &shared); + // A search that failed 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 }); } } Err(error) => { - failures = failures.saturating_add(1); + let attempts = failing.as_ref().map_or(0, |(_, attempts)| *attempts).saturating_add(1); eprintln!( - "huginn: the index could not advance (attempt {failures}, {} pending, retrying in {:?}): {error}", + "huginn: the index could not advance (attempt {attempts}, {} pending, retrying in {:?}): {error}", projector.pending(), delay ); - publish(&shared, projector.failure_status(&error, failures)); + publish(&shared, projector.failure_status(&error, attempts)); refuse_searches(&shared, &format!("the index could not advance: {error}")); + failing = Some((error, attempts)); wait_for_work(&shared, Wait::Backoff(delay)); delay = backoff.after(delay); } @@ -702,25 +789,30 @@ impl WorkerSink { /// Blocks until the health satisfies `predicate`, and returns it. For /// tests and probes that must observe the worker without a clock; it gives - /// up after two minutes rather than hang a run. + /// up after `WAIT_STATUS_MAX` in all, however often the health changes, + /// rather than hang a run. pub fn wait_status(&self, predicate: impl Fn(&IndexStatus) -> bool) -> IndexStatus { + self.wait_status_for(WAIT_STATUS_MAX, predicate) + } + + fn wait_status_for(&self, limit: Duration, predicate: impl Fn(&IndexStatus) -> bool) -> IndexStatus { + let deadline = Instant::now() + limit; let mut guard = lock(&self.shared); loop { let seen = including(&guard.status, u32::try_from(guard.inbox.len()).unwrap_or(u32::MAX)); if predicate(&seen) { return seen; } - let (next, result) = self - .shared - .1 - .wait_timeout(guard, Duration::from_secs(120)) - .unwrap_or_else(|poisoned| poisoned.into_inner()); - guard = next; - assert!(!result.timed_out(), "the index never reached the awaited state; last seen {:?}", guard.status); + let left = deadline.saturating_duration_since(Instant::now()); + assert!(!left.is_zero(), "the index never reached the awaited state; last seen {:?}", guard.status); + guard = self.shared.1.wait_timeout(guard, left).unwrap_or_else(|poisoned| poisoned.into_inner()).0; } } } +/// The longest `WorkerSink::wait_status` waits, in all. +const WAIT_STATUS_MAX: Duration = Duration::from_secs(30); + impl Drop for WorkerSink { fn drop(&mut self) { lock(&self.shared).closed = true; @@ -751,14 +843,20 @@ impl IndexSink for WorkerSink { including(&guard.status, u32::try_from(guard.inbox.len()).unwrap_or(u32::MAX)) } - /// Queues the search for the worker and returns its ticket. A worker that - /// is failing or refused is not asked: its own report is the refusal. + /// 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. fn search(&mut self, text: &str, top_k: u32) -> Result { let mut guard = lock(&self.shared); - match &guard.status { - IndexStatus::Failing { error, .. } => return Err(format!("the semantic index is failing: {error}")), - IndexStatus::Refused { reason, .. } => return Err(format!("the semantic index is refused: {reason}")), - _ => {} + let status = including(&guard.status, u32::try_from(guard.inbox.len()).unwrap_or(u32::MAX)); + if let Some(why) = unsearchable(&status) { + return Err(why); + } + if guard.searches.len() >= SEARCHES_QUEUED_MAX { + return Err(format!( + "the semantic index has {} searches waiting and takes no more until it has answered some", + guard.searches.len() + )); } let ticket = SearchTicket(guard.next_ticket); guard.next_ticket += 1; @@ -771,6 +869,13 @@ impl IndexSink for WorkerSink { fn searched(&mut self) -> Vec<(SearchTicket, Result)> { std::mem::take(&mut lock(&self.shared).found) } + + /// A search still queued is taken out of the queue and never embedded. One + /// 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); + } } /// Plain HTTP to one base address, in text: the adapters own what the text diff --git a/crates/huginn-daemon/src/index/tests.rs b/crates/huginn-daemon/src/index/tests.rs index e0e4ecf..6cce84f 100644 --- a/crates/huginn-daemon/src/index/tests.rs +++ b/crates/huginn-daemon/src/index/tests.rs @@ -12,7 +12,7 @@ use huginn_mind::wire::{HuginnMindRequest, HuginnMindResponse, IndexStatus}; use huginn_mind::{Mind, OwnedRedbMessagePackBackingStore, PipelineAdmissionOutcome}; use tempfile::TempDir; -use super::fakes::{FakeEmbedder, FakeIndex, Gate, pair}; +use super::fakes::{FakeEmbedder, FakeIndex, Gate, Stored, pair}; use super::ollama::OllamaEmbedder; use super::qdrant::QdrantIndex; use super::*; @@ -615,9 +615,7 @@ fn dropping_the_sink_stops_the_worker() { sink.wait_status(|status| *status == IndexStatus::Current); let shared = Arc::downgrade(&sink.shared); drop(sink); - while shared.upgrade().is_some() { - std::thread::yield_now(); - } + eventually("the worker letting go of the shared state", || shared.upgrade().is_none()); } /// F1. The promised curve, pure: 30 s doubling to a 10 min cap, never below @@ -759,7 +757,7 @@ fn a_mind_without_an_identity_makes_no_collection_until_its_first_admission() { /// Waits for a condition the worker will make true, without a clock in the /// assertion: it polls for at most ten seconds and then says what never /// happened. -fn eventually(what: &str, condition: impl Fn() -> bool) { +fn eventually(what: &str, mut condition: impl FnMut() -> bool) { for _ in 0..2000 { if condition() { return; @@ -864,7 +862,7 @@ 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("not been reconciled")); + assert!(format!("{:#}", fresh.search("anything", 3).unwrap_err()).contains("being rebuilt")); assert!(embedder.texts().is_empty() && index.state.lock().unwrap().searches.is_empty()); let mut projector = projected(&embedder, &index, &entries); @@ -931,41 +929,45 @@ fn shared_with( } /// The idle wait is for work: an entry, the mind's identity, a search or the -/// sink closing ends it at once, and only a quiet worker waits out the recheck -/// interval, after which it says the interval passed. +/// sink closing ends it at once, and only a quiet worker waits out its +/// deadline. The deadline is the caller's: a deadline already past ends the +/// wait at once too, and waking for work does not extend it. #[test] -fn an_idle_wait_ends_for_work_and_otherwise_at_the_recheck() { +fn an_idle_wait_ends_for_work_and_otherwise_at_its_deadline() { let (_root, mind) = mind_with_documents(); let entry = mind.index_entries(None).unwrap().remove(0); let waits = |shared: &Lock| { let started = Instant::now(); - let recheck = wait_for_work(shared, Wait::Work { recheck: Duration::from_secs(30) }); - (recheck, started.elapsed() < Duration::from_secs(10)) + wait_for_work(shared, Wait::Work { until: Instant::now() + Duration::from_secs(30) }); + started.elapsed() < Duration::from_secs(10) }; - assert_eq!(waits(&shared_with(vec![entry], None, Vec::new(), false)), (false, true), "an entry"); - assert_eq!(waits(&shared_with(Vec::new(), Some(MIND.into()), Vec::new(), false)), (false, true), "the mind's identity"); - assert_eq!(waits(&shared_with(Vec::new(), None, vec![(SearchTicket(0), "q".into(), 3)], false)), (false, true), "a search"); - assert_eq!(waits(&shared_with(Vec::new(), None, Vec::new(), true)), (false, true), "the sink closing"); + 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::new(), true)), "the sink closing"); let quiet = shared_with(Vec::new(), None, Vec::new(), false); let started = Instant::now(); - assert!(wait_for_work(&quiet, Wait::Work { recheck: Duration::from_millis(60) }), "a quiet worker is told the interval passed"); - assert!(started.elapsed() >= Duration::from_millis(60), "and only after it did"); + wait_for_work(&quiet, Wait::Work { until: started + Duration::from_millis(60) }); + assert!(started.elapsed() >= Duration::from_millis(60), "a quiet worker waits for its deadline"); + let started = Instant::now(); + wait_for_work(&quiet, Wait::Work { until: started.checked_sub(Duration::from_secs(1)).unwrap_or(started) }); + assert!(started.elapsed() < Duration::from_secs(10), "a deadline already past is not waited for"); } -/// A backoff is the whole delay, whatever arrives, and never reports a -/// recheck: a failing index is retried on its schedule. +/// A backoff is the whole delay, whatever arrives: a failing index is retried +/// on its schedule. #[test] 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 started = Instant::now(); - assert!(!wait_for_work(&busy, Wait::Backoff(Duration::from_millis(80)))); + 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"); let closed = shared_with(Vec::new(), None, Vec::new(), true); let started = Instant::now(); - assert!(!wait_for_work(&closed, Wait::Backoff(Duration::from_secs(30)))); + wait_for_work(&closed, Wait::Backoff(Duration::from_secs(30))); assert!(started.elapsed() < Duration::from_secs(10), "a closed sink ends even a backoff"); } @@ -986,3 +988,400 @@ fn queued_searches_are_answered_with_why_when_the_worker_gives_up() { ] ); } + +// The semantic read's index side: when a search is answered, how many wait, +// and what the worker does between them. + +type Answers = Vec<(SearchTicket, Result)>; + +/// The sink's own calls, named: the trait is generic over the store and the +/// worker does not care which. +fn ask_sink(sink: &mut WorkerSink, text: &str, top_k: u32) -> Result { + >::search(sink, text, top_k) +} + +fn collected(sink: &mut WorkerSink) -> Answers { + >::searched(sink) +} + +fn abandon(sink: &mut WorkerSink, ticket: SearchTicket) { + >::abandon(sink, ticket); +} + +/// Waits for `count` answers and returns them, in the order they were given. +fn answers(sink: &mut WorkerSink, count: usize) -> Answers { + let mut all = Vec::new(); + eventually("the searches being answered", || { + all.extend(collected(sink)); + all.len() >= count + }); + all +} + +/// `count` documents that share one document's text and differ in id, for the +/// tests whose subject is how many there are. +fn entries_of(count: usize) -> Vec { + let (_root, mind) = mind_with_documents(); + let template = mind.index_entries(None).unwrap().remove(0); + (0..count) + .map(|n| { + let mut entry = template.clone(); + entry.id.id.0 = format!("{}#{n}", entry.id.id.0); + entry + }) + .collect() +} + +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. +#[test] +fn only_a_current_index_answers_a_search() { + use IndexStatus as S; + assert_eq!(unsearchable(&S::Current), None); + let says = |status: S, wanted: &[&str]| { + let why = unsearchable(&status).unwrap_or_else(|| panic!("{status:?} answered")); + for word in wanted { + assert!(why.contains(word), "{status:?}: {word} in {why}"); + } + }; + 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. + 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(); + 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()); +} + +/// 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. +#[test] +fn a_search_that_fails_forgets_the_reconciliation() { + 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)), + ]; + for (what, breaks) in breaks { + 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}"); + } + + // In the worker the next step is the verification, with nothing to prompt it. + 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); + let listed = index.state.lock().unwrap().listed; + index.state.lock().unwrap().fail_search = true; + let ticket = ask_sink(&mut sink, "anything", 3).unwrap(); + 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); +} + +/// A search asks the store what the collection says about itself: a label for +/// another mind, no label, or no collection is refused before the query is +/// embedded, and the reconciliation is to be redone. +#[test] +fn a_search_re_reads_the_collections_label() { + let (_root, mind) = mind_with_documents(); + let entries = mind.index_entries(None).unwrap(); + let cases: [(&str, Box); 3] = [ + ( + "another mind's label", + Box::new(|stored| { + let Described::Labelled(meta) = &mut stored.label else { panic!() }; + meta.mind = "mind-commit-another".into(); + }), + ), + ("no label", Box::new(|stored| stored.label = Described::Unlabelled)), + ("another text version", Box::new(|stored| { + let Described::Labelled(meta) = &mut stored.label else { panic!() }; + meta.index_text_version += 1; + })), + ]; + for (what, change) in cases { + let (embedder, index) = pair(); + 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()); + 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}"); + assert!(projector.stale(), "{what}"); + } + 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.stale()); +} + +/// The model is read again after the query is embedded: a vector made while the +/// model changed is never compared with the collection. +#[test] +fn a_model_that_changes_while_the_query_is_embedded_is_refused() { + let (_root, mind) = mind_with_documents(); + let entries = mind.index_entries(None).unwrap(); + let gate = Gate::shut(); + let embedder = FakeEmbedder::new("d1").behind(&gate); + let index = FakeIndex::default(); + gate.open(); + let mut projector = projected(&embedder, &index, &entries); + gate.shut_again(); + let searching = std::thread::spawn(move || { + let found = projector.search("anything", 3); + (projector, found) + }); + gate.wait_arrived(1); + embedder.state.lock().unwrap().identity.digest = "d2".into(); + gate.open(); + let (projector, found) = searching.join().unwrap(); + let refusal = format!("{:#}", found.unwrap_err()); + assert!(refusal.contains("while the query was embedded"), "{refusal}"); + assert!(index.state.lock().unwrap().searches.is_empty(), "the vector was not used"); + assert!(projector.stale()); +} + +/// Each search has a ticket of its own, and its answer comes back once, as the +/// worker gave it: nothing is added to it and nothing is left to be +/// collected twice. +#[test] +fn every_search_has_its_own_ticket_and_its_answer_comes_back_exactly_once() { + let (_root, mind) = mind_with_documents(); + let entries = mind.index_entries(None).unwrap(); + let (embedder, index) = pair(); + index.state.lock().unwrap().hits = + vec![Hit { doc_id: "eureka-state:question:Q1".into(), kind: "question".into(), score: 0.9 }]; + let mut sink = WorkerSink::spawn(embedder, index, &slug(INSTANCE), Some(MIND.into()), entries, quick()); + sink.wait_status(|status| *status == IndexStatus::Current); + let expected: Hits = vec![( + PipelineRef { kind: PipelineKind::Question, id: Short("eureka-state:question:Q1".into()) }, + 0.9, + )]; + let mut tickets = Vec::new(); + for text in ["a", "b", "c"] { + let ticket = ask_sink(&mut sink, text, 3).unwrap(); + assert_eq!(answers(&mut sink, 1), vec![(ticket, Ok(expected.clone()))]); + assert!(collected(&mut sink).is_empty(), "an answer is handed over once"); + tickets.push(ticket); + } + assert_eq!(tickets, [SearchTicket(0), SearchTicket(1), SearchTicket(2)]); +} + +/// The queue of searches is bounded and the asker who does not fit is told. +/// One search is inside the embedder and the queue is full; the next is +/// refused by name, and when the worker has answered some it takes more. +#[test] +fn the_search_queue_is_bounded_and_the_asker_is_told_when_it_is_full() { + 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()), Vec::new(), quick()); + sink.wait_status(|status| *status == IndexStatus::Current); + ask_sink(&mut sink, "in flight", 3).unwrap(); + gate.wait_arrived(1); + for n in 0..SEARCHES_QUEUED_MAX { + ask_sink(&mut sink, &format!("waiting {n}"), 3).unwrap_or_else(|why| panic!("{n} of {SEARCHES_QUEUED_MAX}: {why}")); + } + let refusal = ask_sink(&mut sink, "one too many", 3).unwrap_err(); + assert!(refusal.contains(&format!("{SEARCHES_QUEUED_MAX} searches waiting")), "{refusal}"); + + gate.open(); + let answered = answers(&mut sink, SEARCHES_QUEUED_MAX + 1); + assert_eq!(answered.len(), SEARCHES_QUEUED_MAX + 1); + assert!(ask_sink(&mut sink, "room again", 3).is_ok()); +} + +/// A search the asker has stopped waiting for is taken out of the queue and +/// never embedded; the ones on either side of it are. +#[test] +fn an_abandoned_search_is_never_embedded() { + let gate = Gate::shut(); + let (_, index) = pair(); + let embedder = FakeEmbedder::new("d1").behind(&gate); + let mut sink = WorkerSink::spawn(embedder.clone(), index, &slug(INSTANCE), Some(MIND.into()), Vec::new(), quick()); + sink.wait_status(|status| *status == IndexStatus::Current); + let first = ask_sink(&mut sink, "first", 3).unwrap(); + gate.wait_arrived(1); + let second = ask_sink(&mut sink, "second", 3).unwrap(); + let third = ask_sink(&mut sink, "third", 3).unwrap(); + abandon(&mut sink, second); + gate.open(); + let answered = answers(&mut sink, 2); + let tickets: Vec = answered.iter().map(|(ticket, _)| *ticket).collect(); + assert_eq!(tickets, [first, third]); + let texts = queries(&embedder); + assert_eq!(texts.len(), 2, "{texts:?}"); + assert!(texts[0].ends_with("Query: first") && texts[1].ends_with("Query: third"), "{texts:?}"); +} + +/// The worker serves one queued search and returns to its writes: three +/// waiting searches leave two after one turn. +#[test] +fn one_queued_search_is_served_per_turn() { + let (_root, mind) = mind_with_documents(); + 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); + 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.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. +#[test] +fn a_waiting_search_is_answered_between_batches() { + 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()); + gate.wait_arrived(1); + lock(&sink.shared).searches.push_back((SearchTicket(9), "anything".into(), 3)); + 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}"); + sink.wait_status(|status| *status == IndexStatus::Current); +} + +/// A search that was queued when the worker failed is answered with why, by +/// the worker's own failure path. +#[test] +fn a_search_queued_before_the_worker_fails_is_answered_with_why() { + let gate = Gate::shut(); + let (_, index) = pair(); + 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)); + 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}"); +} + +/// While a failure is being reported, the report's pending count is the one +/// the worker holds now, published under the lock that took the entries out of +/// the inbox: never the count from before they moved. +#[test] +fn a_failure_report_carries_the_pending_count_it_has_now() { + let (_root, mind) = mind_with_documents(); + let entries = mind.index_entries(None).unwrap(); + let (embedder, index) = pair(); + let mut projector = Projector::new(embedder, index, &slug(INSTANCE), Some(MIND.into())); + projector.want(entries[..2].iter().cloned()); + let shared = shared_with(entries[2..].to_vec(), None, Vec::new(), false); + let error = StepError::Unavailable(anyhow::anyhow!("down")); + assert!(absorb(&mut projector, &shared, Some((&error, 3)))); + { + let guard = lock(&shared); + assert!(guard.inbox.is_empty()); + assert_eq!(guard.status, IndexStatus::Failing { pending: 4, attempts: 3, error: "down".into() }); + } + assert!(absorb(&mut projector, &shared, None)); + assert_eq!(lock(&shared).status, IndexStatus::Reconciling { pending: 4 }); + lock(&shared).closed = true; + assert!(!absorb(&mut projector, &shared, None), "a closed sink ends the worker"); +} + +/// The worker's recheck is on its own clock: searches arriving faster than the +/// interval do not keep pushing it out. +#[test] +fn steady_searches_do_not_postpone_the_workers_recheck() { + let (_root, mind) = mind_with_documents(); + let entries = mind.index_entries(None).unwrap(); + let (embedder, index) = pair(); + let backoff = Backoff { recheck: Duration::from_millis(200), ..quick() }; + let mut sink = WorkerSink::spawn(embedder, index.clone(), &slug(INSTANCE), Some(MIND.into()), entries, backoff); + sink.wait_status(|status| *status == IndexStatus::Current); + let listed = index.state.lock().unwrap().listed; + let started = Instant::now(); + while index.state.lock().unwrap().listed < listed + 2 { + assert!(started.elapsed() < Duration::from_secs(10), "the worker never rechecked while searches kept arriving"); + let _ = ask_sink(&mut sink, "anything", 3); + collected(&mut sink); + std::thread::sleep(Duration::from_millis(10)); + } +} + +/// `wait_status` gives up at one deadline, not one per wake: a health that +/// keeps changing does not keep it waiting. +#[test] +fn wait_status_gives_up_at_its_overall_deadline_however_often_the_health_changes() { + let (embedder, index) = pair(); + let sink = WorkerSink::spawn(embedder, index, &slug(INSTANCE), Some(MIND.into()), Vec::new(), quick()); + sink.wait_status(|status| *status == IndexStatus::Current); + let shared = Arc::clone(&sink.shared); + let stop = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let ticking = Arc::clone(&stop); + let ticker = std::thread::spawn(move || { + while !ticking.load(std::sync::atomic::Ordering::Relaxed) { + shared.1.notify_all(); + std::thread::sleep(Duration::from_millis(5)); + } + }); + let (tell, told) = std::sync::mpsc::channel(); + std::thread::spawn(move || { + let gave_up = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + sink.wait_status_for(Duration::from_millis(300), |status| *status == IndexStatus::Reconciling { pending: 99 }) + })) + .is_err(); + let _ = tell.send(gave_up); + }); + let gave_up = told.recv_timeout(Duration::from_secs(10)); + stop.store(true, std::sync::atomic::Ordering::Relaxed); + ticker.join().unwrap(); + assert_eq!(gave_up, Ok(true), "the wait did not end at its deadline"); +} diff --git a/crates/huginn-daemon/src/semantic_tests.rs b/crates/huginn-daemon/src/semantic_tests.rs index ad8bb19..406f066 100644 --- a/crates/huginn-daemon/src/semantic_tests.rs +++ b/crates/huginn-daemon/src/semantic_tests.rs @@ -297,10 +297,10 @@ fn a_semantic_query_returns_only_documents_the_mind_holds_and_the_selection_admi assert!(searches.iter().all(|(name, _, limit)| *name == collection_name(&slug(INSTANCE)) && *limit == 20), "{searches:?}"); } -/// A cursor cannot resume a ranked answer: it is refused by name, and nothing -/// is embedded or searched for it. +/// A cursor cannot resume a ranked answer and a ranked answer has no order to +/// reverse: each is refused by name, and nothing is embedded or searched for it. #[test] -fn a_cursor_with_a_semantic_query_is_refused_and_asks_the_index_nothing() { +fn a_cursor_or_a_reversal_with_a_semantic_query_is_refused_and_asks_the_index_nothing() { let harness = healthy(); let mut client = harness.client(); until_index(&mut client, "Current", |status| *status == IndexStatus::Current); @@ -314,6 +314,15 @@ fn a_cursor_with_a_semantic_query_is_refused_and_asks_the_index_nothing() { ); assert_eq!(harness.embedder.texts().len(), embedded); assert!(harness.index.state.lock().unwrap().searches.is_empty()); + + let reversed = Selection { descending: true, ..standing() }; + let refusal = ask(&mut client, "m-2", &semantic(reversed, 5)); + assert!( + matches!(&refusal, HuginnMindResponse::Refused(MindRefusal::SelectionInvalid { field, .. }) if field == "descending"), + "{refusal:?}" + ); + assert_eq!(harness.embedder.texts().len(), embedded); + assert!(harness.index.state.lock().unwrap().searches.is_empty()); } /// The loop never waits on the network. The embedder is shut, the worker is @@ -412,3 +421,59 @@ fn an_index_that_cannot_answer_is_unavailable_and_plain_queries_still_work() { assert!(detail.contains("refused") && detail.contains("no Huginn metadata"), "{detail}"); assert_eq!(documents(&mut client, "m-plain-3"), held); } + +/// A search the client has stopped waiting for is not started. The embedder is +/// held with the first search inside it; a second search queues behind it and +/// times out, and when the embedder is released the worker embeds the first, +/// skips the second and embeds a third: the index is never asked for an answer +/// nobody is waiting for. +#[test] +fn a_search_that_timed_out_while_queued_is_never_embedded() { + let gate = Gate::shut(); + gate.open(); + let embedder = FakeEmbedder::new("d1").behind(&gate); + let harness = Harness::start(shelf(), embedder, FakeIndex::default(), Duration::from_millis(300)); + let mut first = harness.client(); + let mut second = harness.client(); + let mut third = harness.client(); + until_index(&mut third, "Current", |status| *status == IndexStatus::Current); + let queries = |harness: &Harness| harness.embedder.texts().iter().filter(|text| text.starts_with("Instruct: ")).count(); + + gate.shut_again(); + send(&mut first, "m-first", &semantic(standing(), 5)); + gate.wait_arrived(1); + send(&mut second, "m-second", &semantic(standing(), 5)); + let timed_out = receive(&mut second, "m-second", 3000).expect("the second search is answered at its deadline"); + assert!(matches!(&timed_out, HuginnMindResponse::Refused(MindRefusal::Unavailable { detail }) if detail.contains("did not finish")), "{timed_out:?}"); + + gate.open(); + send(&mut third, "m-third", &semantic(standing(), 5)); + let answered = receive(&mut third, "m-third", 3000).expect("the third search is answered"); + assert!(matches!(answered, HuginnMindResponse::Query(_)), "{answered:?}"); + assert_eq!(queries(&harness), 2, "the first and the third were embedded; the second was taken back before it started"); +} + +/// 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. +#[test] +fn a_search_while_the_index_is_being_rebuilt_is_refused_and_asks_the_embedder_nothing() { + let gate = Gate::shut(); + let embedder = FakeEmbedder::new("d1").behind(&gate); + let harness = Harness::start(shelf(), embedder, FakeIndex::default(), Duration::from_secs(30)); + gate.wait_arrived(1); + 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); + + 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"); + + gate.open(); + 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")]); +} diff --git a/crates/huginn-daemon/src/serve.rs b/crates/huginn-daemon/src/serve.rs index 593ae63..628abb9 100644 --- a/crates/huginn-daemon/src/serve.rs +++ b/crates/huginn-daemon/src/serve.rs @@ -519,7 +519,8 @@ pub fn run>( } /// What the index has finished, then what has waited too long: the latter as -/// an `Unavailable` detail. A search that finishes after its deadline finds +/// an `Unavailable` detail, and abandoned, so the index does not start a search +/// nobody is waiting for. A search that finishes after its deadline finds /// nothing waiting for it and is dropped. fn collect_searches>( daemon: &mut Daemon, @@ -530,6 +531,7 @@ fn collect_searches>( let now = Instant::now(); for (ticket, asked) in waiting { if asked.until <= now { + daemon.abandon(*ticket); collected.push((*ticket, Err(format!("the semantic search did not finish within {:?}", options.search_timeout)))); } }