diff --git a/packages/networking/src/lib.rs b/packages/networking/src/lib.rs index 9ad0aca..9e087af 100644 --- a/packages/networking/src/lib.rs +++ b/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 { 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.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 { - 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 { 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 { 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, @@ -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(); diff --git a/packages/networking/src/probe.rs b/packages/networking/src/probe.rs index 63ca93a..d6eb7e0 100644 --- a/packages/networking/src/probe.rs +++ b/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)) => { diff --git a/packages/protocol/src/generated/ChallengePayload.ts b/packages/protocol/src/generated/ChallengePayload.ts index 2520840..4c340c5 100644 --- a/packages/protocol/src/generated/ChallengePayload.ts +++ b/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; +}; diff --git a/packages/protocol/src/generated/ChatPayload.ts b/packages/protocol/src/generated/ChatPayload.ts index bcc124c..5b12881 100644 --- a/packages/protocol/src/generated/ChatPayload.ts +++ b/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 }; diff --git a/packages/protocol/src/generated/Envelope.ts b/packages/protocol/src/generated/Envelope.ts index 3a378ac..58d0403 100644 --- a/packages/protocol/src/generated/Envelope.ts +++ b/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; +}; diff --git a/packages/protocol/src/generated/MatchEndpointPayload.ts b/packages/protocol/src/generated/MatchEndpointPayload.ts index 3882bf4..2ca8c87 100644 --- a/packages/protocol/src/generated/MatchEndpointPayload.ts +++ b/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 }; diff --git a/packages/protocol/src/generated/MatchProbeCompletedPayload.ts b/packages/protocol/src/generated/MatchProbeCompletedPayload.ts index 33596fb..92fd040 100644 --- a/packages/protocol/src/generated/MatchProbeCompletedPayload.ts +++ b/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; +}; diff --git a/packages/protocol/src/generated/PresencePayload.ts b/packages/protocol/src/generated/PresencePayload.ts index a533baa..d640596 100644 --- a/packages/protocol/src/generated/PresencePayload.ts +++ b/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; +}; diff --git a/packages/protocol/src/generated/RoomPayload.ts b/packages/protocol/src/generated/RoomPayload.ts index dda9de4..ea01842 100644 --- a/packages/protocol/src/generated/RoomPayload.ts +++ b/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; +}; diff --git a/packages/protocol/src/generated/RoomState.ts b/packages/protocol/src/generated/RoomState.ts index f24a382..528d39e 100644 --- a/packages/protocol/src/generated/RoomState.ts +++ b/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"; diff --git a/packages/protocol/src/generated/SessionAnswerPayload.ts b/packages/protocol/src/generated/SessionAnswerPayload.ts index d748db1..c63b2a1 100644 --- a/packages/protocol/src/generated/SessionAnswerPayload.ts +++ b/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 }; diff --git a/packages/protocol/src/generated/SessionCandidatePayload.ts b/packages/protocol/src/generated/SessionCandidatePayload.ts index a43ec7e..8e3d1ac 100644 --- a/packages/protocol/src/generated/SessionCandidatePayload.ts +++ b/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; +}; diff --git a/packages/protocol/src/generated/SessionOfferPayload.ts b/packages/protocol/src/generated/SessionOfferPayload.ts index 98aff87..f4c3b6e 100644 --- a/packages/protocol/src/generated/SessionOfferPayload.ts +++ b/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 };