Decide and take a search under one guard; a refill is a rebuild; a first build is lag
S1. serve_search absorbed the inbox and judged pending under one lock and popped the search under a second, so an admission acknowledged before the search was asked could be missed. The absorb, the judgement and the take are now one critical section, and take_search never takes a search while the inbox is non-empty or anything ahead of it is owed. S2. The projector carries Build (Unbuilt, Built, Rebuilding). A collection made again (label mismatch, or lost after it was once built) is Rebuilding until the entries it owes are written; settle() is the one place that clears it, so a failure mid-refill does not end the rebuild. Rebuilding reports Reconciling. unsearchable stays the only refusal predicate. S3. A mind whose index was never built is Unbuilt: its first build reports Behind, so the search waits and answers whichever side looked first. Only a rebuild of an index that exists refuses. S5. ServeOptions::search_timeout is deleted. The sink owns the deadline (Backoff::search_deadline, stamped on every search) and the serve loop asks the daemon for it (IndexSink::search_deadline), so the loop's wait and the worker's bound are one value. A mind with no identity and documents owed is Reconciling, so a search is taken and refused rather than left for a spinning worker. S4. A test pins that a later call's bound shrinks as the search spends time. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens
parent
b48f0cb5ec
commit
e99dc4404e
6 changed files with 429 additions and 60 deletions
|
|
@ -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()
|
||||
}
|
||||
|
|
|
|||
|
|
@ -71,6 +71,8 @@ pub(crate) struct EmbedderState {
|
|||
/// 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)]
|
||||
|
|
@ -81,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, identity_bounds: Vec::new(), hang_identity: false };
|
||||
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 }
|
||||
}
|
||||
|
||||
|
|
@ -107,7 +109,10 @@ impl Embedder for FakeEmbedder {
|
|||
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>>> {
|
||||
|
|
|
|||
|
|
@ -62,23 +62,27 @@
|
|||
//! - 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 it has left: the ticket carries the
|
||||
//! moment `SEARCH_DEADLINE` after it was asked, and every call the search
|
||||
//! 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, 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.
|
||||
//! 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 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
|
||||
//! 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
|
||||
|
|
@ -109,10 +113,10 @@ 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`. Every bound on the search path derives from
|
||||
/// the moment this long after the search was asked, 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 a bulk index-flush embed, the identity and label reads around
|
||||
|
|
@ -277,6 +281,22 @@ impl Budget {
|
|||
}
|
||||
}
|
||||
|
||||
/// 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> {
|
||||
|
|
@ -294,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> {
|
||||
|
|
@ -307,6 +328,7 @@ impl<E: Embedder, V: VectorIndex> Projector<E, V> {
|
|||
known: BTreeMap::new(),
|
||||
wanted: BTreeSet::new(),
|
||||
reconciled: None,
|
||||
build: Build::Unbuilt,
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -398,7 +420,12 @@ impl<E: Embedder, V: VectorIndex> Projector<E, V> {
|
|||
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, INDEX_CALL_TIMEOUT).context("describing the collection")? {
|
||||
Described::Absent => self.recreate(&wanted)?,
|
||||
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",
|
||||
|
|
@ -413,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(())
|
||||
}
|
||||
|
||||
|
|
@ -436,8 +465,20 @@ 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:
|
||||
|
|
@ -483,6 +524,7 @@ impl<E: Embedder, V: VectorIndex> Projector<E, V> {
|
|||
for entry in &batch {
|
||||
self.wanted.remove(&entry.id.id.0);
|
||||
}
|
||||
self.settle();
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
|
@ -537,7 +579,7 @@ impl<E: Embedder, V: VectorIndex> Projector<E, V> {
|
|||
}
|
||||
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}");
|
||||
|
|
@ -566,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 },
|
||||
}
|
||||
}
|
||||
|
|
@ -585,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 }
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -719,20 +767,41 @@ fn unsearchable(status: &IndexStatus) -> Option<String> {
|
|||
/// 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.
|
||||
/// 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) {
|
||||
if !absorb(projector, shared, None) || (projector.pending() > 0 && !projector.stale()) {
|
||||
let mut guard = lock(shared);
|
||||
if !absorb_into(projector, &mut guard, None) {
|
||||
return;
|
||||
}
|
||||
let Some(Queued { ticket, text, top_k, due }) = lock(shared).searches.pop_front() else { 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) {
|
||||
|
|
@ -756,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;
|
||||
}
|
||||
|
|
@ -830,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 {
|
||||
|
|
@ -866,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
|
||||
|
|
@ -944,12 +1024,16 @@ impl<S: MindStore> IndexSink<S> for WorkerSink {
|
|||
}
|
||||
let ticket = SearchTicket(guard.next_ticket);
|
||||
guard.next_ticket += 1;
|
||||
guard.searches.push_back(Queued { ticket, text: text.to_owned(), top_k, due: Instant::now() + SEARCH_DEADLINE });
|
||||
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)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -19,6 +19,20 @@ use super::*;
|
|||
use crate::daemon::Daemon;
|
||||
use crate::daemon::tests::{INSTANCE, OTHER, batch, campaign_seed, now, question_and_ruling, slug};
|
||||
|
||||
thread_local! {
|
||||
/// Runs once, in the worker's own turn, between the inbox being absorbed
|
||||
/// and a search being taken, holding the guard `serve_search` holds.
|
||||
static AFTER_ABSORB: std::cell::RefCell<Option<Box<dyn FnOnce(&mut Shared)>>> = const { std::cell::RefCell::new(None) };
|
||||
}
|
||||
|
||||
/// The hook `serve_search` calls at the point where a second lock once left a
|
||||
/// gap. It is the test's only way in: nothing else can reach that point.
|
||||
pub(super) fn after_absorb(guard: &mut Shared) {
|
||||
if let Some(hook) = AFTER_ABSORB.with(|hook| hook.borrow_mut().take()) {
|
||||
hook(guard);
|
||||
}
|
||||
}
|
||||
|
||||
fn collection() -> String {
|
||||
collection_name(&slug(INSTANCE))
|
||||
}
|
||||
|
|
@ -84,6 +98,17 @@ fn projected(embedder: &FakeEmbedder, index: &FakeIndex, entries: &[IndexEntry])
|
|||
projector
|
||||
}
|
||||
|
||||
/// A projector whose collection was lost while it ran and has been made again:
|
||||
/// it owes every entry, and is rebuilding.
|
||||
fn rebuilding(embedder: &FakeEmbedder, index: &FakeIndex, entries: &[IndexEntry]) -> Projector<FakeEmbedder, FakeIndex> {
|
||||
let mut projector = projected(embedder, index, &entries[..1]);
|
||||
index.state.lock().unwrap().collections.remove(&collection());
|
||||
projector.want(entries[1..].iter().cloned());
|
||||
projector.advance().expect_err("the write meets no collection");
|
||||
assert_eq!(projector.advance().unwrap(), Advance::Progressed, "the reconciliation makes the collection again");
|
||||
projector
|
||||
}
|
||||
|
||||
/// P4's function, as voidbot's `toQdrantPointId` computes it: the first sixteen
|
||||
/// bytes of the SHA-256, version and variant bits set. The fixed vectors were
|
||||
/// computed independently of this crate.
|
||||
|
|
@ -347,8 +372,7 @@ fn every_call_on_the_search_path_is_bounded_by_the_searchs_time_and_the_flush_by
|
|||
.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_eq!(Backoff::default().search_deadline, SEARCH_DEADLINE);
|
||||
assert!(SEARCH_DEADLINE < BULK_EMBED_TIMEOUT);
|
||||
}
|
||||
|
||||
|
|
@ -435,7 +459,7 @@ fn committed(response: HuginnMindResponse) {
|
|||
}
|
||||
|
||||
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) }
|
||||
}
|
||||
|
||||
/// The index never delays or refuses admission: the embedder is shut, the
|
||||
|
|
@ -811,7 +835,7 @@ fn the_backoff_doubles_from_thirty_seconds_to_a_ten_minute_cap() {
|
|||
seen.push(delay.as_secs());
|
||||
}
|
||||
assert_eq!(seen, [30, 60, 120, 240, 480, 600, 600, 600]);
|
||||
let small = Backoff { initial: Duration::from_millis(1), max: Duration::from_millis(3), recheck: Duration::from_secs(3600) };
|
||||
let small = Backoff { initial: Duration::from_millis(1), max: Duration::from_millis(3), recheck: Duration::from_secs(3600), search_deadline: Duration::from_secs(30) };
|
||||
assert_eq!(small.after(Duration::from_millis(1)), Duration::from_millis(2));
|
||||
assert_eq!(small.after(Duration::from_millis(2)), Duration::from_millis(3));
|
||||
}
|
||||
|
|
@ -910,7 +934,7 @@ fn a_mind_without_an_identity_makes_no_collection_until_its_first_admission() {
|
|||
assert_eq!(projector.progress_status(), IndexStatus::Current);
|
||||
assert_eq!(projector.advance().unwrap(), Advance::Idle);
|
||||
let identified = Projector::new(embedder.clone(), index.clone(), &slug(INSTANCE), Some(MIND.into()));
|
||||
assert_eq!(identified.progress_status(), IndexStatus::Reconciling { pending: 0 }, "an identified mind has yet to reconcile");
|
||||
assert_eq!(identified.progress_status(), IndexStatus::Behind { pending: 0 }, "an identified mind has yet to build its first index: lag, not a rebuild");
|
||||
assert!(index.state.lock().unwrap().collections.is_empty());
|
||||
|
||||
let sink = WorkerSink::spawn(embedder, index.clone(), &slug(INSTANCE), None, Vec::new(), quick());
|
||||
|
|
@ -1037,7 +1061,7 @@ fn a_search_refuses_when_the_index_is_not_reconciled_with_the_model() {
|
|||
let entries = mind.index_entries(None).unwrap();
|
||||
let (embedder, index) = pair();
|
||||
let mut fresh = Projector::new(embedder.clone(), index.clone(), &slug(INSTANCE), Some(MIND.into()));
|
||||
assert!(format!("{:#}", fresh.search("anything", 3, SEARCH_DEADLINE).unwrap_err()).contains("being rebuilt"));
|
||||
assert!(format!("{:#}", fresh.search("anything", 3, SEARCH_DEADLINE).unwrap_err()).contains("not been built"));
|
||||
assert!(embedder.texts().is_empty() && index.state.lock().unwrap().searches.is_empty());
|
||||
|
||||
let mut projector = projected(&embedder, &index, &entries);
|
||||
|
|
@ -1238,7 +1262,7 @@ fn a_search_is_refused_only_for_a_rebuild_or_a_failure() {
|
|||
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 }
|
||||
WorkerSink { shared, identified: true, deadline: SEARCH_DEADLINE }
|
||||
};
|
||||
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");
|
||||
|
|
@ -1251,10 +1275,12 @@ fn a_search_is_refused_only_for_a_rebuild_or_a_failure() {
|
|||
let (_root, mind) = mind_with_documents();
|
||||
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()));
|
||||
fresh.want(entries.iter().cloned());
|
||||
let refusal = format!("{:#}", fresh.search("anything", 3, SEARCH_DEADLINE).unwrap_err());
|
||||
let (lost_embedder, lost_index) = pair();
|
||||
let mut rebuilding_now = rebuilding(&lost_embedder, &lost_index, &entries);
|
||||
let searches = lost_index.state.lock().unwrap().searches.len();
|
||||
let refusal = format!("{:#}", rebuilding_now.search("anything", 3, SEARCH_DEADLINE).unwrap_err());
|
||||
assert!(refusal.contains("being rebuilt") && refusal.contains("pending: 4"), "{refusal}");
|
||||
assert_eq!(lost_index.state.lock().unwrap().searches.len(), searches, "the rebuilt collection was not asked");
|
||||
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.
|
||||
|
|
@ -1524,9 +1550,9 @@ fn the_worker_absorbs_admissions_before_a_search_and_waits_for_lag_but_not_for_a
|
|||
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());
|
||||
// A rebuild: the collection was lost and made again, and every entry is owed.
|
||||
let (rebuild_embedder, rebuild_index) = pair();
|
||||
let mut rebuilding = rebuilding(&rebuild_embedder, &rebuild_index, &entries);
|
||||
let shared = shared_with(Vec::new(), None, vec![queued(2, "b")], false);
|
||||
serve_search(&mut rebuilding, &shared);
|
||||
let guard = lock(&shared);
|
||||
|
|
@ -1571,7 +1597,7 @@ fn a_failure_report_carries_the_pending_count_it_has_now() {
|
|||
assert_eq!(guard.status, IndexStatus::Failing { pending: 4, attempts: 3, error: "down".into() });
|
||||
}
|
||||
assert!(absorb(&mut projector, &shared, None));
|
||||
assert_eq!(lock(&shared).status, IndexStatus::Reconciling { pending: 4 });
|
||||
assert_eq!(lock(&shared).status, IndexStatus::Behind { pending: 4 }, "a first build is lag");
|
||||
lock(&shared).closed = true;
|
||||
assert!(!absorb(&mut projector, &shared, None), "a closed sink ends the worker");
|
||||
}
|
||||
|
|
@ -1640,3 +1666,241 @@ fn an_idle_worker_rechecks_once_per_interval() {
|
|||
let listed = index.state.lock().unwrap().listed;
|
||||
assert!((2..=40).contains(&listed), "the startup reconciliation and about five rechecks listed the collection {listed} times");
|
||||
}
|
||||
|
||||
// The search's rulings: what may be waiting ahead of it, when a rebuild
|
||||
// refuses, what a first build is, how time is spent, and who owns the deadline.
|
||||
|
||||
/// A search is never taken while a document sits in the inbox. An admission
|
||||
/// that lands after the inbox was absorbed and before the search was taken (the
|
||||
/// hook puts one there, where a second lock once left the gap) leaves the
|
||||
/// search queued; the next turn absorbs it, and the search is answered only
|
||||
/// once it is written.
|
||||
#[test]
|
||||
fn a_search_is_never_taken_while_an_admission_sits_in_the_inbox() {
|
||||
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::new(), None, vec![queued(1, "a")], false);
|
||||
let late = entries[3].clone();
|
||||
AFTER_ABSORB.with(|hook| *hook.borrow_mut() = Some(Box::new(move |guard| guard.inbox.push(late))));
|
||||
|
||||
serve_search(&mut projector, &shared);
|
||||
{
|
||||
let guard = lock(&shared);
|
||||
assert_eq!(guard.searches.len(), 1, "the search was not taken past the admission");
|
||||
assert!(guard.found.is_empty());
|
||||
assert_eq!(guard.inbox.len(), 1);
|
||||
}
|
||||
serve_search(&mut projector, &shared);
|
||||
assert_eq!(lock(&shared).searches.len(), 1, "absorbed, but not yet written");
|
||||
drain(&mut projector);
|
||||
serve_search(&mut projector, &shared);
|
||||
let guard = lock(&shared);
|
||||
assert!(matches!(guard.found.as_slice(), [(SearchTicket(1), Ok(_))]), "{:?}", guard.found);
|
||||
assert_eq!(index.state.lock().unwrap().held_when_searched, [4], "the search saw the admission");
|
||||
}
|
||||
|
||||
/// Through the sink, an admission acknowledged before the search was asked is
|
||||
/// in the inbox while the worker is inside a batch; the search reads it.
|
||||
#[test]
|
||||
fn a_search_reads_an_admission_acknowledged_before_it_was_asked() {
|
||||
let gate = Gate::shut();
|
||||
let (_, index) = pair();
|
||||
let embedder = FakeEmbedder::new("d1").behind(&gate);
|
||||
let mut sink = WorkerSink::spawn(embedder, index.clone(), &slug(INSTANCE), Some(MIND.into()), entries_of(2), quick());
|
||||
gate.wait_arrived(1);
|
||||
let admitted = entries_of(5).split_off(2);
|
||||
lock(&sink.shared).inbox.extend(admitted);
|
||||
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, ticket);
|
||||
assert!(answered[0].1.is_ok(), "{answered:?}");
|
||||
assert_eq!(index.state.lock().unwrap().held_when_searched, [5]);
|
||||
}
|
||||
|
||||
/// A collection made again is a rebuild until every document it owes is
|
||||
/// written again: searches are refused as rebuilding, however the collection
|
||||
/// came to be made again, and however many steps or failures the refill takes.
|
||||
#[test]
|
||||
fn a_collection_made_again_refuses_searches_until_it_is_refilled() {
|
||||
let refused = |projector: &mut Projector<FakeEmbedder, FakeIndex>, pending: u32| {
|
||||
let why = format!("{:#}", projector.search("anything", 3, SEARCH_DEADLINE).unwrap_err());
|
||||
assert!(why.contains("being rebuilt") && why.contains(&format!("pending: {pending}")), "{why}");
|
||||
};
|
||||
let entries = entries_of(BATCH + 8);
|
||||
|
||||
// Lost while the worker ran.
|
||||
let (embedder, index) = pair();
|
||||
let mut projector = rebuilding(&embedder, &index, &entries);
|
||||
assert_eq!(projector.progress_status(), IndexStatus::Reconciling { pending: 40 });
|
||||
refused(&mut projector, 40);
|
||||
index.state.lock().unwrap().down = true;
|
||||
projector.advance().expect_err("the store is down mid-refill");
|
||||
index.state.lock().unwrap().down = false;
|
||||
assert_eq!(projector.advance().unwrap(), Advance::Progressed, "reconciled again, under a label that already fits");
|
||||
assert_eq!(projector.progress_status(), IndexStatus::Reconciling { pending: 40 }, "a failure does not end the rebuild");
|
||||
assert_eq!(projector.advance().unwrap(), Advance::Progressed);
|
||||
assert_eq!(projector.progress_status(), IndexStatus::Reconciling { pending: 8 }, "half refilled is still a rebuild");
|
||||
refused(&mut projector, 8);
|
||||
assert_eq!(projector.advance().unwrap(), Advance::Progressed);
|
||||
assert_eq!(projector.progress_status(), IndexStatus::Current);
|
||||
assert!(projector.search("anything", 3, SEARCH_DEADLINE).is_ok(), "answered once refilled");
|
||||
assert_eq!(index.state.lock().unwrap().held_when_searched, [40]);
|
||||
projector.want(entries_of(1).into_iter().map(|mut entry| {
|
||||
entry.id.id.0.push_str("-later");
|
||||
entry
|
||||
}));
|
||||
assert_eq!(projector.progress_status(), IndexStatus::Behind { pending: 1 }, "a later admission is lag, not a rebuild");
|
||||
|
||||
// Made again because the model changed.
|
||||
let (embedder, index) = pair();
|
||||
let mut projector = projected(&embedder, &index, &entries);
|
||||
embedder.state.lock().unwrap().identity.digest = "d2".into();
|
||||
assert_eq!(projector.recheck().unwrap(), Advance::Progressed, "the change is seen and the collection made again");
|
||||
assert_eq!(projector.progress_status(), IndexStatus::Reconciling { pending: 40 });
|
||||
projector.advance().unwrap();
|
||||
assert_eq!(projector.progress_status(), IndexStatus::Reconciling { pending: 8 });
|
||||
refused(&mut projector, 8);
|
||||
drain(&mut projector);
|
||||
assert!(projector.search("anything", 3, SEARCH_DEADLINE).is_ok());
|
||||
}
|
||||
|
||||
/// Through the worker: a collection lost while the worker runs is made again,
|
||||
/// and a search asked during the refill is refused as rebuilding, naming how
|
||||
/// many documents are pending; one asked after it is answered.
|
||||
#[test]
|
||||
fn a_search_asked_during_a_refill_is_refused_and_one_after_it_is_answered() {
|
||||
let gate = Gate::shut();
|
||||
gate.open();
|
||||
let (_, index) = pair();
|
||||
let embedder = FakeEmbedder::new("d1").behind(&gate);
|
||||
let backoff = Backoff { recheck: Duration::from_millis(50), ..quick() };
|
||||
let mut sink = WorkerSink::spawn(embedder, index.clone(), &slug(INSTANCE), Some(MIND.into()), entries_of(BATCH + 8), backoff);
|
||||
sink.wait_status(|status| *status == IndexStatus::Current);
|
||||
gate.shut_again();
|
||||
index.state.lock().unwrap().collections.remove(&collection());
|
||||
gate.wait_arrived(1);
|
||||
let refusal = ask_sink(&mut sink, "anything", 3).unwrap_err();
|
||||
assert!(refusal.contains("being rebuilt") && refusal.contains("pending: 40"), "{refusal}");
|
||||
gate.open();
|
||||
sink.wait_status(|status| *status == IndexStatus::Current);
|
||||
let ticket = ask_sink(&mut sink, "anything", 3).expect("refilled");
|
||||
let answered = answers(&mut sink, 1);
|
||||
assert_eq!((answered[0].0, answered[0].1.is_ok()), (ticket, true));
|
||||
assert_eq!(index.state.lock().unwrap().held_when_searched.last(), Some(&40));
|
||||
}
|
||||
|
||||
/// A mind whose index has never been built has nothing stale to protect: its
|
||||
/// first build is lag, whether the search was asked before the worker had
|
||||
/// looked at the first admission or after. Only a rebuild of an index that
|
||||
/// exists refuses.
|
||||
#[test]
|
||||
fn a_minds_first_build_is_lag_whoever_looks_first() {
|
||||
for asked_before_the_worker_looked in [true, false] {
|
||||
let what = if asked_before_the_worker_looked { "asked first" } else { "asked after the worker looked" };
|
||||
let (embedder, index) = pair();
|
||||
let entries = entries_of(3);
|
||||
let (mut projector, shared) = if asked_before_the_worker_looked {
|
||||
let projector = Projector::new(embedder.clone(), index.clone(), &slug(INSTANCE), None);
|
||||
(projector, shared_with(entries, Some(MIND.into()), Vec::new(), false))
|
||||
} else {
|
||||
let mut projector = Projector::new(embedder.clone(), index.clone(), &slug(INSTANCE), Some(MIND.into()));
|
||||
projector.want(entries);
|
||||
let shared = shared_with(Vec::new(), None, Vec::new(), false);
|
||||
lock(&shared).status = projector.progress_status();
|
||||
(projector, shared)
|
||||
};
|
||||
let mut sink = WorkerSink { shared: Arc::clone(&shared), identified: true, deadline: SEARCH_DEADLINE };
|
||||
let ticket = ask_sink(&mut sink, "anything", 3).unwrap_or_else(|why| panic!("{what}: {why}"));
|
||||
|
||||
serve_search(&mut projector, &shared);
|
||||
assert_eq!(projector.progress_status(), IndexStatus::Behind { pending: 3 }, "{what}: not yet reconciled, and lag");
|
||||
assert_eq!(lock(&shared).searches.len(), 1, "{what}: the search waits for the first reconciliation");
|
||||
assert_eq!(projector.advance().unwrap(), Advance::Progressed);
|
||||
serve_search(&mut projector, &shared);
|
||||
assert_eq!(lock(&shared).searches.len(), 1, "{what}: and for the first writes");
|
||||
drain(&mut projector);
|
||||
serve_search(&mut projector, &shared);
|
||||
let guard = lock(&shared);
|
||||
assert!(matches!(guard.found.as_slice(), [(found, Ok(_))] if *found == ticket), "{what}: {:?}", guard.found);
|
||||
assert_eq!(index.state.lock().unwrap().held_when_searched, [3], "{what}");
|
||||
}
|
||||
|
||||
let entries = entries_of(4);
|
||||
let (embedder, index) = pair();
|
||||
projected(&embedder, &index, &entries[..3]);
|
||||
let mut partial = Projector::new(embedder.clone(), index.clone(), &slug(INSTANCE), Some(MIND.into()));
|
||||
partial.want(entries.iter().cloned());
|
||||
partial.advance().unwrap();
|
||||
assert_eq!(partial.progress_status(), IndexStatus::Behind { pending: 1 }, "an index found whole at start-up only lags");
|
||||
|
||||
embedder.state.lock().unwrap().identity.digest = "d2".into();
|
||||
let mut replaced = Projector::new(embedder, index, &slug(INSTANCE), Some(MIND.into()));
|
||||
replaced.want(entries);
|
||||
replaced.advance().unwrap();
|
||||
assert_eq!(replaced.progress_status(), IndexStatus::Reconciling { pending: 4 }, "an index that exists and is made again is a rebuild");
|
||||
}
|
||||
|
||||
/// A mind with no identity and documents owed has nothing to wait for: the
|
||||
/// search is taken and refused, not left queued for a worker that would spin
|
||||
/// on it.
|
||||
#[test]
|
||||
fn a_search_is_refused_when_documents_are_owed_to_a_mind_with_no_identity() {
|
||||
let (embedder, index) = pair();
|
||||
let mut projector = Projector::new(embedder.clone(), index.clone(), &slug(INSTANCE), None);
|
||||
projector.want(entries_of(2));
|
||||
let shared = shared_with(Vec::new(), None, vec![queued(1, "a")], false);
|
||||
serve_search(&mut projector, &shared);
|
||||
let guard = lock(&shared);
|
||||
assert!(guard.searches.is_empty());
|
||||
let [(SearchTicket(1), Err(why))] = guard.found.as_slice() else { panic!("{:?}", guard.found) };
|
||||
assert!(why.contains("being rebuilt") && why.contains("pending: 2"), "{why}");
|
||||
assert!(embedder.texts().is_empty() && index.state.lock().unwrap().searches.is_empty());
|
||||
}
|
||||
|
||||
/// Every call a search makes is bounded by what its time has left, not by the
|
||||
/// time it was given: after a slow identity read the embed's bound is smaller
|
||||
/// by what the read took, and the query's by both reads.
|
||||
#[test]
|
||||
fn a_later_calls_bound_shrinks_as_the_search_spends_its_time() {
|
||||
let entries = entries_of(3);
|
||||
let (embedder, index) = pair();
|
||||
let mut projector = projected(&embedder, &index, &entries);
|
||||
let delay = Duration::from_millis(150);
|
||||
let within = Duration::from_secs(10);
|
||||
{
|
||||
let mut state = embedder.state.lock().unwrap();
|
||||
state.identity_delay = delay;
|
||||
state.bounds.clear();
|
||||
state.identity_bounds.clear();
|
||||
}
|
||||
index.state.lock().unwrap().bounds.clear();
|
||||
projector.search("anything", 3, within).unwrap();
|
||||
|
||||
let embedder = embedder.state.lock().unwrap();
|
||||
let index = index.state.lock().unwrap();
|
||||
assert!(embedder.identity_bounds[0] <= within);
|
||||
assert!(embedder.bounds[0] <= within - delay, "the embed came after one slow read: {:?}", embedder.bounds);
|
||||
assert!(embedder.identity_bounds[1] <= within - delay, "{:?}", embedder.identity_bounds);
|
||||
let queried = index.bounds.iter().find(|(call, _)| *call == "search").unwrap().1;
|
||||
assert!(queried <= within - delay * 2, "the query came after two slow reads: {queried:?}");
|
||||
}
|
||||
|
||||
/// The deadline has one owner: the sink. It stamps every search with it, and
|
||||
/// it is what the serve loop is told (`Daemon::search_deadline`), so the loop's
|
||||
/// wait and the worker's bound are one value.
|
||||
#[test]
|
||||
fn the_sink_owns_the_searchs_deadline_and_tells_the_loop() {
|
||||
let (_root, mind) = mind_with_documents();
|
||||
let deadline = Duration::from_secs(7);
|
||||
let shared = shared_with(Vec::new(), None, Vec::new(), false);
|
||||
let mut sink = WorkerSink { shared: Arc::clone(&shared), identified: true, deadline };
|
||||
let asked = Instant::now();
|
||||
ask_sink(&mut sink, "anything", 3).unwrap();
|
||||
let due = lock(&shared).searches.back().expect("no worker took it").due;
|
||||
assert!(due >= asked + deadline && due <= Instant::now() + deadline, "stamped with the sink's deadline");
|
||||
assert_eq!(<WorkerSink as IndexSink<OwnedRedbMessagePackBackingStore>>::search_deadline(&sink), deadline);
|
||||
assert_eq!(Daemon::new(mind, sink).search_deadline(), deadline);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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, ®istry, &loop_stopping, &options));
|
||||
Self { endpoint, stopping, serving: Some(serving), embedder, index, _root: root }
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Reference in a new issue