GameCult/Huginn
Watch
9
0
Fork
You've already forked Huginn
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Semantic search refuses while the index rebuilds; bound, cancel and re-verify

Browse source 
Operator ruling 2026-09-30: while the index is rebuilding (a re-pulled model,
a lost collection, Reconciling or Behind) a semantic search refuses with a
typed Unavailable naming the state and the pending count, at the sink and
again in Projector::search, and never answers from a half-filled collection.

Soul's findings on 11b:
- a failed search clears the reconciliation, so whoami stops reporting Current
  and the worker verifies again; the idle recheck is an absolute time that
  searches no longer push out;
- the queue holds at most SEARCHES_QUEUED_MAX and refuses the rest by name;
  the worker serves one search per turn and returns to its writes; the serve
  loop takes back a search it timed out (IndexSink::abandon), so a queued one
  is never embedded;
- the model identity is read again after the query is embedded, and the
  collection's label is re-read before it: a search is one decision;
- a failure report's pending count is refreshed under the lock that takes the
  inbox (absorb);
- wait_status has one overall deadline, and the test gate and worker-stop
  waits are bounded, so a mutant fails instead of hanging.

Tests pin each rule: ticket uniqueness, searched() returning exactly what the
worker answered, the refuse_searches call site, serving between batches, the
bound, abandonment, the rechecks and the label.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-29 16:25:55 +02:00
commit 4b5eaf9b8f
6 changed files with 718 additions and 123 deletions
Download patch file Download diff file Expand all files Collapse all files

15
crates/huginn-daemon/src/daemon.rs
View file

@ -28,6 +28,10 @@ pub trait IndexSink<S: MindStore> {
/// 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<Hits, String>)>;
/// 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<S: MindStore, I: IndexSink<S>> Daemon<S, I> {
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<Hits, String>)> {
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<Hits, String>)> {
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<Hits, String>)> {
Vec::new()
}
fn abandon(&mut self, _ticket: SearchTicket) {}
}
fn status(daemon: &mut Daemon<OwnedRedbMessagePackBackingStore, impl IndexSink<OwnedRedbMessagePackBackingStore>>) -> MindStatus {

13
crates/huginn-daemon/src/index/fakes.rs
View file

@ -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<Hit>,
/// 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<BTreeSet<String>> {
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())
}

297
crates/huginn-daemon/src/index/mod.rs
View file

@ -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<E: Embedder, V: VectorIndex> Projector<E, V> {
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<E: Embedder, V: VectorIndex> Projector<E, V> {
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<E: Embedder, V: VectorIndex> Projector<E, V> {
/// 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<Hits> {
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<Hits> {
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<IndexEntry>,
/// The mind's identity, once a commit has revealed it to a mind that had none.
mind: Option<String>,
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<Hits, String>)>,
@ -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<E: Embedder, V: VectorIndex>(projector: &mut Projector<E, V>, 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<String> {
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<E: Embedder, V: VectorIndex>(projector: &mut Projector<E, V>, 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<E: Embedder, V: VectorIndex>(
projector: &mut Projector<E, V>,
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<E: Embedder, V: VectorIndex>(mut projector: Projector<E, V>, 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<S: MindStore> IndexSink<S> 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<SearchTicket, String> {
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<S: MindStore> IndexSink<S> for WorkerSink {
fn searched(&mut self) -> Vec<(SearchTicket, Result<Hits, String>)> {
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

441
crates/huginn-daemon/src/index/tests.rs
View file

@ -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<Hits, String>)>;
/// 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<SearchTicket, String> {
<WorkerSink as IndexSink<OwnedRedbMessagePackBackingStore>>::search(sink, text, top_k)
}
fn collected(sink: &mut WorkerSink) -> Answers {
<WorkerSink as IndexSink<OwnedRedbMessagePackBackingStore>>::searched(sink)
}
fn abandon(sink: &mut WorkerSink, ticket: SearchTicket) {
<WorkerSink as IndexSink<OwnedRedbMessagePackBackingStore>>::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<IndexEntry> {
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<String> {
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<dyn Fn(&FakeEmbedder, &FakeIndex)>); 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<dyn Fn(&mut Stored)>); 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<SearchTicket> = 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::<Vec<_>>(), [SearchTicket(2), SearchTicket(3)]);
assert_eq!(guard.found.iter().map(|(ticket, _)| *ticket).collect::<Vec<_>>(), [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");
}

71
crates/huginn-daemon/src/semantic_tests.rs
View file

@ -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")]);
}

4
crates/huginn-daemon/src/serve.rs
View file

@ -519,7 +519,8 @@ pub fn run<S: MindStore, I: IndexSink<S>>(
}
/// 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<S: MindStore, I: IndexSink<S>>(
daemon: &mut Daemon<S, I>,
@ -530,6 +531,7 @@ fn collect_searches<S: MindStore, I: IndexSink<S>>(
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))));
}
}