feat(server): network traffic, every interface, CPU threads

crates/sbm_ffi/src/api/parser.rs

@@ -45,16 +45,6 @@ pub fn parse_status_json(
     serde_json::to_string(&status).map_err(|e| e.to_string())
 }
 
-/// Windows WMI double-sample net speed delta, returning `[{name, rx, tx}]` JSON (bytes/sec)
-#[flutter_rust_bridge::frb(sync)]
-pub fn parse_windows_net_speed_json(raw: String) -> String {
-    let speeds: Vec<_> = sbm_parser::windows::parse_net_speed(&raw)
-        .into_iter()
-        .map(|(name, rx, tx)| serde_json::json!({ "name": name, "rx": rx, "tx": tx }))
-        .collect();
-    serde_json::Value::Array(speeds).to_string()
-}
-
 /// Per-platform collection command manifest (the app's script generation derives from it, versioned with the parsers)
 #[flutter_rust_bridge::frb(sync)]
 pub fn command_specs(system: String) -> Result<Vec<CommandSpec>, String> {

crates/sbm_parser/src/lib.rs

@@ -211,9 +211,8 @@ pub fn parse_status_opts(
             status.batteries = windows::parse_batteries(get(commands::BATTERY));
             status.sensors = windows::parse_sensors(get(commands::SENSORS));
             status.disk_smart = windows::parse_disk_smart(get(commands::DISK_SMART));
-            // NET is a WMI double sample: cumulative counters go into status.net
-            // (consistent with other platforms); instantaneous rates are separately
-            // derived by windows::parse_net_speed deltas
+            // Cumulative counters, as on the other platforms: the caller derives
+            // rates across polls
             status.net = windows::parse_net(get(commands::NET));
             status.diskio = windows::parse_diskio(get(commands::DISKIO));
         }

crates/sbm_parser/src/windows.rs

@@ -178,15 +178,7 @@ pub fn parse_temps(raw: &str) -> Temperatures {
     temps
 }
 
-/// WMI raw perf-counter two-sample delta (Dart `_parseWindowsWmiDelta`):
-/// input JSON contains >= 2 sample groups; returns (name, rx_speed, tx_speed) in bytes/sec.
-/// `_Total` and empty names are skipped; timestamps are in 100ns units
-pub fn parse_net_speed(raw: &str) -> Vec<(String, f64, f64)> {
-    parse_wmi_delta(raw, "BytesReceivedPersec", "BytesSentPersec")
-}
-
-/// Same WMI sampling as `parse_net_speed`, taking the raw cumulative counters of
-/// the last group: the `*Persec` fields of Win32_PerfRawData are actually cumulative
+/// The WMI sample's raw cumulative counters, of the last group: the `*Persec` fields of Win32_PerfRawData are actually cumulative
 /// byte counts. Produces NetIface cumulative counters consistent with Linux/BSD;
 /// rates are derived by the caller via cross-sample deltas
 pub fn parse_net(raw: &str) -> Vec<NetIface> {
@@ -218,51 +210,6 @@ pub fn parse_net(raw: &str) -> Vec<NetIface> {
         .collect()
 }
 
-fn parse_wmi_delta(raw: &str, field1: &str, field2: &str) -> Vec<(String, f64, f64)> {
-    let Some(Value::Array(samples)) = decode(raw) else {
-        return Vec::new();
-    };
-    if samples.len() < 2 {
-        return Vec::new();
-    }
-    let unwrap_value = |v: &Value| -> Value {
-        match v.get("value") {
-            Some(inner) => inner.clone(),
-            None => v.clone(),
-        }
-    };
-    let (s1, s2) = (
-        unwrap_value(&samples[samples.len() - 2]),
-        unwrap_value(&samples[samples.len() - 1]),
-    );
-    let (Value::Array(list1), Value::Array(list2)) = (s1, s2) else {
-        return Vec::new();
-    };
-    if list1.len() != list2.len() {
-        return Vec::new();
-    }
-
-    let num = |v: &Value, key: &str| v[key].as_f64().unwrap_or(0.0);
-    let mut result = Vec::new();
-    for (a, b) in list1.iter().zip(list2.iter()) {
-        let name = a["Name"].as_str().unwrap_or("");
-        if name.is_empty() || name == "_Total" {
-            continue;
-        }
-        let time_delta = (num(b, "Timestamp_Sys100NS") - num(a, "Timestamp_Sys100NS")) / 1e7;
-        if time_delta <= 0.0 {
-            continue;
-        }
-        let d1 = num(b, field1) - num(a, field1);
-        let d2 = num(b, field2) - num(a, field2);
-        if d1 < 0.0 || d2 < 0.0 {
-            continue;
-        }
-        result.push((name.to_string(), d1 / time_delta, d2 / time_delta));
-    }
-    result
-}
-
 /// Win32_Battery JSON (Dart `_parseWindowsBatteries`).
 ///
 /// `BatteryStatus` follows Win32_Battery's enumeration: 3 is fully charged,

crates/sbm_parser/tests/dart_compat.rs

@@ -919,41 +919,6 @@ fn windows_real_net_totals() {
     }
 }
 
-/// Real WMI double-sample output: 5 NICs, `{"value": [...], "Count"}` wrapper form
-#[test]
-fn windows_real_net_speed() {
-    let speeds = windows::parse_net_speed(include_str!("fixtures/win_net.json"));
-    assert_eq!(speeds.len(), 5);
-    for (name, rx, tx) in &speeds {
-        assert!(!name.is_empty() && name != "_Total");
-        assert!(*rx >= 0.0 && *tx >= 0.0, "{}: rx={} tx={}", name, rx, tx);
-    }
-    // The active NIC (Wi-Fi) is present
-    assert!(speeds.iter().any(|(n, _, _)| n.contains("Wi-Fi")));
-}
-
-/// Dart `_parseWindowsWmiDelta`: two-sample delta, `_Total` skipped, 100ns timestamps
-#[test]
-fn windows_parse_net_speed_delta() {
-    let raw = r#"[
-        [
-            {"Name": "_Total", "BytesReceivedPersec": 0, "BytesSentPersec": 0, "Timestamp_Sys100NS": 10000000},
-            {"Name": "Ethernet", "BytesReceivedPersec": 1000, "BytesSentPersec": 500, "Timestamp_Sys100NS": 10000000}
-        ],
-        [
-            {"Name": "_Total", "BytesReceivedPersec": 99, "BytesSentPersec": 99, "Timestamp_Sys100NS": 20000000},
-            {"Name": "Ethernet", "BytesReceivedPersec": 3000, "BytesSentPersec": 1500, "Timestamp_Sys100NS": 20000000}
-        ]
-    ]"#;
-    let speeds = windows::parse_net_speed(raw);
-    assert_eq!(speeds.len(), 1);
-    let (name, rx, tx) = &speeds[0];
-    assert_eq!(name, "Ethernet");
-    // 1s interval (10000000 * 100ns), deltas 2000/1000
-    assert_eq!(*rx, 2000.0);
-    assert_eq!(*tx, 1000.0);
-}
-
 // ---------- Conn:conn_test.dart ----------
 
 /// Dart 'Conn.parse reads MaxConn and AttemptFails from /proc/net/snmp'

crates/sbm_parser/tests/ssh_e2e.rs

@@ -939,6 +939,13 @@ fn ssh_e2e_windows_script_parse_matches_direct_commands() {
         None,
     )
     .expect("run status script");
+    // A second poll, for the net counters to be compared across the two
+    let raw_next = ssh_raw(
+        &host,
+        &script::exec_command(SystemType::Windows, &path, ShellFunc::Status),
+        None,
+    )
+    .expect("run status script again");
     let _ = ssh_raw(
         &host,
         &script::encoded_powershell_command(&format!("Remove-Item -Recurse -Force '{dir}'")),
@@ -965,6 +972,27 @@ fn ssh_e2e_windows_script_parse_matches_direct_commands() {
         "net counters parsed from WMI double sample"
     );
 
+    // Counters since boot, as on Unix: none goes down between two polls, and a
+    // host with a session open moves some bytes. Rates would do neither
+    // reliably — which is what the app once read these as.
+    let next = sbm_parser::parse_status(
+        SystemType::Windows,
+        &script::parse_script_output(&raw_next),
+    );
+    let mut grew = false;
+    for a in &status.net {
+        let Some(b) = next.net.iter().find(|b| b.device == a.device) else {
+            continue;
+        };
+        assert!(
+            b.rx_bytes >= a.rx_bytes && b.tx_bytes >= a.tx_bytes,
+            "{}: counters went down: {a:?} -> {b:?}",
+            a.device
+        );
+        grew |= b.rx_bytes > a.rx_bytes || b.tx_bytes > a.tx_bytes;
+    }
+    assert!(grew, "no interface counted any traffic between two polls");
+
     let remote_epoch: i64 = segments["time"]
         .trim()
         .parse()
@@ -1019,13 +1047,13 @@ fn ssh_e2e_windows_script_parse_matches_direct_commands() {
     assert_eq!(script_names, direct_names, "GPU names: script vs direct");
 
     eprintln!(
-        "windows ssh e2e ok: host={:?} sys={:?} cores={} mem_total={}KiB disks={} net_ifaces={} gpus={}",
+        "windows ssh e2e ok: host={:?} sys={:?} cores={} mem_total={}KiB disks={} net_ifaces={:?} gpus={}",
         status.host,
         status.sys,
         status.cpu.len(),
         mem.total,
         status.disks.len(),
-        status.net.len(),
+        status.net.iter().map(|n| &n.device).collect::<Vec<_>>(),
         status.nvidia.len()
     );
 }

lib/data/model/server/net_speed.dart

@@ -31,6 +31,7 @@ class NetSpeed extends TimeSeq<NetSpeedPart> {
   NetSpeed.copy(NetSpeed source) : super.copy(source) {
     devices.addAll(source.devices);
     realIfaces.addAll(source.realIfaces);
+    ifaces.addAll(source.ifaces);
     _realIfaceIndices.addAll(source._realIfaceIndices);
     cachedVals = source.cachedVals;
   }
@@ -54,11 +55,26 @@ class NetSpeed extends TimeSeq<NetSpeedPart> {
 
     realIfaces.clear();
     _realIfaceIndices.clear();
+    ifaces.clear();
+    final others = <int>[];
     for (var i = 0; i < devices.length; i++) {
       final dev = devices[i];
+      if (isLoopback(dev)) continue;
+      ifaces.add(dev);
       if (realIfacePrefixs.any((prefix) => dev.startsWith(prefix))) {
         realIfaces.add(dev);
         _realIfaceIndices.add(i);
+      } else {
+        others.add(i);
+      }
+    }
+    // Windows names an interface by its adapter ("Realtek PCIe GbE Family
+    // Controller", or "Ethernet" through an agent), which no prefix matches.
+    // Totals of nothing read as no traffic at all; every interface is closer.
+    if (realIfaces.isEmpty) {
+      for (final i in others) {
+        realIfaces.add(devices[i]);
+        _realIfaceIndices.add(i);
       }
     }
 
@@ -77,9 +93,23 @@ class NetSpeed extends TimeSeq<NetSpeedPart> {
   /// Non-virtual network device prefix
   static const realIfacePrefixs = ['eth', 'wlan', 'en', 'ww', 'wl'];
 
-  /// Cached non-virtual network device prefix
+  /// `lo` (Linux), `lo0` (BSD, macOS), `Loopback Pseudo-Interface 1`
+  /// (Windows).
+  static bool isLoopback(String dev) {
+    final name = dev.toLowerCase();
+    return name.startsWith('loopback') || _loopbackName.hasMatch(name);
+  }
+
+  static final _loopbackName = RegExp(r'^lo\d*$');
+
+  /// The interfaces the totals add up: those matching [realIfacePrefixs], or
+  /// every one but loopback where none does.
   final realIfaces = <String>[];
 
+  /// Every interface but loopback: what a reader can pick from, bridges and
+  /// tunnels included. The totals stay the sum of [realIfaces].
+  final ifaces = <String>[];
+
   /// Cached indices of real (non-virtual) interfaces in [devices]
   final _realIfaceIndices = <int>[];
 
@@ -99,6 +129,7 @@ class NetSpeed extends TimeSeq<NetSpeedPart> {
   double? _speed(int i, BigInt Function(NetSpeedPart) counter) {
     final elapsed = _elapsed;
     if (elapsed == null || i >= now.length || i >= pre.length) return null;
+    if (appeared(i)) return null;
     final delta = counterDeltaBig(counter(pre[i]), counter(now[i]));
     if (delta == null) return null;
     return delta.toDouble() / elapsed;
@@ -136,26 +167,28 @@ class NetSpeed extends TimeSeq<NetSpeedPart> {
   String speedOut({String? device}) =>
       _fmtSpeed(speedOutBytesOf(device: device));
 
-  String sizeIn({String? device}) {
-    if (now.isEmpty) return noReading;
-    if (device != null) return sizeInBytes(deviceIdx(device)).bytes2Str;
-    var size = BigInt.zero;
-    for (final i in _realIfaceIndices) {
-      size += sizeInBytes(i);
-    }
-    return size.bytes2Str;
-  }
+  /// Bytes moved since boot, summed over real interfaces when [device] is
+  /// null. `null` before the first sample.
+  BigInt? sizeInBytesOf({String? device}) => _total(device, sizeInBytes);
+
+  BigInt? sizeOutBytesOf({String? device}) => _total(device, sizeOutBytes);
 
-  String sizeOut({String? device}) {
-    if (now.isEmpty) return noReading;
-    if (device != null) return sizeOutBytes(deviceIdx(device)).bytes2Str;
+  BigInt? _total(String? device, BigInt Function(int) of) {
+    if (now.isEmpty) return null;
+    if (device != null) return of(deviceIdx(device));
     var size = BigInt.zero;
     for (final i in _realIfaceIndices) {
-      size += sizeOutBytes(i);
+      size += of(i);
     }
-    return size.bytes2Str;
+    return size;
   }
 
+  String sizeIn({String? device}) =>
+      sizeInBytesOf(device: device)?.bytes2Str ?? noReading;
+
+  String sizeOut({String? device}) =>
+      sizeOutBytesOf(device: device)?.bytes2Str ?? noReading;
+
   int deviceIdx(String? device) {
     if (device != null) {
       for (var i = 0; i < now.length; i++) {

lib/data/model/server/server_status_update_req.dart

@@ -67,7 +67,7 @@ Future<ServerStatus> getStatus(ServerStatusUpdateReq req) async {
   _apply(ss, 'mem', () => _applyMemory(ss, status));
   _apply(ss, 'swap', () => _applySwap(ss, status));
   _apply(ss, 'disk', () => _applyDisks(ss, status));
-  _apply(ss, 'net', () => _applyNet(ss, status, req, time));
+  _apply(ss, 'net', () => _applyNet(ss, status, time));
   _apply(ss, 'temps', () => _applyTemps(ss, status));
   _apply(ss, 'conn', () => _applyConn(ss, status));
   _apply(ss, 'more', () => _applyMore(ss, status));
@@ -359,41 +359,19 @@ void _applyDisks(ServerStatus ss, Map<String, dynamic> status) {
       .toList();
 }
 
-void _applyNet(
-  ServerStatus ss,
-  Map<String, dynamic> status,
-  ServerStatusUpdateReq req,
-  int time,
-) {
-  final List<NetSpeedPart> parts;
-  if (req.system == SystemType.windows) {
-    // Windows network speed comes from a two-sample WMI delta; the FFI returns
-    // the rates directly.
-    final speedsJson = ffi.parseWindowsNetSpeedJson(
-      raw: WindowsStatusCmdType.net.findInMap(req.parsedOutput),
-    );
-    parts = (jsonDecode(speedsJson) as List)
-        .map(
-          (s) => NetSpeedPart(
-            s['name'] as String,
-            BigInt.from((s['rx'] as num).toInt()),
-            BigInt.from((s['tx'] as num).toInt()),
-            time,
-          ),
-        )
-        .toList();
-  } else {
-    parts = (status['net'] as List)
-        .map(
-          (n) => NetSpeedPart(
-            n['device'] as String,
-            BigInt.from(n['rx_bytes'] as int),
-            BigInt.from(n['tx_bytes'] as int),
-            time,
-          ),
-        )
-        .toList();
-  }
+/// Cumulative counters on every system, Windows included: the rates are
+/// taken across polls by [NetSpeed].
+void _applyNet(ServerStatus ss, Map<String, dynamic> status, int time) {
+  final parts = (status['net'] as List)
+      .map(
+        (n) => NetSpeedPart(
+          n['device'] as String,
+          BigInt.from(n['rx_bytes'] as int),
+          BigInt.from(n['tx_bytes'] as int),
+          time,
+        ),
+      )
+      .toList();
   if (parts.isNotEmpty) {
     ss.netSpeed.update(parts);
   }

lib/data/model/server/time_seq.dart

@@ -72,6 +72,12 @@ abstract class TimeSeq<T extends TimeSeqIface<T>> {
     return pre != null && pre.length == _now.length && _now.isNotEmpty;
   }
 
+  /// Whether item [i] of [now] appeared this refresh. It is paired with
+  /// itself in [pre] (see [_alignPre]), so it has no baseline, and a rate a
+  /// subclass takes of it is no reading rather than a zero.
+  bool appeared(int i) =>
+      hasWindow && i < _now.length && identical(_pre![i], _now[i]);
+
   /// Called after each [update], for subclasses to refresh cached values.
   void onUpdate();
 

lib/data/provider/server/data_source.dart

@@ -80,10 +80,10 @@ extension ServerStatusHistoryX on ServerStatus {
         for (final d in diskIO.devices) d: ?diskIO.speedBytes(d).$2,
       },
       netRxs: {
-        for (final d in netSpeed.realIfaces) d: ?netSpeed.speedInBytesOf(device: d),
+        for (final d in netSpeed.ifaces) d: ?netSpeed.speedInBytesOf(device: d),
       },
       netTxs: {
-        for (final d in netSpeed.realIfaces) d: ?netSpeed.speedOutBytesOf(device: d),
+        for (final d in netSpeed.ifaces) d: ?netSpeed.speedOutBytesOf(device: d),
       },
       battery: batteries.firstOrNull?.percent?.toDouble(),
     );