fix(networking): tolerate UDP probe startup races

packages/networking/src/lib.rs

@@ -52,6 +52,8 @@ pub enum TransportError {
     Serialization(String),
     #[error("udp transport failed: {0}")]
     Io(String),
+    #[error("udp peer is temporarily unavailable")]
+    PeerUnavailable,
     #[error("invalid match probe configuration: {0}")]
     InvalidConfiguration(String),
     #[error("received a datagram for a different match or peer")]
@@ -78,22 +80,17 @@ impl UdpPeer {
     pub async fn bind_unconnected(local: SocketAddr) -> Result<Self, TransportError> {
         let socket = tokio::net::UdpSocket::bind(local)
             .await
-            .map_err(|error| TransportError::Io(error.to_string()))?;
+            .map_err(map_udp_error)?;
         Ok(Self { socket })
     }
 
     pub async fn connect(self, peer: SocketAddr) -> Result<Self, TransportError> {
-        self.socket
-            .connect(peer)
-            .await
-            .map_err(|error| TransportError::Io(error.to_string()))?;
+        self.socket.connect(peer).await.map_err(map_udp_error)?;
         Ok(self)
     }
 
     pub fn local_addr(&self) -> Result<SocketAddr, TransportError> {
-        self.socket
-            .local_addr()
-            .map_err(|error| TransportError::Io(error.to_string()))
+        self.socket.local_addr().map_err(map_udp_error)
     }
 
     pub async fn send(&self, frame: &InputFrame) -> Result<(), TransportError> {
@@ -102,20 +99,13 @@ impl UdpPeer {
         }
         let encoded = serde_json::to_vec(frame)
             .map_err(|error| TransportError::Serialization(error.to_string()))?;
-        self.socket
-            .send(&encoded)
-            .await
-            .map_err(|error| TransportError::Io(error.to_string()))?;
+        self.socket.send(&encoded).await.map_err(map_udp_error)?;
         Ok(())
     }
 
     pub async fn receive(&self) -> Result<InputFrame, TransportError> {
         let mut buffer = [0_u8; MAX_DATAGRAM_BYTES];
-        let received = self
-            .socket
-            .recv(&mut buffer)
-            .await
-            .map_err(|error| TransportError::Io(error.to_string()))?;
+        let received = self.socket.recv(&mut buffer).await.map_err(map_udp_error)?;
         let frame: InputFrame = serde_json::from_slice(&buffer[..received])
             .map_err(|error| TransportError::Serialization(error.to_string()))?;
         InputFrame::new(frame.frame, frame.player_id, frame.input)
@@ -129,25 +119,29 @@ impl UdpPeer {
                 "match probe datagram exceeds maximum size".into(),
             ));
         }
-        self.socket
-            .send(&encoded)
-            .await
-            .map_err(|error| TransportError::Io(error.to_string()))?;
+        self.socket.send(&encoded).await.map_err(map_udp_error)?;
         Ok(())
     }
 
     async fn receive_packet(&self) -> Result<probe::ProbePacket, TransportError> {
         let mut buffer = [0_u8; MAX_DATAGRAM_BYTES];
-        let received = self
-            .socket
-            .recv(&mut buffer)
-            .await
-            .map_err(|error| TransportError::Io(error.to_string()))?;
+        let received = self.socket.recv(&mut buffer).await.map_err(map_udp_error)?;
         serde_json::from_slice(&buffer[..received])
             .map_err(|error| TransportError::Serialization(error.to_string()))
     }
 }
 
+fn map_udp_error(error: std::io::Error) -> TransportError {
+    if matches!(
+        error.kind(),
+        std::io::ErrorKind::ConnectionRefused | std::io::ErrorKind::ConnectionReset
+    ) {
+        TransportError::PeerUnavailable
+    } else {
+        TransportError::Io(error.to_string())
+    }
+}
+
 /// Deterministic, bounded transport used to prove the match contract without a network.
 pub struct InMemoryPeer {
     outbound: mpsc::Sender<InputFrame>,
@@ -265,6 +259,19 @@ mod tests {
         assert_eq!(error, Err(TransportError::FrameTooLarge(257)));
     }
 
+    #[test]
+    fn connected_udp_icmp_errors_are_retryable_startup_races() {
+        for kind in [
+            std::io::ErrorKind::ConnectionRefused,
+            std::io::ErrorKind::ConnectionReset,
+        ] {
+            assert_eq!(
+                map_udp_error(std::io::Error::from(kind)),
+                TransportError::PeerUnavailable
+            );
+        }
+    }
+
     #[test]
     fn bounded_queue_applies_backpressure() {
         let (host, _guest) = InMemoryPeer::pair();

packages/networking/src/probe.rs

@@ -5,7 +5,10 @@ use std::time::{Duration, Instant};
 
 pub const MAX_PROBE_FRAMES: u64 = 10_000;
 const RETRY_INTERVAL: Duration = Duration::from_millis(25);
-const COMPLETION_GRACE: Duration = Duration::from_millis(250);
+// Keep answering lagging final-frame retries after the local transcript completes. A full second
+// covers process-start and scheduler skew observed on shared CI runners without extending the
+// bounded match deadline or changing the deterministic transcript.
+const COMPLETION_GRACE: Duration = Duration::from_secs(1);
 
 // Kept private to the crate so the proof protocol can evolve independently of the data-plane API.
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
@@ -115,15 +118,12 @@ pub async fn run_match_probe(
                     send_probe_frame(peer, config, local_frame.clone()).await?;
                     frames_sent += 1;
                 }
-                Ok(Err(error)) => {
-                    if matches!(&error, TransportError::Io(message) if message.contains("10054") || message.to_lowercase().contains("forcibly closed"))
-                    {
-                        send_probe_frame(peer, config, local_frame.clone()).await?;
-                        frames_sent += 1;
-                        continue;
-                    }
-                    return Err(error);
+                Ok(Err(TransportError::PeerUnavailable)) => {
+                    tokio::time::sleep(wait).await;
+                    send_probe_frame(peer, config, local_frame.clone()).await?;
+                    frames_sent += 1;
                 }
+                Ok(Err(error)) => return Err(error),
                 Ok(Ok(packet)) => {
                     let remote = validated_remote_frame(packet, config)?;
                     if remote.frame < frame_number {
@@ -172,12 +172,20 @@ async fn send_probe_frame(
     config: &MatchProbeConfig,
     frame: InputFrame,
 ) -> Result<(), TransportError> {
-    peer.send_packet(&ProbePacket::Input {
-        room_id: config.descriptor.room_id.clone(),
-        session_key: config.session_key.clone(),
-        frame,
-    })
-    .await
+    match peer
+        .send_packet(&ProbePacket::Input {
+            room_id: config.descriptor.room_id.clone(),
+            session_key: config.session_key.clone(),
+            frame,
+        })
+        .await
+    {
+        // A connected UDP socket can receive an ICMP port-unreachable response while the other
+        // process is still binding. The probe's bounded retry loop treats that startup race like
+        // a dropped datagram instead of failing the match immediately.
+        Err(TransportError::PeerUnavailable) => Ok(()),
+        result => result,
+    }
 }
 
 fn validated_remote_frame(
@@ -218,9 +226,10 @@ async fn linger_for_peer(
                 send_probe_frame(peer, config, final_frame.clone()).await?;
                 *frames_sent += 1;
             }
-            Ok(Err(error)) if matches!(&error, TransportError::Io(message) if message.contains("10054") || message.to_lowercase().contains("forcibly closed")) =>
-            {
-                continue;
+            Ok(Err(TransportError::PeerUnavailable)) => {
+                tokio::time::sleep(wait).await;
+                send_probe_frame(peer, config, final_frame.clone()).await?;
+                *frames_sent += 1;
             }
             Ok(Err(error)) => return Err(error),
             Ok(Ok(packet)) => {

packages/protocol/src/generated/ChallengePayload.ts

@@ -4,4 +4,11 @@ import type { ChallengeState } from "./ChallengeState";
 /**
  * Challenge request between two users for a game/room.
  */
-export type ChallengePayload = { challenge_id: string, room_id: string, game_id: string, challenger_id: string, challenged_id: string, state: ChallengeState, };
+export type ChallengePayload = {
+  challenge_id: string;
+  room_id: string;
+  game_id: string;
+  challenger_id: string;
+  challenged_id: string;
+  state: ChallengeState;
+};

packages/protocol/src/generated/ChatPayload.ts

@@ -3,4 +3,4 @@
 /**
  * Chat message. Used for `chat.message` in rooms/lobbies.
  */
-export type ChatPayload = { channel: string, body: string, author_id: string, };
+export type ChatPayload = { channel: string; body: string; author_id: string };

packages/protocol/src/generated/Envelope.ts

@@ -3,24 +3,25 @@
 /**
  * Versioned envelope wrapping every protocol message.
  */
-export type Envelope = { 
-/**
- * Discriminator for message kind, e.g. `lobby.create`, `challenge.offer`.
- */
-type: string, 
-/**
- * Protocol version, e.g. "1.0".
- */
-version: string, 
-/**
- * Unique id for request/response correlation.
- */
-request_id: string, 
-/**
- * RFC 3339 timestamp (UTC).
- */
-timestamp: string, 
-/**
- * Arbitrary payload.
- */
-payload: unknown, };
+export type Envelope = {
+  /**
+   * Discriminator for message kind, e.g. `lobby.create`, `challenge.offer`.
+   */
+  type: string;
+  /**
+   * Protocol version, e.g. "1.0".
+   */
+  version: string;
+  /**
+   * Unique id for request/response correlation.
+   */
+  request_id: string;
+  /**
+   * RFC 3339 timestamp (UTC).
+   */
+  timestamp: string;
+  /**
+   * Arbitrary payload.
+   */
+  payload: unknown;
+};

packages/protocol/src/generated/MatchEndpointPayload.ts

@@ -3,4 +3,4 @@
 /**
  * Direct-UDP endpoint candidate exchanged between authenticated room members.
  */
-export type MatchEndpointPayload = { room_id: string, endpoint: string, nonce: string, };
+export type MatchEndpointPayload = { room_id: string; endpoint: string; nonce: string };

packages/protocol/src/generated/MatchProbeCompletedPayload.ts

@@ -3,4 +3,8 @@
 /**
  * Result of a completed direct-UDP proof, relayed to the other room member.
  */
-export type MatchProbeCompletedPayload = { room_id: string, frames_received: number, transcript_checksum: string, };
+export type MatchProbeCompletedPayload = {
+  room_id: string;
+  frames_received: number;
+  transcript_checksum: string;
+};

packages/protocol/src/generated/PresencePayload.ts

@@ -3,4 +3,10 @@
 /**
  * Presence / latency update. Mirrors `diagnose_network` and `presence.update`.
  */
-export type PresencePayload = { user_id: string, rtt_ms: number, loss: number, jitter: number, relay_reachable: boolean, };
+export type PresencePayload = {
+  user_id: string;
+  rtt_ms: number;
+  loss: number;
+  jitter: number;
+  relay_reachable: boolean;
+};

packages/protocol/src/generated/RoomPayload.ts

@@ -4,4 +4,10 @@ import type { RoomState } from "./RoomState";
 /**
  * Room snapshot pushed via `room.state`.
  */
-export type RoomPayload = { id: string, game_id: string, host_id: string, guest_id: string | null, state: RoomState, };
+export type RoomPayload = {
+  id: string;
+  game_id: string;
+  host_id: string;
+  guest_id: string | null;
+  state: RoomState;
+};

packages/protocol/src/generated/RoomState.ts

@@ -6,4 +6,5 @@
  * AGENTS.md describes WAITING→CHALLENGING→CONNECTING→PLAYING→FINISHED|CANCELLED.
  * Rust payload is snake_case lowercase for the wire; `Ready` is kept for compat with ARCH's READY.
  */
-export type RoomState = "waiting" | "ready" | "challenging" | "connecting" | "playing" | "finished" | "cancelled";
+export type RoomState =
+  "waiting" | "ready" | "challenging" | "connecting" | "playing" | "finished" | "cancelled";

packages/protocol/src/generated/SessionAnswerPayload.ts

@@ -3,4 +3,4 @@
 /**
  * SDP answer relayed verbatim between authenticated room members.
  */
-export type SessionAnswerPayload = { room_id: string, sdp: string, };
+export type SessionAnswerPayload = { room_id: string; sdp: string };

packages/protocol/src/generated/SessionCandidatePayload.ts

@@ -3,4 +3,9 @@
 /**
  * ICE candidate relayed verbatim between authenticated room members.
  */
-export type SessionCandidatePayload = { room_id: string, candidate: string, sdp_mid: string | null, sdp_mline_index: number | null, };
+export type SessionCandidatePayload = {
+  room_id: string;
+  candidate: string;
+  sdp_mid: string | null;
+  sdp_mline_index: number | null;
+};

packages/protocol/src/generated/SessionOfferPayload.ts

@@ -3,4 +3,4 @@
 /**
  * SDP offer relayed verbatim between authenticated room members.
  */
-export type SessionOfferPayload = { room_id: string, sdp: string, };
+export type SessionOfferPayload = { room_id: string; sdp: string };