diff --git a/crates/huginn-daemon/src/daemon.rs b/crates/huginn-daemon/src/daemon.rs index 8b831a3..ffbf813 100644 --- a/crates/huginn-daemon/src/daemon.rs +++ b/crates/huginn-daemon/src/daemon.rs @@ -25,6 +25,9 @@ pub trait IndexSink { /// 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; + /// 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)>; @@ -174,6 +177,11 @@ impl> Daemon { } } + /// 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)> { 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)> { 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)> { 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)> { Vec::new() } diff --git a/crates/huginn-daemon/src/index/fakes.rs b/crates/huginn-daemon/src/index/fakes.rs index 61c98a2..20bcc8c 100644 --- a/crates/huginn-daemon/src/index/fakes.rs +++ b/crates/huginn-daemon/src/index/fakes.rs @@ -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>> { diff --git a/crates/huginn-daemon/src/index/mod.rs b/crates/huginn-daemon/src/index/mod.rs index 367e675..30a7b4c 100644 --- a/crates/huginn-daemon/src/index/mod.rs +++ b/crates/huginn-daemon/src/index/mod.rs @@ -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 { @@ -294,6 +314,7 @@ pub struct Projector { /// The model the collection was reconciled under; `None` until it is, and /// again whenever that reconciliation is forgotten. reconciled: Option, + build: Build, } impl Projector { @@ -307,6 +328,7 @@ impl Projector { known: BTreeMap::new(), wanted: BTreeSet::new(), reconciled: None, + build: Build::Unbuilt, } } @@ -398,7 +420,12 @@ impl Projector { 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 Projector { 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 Projector { } } - 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 Projector { for entry in &batch { self.wanted.remove(&entry.id.id.0); } + self.settle(); Ok(()) } @@ -537,7 +579,7 @@ impl Projector { } 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 Projector { 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 Projector { /// 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 { /// 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(projector: &mut Projector, 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(projector: &Projector, guard: &mut Shared) -> Option { + 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( 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( + projector: &mut Projector, + 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 IndexSink 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)> { std::mem::take(&mut lock(&self.shared).found) } diff --git a/crates/huginn-daemon/src/index/tests.rs b/crates/huginn-daemon/src/index/tests.rs index 76c3ddd..17ead91 100644 --- a/crates/huginn-daemon/src/index/tests.rs +++ b/crates/huginn-daemon/src/index/tests.rs @@ -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>> = 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 { + 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, 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!(>::search_deadline(&sink), deadline); + assert_eq!(Daemon::new(mind, sink).search_deadline(), deadline); +} diff --git a/crates/huginn-daemon/src/semantic_tests.rs b/crates/huginn-daemon/src/semantic_tests.rs index 9a5646b..513f0e9 100644 --- a/crates/huginn-daemon/src/semantic_tests.rs +++ b/crates/huginn-daemon/src/semantic_tests.rs @@ -140,7 +140,7 @@ fn ids(response: &HuginnMindResponse) -> Vec { } 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 = 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 } } diff --git a/crates/huginn-daemon/src/serve.rs b/crates/huginn-daemon/src/serve.rs index 5af8589..a5efc02 100644 --- a/crates/huginn-daemon/src/serve.rs +++ b/crates/huginn-daemon/src/serve.rs @@ -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>( 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>( fn collect_searches>( daemon: &mut Daemon, waiting: &BTreeMap, - options: &ServeOptions, ) -> Vec<(SearchTicket, Result)> { + 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