multiple feature and performance improvements (see description)

Here are concise release notes you can paste into your GitHub release.

Release notes — 2025-08-12

Highlights

Agent back to near-zero CPU when idle (request-driven, no background samplers).
Accurate per-process CPU% via /proc deltas; only top-level processes (threads hidden).
TUI: processes pane gets scrollbar, click-to-sort (CPU% or Mem) with indicator, stable total count.
Network panes made taller; disks slightly reduced.
README revamped: rustup prereqs, crates.io install, update/systemd instructions.
Clippy cleanups across agent and client.
Agent

Reverted precompressed caches and background samplers; WebSocket path is request-driven again.
Ensured on-demand gzip for larger replies; no per-request overhead when small.
Processes: switched to refresh_processes_specifics with ProcessRefreshKind::everything().without_tasks() to exclude threads.
Per-process CPU% now computed from /proc jiffies deltas using a small ProcCpuTracker (fixes “always 0%”/scaling issues).
Optional metrics and light caching:
CPU temp and GPU metrics gated by env (SOCKTOP_AGENT_TEMP=0, SOCKTOP_AGENT_GPU=0).
Tiny TTL caches via once_cell to avoid rescanning sensors every tick.
Dependencies: added once_cell = "1.19".
No API changes to WS endpoints.
Client (TUI)

Processes pane:
Scrollbar (mouse wheel, drag; keyboard arrows/PageUp/PageDown/Home/End).
Click header to sort by CPU% or Mem; dot indicator on active column.
Preserves process_count across fast metrics updates to avoid flicker.
UI/theme:
Shared scrollbar colors moved to ui/theme.rs; both CPU and Processes reuse them.
Cached pane rect to fix input handling; removed unused vars.
Layout: network download/upload get more vertical space; disks shrink slightly.
Clippy fixes: derive Default for ProcSortBy; style/import cleanups.
Docs

README: added rustup install steps (with proper shell reload), install via cargo install socktop and cargo install socktop_agent, and a clear Updating section (systemd service steps included).
Features list updated; roadmap marks independent cadences as done.
Upgrade notes

Agent: cargo install socktop_agent --force, then restart your systemd service; if unit changed, systemctl daemon-reload.
TUI: cargo install socktop --force.
Optional envs to trim overhead: SOCKTOP_AGENT_GPU=0, SOCKTOP_AGENT_TEMP=0.
No config or API breaking changes.

socktop_agent/src/main.rs

@@ -9,64 +9,25 @@ mod types;
 mod ws;
 
 use axum::{routing::get, Router};
-use std::{net::SocketAddr, sync::atomic::AtomicUsize, sync::Arc, time::Duration};
+use std::net::SocketAddr;
 
-use sysinfo::{
-    Components, CpuRefreshKind, Disks, MemoryRefreshKind, Networks, ProcessRefreshKind,
-    RefreshKind, System,
-};
-use tokio::sync::{Mutex, Notify, RwLock};
-use tracing_subscriber::EnvFilter;
-
-use sampler::spawn_sampler;
+use crate::sampler::{spawn_disks_sampler, spawn_process_sampler, spawn_sampler};
 use state::AppState;
 use ws::ws_handler;
 
 #[tokio::main]
 async fn main() {
-    // Init logging; configure with RUST_LOG (e.g., RUST_LOG=info).
-    tracing_subscriber::fmt()
-        .with_env_filter(EnvFilter::from_default_env())
-        .with_target(false)
-        .compact()
-        .init();
+    tracing_subscriber::fmt::init();
 
-    // sysinfo build specifics (scopes what refresh_all() will touch internally)
-    let refresh_kind = RefreshKind::nothing()
-        .with_cpu(CpuRefreshKind::everything())
-        .with_memory(MemoryRefreshKind::everything())
-        .with_processes(ProcessRefreshKind::everything());
-
-    // Initialize sysinfo handles once and keep them alive
-    let mut sys = System::new_with_specifics(refresh_kind);
-    sys.refresh_all();
-
-    let mut nets = Networks::new();
-    nets.refresh(true);
-
-    let mut components = Components::new();
-    components.refresh(true);
-
-    let mut disks = Disks::new();
-    disks.refresh(true);
-
-    // Shared state across requests
-    let state = AppState {
-        sys: Arc::new(Mutex::new(sys)),
-        last_json: Arc::new(RwLock::new(String::new())),
-        components: Arc::new(Mutex::new(components)),
-        disks: Arc::new(Mutex::new(disks)),
-        networks: Arc::new(Mutex::new(nets)),
-        // new: adaptive sampling controls
-        client_count: Arc::new(AtomicUsize::new(0)),
-        wake_sampler: Arc::new(Notify::new()),
-        auth_token: std::env::var("SOCKTOP_TOKEN")
-            .ok()
-            .filter(|s| !s.is_empty()),
-    };
+    let state = AppState::new();
 
     // Start background sampler (adjust cadence as needed)
-    let _sampler = spawn_sampler(state.clone(), Duration::from_millis(500));
+    // 500ms fast metrics
+    let _h_fast = spawn_sampler(state.clone(), std::time::Duration::from_millis(500));
+    // 2s processes (top 50)
+    let _h_procs = spawn_process_sampler(state.clone(), std::time::Duration::from_secs(2), 50);
+    // 5s disks
+    let _h_disks = spawn_disks_sampler(state.clone(), std::time::Duration::from_secs(5));
 
     // Web app
     let port = resolve_port();

socktop_agent/src/metrics.rs

@@ -2,75 +2,145 @@
 
 use crate::gpu::collect_all_gpus;
 use crate::state::AppState;
-use crate::types::{DiskInfo, Metrics, NetworkInfo, ProcessInfo};
-
-use std::cmp::Ordering;
-use sysinfo::{ProcessesToUpdate, System};
+use crate::types::{DiskInfo, Metrics, NetworkInfo, ProcessInfo, ProcessesPayload};
+use once_cell::sync::OnceCell;
+use std::collections::HashMap;
+use std::fs;
+use std::io;
+use std::sync::Mutex;
+use std::time::{Duration, Instant};
+use sysinfo::{ProcessRefreshKind, ProcessesToUpdate, System};
 use tracing::warn;
 
-pub async fn collect_metrics(state: &AppState) -> Metrics {
-    let mut sys = state.sys.lock().await;
+// Runtime toggles (read once)
+fn gpu_enabled() -> bool {
+    static ON: OnceCell<bool> = OnceCell::new();
+    *ON.get_or_init(|| {
+        std::env::var("SOCKTOP_AGENT_GPU")
+            .map(|v| v != "0")
+            .unwrap_or(true)
+    })
+}
+fn temp_enabled() -> bool {
+    static ON: OnceCell<bool> = OnceCell::new();
+    *ON.get_or_init(|| {
+        std::env::var("SOCKTOP_AGENT_TEMP")
+            .map(|v| v != "0")
+            .unwrap_or(true)
+    })
+}
 
-    // Targeted refresh: CPU/mem/processes only
+// Tiny TTL caches to avoid rescanning sensors every 500ms
+const TTL: Duration = Duration::from_millis(1500);
+struct TempCache {
+    at: Option<Instant>,
+    v: Option<f32>,
+}
+static TEMP: OnceCell<Mutex<TempCache>> = OnceCell::new();
+
+struct GpuCache {
+    at: Option<Instant>,
+    v: Option<Vec<crate::gpu::GpuMetrics>>,
+}
+static GPUC: OnceCell<Mutex<GpuCache>> = OnceCell::new();
+
+fn cached_temp() -> Option<f32> {
+    if !temp_enabled() {
+        return None;
+    }
+    let now = Instant::now();
+    let lock = TEMP.get_or_init(|| Mutex::new(TempCache { at: None, v: None }));
+    let mut c = lock.lock().ok()?;
+    if c.at.is_none_or(|t| now.duration_since(t) >= TTL) {
+        c.at = Some(now);
+        // caller will fill this; we just hold a slot
+        c.v = None;
+    }
+    c.v
+}
+
+fn set_temp(v: Option<f32>) {
+    if let Some(lock) = TEMP.get() {
+        if let Ok(mut c) = lock.lock() {
+            c.v = v;
+            c.at = Some(Instant::now());
+        }
+    }
+}
+
+fn cached_gpus() -> Option<Vec<crate::gpu::GpuMetrics>> {
+    if !gpu_enabled() {
+        return None;
+    }
+    let now = Instant::now();
+    let lock = GPUC.get_or_init(|| Mutex::new(GpuCache { at: None, v: None }));
+    let mut c = lock.lock().ok()?;
+    if c.at.is_none_or(|t| now.duration_since(t) >= TTL) {
+        // mark stale; caller will refresh
+        c.at = Some(now);
+        c.v = None;
+    }
+    c.v.clone()
+}
+
+fn set_gpus(v: Option<Vec<crate::gpu::GpuMetrics>>) {
+    if let Some(lock) = GPUC.get() {
+        if let Ok(mut c) = lock.lock() {
+            c.v = v.clone();
+            c.at = Some(Instant::now());
+        }
+    }
+}
+
+// Collect only fast-changing metrics (CPU/mem/net + optional temps/gpus).
+pub async fn collect_fast_metrics(state: &AppState) -> Metrics {
+    let mut sys = state.sys.lock().await;
     if let Err(e) = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
         sys.refresh_cpu_usage();
         sys.refresh_memory();
-        sys.refresh_processes(ProcessesToUpdate::All, true);
     })) {
         warn!("sysinfo selective refresh panicked: {e:?}");
     }
 
-    // Hostname
     let hostname = System::host_name().unwrap_or_else(|| "unknown".to_string());
-
-    // CPU usage
     let cpu_total = sys.global_cpu_usage();
     let cpu_per_core: Vec<f32> = sys.cpus().iter().map(|c| c.cpu_usage()).collect();
-
-    // Memory / swap
     let mem_total = sys.total_memory();
     let mem_used = mem_total.saturating_sub(sys.available_memory());
     let swap_total = sys.total_swap();
     let swap_used = sys.used_swap();
+    drop(sys);
 
-    drop(sys); // release quickly before touching other locks
-
-    // Components (cached): just refresh temps
-    let cpu_temp_c = {
-        let mut components = state.components.lock().await;
-        components.refresh(true);
-        components.iter().find_map(|c| {
-            let l = c.label().to_ascii_lowercase();
-            if l.contains("cpu")
-                || l.contains("package")
-                || l.contains("tctl")
-                || l.contains("tdie")
-            {
-                c.temperature()
-            } else {
-                None
-            }
-        })
-    };
-
-    // Disks (cached): refresh sizes/usage, reuse enumeration
-    let disks: Vec<DiskInfo> = {
-        let mut disks_list = state.disks.lock().await;
-        disks_list.refresh(true);
-        disks_list
-            .iter()
-            .map(|d| DiskInfo {
-                name: d.name().to_string_lossy().into_owned(),
-                total: d.total_space(),
-                available: d.available_space(),
+    // CPU temperature: only refresh sensors if cache is stale
+    let cpu_temp_c = if cached_temp().is_some() {
+        cached_temp()
+    } else if temp_enabled() {
+        let val = {
+            let mut components = state.components.lock().await;
+            components.refresh(false);
+            components.iter().find_map(|c| {
+                let l = c.label().to_ascii_lowercase();
+                if l.contains("cpu")
+                    || l.contains("package")
+                    || l.contains("tctl")
+                    || l.contains("tdie")
+                {
+                    c.temperature()
+                } else {
+                    None
+                }
             })
-            .collect()
+        };
+        set_temp(val);
+        val
+    } else {
+        None
     };
 
-    // Networks (cached): refresh counters
+    // Networks
     let networks: Vec<NetworkInfo> = {
         let mut nets = state.networks.lock().await;
-        nets.refresh(true);
+        nets.refresh(false);
         nets.iter()
             .map(|(name, data)| NetworkInfo {
                 name: name.to_string(),
@@ -80,48 +150,22 @@ pub async fn collect_metrics(state: &AppState) -> Metrics {
             .collect()
     };
 
-    // Processes: only collect fields we use (pid, name, cpu, mem), keep top K efficiently
-    const TOP_K: usize = 30;
-    let mut procs: Vec<ProcessInfo> = {
-        let sys = state.sys.lock().await; // re-lock briefly to read processes
-        sys.processes()
-            .values()
-            .map(|p| ProcessInfo {
-                pid: p.pid().as_u32(),
-                name: p.name().to_string_lossy().into_owned(),
-                cpu_usage: p.cpu_usage(),
-                mem_bytes: p.memory(),
-            })
-            .collect()
-    };
-
-    if procs.len() > TOP_K {
-        procs.select_nth_unstable_by(TOP_K, |a, b| {
-            b.cpu_usage
-                .partial_cmp(&a.cpu_usage)
-                .unwrap_or(Ordering::Equal)
-        });
-        procs.truncate(TOP_K);
-    }
-    procs.sort_by(|a, b| {
-        b.cpu_usage
-            .partial_cmp(&a.cpu_usage)
-            .unwrap_or(Ordering::Equal)
-    });
-
-    let process_count = {
-        let sys = state.sys.lock().await;
-        sys.processes().len()
-    };
-
-    // GPU(s)
-    let gpus = match collect_all_gpus() {
-        Ok(v) if !v.is_empty() => Some(v),
-        Ok(_) => None,
-        Err(e) => {
-            warn!("gpu collection failed: {e}");
-            None
-        }
+    // GPUs: refresh only when cache is stale
+    let gpus = if cached_gpus().is_some() {
+        cached_gpus()
+    } else if gpu_enabled() {
+        let v = match collect_all_gpus() {
+            Ok(v) if !v.is_empty() => Some(v),
+            Ok(_) => None,
+            Err(e) => {
+                warn!("gpu collection failed: {e}");
+                None
+            }
+        };
+        set_gpus(v.clone());
+        v
+    } else {
+        None
     };
 
     Metrics {
@@ -131,12 +175,153 @@ pub async fn collect_metrics(state: &AppState) -> Metrics {
         mem_used,
         swap_total,
         swap_used,
-        process_count,
         hostname,
         cpu_temp_c,
-        disks,
+        disks: Vec::new(),
         networks,
-        top_processes: procs,
+        top_processes: Vec::new(),
         gpus,
     }
 }
+
+// Cached disks
+pub async fn collect_disks(state: &AppState) -> Vec<DiskInfo> {
+    let mut disks_list = state.disks.lock().await;
+    disks_list.refresh(false); // don't drop missing disks
+    disks_list
+        .iter()
+        .map(|d| DiskInfo {
+            name: d.name().to_string_lossy().into_owned(),
+            total: d.total_space(),
+            available: d.available_space(),
+        })
+        .collect()
+}
+
+#[inline]
+fn read_total_jiffies() -> io::Result<u64> {
+    // /proc/stat first line: "cpu  user nice system idle iowait irq softirq steal ..."
+    let s = fs::read_to_string("/proc/stat")?;
+    if let Some(line) = s.lines().next() {
+        let mut it = line.split_whitespace();
+        let _cpu = it.next(); // "cpu"
+        let mut sum: u64 = 0;
+        for tok in it.take(8) {
+            if let Ok(v) = tok.parse::<u64>() {
+                sum = sum.saturating_add(v);
+            }
+        }
+        return Ok(sum);
+    }
+    Err(io::Error::other("no cpu line"))
+}
+
+#[inline]
+fn read_proc_jiffies(pid: u32) -> Option<u64> {
+    let path = format!("/proc/{pid}/stat");
+    let s = fs::read_to_string(path).ok()?;
+    // Find the right parenthesis that terminates comm; everything after is space-separated fields starting at "state"
+    let rpar = s.rfind(')')?;
+    let after = s.get(rpar + 2..)?; // skip ") "
+    let mut it = after.split_whitespace();
+    // utime (14th field) is offset 11 from "state", stime (15th) is next
+    let utime = it.nth(11)?.parse::<u64>().ok()?;
+    let stime = it.next()?.parse::<u64>().ok()?;
+    Some(utime.saturating_add(stime))
+}
+
+// Replace the body of collect_processes_top_k to use /proc deltas.
+// This makes CPU% = (delta_proc / delta_total) * 100 over the 2s interval.
+pub async fn collect_processes_top_k(state: &AppState, k: usize) -> ProcessesPayload {
+    // Fresh view to avoid lingering entries and select "no tasks" (no per-thread rows).
+    // Only processes, no per-thread entries.
+    let mut sys = System::new();
+    sys.refresh_processes_specifics(
+        ProcessesToUpdate::All,
+        false,
+        ProcessRefreshKind::everything().without_tasks(),
+    );
+
+    let total_count = sys.processes().len();
+
+    // Snapshot current per-pid jiffies
+    let mut current: HashMap<u32, u64> = HashMap::with_capacity(total_count);
+    for p in sys.processes().values() {
+        let pid = p.pid().as_u32();
+        if let Some(j) = read_proc_jiffies(pid) {
+            current.insert(pid, j);
+        }
+    }
+    let total_now = read_total_jiffies().unwrap_or(0);
+
+    // Compute deltas vs last sample
+    let (last_total, mut last_map) = {
+        let mut t = state.proc_cpu.lock().await;
+        let lt = t.last_total;
+        let lm = std::mem::take(&mut t.last_per_pid);
+        t.last_total = total_now;
+        t.last_per_pid = current.clone();
+        (lt, lm)
+    };
+
+    // On first run or if total delta is tiny, report zeros
+    if last_total == 0 || total_now <= last_total {
+        let procs: Vec<ProcessInfo> = sys
+            .processes()
+            .values()
+            .map(|p| ProcessInfo {
+                pid: p.pid().as_u32(),
+                name: p.name().to_string_lossy().into_owned(),
+                cpu_usage: 0.0,
+                mem_bytes: p.memory(),
+            })
+            .collect();
+        return ProcessesPayload {
+            process_count: total_count,
+            top_processes: top_k_sorted(procs, k),
+        };
+    }
+
+    let dt = total_now.saturating_sub(last_total).max(1) as f32;
+
+    let procs: Vec<ProcessInfo> = sys
+        .processes()
+        .values()
+        .map(|p| {
+            let pid = p.pid().as_u32();
+            let now = current.get(&pid).copied().unwrap_or(0);
+            let prev = last_map.remove(&pid).unwrap_or(0);
+            let du = now.saturating_sub(prev) as f32;
+            let cpu = ((du / dt) * 100.0).clamp(0.0, 100.0);
+            ProcessInfo {
+                pid,
+                name: p.name().to_string_lossy().into_owned(),
+                cpu_usage: cpu,
+                mem_bytes: p.memory(),
+            }
+        })
+        .collect();
+
+    ProcessesPayload {
+        process_count: total_count,
+        top_processes: top_k_sorted(procs, k),
+    }
+}
+
+// Small helper to select and sort top-k by cpu
+fn top_k_sorted(mut v: Vec<ProcessInfo>, k: usize) -> Vec<ProcessInfo> {
+    if v.len() > k {
+        v.select_nth_unstable_by(k, |a, b| {
+            b.cpu_usage
+                .partial_cmp(&a.cpu_usage)
+                .unwrap_or(std::cmp::Ordering::Equal)
+        });
+        v.truncate(k);
+    }
+    v.sort_by(|a, b| {
+        b.cpu_usage
+            .partial_cmp(&a.cpu_usage)
+            .unwrap_or(std::cmp::Ordering::Equal)
+    });
+    v
+}

socktop_agent/src/sampler.rs

@@ -1,39 +1,34 @@
-//! Background sampler: periodically collects metrics and updates a JSON cache,
-//! so WS replies are just a read of the cached string.
+//! Background sampler: periodically collects metrics and updates precompressed caches,
+//! so WS replies just read and send cached bytes.
 
-use crate::metrics::collect_metrics;
 use crate::state::AppState;
-//use serde_json::to_string;
 use tokio::task::JoinHandle;
-use tokio::time::{interval, Duration, MissedTickBehavior};
+use tokio::time::{sleep, Duration};
 
-pub fn spawn_sampler(state: AppState, period: Duration) -> JoinHandle<()> {
+// 500ms: fast path (cpu/mem/net/temp/gpu)
+pub fn spawn_sampler(_state: AppState, _period: Duration) -> JoinHandle<()> {
     tokio::spawn(async move {
-        let idle_period = Duration::from_secs(10);
+        // no-op background sampler (request-driven collection elsewhere)
         loop {
-            let active = state
-                .client_count
-                .load(std::sync::atomic::Ordering::Relaxed)
-                > 0;
-            let mut ticker = interval(if active { period } else { idle_period });
-            ticker.set_missed_tick_behavior(MissedTickBehavior::Skip);
-            ticker.tick().await;
-
-            if !active {
-                tokio::select! {
-                    _ = ticker.tick() => {},
-                    _ = state.wake_sampler.notified() => continue,
-                }
-            }
-
-            if let Ok(json) = async {
-                let m = collect_metrics(&state).await;
-                serde_json::to_string(&m)
-            }
-            .await
-            {
-                *state.last_json.write().await = json;
-            }
+            sleep(Duration::from_secs(3600)).await;
+        }
+    })
+}
+
+// 2s: processes top-k
+pub fn spawn_process_sampler(_state: AppState, _period: Duration, _top_k: usize) -> JoinHandle<()> {
+    tokio::spawn(async move {
+        loop {
+            sleep(Duration::from_secs(3600)).await;
+        }
+    })
+}
+
+// 5s: disks
+pub fn spawn_disks_sampler(_state: AppState, _period: Duration) -> JoinHandle<()> {
+    tokio::spawn(async move {
+        loop {
+            sleep(Duration::from_secs(3600)).await;
         }
     })
 }

socktop_agent/src/state.rs

@@ -1,49 +1,52 @@
 //! Shared agent state: sysinfo handles and hot JSON cache.
 
+use std::collections::HashMap;
 use std::sync::atomic::AtomicUsize;
 use std::sync::Arc;
 use sysinfo::{Components, Disks, Networks, System};
-use tokio::sync::{Mutex, Notify, RwLock};
+use tokio::sync::Mutex;
 
 pub type SharedSystem = Arc<Mutex<System>>;
 pub type SharedComponents = Arc<Mutex<Components>>;
 pub type SharedDisks = Arc<Mutex<Disks>>;
 pub type SharedNetworks = Arc<Mutex<Networks>>;
 
+#[derive(Default)]
+pub struct ProcCpuTracker {
+    pub last_total: u64,
+    pub last_per_pid: HashMap<u32, u64>,
+}
+
 #[derive(Clone)]
 pub struct AppState {
-    // Persistent sysinfo handles
     pub sys: SharedSystem,
-
-    // Last serialized JSON snapshot for fast WS responses
-    pub last_json: Arc<RwLock<String>>,
-
-    // Adaptive sampling controls
-    pub client_count: Arc<AtomicUsize>,
-    pub wake_sampler: Arc<Notify>,
-    pub auth_token: Option<String>,
-
-    // Cached containers (enumerated once; refreshed per tick)
     pub components: SharedComponents,
     pub disks: SharedDisks,
     pub networks: SharedNetworks,
+
+    // For correct per-process CPU% using /proc deltas
+    pub proc_cpu: Arc<Mutex<ProcCpuTracker>>,
+
+    // Connection tracking (to allow future idle sleeps if desired)
+    pub client_count: Arc<AtomicUsize>,
+
+    pub auth_token: Option<String>,
 }
 
 impl AppState {
-    #[allow(dead_code)]
     pub fn new() -> Self {
-        let sys = System::new(); // targeted refreshes per tick
-        let components = Components::new_with_refreshed_list(); // enumerate once
+        let sys = System::new();
+        let components = Components::new_with_refreshed_list();
         let disks = Disks::new_with_refreshed_list();
         let networks = Networks::new_with_refreshed_list();
+
         Self {
             sys: Arc::new(Mutex::new(sys)),
             components: Arc::new(Mutex::new(components)),
             disks: Arc::new(Mutex::new(disks)),
             networks: Arc::new(Mutex::new(networks)),
-            last_json: Arc::new(RwLock::new(String::new())),
+            proc_cpu: Arc::new(Mutex::new(ProcCpuTracker::default())),
             client_count: Arc::new(AtomicUsize::new(0)),
-            wake_sampler: Arc::new(Notify::new()),
             auth_token: std::env::var("SOCKTOP_TOKEN")
                 .ok()
                 .filter(|s| !s.is_empty()),

socktop_agent/src/types.rs

@@ -4,7 +4,21 @@
 use crate::gpu::GpuMetrics;
 use serde::Serialize;
 
-#[derive(Debug, Serialize, Clone)]
+#[derive(Debug, Clone, Serialize)]
+pub struct DiskInfo {
+    pub name: String,
+    pub total: u64,
+    pub available: u64,
+}
+
+#[derive(Debug, Clone, Serialize)]
+pub struct NetworkInfo {
+    pub name: String,
+    pub received: u64,
+    pub transmitted: u64,
+}
+
+#[derive(Debug, Clone, Serialize)]
 pub struct ProcessInfo {
     pub pid: u32,
     pub name: String,
@@ -12,22 +26,7 @@ pub struct ProcessInfo {
     pub mem_bytes: u64,
 }
 
-#[derive(Debug, Serialize, Clone)]
-pub struct DiskInfo {
-    pub name: String,
-    pub total: u64,
-    pub available: u64,
-}
-
-#[derive(Debug, Serialize, Clone)]
-pub struct NetworkInfo {
-    pub name: String,
-    // cumulative totals since the agent started (client should diff to get rates)
-    pub received: u64,
-    pub transmitted: u64,
-}
-
-#[derive(Serialize)]
+#[derive(Debug, Clone, Serialize)]
 pub struct Metrics {
     pub cpu_total: f32,
     pub cpu_per_core: Vec<f32>,
@@ -35,7 +34,6 @@ pub struct Metrics {
     pub mem_used: u64,
     pub swap_total: u64,
     pub swap_used: u64,
-    pub process_count: usize,
     pub hostname: String,
     pub cpu_temp_c: Option<f32>,
     pub disks: Vec<DiskInfo>,
@@ -43,3 +41,9 @@ pub struct Metrics {
     pub top_processes: Vec<ProcessInfo>,
     pub gpus: Option<Vec<GpuMetrics>>,
 }
+
+#[derive(Debug, Clone, Serialize)]
+pub struct ProcessesPayload {
+    pub process_count: usize,
+    pub top_processes: Vec<ProcessInfo>,
+}

socktop_agent/src/ws.rs

@@ -1,66 +1,69 @@
-//! WebSocket upgrade and per-connection handler. Serves cached JSON quickly.
+//! WebSocket upgrade and per-connection handler (request-driven).
 
 use axum::{
-    extract::{
-        ws::{Message, WebSocket, WebSocketUpgrade},
-        Query, State,
-    },
-    http::StatusCode,
-    response::{IntoResponse, Response},
+    extract::ws::{Message, WebSocket},
+    extract::{Query, State, WebSocketUpgrade},
+    response::Response,
 };
-use futures_util::stream::StreamExt;
-
-use crate::metrics::collect_metrics;
-use crate::state::AppState;
-
+use flate2::{write::GzEncoder, Compression};
+use futures_util::StreamExt;
 use std::collections::HashMap;
-use std::sync::atomic::Ordering;
+use std::io::Write;
+
+use crate::metrics::{collect_disks, collect_fast_metrics, collect_processes_top_k};
+use crate::state::AppState;
 
 pub async fn ws_handler(
     ws: WebSocketUpgrade,
     State(state): State<AppState>,
     Query(q): Query<HashMap<String, String>>,
 ) -> Response {
+    // optional auth
     if let Some(expected) = state.auth_token.as_ref() {
-        match q.get("token") {
-            Some(t) if t == expected => {}
-            _ => return StatusCode::UNAUTHORIZED.into_response(),
+        if q.get("token") != Some(expected) {
+            return ws.on_upgrade(|socket| async move {
+                let _ = socket.close().await;
+            });
         }
     }
     ws.on_upgrade(move |socket| handle_socket(socket, state))
 }
 
 async fn handle_socket(mut socket: WebSocket, state: AppState) {
-    // Bump client count on connect and wake the sampler.
-    state.client_count.fetch_add(1, Ordering::Relaxed);
-    state.wake_sampler.notify_waiters();
-
-    // Ensure we decrement on disconnect (drop).
-    struct ClientGuard(AppState);
-    impl Drop for ClientGuard {
-        fn drop(&mut self) {
-            self.0.client_count.fetch_sub(1, Ordering::Relaxed);
-            self.0.wake_sampler.notify_waiters();
-        }
-    }
-    let _guard = ClientGuard(state.clone());
-
+    state
+        .client_count
+        .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
     while let Some(Ok(msg)) = socket.next().await {
         match msg {
-            Message::Text(text) if text == "get_metrics" => {
-                // Serve the cached JSON quickly; if empty (cold start), collect once.
-                let cached = state.last_json.read().await.clone();
-                if !cached.is_empty() {
-                    let _ = socket.send(Message::Text(cached)).await;
-                } else {
-                    let metrics = collect_metrics(&state).await;
-                    if let Ok(js) = serde_json::to_string(&metrics) {
-                        let _ = socket.send(Message::Text(js)).await;
-                    }
-                }
+            Message::Text(ref text) if text == "get_metrics" => {
+                let m = collect_fast_metrics(&state).await;
+                let _ = send_json(&mut socket, &m).await;
+            }
+            Message::Text(ref text) if text == "get_disks" => {
+                let d = collect_disks(&state).await;
+                let _ = send_json(&mut socket, &d).await;
+            }
+            Message::Text(ref text) if text == "get_processes" => {
+                let p = collect_processes_top_k(&state, 50).await;
+                let _ = send_json(&mut socket, &p).await;
             }
             Message::Close(_) => break,
             _ => {}
         }
     }
+    state
+        .client_count
+        .fetch_sub(1, std::sync::atomic::Ordering::Relaxed);
+}
+
+// Small, cheap gzip for larger payloads; send text for small.
+async fn send_json<T: serde::Serialize>(ws: &mut WebSocket, value: &T) -> Result<(), axum::Error> {
+    let json = serde_json::to_string(value).expect("serialize");
+    if json.len() <= 768 {
+        return ws.send(Message::Text(json)).await;
+    }
+    let mut enc = GzEncoder::new(Vec::new(), Compression::fast());
+    enc.write_all(json.as_bytes()).ok();
+    let bin = enc.finish().unwrap_or_else(|_| json.into_bytes());
+    ws.send(Message::Binary(bin)).await
 }