feat(obs): report broadcast frame counters

PLAN.md calls out "Broadcast traffic flowing" as a user-facing diagnostic.
The tunnel stats only reported total Ethernet frame counts, so the client
could not distinguish whether broadcast traffic was actually crossing the
tunnel.

Add defaulted broadcast tx/rx counters to TunnelStats while preserving the
existing constructor and old JSON compatibility. Client and gateway accounting
now increments those counters from validated Ethernet frames, and the client
diagnostics line reports the broadcast flow next to total frame counts.

Relay peer stats logs include the new counters so operators can see broadcast
activity from forwarded stats snapshots too.

Test Plan:
- cargo fmt --check
- cargo test -p lanparty-obs -p lanparty-client-core -p lanparty-gateway \
  -p lanparty-client-win -p lanparty-relay
- cargo test --workspace
- cargo clippy --workspace --all-targets -- -D warnings
- git diff --check

Refs: PLAN.md

crates/lanparty-gateway/src/lib.rs

@@ -366,10 +366,13 @@ fn send_gateway_ethernet(
     stats: &GatewayTunnelStats,
     frame: &[u8],
 ) -> Result<()> {
-    if let Err(error) = EthernetFrame::parse(frame) {
-        stats.record_malformed_frame();
-        return Err(error).context("gateway Ethernet frame is malformed");
-    }
+    let ethernet_frame = match EthernetFrame::parse(frame) {
+        Ok(frame) => frame,
+        Err(error) => {
+            stats.record_malformed_frame();
+            return Err(error).context("gateway Ethernet frame is malformed");
+        }
+    };
     let datagram = encode_datagram(
         FrameType::Ethernet,
         welcome.room_id(),
@@ -382,7 +385,7 @@ fn send_gateway_ethernet(
     connection
         .send_datagram(Bytes::from(datagram))
         .context("failed to send gateway Ethernet datagram")?;
-    stats.record_ethernet_tx();
+    stats.record_ethernet_tx(ethernet_frame);
 
     Ok(())
 }
@@ -407,12 +410,15 @@ async fn recv_gateway_ethernet(
             stats.record_dropped_frame();
             continue;
         }
-        if EthernetFrame::parse(packet.payload()).is_err() {
-            stats.record_malformed_frame();
-            continue;
-        }
+        let ethernet_frame = match EthernetFrame::parse(packet.payload()) {
+            Ok(frame) => frame,
+            Err(_) => {
+                stats.record_malformed_frame();
+                continue;
+            }
+        };
 
-        stats.record_ethernet_rx();
+        stats.record_ethernet_rx(ethernet_frame);
         return Ok(ReceivedEthernetFrame {
             source_peer_id: header.peer_id(),
             payload: Bytes::copy_from_slice(packet.payload()),
@@ -459,6 +465,8 @@ async fn send_gateway_control_event(
 struct GatewayTunnelStats {
     ethernet_frames_tx: AtomicU64,
     ethernet_frames_rx: AtomicU64,
+    broadcast_frames_tx: AtomicU64,
+    broadcast_frames_rx: AtomicU64,
     datagrams_tx: AtomicU64,
     datagrams_rx: AtomicU64,
     dropped_frames: AtomicU64,
@@ -466,13 +474,19 @@ struct GatewayTunnelStats {
 }
 
 impl GatewayTunnelStats {
-    fn record_ethernet_tx(&self) {
+    fn record_ethernet_tx(&self, frame: EthernetFrame<'_>) {
         self.ethernet_frames_tx.fetch_add(1, Ordering::Relaxed);
+        if frame.is_broadcast() {
+            self.broadcast_frames_tx.fetch_add(1, Ordering::Relaxed);
+        }
         self.datagrams_tx.fetch_add(1, Ordering::Relaxed);
     }
 
-    fn record_ethernet_rx(&self) {
+    fn record_ethernet_rx(&self, frame: EthernetFrame<'_>) {
         self.ethernet_frames_rx.fetch_add(1, Ordering::Relaxed);
+        if frame.is_broadcast() {
+            self.broadcast_frames_rx.fetch_add(1, Ordering::Relaxed);
+        }
     }
 
     fn record_datagram_rx(&self) {
@@ -497,6 +511,10 @@ impl GatewayTunnelStats {
             self.dropped_frames.load(Ordering::Relaxed),
             self.malformed_frames.load(Ordering::Relaxed),
         )
+        .with_broadcast_frames(
+            self.broadcast_frames_tx.load(Ordering::Relaxed),
+            self.broadcast_frames_rx.load(Ordering::Relaxed),
+        )
     }
 }
 
@@ -986,6 +1004,31 @@ mod tests {
             .unwrap();
     }
 
+    #[test]
+    fn snapshots_gateway_broadcast_stats() {
+        let stats = GatewayTunnelStats::default();
+        let broadcast_tx_bytes = broadcast_ethernet_frame(b"broadcast tx");
+        let broadcast_rx_bytes = broadcast_ethernet_frame(b"broadcast rx");
+        let broadcast_tx = EthernetFrame::parse(&broadcast_tx_bytes).unwrap();
+        let broadcast_rx = EthernetFrame::parse(&broadcast_rx_bytes).unwrap();
+
+        stats.record_ethernet_tx(broadcast_tx);
+        stats.record_datagram_rx();
+        stats.record_ethernet_rx(broadcast_rx);
+        stats.record_dropped_frame();
+        stats.record_malformed_frame();
+        let snapshot = stats.snapshot();
+
+        assert_eq!(snapshot.ethernet_frames_tx(), 1);
+        assert_eq!(snapshot.ethernet_frames_rx(), 1);
+        assert_eq!(snapshot.broadcast_frames_tx(), 1);
+        assert_eq!(snapshot.broadcast_frames_rx(), 1);
+        assert_eq!(snapshot.datagrams_tx(), 1);
+        assert_eq!(snapshot.datagrams_rx(), 1);
+        assert_eq!(snapshot.dropped_frames(), 2);
+        assert_eq!(snapshot.malformed_frames(), 1);
+    }
+
     #[cfg(target_os = "linux")]
     #[test]
     fn builds_padded_cam_refresh_frame() {
@@ -1106,4 +1149,13 @@ mod tests {
         frame.extend_from_slice(payload);
         frame
     }
+
+    fn broadcast_ethernet_frame(payload: &[u8]) -> Vec<u8> {
+        let mut frame = Vec::new();
+        frame.extend_from_slice(&MacAddr::BROADCAST.octets());
+        frame.extend_from_slice(&[0x02, 0, 0, 0, 0, 1]);
+        frame.extend_from_slice(&0x0800_u16.to_be_bytes());
+        frame.extend_from_slice(payload);
+        frame
+    }
 }