diff --git a/crates/huginn-daemon/src/daemon.rs b/crates/huginn-daemon/src/daemon.rs index c42d644..8b831a3 100644 --- a/crates/huginn-daemon/src/daemon.rs +++ b/crates/huginn-daemon/src/daemon.rs @@ -70,6 +70,20 @@ impl Handled { } } +/// What the log says of the hits the mind held no document for: how many, and +/// the first few by name (a restored mind can leave hundreds), or nothing when +/// the mind held them all. +fn dropped_note(dropped: &[PipelineRef]) -> Option { + if dropped.is_empty() { + return None; + } + let named: Vec<&str> = dropped.iter().take(5).map(|hit| hit.id.0.as_str()).collect(); + Some(format!( + "huginn: the index returned {} ids this mind does not hold (first {named:?}); they are left out", + dropped.len() + )) +} + /// The envelope's `source_runtime_id` on every response a daemon serving this /// instance sends. Also what the socket is bound under, before a daemon exists. pub fn runtime_id(instance: &Slug) -> String { @@ -181,13 +195,8 @@ impl> Daemon { }; match self.mind.rank(&search.selection, &search.semantic, &hits) { Ok(ranked) => { - if !ranked.dropped.is_empty() { - let named: Vec<&str> = ranked.dropped.iter().take(5).map(|hit| hit.id.0.as_str()).collect(); - eprintln!( - "huginn: the index returned {} ids this mind does not hold (first {:?}); they are left out", - ranked.dropped.len(), - named - ); + if let Some(note) = dropped_note(&ranked.dropped) { + eprintln!("{note}"); } HuginnMindResponse::Query(ranked.page) } @@ -428,6 +437,19 @@ pub(crate) mod tests { (root, daemon) } + /// The log names how many hits the mind did not hold and the first five, and + /// says nothing when it held them all. + #[test] + fn the_dropped_hits_are_reported_by_count_and_the_first_five_names() { + assert_eq!(dropped_note(&[]), None); + let dropped: Vec = + (0..7).map(|n| PipelineRef { kind: PipelineKind::Question, id: Short(format!("gone-{n}")) }).collect(); + let note = dropped_note(&dropped).unwrap(); + assert!(note.contains("returned 7 ids"), "{note}"); + assert!(note.contains("gone-0") && note.contains("gone-4") && !note.contains("gone-5"), "{note}"); + assert_eq!(dropped_note(&dropped[..1]).unwrap().matches("gone-").count(), 1); + } + /// No index: for the tests whose subject is not the projection. pub(crate) struct NoIndex; diff --git a/crates/huginn-daemon/src/index/fakes.rs b/crates/huginn-daemon/src/index/fakes.rs index 3ed7560..ffb8570 100644 --- a/crates/huginn-daemon/src/index/fakes.rs +++ b/crates/huginn-daemon/src/index/fakes.rs @@ -24,7 +24,7 @@ impl Gate { /// Blocks until `count` calls to `embed` have arrived at the gate. pub(crate) fn wait_arrived(&self, count: u32) { - let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30); + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10); let mut state = self.0.0.lock().unwrap(); while state.0 < count { let left = deadline.saturating_duration_since(std::time::Instant::now()); diff --git a/crates/huginn-daemon/src/index/mod.rs b/crates/huginn-daemon/src/index/mod.rs index 1d7aaa6..c0980c8 100644 --- a/crates/huginn-daemon/src/index/mod.rs +++ b/crates/huginn-daemon/src/index/mod.rs @@ -699,7 +699,9 @@ fn work(mut projector: Projector, shared: Loc // When the worker next verifies the model and the collection whether or // not anything has woken it: an absolute time, moved only by a // verification, so a steady stream of searches cannot keep pushing it out. - let mut recheck_at = Instant::now() + backoff.recheck; + // The first step is a verification (the startup reconciliation), so it is + // due now. + let mut recheck_at = Instant::now(); loop { if !absorb(&mut projector, &shared, failing.as_ref().map(|(error, attempts)| (error, *attempts))) { return; @@ -811,7 +813,7 @@ impl WorkerSink { } /// The longest `WorkerSink::wait_status` waits, in all. -const WAIT_STATUS_MAX: Duration = Duration::from_secs(30); +const WAIT_STATUS_MAX: Duration = Duration::from_secs(10); impl Drop for WorkerSink { fn drop(&mut self) { diff --git a/crates/huginn-daemon/src/index/tests.rs b/crates/huginn-daemon/src/index/tests.rs index 6cce84f..2e7387c 100644 --- a/crates/huginn-daemon/src/index/tests.rs +++ b/crates/huginn-daemon/src/index/tests.rs @@ -59,8 +59,15 @@ fn counts(index: &FakeIndex) -> (usize, usize) { (state.recreated.len(), state.upserts.len()) } +/// Steps until the projector has nothing to do. A step that never settles is a +/// failure of the test, not a hang: it is bounded. fn drain(projector: &mut Projector) { - while projector.advance().unwrap() == Advance::Progressed {} + for _ in 0..1000 { + if projector.advance().unwrap() == Advance::Idle { + return; + } + } + panic!("the projector was still stepping after 1000 steps"); } fn point_ids(entries: &[IndexEntry]) -> BTreeSet { @@ -758,7 +765,7 @@ fn a_mind_without_an_identity_makes_no_collection_until_its_first_admission() { /// assertion: it polls for at most ten seconds and then says what never /// happened. fn eventually(what: &str, mut condition: impl FnMut() -> bool) { - for _ in 0..2000 { + for _ in 0..1000 { if condition() { return; } @@ -1349,7 +1356,7 @@ fn steady_searches_do_not_postpone_the_workers_recheck() { let listed = index.state.lock().unwrap().listed; let started = Instant::now(); while index.state.lock().unwrap().listed < listed + 2 { - assert!(started.elapsed() < Duration::from_secs(10), "the worker never rechecked while searches kept arriving"); + assert!(started.elapsed() < Duration::from_secs(5), "the worker never rechecked while searches kept arriving"); let _ = ask_sink(&mut sink, "anything", 3); collected(&mut sink); std::thread::sleep(Duration::from_millis(10)); @@ -1385,3 +1392,18 @@ fn wait_status_gives_up_at_its_overall_deadline_however_often_the_health_changes ticker.join().unwrap(); assert_eq!(gave_up, Ok(true), "the wait did not end at its deadline"); } + +/// An idle worker verifies once per interval, not in a loop: the recheck moves +/// its own deadline when it runs. +#[test] +fn an_idle_worker_rechecks_once_per_interval() { + let (_root, mind) = mind_with_documents(); + let entries = mind.index_entries(None).unwrap(); + let (embedder, index) = pair(); + let backoff = Backoff { recheck: Duration::from_millis(100), ..quick() }; + let sink = WorkerSink::spawn(embedder, index.clone(), &slug(INSTANCE), Some(MIND.into()), entries, backoff); + sink.wait_status(|status| *status == IndexStatus::Current); + std::thread::sleep(Duration::from_millis(600)); + let listed = index.state.lock().unwrap().listed; + assert!((2..=40).contains(&listed), "the startup reconciliation and about five rechecks listed the collection {listed} times"); +}