From 22c18c36bd78f688adca01d2aad0859228190201 Mon Sep 17 00:00:00 2001 From: Jake Magar Date: Tue, 25 Aug 2026 09:32:36 -0400 Subject: [PATCH 1/2] test: stop process-global test state from failing unrelated tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `cargo test` intermittently failed 7-9 tests in inventory, setup::doctor, and agent_observatory that had nothing to do with whatever was being changed. The failing set varied run to run, and CI never saw any of it, so the failures read as "your branch broke something" to whoever hit them. Three separate causes, all process-global state that `cargo nextest` hides by giving each test its own process. PATH override leaked. `container_probe_reports_unreachable_when_docker_ps_fails` (#205) replaced PATH with a bare tempdir and never restored it, so every later bare-name spawn in the binary resolved against a deleted directory and failed ENOENT. This reproduces at --test-threads=1, so it is a leak, not a race, and `#[serial]` alone would not have fixed it. Now prepends and restores via the file's own EnvGuard, and is `#[serial]` so it cannot collide with the three docker-stubbing tests in setup_tests.rs. PATH replacement had no scoped alternative. `collect_warns_when_optional_device_ commands_are_missing` legitimately wiped PATH to assert `ip`/`ss`/`df` are not installed — which took every concurrent test's spawns down with it. Adding `env::mask_test_programs`, which makes named programs resolve to an absent path so they spawn with the same NotFound an uninstalled binary gives. The test now says what it means, and no test in the tree replaces PATH any more. Projector tests asserted per-cycle values. `oversized_first_rows=1` and `"attempts":2` are overwritten by the projector's next cycle, so each was true for roughly one 10ms window and false forever after — no timeout could fix that, and `notify_projection_work` broadcasts on a process-global channel that any test's `insert_logs_batch` rings, forcing extra cycles. They now assert monotone facts: attempts >= 2, durable cursors, and that health reports the counter rather than what it currently reads. Deadlines also moved to named constants documenting the shared write lock, since every test pool's `init_pool` migrates while holding it. Verified on this tree: `cargo test --no-fail-fast` 2471/2471 (baseline 79adf1f4: 8 failures) and `cargo nextest run` 3061/3061. Refs: syslog-mcp-g4frk --- src/env.rs | 140 ++++++++++++++++++++++++- src/heartbeat_agent_tests.rs | 13 ++- src/inventory/device_tests.rs | 15 ++- src/runtime/agent_observatory_tests.rs | 100 ++++++++++++++---- 4 files changed, 243 insertions(+), 25 deletions(-) diff --git a/src/env.rs b/src/env.rs index a7cd6219..b900da1f 100644 --- a/src/env.rs +++ b/src/env.rs @@ -8,7 +8,7 @@ use std::ffi::{OsStr, OsString}; #[cfg(any(test, feature = "test-support"))] -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; #[cfg(any(test, feature = "test-support"))] use std::sync::{OnceLock, RwLock}; @@ -79,7 +79,80 @@ pub fn remove_test_var>(key: K) { .insert(key.as_ref().to_os_string(), None); } -/// Resolve an executable only when tests explicitly override `PATH`. +// Program names that must resolve as missing, regardless of what is really on +// `PATH`. See [`mask_test_programs`]. +#[cfg(any(test, feature = "test-support"))] +static MASKED_PROGRAMS: OnceLock>> = OnceLock::new(); + +/// Directory component that cannot exist, so a masked program spawns with the +/// same `ErrorKind::NotFound` a genuinely absent binary would produce. Any path +/// with more than one component bypasses the platform's `PATH` search entirely. +#[cfg(any(test, feature = "test-support"))] +const MASKED_PROGRAM_DIR: &str = "cortex-test-masked-program-does-not-exist"; + +/// Make specific bare program names resolve as missing until the returned guard +/// is dropped. +/// +/// This exists so a test asserting "this command is not installed" does not have +/// to replace `PATH` to say so. The override map is process-global, so replacing +/// `PATH` with a fixture directory takes every *other* test's subprocess spawns +/// down with it — they stop finding `sh`, `git`, and everything else — and under +/// plain `cargo test` those tests run concurrently. Masking states the actual +/// intent instead, and scopes the blast radius to the named programs. +/// +/// The mask is still process-global: only mask programs whose spawning tests are +/// serialized against this one. +#[cfg(any(test, feature = "test-support"))] +#[doc(hidden)] +#[must_use = "the mask is lifted when the guard is dropped"] +pub fn mask_test_programs(names: I) -> MaskedPrograms +where + I: IntoIterator, + S: AsRef, +{ + let names: Vec = names + .into_iter() + .map(|name| name.as_ref().to_os_string()) + .collect(); + let mut masked = MASKED_PROGRAMS + .get_or_init(|| RwLock::new(HashSet::new())) + .write() + .expect("masked program set lock poisoned"); + for name in &names { + masked.insert(name.clone()); + } + drop(masked); + MaskedPrograms(names) +} + +/// Guard returned by [`mask_test_programs`]; lifts the mask on drop. +#[cfg(any(test, feature = "test-support"))] +#[doc(hidden)] +pub struct MaskedPrograms(Vec); + +#[cfg(any(test, feature = "test-support"))] +impl Drop for MaskedPrograms { + fn drop(&mut self) { + if let Some(masked) = MASKED_PROGRAMS.get() { + let mut masked = masked.write().expect("masked program set lock poisoned"); + for name in &self.0 { + masked.remove(name); + } + } + } +} + +#[cfg(any(test, feature = "test-support"))] +fn program_is_masked(program: &OsStr) -> bool { + MASKED_PROGRAMS.get().is_some_and(|masked| { + masked + .read() + .expect("masked program set lock poisoned") + .contains(program) + }) +} + +/// Resolve an executable when tests mask it or explicitly override `PATH`. /// Normal test-support builds keep the platform's native command lookup semantics. #[cfg(any(test, feature = "test-support"))] fn resolve_test_program(program: &OsStr) -> Option { @@ -87,6 +160,9 @@ fn resolve_test_program(program: &OsStr) -> Option { if program_path.components().count() != 1 { return None; } + if program_is_masked(program) { + return Some(std::path::Path::new(MASKED_PROGRAM_DIR).join(program)); + } let search_path = test_override(OsStr::new("PATH"))??; std::env::split_paths(&search_path) .map(|dir| dir.join(program)) @@ -180,6 +256,66 @@ mod tests { assert!(var_os(KEY).is_none()); } + /// Masking must make a program that *is* on `PATH` resolve to something that + /// cannot be spawned, and must lift cleanly — otherwise it becomes the same + /// process-global leak that replacing `PATH` was. + #[cfg(unix)] + #[test] + fn masking_hides_a_resolvable_program_and_lifts_on_drop() { + // Restore the effective PATH: `remove_test_var` masks the key outright, + // which would leave every later test in this binary with no PATH at all. + struct PathGuard(Option); + impl Drop for PathGuard { + fn drop(&mut self) { + match self.0.take() { + Some(previous) => set_test_var("PATH", previous), + None => remove_test_var("PATH"), + } + } + } + + // A name nothing else in this binary spawns, so the process-global mask + // cannot disturb a concurrently running test. + const PROGRAM: &str = "cortex-test-mask-probe"; + + let dir = tempfile::tempdir().unwrap(); + let binary = dir.path().join(PROGRAM); + std::fs::write(&binary, "#!/bin/sh\nexit 0\n").unwrap(); + use std::os::unix::fs::PermissionsExt; + std::fs::set_permissions(&binary, std::fs::Permissions::from_mode(0o755)).unwrap(); + + // Prepend, never replace: a bare fixture PATH breaks unrelated spawns. + let mut search_path = vec![dir.path().to_path_buf()]; + if let Some(existing) = var_os("PATH") { + search_path.extend(std::env::split_paths(&existing)); + } + let _path = PathGuard(var_os("PATH")); + set_test_var("PATH", std::env::join_paths(search_path).unwrap()); + + assert_eq!( + resolve_test_program(OsStr::new(PROGRAM)).as_deref(), + Some(binary.as_path()), + "the fixture directory should resolve the program before masking" + ); + + { + let _masked = mask_test_programs([PROGRAM]); + let resolved = resolve_test_program(OsStr::new(PROGRAM)) + .expect("a masked program still resolves, to a path that cannot exist"); + assert!( + !resolved.exists(), + "masking must resolve to an absent path so the spawn fails with \ + NotFound, exactly as an uninstalled binary does" + ); + } + + assert_eq!( + resolve_test_program(OsStr::new(PROGRAM)).as_deref(), + Some(binary.as_path()), + "dropping the guard must lift the mask" + ); + } + #[test] fn unrelated_keys_do_not_interfere() { const LEFT: &str = "CORTEX_TEST_ENV_OVERLAY_LEFT"; diff --git a/src/heartbeat_agent_tests.rs b/src/heartbeat_agent_tests.rs index c8a0d3d6..6085ad82 100644 --- a/src/heartbeat_agent_tests.rs +++ b/src/heartbeat_agent_tests.rs @@ -720,6 +720,7 @@ fn transcript_forward_env_does_not_gate_local_sessions_watch_service() { /// dead or permission-denied daemon — a CI container, for instance — reported a /// probe error instead. Both are the same fact about the host. #[tokio::test] +#[serial] async fn container_probe_reports_unreachable_when_docker_ps_fails() { let dir = tempfile::tempdir().unwrap(); let fake_docker = dir.path().join("docker"); @@ -733,7 +734,17 @@ async fn container_probe_reports_unreachable_when_docker_ps_fails() { use std::os::unix::fs::PermissionsExt; std::fs::set_permissions(&fake_docker, std::fs::Permissions::from_mode(0o755)).unwrap(); } - crate::env::set_test_var("PATH", dir.path().to_str().unwrap()); + // `PATH` overrides live in a process-global map that every subprocess spawn + // in this test binary resolves against, so this has to both prepend rather + // than replace and restore on drop. A bare, unrestored fixture directory + // outlives the test and leaves every later `sh`, `git`, or `docker` spawn + // in the binary resolving against a deleted temp dir. `cargo nextest` hides + // that behind one process per test; plain `cargo test` does not. + let mut search_path = vec![dir.path().to_path_buf()]; + if let Some(existing) = crate::env::var_os("PATH") { + search_path.extend(std::env::split_paths(&existing)); + } + let _path = EnvGuard::set("PATH", std::env::join_paths(search_path).unwrap()); let output = LinuxContainerProbe.collect().await.expect( "a failing `docker ps` is an unreachable runtime, not a probe error — \ diff --git a/src/inventory/device_tests.rs b/src/inventory/device_tests.rs index e4c4db56..d688065c 100644 --- a/src/inventory/device_tests.rs +++ b/src/inventory/device_tests.rs @@ -117,7 +117,20 @@ async fn collect_warns_when_optional_device_commands_are_missing() { let bin_dir = dir.path().join("bin"); std::fs::create_dir_all(&bin_dir).unwrap(); executable_file(&bin_dir.join("hostname"), "#!/bin/sh\nprintf '\\n'\n"); - let _path_guard = EnvGuard::set("PATH", bin_dir.as_os_str()); + // Mask the optional commands instead of replacing `PATH` with `bin_dir`. + // `PATH` overrides are process-global, so wiping it here also stopped every + // concurrently-running test in this binary from finding `sh`, `git`, and + // friends. Masking says what this test actually means — these four commands + // are not installed — and leaves every other program resolving normally. + // Safe to mask process-wide because `inventory::device` is the only caller + // of them and its tests are `#[serial]` with each other. + let _masked = crate::env::mask_test_programs(["uname", "ip", "ss", "df"]); + let path = format!( + "{}:{}", + bin_dir.display(), + crate::env::var("PATH").unwrap_or_default() + ); + let _path_guard = EnvGuard::set("PATH", path); let output = collect(std::time::Duration::from_millis(50)).await; diff --git a/src/runtime/agent_observatory_tests.rs b/src/runtime/agent_observatory_tests.rs index f7d92bee..b6066481 100644 --- a/src/runtime/agent_observatory_tests.rs +++ b/src/runtime/agent_observatory_tests.rs @@ -11,6 +11,50 @@ fn pool() -> (tempfile::TempDir, Arc) { (directory, pool) } +/// Deadline for waits on a background worker making progress. +/// +/// Every `DbPool` in this test binary shares one process-wide SQLite write lock +/// (`crate::db::pool::write_lock`), and each test pool's `init_pool` runs the +/// full migration set — VACUUM, CREATE INDEX, ANALYZE — while holding it. Under +/// full-suite parallelism a worker's cursor or health write therefore queues +/// behind however many other tests are migrating, which is seconds, not +/// milliseconds. `cargo nextest` gives each test its own process and never sees +/// this contention; plain `cargo test` does, and the old 2s/5s deadlines flaked +/// there. +/// +/// These are liveness backstops for a genuinely stuck worker, not latency +/// assertions — a passing run exits as soon as the condition holds, so the +/// headroom is free. Anything that needs to assert *timing* must do it against a +/// deliberately separated interval, the way the wake test uses `IDLE_POLL_MS`. +const PROGRESS_TIMEOUT: Duration = Duration::from_secs(60); + +/// Deadline for a cancelled worker to unwind. The same contention applies: an +/// in-flight write may still be queued behind the shared write lock. +const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(30); + +/// Poll interval high enough that the projector's fallback timer never fires +/// during a test, so it runs its first cycle and then idles. Observing progress +/// within `PROGRESS_TIMEOUT` therefore means something woke it, not that the +/// timer came round. +const IDLE_POLL_MS: u64 = 600_000; + +/// Reads `attempts` out of a projector health row. +/// +/// `record_projection_health` bumps this on every health write, so it only ever +/// climbs — which is the whole point. The projector's other health fields +/// (`projected`, `oversized_first_rows`) are per-cycle snapshots, and +/// `notify_projection_work` broadcasts on a **process-global** channel that every +/// `insert_logs_batch` in this test binary rings. Any test can therefore drive +/// any other test's projector through another cycle at any moment, overwriting a +/// per-cycle value microseconds after it appears. Assert on monotone facts — +/// this counter, and the durable cursors — never on a per-cycle snapshot. +fn projection_attempts(health: &str) -> u64 { + serde_json::from_str::(health) + .ok() + .and_then(|value| value.get("attempts").and_then(serde_json::Value::as_u64)) + .unwrap_or_default() +} + #[test] fn projector_sqlite_retry_backoff_reuses_ingest_policy_and_caps() { assert_eq!(projector_retry_delay_ms(1), 25); @@ -49,7 +93,7 @@ fn enabled_projector_advances_durable_log_cursor_and_shuts_down() { .unwrap(); let config = AgentObservatoryConfig { enabled: true, - projector_poll_ms: 10, + projector_poll_ms: IDLE_POLL_MS, projector_page_bytes: 1, ..AgentObservatoryConfig::default() }; @@ -60,7 +104,7 @@ fn enabled_projector_advances_durable_log_cursor_and_shuts_down() { .unwrap(); runtime.block_on(async { let handle = spawn_projector(token.clone(), Arc::clone(&pool), config).unwrap(); - tokio::time::timeout(Duration::from_secs(2), async { + tokio::time::timeout(PROGRESS_TIMEOUT, async { loop { if projection_cursor(&pool, "logs").unwrap() == "1" { break; @@ -78,23 +122,31 @@ fn enabled_projector_advances_durable_log_cursor_and_shuts_down() { ] { assert_eq!(projection_cursor(&pool, source).unwrap(), ""); } - // Parallel DB tests share Cortex's process-wide SQLite write lock; under - // suite load health persistence can legitimately queue behind other writers. - tokio::time::timeout(Duration::from_secs(5), async { + // The cursor reaching "1" above is what proves the page-bytes guard does + // not stall on a first row bigger than the whole budget — with + // projector_page_bytes = 1 and a 16-byte row, a stalling guard never + // advances it. Health is only checked for *reporting* the counter; + // asserting the count itself would race the global wake broadcast (see + // `projection_attempts`). + // Wait for the health row to exist rather than reading straight after the + // cursor wait: the cursor is written during projection, health only at + // the end of the cycle. Existence is monotone, so this cannot flap. + let health = tokio::time::timeout(PROGRESS_TIMEOUT, async { loop { - if projection_health(&pool, "projector") - .unwrap() - .is_some_and(|health| health.contains("oversized_first_rows=1")) - { - break; + if let Some(health) = projection_health(&pool, "projector").unwrap() { + return health; } tokio::time::sleep(Duration::from_millis(10)).await; } }) .await - .unwrap(); + .expect("a completed cycle records health"); + assert!( + health.contains("oversized_first_rows="), + "projector health must report the oversized-first-row counter: {health}" + ); token.cancel(); - tokio::time::timeout(Duration::from_secs(1), handle) + tokio::time::timeout(SHUTDOWN_TIMEOUT, handle) .await .unwrap() .unwrap(); @@ -106,7 +158,7 @@ fn projector_wakes_on_committed_log_ingest_before_fallback_poll() { let (_directory, pool) = pool(); let config = AgentObservatoryConfig { enabled: true, - projector_poll_ms: 60_000, + projector_poll_ms: IDLE_POLL_MS, ..AgentObservatoryConfig::default() }; let token = CancellationToken::new(); @@ -116,7 +168,7 @@ fn projector_wakes_on_committed_log_ingest_before_fallback_poll() { .unwrap(); runtime.block_on(async { let handle = spawn_projector(token.clone(), Arc::clone(&pool), config).unwrap(); - tokio::time::timeout(Duration::from_secs(2), async { + tokio::time::timeout(PROGRESS_TIMEOUT, async { loop { if projection_health(&pool, "projector").unwrap().is_some() { break; @@ -154,7 +206,10 @@ fn projector_wakes_on_committed_log_ingest_before_fallback_poll() { ) .unwrap(); - tokio::time::timeout(Duration::from_secs(1), async { + // Still proves the wake came from the committed ingest rather than the + // fallback poll: PROGRESS_TIMEOUT is an order of magnitude below + // IDLE_POLL_MS. + tokio::time::timeout(PROGRESS_TIMEOUT, async { loop { if projection_cursor(&pool, "logs").unwrap() == "1" { break; @@ -166,7 +221,7 @@ fn projector_wakes_on_committed_log_ingest_before_fallback_poll() { .expect("committed ingest should wake projector without waiting for fallback poll"); token.cancel(); - tokio::time::timeout(Duration::from_secs(1), handle) + tokio::time::timeout(SHUTDOWN_TIMEOUT, handle) .await .unwrap() .unwrap(); @@ -213,7 +268,7 @@ fn enabled_git_worker_records_progress_and_shuts_down() { .unwrap(); runtime.block_on(async { let handle = spawn_git_reconcile(token.clone(), Arc::clone(&pool), config).unwrap(); - tokio::time::timeout(Duration::from_secs(5), async { + tokio::time::timeout(PROGRESS_TIMEOUT, async { loop { if !projection_cursor(&pool, "git").unwrap().is_empty() { break; @@ -224,7 +279,7 @@ fn enabled_git_worker_records_progress_and_shuts_down() { .await .unwrap(); token.cancel(); - tokio::time::timeout(Duration::from_secs(1), handle) + tokio::time::timeout(SHUTDOWN_TIMEOUT, handle) .await .unwrap() .unwrap(); @@ -248,12 +303,15 @@ fn projector_failure_is_queryable_retries_without_advancing_and_stops_after_canc .unwrap(); runtime.block_on(async { let handle = spawn_projector(token.clone(), Arc::clone(&pool), config).unwrap(); - tokio::time::timeout(Duration::from_secs(10), async { + tokio::time::timeout(PROGRESS_TIMEOUT, async { loop { let health = projection_health(&pool, "projector") .unwrap() .unwrap_or_default(); - if health.contains("\"status\":\"error\"") && health.contains("\"attempts\":2") { + // `>= 2`, not `== 2`: attempts only climbs, so this latches once + // true. Pinning it to an exact value made the assertion true for + // one ~10ms retry window and false forever after. + if health.contains("\"status\":\"error\"") && projection_attempts(&health) >= 2 { assert!(health.contains("retry_safe=false")); assert!(health.contains("retry_delay_ms=0")); break; @@ -265,7 +323,7 @@ fn projector_failure_is_queryable_retries_without_advancing_and_stops_after_canc .unwrap(); assert_eq!(projection_cursor(&pool, "logs").unwrap(), "invalid"); token.cancel(); - tokio::time::timeout(Duration::from_secs(5), handle) + tokio::time::timeout(SHUTDOWN_TIMEOUT, handle) .await .unwrap() .unwrap(); From 3ce68481e97238e43d531230701756d6a97d5fc5 Mon Sep 17 00:00:00 2001 From: Jake Magar Date: Tue, 25 Aug 2026 01:27:03 -0400 Subject: [PATCH 2/2] fix(tests): wait on a projector progress signal instead of a wall-clock deadline MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `projector_failure_is_queryable_retries_without_advancing_and_stops_after_cancel` failed under `cargo test --lib` while passing in isolation. Captured tracing from a reproduction shows why: the projector's *first* cycle took ~23s (cursor read at 04:59:25.473, cycle completed at 04:59:43.169), so the test's 10s deadline expired before a second retry cycle could ever be recorded. A cycle has no wall-clock bound. It acquires the process-wide SQLite write serialization lock (`db::pool::write_lock`) several times — four source-cursor initialisations plus the health write — and under full-suite parallelism every test's pool queues on that one lock. Nothing about the projector is wrong; the tests were asserting a scheduling property they cannot guarantee. Publish `ProjectorProgress` from the projector on a `watch` channel after each cycle and have the tests await that signal instead. All fields are cumulative and monotonic, so a late observer cannot miss a transition — which matters because the projector only yields when every select branch is pending and can therefore run several cycles between two polls of a waiter. That also removes the old `"attempts":2` substring check, which silently matched 20-29 and 200-299 as well. The three projector tests now have no deadline on the success path, except in `projector_wakes_on_committed_log_ingest_before_fallback_poll`, where the bound *is* the assertion: it discriminates "woken by the commit notification" from "woken by the 60s fallback poll". That one is sized for the gap between those two outcomes (20s) rather than for how fast a cycle usually is. Production drops the receiver; `send_replace` never fails on a dropped receiver, so the signal costs one `watch` cell. Verified with the full `cargo test --lib` suite (676s wall, 2463 passed): the two projector failures reproduced on origin/main are gone. --- src/runtime.rs | 5 +- src/runtime/agent_observatory.rs | 51 +++++- src/runtime/agent_observatory_tests.rs | 226 +++++++++++-------------- 3 files changed, 150 insertions(+), 132 deletions(-) diff --git a/src/runtime.rs b/src/runtime.rs index 0182b9b8..a366ca54 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -511,11 +511,14 @@ impl RuntimeCore { let timeline_rollup = self.spawn_timeline_rollup_task(token.clone()); let optimize = self.spawn_optimize_task(token.clone()); let file_tail = self.spawn_file_tail_task(token.clone()); + // The projector's per-cycle progress receiver is a test observation + // seam; the runtime only needs the join handle for shutdown. let agent_observatory_projector = agent_observatory::spawn_projector( token.clone(), Arc::clone(&self.pool), self.config.agent_observatory.clone(), - ); + ) + .map(|(task, _progress)| task); let agent_observatory_git = agent_observatory::spawn_git_reconcile( token.clone(), Arc::clone(&self.pool), diff --git a/src/runtime/agent_observatory.rs b/src/runtime/agent_observatory.rs index e36433b2..70881280 100644 --- a/src/runtime/agent_observatory.rs +++ b/src/runtime/agent_observatory.rs @@ -16,6 +16,7 @@ use crate::scanner::local_hostname; use std::path::PathBuf; use std::sync::Arc; use std::time::Duration; +use tokio::sync::watch; use tokio::task::JoinHandle; use tokio_util::sync::CancellationToken; @@ -37,14 +38,42 @@ fn projector_retry_delay_ms(attempt: usize) -> u64 { TRANSIENT_SQLITE_RETRY_DELAYS_MS[index] } +/// Cumulative, monotonic progress of one projector instance. +/// +/// A projector cycle has no wall-clock bound: it takes the process-wide SQLite +/// write serialization lock (`db::pool::write_lock`) several times, so under +/// contention — most visibly the parallel test suite, where every test's pool +/// queues on that one lock — a single cycle can take tens of seconds. Anything +/// that needs to know a cycle happened must therefore observe this signal +/// rather than assume a cycle fits inside some interval. +/// +/// Every field is cumulative and monotonic, so an observer can never miss a +/// transition by sampling late — unlike the health row, whose `detail` describes +/// only the cycle that wrote it and is overwritten by the next one. +/// `health_records` counts only successfully persisted health rows, so it +/// tracks the `attempts` counter inside the stored health JSON exactly. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub(super) struct ProjectorProgress { + pub(super) cycles: u64, + pub(super) health_records: u64, + pub(super) projected: u64, + pub(super) oversized_first_rows: u64, +} + +/// Spawn the projector, returning its join handle plus a receiver that reports +/// [`ProjectorProgress`] after every completed cycle. Production drops the +/// receiver; `watch::Sender::send_replace` never fails on a dropped receiver, +/// so the signal costs one `watch` cell whether or not anyone is listening. pub(super) fn spawn_projector( token: CancellationToken, pool: Arc, config: AgentObservatoryConfig, -) -> Option> { +) -> Option<(JoinHandle<()>, watch::Receiver)> { config.enabled.then(|| { let mut wake = projection_wake_receiver(); - tokio::spawn(async move { + let (progress_tx, progress_rx) = watch::channel(ProjectorProgress::default()); + let task = tokio::spawn(async move { + let mut progress = ProjectorProgress::default(); let mut consecutive_retry_safe_failures = 0usize; loop { if token.is_cancelled() { break; } @@ -158,7 +187,7 @@ pub(super) fn spawn_projector( None }; let status = if healthy { "ok" } else { "error" }; - if let Err(error) = record_projection_health( + match record_projection_health( &pool, "projector", status, @@ -167,9 +196,20 @@ pub(super) fn spawn_projector( retry_delay_ms.unwrap_or(0) ), ) { - tracing::error!(error = %error, "Agent Observatory projector health write failed"); + Ok(()) => progress.health_records = progress.health_records.saturating_add(1), + Err(error) => { + tracing::error!(error = %error, "Agent Observatory projector health write failed"); + } } tracing::debug!(projected, "Agent Observatory projector cycle completed"); + progress.cycles = progress.cycles.saturating_add(1); + progress.projected = progress.projected.saturating_add(projected as u64); + progress.oversized_first_rows = progress + .oversized_first_rows + .saturating_add(oversized_first_rows as u64); + // Published after the health row is committed so an observer + // woken by this signal always reads the row this cycle wrote. + progress_tx.send_replace(progress); if healthy { tokio::select! { biased; @@ -186,7 +226,8 @@ pub(super) fn spawn_projector( } } } - }) + }); + (task, progress_rx) }) } diff --git a/src/runtime/agent_observatory_tests.rs b/src/runtime/agent_observatory_tests.rs index b6066481..c07fc1be 100644 --- a/src/runtime/agent_observatory_tests.rs +++ b/src/runtime/agent_observatory_tests.rs @@ -3,6 +3,7 @@ use crate::db::agent_observatory::{ advance_projection_cursor, projection_cursor, projection_health, }; use crate::db::{LogBatchEntry, insert_logs_batch}; +use tokio::sync::watch; fn pool() -> (tempfile::TempDir, Arc) { let directory = tempfile::tempdir().unwrap(); @@ -11,48 +12,46 @@ fn pool() -> (tempfile::TempDir, Arc) { (directory, pool) } -/// Deadline for waits on a background worker making progress. +/// Block until the projector's own progress signal satisfies `ready`. /// -/// Every `DbPool` in this test binary shares one process-wide SQLite write lock -/// (`crate::db::pool::write_lock`), and each test pool's `init_pool` runs the -/// full migration set — VACUUM, CREATE INDEX, ANALYZE — while holding it. Under -/// full-suite parallelism a worker's cursor or health write therefore queues -/// behind however many other tests are migrating, which is seconds, not -/// milliseconds. `cargo nextest` gives each test its own process and never sees -/// this contention; plain `cargo test` does, and the old 2s/5s deadlines flaked -/// there. -/// -/// These are liveness backstops for a genuinely stuck worker, not latency -/// assertions — a passing run exits as soon as the condition holds, so the -/// headroom is free. Anything that needs to assert *timing* must do it against a -/// deliberately separated interval, the way the wake test uses `IDLE_POLL_MS`. -const PROGRESS_TIMEOUT: Duration = Duration::from_secs(60); - -/// Deadline for a cancelled worker to unwind. The same contention applies: an -/// in-flight write may still be queued behind the shared write lock. -const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(30); +/// A projector cycle is not wall-clock bounded — it takes the process-wide +/// SQLite write lock several times, and under full-suite parallelism that lock +/// can hold one cycle for tens of seconds — so these tests wait on a real +/// signal instead of polling against a deadline. Predicates must be monotonic +/// over [`ProjectorProgress`]'s cumulative counters: the projector only yields +/// when all of its select branches are pending, so it can run several cycles +/// between two polls of the waiter and a "counter equals N exactly" predicate +/// would be missable. `Err` means the task ended first, which is a real failure +/// rather than a slow machine. +async fn await_progress( + progress: &mut watch::Receiver, + expectation: &str, + ready: impl Fn(ProjectorProgress) -> bool, +) -> ProjectorProgress { + loop { + let seen = *progress.borrow_and_update(); + if ready(seen) { + return seen; + } + progress + .changed() + .await + .unwrap_or_else(|_| panic!("projector exited before {expectation}")); + } +} -/// Poll interval high enough that the projector's fallback timer never fires -/// during a test, so it runs its first cycle and then idles. Observing progress -/// within `PROGRESS_TIMEOUT` therefore means something woke it, not that the -/// timer came round. +/// Fallback poll interval high enough that the projector's own timer never +/// fires during a test, so it runs its first cycle and then idles. Progress +/// observed after that therefore means something *woke* it rather than that the +/// timer came round. Tests that need repeated cycles set a small interval +/// instead. const IDLE_POLL_MS: u64 = 600_000; -/// Reads `attempts` out of a projector health row. -/// -/// `record_projection_health` bumps this on every health write, so it only ever -/// climbs — which is the whole point. The projector's other health fields -/// (`projected`, `oversized_first_rows`) are per-cycle snapshots, and -/// `notify_projection_work` broadcasts on a **process-global** channel that every -/// `insert_logs_batch` in this test binary rings. Any test can therefore drive -/// any other test's projector through another cycle at any moment, overwriting a -/// per-cycle value microseconds after it appears. Assert on monotone facts — -/// this counter, and the durable cursors — never on a per-cycle snapshot. -fn projection_attempts(health: &str) -> u64 { +fn health_attempts(health: &str) -> u64 { serde_json::from_str::(health) - .ok() - .and_then(|value| value.get("attempts").and_then(serde_json::Value::as_u64)) - .unwrap_or_default() + .unwrap_or_else(|error| panic!("health is not JSON: {error}: {health}"))["attempts"] + .as_u64() + .unwrap_or_else(|| panic!("health has no numeric attempts: {health}")) } #[test] @@ -103,17 +102,19 @@ fn enabled_projector_advances_durable_log_cursor_and_shuts_down() { .build() .unwrap(); runtime.block_on(async { - let handle = spawn_projector(token.clone(), Arc::clone(&pool), config).unwrap(); - tokio::time::timeout(PROGRESS_TIMEOUT, async { - loop { - if projection_cursor(&pool, "logs").unwrap() == "1" { - break; - } - tokio::time::sleep(Duration::from_millis(10)).await; - } + let (handle, mut progress) = + spawn_projector(token.clone(), Arc::clone(&pool), config).unwrap(); + let seen = await_progress(&mut progress, "the log row was projected", |p| { + p.projected >= 1 }) - .await - .unwrap(); + .await; + // The single row is over `projector_page_bytes` but is the page's first + // row, so the guard counts it and projects it anyway. Both counters are + // cumulative, and no later cycle has a row to project, so these hold no + // matter how many cycles ran before this task was polled. + assert_eq!(seen.projected, 1); + assert_eq!(seen.oversized_first_rows, 1); + assert_eq!(projection_cursor(&pool, "logs").unwrap(), "1"); for source in [ "mcp_events", "hook_events", @@ -122,34 +123,10 @@ fn enabled_projector_advances_durable_log_cursor_and_shuts_down() { ] { assert_eq!(projection_cursor(&pool, source).unwrap(), ""); } - // The cursor reaching "1" above is what proves the page-bytes guard does - // not stall on a first row bigger than the whole budget — with - // projector_page_bytes = 1 and a 16-byte row, a stalling guard never - // advances it. Health is only checked for *reporting* the counter; - // asserting the count itself would race the global wake broadcast (see - // `projection_attempts`). - // Wait for the health row to exist rather than reading straight after the - // cursor wait: the cursor is written during projection, health only at - // the end of the cycle. Existence is monotone, so this cannot flap. - let health = tokio::time::timeout(PROGRESS_TIMEOUT, async { - loop { - if let Some(health) = projection_health(&pool, "projector").unwrap() { - return health; - } - tokio::time::sleep(Duration::from_millis(10)).await; - } - }) - .await - .expect("a completed cycle records health"); - assert!( - health.contains("oversized_first_rows="), - "projector health must report the oversized-first-row counter: {health}" - ); + let health = projection_health(&pool, "projector").unwrap().unwrap(); + assert!(health.contains("\"status\":\"ok\""), "{health}"); token.cancel(); - tokio::time::timeout(SHUTDOWN_TIMEOUT, handle) - .await - .unwrap() - .unwrap(); + handle.await.unwrap(); }); } @@ -167,17 +144,12 @@ fn projector_wakes_on_committed_log_ingest_before_fallback_poll() { .build() .unwrap(); runtime.block_on(async { - let handle = spawn_projector(token.clone(), Arc::clone(&pool), config).unwrap(); - tokio::time::timeout(PROGRESS_TIMEOUT, async { - loop { - if projection_health(&pool, "projector").unwrap().is_some() { - break; - } - tokio::time::sleep(Duration::from_millis(10)).await; - } + let (handle, mut progress) = + spawn_projector(token.clone(), Arc::clone(&pool), config).unwrap(); + await_progress(&mut progress, "the first cycle completed", |p| { + p.cycles >= 1 }) - .await - .unwrap(); + .await; insert_logs_batch( &pool, @@ -206,25 +178,23 @@ fn projector_wakes_on_committed_log_ingest_before_fallback_poll() { ) .unwrap(); - // Still proves the wake came from the committed ingest rather than the - // fallback poll: PROGRESS_TIMEOUT is an order of magnitude below - // IDLE_POLL_MS. - tokio::time::timeout(PROGRESS_TIMEOUT, async { - loop { - if projection_cursor(&pool, "logs").unwrap() == "1" { - break; - } - tokio::time::sleep(Duration::from_millis(10)).await; - } - }) + // This deadline *is* the assertion: it has to discriminate "woken by the + // commit notification" from "woken by the fallback poll", so a + // wall-clock bound is the correct mechanism here. It is sized for the + // gap between those two outcomes — 20s against a 600s fallback — not for + // how fast a cycle usually is. + tokio::time::timeout( + Duration::from_secs(20), + await_progress(&mut progress, "the committed row was projected", |p| { + p.projected >= 1 + }), + ) .await .expect("committed ingest should wake projector without waiting for fallback poll"); + assert_eq!(projection_cursor(&pool, "logs").unwrap(), "1"); token.cancel(); - tokio::time::timeout(SHUTDOWN_TIMEOUT, handle) - .await - .unwrap() - .unwrap(); + handle.await.unwrap(); }); } @@ -268,7 +238,7 @@ fn enabled_git_worker_records_progress_and_shuts_down() { .unwrap(); runtime.block_on(async { let handle = spawn_git_reconcile(token.clone(), Arc::clone(&pool), config).unwrap(); - tokio::time::timeout(PROGRESS_TIMEOUT, async { + tokio::time::timeout(Duration::from_secs(5), async { loop { if !projection_cursor(&pool, "git").unwrap().is_empty() { break; @@ -279,7 +249,7 @@ fn enabled_git_worker_records_progress_and_shuts_down() { .await .unwrap(); token.cancel(); - tokio::time::timeout(SHUTDOWN_TIMEOUT, handle) + tokio::time::timeout(Duration::from_secs(1), handle) .await .unwrap() .unwrap(); @@ -302,33 +272,37 @@ fn projector_failure_is_queryable_retries_without_advancing_and_stops_after_canc .build() .unwrap(); runtime.block_on(async { - let handle = spawn_projector(token.clone(), Arc::clone(&pool), config).unwrap(); - tokio::time::timeout(PROGRESS_TIMEOUT, async { - loop { - let health = projection_health(&pool, "projector") - .unwrap() - .unwrap_or_default(); - // `>= 2`, not `== 2`: attempts only climbs, so this latches once - // true. Pinning it to an exact value made the assertion true for - // one ~10ms retry window and false forever after. - if health.contains("\"status\":\"error\"") && projection_attempts(&health) >= 2 { - assert!(health.contains("retry_safe=false")); - assert!(health.contains("retry_delay_ms=0")); - break; - } - tokio::time::sleep(Duration::from_millis(10)).await; - } + let (handle, mut progress) = + spawn_projector(token.clone(), Arc::clone(&pool), config).unwrap(); + // Two persisted health rows are two completed cycles, i.e. the failure + // was retried. Awaiting the projector's signal keeps that independent of + // how long a cycle takes under suite load. + await_progress(&mut progress, "two failing cycles were recorded", |p| { + p.health_records >= 2 }) - .await - .unwrap(); + .await; + + let health = projection_health(&pool, "projector").unwrap().unwrap(); + assert!(health.contains("\"status\":\"error\""), "{health}"); + assert!(health.contains("retry_safe=false"), "{health}"); + assert!(health.contains("retry_delay_ms=0"), "{health}"); + assert!(health_attempts(&health) >= 2, "{health}"); assert_eq!(projection_cursor(&pool, "logs").unwrap(), "invalid"); + token.cancel(); - tokio::time::timeout(SHUTDOWN_TIMEOUT, handle) - .await - .unwrap() - .unwrap(); - let stopped = projection_health(&pool, "projector").unwrap(); - tokio::time::sleep(Duration::from_millis(30)).await; - assert_eq!(projection_health(&pool, "projector").unwrap(), stopped); + handle.await.unwrap(); + // The task has been joined, so its progress sender is dropped. A closed + // channel with no unseen update proves no further cycle ran after + // cancellation — the signal replaces a "sleep and re-read" check. + let after_cancel = *progress.borrow_and_update(); + assert!( + progress.changed().await.is_err(), + "projector must not run another cycle after cancellation" + ); + assert_eq!(*progress.borrow(), after_cancel); + assert_eq!( + health_attempts(&projection_health(&pool, "projector").unwrap().unwrap()), + after_cancel.health_records + ); }); }