Pin what the mutant run left open; make failing tests fail fast

cargo-mutants on 11b's diff: three timeouts (a drain loop that never settled
under Projector::reconcile -> Ok(()), a dead worker leaving thirty-second
waits to add up past the harness deadline) and three survivors. Now:
- drain is bounded, the status and gate waits are ten seconds, so a mutant
  fails inside the harness deadline instead of hanging it;
- the worker's first step is due immediately rather than one interval after
  a startup that is itself a verification, and an idle worker is pinned to
  one recheck per interval (a recheck that did not move its own deadline
  would list the collection in a loop);
- the log line for hits the mind does not hold is a function with a test.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-29 16:54:18 +02:00
commit 2052d314e3
4 changed files with 59 additions and 13 deletions

View file

@ -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<String> {
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<S: MindStore, I: IndexSink<S>> Daemon<S, I> {
};
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<PipelineRef> =
(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;

View file

@ -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());

View file

@ -699,7 +699,9 @@ fn work<E: Embedder, V: VectorIndex>(mut projector: Projector<E, V>, 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) {

View file

@ -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<E: Embedder, V: VectorIndex>(projector: &mut Projector<E, V>) {
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<String> {
@ -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");
}