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.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 f7d92bee..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,6 +12,48 @@ fn pool() -> (tempfile::TempDir, Arc) { (directory, pool) } +/// Block until the projector's own progress signal satisfies `ready`. +/// +/// 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}")); + } +} + +/// 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; + +fn health_attempts(health: &str) -> u64 { + serde_json::from_str::(health) + .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] fn projector_sqlite_retry_backoff_reuses_ingest_policy_and_caps() { assert_eq!(projector_retry_delay_ms(1), 25); @@ -49,7 +92,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() }; @@ -59,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(Duration::from_secs(2), 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", @@ -78,26 +123,10 @@ 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 { - loop { - if projection_health(&pool, "projector") - .unwrap() - .is_some_and(|health| health.contains("oversized_first_rows=1")) - { - break; - } - tokio::time::sleep(Duration::from_millis(10)).await; - } - }) - .await - .unwrap(); + let health = projection_health(&pool, "projector").unwrap().unwrap(); + assert!(health.contains("\"status\":\"ok\""), "{health}"); token.cancel(); - tokio::time::timeout(Duration::from_secs(1), handle) - .await - .unwrap() - .unwrap(); + handle.await.unwrap(); }); } @@ -106,7 +135,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(); @@ -115,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(Duration::from_secs(2), 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, @@ -154,22 +178,23 @@ fn projector_wakes_on_committed_log_ingest_before_fallback_poll() { ) .unwrap(); - tokio::time::timeout(Duration::from_secs(1), 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(Duration::from_secs(1), handle) - .await - .unwrap() - .unwrap(); + handle.await.unwrap(); }); } @@ -247,30 +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(Duration::from_secs(10), async { - loop { - let health = projection_health(&pool, "projector") - .unwrap() - .unwrap_or_default(); - if health.contains("\"status\":\"error\"") && health.contains("\"attempts\":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(Duration::from_secs(5), 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 + ); }); }