fix(relay): enforce ingress datagram budget

Peers advertise a datagram budget during hello, and the relay folds that into
the room MTU/no-fragmentation model. Honest clients already avoid sending
larger encoded frames, but the relay was still trusting ingress traffic to obey
that contract before forwarding it.

Drop datagrams that exceed the accepted peer's negotiated max before decode or
forwarding, and log them as datagram_budget. This keeps malformed datagram
disconnect accounting reserved for invalid overlay/ethernet bytes instead of
policy budget drops.

Test Plan:
- cargo test -p lanparty-relay ingress_budget
- cargo test -p lanparty-relay
- cargo fmt --check
- cargo test --workspace
- cargo clippy --workspace --all-targets -- -D warnings
- git diff --check
- git diff --cached --check

Refs: MVP relay datagram budget audit

README.md

@@ -143,6 +143,8 @@ Malformed peer datagrams log their per-peer count before the relay disconnects
 peers that cross the malformed-datagram threshold.
 Relay egress skips caused by a target peer's smaller datagram budget are logged
 with the ingress peer, target peer, encoded length, and target budget.
+Ingress datagrams larger than the sending peer's negotiated datagram budget are
+dropped before decode/forwarding and logged with `reason=datagram_budget`.
 Unknown unicast from a client is forwarded only to the gateway port; unknown
 unicast from the gateway is dropped instead of flooded to every remote client.
 When a peer joins or leaves, the relay sends a reliable lifecycle control event

crates/lanparty-relay/src/server.rs

@@ -66,6 +66,7 @@ struct PeerSession {
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 enum PeerDatagramOutcome {
     Accepted,
+    Dropped(DropReason),
     Malformed,
 }
 
@@ -336,6 +337,7 @@ async fn run_peer_io(
 
                 match forward_peer_datagram(rooms, sessions, accepted, datagram).await {
                     Ok(PeerDatagramOutcome::Accepted) => {}
+                    Ok(PeerDatagramOutcome::Dropped(_reason)) => {}
                     Ok(PeerDatagramOutcome::Malformed) => {
                         if let Some(reason) = malformed_tracker.record_malformed() {
                             eprintln!(
@@ -396,6 +398,11 @@ async fn forward_peer_datagram(
     accepted: &AcceptedPeer,
     datagram: Bytes,
 ) -> Result<PeerDatagramOutcome> {
+    if datagram.len() > accepted.max_datagram_size {
+        eprintln!("{}", ingress_budget_drop_log_line(accepted, datagram.len()));
+        return Ok(PeerDatagramOutcome::Dropped(DropReason::DatagramBudget));
+    }
+
     let Ok(packet) = decode_datagram(&datagram) else {
         return Ok(PeerDatagramOutcome::Malformed);
     };
@@ -626,6 +633,17 @@ fn egress_budget_skip_log_line(
     )
 }
 
+fn ingress_budget_drop_log_line(accepted: &AcceptedPeer, datagram_len: usize) -> String {
+    format!(
+        "relay ingress dropped room={} peer_id={} role={:?} len={} max_datagram_size={} reason=datagram_budget",
+        accepted.room,
+        accepted.peer.peer_id(),
+        accepted.peer.role(),
+        datagram_len,
+        accepted.max_datagram_size
+    )
+}
+
 async fn collect_target_sessions(
     sessions: &Arc<Mutex<HashMap<PeerKey, PeerSession>>>,
     room: &RoomCode,
@@ -1158,6 +1176,23 @@ mod tests {
         assert!(line.contains("reason=datagram_budget"));
     }
 
+    #[tokio::test]
+    async fn formats_ingress_budget_drop_log_line() {
+        let rooms = Arc::new(Mutex::new(RoomRegistry::default()));
+        let mut accepted = accepted_client_for_forwarding(&rooms, client_mac(1)).await;
+        accepted.max_datagram_size = 64;
+
+        let line = ingress_budget_drop_log_line(&accepted, 65);
+
+        assert!(line.contains("relay ingress dropped"));
+        assert!(line.contains("room=TESTROOM"));
+        assert!(line.contains(&format!("peer_id={}", accepted.peer.peer_id())));
+        assert!(line.contains("role=Client"));
+        assert!(line.contains("len=65"));
+        assert!(line.contains("max_datagram_size=64"));
+        assert!(line.contains("reason=datagram_budget"));
+    }
+
     #[tokio::test]
     async fn classifies_bad_peer_datagrams_as_malformed() {
         let rooms = Arc::new(Mutex::new(RoomRegistry::default()));
@@ -1193,6 +1228,32 @@ mod tests {
         assert_eq!(outcome, PeerDatagramOutcome::Malformed);
     }
 
+    #[tokio::test]
+    async fn drops_peer_datagrams_above_negotiated_ingress_budget() {
+        let rooms = Arc::new(Mutex::new(RoomRegistry::default()));
+        let sessions = Arc::new(Mutex::new(HashMap::new()));
+        let mut accepted = accepted_client_for_forwarding(&rooms, client_mac(1)).await;
+        accepted.max_datagram_size = 32;
+        let datagram = encode_datagram(
+            FrameType::Ethernet,
+            accepted.welcome.room_id(),
+            accepted.peer.peer_id(),
+            OVERLAY_FLAGS_NONE,
+            &ethernet_frame(client_mac(2), client_mac(1)),
+        )
+        .unwrap();
+
+        assert!(datagram.len() > accepted.max_datagram_size);
+        let outcome = forward_peer_datagram(&rooms, &sessions, &accepted, Bytes::from(datagram))
+            .await
+            .unwrap();
+
+        assert_eq!(
+            outcome,
+            PeerDatagramOutcome::Dropped(DropReason::DatagramBudget)
+        );
+    }
+
     #[tokio::test]
     async fn forwards_ethernet_datagrams_between_joined_peers() {
         let (server, certificate) = bind_test_server(DEFAULT_MAX_CLIENTS_PER_ROOM);