diff --git a/crates/freshell-server/src/host_stats.rs b/crates/freshell-server/src/host_stats.rs index 370d79045..39196c867 100644 --- a/crates/freshell-server/src/host_stats.rs +++ b/crates/freshell-server/src/host_stats.rs @@ -18,7 +18,9 @@ //! //! `set_active(true)` runs ONE immediate fast tick (a fresh subscriber gets a //! shaped snapshot at once); the slow tier only ticks on its own interval. -//! `set_active(false)` aborts ALL collection tasks (true zero cost). +//! `set_active(false)` aborts ALL collection tasks (true zero cost) and clears +//! every delta base + the merged live cache, so a restart takes the null-safe +//! first-sample path like a fresh start (Node stop() parity). //! `snapshot()` never blocks on I/O — ticks write caches, snapshots read //! caches. //! @@ -841,6 +843,20 @@ impl HostStatsCollector for HostStatsCollectorService { handles.fast.abort(); handles.slow.abort(); handles.drift.abort(); + // Restart = fresh start (Node stop() parity): drop every + // cumulative-counter delta base so the restart's immediate tick + // reports the null-safe first sample instead of deltas spanning + // the whole unwatched interval; clear queued drift samples (Node + // disables/nulls its histogram at stop); reset the merged live + // cache so no stale slow sections survive. The `manual` refresh + // cache is request-owned (not sampled state) and stays. Cadence + // is locked first; these are disjoint mutexes — no lock cycle. + *self.ctx.share.prev_cpu.lock().unwrap() = None; + *self.ctx.share.prev_vmstat.lock().unwrap() = None; + *self.ctx.share.prev_disks.lock().unwrap() = None; + *self.ctx.share.prev_net.lock().unwrap() = None; + self.ctx.share.lag_samples.lock().unwrap().clear(); + *self.ctx.share.live.lock().unwrap() = zero_live(&self.ctx.machine); } } } @@ -2190,6 +2206,201 @@ mod tests { collector.set_active(false); } + /// Rewrite the first `name value` line inside a copied fixture file — the + /// overlay tree's counters grow in place to simulate an unwatched + /// interval (Node suite parity: re-seeded reader mocks). + fn rewrite_counter(path: &Path, name: &str, value: u64) { + let text = std::fs::read_to_string(path).unwrap(); + let mut out = String::new(); + let mut replaced = false; + for line in text.lines() { + if !replaced && line.starts_with(&format!("{name} ")) { + out.push_str(&format!("{name} {value}\n")); + replaced = true; + } else { + out.push_str(line); + out.push('\n'); + } + } + assert!(replaced, "counter `{name}` present in {}", path.display()); + std::fs::write(path, out).unwrap(); + } + + // Full-file fixture rewrites for the restart rebase test. `lo`/headers/ + // partitions/`docker0` lines are always identical to the committed + // fixtures; only the named device counters grow. (Trailing newline each.) + + /// diskstats growth over the unwatched interval (every device grows). + const DISKSTATS_G1: &str = r" 8 0 sda 6000 150 440000 7000 2400 75 240800 4000 0 4500 9000 + 8 1 sda1 4000 80 300000 5000 1500 40 150000 2500 0 3000 7500 + 7 0 loop0 100 0 800 10 0 0 0 0 0 10 10 + 259 0 nvme0n1 9000 200 700000 8000 3000 60 300000 4000 0 5000 12000 + 259 1 nvme0n1p1 8000 150 600000 7000 2500 55 250000 3500 0 4500 10500 +"; + + /// diskstats after the rebase proof bump — only the sda line changes + /// (dReads 100, dReadMs 2000, dWrites 400, dWriteMs 2000, dIosMs 1000). + const DISKSTATS_G2: &str = r" 8 0 sda 6100 150 491200 9000 2800 75 343200 6000 0 5500 9000 + 8 1 sda1 4000 80 300000 5000 1500 40 150000 2500 0 3000 7500 + 7 0 loop0 100 0 800 10 0 0 0 0 0 10 10 + 259 0 nvme0n1 9000 200 700000 8000 3000 60 300000 4000 0 5000 12000 + 259 1 nvme0n1p1 8000 150 600000 7000 2500 55 250000 3500 0 4500 10500 +"; + + /// net/dev growth over the unwatched interval — only eth0 errs 7->17. + const NET_DEV_G1: &str = r"Inter-| Receive | Transmit + face |bytes packets errs drop fifo frame compressed multicast|bytes packets errs drop fifo colls carrier compressed + lo: 1000000 10000 0 0 0 0 0 0 1000000 10000 0 0 0 0 0 0 + eth0: 5000000 50000 17 3 0 0 0 0 8000000 80000 11 4 0 0 0 0 +docker0: 2000000 20000 2 1 0 0 0 0 3000000 30000 5 2 0 0 0 0 +"; + + /// net/dev after the rebase proof bump — only eth0 errs 17->18. + const NET_DEV_G2: &str = r"Inter-| Receive | Transmit + face |bytes packets errs drop fifo frame compressed multicast|bytes packets errs drop fifo colls carrier compressed + lo: 1000000 10000 0 0 0 0 0 0 1000000 10000 0 0 0 0 0 0 + eth0: 5000000 50000 18 3 0 0 0 0 8000000 80000 11 4 0 0 0 0 +docker0: 2000000 20000 2 1 0 0 0 0 3000000 30000 5 2 0 0 0 0 +"; + + #[tokio::test] + async fn host_stats_set_active_restart_rebases_deltas_and_resets_live_cache() { + // Full tmpdir overlay: the config roots are immutable after + // construction, so the fixture counters grow via file rewrites. + let tmp = tempfile::tempdir().unwrap(); + let proc_root = tmp.path().join("proc"); + let sys_root = tmp.path().join("sys"); + copy_tree(&proc_fixture(), &proc_root); + copy_tree(&sys_fixture(), &sys_root); + + let interest = HostStatsInterestRegistry::default(); + let collector = test_collector(proc_root.clone(), sys_root, &interest); + + // Cycle 1: run until every prev_* store AND the slow-owned live + // sections have populated from the committed fixture counters. + collector.set_active(true); + assert!( + wait_until(Duration::from_millis(500), || { + collector.snapshot().live.disk_io.available + }) + .await, + "slow sections populate before stop" + ); + collector.set_active(false); + // A stopped collector publishes the documented fresh-subscriber shape + // again — no stale slow sections survive a stop. + let stopped = collector.snapshot(); + assert!(!stopped.live.disk_io.available); + assert!(!stopped.live.network.available); + assert!(!stopped.live.limits.available); + + // The unwatched counters GROW: aggregate cpu line x10 (busy 78850 / + // total 1742360) plus pswpin/pswpout/pgmajfault/oom_kill growth, and + // the diskstats/net-dev G1 files below. + let stat_path = proc_root.join("stat"); + let stat = std::fs::read_to_string(&stat_path).unwrap(); + let grown = stat.replacen( + "cpu 4705 356 1622 164331 2020 80 345 777 0 0", + "cpu 47050 3560 16220 1643310 20200 800 3450 7770 0 0", + 1, + ); + assert!(grown != stat, "aggregate cpu line rewritten"); + std::fs::write(&stat_path, grown).unwrap(); + rewrite_counter(&proc_root.join("vmstat"), "pswpin", 11234); + rewrite_counter(&proc_root.join("vmstat"), "pswpout", 15678); + rewrite_counter(&proc_root.join("vmstat"), "pgmajfault", 1890); + rewrite_counter(&proc_root.join("vmstat"), "oom_kill", 13); + std::fs::write(proc_root.join("diskstats"), DISKSTATS_G1).unwrap(); + std::fs::write(proc_root.join("net/dev"), NET_DEV_G1).unwrap(); + + collector.set_active(true); + let restarted = collector.snapshot(); + // The restart's immediate tick is the documented null-safe first + // sample: zero rates/deltas, totals carried from the CURRENT (grown) + // sample, slow sections still unavailable. + assert!(restarted.live.cpu.available); + assert_eq!(restarted.live.cpu.usage_pct, 0.0); + assert_eq!(restarted.live.cpu.steal_pct, Some(0.0)); + assert_eq!(restarted.live.paging.swap_in_kbps, 0.0); + assert_eq!(restarted.live.paging.swap_out_kbps, 0.0); + assert_eq!(restarted.live.paging.maj_faults_per_sec, 0.0); + assert_eq!(restarted.live.paging.oom_kills_delta, 0); + assert_eq!(restarted.live.paging.oom_kills_total, 13); + assert!(!restarted.live.disk_io.available); + assert!(!restarted.live.network.available); + assert!(!restarted.live.limits.available); + + // Positive re-base proof #1: bump ONLY oom_kill by 1; the next fast + // tick reports delta 1 against the RESTART sample — never 11 against + // the pre-stop sample. + rewrite_counter(&proc_root.join("vmstat"), "oom_kill", 14); + assert!( + wait_until(Duration::from_millis(500), || { + let paging = &collector.snapshot().live.paging; + paging.oom_kills_delta == 1 && paging.oom_kills_total == 14 + }) + .await, + "paging deltas re-base on the restart sample (delta 1, total 14)" + ); + + // Slow tier (Finding 1 disposition): the FIRST post-restart slow tick + // is the null-safe first slow sample — zero rates/deltas, totals + // carried from the G1 growth (eth0 rx errs 7->17 plus docker0 2 = 19). + assert!( + wait_until(Duration::from_millis(500), || { + collector.snapshot().live.disk_io.available + }) + .await, + "first post-restart slow tick populates the slow sections" + ); + assert!( + wait_until(Duration::from_millis(500), || { + let live = &collector.snapshot().live; + live.disk_io.available + && live.disk_io.read_bps == 0.0 + && live.disk_io.write_bps == 0.0 + && live.disk_io.util_pct.is_none() + && live.disk_io.weighted_await_ms.is_none() + && live.network.available + && live.network.rx_bps == 0.0 + && live.network.tx_bps == 0.0 + && live.network.rx_errors_total == 19 + && live.network.tx_errors_total == 16 + && live.network.rx_dropped_total == 4 + && live.network.tx_dropped_total == 6 + && live.network.rx_errors_delta == 0 + }) + .await, + "first post-restart slow tick is the null-safe first slow sample with G1 totals" + ); + + // Positive re-base proof #2 (dt-free witnesses only): grow eth0 rx + // errs by exactly 1 (17->18) and grow ONLY sda's diskstats counters + // (G1->G2: dReads 100, dReadMs 2000, dWrites 400, dWriteMs 2000, + // dIosMs 1000; every other device unchanged so sda is the worst-util + // device). The next slow tick must report counts/ratios against the + // RESTART samples: rx_errors_delta == 1 (never 11) and + // weighted_await_ms == (2000+2000)/(100+400) == 8.0 (never the + // gap-spanning value). BEFORE finalizing, verify the exact field + // mapping, loopback/partition exclusion, and worst-device selection + // in freshell-platform/src/host_stats_readers.rs and adjust THESE + // COMMENTS/VALUES only if the code contradicts them (record any + // adjustment in your report). + std::fs::write(proc_root.join("diskstats"), DISKSTATS_G2).unwrap(); + std::fs::write(proc_root.join("net/dev"), NET_DEV_G2).unwrap(); + assert!( + wait_until(Duration::from_millis(2000), || { + let live = &collector.snapshot().live; + live.network.rx_errors_delta == 1 + && live.network.rx_errors_total == 20 + && live.disk_io.weighted_await_ms == Some(8.0) + }) + .await, + "network/disk re-base on the restart sample (dt-free witnesses)" + ); + collector.set_active(false); + } + #[test] fn host_stats_snapshot_zero_shape_before_first_tick() { let interest = HostStatsInterestRegistry::default(); diff --git a/server/host-stats/service.ts b/server/host-stats/service.ts index 827d2a674..e8b51eccc 100644 --- a/server/host-stats/service.ts +++ b/server/host-stats/service.ts @@ -11,7 +11,9 @@ * * start() runs ONE immediate fast tick (a fresh subscriber gets a shaped snapshot at once); * the slow tier only ticks on its own interval. stop() halts ALL collection (true zero - * cost). getSnapshot() never blocks on I/O — ticks write caches, snapshots read caches. + * cost) and clears every delta base + the merged live cache, so a restart takes the + * null-safe first-sample path like a fresh start. getSnapshot() never blocks on I/O — + * ticks write caches, snapshots read caches. * * refresh() (on-request manual data — process table, disks, inotify, thermals/battery) is * single-flight with a 1s post-completion cooldown (connection-agnostic, R3M6). Section @@ -297,6 +299,18 @@ export class HostStatsService { this.slowTimer = undefined this.histogram?.disable() this.histogram = null + // Restart = fresh start (contract point 8: what a fresh subscriber + // receives): drop every cumulative-counter delta base so the restart's + // immediate tick takes the documented null-safe first-sample path instead + // of spanning the whole unwatched interval, and reset the merged live + // cache so no stale slow sections survive. The refresh-owned manualCache / + // manualAt are request state, not sampled state, and are retained. + this.prevCpu = null + this.prevDarwinCpu = null + this.prevVmstat = null + this.prevDisks = null + this.prevNet = null + this.liveCache = zeroLive(this.machine) } isRunning(): boolean { diff --git a/test/unit/server/host-stats/service.test.ts b/test/unit/server/host-stats/service.test.ts index 376622213..7eebee8b6 100644 --- a/test/unit/server/host-stats/service.test.ts +++ b/test/unit/server/host-stats/service.test.ts @@ -128,6 +128,46 @@ const DISK_T1 = new Map([ const NET_T0 = { rxBytes: 1_000_000, txBytes: 500_000, rxErr: 3, txErr: 1, rxDrop: 2, txDrop: 4 } const NET_T1 = { rxBytes: 1_500_000, txBytes: 600_000, rxErr: 5, txErr: 3, rxDrop: 3, txDrop: 5 } +const CPU_T2 = { + total: 6000, + busy: 600, + steal: 60, + perCore: [ + { total: 1500, busy: 150 }, + { total: 1500, busy: 150 }, + { total: 1500, busy: 150 }, + { total: 1500, busy: 150 }, + ], +} +const CPU_T3 = { + total: 8000, + busy: 1200, + steal: 100, + perCore: [ + { total: 2000, busy: 300 }, + { total: 2000, busy: 300 }, + { total: 2000, busy: 300 }, + { total: 2000, busy: 300 }, + ], +} +// T2->T3 over one 2s fast tick: dBusy 600/dTotal 2000 = 30%; dSteal 40/2000 = 2%; +// per-core 150/500 = 30%; swap in 24*4/2 = 48 KB/s; out 12*4/2 = 24 KB/s; +// majfaults 40/2 = 20/s; oom 9->11. +const VMSTAT_T2 = { pswpin: 200, pswpout: 80, pgmajfault: 100, oomKill: 9 } +const VMSTAT_T3 = { pswpin: 224, pswpout: 92, pgmajfault: 140, oomKill: 11 } +// Slow tier growth for the restart cycle (re-based T2->T3 values over 5s equal +// the familiar disk/net delta math): read 51200*512/5 = 5,242,880 B/s; write +// 102400*512/5 = 10,485,760 B/s; util 1000/5000 = 20%; await (2000+2000)/(100+400) = 8ms; +// rx (2.5M-2M)/5 = 100,000 B/s; tx (1.1M-1M)/5 = 20,000 B/s; err/drop deltas 2/3/1/2. +const DISK_T2 = new Map([ + ['sda', { readsCompleted: 2000, readMs: 8000, writesCompleted: 4000, writeMs: 16000, readSectors: 200_000, writtenSectors: 800_000, timeDoingIosMs: 1000 }], +]) +const DISK_T3 = new Map([ + ['sda', { readsCompleted: 2100, readMs: 10000, writesCompleted: 4400, writeMs: 18000, readSectors: 251_200, writtenSectors: 902_400, timeDoingIosMs: 2000 }], +]) +const NET_T2 = { rxBytes: 2_000_000, txBytes: 1_000_000, rxErr: 7, txErr: 2, rxDrop: 5, txDrop: 6 } +const NET_T3 = { rxBytes: 2_500_000, txBytes: 1_100_000, rxErr: 9, txErr: 5, rxDrop: 6, txDrop: 8 } + const TABLE = { top: [{ pid: 5, name: 'node', cpuPct: 12.3, rssBytes: 1e6, state: 'S' }], zombies: 1, @@ -334,6 +374,77 @@ describe('start/stop (contract points 1, 5)', () => { vi.advanceTimersByTime(2000) expect(readerFn('readCpuTimes')).toHaveBeenCalledTimes(2) }) + + it('restart after stop re-bases every delta family and resets the live cache (fresh-start restart)', () => { + // Cycle 1: run the default T0 fixtures long enough for every prev-* store + // AND the slow-owned liveCache keys to populate (fast ticks at 2/4/6s, the + // first slow tick at 5s). + const service = makeService() + service.start() + vi.advanceTimersByTime(6000) + expect(service.getSnapshot().live.diskIo.available).toBe(true) // slow cache populated pre-stop + + service.stop() + // A stopped service publishes the documented fresh-subscriber shape again: + // no stale slow sections survive a stop. + const stopped = service.getSnapshot().live + expect(stopped.diskIo.available).toBe(false) + expect(stopped.network.available).toBe(false) + expect(stopped.limits.available).toBe(false) + + // The cumulative counters GROW while the service is unwatched — the + // long-window bug case (all values below are grown from the T0 cycle). + // Fake timers advance Date.now(), the dt basis. + vi.advanceTimersByTime(20000) + vi.mocked(readersMock.readCpuTimes).mockReturnValue(CPU_T2) + vi.mocked(readersMock.readVmstat).mockReturnValue(VMSTAT_T2) + vi.mocked(readersMock.readDiskStats).mockReturnValue(DISK_T2) + vi.mocked(readersMock.readNetDev).mockReturnValue(NET_T2) + + service.start() + // The restart's immediate tick is the documented null-safe first sample: + // zero rates/deltas, totals carried from the CURRENT (T2) sample, and the + // slow sections stay unavailable until their first new slow tick. + const restarted = service.getSnapshot().live + expect(restarted.cpu).toEqual({ available: true, usagePct: 0, stealPct: 0, perCorePct: [0, 0, 0, 0], freqMHz: null }) + expect(restarted.paging).toEqual({ available: true, swapInKbps: 0, swapOutKbps: 0, majFaultsPerSec: 0, oomKillsDelta: 0, oomKillsTotal: 9 }) + expect(restarted.diskIo.available).toBe(false) + expect(restarted.network.available).toBe(false) + expect(restarted.limits.available).toBe(false) + + // The next fast tick deltas against the RESTART sample (T2), never against + // the pre-stop sample (T0): every value below derives from T2->T3, dt=2s. + vi.mocked(readersMock.readCpuTimes).mockReturnValue(CPU_T3) + vi.mocked(readersMock.readVmstat).mockReturnValue(VMSTAT_T3) + vi.advanceTimersByTime(2000) + const live = service.getSnapshot().live + expect(live.cpu).toEqual({ available: true, usagePct: 30, stealPct: 2, perCorePct: [30, 30, 30, 30], freqMHz: null }) + expect(live.paging).toEqual({ available: true, swapInKbps: 48, swapOutKbps: 24, majFaultsPerSec: 20, oomKillsDelta: 2, oomKillsTotal: 11 }) + + // First post-restart slow tick (t=+5000): the null-safe first slow sample + // — zero rates/deltas, carried totals from the GROWN samples. + vi.advanceTimersByTime(3000) + const firstSlow = service.getSnapshot().live + expect(firstSlow.diskIo).toEqual({ available: true, readBps: 0, writeBps: 0, utilPct: null, weightedAwaitMs: null }) + expect(firstSlow.network).toEqual({ + available: true, rxBps: 0, txBps: 0, + rxErrorsTotal: 7, txErrorsTotal: 2, rxDroppedTotal: 5, txDroppedTotal: 6, + rxErrorsDelta: 0, txErrorsDelta: 0, rxDroppedDelta: 0, txDroppedDelta: 0, + }) + + // Second post-restart slow tick (t=+10000): disk/net re-based on the FIRST + // post-restart slow sample (T2), never the pre-stop sample (T0). + vi.mocked(readersMock.readDiskStats).mockReturnValue(DISK_T3) + vi.mocked(readersMock.readNetDev).mockReturnValue(NET_T3) + vi.advanceTimersByTime(5000) + const rebased = service.getSnapshot().live + expect(rebased.diskIo).toEqual({ available: true, readBps: 5_242_880, writeBps: 10_485_760, utilPct: 20, weightedAwaitMs: 8 }) + expect(rebased.network).toEqual({ + available: true, rxBps: 100_000, txBps: 20_000, + rxErrorsTotal: 9, txErrorsTotal: 5, rxDroppedTotal: 6, txDroppedTotal: 8, + rxErrorsDelta: 2, txErrorsDelta: 3, rxDroppedDelta: 1, txDroppedDelta: 2, + }) + }) }) describe('event-loop lag histogram lifecycle (contract point 3)', () => { @@ -362,7 +473,7 @@ describe('event-loop lag histogram lifecycle (contract point 3)', () => { expect(fakeHistogram.disable).toHaveBeenCalledTimes(1) }) - it('collects no lag samples while stopped (cache retains last tick), then resumes per-tick on restart', () => { + it('collects no lag samples while stopped, then resumes per-tick on restart', () => { const service = makeService() service.start() service.stop()