From 2691645ee5a436dd7425b894e58625d1736d33e9 Mon Sep 17 00:00:00 2001 From: HackPoint <6758579+HackPoint@users.noreply.github.com> Date: Mon, 12 Oct 2026 14:37:18 +0300 Subject: [PATCH 1/2] test(daemon): the vanished-transcript tests wait for the watcher and the faults they assert on They took the first ingested turn as proof that the notify watcher was registered, made one directory, and read the first fault in the whole spool. Under load that directory could land before the watcher existed and never be reported, and a ws_restart from a lost port race could be spooled ahead of it. The tests now make a fresh directory per check until one is reported, filter the spool to ingest_failed faults for the paths they made, assert that no such fault names a vanished path, and wait on the turn rows they assert on instead of a running total. --- crates/lumen-daemon/tests/e2e_pipeline.rs | 152 ++++++++++++---------- 1 file changed, 84 insertions(+), 68 deletions(-) diff --git a/crates/lumen-daemon/tests/e2e_pipeline.rs b/crates/lumen-daemon/tests/e2e_pipeline.rs index fc907c5..648affe 100644 --- a/crates/lumen-daemon/tests/e2e_pipeline.rs +++ b/crates/lumen-daemon/tests/e2e_pipeline.rs @@ -163,11 +163,54 @@ fn turn_count(db: &std::path::Path) -> i64 { .unwrap_or(0) } +/// Whether the turn with this message id is in the ledger. +fn has_turn(db: &std::path::Path, message_id: &str) -> bool { + let Ok(conn) = lumen_core::meter::connect_db(db) else { + return false; + }; + conn.query_row( + "SELECT COUNT(*) FROM turns WHERE message_id = ?1", + [message_id], + |r| r.get::<_, i64>(0), + ) + .is_ok_and(|n| n > 0) +} + /// The faults the daemon has spooled, in the order it recorded them. fn spooled_faults(spool: &std::path::Path) -> Vec { lumen_core::faults::peek_spool_at(spool).expect("read the fault spool") } +/// The `ingest_failed` faults spooled for a file whose name satisfies `named`, oldest first. +/// +/// A daemon files other faults too (a WebSocket port a neighbouring daemon got first, say), and +/// those say nothing about a transcript. +fn ingest_failures( + spool: &std::path::Path, + named: impl Fn(&str) -> bool, +) -> Vec { + spooled_faults(spool) + .into_iter() + .filter(|f| f.kind == "ingest_failed") + .filter(|f| { + f.path + .as_deref() + .and_then(|p| std::path::Path::new(p).file_name()) + .and_then(|n| n.to_str()) + .is_some_and(&named) + }) + .collect() +} + +/// A directory with a transcript's name, which the daemon cannot read: refused at open on +/// Windows, at read elsewhere, and skipped by the poll, so only the watcher reports it. +/// +/// Numbered so a test can keep making new ones: the watcher reports only what happens after it +/// is registered, and a fault for one of these is the first sign that it is. +fn make_unreadable(projects: &std::path::Path, n: usize) { + std::fs::create_dir(projects.join(format!("unreadable-{n}.jsonl"))).unwrap(); +} + // Each test gets its own HOME, its own DB and its own WS port, so they are safe // to run in parallel and cannot see a daemon running outside the test suite. @@ -545,53 +588,52 @@ async fn a_snapshot_labels_the_session_with_its_project() { #[tokio::test] async fn a_transcript_deleted_before_the_daemon_reads_it_is_not_a_fault() { let fx = start_daemon().await; - let db = fx.db.clone(); - let spool = fx.spool.clone(); - let gone = fx.projects.join("gone.jsonl"); - append( - &gone, - &[transcript_line( - "gone-1", - "sess-g", - "2026-01-01T10:00:00Z", - (1, 1, 1, 1), - )], - ); - wait_until("the turn", || turn_count(&db) == 1).await; - - // The watcher reports the deletion, and the daemon opens a path that is no longer there. - std::fs::remove_file(&gone).unwrap(); - // Watcher events are handled in order, so the fault on this directory is filed after - // whatever the deletion produced. The daemon records one fault per variant a minute, so - // the first one spooled says which of the two it was. - std::fs::create_dir(fx.projects.join("unreadable.jsonl")).unwrap(); - wait_until("the unreadable directory to be reported", || { - !spooled_faults(&spool).is_empty() + // A round deletes a transcript the moment it is written, so the watcher reports a path the + // daemon then finds gone, and makes a directory the daemon does file a fault for. Events are + // handled in order, and the daemon records one fault per variant a minute, so the first one + // spooled says whether the deletion was one. A round that ran before the watcher was + // registered is lost, so rounds repeat, with names of their own, until one is heard. + let mut round = 0; + wait_until("a fault for the directory made after a deletion", || { + round += 1; + let gone = fx.projects.join(format!("gone-{round}.jsonl")); + append( + &gone, + &[transcript_line( + &format!("gone-{round}"), + "sess-g", + "2026-01-01T10:00:00Z", + (1, 1, 1, 1), + )], + ); + std::fs::remove_file(&gone).unwrap(); + make_unreadable(&fx.projects, round); + !ingest_failures(&fx.spool, |n| { + n.starts_with("gone-") || n.starts_with("unreadable-") + }) + .is_empty() }) .await; - let faults = spooled_faults(&spool); + let faults = ingest_failures(&fx.spool, |n| n.starts_with("gone-")); assert!( - faults[0] - .path - .as_deref() - .is_some_and(|p| p.ends_with("unreadable.jsonl")), + faults.is_empty(), "a transcript deleted before it was read was filed as a fault: {faults:#?}" ); - // The deletion left the daemon ingesting. + // The deletions left the daemon ingesting. append( &fx.projects.join("next.jsonl"), &[transcript_line( - "gone-2", + "after-deletion", "sess-g", "2026-01-01T10:05:00Z", (1, 1, 1, 1), )], ); - wait_until("a transcript written after the deletion", || { - turn_count(&db) == 2 + wait_until("a transcript written after the deletions", || { + has_turn(&fx.db, "after-deletion") }) .await; } @@ -601,43 +643,17 @@ async fn a_transcript_the_daemon_cannot_read_is_still_filed_as_a_fault() { // Only a missing file is excused. Any other reason the daemon cannot read a transcript is // what the fault report is for. let fx = start_daemon().await; - let db = fx.db.clone(); - let spool = fx.spool.clone(); - - // A turn first, so the daemon is past its startup pass before the directory appears. - append( - &fx.projects.join("first.jsonl"), - &[transcript_line( - "first-1", - "sess-f", - "2026-01-01T10:00:00Z", - (1, 1, 1, 1), - )], - ); - wait_until("the turn", || turn_count(&db) == 1).await; - // A directory with a transcript's name: refused at open on Windows, at read elsewhere, - // and skipped by the poll, so only the watcher reports it. - std::fs::create_dir(fx.projects.join("unreadable.jsonl")).unwrap(); - wait_until("the unreadable directory to be reported", || { - !spooled_faults(&spool).is_empty() + let mut made = 0; + wait_until("a fault for an unreadable directory", || { + made += 1; + make_unreadable(&fx.projects, made); + !ingest_failures(&fx.spool, |n| n.starts_with("unreadable-")).is_empty() }) .await; - let faults = spooled_faults(&spool); - let first = &faults[0]; - assert_eq!( - (first.kind.as_str(), first.variant.as_str()), - ("ingest_failed", "notify"), - "{faults:#?}" - ); - assert!( - first - .path - .as_deref() - .is_some_and(|p| p.ends_with("unreadable.jsonl")), - "{faults:#?}" - ); + let faults = ingest_failures(&fx.spool, |n| n.starts_with("unreadable-")); + assert_eq!(faults[0].variant, "notify", "{faults:#?}"); } #[cfg(unix)] @@ -657,22 +673,22 @@ async fn a_listed_transcript_that_is_gone_when_opened_is_not_a_fault() { .unwrap(); }) .await; - let db = fx.db.clone(); for i in 1..=2 { + let id = format!("poll-{i}"); append( &target, &[transcript_line( - &format!("poll-{i}"), + &id, "sess-poll", "2026-01-01T10:00:00Z", (1, 1, 1, 1), )], ); - wait_until(&format!("{i} turns"), || turn_count(&db) == i).await; + wait_until(&format!("the turn {id}"), || has_turn(&fx.db, &id)).await; } - let faults = spooled_faults(&fx.spool); + let faults = ingest_failures(&fx.spool, |n| n == "gone.jsonl"); assert!( faults.is_empty(), "a transcript that was gone when the daemon opened it was filed as a fault: {faults:#?}" From fea220fa0c1ff33361ad98b4a0ee61031b3256f0 Mon Sep 17 00:00:00 2001 From: HackPoint <6758579+HackPoint@users.noreply.github.com> Date: Mon, 12 Oct 2026 17:57:54 +0300 Subject: [PATCH 2/2] test(daemon): the vanished-transcript helpers carry one-line comments The three comments the review found longer than the repository's one-line rule are cut to one line each; no code changes. --- crates/lumen-daemon/tests/e2e_pipeline.rs | 17 +++-------------- 1 file changed, 3 insertions(+), 14 deletions(-) diff --git a/crates/lumen-daemon/tests/e2e_pipeline.rs b/crates/lumen-daemon/tests/e2e_pipeline.rs index 648affe..f4d73dc 100644 --- a/crates/lumen-daemon/tests/e2e_pipeline.rs +++ b/crates/lumen-daemon/tests/e2e_pipeline.rs @@ -181,10 +181,7 @@ fn spooled_faults(spool: &std::path::Path) -> Vec bool, @@ -202,11 +199,7 @@ fn ingest_failures( .collect() } -/// A directory with a transcript's name, which the daemon cannot read: refused at open on -/// Windows, at read elsewhere, and skipped by the poll, so only the watcher reports it. -/// -/// Numbered so a test can keep making new ones: the watcher reports only what happens after it -/// is registered, and a fault for one of these is the first sign that it is. +/// A directory with a transcript's name: the daemon cannot read it, and only the watcher reports it. fn make_unreadable(projects: &std::path::Path, n: usize) { std::fs::create_dir(projects.join(format!("unreadable-{n}.jsonl"))).unwrap(); } @@ -589,11 +582,7 @@ async fn a_snapshot_labels_the_session_with_its_project() { async fn a_transcript_deleted_before_the_daemon_reads_it_is_not_a_fault() { let fx = start_daemon().await; - // A round deletes a transcript the moment it is written, so the watcher reports a path the - // daemon then finds gone, and makes a directory the daemon does file a fault for. Events are - // handled in order, and the daemon records one fault per variant a minute, so the first one - // spooled says whether the deletion was one. A round that ran before the watcher was - // registered is lost, so rounds repeat, with names of their own, until one is heard. + // Rounds repeat under new names until one is heard: a round before the watcher registers is lost. let mut round = 0; wait_until("a fault for the directory made after a deletion", || { round += 1;