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

Merge the Cut 11b follow-ups: read-your-writes, rebuild refusal, one deadline

Browse source 
A search waits behind acknowledged admissions and answers; a recreated
index refuses until refilled; a first build is lag; the sink owns the
search deadline; every search-path call is bounded by it. Soul pass 4:
merge; its two pins landed. Follow-up: an ask-time watermark would stop
sustained admissions from starving searches.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-29 22:03:15 +02:00
commit ff39d80297
9 changed files with 2285 additions and 1653 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -25,6 +25,9 @@ pub trait IndexSink<S: MindStore> {
/// Starts a search for the documents nearest `text`, at most `top_k` of
/// them wanted. `Err` says why the index cannot be asked now.
fn search(&mut self, text: &str, top_k: u32) -> Result<SearchTicket, String>;
/// How long after it was asked a search is worth answering: the index's
/// own bound on it, which the loop waiting for the answer must share.
fn search_deadline(&self) -> std::time::Duration;
/// 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>)>;
@ -174,6 +177,11 @@ impl<S: MindStore, I: IndexSink<S>> Daemon<S, I> {
}
}
/// How long a search is worth waiting for.
pub fn search_deadline(&self) -> std::time::Duration {
self.index.search_deadline()
}
/// Searches the index has finished since this was last called.
pub fn searched(&mut self) -> Vec<(SearchTicket, Result<Hits, String>)> {
self.index.searched()
@ -466,6 +474,10 @@ pub(crate) mod tests {
Err("this index cannot search".into())
}
fn search_deadline(&self) -> std::time::Duration {
std::time::Duration::ZERO
}
fn searched(&mut self) -> Vec<(SearchTicket, Result<Hits, String>)> {
Vec::new()
}
@ -502,6 +514,10 @@ pub(crate) mod tests {
Err("this index cannot search".into())
}
fn search_deadline(&self) -> std::time::Duration {
std::time::Duration::ZERO
}
fn searched(&mut self) -> Vec<(SearchTicket, Result<Hits, String>)> {
Vec::new()
}
@ -524,6 +540,10 @@ pub(crate) mod tests {
Err("this index cannot search".into())
}
fn search_deadline(&self) -> std::time::Duration {
std::time::Duration::ZERO
}
fn searched(&mut self) -> Vec<(SearchTicket, Result<Hits, String>)> {
Vec::new()
}

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

@ -66,6 +66,13 @@ 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,
/// How long an identity read takes to answer, when it answers.
pub(crate) identity_delay: Duration,
}
#[derive(Clone)]
@ -76,7 +83,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, identity_delay: Duration::ZERO };
Self { state: Arc::new(Mutex::new(state)), gate: None }
}
@ -91,12 +98,21 @@ 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");
}
Ok(state.identity.clone())
let (identity, delay) = (state.identity.clone(), state.identity_delay);
drop(state);
std::thread::sleep(delay);
Ok(identity)
}
fn embed(&mut self, texts: &[String], within: Duration) -> Result<Vec<Vec<f32>>> {
@ -135,6 +151,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 +175,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 +218,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())
}
}

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

@ -62,18 +62,31 @@
//! - A search runs on the worker between batches, so it waits for the batch in
//! 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
//! (`IndexSink::search_deadline` 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 `Backoff::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, and the worker judges a search and takes it in one critical
//! section, admissions first: a search waits in the queue while anything
//! ahead of it is owed (documents in the inbox, documents to write, a first
//! reconciliation to make), and is answered once the index has caught up. Lag
//! is not a rebuild and refuses nothing, and a mind's first build is lag:
//! there is no earlier index to protect. 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 collection made again because the model, its
//! digest or the text version changed or because it was lost, until every
//! document it owes is written again), `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;
@ -100,22 +113,22 @@ pub const MANAGED_BY: &str = "huginn";
/// Documents embedded and written per call.
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.
/// How long a search may take, queue wait included, by default
/// (`Backoff::search_deadline`). The sink stamps it on every search when it is
/// asked, and the serve loop asks the sink for it (`IndexSink::search_deadline`),
/// so the loop's wait and the worker's bound are one value.
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 +192,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 +209,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 +262,41 @@ 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)
}
}
/// Whether the collection holds an index worth protecting, and whether it is
/// being made again. The state that marks a rebuild lasts until the entries it
/// owes have been written; nothing else clears it.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum Build {
/// No reconciliation has yet found or made this mind's collection. Its
/// first build has nothing stale to protect, so it is lag.
Unbuilt,
/// The collection was built by this model and only lags the mind.
Built,
/// The collection was made again (another model, digest or text version,
/// or lost while the worker ran) and is refilled from what the projector
/// knows: until every entry it owes is written, it answers nothing.
Rebuilding,
}
/// 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> {
@ -262,6 +314,7 @@ pub struct Projector<E: Embedder, V: VectorIndex> {
/// The model the collection was reconciled under; `None` until it is, and
/// again whenever that reconciliation is forgotten.
reconciled: Option<ModelIdentity>,
build: Build,
}
impl<E: Embedder, V: VectorIndex> Projector<E, V> {
@ -275,6 +328,7 @@ impl<E: Embedder, V: VectorIndex> Projector<E, V> {
known: BTreeMap::new(),
wanted: BTreeSet::new(),
reconciled: None,
build: Build::Unbuilt,
}
}
@ -344,8 +398,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,10 +416,16 @@ 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")? {
Described::Absent => self.recreate(&wanted)?,
match self.index.describe(&self.collection, INDEX_CALL_TIMEOUT).context("describing the collection")? {
Described::Absent => {
// A collection lost while the worker ran was an index; one that
// was never made is a first build.
let build = if self.build == Build::Unbuilt { Build::Built } else { Build::Rebuilding };
self.recreate(&wanted, build)?;
}
Described::Unlabelled => {
return Err(StepError::Refused(format!(
"collection {} exists and carries no Huginn metadata; it is not this organ's to rewrite",
@ -380,12 +440,14 @@ impl<E: Embedder, V: VectorIndex> Projector<E, V> {
self.collection, found.managed_by, found.instance, found.mind, self.instance, mind
)));
}
Described::Labelled(found) if found != wanted => self.recreate(&wanted)?,
Described::Labelled(found) if found != wanted => self.recreate(&wanted, Build::Rebuilding)?,
Described::Labelled(_) if self.build == Build::Unbuilt => self.build = Build::Built,
Described::Labelled(_) => {}
}
let present = self.index.ids(&self.collection).context("listing the collection's points")?;
self.wanted = self.known.keys().filter(|document_id| !present.contains(&point_id(document_id))).cloned().collect();
self.reconciled = Some(identity);
self.settle();
Ok(())
}
@ -403,18 +465,30 @@ impl<E: Embedder, V: VectorIndex> Projector<E, V> {
}
}
fn recreate(&mut self, meta: &CollectionMeta) -> Result<()> {
self.index.recreate(&self.collection, meta).context("creating the collection")
/// Makes the collection again. `build` is what it is once made, recorded
/// with the collection so a failure before the refill leaves it recorded.
fn recreate(&mut self, meta: &CollectionMeta, build: Build) -> Result<()> {
self.index.recreate(&self.collection, meta).context("creating the collection")?;
self.build = build;
Ok(())
}
/// A rebuild ends when the entries it owes are all written: the one place
/// it is cleared.
fn settle(&mut self) {
if self.build == Build::Rebuilding && self.wanted.is_empty() {
self.build = Build::Built;
}
}
/// The one embed-and-upsert routine, for startup and live entries alike:
/// 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 +498,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) {
@ -450,31 +524,33 @@ impl<E: Embedder, V: VectorIndex> Projector<E, V> {
for entry in &batch {
self.wanted.remove(&entry.id.id.0);
}
self.settle();
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 +562,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");
bail!("the index has not been built yet");
};
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) {
@ -535,8 +608,12 @@ impl<E: Embedder, V: VectorIndex> Projector<E, V> {
pub fn progress_status(&self) -> IndexStatus {
match (&self.reconciled, self.pending()) {
(None, 0) if self.mind.is_none() => IndexStatus::Current,
// The first build of an identified mind has nothing stale to
// protect: the wait for it is lag.
(None, pending) if self.mind.is_some() && self.build == Build::Unbuilt => IndexStatus::Behind { pending },
(None, pending) => IndexStatus::Reconciling { pending },
(Some(_), 0) => IndexStatus::Current,
(Some(_), pending) if self.build == Build::Rebuilding => IndexStatus::Reconciling { pending },
(Some(_), pending) => IndexStatus::Behind { pending },
}
}
@ -554,17 +631,19 @@ impl<E: Embedder, V: VectorIndex> Projector<E, V> {
/// The worker's clock: the retry delay starts at `initial`, doubles per
/// consecutive failure and stops at `max`; an idle worker verifies the model
/// and the collection again every `recheck`.
/// and the collection again every `recheck`; a search is worth waiting for
/// `search_deadline` after it was asked.
#[derive(Clone, Copy, Debug)]
pub struct Backoff {
pub initial: Duration,
pub max: Duration,
pub recheck: Duration,
pub search_deadline: Duration,
}
impl Default for Backoff {
fn default() -> Self {
Self { initial: Duration::from_secs(30), max: Duration::from_secs(600), recheck: Duration::from_secs(60) }
Self { initial: Duration::from_secs(30), max: Duration::from_secs(600), recheck: Duration::from_secs(60), search_deadline: SEARCH_DEADLINE }
}
}
@ -579,6 +658,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 +677,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 +744,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,18 +766,48 @@ 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, and the judgement and the take are one critical
/// section: the inbox is absorbed and the search taken under one guard, so a
/// document acknowledged before the search was asked is wanted when the search
/// is judged, and none can slip in between. A search is taken only with nothing
/// ahead of it (`take_search`). 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:#}"));
let mut guard = lock(shared);
if !absorb_into(projector, &mut guard, None) {
return;
}
#[cfg(test)]
tests::after_absorb(&mut guard);
let Some(Queued { ticket, text, top_k, due }) = take_search(projector, &mut guard) else { return };
drop(guard);
let found = projector.search(&text, top_k, due.saturating_duration_since(Instant::now())).map_err(|error| format!("{error:#}"));
lock(shared).found.push((ticket, found));
}
/// The search at the front of the queue, if it may be served now. Never while
/// a document sits in the inbox, and never while anything ahead of it is
/// owed (documents to write, a reconciliation to make) unless the index
/// cannot be asked at all, in which case the search is taken to be refused:
/// waiting would not change the answer.
fn take_search<E: Embedder, V: VectorIndex>(projector: &Projector<E, V>, guard: &mut Shared) -> Option<Queued> {
if !guard.inbox.is_empty() {
return None;
}
let caught_up = projector.pending() == 0 && !projector.stale();
if !caught_up && unsearchable(&projector.progress_status()).is_none() {
return None;
}
guard.searches.pop_front()
}
/// 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) {
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())));
}
}
@ -705,7 +825,16 @@ fn absorb<E: Embedder, V: VectorIndex>(
shared: &Lock,
failing: Option<(&StepError, u32)>,
) -> bool {
let mut guard = lock(shared);
absorb_into(projector, &mut lock(shared), failing)
}
/// `absorb` under a guard the caller already holds, for the caller that must
/// decide something in the same critical section.
fn absorb_into<E: Embedder, V: VectorIndex>(
projector: &mut Projector<E, V>,
guard: &mut Shared,
failing: Option<(&StepError, u32)>,
) -> bool {
if guard.closed {
return false;
}
@ -748,8 +877,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 +891,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);
@ -779,6 +908,8 @@ pub struct WorkerSink {
shared: Lock,
/// Whether the worker has been told the mind's identity.
identified: bool,
/// How long after it was asked a search is worth answering.
deadline: Duration,
}
impl WorkerSink {
@ -815,7 +946,7 @@ impl WorkerSink {
));
let worker = Arc::clone(&shared);
std::thread::spawn(move || work(projector, worker, backoff));
Self { shared, identified }
Self { shared, identified, deadline: backoff.search_deadline }
}
/// Blocks until the health satisfies `predicate`, and returns it. For
@ -875,8 +1006,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,12 +1024,16 @@ 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() + self.deadline });
drop(guard);
self.shared.1.notify_all();
Ok(ticket)
}
fn search_deadline(&self) -> Duration {
self.deadline
}
fn searched(&mut self) -> Vec<(SearchTicket, Result<Hits, String>)> {
std::mem::take(&mut lock(&self.shared).found)
}
@ -905,7 +1042,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 +1069,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()

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

File diff suppressed because it is too large Load diff

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

@ -140,7 +140,7 @@ fn ids(response: &HuginnMindResponse) -> Vec<String> {
}
fn quick() -> Backoff {
Backoff { initial: Duration::from_millis(1), max: Duration::from_millis(2), recheck: Duration::from_secs(3600) }
Backoff { initial: Duration::from_millis(1), max: Duration::from_millis(2), recheck: Duration::from_secs(3600), search_deadline: Duration::from_secs(30) }
}
/// A daemon serving a mind over a real socket, with the ports the test gave it.
@ -167,7 +167,7 @@ impl Harness {
&slug(INSTANCE),
mind.genesis_receipt_id().unwrap(),
mind.index_entries(None).unwrap(),
quick(),
Backoff { search_deadline: search_timeout, ..quick() },
);
let mut daemon: Daemon<OwnedRedbMessagePackBackingStore, WorkerSink> = Daemon::new(mind, sink);
let mut hub = bind("127.0.0.1:0".parse().unwrap(), &daemon.runtime_id()).unwrap();
@ -175,7 +175,7 @@ impl Harness {
let registry = schema_registry().unwrap();
let stopping = Arc::new(AtomicBool::new(false));
let loop_stopping = Arc::clone(&stopping);
let options = ServeOptions { search_timeout, ..ServeOptions::default() };
let options = ServeOptions::default();
let serving = std::thread::spawn(move || run(&mut daemon, &mut hub, &registry, &loop_stopping, &options));
Self { endpoint, stopping, serving: Some(serving), embedder, index, _root: root }
}
@ -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);
}

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

@ -88,10 +88,6 @@ pub struct ServeOptions {
pub session_timeout: Duration,
pub idle_sleep: Duration,
pub deferred_ttl: Duration,
/// How long a semantic query may wait for the index before it is answered
/// `Unavailable`. The worker's own calls time out on their own; this is
/// what the client is promised.
pub search_timeout: Duration,
}
impl Default for ServeOptions {
@ -100,7 +96,6 @@ impl Default for ServeOptions {
session_timeout: Duration::from_secs(30),
idle_sleep: Duration::from_millis(2),
deferred_ttl: Duration::from_secs(60),
search_timeout: crate::index::SEARCH_DEADLINE,
}
}
}
@ -449,7 +444,8 @@ struct Waiting {
/// Until `stopping`: expire what timed out, resend what was not acknowledged,
/// answer every frame waiting on the socket, then answer every semantic query
/// the index has finished or that has waited out `search_timeout`. A semantic
/// the index has finished or that has waited out the index's own
/// `search_deadline`. A semantic
/// query is never answered inline: `answer` hands back a `Search`, the loop
/// holds it beside its session and goes on serving, and the embedder and the
/// vector store are the worker's to wait on. A hostile datagram and a departed
@ -497,13 +493,13 @@ pub fn run<S: MindStore, I: IndexSink<S>>(
match answer(daemon, registry, &mut bodies, message, Utc::now()) {
Routed::Reply(reply) => send(hub, &session, &reply),
Routed::Search { message_id, operation, search } => {
let until = Instant::now() + options.search_timeout;
let until = Instant::now() + daemon.search_deadline();
waiting.insert(search.ticket, Waiting { session, message_id, operation, search, until });
}
}
served += 1;
}
let finished = collect_searches(daemon, &waiting, options);
let finished = collect_searches(daemon, &waiting);
let answered = !finished.is_empty();
for (ticket, found) in finished {
let Some(asked) = waiting.remove(&ticket) else { continue };
@ -525,14 +521,14 @@ pub fn run<S: MindStore, I: IndexSink<S>>(
fn collect_searches<S: MindStore, I: IndexSink<S>>(
daemon: &mut Daemon<S, I>,
waiting: &BTreeMap<SearchTicket, Waiting>,
options: &ServeOptions,
) -> Vec<(SearchTicket, Result<crate::daemon::Hits, String>)> {
let deadline = daemon.search_deadline();
let mut collected = daemon.searched();
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))));
collected.push((*ticket, Err(format!("the semantic search did not finish within {deadline:?}"))));
}
}
collected

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:?}");