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

Bound the whole search path by the search's own time; lag is not a rebuild

Browse source 
Soul pass 2 follow-ups F1, F2, F4, F5 on the semantic index.

F1: every call a search makes (identity reads, label read, embed, query) is
bounded by what remains of one Budget, and the queued ticket carries the moment
SEARCH_DEADLINE after it was asked. The ports take the bound per call
(model_identity, describe, search); QUERY_EMBED_TIMEOUT is gone, since the
ticket's deadline is the one number. A hung identity read frees the worker
within the search's time.

F2: one refusal predicate. Behind is lag and no longer refuses; only
Reconciling, Failing and Refused do, for the sink and the projector alike. The
worker absorbs the inbox before taking a search and leaves the search queued
while the writes ahead of it are unfinished, so a client that admits and then
searches reads its own writes. An index being rebuilt is still refused.

F4: the model-changed refusal and the worker's failure answer to queued
searches name the state and the pending count.

F5: a search that fails for its own reasons no longer forgets the
reconciliation; only a changed model or label does.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-29 19:42:03 +02:00
commit 4a4d8b029e
7 changed files with 436 additions and 216 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -66,6 +66,11 @@ pub(crate) struct EmbedderState {
/// The length of the vectors it answers with, where that is not the
/// identity's.
pub(crate) vector_length: Option<usize>,
/// The bound each identity read was given, in order.
pub(crate) identity_bounds: Vec<Duration>,
/// Whether an identity read is never answered: it fails when its bound
/// has passed, as an adapter that honours its bound does.
pub(crate) hang_identity: bool,
}
#[derive(Clone)]
@ -76,7 +81,7 @@ pub(crate) struct FakeEmbedder {
impl FakeEmbedder {
pub(crate) fn new(digest: &str) -> Self {
let state = EmbedderState { identity: identity(digest), down: false, embedded: Vec::new(), bounds: Vec::new(), vector_length: None };
let state = EmbedderState { identity: identity(digest), down: false, embedded: Vec::new(), bounds: Vec::new(), vector_length: None, identity_bounds: Vec::new(), hang_identity: false };
Self { state: Arc::new(Mutex::new(state)), gate: None }
}
@ -91,8 +96,14 @@ impl FakeEmbedder {
}
impl Embedder for FakeEmbedder {
fn model_identity(&mut self) -> Result<ModelIdentity> {
let state = self.state.lock().unwrap();
fn model_identity(&mut self, within: Duration) -> Result<ModelIdentity> {
let mut state = self.state.lock().unwrap();
state.identity_bounds.push(within);
if state.hang_identity {
drop(state);
std::thread::sleep(within);
bail!("the embedder did not answer within {within:?}");
}
if state.down {
bail!("the embedder is down");
}
@ -135,6 +146,10 @@ pub(crate) struct IndexState {
pub(crate) listed: u32,
/// Whether a search fails at the vector store, the rest of it working.
pub(crate) fail_search: bool,
/// The bound each describe and each search was given, in order, by call.
pub(crate) bounds: Vec<(&'static str, Duration)>,
/// How many points the collection held when each search reached it.
pub(crate) held_when_searched: Vec<usize>,
}
#[derive(Clone, Default)]
@ -155,8 +170,9 @@ impl FakeIndex {
}
impl VectorIndex for FakeIndex {
fn describe(&mut self, collection: &str) -> Result<Described> {
let state = self.state.lock().unwrap();
fn describe(&mut self, collection: &str, within: Duration) -> Result<Described> {
let mut state = self.state.lock().unwrap();
state.bounds.push(("describe", within));
if state.down {
bail!("the vector store is down");
}
@ -197,14 +213,17 @@ impl VectorIndex for FakeIndex {
Ok(())
}
fn search(&mut self, collection: &str, vector: &[f32], limit: u32) -> Result<Vec<Hit>> {
fn search(&mut self, collection: &str, vector: &[f32], limit: u32, within: Duration) -> Result<Vec<Hit>> {
let mut state = self.state.lock().unwrap();
state.bounds.push(("search", within));
if state.down {
bail!("the vector store is down");
}
state.searches.push((collection.into(), vector.len(), limit));
ensure!(!state.fail_search, "the vector store failed the search");
ensure!(state.collections.contains_key(collection), "no collection {collection}");
let held = state.collections[collection].points.len();
state.held_when_searched.push(held);
Ok(state.hits.clone())
}
}

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

@ -65,15 +65,24 @@
//! (`ServeOptions::search_timeout` bounds the wait), and takes back a search
//! it has stopped waiting for. At most `SEARCHES_QUEUED_MAX` wait; one more is
//! refused. A search whose embed is already running cannot be recalled, so it
//! can hold the writes off for as long as its bound, `QUERY_EMBED_TIMEOUT`,
//! which is `SEARCH_DEADLINE`, the deadline the serve loop gives the search.
//! - Only a current index answers a search. While the worker is `Reconciling`
//! or `Behind` (a re-pulled model, a lost collection, entries not yet
//! written), `Failing` or `Refused`, a search is refused at once with an
//! `Unavailable` that names the state and how many documents are pending; it
//! is never answered from a half-filled collection. A search that fails after
//! it was accepted forgets the reconciliation, so `whoami` stops reporting
//! `Current` and the worker verifies the model and the collection again.
//! can hold the writes off for as long as it has left: the ticket carries the
//! moment `SEARCH_DEADLINE` after it was asked, and every call the search
//! makes (the model's identity, the collection's label, the embed, the
//! query) is bounded by what remains of it.
//! - A search reads its own writes. Admissions and searches meet under one
//! lock, admissions first: a search waits in the queue while the worker
//! absorbs and writes what was admitted before it, and is answered once the
//! index has caught up. Lag is not a rebuild and refuses nothing. Under a
//! steady stream of admissions a search may never see the index caught up;
//! its deadline then answers it.
//! - A search is refused only while the index is being rebuilt or cannot be
//! asked: `Reconciling` (a re-pulled model, a lost collection, a mind's first
//! admission), `Failing` or `Refused`. The refusal is an `Unavailable` that
//! names the state and how many documents are pending; it is never answered
//! from a collection being rebuilt. A search that fails for its own reasons
//! (a timeout, a store error) fails that search alone and the
//! reconciliation stands. Only what the check-then-embed finds changed (the
//! model, the collection's label) forgets it.
pub mod ollama;
pub mod qdrant;
@ -101,21 +110,21 @@ pub const MANAGED_BY: &str = "huginn";
pub const BATCH: usize = 32;
/// How long a search may take, queue wait included: the serve loop's default
/// `ServeOptions::search_timeout`. It is the one number the query embed's
/// bound derives from, so the two cannot drift apart.
/// `ServeOptions::search_timeout`. Every bound on the search path derives from
/// the moment this long after the search was asked, so the two cannot drift
/// apart.
pub const SEARCH_DEADLINE: Duration = Duration::from_secs(30);
/// The bound on one query embed. A running embed cannot be recalled, so it
/// holds the index's writes off for as long as it lasts; it is held to the
/// search's own deadline, after which nobody is waiting for its vector.
pub const QUERY_EMBED_TIMEOUT: Duration = SEARCH_DEADLINE;
/// The bound on a bulk index-flush embed, and on the embedder's identity
/// reads. A cold model takes seconds to load. It does not block serving: the
/// serve loop never waits on the worker, and a search queued behind a flush is
/// answered by its own deadline.
/// The bound on a bulk index-flush embed, the identity and label reads around
/// it, and the reconciliation's identity read. A cold model takes seconds to
/// load. It does not block serving: the serve loop never waits on the worker,
/// and a search queued behind a flush is answered by its own deadline.
pub const BULK_EMBED_TIMEOUT: Duration = Duration::from_secs(300);
/// The bound on a call to the vector store that is not a search: the
/// adapter's default.
pub const INDEX_CALL_TIMEOUT: Duration = Duration::from_secs(60);
/// The most candidates one search asks the index for.
pub const OVERSAMPLE_MAX: u32 = 200;
@ -179,13 +188,16 @@ pub struct Hit {
}
pub trait Embedder {
fn model_identity(&mut self) -> Result<ModelIdentity>;
/// The model's identity, or an error once `within` has passed.
fn model_identity(&mut self, within: Duration) -> Result<ModelIdentity>;
/// One vector per text, in order, or an error once `within` has passed.
fn embed(&mut self, texts: &[String], within: Duration) -> Result<Vec<Vec<f32>>>;
}
pub trait VectorIndex {
fn describe(&mut self, collection: &str) -> Result<Described>;
/// What the collection says about itself, or an error once `within` has
/// passed.
fn describe(&mut self, collection: &str, within: Duration) -> Result<Described>;
/// Drops the collection if it exists and creates it empty with this
/// metadata.
fn recreate(&mut self, collection: &str, meta: &CollectionMeta) -> Result<()>;
@ -193,8 +205,9 @@ pub trait VectorIndex {
fn ids(&mut self, collection: &str) -> Result<BTreeSet<String>>;
fn upsert(&mut self, collection: &str, points: &[Point]) -> Result<()>;
/// The `limit` nearest points to `vector`, best first, with no filter: which
/// documents may be shown is the mind's to decide.
fn search(&mut self, collection: &str, vector: &[f32], limit: u32) -> Result<Vec<Hit>>;
/// documents may be shown is the mind's to decide. An error once `within`
/// has passed.
fn search(&mut self, collection: &str, vector: &[f32], limit: u32, within: Duration) -> Result<Vec<Hit>>;
}
/// The mind's collection: one per instance.
@ -245,6 +258,25 @@ pub enum Advance {
Idle,
}
/// The time one operation has left. Every call the operation makes is bounded
/// by what remains, not by a bound of its own, so their sum cannot outlast it.
struct Budget {
end: Instant,
total: Duration,
}
impl Budget {
fn new(total: Duration) -> Self {
Self { end: Instant::now() + total, total }
}
fn left(&self) -> Result<Duration> {
let left = self.end.saturating_duration_since(Instant::now());
ensure!(!left.is_zero(), "the time allowed ({:?}) has run out", self.total);
Ok(left)
}
}
/// The projection's state: the entries not yet known to be in the collection,
/// and, once reconciled, the length of vector it accepts.
pub struct Projector<E: Embedder, V: VectorIndex> {
@ -344,8 +376,8 @@ impl<E: Embedder, V: VectorIndex> Projector<E, V> {
/// collection was reconciled under. A different model, digest or length
/// forgets the reconciliation, so the next step reconciles (and rebuilds)
/// instead of writing or searching with vectors the collection cannot mix.
fn model_unchanged(&mut self) -> Result<bool, StepError> {
let now = self.embedder.model_identity().context("re-reading the embedding model's identity")?;
fn model_unchanged(&mut self, budget: &Budget) -> Result<bool> {
let now = self.embedder.model_identity(budget.left()?).context("re-reading the embedding model's identity")?;
if self.reconciled.as_ref() == Some(&now) {
return Ok(true);
}
@ -362,9 +394,10 @@ impl<E: Embedder, V: VectorIndex> Projector<E, V> {
let Some(mind) = self.mind.clone() else {
return Err(StepError::Unavailable(anyhow::anyhow!("the mind has no identity yet")));
};
let identity = self.embedder.model_identity().context("reading the embedding model's identity")?;
let identity =
self.embedder.model_identity(BULK_EMBED_TIMEOUT).context("reading the embedding model's identity")?;
let wanted = self.collection_meta(&mind, &identity);
match self.index.describe(&self.collection).context("describing the collection")? {
match self.index.describe(&self.collection, INDEX_CALL_TIMEOUT).context("describing the collection")? {
Described::Absent => self.recreate(&wanted)?,
Described::Unlabelled => {
return Err(StepError::Refused(format!(
@ -411,10 +444,10 @@ impl<E: Embedder, V: VectorIndex> Projector<E, V> {
/// the first `BATCH` wanted entries are embedded, written, and only then
/// no longer wanted.
pub fn flush_batch(&mut self) -> Result<(), StepError> {
self.flush().map_err(StepError::Unavailable)
self.flush(&Budget::new(BULK_EMBED_TIMEOUT)).map_err(StepError::Unavailable)
}
fn flush(&mut self) -> Result<()> {
fn flush(&mut self, budget: &Budget) -> Result<()> {
let dimensions = self.reconciled.as_ref().context("flush before reconcile")?.dimensions as usize;
let batch: Vec<IndexEntry> =
self.wanted.iter().take(BATCH).filter_map(|document_id| self.known.get(document_id)).cloned().collect();
@ -424,7 +457,7 @@ impl<E: Embedder, V: VectorIndex> Projector<E, V> {
let texts: Vec<String> = batch.iter().map(|entry| entry.text.clone()).collect();
// A model that changed leaves nothing written and the reconciliation
// forgotten: the next step reconciles and rebuilds.
let Ok(vectors) = self.embed_verified(&texts, BULK_EMBED_TIMEOUT)? else { return Ok(()) };
let Ok(vectors) = self.embed_verified(&texts, budget)? else { return Ok(()) };
ensure!(vectors.len() == batch.len(), "the embedder returned {} vectors for {} texts", vectors.len(), batch.len());
let mut points = Vec::with_capacity(batch.len());
for (entry, vector) in batch.iter().zip(vectors) {
@ -453,28 +486,29 @@ impl<E: Embedder, V: VectorIndex> Projector<E, V> {
Ok(())
}
/// The one check-then-embed: reads the model's identity and the
/// collection's label, embeds, reads the identity again, and gives the
/// vectors back only if the model is still the one the collection was
/// reconciled under. A model that changed forgets the reconciliation and
/// answers `Err(when)`; a collection that is no longer this mind's, built
/// as reconciled, is an error and forgets it too. Search and flush both
/// embed through here, so neither compares or writes vectors made by a
/// model the collection cannot mix.
fn embed_verified(&mut self, texts: &[String], within: Duration) -> Result<Result<Vec<Vec<f32>>, &'static str>> {
/// The one check-then-embed, every call bounded by what `budget` has left:
/// reads the model's identity and the collection's label, embeds, reads the
/// identity again, and gives the vectors back only if the model is still
/// the one the collection was reconciled under. A model that changed
/// forgets the reconciliation and answers `Err(when)`; a collection that
/// is no longer this mind's, built as reconciled, is an error and forgets
/// it too. A call that fails or times out forgets nothing. Search and
/// flush both embed through here, so neither compares or writes vectors
/// made by a model the collection cannot mix.
fn embed_verified(&mut self, texts: &[String], budget: &Budget) -> Result<Result<Vec<Vec<f32>>, &'static str>> {
let (Some(mind), Some(reconciled)) = (self.mind.clone(), self.reconciled.clone()) else {
bail!("the mind has no identity yet, so it has no index");
};
if !self.model_unchanged().map_err(|error| anyhow::anyhow!("{error}"))? {
if !self.model_unchanged(budget)? {
return Ok(Err("since the collection was built"));
}
let described = self.index.describe(&self.collection).context("describing the collection")?;
let described = self.index.describe(&self.collection, budget.left()?).context("describing the collection")?;
if described != Described::Labelled(self.collection_meta(&mind, &reconciled)) {
self.reconciled = None;
bail!("collection {} is no longer the one this mind's index was built as: {described:?}", self.collection);
}
let vectors = self.embedder.embed(texts, within).context("embedding")?;
if !self.model_unchanged().map_err(|error| anyhow::anyhow!("{error}"))? {
let vectors = self.embedder.embed(texts, budget.left()?).context("embedding")?;
if !self.model_unchanged(budget)? {
return Ok(Err("while the texts were embedded"));
}
Ok(Ok(vectors))
@ -486,41 +520,38 @@ impl<E: Embedder, V: VectorIndex> Projector<E, V> {
/// 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> {
/// A search is refused while the index is being rebuilt or cannot be asked
/// (`unsearchable`), naming the state and how many documents are pending.
/// Catching up on lag is the worker's: it serves a search only once the
/// documents admitted before it are written. The answer is also only from
/// a collection that is still this mind's, built by the model that is
/// still the embedder: the identity is read before and after the query is
/// embedded and the label before it, and the query is refused, with the
/// reconciliation forgotten, if either disagrees with what the collection
/// was reconciled under. Every call is bounded by what remains of
/// `within`. A search that fails for any other reason (a timeout, a store
/// error) fails alone and leaves the reconciliation standing.
pub fn search(&mut self, text: &str, top_k: u32, within: Duration) -> Result<Hits> {
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 budget = Budget::new(within);
let Some(reconciled) = self.reconciled.clone() else {
bail!("the mind has no identity yet, so it has no index");
};
let dimensions = reconciled.dimensions as usize;
let query = format!("Instruct: {QUERY_INSTRUCTION}\nQuery: {text}");
let mut vectors = match self.embed_verified(&[query], QUERY_EMBED_TIMEOUT)? {
let mut vectors = match self.embed_verified(&[query], &budget)? {
Ok(vectors) => vectors,
Err(when) => bail!("the embedding model changed {when}; the index is being rebuilt for it"),
Err(when) => {
bail!("the embedding model changed {when}: {}", unsearchable(&self.progress_status()).unwrap_or_default())
}
};
ensure!(vectors.len() == 1, "the embedder returned {} vectors for one query", vectors.len());
let vector = vectors.remove(0);
ensure!(vector.len() == dimensions, "the embedder returned a query vector of {} where the collection holds {dimensions}", vector.len());
let found = self.index.search(&self.collection, &vector, top_k.saturating_mul(4).min(OVERSAMPLE_MAX))?;
let limit = top_k.saturating_mul(4).min(OVERSAMPLE_MAX);
let found = self.index.search(&self.collection, &vector, limit, budget.left()?)?;
let mut hits = Vec::with_capacity(found.len());
for hit in found {
match PipelineKind::ALL.into_iter().find(|kind| kind.name() == hit.kind) {
@ -579,6 +610,16 @@ impl Backoff {
/// told, not by the worker's patience.
pub const SEARCHES_QUEUED_MAX: usize = 16;
/// A search waiting for the worker: what was asked, and the moment nobody is
/// waiting for the answer any more. Every call the search makes is bounded by
/// what remains until `due`.
struct Queued {
ticket: SearchTicket,
text: String,
top_k: u32,
due: Instant,
}
struct Shared {
inbox: Vec<IndexEntry>,
/// The mind's identity, once a commit has revealed it to a mind that had none.
@ -588,7 +629,7 @@ struct Shared {
/// Searches started and not yet taken by the worker, at most
/// `SEARCHES_QUEUED_MAX`; the serve loop takes back one it has stopped
/// waiting for.
searches: VecDeque<(SearchTicket, String, u32)>,
searches: VecDeque<Queued>,
/// Searches the worker finished and the serve loop has not collected.
found: Vec<(SearchTicket, Result<Hits, String>)>,
next_ticket: u64,
@ -655,18 +696,19 @@ fn including(status: &IndexStatus, queued: u32) -> IndexStatus {
}
}
/// Why a status cannot answer a search, or `None` when it can. Only a current
/// index answers: one that is being reconciled or written to holds a
/// half-filled collection, and one that is failing or refused cannot be asked.
/// The reason names the state and how many documents are pending.
/// Why a status cannot answer a search, or `None` when it can: the one
/// predicate for asking and for answering. An index being rebuilt
/// (`Reconciling`) holds a collection that is not the model's yet, and one
/// that is failing or refused cannot be asked. `Behind` is lag, not a
/// rebuild: the worker writes what was admitted before a search and then
/// answers it. The reason names the state and how many documents are pending.
fn unsearchable(status: &IndexStatus) -> Option<String> {
use IndexStatus as S;
match status {
S::Current => None,
S::Current | S::Behind { .. } => None,
S::Reconciling { pending } => Some(format!(
"the semantic index is being rebuilt: reconciling with the model and the collection (pending: {pending})"
)),
S::Behind { pending } => Some(format!("the semantic index is being rebuilt: behind the mind (pending: {pending})")),
S::Failing { pending, error, .. } => Some(format!("the semantic index is failing: {error} (pending: {pending})")),
S::Refused { pending, reason, .. } => Some(format!("the semantic index is refused: {reason} (pending: {pending})")),
}
@ -676,9 +718,18 @@ fn unsearchable(status: &IndexStatus) -> Option<String> {
/// for the loop to collect. Between batches, never inside one, and only one:
/// the worker returns to its writes after each, so a queue of searches cannot
/// hold the index's writes off for the sum of their embeddings.
///
/// Admissions come first: the inbox is absorbed before the search is taken, so
/// every document committed before the search was asked is wanted, and a search
/// behind lag stays queued until the worker's writes have caught up. An index
/// being rebuilt does not wait: the search is taken and refused. The search's
/// calls are bounded by what remains of the time it was given when it was asked.
fn serve_search<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:#}"));
if !absorb(projector, shared, None) || (projector.pending() > 0 && !projector.stale()) {
return;
}
let Some(Queued { ticket, text, top_k, due }) = lock(shared).searches.pop_front() else { return };
let found = projector.search(&text, top_k, due.saturating_duration_since(Instant::now())).map_err(|error| format!("{error:#}"));
lock(shared).found.push((ticket, found));
}
@ -686,8 +737,8 @@ fn serve_search<E: Embedder, V: VectorIndex>(projector: &mut Projector<E, V>, sh
/// while it was healthy and it has failed since.
fn refuse_searches(shared: &Lock, why: &str) {
let mut guard = lock(shared);
while let Some((ticket, ..)) = guard.searches.pop_front() {
guard.found.push((ticket, Err(why.to_owned())));
while let Some(queued) = guard.searches.pop_front() {
guard.found.push((queued.ticket, Err(why.to_owned())));
}
}
@ -748,8 +799,8 @@ fn work<E: Embedder, V: VectorIndex>(mut projector: Projector<E, V>, shared: Loc
}
publish(&shared, projector.progress_status());
serve_search(&mut projector, &shared);
// A search that failed has left the reconciliation to redo:
// do it now, not at the next wake.
// A model or collection the search found changed has left the
// reconciliation to redo: do it now, not at the next wake.
if advance == Advance::Idle && !projector.stale() {
wait_for_work(&shared, Wait::Work { until: recheck_at });
}
@ -762,7 +813,7 @@ fn work<E: Embedder, V: VectorIndex>(mut projector: Projector<E, V>, shared: Loc
delay
);
publish(&shared, projector.failure_status(&error, attempts));
refuse_searches(&shared, &format!("the index could not advance: {error}"));
refuse_searches(&shared, &unsearchable(&projector.failure_status(&error, attempts)).unwrap_or_default());
failing = Some((error, attempts));
wait_for_work(&shared, Wait::Backoff(delay));
delay = backoff.after(delay);
@ -875,8 +926,10 @@ impl<S: MindStore> IndexSink<S> for WorkerSink {
}
/// Queues the search for the worker and returns its ticket. An index that
/// is not current is not asked, and neither is a full queue: the refusal
/// is the answer, naming why and how many documents are pending.
/// is being rebuilt or cannot be asked is not asked, and neither is a full
/// queue: the refusal is the answer, naming why and how many documents are
/// pending. Documents admitted and not yet written are not a refusal: the
/// search waits behind them.
fn search(&mut self, text: &str, top_k: u32) -> Result<SearchTicket, String> {
let mut guard = lock(&self.shared);
let status = including(&guard.status, u32::try_from(guard.inbox.len()).unwrap_or(u32::MAX));
@ -891,7 +944,7 @@ impl<S: MindStore> IndexSink<S> for WorkerSink {
}
let ticket = SearchTicket(guard.next_ticket);
guard.next_ticket += 1;
guard.searches.push_back((ticket, text.to_owned(), top_k));
guard.searches.push_back(Queued { ticket, text: text.to_owned(), top_k, due: Instant::now() + SEARCH_DEADLINE });
drop(guard);
self.shared.1.notify_all();
Ok(ticket)
@ -905,7 +958,7 @@ impl<S: MindStore> IndexSink<S> for WorkerSink {
/// the worker already holds cannot be recalled; its answer is collected
/// and dropped by the loop, which no longer waits for it.
fn abandon(&mut self, ticket: SearchTicket) {
lock(&self.shared).searches.retain(|(queued, ..)| *queued != ticket);
lock(&self.shared).searches.retain(|queued| queued.ticket != ticket);
}
}
@ -932,6 +985,11 @@ impl Http {
Self { agent, base: base_url.trim_end_matches('/').to_owned() }
}
/// The same address under a bound of its own.
pub(crate) fn bounded(&self, timeout: Duration) -> Self {
Self::new(&self.base, timeout)
}
/// The status and the body, whatever the status is.
pub(crate) fn request(&self, method: Method, path: &str, body: Option<&str>) -> Result<(u16, String)> {
let url = format!("{}{path}", self.base);

16
crates/huginn-daemon/src/index/ollama.rs
View file

@ -6,7 +6,7 @@ use std::time::Duration;
use anyhow::{Context, Result, bail, ensure};
use serde_json::{Value, json};
use super::{BULK_EMBED_TIMEOUT, Embedder, Http, Method, ModelIdentity};
use super::{Embedder, Http, Method, ModelIdentity};
/// The instruction Qwen3 embedding models take in front of a query (never in
/// front of a document). Cut 11b's semantic read prepends it as
@ -14,18 +14,15 @@ use super::{BULK_EMBED_TIMEOUT, Embedder, Http, Method, ModelIdentity};
pub const QUERY_INSTRUCTION: &str =
"Given a question about a project's pipeline state, retrieve the documents that answer it";
/// One model on one Ollama. A cold model takes seconds to load, so the
/// identity reads get `BULK_EMBED_TIMEOUT`; the worker owns the schedule
/// around it. Each embed call carries its own bound.
/// One model on one Ollama. Every call carries its own bound.
pub struct OllamaEmbedder {
http: Http,
base_url: String,
model: String,
}
impl OllamaEmbedder {
pub fn new(base_url: &str, model: &str) -> Self {
Self { http: Http::new(base_url, BULK_EMBED_TIMEOUT), base_url: base_url.to_owned(), model: model.to_owned() }
Self { base_url: base_url.to_owned(), model: model.to_owned() }
}
fn ask(&self, http: &Http, method: Method, path: &str, body: Option<&Value>) -> Result<Value> {
@ -39,8 +36,9 @@ impl OllamaEmbedder {
impl Embedder for OllamaEmbedder {
/// The configured name, the digest Ollama lists for it, and the model's
/// embedding length. The address is not part of the identity.
fn model_identity(&mut self) -> Result<ModelIdentity> {
let tags = self.ask(&self.http, Method::Get, "/api/tags", None)?;
fn model_identity(&mut self, within: Duration) -> Result<ModelIdentity> {
let http = Http::new(&self.base_url, within);
let tags = self.ask(&http, Method::Get, "/api/tags", None)?;
let listed = tags["models"].as_array().context("Ollama's tag list has no models")?;
let digest = listed
.iter()
@ -48,7 +46,7 @@ impl Embedder for OllamaEmbedder {
.and_then(|entry| entry["digest"].as_str())
.with_context(|| format!("Ollama does not list the model {}", self.model))?
.to_owned();
let shown = self.ask(&self.http, Method::Post, "/api/show", Some(&json!({ "model": self.model })))?;
let shown = self.ask(&http, Method::Post, "/api/show", Some(&json!({ "model": self.model })))?;
let info = shown["model_info"].as_object().context("Ollama's model description has no model_info")?;
let Some(length) = info.iter().find(|(key, _)| key.ends_with(".embedding_length")).and_then(|(_, value)| value.as_u64())
else {

21
crates/huginn-daemon/src/index/qdrant.rs
View file

@ -7,7 +7,7 @@ use std::time::Duration;
use anyhow::{Context, Result, ensure};
use serde_json::{Value, json};
use super::{CollectionMeta, Described, Hit, Http, Method, Point, VectorIndex};
use super::{CollectionMeta, INDEX_CALL_TIMEOUT, Described, Hit, Http, Method, Point, VectorIndex};
/// One Qdrant. Every path names a collection this organ was told to own; it
/// never lists or touches another.
@ -20,12 +20,16 @@ const SCROLL_PAGE: u64 = 1000;
impl QdrantIndex {
pub fn new(base_url: &str) -> Self {
Self { http: Http::new(base_url, Duration::from_secs(60)) }
Self { http: Http::new(base_url, INDEX_CALL_TIMEOUT) }
}
fn ask(&self, method: Method, path: &str, body: Option<&Value>, ok: &[u16]) -> Result<(u16, Value)> {
self.ask_with(&self.http, method, path, body, ok)
}
fn ask_with(&self, http: &Http, method: Method, path: &str, body: Option<&Value>, ok: &[u16]) -> Result<(u16, Value)> {
let text = body.map(Value::to_string);
let (status, answer) = self.http.request(method, path, text.as_deref())?;
let (status, answer) = http.request(method, path, text.as_deref())?;
ensure!(ok.contains(&status), "Qdrant answered {status} to {path}: {}", answer.chars().take(300).collect::<String>());
let parsed = if answer.is_empty() { Value::Null } else { serde_json::from_str(&answer).unwrap_or(Value::Null) };
Ok((status, parsed))
@ -67,8 +71,9 @@ fn described(metadata: &Value) -> Described {
}
impl VectorIndex for QdrantIndex {
fn describe(&mut self, collection: &str) -> Result<Described> {
let (status, answer) = self.ask(Method::Get, &format!("/collections/{collection}"), None, &[200, 404])?;
fn describe(&mut self, collection: &str, within: Duration) -> Result<Described> {
let http = self.http.bounded(within);
let (status, answer) = self.ask_with(&http, Method::Get, &format!("/collections/{collection}"), None, &[200, 404])?;
if status == 404 {
return Ok(Described::Absent);
}
@ -133,9 +138,11 @@ impl VectorIndex for QdrantIndex {
Ok(())
}
fn search(&mut self, collection: &str, vector: &[f32], limit: u32) -> Result<Vec<Hit>> {
fn search(&mut self, collection: &str, vector: &[f32], limit: u32, within: Duration) -> Result<Vec<Hit>> {
let http = self.http.bounded(within);
let body = json!({ "vector": vector, "limit": limit, "with_payload": ["doc_id", "kind"], "with_vector": false });
let (_, answer) = self.ask(Method::Post, &format!("/collections/{collection}/points/search"), Some(&body), &[200])?;
let (_, answer) =
self.ask_with(&http, Method::Post, &format!("/collections/{collection}/points/search"), Some(&body), &[200])?;
let found = answer["result"].as_array().context("Qdrant's search has no result")?;
found
.iter()

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

@ -23,6 +23,9 @@ fn collection() -> String {
collection_name(&slug(INSTANCE))
}
/// The bound the adapter tests give a call.
const TIME: Duration = Duration::from_secs(30);
/// A mind identity for tests that project entries without a mind behind them.
const MIND: &str = "mind-commit-first";
@ -308,25 +311,92 @@ fn change_label(index: &FakeIndex, change: impl FnOnce(&mut CollectionMeta)) {
let Described::Labelled(meta) = &mut state.collections.get_mut(&collection()).unwrap().label else { panic!() };
change(meta);
}
/// The query embed is bounded by the search's deadline, one constant behind
/// both, so a query embed that cannot be recalled cannot hold the writes off
/// past the moment nobody is waiting for it. A bulk flush keeps its own bound.
/// Every call on the search path is bounded by what remains of the search's own
/// time: the identity reads, the label read, the embed and the query. A search
/// whose embed cannot be recalled therefore cannot hold the writes off past the
/// moment nobody is waiting for it. A bulk flush keeps its own, longer, bound.
#[test]
fn the_query_embed_is_bounded_by_the_search_deadline_and_the_flush_by_its_own() {
fn every_call_on_the_search_path_is_bounded_by_the_searchs_time_and_the_flush_by_its_own() {
let (_root, mind) = mind_with_documents();
let entries = mind.index_entries(None).unwrap();
let (embedder, index) = pair();
let mut projector = projected(&embedder, &index, &entries);
assert!(!embedder.state.lock().unwrap().bounds.is_empty());
assert!(embedder.state.lock().unwrap().bounds.iter().all(|bound| *bound == BULK_EMBED_TIMEOUT), "the flushes");
projector.search("anything", 3).unwrap();
let bounds = embedder.state.lock().unwrap().bounds.clone();
assert_eq!(bounds.last(), Some(&QUERY_EMBED_TIMEOUT), "the query");
{
let state = embedder.state.lock().unwrap();
assert!(!state.bounds.is_empty());
let near = |bound: &Duration| *bound <= BULK_EMBED_TIMEOUT && *bound > BULK_EMBED_TIMEOUT - Duration::from_secs(60);
assert!(state.bounds.iter().all(near), "the flushes' embeds: {:?}", state.bounds);
assert!(state.identity_bounds.iter().all(near), "the reconciliation's and the flushes' identity reads: {:?}", state.identity_bounds);
}
embedder.state.lock().unwrap().bounds.clear();
embedder.state.lock().unwrap().identity_bounds.clear();
index.state.lock().unwrap().bounds.clear();
let within = Duration::from_secs(20);
projector.search("anything", 3, within).unwrap();
let embedder = embedder.state.lock().unwrap();
let index = index.state.lock().unwrap();
assert_eq!(embedder.identity_bounds.len(), 2, "the model is read before and after the embed");
assert_eq!(embedder.bounds.len(), 1);
assert_eq!(index.bounds.iter().map(|(call, _)| *call).collect::<Vec<_>>(), ["describe", "search"]);
let bounds: Vec<Duration> = embedder
.identity_bounds
.iter()
.chain(&embedder.bounds)
.copied()
.chain(index.bounds.iter().map(|(_, bound)| *bound))
.collect();
assert!(bounds.iter().all(|bound| !bound.is_zero() && *bound <= within), "{bounds:?}");
let served = crate::serve::ServeOptions::default().search_timeout;
assert_eq!(served, SEARCH_DEADLINE);
assert!(QUERY_EMBED_TIMEOUT <= served);
assert!(QUERY_EMBED_TIMEOUT < BULK_EMBED_TIMEOUT);
assert!(SEARCH_DEADLINE < BULK_EMBED_TIMEOUT);
}
/// A hung Ollama does not hold the worker past the search that asked: the
/// identity read never answers, and the search is answered, and the worker
/// free, when the search's own time has passed. A search that waited past its
/// time in the queue is answered without asking the embedder anything. A
/// ticket's time is `SEARCH_DEADLINE` from the moment it was asked.
#[test]
fn a_hung_embedder_holds_the_worker_no_longer_than_the_search_that_asked() {
let (_root, mind) = mind_with_documents();
let entries = mind.index_entries(None).unwrap();
let (embedder, index) = pair();
let mut sink = WorkerSink::spawn(embedder.clone(), index, &slug(INSTANCE), Some(MIND.into()), entries, quick());
sink.wait_status(|status| *status == IndexStatus::Current);
let asked = Instant::now();
ask_sink(&mut sink, "the ticket's time", 3).unwrap();
let due = lock(&sink.shared).searches.back().expect("the search is queued").due;
assert!(due >= asked + SEARCH_DEADLINE && due <= Instant::now() + SEARCH_DEADLINE, "due is SEARCH_DEADLINE after it was asked");
answers(&mut sink, 1);
embedder.state.lock().unwrap().hang_identity = true;
let started = Instant::now();
let due = started + Duration::from_millis(400);
lock(&sink.shared).searches.push_back(Queued { ticket: SearchTicket(70), text: "hung".into(), top_k: 3, due });
sink.shared.1.notify_all();
let answered = answers(&mut sink, 1);
assert_eq!(answered[0].0, SearchTicket(70));
assert!(answered[0].1.is_err(), "{answered:?}");
assert!(started.elapsed() < Duration::from_secs(5), "the worker was held for {:?}", started.elapsed());
embedder.state.lock().unwrap().hang_identity = false;
let reads = embedder.state.lock().unwrap().identity_bounds.len();
lock(&sink.shared).searches.push_back(Queued {
ticket: SearchTicket(71),
text: "late".into(),
top_k: 3,
due: Instant::now().checked_sub(Duration::from_secs(1)).unwrap_or_else(Instant::now),
});
sink.shared.1.notify_all();
let late = answers(&mut sink, 1);
let Err(why) = &late[0].1 else { panic!("{late:?}") };
assert!(why.contains("has run out"), "{why}");
assert_eq!(embedder.state.lock().unwrap().identity_bounds.len(), reads, "nothing was asked for a search nobody waits for");
let ticket = ask_sink(&mut sink, "the worker is free", 3).expect("the worker is still current");
assert_eq!(answers(&mut sink, 1)[0].0, ticket);
}
/// The Ollama adapter enforces the bound it is given: an embed that is never
@ -543,7 +613,7 @@ const EMBED: &str = r#"{"embeddings":[[0.5,1.0,2.0,3.0],[0.5,1.0,2.0,3.0],[0.5,1
fn ollama_identity_is_the_listed_digest_and_the_models_embedding_length() {
let ollama = stub(&[("GET /api/tags", 200, TAGS), ("POST /api/show", 200, SHOW), ("POST /api/embed", 200, EMBED)]);
let mut embedder = OllamaEmbedder::new(&ollama.url, "m:1");
assert_eq!(embedder.model_identity().unwrap(), ModelIdentity { name: "m:1".into(), digest: "ac6d".into(), dimensions: 4 });
assert_eq!(embedder.model_identity(TIME).unwrap(), ModelIdentity { name: "m:1".into(), digest: "ac6d".into(), dimensions: 4 });
let vectors = embedder.embed(&["a".into(), "b".into(), "c".into(), "d".into()], Duration::from_secs(5)).unwrap();
assert_eq!(vectors.len(), 4);
assert_eq!(vectors[0], vec![0.5, 1.0, 2.0, 3.0]);
@ -557,10 +627,10 @@ fn ollama_identity_is_the_listed_digest_and_the_models_embedding_length() {
("GET /api/tags", 200, r#"{"models":[{"name":"m:1","digest":"ac6d"}]}"#),
("POST /api/show", 200, SHOW),
]);
assert_eq!(OllamaEmbedder::new(&bare.url, "m:1").model_identity().unwrap().digest, "ac6d");
assert_eq!(OllamaEmbedder::new(&bare.url, "m:1").model_identity(TIME).unwrap().digest, "ac6d");
let mut absent = OllamaEmbedder::new(&ollama.url, "not-pulled:1");
assert!(format!("{:#}", absent.model_identity().unwrap_err()).contains("does not list the model"));
assert!(format!("{:#}", absent.model_identity(TIME).unwrap_err()).contains("does not list the model"));
let refusing = stub(&[("POST /api/embed", 500, r#"{"error":"boom"}"#)]);
assert!(OllamaEmbedder::new(&refusing.url, "m:1").embed(&["a".into()], Duration::from_secs(5)).is_err());
}
@ -600,12 +670,12 @@ fn the_embedders_address_is_not_part_of_the_collections_compatibility() {
fn qdrant_describes_creates_lists_and_writes_what_the_projector_asks() {
let name = collection();
let absent = stub(&[("GET /collections/", 404, r#"{"status":{"error":"not found"}}"#)]);
assert_eq!(QdrantIndex::new(&absent.url).describe(&name).unwrap(), Described::Absent);
assert_eq!(QdrantIndex::new(&absent.url).describe(&name, TIME).unwrap(), Described::Absent);
let unlabelled = stub(&[("GET /collections/", 200, r#"{"result":{"config":{"metadata":null}}}"#)]);
assert_eq!(QdrantIndex::new(&unlabelled.url).describe(&name).unwrap(), Described::Unlabelled);
assert_eq!(QdrantIndex::new(&unlabelled.url).describe(&name, TIME).unwrap(), Described::Unlabelled);
let voidbot = stub(&[("GET /collections/", 200, r#"{"result":{"config":{"metadata":{"managedBy":"voidbot"}}}}"#)]);
assert_eq!(QdrantIndex::new(&voidbot.url).describe(&name).unwrap(), Described::Unlabelled);
assert_eq!(QdrantIndex::new(&voidbot.url).describe(&name, TIME).unwrap(), Described::Unlabelled);
let labelled = stub(&[(
"GET /collections/",
@ -613,7 +683,7 @@ fn qdrant_describes_creates_lists_and_writes_what_the_projector_asks() {
r#"{"result":{"config":{"metadata":{"managed_by":"huginn","instance":"yggdrasil","mind":"mind-commit-first","model":"m:1","model_digest":"ac6d","dimensions":4,"index_text_version":1}}}}"#,
)]);
assert_eq!(
QdrantIndex::new(&labelled.url).describe(&name).unwrap(),
QdrantIndex::new(&labelled.url).describe(&name, TIME).unwrap(),
Described::Labelled(CollectionMeta {
managed_by: "huginn".into(),
instance: "yggdrasil".into(),
@ -943,14 +1013,14 @@ fn a_search_embeds_the_query_with_its_instruction_and_oversamples() {
Hit { doc_id: "eureka-state:mystery:X".into(), kind: "mystery".into(), score: 0.8 },
];
let hits = projector.search("who owns the state", 3).unwrap();
let hits = projector.search("who owns the state", 3, SEARCH_DEADLINE).unwrap();
assert_eq!(hits.len(), 1, "the unknown kind is left out");
assert_eq!(hits[0].0.id.0, "eureka-state:question:Q1");
assert_eq!(hits[0].0.kind, PipelineKind::Question);
assert_eq!(hits[0].1, 0.9);
let embedded = embedder.texts();
assert_eq!(embedded.last().unwrap(), &format!("Instruct: {QUERY_INSTRUCTION}\nQuery: who owns the state"));
projector.search("again", 100).unwrap();
projector.search("again", 100, SEARCH_DEADLINE).unwrap();
let searches = index.state.lock().unwrap().searches.clone();
assert_eq!(searches.iter().map(|(name, length, limit)| (name.as_str(), *length, *limit)).collect::<Vec<_>>(), [
(collection().as_str(), 4, 12),
@ -967,13 +1037,13 @@ fn a_search_refuses_when_the_index_is_not_reconciled_with_the_model() {
let entries = mind.index_entries(None).unwrap();
let (embedder, index) = pair();
let mut fresh = Projector::new(embedder.clone(), index.clone(), &slug(INSTANCE), Some(MIND.into()));
assert!(format!("{:#}", fresh.search("anything", 3).unwrap_err()).contains("being rebuilt"));
assert!(format!("{:#}", fresh.search("anything", 3, SEARCH_DEADLINE).unwrap_err()).contains("being rebuilt"));
assert!(embedder.texts().is_empty() && index.state.lock().unwrap().searches.is_empty());
let mut projector = projected(&embedder, &index, &entries);
embedder.state.lock().unwrap().identity.digest = "d2".into();
let before = embedder.texts().len();
assert!(format!("{:#}", projector.search("anything", 3).unwrap_err()).contains("model changed"));
assert!(format!("{:#}", projector.search("anything", 3, SEARCH_DEADLINE).unwrap_err()).contains("model changed"));
assert_eq!(embedder.texts().len(), before, "the query was not embedded under the new model");
assert!(index.state.lock().unwrap().searches.is_empty());
assert!(projector.stale(), "and the reconciliation is to be redone");
@ -991,7 +1061,7 @@ fn qdrant_search_sends_no_filter_and_reads_ids_kinds_and_scores() {
200,
r#"{"result":[{"id":"p1","score":0.75,"payload":{"doc_id":"eureka-state:question:Q1","kind":"question"}},{"id":"p2","score":0.5,"payload":{"doc_id":"eureka-state:ruling:R1","kind":"ruling"}}]}"#,
)]);
let hits = QdrantIndex::new(&answering.url).search(&name, &[0.5, 1.0, 2.0, 3.0], 12).unwrap();
let hits = QdrantIndex::new(&answering.url).search(&name, &[0.5, 1.0, 2.0, 3.0], 12, TIME).unwrap();
assert_eq!(
hits,
[
@ -1008,15 +1078,19 @@ fn qdrant_search_sends_no_filter_and_reads_ids_kinds_and_scores() {
drop(seen);
let nameless = stub(&[("POST /collections/", 200, r#"{"result":[{"id":"p","score":0.5,"payload":{"kind":"question"}}]}"#)]);
assert!(QdrantIndex::new(&nameless.url).search(&name, &[1.0], 1).is_err());
assert!(QdrantIndex::new(&nameless.url).search(&name, &[1.0], 1, TIME).is_err());
let failing = stub(&[("POST /collections/", 404, r#"{"status":{"error":"not found"}}"#)]);
assert!(QdrantIndex::new(&failing.url).search(&name, &[1.0], 1).is_err());
assert!(QdrantIndex::new(&failing.url).search(&name, &[1.0], 1, TIME).is_err());
}
fn queued(ticket: u64, text: &str) -> Queued {
Queued { ticket: SearchTicket(ticket), text: text.into(), top_k: 3, due: Instant::now() + SEARCH_DEADLINE }
}
fn shared_with(
inbox: Vec<IndexEntry>,
mind: Option<String>,
searches: Vec<(SearchTicket, String, u32)>,
searches: Vec<Queued>,
closed: bool,
) -> Lock {
Arc::new((
@ -1048,7 +1122,7 @@ fn an_idle_wait_ends_for_work_and_otherwise_at_its_deadline() {
};
assert!(waits(&shared_with(vec![entry], None, Vec::new(), false)), "an entry");
assert!(waits(&shared_with(Vec::new(), Some(MIND.into()), Vec::new(), false)), "the mind's identity");
assert!(waits(&shared_with(Vec::new(), None, vec![(SearchTicket(0), "q".into(), 3)], false)), "a search");
assert!(waits(&shared_with(Vec::new(), None, vec![queued(0, "q")], false)), "a search");
assert!(waits(&shared_with(Vec::new(), None, Vec::new(), true)), "the sink closing");
let quiet = shared_with(Vec::new(), None, Vec::new(), false);
@ -1066,7 +1140,7 @@ fn an_idle_wait_ends_for_work_and_otherwise_at_its_deadline() {
fn a_backoff_wait_is_the_whole_delay_whatever_arrives() {
let (_root, mind) = mind_with_documents();
let entry = mind.index_entries(None).unwrap().remove(0);
let busy = shared_with(vec![entry], Some(MIND.into()), vec![(SearchTicket(0), "q".into(), 3)], false);
let busy = shared_with(vec![entry], Some(MIND.into()), vec![queued(0, "q")], false);
let started = Instant::now();
wait_for_work(&busy, Wait::Backoff(Duration::from_millis(80)));
assert!(started.elapsed() >= Duration::from_millis(80), "the delay was waited out although work was queued");
@ -1080,8 +1154,7 @@ fn a_backoff_wait_is_the_whole_delay_whatever_arrives() {
/// reason, when the worker fails: nothing is left to wait for its deadline.
#[test]
fn queued_searches_are_answered_with_why_when_the_worker_gives_up() {
let queued = vec![(SearchTicket(3), "a".into(), 1), (SearchTicket(4), "b".into(), 2)];
let shared = shared_with(Vec::new(), None, queued, false);
let shared = shared_with(Vec::new(), None, vec![queued(3, "a"), queued(4, "b")], false);
refuse_searches(&shared, "the index could not advance: down");
let guard = lock(&shared);
assert!(guard.searches.is_empty());
@ -1141,13 +1214,15 @@ fn queries(embedder: &FakeEmbedder) -> Vec<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.
/// A search is refused only while the index is being rebuilt or cannot be
/// asked, and the words name the state and how many documents are pending. Lag
/// is not a rebuild: the sink and the projector ask the one predicate and both
/// let it through.
#[test]
fn only_a_current_index_answers_a_search() {
fn a_search_is_refused_only_for_a_rebuild_or_a_failure() {
use IndexStatus as S;
assert_eq!(unsearchable(&S::Current), None);
assert_eq!(unsearchable(&S::Behind { pending: 5 }), None, "lag is waited out, not refused");
let says = |status: S, wanted: &[&str]| {
let why = unsearchable(&status).unwrap_or_else(|| panic!("{status:?} answered"));
for word in wanted {
@ -1155,65 +1230,68 @@ fn only_a_current_index_answers_a_search() {
}
};
says(S::Reconciling { pending: 7 }, &["being rebuilt", "reconciling", "pending: 7"]);
says(S::Behind { pending: 5 }, &["being rebuilt", "behind", "pending: 5"]);
says(S::Failing { pending: 3, attempts: 2, error: "boom".into() }, &["failing", "boom", "pending: 3"]);
says(S::Refused { pending: 4, attempts: 2, reason: "not ours".into() }, &["refused", "not ours", "pending: 4"]);
// The projector never answers from a half-filled collection, and never
// touches the embedder or the vector store to find out.
// The sink: what the worker last published, and the documents admitted
// since, which are lag whatever the published state says of a current index.
let sink_in = |status: S, inbox: usize| {
let shared = shared_with(entries_of(inbox), None, Vec::new(), false);
lock(&shared).status = status;
WorkerSink { shared, identified: true }
};
assert!(ask_sink(&mut sink_in(S::Current, 2), "anything", 3).is_ok(), "admitted a moment ago and not yet absorbed");
assert!(ask_sink(&mut sink_in(S::Behind { pending: 4 }, 0), "anything", 3).is_ok(), "the worker is writing");
let refusal = ask_sink(&mut sink_in(S::Reconciling { pending: 4 }, 2), "anything", 3).unwrap_err();
assert!(refusal.contains("being rebuilt") && refusal.contains("pending: 6"), "{refusal}");
let refusal = ask_sink(&mut sink_in(S::Failing { pending: 1, attempts: 1, error: "boom".into() }, 0), "anything", 3).unwrap_err();
assert!(refusal.contains("failing") && refusal.contains("boom"), "{refusal}");
// The projector: a rebuild refuses without touching the ports to find out.
let (_root, mind) = mind_with_documents();
let entries = mind.index_entries(None).unwrap();
let (embedder, index) = pair();
let mut projector = projected(&embedder, &index, &entries[..3]);
projector.want(entries[3..].iter().cloned());
let embedded = embedder.texts().len();
let refusal = format!("{:#}", projector.search("anything", 3).unwrap_err());
assert!(refusal.contains("being rebuilt") && refusal.contains("pending: 1"), "{refusal}");
assert_eq!(embedder.texts().len(), embedded);
assert!(index.state.lock().unwrap().searches.is_empty());
drain(&mut projector);
assert!(projector.search("anything", 3).is_ok(), "once it has caught up it answers");
// The sink refuses before the worker is asked, while the worker is inside a batch.
let gate = Gate::shut();
let mut sink = WorkerSink::spawn(
FakeEmbedder::new("d1").behind(&gate),
FakeIndex::default(),
&slug(INSTANCE),
Some(MIND.into()),
entries.clone(),
quick(),
);
gate.wait_arrived(1);
let refusal = ask_sink(&mut sink, "anything", 3).unwrap_err();
let mut fresh = Projector::new(embedder.clone(), index.clone(), &slug(INSTANCE), Some(MIND.into()));
fresh.want(entries.iter().cloned());
let refusal = format!("{:#}", fresh.search("anything", 3, SEARCH_DEADLINE).unwrap_err());
assert!(refusal.contains("being rebuilt") && refusal.contains("pending: 4"), "{refusal}");
gate.open();
sink.wait_status(|status| *status == IndexStatus::Current);
assert!(ask_sink(&mut sink, "anything", 3).is_ok());
assert!(embedder.texts().is_empty() && index.state.lock().unwrap().searches.is_empty());
// Lag it lets through: catching up is the worker's, before the search is taken.
let mut lagging = projected(&embedder, &index, &entries[..3]);
lagging.want(entries[3..].iter().cloned());
assert_eq!(lagging.progress_status(), IndexStatus::Behind { pending: 1 });
assert!(lagging.search("anything", 3, SEARCH_DEADLINE).is_ok());
}
/// A search that fails after it was accepted has found the index not what it
/// claimed: the reconciliation is forgotten, so the status stops reading
/// `Current` and the worker verifies the model and the collection again.
/// A search that fails for its own reasons (the vector store errs, the
/// embedder is down, an identity read times out) fails that search alone: the
/// reconciliation stands, the status stays `Current`, the collection is not
/// listed again, and the next search is asked and answered.
#[test]
fn a_search_that_fails_forgets_the_reconciliation() {
fn a_failed_search_fails_that_search_alone() {
let (_root, mind) = mind_with_documents();
let entries = mind.index_entries(None).unwrap();
let breaks: [(&str, Box<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)),
type Fault = Box<dyn Fn(&FakeEmbedder, &FakeIndex, bool)>;
let faults: [(&str, Fault); 3] = [
("the vector store fails the search", Box::new(|_, index, on| index.state.lock().unwrap().fail_search = on)),
("the embedder is down", Box::new(|embedder, _, on| embedder.state.lock().unwrap().down = on)),
("the identity read never answers", Box::new(|embedder, _, on| embedder.state.lock().unwrap().hang_identity = on)),
];
for (what, breaks) in breaks {
for (what, fault) in faults {
let (embedder, index) = pair();
let mut projector = projected(&embedder, &index, &entries);
assert_eq!(projector.progress_status(), IndexStatus::Current);
breaks(&embedder, &index);
assert!(projector.search("anything", 3).is_err(), "{what}");
assert_eq!(projector.progress_status(), IndexStatus::Reconciling { pending: 0 }, "{what}");
assert!(projector.stale(), "{what}");
let listed = index.state.lock().unwrap().listed;
fault(&embedder, &index, true);
assert!(projector.search("anything", 3, Duration::from_millis(50)).is_err(), "{what}");
assert_eq!(projector.progress_status(), IndexStatus::Current, "{what}");
assert!(!projector.stale(), "{what}");
fault(&embedder, &index, false);
assert!(projector.search("anything", 3, SEARCH_DEADLINE).is_ok(), "{what}: the next search is answered");
assert_eq!(index.state.lock().unwrap().listed, listed, "{what}: the collection was not listed again");
}
// In the worker the next step is the verification, with nothing to prompt it.
// In the worker the failed search is answered and nothing else moves.
let (embedder, index) = pair();
let mut sink = WorkerSink::spawn(embedder, index.clone(), &slug(INSTANCE), Some(MIND.into()), entries, quick());
sink.wait_status(|status| *status == IndexStatus::Current);
@ -1223,7 +1301,24 @@ fn a_search_that_fails_forgets_the_reconciliation() {
let failed = answers(&mut sink, 1);
assert_eq!(failed[0].0, ticket);
assert!(failed[0].1.is_err());
eventually("the worker listing the collection again", || index.state.lock().unwrap().listed > listed);
index.state.lock().unwrap().fail_search = false;
let next = ask_sink(&mut sink, "anything", 3).expect("the index is still current, not being rebuilt");
let answered = answers(&mut sink, 1);
assert_eq!((answered[0].0, answered[0].1.is_ok()), (next, true));
assert_eq!(index.state.lock().unwrap().listed, listed, "the worker did not list the collection again");
}
/// A model the search finds changed is a rebuild, and its refusal names the
/// state and how many documents are pending, as every other refusal does.
#[test]
fn a_model_changed_refusal_names_the_state_and_the_pending_count() {
let (_root, mind) = mind_with_documents();
let entries = mind.index_entries(None).unwrap();
let (embedder, index) = pair();
let mut projector = projected(&embedder, &index, &entries);
embedder.state.lock().unwrap().identity.digest = "d2".into();
let refusal = format!("{:#}", projector.search("anything", 3, SEARCH_DEADLINE).unwrap_err());
assert!(refusal.contains("model changed") && refusal.contains("being rebuilt") && refusal.contains("pending: 4"), "{refusal}");
}
/// A search asks the store what the collection says about itself: a label for
@ -1252,7 +1347,7 @@ fn a_search_re_reads_the_collections_label() {
let mut projector = projected(&embedder, &index, &entries);
change(index.state.lock().unwrap().collections.get_mut(&collection()).unwrap());
let embedded = embedder.texts().len();
let refusal = format!("{:#}", projector.search("anything", 3).unwrap_err());
let refusal = format!("{:#}", projector.search("anything", 3, SEARCH_DEADLINE).unwrap_err());
assert!(refusal.contains("no longer the one this mind's index was built as"), "{what}: {refusal}");
assert_eq!(embedder.texts().len(), embedded, "{what}: nothing was embedded");
assert!(index.state.lock().unwrap().searches.is_empty(), "{what}");
@ -1261,7 +1356,7 @@ fn a_search_re_reads_the_collections_label() {
let (embedder, index) = pair();
let mut projector = projected(&embedder, &index, &entries);
index.state.lock().unwrap().collections.clear();
assert!(projector.search("anything", 3).is_err(), "no collection");
assert!(projector.search("anything", 3, SEARCH_DEADLINE).is_err(), "no collection");
assert!(projector.stale());
}
@ -1278,7 +1373,7 @@ fn a_model_that_changes_while_the_query_is_embedded_is_refused() {
let mut projector = projected(&embedder, &index, &entries);
gate.shut_again();
let searching = std::thread::spawn(move || {
let found = projector.search("anything", 3);
let found = projector.search("anything", 3, SEARCH_DEADLINE);
(projector, found)
});
gate.wait_arrived(1);
@ -1372,33 +1467,72 @@ fn one_queued_search_is_served_per_turn() {
let entries = mind.index_entries(None).unwrap();
let (embedder, index) = pair();
let mut projector = projected(&embedder, &index, &entries);
let queued = vec![(SearchTicket(1), "a".into(), 3), (SearchTicket(2), "b".into(), 3), (SearchTicket(3), "c".into(), 3)];
let shared = shared_with(Vec::new(), None, queued, false);
let shared = shared_with(Vec::new(), None, vec![queued(1, "a"), queued(2, "b"), queued(3, "c")], false);
serve_search(&mut projector, &shared);
let guard = lock(&shared);
assert_eq!(guard.searches.iter().map(|(ticket, ..)| *ticket).collect::<Vec<_>>(), [SearchTicket(2), SearchTicket(3)]);
assert_eq!(guard.searches.iter().map(|queued| queued.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.
/// A search reads its own writes. Asked while a batch is in the embedder and
/// eight more documents wait, it is accepted, waits behind them, and is
/// answered only once every one is in the collection.
#[test]
fn a_waiting_search_is_answered_between_batches() {
fn a_search_waits_for_the_writes_ahead_of_it_and_reads_them() {
let gate = Gate::shut();
let (_, index) = pair();
let embedder = FakeEmbedder::new("d1").behind(&gate);
let mut sink = WorkerSink::spawn(embedder, index, &slug(INSTANCE), Some(MIND.into()), entries_of(BATCH + 8), quick());
let mut sink = WorkerSink::spawn(embedder, index.clone(), &slug(INSTANCE), Some(MIND.into()), entries_of(BATCH + 8), quick());
gate.wait_arrived(1);
lock(&sink.shared).searches.push_back((SearchTicket(9), "anything".into(), 3));
let ticket = ask_sink(&mut sink, "anything", 3).expect("lag is not a rebuild");
gate.open();
let answered = answers(&mut sink, 1);
assert_eq!(answered[0].0, SearchTicket(9));
let Err(refusal) = &answered[0].1 else { panic!("a search was answered from a half-filled collection: {answered:?}") };
assert!(refusal.contains("being rebuilt") && refusal.contains("pending: 8"), "{refusal}");
assert_eq!(answered[0].0, ticket);
assert!(answered[0].1.is_ok(), "{answered:?}");
assert_eq!(index.state.lock().unwrap().held_when_searched, [BATCH + 8], "the search saw every document admitted before it");
sink.wait_status(|status| *status == IndexStatus::Current);
}
/// The worker takes admissions before the search, and leaves the search queued
/// while the writes ahead of it are unfinished: a search asked with a document
/// still in the inbox absorbs it and waits, and once the index has caught up
/// it is answered. An index being rebuilt is not waited on: the search is taken
/// and refused, naming the state and the pending count.
#[test]
fn the_worker_absorbs_admissions_before_a_search_and_waits_for_lag_but_not_for_a_rebuild() {
let (_root, mind) = mind_with_documents();
let entries = mind.index_entries(None).unwrap();
let (embedder, index) = pair();
let mut projector = projected(&embedder, &index, &entries[..3]);
let shared = shared_with(vec![entries[3].clone()], None, vec![queued(1, "a")], false);
serve_search(&mut projector, &shared);
{
let guard = lock(&shared);
assert!(guard.inbox.is_empty(), "the admission ahead of the search was absorbed");
assert_eq!(guard.searches.len(), 1, "the search waits for it to be written");
assert!(guard.found.is_empty());
}
assert_eq!(projector.pending(), 1);
drain(&mut projector);
serve_search(&mut projector, &shared);
let guard = lock(&shared);
assert!(guard.searches.is_empty());
assert!(matches!(guard.found.as_slice(), [(SearchTicket(1), Ok(_))]), "{:?}", guard.found);
assert_eq!(index.state.lock().unwrap().held_when_searched, [4]);
drop(guard);
// A rebuild: the projector has forgotten its reconciliation with the entries owed.
let mut rebuilding = Projector::new(embedder.clone(), index.clone(), &slug(INSTANCE), Some(MIND.into()));
rebuilding.want(entries.iter().cloned());
let shared = shared_with(Vec::new(), None, vec![queued(2, "b")], false);
serve_search(&mut rebuilding, &shared);
let guard = lock(&shared);
let [(SearchTicket(2), Err(refusal))] = guard.found.as_slice() else { panic!("{:?}", guard.found) };
assert!(refusal.contains("being rebuilt") && refusal.contains("pending: 4"), "{refusal}");
}
/// A search that was queued when the worker failed is answered with why, by
/// the worker's own failure path.
#[test]
@ -1408,13 +1542,13 @@ fn a_search_queued_before_the_worker_fails_is_answered_with_why() {
let embedder = FakeEmbedder::new("d1").behind(&gate);
let mut sink = WorkerSink::spawn(embedder.clone(), index, &slug(INSTANCE), Some(MIND.into()), entries_of(1), quick());
gate.wait_arrived(1);
lock(&sink.shared).searches.push_back((SearchTicket(5), "anything".into(), 3));
lock(&sink.shared).searches.push_back(queued(5, "anything"));
embedder.state.lock().unwrap().down = true;
gate.open();
let answered = answers(&mut sink, 1);
assert_eq!(answered[0].0, SearchTicket(5));
let Err(why) = &answered[0].1 else { panic!("{answered:?}") };
assert!(why.contains("the index could not advance") && why.contains("the embedder is down"), "{why}");
assert!(why.contains("the semantic index is failing") && why.contains("pending: 1") && why.contains("the embedder is down"), "{why}");
}
/// While a failure is being reported, the report's pending count is the one

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

@ -461,12 +461,13 @@ fn a_search_that_timed_out_while_queued_is_never_embedded() {
assert!(embedded[0].ends_with("Query: first") && embedded[1].ends_with("Query: third"), "{embedded:?}");
}
/// While the index is rebuilding, a semantic query is refused with an
/// `Unavailable` that names the state and how many documents are pending, and
/// the embedder is not asked for it. When the index has caught up the same
/// query is answered.
/// A search asked while the index lags the mind is not refused: it waits behind
/// the writes ahead of it, through the real serve loop, and reads them. The
/// worker is inside its first batch with the rest of the mind still to write;
/// the reply does not come until the gate opens, and it comes with every
/// document already in the collection.
#[test]
fn a_search_while_the_index_is_being_rebuilt_is_refused_and_asks_the_embedder_nothing() {
fn a_search_asked_while_the_index_lags_waits_for_the_writes_ahead_of_it_and_reads_them() {
let gate = Gate::shut();
let embedder = FakeEmbedder::new("d1").behind(&gate);
let harness = Harness::start(shelf(), embedder, FakeIndex::default(), Duration::from_secs(30));
@ -474,14 +475,17 @@ fn a_search_while_the_index_is_being_rebuilt_is_refused_and_asks_the_embedder_no
let mut client = harness.client();
let IndexStatus::Behind { pending } = index_status(&mut client, "m-status") else { panic!("the worker is inside its first batch") };
assert!(pending > 0);
harness.index.state.lock().unwrap().hits = vec![hit("ruling", id("ruling", "R4"), 0.5)];
let refused = ask(&mut client, "m-behind", &semantic(standing(), 5));
let HuginnMindResponse::Refused(MindRefusal::Unavailable { detail }) = &refused else { panic!("{refused:?}") };
assert!(detail.contains("being rebuilt") && detail.contains(&format!("pending: {pending}")), "{detail}");
assert!(harness.embedder.texts().is_empty(), "nothing was embedded, not even the documents, for a refused search");
send(&mut client, "m-behind", &semantic(standing(), 5));
assert!(receive(&mut client, "m-behind", 100).is_none(), "the search waits: it is neither refused nor answered from a half-filled collection");
gate.open();
let answered = receive(&mut client, "m-behind", 3000).expect("the search is answered once the index has caught up");
assert_eq!(ids(&answered), vec![id("ruling", "R4")]);
until_index(&mut client, "Current", |status| *status == IndexStatus::Current);
harness.index.state.lock().unwrap().hits = vec![hit("ruling", id("ruling", "R4"), 0.5)];
assert_eq!(ids(&ask(&mut client, "m-current", &semantic(standing(), 5))), vec![id("ruling", "R4")]);
let state = harness.index.state.lock().unwrap();
let written = state.collections[&collection_name(&slug(INSTANCE))].points.len();
assert_eq!(state.held_when_searched, [written], "the search saw every document");
assert!(written >= pending as usize);
}

8
crates/huginn-daemon/tests/live_index_smoke.rs
View file

@ -44,7 +44,7 @@ fn the_adapters_embed_write_query_and_clean_up_a_scratch_collection() {
let dotted = format!("{plain}.dotted");
let mut embedder = OllamaEmbedder::new(&ollama, &model);
let identity = embedder.model_identity().expect("the model is listed");
let identity = embedder.model_identity(Duration::from_secs(60)).expect("the model is listed");
println!("identity: {identity:?}");
let texts = ["The organ owns the mind.".to_string(), "A recipe for lentil soup.".to_string()];
let vectors = embedder.embed(&texts, Duration::from_secs(60)).expect("two texts embed");
@ -63,9 +63,9 @@ fn the_adapters_embed_write_query_and_clean_up_a_scratch_collection() {
let mut store = QdrantIndex::new(&qdrant);
let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
for name in [&plain, &dotted] {
assert_eq!(store.describe(name).unwrap(), Described::Absent);
assert_eq!(store.describe(name, Duration::from_secs(60)).unwrap(), Described::Absent);
store.recreate(name, &meta).unwrap();
assert_eq!(store.describe(name).unwrap(), Described::Labelled(meta.clone()));
assert_eq!(store.describe(name, Duration::from_secs(60)).unwrap(), Described::Labelled(meta.clone()));
assert!(store.ids(name).unwrap().is_empty());
let points: Vec<Point> = texts
.iter()
@ -97,7 +97,7 @@ fn the_adapters_embed_write_query_and_clean_up_a_scratch_collection() {
let asked = format!("Instruct: {QUERY_INSTRUCTION}
Query: Who is responsible for keeping the memory?");
let query = embedder.embed(&[asked], Duration::from_secs(60)).expect("the query embeds").remove(0);
let hits = store.search(name, &query, 2).unwrap();
let hits = store.search(name, &query, 2, Duration::from_secs(60)).unwrap();
assert_eq!(hits.len(), 2, "{hits:?}");
assert_eq!((hits[0].doc_id.as_str(), hits[0].kind.as_str()), ("smoke:doc:0", "question"), "{hits:?}");
assert!(hits[0].score > hits[1].score, "{hits:?}");