diff --git a/.config/nextest.toml b/.config/nextest.toml index 070069fff8..bc3f72a109 100644 --- a/.config/nextest.toml +++ b/.config/nextest.toml @@ -34,6 +34,14 @@ telemetry-contract = { max-threads = 1 } # parallel load the file never appears (#5355). Serialize them; do not drop # the tests. exec-persistent-service = { max-threads = 1 } +# Fleet manager lifecycle tests start real `/bin/sh` workers under setsid and +# wait 10-15 s for the child's first line. Under full macOS CI load that line +# never ran for three overlapping tests (#6424, #6428 runs), while each +# passed alone from the same binary in under 1.2 s. Serialize the module; do +# not lengthen its deadlines or drop tests. The product side of the load +# (a durable heartbeat and a `ps` sample on every scheduler tick) is now +# bounded to once a second per worker. +fleet-manager-lifecycle = { max-threads = 1 } # First matching test-group override wins. Keep these more-specific # integration filters before binary(integration) so they are not stolen @@ -56,6 +64,10 @@ test-group = 'telemetry-contract' filter = 'binary(telemetry_kill_switch_dispatch)' test-group = 'telemetry-contract' +[[profile.default.overrides]] +filter = 'package(codewhale-tui) & kind(lib) & test(/^fleet::manager::tests::/)' +test-group = 'fleet-manager-lifecycle' + [[profile.default.overrides]] filter = 'binary(integration) & test(/^exec_persistent_service::/)' test-group = 'exec-persistent-service' diff --git a/crates/tui/src/fleet/host.rs b/crates/tui/src/fleet/host.rs index d332ff5aa3..167b13b37a 100644 --- a/crates/tui/src/fleet/host.rs +++ b/crates/tui/src/fleet/host.rs @@ -190,8 +190,14 @@ struct LocalWorkerProcess { stopped: bool, last_exit: Option, last_memory_mb: Option, + /// When `ps` last sampled this worker. Status polls can run every few + /// milliseconds; memory is display data, so one sample a second is ample. + last_memory_sample: Option, } +/// Minimum spacing between `ps` memory samples for one worker. +const MEMORY_SAMPLE_INTERVAL: Duration = Duration::from_secs(1); + impl LocalProcessFleetHostAdapter { pub fn new(workspace: impl AsRef) -> Self { Self { @@ -303,6 +309,7 @@ impl LocalProcessFleetHostAdapter { stopped: false, last_exit: None, last_memory_mb: None, + last_memory_sample: None, }, ); Ok(handle) @@ -356,7 +363,11 @@ impl FleetHostAdapter for LocalProcessFleetHostAdapter { match process.child.try_wait() { Ok(None) => { let pid = process.child.id(); - let memory_mb = if process.host_kind == FleetHostKind::LocalProcess { + let due = process + .last_memory_sample + .is_none_or(|sampled| sampled.elapsed() >= MEMORY_SAMPLE_INTERVAL); + let memory_mb = if process.host_kind == FleetHostKind::LocalProcess && due { + process.last_memory_sample = Some(std::time::Instant::now()); sample_process_memory_mb(pid) } else { None diff --git a/crates/tui/src/fleet/manager.rs b/crates/tui/src/fleet/manager.rs index 76534ac1ec..6b81ac471f 100644 --- a/crates/tui/src/fleet/manager.rs +++ b/crates/tui/src/fleet/manager.rs @@ -608,11 +608,20 @@ impl FleetManager { .ok_or_else(|| anyhow!("Fleet run {} does not exist", run_id.0))?; let worker_ids = worker_ids_for_run(&run, max_workers); + // Heartbeats are durable ledger appends (a full-drive flush on + // macOS). Timestamps have whole-second resolution and the stale window + // is minutes, so a worker already stamped this second needs no second + // record; a fast driver tick must not turn into a flush storm. + let now = timestamp(); for task in active_tasks_for_run(&state, run_id) { if let Some(worker_id) = task.leased_to.as_deref() && worker_ids.iter().any(|id| id == worker_id) + && state + .heartbeats + .get(worker_id) + .is_none_or(|heartbeat| heartbeat.timestamp != now) { - self.ledger.heartbeat(worker_id, ×tamp(), None, None)?; + self.ledger.heartbeat(worker_id, &now, None, None)?; report.heartbeats += 1; } }