diff --git a/Cargo.lock b/Cargo.lock index d2fac60..2b030ec 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2,19 +2,6 @@ # It is not intended for manual editing. version = 3 -[[package]] -name = "ahash" -version = "0.8.12" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5a15f179cd60c4584b8a8c596927aadc462e27f2ca70c04e0071964a73ba7a75" -dependencies = [ - "cfg-if", - "getrandom 0.3.4", - "once_cell", - "version_check", - "zerocopy", -] - [[package]] name = "aho-corasick" version = "1.1.5" @@ -91,7 +78,7 @@ checksum = "edca88bc138befd0323b20752846e6587272d3b03b0343c8ea28a6f819e6e71f" dependencies = [ "async-trait", "axum-core", - "base64 0.22.1", + "base64", "bytes", "futures-util", "http", @@ -141,12 +128,6 @@ dependencies = [ "tracing", ] -[[package]] -name = "base64" -version = "0.21.7" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9d297deb1925b89f2ccc13d7635fa0714f12c87adce1c75356b39ca9b7178567" - [[package]] name = "base64" version = "0.22.1" @@ -365,7 +346,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -381,9 +362,13 @@ dependencies = [ [[package]] name = "event-listener" -version = "2.5.3" +version = "5.4.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0206175f82b8d6bf6652ff7d71a1e27fd2e4efde587fd368662814d6ec1d9ce0" +checksum = "5a23add41df1562121a9393cb065eab5146a1242410f23a644851e90cfd669d2" +dependencies = [ + "parking", + "pin-project-lite", +] [[package]] name = "fastrand" @@ -408,6 +393,12 @@ dependencies = [ "spin", ] +[[package]] +name = "foldhash" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d9c4f5dac5e15c24eb999c26181a6ca40b39fe946cbe4c263c7209467bc83af2" + [[package]] name = "form_urlencoded" version = "1.2.2" @@ -526,34 +517,32 @@ name = "hashbrown" version = "0.14.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e5274423e17b7c9fc20b6e7e208532f9b19825d82dfd615708b70edd83df41f1" -dependencies = [ - "ahash", - "allocator-api2", -] [[package]] name = "hashbrown" version = "0.15.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9229cfe53dfd69f0609a49f65461bd93001ea1ef889cd5529dd176593f5338a1" +dependencies = [ + "allocator-api2", + "equivalent", + "foldhash", +] [[package]] name = "hashlink" -version = "0.8.4" +version = "0.10.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e8094feaf31ff591f651a2664fb9cfd92bba7a60ce3197265e9482ebe753c8f7" +checksum = "7382cf6263419f2d8df38c55d7da83da5c18aef87fc7a7fc1fb1e344edfe14c1" dependencies = [ - "hashbrown 0.14.5", + "hashbrown 0.15.5", ] [[package]] name = "heck" -version = "0.4.1" +version = "0.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "95505c38b4572b2d910cecb0281560f54b440a19336cbbcb27bf6ce6adc6f5a8" -dependencies = [ - "unicode-segmentation", -] +checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" [[package]] name = "hex" @@ -764,11 +753,10 @@ dependencies = [ [[package]] name = "libsqlite3-sys" -version = "0.27.0" +version = "0.30.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cf4e226dcd58b4be396f7bd3c20da8fdee2911400705297ba7d2d7cc2c30f716" +checksum = "2e99fb7a497b1e3339bc746195567ed8d3e24945ecd636e3619d20b9de9e9149" dependencies = [ - "cc", "pkg-config", "vcpkg", ] @@ -831,12 +819,6 @@ version = "0.3.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a" -[[package]] -name = "minimal-lexical" -version = "0.2.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "68354c5c6bd36d73ff3feceb05efa59b6acb7626617f4962be322a825e61f79a" - [[package]] name = "mio" version = "1.2.2" @@ -848,16 +830,6 @@ dependencies = [ "windows-sys 0.61.2", ] -[[package]] -name = "nom" -version = "7.1.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d273983c5a657a70a3e8f2a01329822f3b8c8172b73826411a55751e404a0a4a" -dependencies = [ - "memchr", - "minimal-lexical", -] - [[package]] name = "nu-ansi-term" version = "0.50.3" @@ -924,7 +896,7 @@ version = "0.1.0" dependencies = [ "openfight-emulator-sdk", "serde", - "thiserror", + "thiserror 1.0.69", ] [[package]] @@ -932,7 +904,7 @@ name = "openfight-emulator-sdk" version = "0.1.0" dependencies = [ "serde", - "thiserror", + "thiserror 1.0.69", ] [[package]] @@ -943,7 +915,7 @@ dependencies = [ "serde", "serde_json", "tempfile", - "thiserror", + "thiserror 1.0.69", "toml", ] @@ -954,7 +926,7 @@ dependencies = [ "openfight-emulator-sdk", "serde", "serde_json", - "thiserror", + "thiserror 1.0.69", "tokio", ] @@ -985,7 +957,7 @@ dependencies = [ "serial_test", "sha2", "sqlx", - "thiserror", + "thiserror 1.0.69", "tokio", "tokio-tungstenite", "tower 0.4.13", @@ -1000,9 +972,15 @@ name = "openfight-shared" version = "0.1.0" dependencies = [ "serde", - "thiserror", + "thiserror 1.0.69", ] +[[package]] +name = "parking" +version = "2.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f38d5652c16fde515bb1ecef450ab0f6a219d619a7274976324d5e377f7dceba" + [[package]] name = "parking_lot" version = "0.12.5" @@ -1037,12 +1015,6 @@ dependencies = [ "subtle", ] -[[package]] -name = "paste" -version = "1.0.15" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "57c0d7b74b563b49d38dae00a0c37d4d6de9b432382b2892f0574ddcae73fd0a" - [[package]] name = "pem-rfc7468" version = "0.7.0" @@ -1257,7 +1229,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -1446,6 +1418,9 @@ name = "smallvec" version = "1.15.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8ed6a63f02c8539c91a8685a86f4099661ba3da017932f6ebbea6de3f0fa7c90" +dependencies = [ + "serde", +] [[package]] name = "socket2" @@ -1476,21 +1451,11 @@ dependencies = [ "der", ] -[[package]] -name = "sqlformat" -version = "0.2.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7bba3a93db0cc4f7bdece8bb09e77e2e785c20bfebf79eb8340ed80708048790" -dependencies = [ - "nom", - "unicode_categories", -] - [[package]] name = "sqlx" -version = "0.7.4" +version = "0.8.6" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c9a2ccff1a000a5a59cd33da541d9f2fdcd9e6e8229cc200565942bff36d0aaa" +checksum = "1fefb893899429669dcdd979aff487bd78f4064e5e7907e4269081e0ef7d97dc" dependencies = [ "sqlx-core", "sqlx-macros", @@ -1501,38 +1466,33 @@ dependencies = [ [[package]] name = "sqlx-core" -version = "0.7.4" +version = "0.8.6" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "24ba59a9342a3d9bab6c56c118be528b27c9b60e490080e9711a04dccac83ef6" +checksum = "ee6798b1838b6a0f69c007c133b8df5866302197e404e8b6ee8ed3e3a5e68dc6" dependencies = [ - "ahash", - "atoi", - "byteorder", + "base64", "bytes", "chrono", "crc", "crossbeam-queue", "either", "event-listener", - "futures-channel", "futures-core", "futures-intrusive", "futures-io", "futures-util", + "hashbrown 0.15.5", "hashlink", - "hex", "indexmap", "log", "memchr", "once_cell", - "paste", "percent-encoding", "serde", "serde_json", "sha2", "smallvec", - "sqlformat", - "thiserror", + "thiserror 2.0.20", "tokio", "tokio-stream", "tracing", @@ -1542,22 +1502,22 @@ dependencies = [ [[package]] name = "sqlx-macros" -version = "0.7.4" +version = "0.8.6" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4ea40e2345eb2faa9e1e5e326db8c34711317d2b5e08d0d5741619048a803127" +checksum = "a2d452988ccaacfbf5e0bdbc348fb91d7c8af5bee192173ac3636b5fb6e6715d" dependencies = [ "proc-macro2", "quote", "sqlx-core", "sqlx-macros-core", - "syn 1.0.109", + "syn 2.0.119", ] [[package]] name = "sqlx-macros-core" -version = "0.7.4" +version = "0.8.6" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5833ef53aaa16d860e92123292f1f6a3d53c34ba8b1969f152ef1a7bb803f3c8" +checksum = "19a9c1841124ac5a61741f96e1d9e2ec77424bf323962dd894bdb93f37d5219b" dependencies = [ "dotenvy", "either", @@ -1573,20 +1533,19 @@ dependencies = [ "sqlx-mysql", "sqlx-postgres", "sqlx-sqlite", - "syn 1.0.109", - "tempfile", + "syn 2.0.119", "tokio", "url", ] [[package]] name = "sqlx-mysql" -version = "0.7.4" +version = "0.8.6" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1ed31390216d20e538e447a7a9b959e06ed9fc51c37b514b46eb758016ecd418" +checksum = "aa003f0038df784eb8fecbbac13affe3da23b45194bd57dba231c8f48199c526" dependencies = [ "atoi", - "base64 0.21.7", + "base64", "bitflags", "byteorder", "bytes", @@ -1617,7 +1576,7 @@ dependencies = [ "smallvec", "sqlx-core", "stringprep", - "thiserror", + "thiserror 2.0.20", "tracing", "uuid", "whoami", @@ -1625,12 +1584,12 @@ dependencies = [ [[package]] name = "sqlx-postgres" -version = "0.7.4" +version = "0.8.6" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7c824eb80b894f926f89a0b9da0c7f435d27cdd35b8c655b114e58223918577e" +checksum = "db58fcd5a53cf07c184b154801ff91347e4c30d17a3562a635ff028ad5deda46" dependencies = [ "atoi", - "base64 0.21.7", + "base64", "bitflags", "byteorder", "chrono", @@ -1639,7 +1598,6 @@ dependencies = [ "etcetera", "futures-channel", "futures-core", - "futures-io", "futures-util", "hex", "hkdf", @@ -1657,7 +1615,7 @@ dependencies = [ "smallvec", "sqlx-core", "stringprep", - "thiserror", + "thiserror 2.0.20", "tracing", "uuid", "whoami", @@ -1665,9 +1623,9 @@ dependencies = [ [[package]] name = "sqlx-sqlite" -version = "0.7.4" +version = "0.8.6" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b244ef0a8414da0bed4bb1910426e890b19e5e9bccc27ada6b797d05c55ae0aa" +checksum = "c2d12fe70b2c1b4401038055f90f151b78208de1f9f89a7dbfd41587a10c3eea" dependencies = [ "atoi", "chrono", @@ -1681,10 +1639,11 @@ dependencies = [ "log", "percent-encoding", "serde", + "serde_urlencoded", "sqlx-core", + "thiserror 2.0.20", "tracing", "url", - "urlencoding", "uuid", ] @@ -1705,17 +1664,6 @@ version = "2.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292" -[[package]] -name = "syn" -version = "1.0.109" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "72b64191b275b66ffe2469e8af2c1cfe3bafa67b529ead792a6d0160888b4237" -dependencies = [ - "proc-macro2", - "quote", - "unicode-ident", -] - [[package]] name = "syn" version = "2.0.119" @@ -1754,7 +1702,7 @@ dependencies = [ "getrandom 0.3.4", "once_cell", "rustix", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -1772,7 +1720,16 @@ version = "1.0.69" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6aaf5339b578ea85b50e080feb250a3e8ae8cfcdff9a461c9ec2904bc923f52" dependencies = [ - "thiserror-impl", + "thiserror-impl 1.0.69", +] + +[[package]] +name = "thiserror" +version = "2.0.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ec86235f5fcc2a73650310756d2ac5b138a5780bbbdfae3eeccec992c435ba4f" +dependencies = [ + "thiserror-impl 2.0.20", ] [[package]] @@ -1786,6 +1743,17 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "thiserror-impl" +version = "2.0.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bc04cd3e1236dd4a98afca4569f2deb3f120e5422a4023be2cb683f8486292af" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.3", +] + [[package]] name = "thread_local" version = "1.1.10" @@ -2031,7 +1999,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b44017f9f875786e543595076374b9ef7d13465a518dd93d6ccdbf5b432dde8c" dependencies = [ "chrono", - "thiserror", + "thiserror 1.0.69", "ts-rs-macros", "uuid", ] @@ -2062,7 +2030,7 @@ dependencies = [ "log", "rand", "sha1", - "thiserror", + "thiserror 1.0.69", "utf-8", ] @@ -2099,18 +2067,6 @@ version = "0.1.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7df058c713841ad818f1dc5d3fd88063241cc61f49f5fbea4b951e8cf5a8d71d" -[[package]] -name = "unicode-segmentation" -version = "1.12.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f6ccf251212114b54433ec949fd6a7841275f9ada20dddd2f29e9ceea4501493" - -[[package]] -name = "unicode_categories" -version = "0.1.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "39ec24b3121d976906ece63c9daad25b85969647682eee313cb5779fdd69e14e" - [[package]] name = "url" version = "2.5.0" @@ -2122,12 +2078,6 @@ dependencies = [ "percent-encoding", ] -[[package]] -name = "urlencoding" -version = "2.1.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "daf8dba3b7eb870caf1ddeed7bc9d2a049f3cfdfae7cb521b087cc33ae4c49da" - [[package]] name = "utf-8" version = "0.7.6" @@ -2244,7 +2194,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.48.0", + "windows-sys 0.61.2", ] [[package]] diff --git a/apps/client/src-tauri/Cargo.lock b/apps/client/src-tauri/Cargo.lock index 2a3ae59..8c05d11 100644 --- a/apps/client/src-tauri/Cargo.lock +++ b/apps/client/src-tauri/Cargo.lock @@ -1684,6 +1684,7 @@ version = "0.1.0" dependencies = [ "openfight-adapter-fbneo", "openfight-emulator-sdk", + "openfight-networking", "serde", "serde_json", "tauri", @@ -1691,6 +1692,7 @@ dependencies = [ "tokio", "tracing", "tracing-subscriber", + "uuid", ] [[package]] @@ -1701,6 +1703,17 @@ dependencies = [ "thiserror", ] +[[package]] +name = "openfight-networking" +version = "0.1.0" +dependencies = [ + "openfight-emulator-sdk", + "serde", + "serde_json", + "thiserror", + "tokio", +] + [[package]] name = "pango" version = "0.15.10" diff --git a/apps/client/src-tauri/Cargo.toml b/apps/client/src-tauri/Cargo.toml index d97d6d3..98f72b7 100644 --- a/apps/client/src-tauri/Cargo.toml +++ b/apps/client/src-tauri/Cargo.toml @@ -20,6 +20,8 @@ serde = { version = "1", features = ["derive"] } serde_json = "1" openfight-adapter-fbneo = { path = "../../../adapters/fbneo" } openfight-emulator-sdk = { path = "../../../packages/emulator-sdk" } +openfight-networking = { path = "../../../packages/networking" } +uuid = { version = "1", features = ["v4"] } [features] default = ["custom-protocol"] diff --git a/apps/client/src-tauri/src/commands/match_probe.rs b/apps/client/src-tauri/src/commands/match_probe.rs new file mode 100644 index 0000000..b5a4b4a --- /dev/null +++ b/apps/client/src-tauri/src/commands/match_probe.rs @@ -0,0 +1,338 @@ +use openfight_emulator_sdk::{MatchDescriptor, PeerRole, TransportKind}; +use openfight_networking::{run_match_probe, MatchProbeConfig, MatchProbeReport, UdpPeer}; +use serde::{Deserialize, Serialize}; +use std::collections::HashMap; +use std::future::Future; +use std::net::{IpAddr, Ipv4Addr, SocketAddr}; +use std::time::Duration; +use tokio::sync::{watch, Mutex}; +use uuid::Uuid; + +const MAX_ACTIVE_PROBES: usize = 8; + +#[derive(Default)] +pub struct MatchProbeState { + probes: Mutex>, +} + +enum MatchProbeSlot { + Prepared(PreparedProbe), + Running { + run_id: Uuid, + cancel: watch::Sender, + }, +} + +struct PreparedProbe { + peer: UdpPeer, + candidate: MatchEndpointCandidate, +} + +#[derive(Debug, Clone, Serialize)] +pub struct MatchEndpointCandidate { + pub endpoint: SocketAddr, + pub nonce: String, +} + +#[derive(Debug, Deserialize)] +pub struct ReserveMatchProbeRequest { + pub room_id: String, + pub advertised_host: Option, + pub bind_port: Option, +} + +#[derive(Debug, Deserialize)] +pub struct RunMatchProbeRequest { + pub room_id: String, + pub game_id: String, + pub local_user_id: String, + pub peer_user_id: String, + pub role: PeerRole, + pub peer_endpoint: SocketAddr, + pub peer_nonce: String, + #[serde(default = "default_frame_count")] + pub frame_count: u64, + #[serde(default = "default_timeout_ms")] + pub timeout_ms: u64, +} + +fn default_frame_count() -> u64 { + 60 +} + +fn default_timeout_ms() -> u64 { + 5_000 +} + +#[tauri::command] +pub async fn reserve_match_probe( + state: tauri::State<'_, MatchProbeState>, + request: ReserveMatchProbeRequest, +) -> Result { + validate_room_id(&request.room_id)?; + if let Some(slot) = state.probes.lock().await.get(&request.room_id) { + return match slot { + MatchProbeSlot::Prepared(prepared) => Ok(prepared.candidate.clone()), + MatchProbeSlot::Running { .. } => { + Err("match probe is already running for this room".into()) + } + }; + } + + let bind_address = SocketAddr::from((Ipv4Addr::UNSPECIFIED, request.bind_port.unwrap_or(0))); + let peer = UdpPeer::bind_unconnected(bind_address) + .await + .map_err(|error| error.to_string())?; + let port = peer.local_addr().map_err(|error| error.to_string())?.port(); + let host = match request.advertised_host { + Some(host) if !host.is_unspecified() => host, + Some(_) => return Err("advertised_host must not be unspecified".into()), + None => discover_local_ipv4()?, + }; + let candidate = MatchEndpointCandidate { + endpoint: SocketAddr::from((host, port)), + nonce: Uuid::new_v4().to_string(), + }; + let mut probes = state.probes.lock().await; + if let Some(existing) = probes.get(&request.room_id) { + return match existing { + MatchProbeSlot::Prepared(prepared) => Ok(prepared.candidate.clone()), + MatchProbeSlot::Running { .. } => { + Err("match probe is already running for this room".into()) + } + }; + } + if probes.len() >= MAX_ACTIVE_PROBES { + return Err("too many active match probes; cancel an older room first".into()); + } + probes.insert( + request.room_id, + MatchProbeSlot::Prepared(PreparedProbe { + peer, + candidate: candidate.clone(), + }), + ); + Ok(candidate) +} + +#[tauri::command] +pub async fn run_reserved_match_probe( + state: tauri::State<'_, MatchProbeState>, + request: RunMatchProbeRequest, +) -> Result { + validate_room_id(&request.room_id)?; + let room_id = request.room_id.clone(); + let run_id = Uuid::new_v4(); + let (cancel_tx, cancel_rx) = watch::channel(false); + let prepared = { + let mut probes = state.probes.lock().await; + let slot = probes + .remove(&request.room_id) + .ok_or_else(|| "match probe is not reserved for this room".to_string())?; + let MatchProbeSlot::Prepared(prepared) = slot else { + probes.insert(request.room_id.clone(), slot); + return Err("match probe is already running for this room".into()); + }; + probes.insert( + request.room_id.clone(), + MatchProbeSlot::Running { + run_id, + cancel: cancel_tx, + }, + ); + prepared + }; + let probe = async move { + let descriptor = MatchDescriptor { + room_id: request.room_id, + game_id: request.game_id, + local_user_id: request.local_user_id, + peer_user_id: request.peer_user_id, + role: request.role, + transport: TransportKind::DirectUdp, + local_endpoint: prepared.candidate.endpoint, + peer_endpoint: request.peer_endpoint, + input_delay_frames: 2, + }; + let session_key = combined_session_key(&prepared.candidate.nonce, &request.peer_nonce)?; + let config = MatchProbeConfig::new( + descriptor, + session_key, + request.frame_count, + Duration::from_millis(request.timeout_ms), + ) + .map_err(|error| error.to_string())?; + let peer = prepared + .peer + .connect(request.peer_endpoint) + .await + .map_err(|error| error.to_string())?; + run_match_probe(&peer, &config) + .await + .map_err(|error| error.to_string()) + }; + let result = cancel_or_complete(probe, cancel_rx).await; + remove_running_probe(&state, &room_id, run_id).await; + let report = result?; + tracing::info!( + room_id = %report.room_id, + frames = report.frames_received, + elapsed_ms = report.elapsed_ms, + "LAN match probe completed" + ); + Ok(report) +} + +#[tauri::command] +pub async fn cancel_match_probe( + state: tauri::State<'_, MatchProbeState>, + room_id: String, +) -> Result<(), String> { + validate_room_id(&room_id)?; + cancel_probe(&state, &room_id).await; + Ok(()) +} + +async fn cancel_probe(state: &MatchProbeState, room_id: &str) { + if let Some(MatchProbeSlot::Running { cancel, .. }) = state.probes.lock().await.remove(room_id) { + let _ = cancel.send(true); + } +} + +async fn remove_running_probe(state: &MatchProbeState, room_id: &str, run_id: Uuid) { + let mut probes = state.probes.lock().await; + if matches!( + probes.get(room_id), + Some(MatchProbeSlot::Running { + run_id: current_id, + .. + }) if *current_id == run_id + ) { + probes.remove(room_id); + } +} + +async fn cancel_or_complete(future: F, mut cancel: watch::Receiver) -> Result +where + F: Future>, +{ + if *cancel.borrow() { + return Err("match probe cancelled".into()); + } + tokio::select! { + result = future => result, + _ = cancel.changed() => Err("match probe cancelled".into()), + } +} + +fn validate_room_id(room_id: &str) -> Result<(), String> { + Uuid::parse_str(room_id) + .map(|_| ()) + .map_err(|_| "room_id must be a UUID".into()) +} + +fn discover_local_ipv4() -> Result { + let socket = std::net::UdpSocket::bind((Ipv4Addr::UNSPECIFIED, 0)) + .map_err(|error| format!("failed to inspect local network: {error}"))?; + socket + .connect((Ipv4Addr::new(192, 0, 2, 1), 9)) + .map_err(|error| format!("failed to select a LAN interface: {error}"))?; + match socket + .local_addr() + .map_err(|error| format!("failed to read LAN address: {error}"))? + .ip() + { + IpAddr::V4(address) if !address.is_unspecified() => Ok(address), + _ => Err("no usable IPv4 LAN address was detected".into()), + } +} + +fn combined_session_key(local_nonce: &str, peer_nonce: &str) -> Result { + let local = Uuid::parse_str(local_nonce).map_err(|_| "local nonce is invalid")?; + let peer = Uuid::parse_str(peer_nonce).map_err(|_| "peer nonce is invalid")?; + if local == peer { + return Err("peer nonce must differ from local nonce".into()); + } + let (first, second) = if local.as_bytes() < peer.as_bytes() { + (local, peer) + } else { + (peer, local) + }; + Ok(format!("{first}:{second}")) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn session_key_is_independent_of_peer_order() { + let first = "8a1110d5-8dd2-4ad2-9c88-ad9768bc4905"; + let second = "7fa7b981-f41c-4d40-a4c9-d2170067f12f"; + assert_eq!( + combined_session_key(first, second), + combined_session_key(second, first) + ); + } + + #[test] + fn session_key_rejects_duplicate_nonces() { + let nonce = "8a1110d5-8dd2-4ad2-9c88-ad9768bc4905"; + assert!(combined_session_key(nonce, nonce).is_err()); + } + + #[tokio::test] + async fn cancellation_interrupts_pending_probe() { + let (cancel_tx, cancel_rx) = watch::channel(false); + cancel_tx.send(true).expect("receiver remains active"); + let result = cancel_or_complete(std::future::pending::>(), cancel_rx).await; + assert_eq!(result, Err("match probe cancelled".into())); + } + + #[tokio::test] + async fn completed_probe_returns_its_result() { + let (_cancel_tx, cancel_rx) = watch::channel(false); + let result = cancel_or_complete(async { Ok::<_, String>(42) }, cancel_rx).await; + assert_eq!(result, Ok(42)); + } + + #[tokio::test] + async fn cancelling_running_probe_signals_and_removes_it() { + let state = MatchProbeState::default(); + let (cancel, cancel_rx) = watch::channel(false); + state.probes.lock().await.insert( + "room".into(), + MatchProbeSlot::Running { + run_id: Uuid::new_v4(), + cancel, + }, + ); + + cancel_probe(&state, "room").await; + + assert!(*cancel_rx.borrow()); + assert!(!state.probes.lock().await.contains_key("room")); + } + + #[tokio::test] + async fn old_run_cleanup_preserves_newer_run() { + let state = MatchProbeState::default(); + let old_run_id = Uuid::new_v4(); + let new_run_id = Uuid::new_v4(); + let (cancel, _cancel_rx) = watch::channel(false); + state.probes.lock().await.insert( + "room".into(), + MatchProbeSlot::Running { + run_id: new_run_id, + cancel, + }, + ); + + remove_running_probe(&state, "room", old_run_id).await; + + assert!(matches!( + state.probes.lock().await.get("room"), + Some(MatchProbeSlot::Running { run_id, .. }) if *run_id == new_run_id + )); + } +} diff --git a/apps/client/src-tauri/src/commands/mod.rs b/apps/client/src-tauri/src/commands/mod.rs index a3dc93f..565bb65 100644 --- a/apps/client/src-tauri/src/commands/mod.rs +++ b/apps/client/src-tauri/src/commands/mod.rs @@ -1,3 +1,4 @@ pub mod diag; pub mod fs; +pub mod match_probe; pub mod process; diff --git a/apps/client/src-tauri/src/main.rs b/apps/client/src-tauri/src/main.rs index 6c45031..cfa2a0a 100644 --- a/apps/client/src-tauri/src/main.rs +++ b/apps/client/src-tauri/src/main.rs @@ -6,11 +6,15 @@ fn main() { tracing_subscriber::fmt::init(); tauri::Builder::default() .manage(commands::process::ProcessState::default()) + .manage(commands::match_probe::MatchProbeState::default()) .invoke_handler(tauri::generate_handler![ commands::fs::scan_game, commands::process::launch_game, commands::process::stop_game, commands::diag::network_test, + commands::match_probe::reserve_match_probe, + commands::match_probe::run_reserved_match_probe, + commands::match_probe::cancel_match_probe, ]) .run(tauri::generate_context!()) .expect("error while running tauri application"); diff --git a/apps/client/src/App.tsx b/apps/client/src/App.tsx index 8c56baf..f563c56 100644 --- a/apps/client/src/App.tsx +++ b/apps/client/src/App.tsx @@ -1,8 +1,14 @@ import { useEffect, useState } from "react"; import { useQuery, useQueryClient } from "@tanstack/react-query"; -import type { Envelope } from "@openfight/protocol"; +import type { + Envelope, + MatchEndpointPayload, + MatchProbeCompletedPayload, +} from "@openfight/protocol"; import DiagnosticsButton from "./components/DiagnosticsButton"; import { ApiError, api } from "./lib/api"; +import { parseMatchCompletion, parseMatchEndpoint } from "./lib/match"; +import { cancelMatchProbe } from "./lib/native"; import { useSessionStore } from "./lib/store"; import { OpenFightSocket, type ConnectionState } from "./lib/ws"; import Auth from "./routes/Auth"; @@ -19,6 +25,9 @@ export default function App() { const queryClient = useQueryClient(); const [view, setView] = useState({ name: "games" }); const [connection, setConnection] = useState("idle"); + const [activeSocket, setActiveSocket] = useState(null); + const [peerEndpoint, setPeerEndpoint] = useState(null); + const [peerCompletion, setPeerCompletion] = useState(null); const me = useQuery({ queryKey: ["me", token], queryFn: () => { @@ -37,6 +46,7 @@ export default function App() { useEffect(() => { if (!token) return; const socket = new OpenFightSocket(API_URL, token, setConnection); + setActiveSocket(socket); const unsubscribe = socket.subscribe((message: Envelope) => { if (message.type.startsWith("challenge.")) { void queryClient.invalidateQueries({ queryKey: ["challenges"] }); @@ -45,11 +55,22 @@ export default function App() { const roomId = roomIdFromPayload(message.payload); if (roomId) setView({ name: "match", roomId }); } + if (message.type === "match.endpoint") { + const endpoint = parseMatchEndpoint(message.payload); + if (endpoint) { + setPeerEndpoint(endpoint); + } + } + if (message.type === "match.probe.completed") { + const completion = parseMatchCompletion(message.payload); + if (completion) setPeerCompletion(completion); + } }); socket.connect(); return () => { unsubscribe(); socket.close(); + setActiveSocket((current) => (current === socket ? null : current)); }; }, [token, queryClient]); @@ -63,19 +84,22 @@ export default function App() { } const logout = async () => { try { + if (view.name === "match") await cancelMatchProbe(view.roomId); await api.logout(token); } finally { clearSession(); } }; + const returnToGames = () => { + if (view.name === "match") void cancelMatchProbe(view.roomId); + setPeerEndpoint(null); + setPeerCompletion(null); + setView({ name: "games" }); + }; return (
- @@ -102,7 +126,19 @@ export default function App() { /> )} {view.name === "match" && ( - setView({ name: "games" })} /> + { + setPeerEndpoint(null); + setPeerCompletion(null); + }} + onDone={returnToGames} + /> )}
diff --git a/apps/client/src/lib/match.test.ts b/apps/client/src/lib/match.test.ts new file mode 100644 index 0000000..5cb2c35 --- /dev/null +++ b/apps/client/src/lib/match.test.ts @@ -0,0 +1,67 @@ +import { describe, expect, it } from "vitest"; +import type { RoomPayload } from "@openfight/protocol"; +import { matchParticipants, parseMatchCompletion, parseMatchEndpoint } from "./match"; + +const room = (guestId: string | null = "guest"): RoomPayload => ({ + id: "room-1", + game_id: "sfiii3", + host_id: "host", + guest_id: guestId, + state: "connecting", +}); + +describe("parseMatchEndpoint", () => { + it("accepts the transport candidate fields", () => { + expect( + parseMatchEndpoint({ + room_id: "room-1", + endpoint: "192.168.1.20:42000", + nonce: "nonce-1", + }) + ).toEqual({ + room_id: "room-1", + endpoint: "192.168.1.20:42000", + nonce: "nonce-1", + }); + }); + + it("rejects missing or non-string fields", () => { + expect(parseMatchEndpoint(null)).toBeUndefined(); + expect( + parseMatchEndpoint({ room_id: "room-1", endpoint: 42000, nonce: "nonce-1" }) + ).toBeUndefined(); + expect(parseMatchEndpoint({ room_id: "room-1", endpoint: "127.0.0.1:1" })).toBeUndefined(); + }); +}); + +describe("matchParticipants", () => { + it("derives complementary host and guest peers", () => { + expect(matchParticipants(room(), "host")).toEqual({ role: "host", peerUserId: "guest" }); + expect(matchParticipants(room(), "guest")).toEqual({ role: "guest", peerUserId: "host" }); + }); + + it("rejects incomplete rooms and non-members", () => { + expect(matchParticipants(room(null), "host")).toBeUndefined(); + expect(matchParticipants(room(), "outsider")).toBeUndefined(); + }); +}); + +describe("parseMatchCompletion", () => { + it("accepts a peer transcript result", () => { + expect( + parseMatchCompletion({ + room_id: "room-1", + frames_received: 60, + transcript_checksum: "0376c2e852f4fd25", + }) + ).toEqual({ + room_id: "room-1", + frames_received: 60, + transcript_checksum: "0376c2e852f4fd25", + }); + }); + + it("rejects incomplete peer results", () => { + expect(parseMatchCompletion({ room_id: "room-1", frames_received: "60" })).toBeUndefined(); + }); +}); diff --git a/apps/client/src/lib/match.ts b/apps/client/src/lib/match.ts new file mode 100644 index 0000000..3b56f9c --- /dev/null +++ b/apps/client/src/lib/match.ts @@ -0,0 +1,45 @@ +import type { + MatchEndpointPayload, + MatchProbeCompletedPayload, + RoomPayload, +} from "@openfight/protocol"; + +export function parseMatchEndpoint(payload: unknown): MatchEndpointPayload | undefined { + if (typeof payload !== "object" || payload === null) return undefined; + const roomId = Reflect.get(payload, "room_id"); + const endpoint = Reflect.get(payload, "endpoint"); + const nonce = Reflect.get(payload, "nonce"); + if (typeof roomId !== "string" || typeof endpoint !== "string" || typeof nonce !== "string") { + return undefined; + } + return { room_id: roomId, endpoint, nonce }; +} + +export function parseMatchCompletion(payload: unknown): MatchProbeCompletedPayload | undefined { + if (typeof payload !== "object" || payload === null) return undefined; + const roomId = Reflect.get(payload, "room_id"); + const framesReceived = Reflect.get(payload, "frames_received"); + const transcriptChecksum = Reflect.get(payload, "transcript_checksum"); + if ( + typeof roomId !== "string" || + typeof framesReceived !== "number" || + typeof transcriptChecksum !== "string" + ) { + return undefined; + } + return { + room_id: roomId, + frames_received: framesReceived, + transcript_checksum: transcriptChecksum, + }; +} + +export function matchParticipants( + room: RoomPayload, + localUserId: string +): { role: "host" | "guest"; peerUserId: string } | undefined { + if (!room.guest_id) return undefined; + if (room.host_id === localUserId) return { role: "host", peerUserId: room.guest_id }; + if (room.guest_id === localUserId) return { role: "guest", peerUserId: room.host_id }; + return undefined; +} diff --git a/apps/client/src/lib/native.ts b/apps/client/src/lib/native.ts index d2721c2..853f90a 100644 --- a/apps/client/src/lib/native.ts +++ b/apps/client/src/lib/native.ts @@ -6,6 +6,18 @@ export type NetworkDiagnostics = { rtt_ms: number | null; relay_reachable: boolean; }; +export type MatchEndpointCandidate = { endpoint: string; nonce: string }; +export type MatchProbeReport = { + room_id: string; + local_user_id: string; + peer_user_id: string; + role: "host" | "guest"; + transport: "direct_udp"; + frames_sent: number; + frames_received: number; + transcript_checksum: string; + elapsed_ms: number; +}; export function isDesktopRuntime(): boolean { return "__TAURI_IPC__" in window; @@ -34,3 +46,30 @@ export async function runNetworkTest(): Promise { } return invoke("network_test"); } + +export async function reserveMatchProbe(roomId: string): Promise { + if (!isDesktopRuntime()) throw new Error("LAN match probe requires the desktop client"); + return invoke("reserve_match_probe", { + request: { room_id: roomId }, + }); +} + +export async function runReservedMatchProbe(request: { + room_id: string; + game_id: string; + local_user_id: string; + peer_user_id: string; + role: "host" | "guest"; + peer_endpoint: string; + peer_nonce: string; + frame_count?: number; + timeout_ms?: number; +}): Promise { + if (!isDesktopRuntime()) throw new Error("LAN match probe requires the desktop client"); + return invoke("run_reserved_match_probe", { request }); +} + +export async function cancelMatchProbe(roomId: string): Promise { + if (!isDesktopRuntime()) return; + await invoke("cancel_match_probe", { roomId }); +} diff --git a/apps/client/src/lib/report.test.ts b/apps/client/src/lib/report.test.ts index 608f6f7..66917d4 100644 --- a/apps/client/src/lib/report.test.ts +++ b/apps/client/src/lib/report.test.ts @@ -19,4 +19,39 @@ describe("buildMatchReport", () => { expect(serialized).not.toContain("token"); expect(serialized).not.toContain("rom_path"); }); + + it("includes useful UDP evidence without private session material", () => { + const report = buildMatchReport( + { + id: "room-123", + game_id: "sfiii3", + host_id: "host", + guest_id: "guest", + state: "finished", + }, + new Date("2026-08-23T12:00:00Z"), + { + room_id: "room-123", + local_user_id: "host", + peer_user_id: "guest", + role: "host", + transport: "direct_udp", + frames_sent: 60, + frames_received: 60, + transcript_checksum: "0123456789abcdef", + elapsed_ms: 240, + } + ); + + expect(report.probe).toEqual({ + transport: "direct_udp", + frames_sent: 60, + frames_received: 60, + transcript_checksum: "0123456789abcdef", + elapsed_ms: 240, + }); + expect(JSON.stringify(report.probe)).not.toContain("user_id"); + expect(JSON.stringify(report.probe)).not.toContain("nonce"); + expect(JSON.stringify(report.probe)).not.toContain("endpoint"); + }); }); diff --git a/apps/client/src/lib/report.ts b/apps/client/src/lib/report.ts index 04f9340..534ea64 100644 --- a/apps/client/src/lib/report.ts +++ b/apps/client/src/lib/report.ts @@ -1,17 +1,37 @@ import type { RoomPayload } from "@openfight/protocol"; +import type { MatchProbeReport } from "./native"; + +export type MatchProbeEvidence = Pick< + MatchProbeReport, + "transport" | "frames_sent" | "frames_received" | "transcript_checksum" | "elapsed_ms" +>; export type MatchReport = { schema_version: 1; exported_at: string; room: RoomPayload; + probe?: MatchProbeEvidence; client: { platform: string; user_agent: string }; }; -export function buildMatchReport(room: RoomPayload, now = new Date()): MatchReport { +export function buildMatchReport( + room: RoomPayload, + now = new Date(), + probe?: MatchProbeReport +): MatchReport { return { schema_version: 1, exported_at: now.toISOString(), room, + ...(probe && { + probe: { + transport: probe.transport, + frames_sent: probe.frames_sent, + frames_received: probe.frames_received, + transcript_checksum: probe.transcript_checksum, + elapsed_ms: probe.elapsed_ms, + }, + }), client: { platform: typeof navigator === "undefined" ? "unknown" : navigator.platform, user_agent: typeof navigator === "undefined" ? "unknown" : navigator.userAgent, @@ -19,8 +39,8 @@ export function buildMatchReport(room: RoomPayload, now = new Date()): MatchRepo }; } -export function downloadMatchReport(room: RoomPayload): void { - const report = buildMatchReport(room); +export function downloadMatchReport(room: RoomPayload, probe?: MatchProbeReport): void { + const report = buildMatchReport(room, new Date(), probe); const url = URL.createObjectURL( new Blob([JSON.stringify(report, null, 2)], { type: "application/json" }) ); diff --git a/apps/client/src/lib/useLanMatchProbe.ts b/apps/client/src/lib/useLanMatchProbe.ts new file mode 100644 index 0000000..ab9a867 --- /dev/null +++ b/apps/client/src/lib/useLanMatchProbe.ts @@ -0,0 +1,231 @@ +import { useEffect, useRef, useState } from "react"; +import { useQueryClient } from "@tanstack/react-query"; +import type { + MatchEndpointPayload, + MatchProbeCompletedPayload, + RoomPayload, +} from "@openfight/protocol"; +import { api } from "./api"; +import { matchParticipants } from "./match"; +import { + cancelMatchProbe, + reserveMatchProbe, + runReservedMatchProbe, + type MatchEndpointCandidate, + type MatchProbeReport, +} from "./native"; +import type { OpenFightSocket } from "./ws"; + +type LanMatchProbeOptions = { + token: string; + userId: string; + roomId: string; + room?: RoomPayload; + socket: OpenFightSocket | null; + peerEndpoint?: MatchEndpointPayload; + peerCompletion?: MatchProbeCompletedPayload; + onRetry: () => void; +}; + +export function useLanMatchProbe({ + token, + userId, + roomId, + room, + socket, + peerEndpoint, + peerCompletion, + onRetry, +}: LanMatchProbeOptions) { + const queryClient = useQueryClient(); + const [localEndpoint, setLocalEndpoint] = useState(null); + const [probeReport, setProbeReport] = useState(null); + const [probeError, setProbeError] = useState(null); + const [candidateRelayed, setCandidateRelayed] = useState(false); + const [isResetting, setIsResetting] = useState(false); + const probeStarted = useRef(false); + const roomAdvanced = useRef(false); + const resetStarted = useRef(false); + const probeAttempt = useRef(0); + const preparationGeneration = useRef(0); + const reservationInFlight = useRef | null>(null); + + useEffect(() => { + if (!socket || !room || room.state !== "connecting" || localEndpoint) return; + const attempt = probeAttempt.current; + const preparation = ++preparationGeneration.current; + let cancelled = false; + const prepare = async () => { + try { + const reservation = reservationInFlight.current ?? reserveMatchProbe(roomId); + if (!reservationInFlight.current) reservationInFlight.current = reservation; + let candidate: MatchEndpointCandidate; + try { + candidate = await reservation; + } finally { + if (reservationInFlight.current === reservation) reservationInFlight.current = null; + } + if (cancelled || probeAttempt.current !== attempt) { + if (preparationGeneration.current === preparation) await cancelMatchProbe(roomId); + return; + } + setLocalEndpoint(candidate); + const delivered = await relayUntilDelivered( + socket, + "match.endpoint", + { + room_id: roomId, + endpoint: candidate.endpoint, + nonce: candidate.nonce, + }, + () => cancelled || probeAttempt.current !== attempt, + (message) => { + if (!cancelled && probeAttempt.current === attempt) setProbeError(message); + } + ); + if (delivered && !cancelled && probeAttempt.current === attempt) { + setCandidateRelayed(true); + setProbeError(null); + } + } catch (error) { + if (!cancelled && probeAttempt.current === attempt) { + setProbeError(errorMessage(error, "Failed to reserve LAN probe")); + } + } + }; + void prepare(); + return () => { + cancelled = true; + }; + }, [localEndpoint, room, roomId, socket]); + + useEffect(() => { + if ( + !room || + !socket || + !localEndpoint || + !candidateRelayed || + !peerEndpoint || + peerEndpoint.room_id !== roomId || + probeStarted.current + ) { + return; + } + const participants = matchParticipants(room, userId); + if (!participants) return; + const attempt = probeAttempt.current; + probeStarted.current = true; + setProbeError(null); + let cancelled = false; + const run = async () => { + try { + const report = await runReservedMatchProbe({ + room_id: roomId, + game_id: room.game_id, + local_user_id: userId, + peer_user_id: participants.peerUserId, + role: participants.role, + peer_endpoint: peerEndpoint.endpoint, + peer_nonce: peerEndpoint.nonce, + }); + if (cancelled || probeAttempt.current !== attempt) return; + setProbeReport(report); + await relayUntilDelivered( + socket, + "match.probe.completed", + { + room_id: roomId, + frames_received: report.frames_received, + transcript_checksum: report.transcript_checksum, + }, + () => cancelled || probeAttempt.current !== attempt, + (message) => { + if (!cancelled && probeAttempt.current === attempt) setProbeError(message); + } + ); + } catch (error) { + if (!cancelled && probeAttempt.current === attempt) { + setProbeError(errorMessage(error, "LAN match probe failed")); + } + } + }; + void run(); + return () => { + cancelled = true; + }; + }, [candidateRelayed, localEndpoint, peerEndpoint, room, roomId, socket, userId]); + + useEffect(() => { + if (!probeReport || !peerCompletion || !room || roomAdvanced.current) return; + const participants = matchParticipants(room, userId); + if (!participants || participants.role !== "host") return; + if ( + peerCompletion.frames_received !== probeReport.frames_received || + peerCompletion.transcript_checksum !== probeReport.transcript_checksum + ) { + setProbeError("Peer transcript does not match the local LAN probe"); + return; + } + roomAdvanced.current = true; + const advance = async () => { + try { + await api.startRoom(token, roomId); + await api.finishRoom(token, roomId); + await queryClient.invalidateQueries({ queryKey: ["room", roomId] }); + } catch (error) { + roomAdvanced.current = false; + setProbeError(errorMessage(error, "Failed to finish match room")); + } + }; + void advance(); + }, [peerCompletion, probeReport, queryClient, room, roomId, token, userId]); + + const retry = async () => { + if (resetStarted.current) return; + resetStarted.current = true; + probeAttempt.current += 1; + setIsResetting(true); + try { + await reservationInFlight.current?.catch(() => undefined); + await cancelMatchProbe(roomId); + probeStarted.current = false; + roomAdvanced.current = false; + setProbeError(null); + setProbeReport(null); + setCandidateRelayed(false); + setLocalEndpoint(null); + onRetry(); + } catch (error) { + setProbeError(errorMessage(error, "Could not reset the LAN probe; try again")); + } finally { + resetStarted.current = false; + setIsResetting(false); + } + }; + + return { localEndpoint, probeReport, probeError, isResetting, retry }; +} + +async function relayUntilDelivered( + socket: OpenFightSocket, + type: string, + payload: unknown, + cancelled: () => boolean, + onWaiting: (message: string | null) => void +): Promise { + while (!cancelled()) { + try { + await socket.send(type, payload); + onWaiting(null); + return true; + } catch (error) { + onWaiting(`${errorMessage(error, "Peer unavailable")}; waiting for peer connection`); + await new Promise((resolve) => window.setTimeout(resolve, 1_000)); + } + } + return false; +} + +function errorMessage(error: unknown, fallback: string): string { + return error instanceof Error ? error.message : fallback; +} diff --git a/apps/client/src/routes/Match.tsx b/apps/client/src/routes/Match.tsx index 1363f9b..1298305 100644 --- a/apps/client/src/routes/Match.tsx +++ b/apps/client/src/routes/Match.tsx @@ -1,14 +1,27 @@ import { useQuery } from "@tanstack/react-query"; +import type { MatchEndpointPayload, MatchProbeCompletedPayload } from "@openfight/protocol"; import { api } from "../lib/api"; import { downloadMatchReport } from "../lib/report"; +import { useLanMatchProbe } from "../lib/useLanMatchProbe"; +import type { OpenFightSocket } from "../lib/ws"; export default function Match({ token, + userId, roomId, + socket, + peerEndpoint, + peerCompletion, + onProbeRetry, onDone, }: { token: string; + userId: string; roomId: string; + socket: OpenFightSocket | null; + peerEndpoint?: MatchEndpointPayload; + peerCompletion?: MatchProbeCompletedPayload; + onProbeRetry: () => void; onDone: () => void; }) { const room = useQuery({ @@ -16,6 +29,16 @@ export default function Match({ queryFn: () => api.room(token, roomId), refetchInterval: 2_000, }); + const { localEndpoint, probeReport, probeError, isResetting, retry } = useLanMatchProbe({ + token, + userId, + roomId, + room: room.data, + socket, + peerEndpoint, + peerCompletion, + onRetry: onProbeRetry, + }); const state = room.data?.state ?? "connecting"; const steps = ["connecting", "playing", "finished"]; const active = Math.max(0, steps.indexOf(state)); @@ -31,22 +54,50 @@ export default function Match({

Room {roomId.slice(0, 8)}

{heading}

-
+ -
    +
      {steps.map((step, index) => (
    1. {step}
      2. ))}
    - {room.isError &&

    {room.error.message}

    } + {localEndpoint && !probeReport && ( +

    + LAN endpoint reserved at {localEndpoint.endpoint} +

    + )} + {probeReport && ( +

    + Direct UDP verified: {probeReport.frames_received} frames in {probeReport.elapsed_ms} ms · + transcript {probeReport.transcript_checksum} +

    + )} + {probeError && ( +

    + {probeError} +

    + )} + {room.isError && ( +

    + {room.error.message} +

    + )}
    + {probeError && ( + + )} {room.data && ( - )} diff --git a/apps/client/src/styles.css b/apps/client/src/styles.css index 9969dbe..0f5a5cb 100644 --- a/apps/client/src/styles.css +++ b/apps/client/src/styles.css @@ -278,6 +278,10 @@ input { border-color: #45464b; color: #e8e5df; } +.secondary:disabled { + opacity: 0.5; + cursor: wait; +} .form-error { color: #ff8e86; margin: 0; @@ -374,8 +378,12 @@ input { } .match-actions { display: flex; + flex-wrap: wrap; gap: 12px; } +.match-actions .secondary { + min-height: 44px; +} @media (max-width: 760px) { .auth-layout { grid-template-columns: 1fr; diff --git a/apps/server/Cargo.toml b/apps/server/Cargo.toml index d362cc8..1a4d8b6 100644 --- a/apps/server/Cargo.toml +++ b/apps/server/Cargo.toml @@ -15,7 +15,7 @@ tower = { version = "0.4", features = ["util"] } tower-http = { version = "0.5", features = ["cors", "trace"] } serde = { version = "1", features = ["derive"] } serde_json = "1" -sqlx = { version = "0.7", features = ["runtime-tokio", "postgres", "chrono", "uuid", "migrate", "macros"] } +sqlx = { version = "0.8.6", default-features = false, features = ["runtime-tokio", "postgres", "chrono", "uuid", "migrate", "macros"] } argon2 = "0.5" chrono = { version = "0.4", features = ["serde"] } uuid = { version = "1", features = ["v4", "serde"] } diff --git a/apps/server/Dockerfile b/apps/server/Dockerfile index 6aff98c..060b0d7 100644 --- a/apps/server/Dockerfile +++ b/apps/server/Dockerfile @@ -25,14 +25,16 @@ RUN mkdir -p apps/server/src packages/protocol/src packages/networking/src \ && echo "" > packages/shared/src/lib.rs \ && echo "" > packages/game-definitions/src/lib.rs \ && echo "" > adapters/fbneo/src/lib.rs \ - && cargo build --release 2>/dev/null || true + && cargo build --release --locked # Copy the real sources (invalidates cache only when source changes). COPY . . -# Build the server binary. -RUN cargo build --release --locked -p openfight-server \ - && ls -lh target/release/openfight-server* 2>/dev/null || ls -lh target/release/ +# Remove only the dummy workspace artifacts before compiling the real sources. Dependency +# artifacts remain cached, while Cargo can no longer mistake a stub crate for the application. +RUN cargo clean --release -p openfight-server -p openfight-protocol \ + && cargo build --release --locked -p openfight-server \ + && test -x target/release/openfight-server # --- runtime --------------------------------------------------------------- FROM debian:bookworm-slim AS runtime diff --git a/apps/server/src/ws.rs b/apps/server/src/ws.rs index e77df4a..41fe1e7 100644 --- a/apps/server/src/ws.rs +++ b/apps/server/src/ws.rs @@ -6,7 +6,10 @@ use axum::{ http::{header, HeaderMap}, response::{IntoResponse, Response}, }; -use openfight_protocol::{is_supported_version, Envelope, PROTOCOL_VERSION}; +use openfight_protocol::{ + is_supported_version, Envelope, MatchEndpointPayload, MatchProbeCompletedPayload, + PROTOCOL_VERSION, +}; use serde_json::{json, Value}; use sqlx::Row; use std::borrow::Cow; @@ -181,7 +184,32 @@ async fn handle_text( ) .await } - "signaling.offer" | "signaling.answer" | "signaling.candidate" => { + "signaling.offer" + | "signaling.answer" + | "signaling.candidate" + | "match.endpoint" + | "match.probe.completed" => { + if envelope.msg_type == "match.endpoint" && validate_match_endpoint(&envelope).is_err() + { + return send_error( + socket, + "invalid_candidate", + "match endpoint candidate is invalid", + Some(&envelope.request_id), + ) + .await; + } + if envelope.msg_type == "match.probe.completed" + && validate_match_completion(&envelope).is_err() + { + return send_error( + socket, + "invalid_probe_report", + "match probe completion is invalid", + Some(&envelope.request_id), + ) + .await; + } if let Err(error) = relay_to_room_members(state, user.id, &envelope, text).await { return send_error( socket, @@ -194,7 +222,11 @@ async fn handle_text( send_envelope( socket, Envelope::reply( - "signaling.relayed", + match envelope.msg_type.as_str() { + "match.endpoint" => "match.endpoint.relayed", + "match.probe.completed" => "match.probe.completed.relayed", + _ => "signaling.relayed", + }, envelope.request_id, json!({ "status": "relayed" }), ), @@ -213,6 +245,34 @@ async fn handle_text( } } +fn validate_match_endpoint(envelope: &Envelope) -> Result<(), ()> { + let candidate: MatchEndpointPayload = + serde_json::from_value(envelope.payload.clone()).map_err(|_| ())?; + candidate + .endpoint + .parse::() + .map_err(|_| ())?; + Uuid::parse_str(&candidate.nonce).map_err(|_| ())?; + Ok(()) +} + +fn validate_match_completion(envelope: &Envelope) -> Result<(), ()> { + let completion: MatchProbeCompletedPayload = + serde_json::from_value(envelope.payload.clone()).map_err(|_| ())?; + if completion.frames_received == 0 || completion.frames_received > 10_000 { + return Err(()); + } + if completion.transcript_checksum.len() != 16 + || !completion + .transcript_checksum + .bytes() + .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase()) + { + return Err(()); + } + Ok(()) +} + async fn relay_to_room_members( state: &AppState, sender_id: Uuid, @@ -342,4 +402,60 @@ mod tests { ); assert_eq!(websocket_token(&headers), Some("secret-token")); } + + #[test] + fn match_endpoint_requires_a_socket_address_and_uuid_nonce() { + let valid = Envelope::new( + "match.endpoint", + json!({ + "room_id": Uuid::new_v4(), + "endpoint": "192.168.1.20:42000", + "nonce": Uuid::new_v4() + }), + ); + assert!(validate_match_endpoint(&valid).is_ok()); + + let invalid_endpoint = Envelope::new( + "match.endpoint", + json!({ + "room_id": Uuid::new_v4(), + "endpoint": "not-an-endpoint", + "nonce": Uuid::new_v4() + }), + ); + assert!(validate_match_endpoint(&invalid_endpoint).is_err()); + + let invalid_nonce = Envelope::new( + "match.endpoint", + json!({ + "room_id": Uuid::new_v4(), + "endpoint": "192.168.1.20:42000", + "nonce": "predictable" + }), + ); + assert!(validate_match_endpoint(&invalid_nonce).is_err()); + } + + #[test] + fn match_completion_requires_bounded_frames_and_a_checksum() { + let valid = Envelope::new( + "match.probe.completed", + json!({ + "room_id": Uuid::new_v4(), + "frames_received": 60, + "transcript_checksum": "0376c2e852f4fd25" + }), + ); + assert!(validate_match_completion(&valid).is_ok()); + + let invalid = Envelope::new( + "match.probe.completed", + json!({ + "room_id": Uuid::new_v4(), + "frames_received": 0, + "transcript_checksum": "not-a-checksum" + }), + ); + assert!(validate_match_completion(&invalid).is_err()); + } } diff --git a/apps/server/tests/api_flow.rs b/apps/server/tests/api_flow.rs index f8d321f..a045a18 100644 --- a/apps/server/tests/api_flow.rs +++ b/apps/server/tests/api_flow.rs @@ -241,6 +241,81 @@ async fn authenticated_users_create_and_accept_a_room(pool: PgPool) { assert_eq!(acknowledgement.msg_type, "signaling.relayed"); assert_eq!(acknowledgement.request_id, request_id); + let endpoint_request_id = "endpoint-integration-1"; + let endpoint = json!({ + "type": "match.endpoint", + "version": "1.0", + "request_id": endpoint_request_id, + "timestamp": chrono::Utc::now(), + "payload": { + "room_id": room_id, + "endpoint": "192.168.1.20:42000", + "nonce": "8a1110d5-8dd2-4ad2-9c88-ad9768bc4905" + } + }); + host_socket + .send(Message::Text(endpoint.to_string())) + .await + .expect("send endpoint"); + let relayed_endpoint = guest_socket + .next() + .await + .expect("relayed endpoint") + .expect("guest endpoint frame"); + let relayed_endpoint: Envelope = + serde_json::from_str(relayed_endpoint.to_text().expect("text endpoint")) + .expect("endpoint envelope"); + assert_eq!(relayed_endpoint.msg_type, "match.endpoint"); + assert_eq!(relayed_endpoint.request_id, endpoint_request_id); + assert_eq!(relayed_endpoint.payload["endpoint"], "192.168.1.20:42000"); + let endpoint_ack = host_socket + .next() + .await + .expect("endpoint ack") + .expect("host endpoint frame"); + let endpoint_ack: Envelope = + serde_json::from_str(endpoint_ack.to_text().expect("text endpoint ack")) + .expect("endpoint ack envelope"); + assert_eq!(endpoint_ack.msg_type, "match.endpoint.relayed"); + assert_eq!(endpoint_ack.request_id, endpoint_request_id); + + let completion_request_id = "completion-integration-1"; + let completion = json!({ + "type": "match.probe.completed", + "version": "1.0", + "request_id": completion_request_id, + "timestamp": chrono::Utc::now(), + "payload": { + "room_id": room_id, + "frames_received": 60, + "transcript_checksum": "0376c2e852f4fd25" + } + }); + guest_socket + .send(Message::Text(completion.to_string())) + .await + .expect("send completion"); + let relayed_completion = host_socket + .next() + .await + .expect("relayed completion") + .expect("host completion frame"); + let relayed_completion: Envelope = + serde_json::from_str(relayed_completion.to_text().expect("text completion")) + .expect("completion envelope"); + assert_eq!(relayed_completion.msg_type, "match.probe.completed"); + assert_eq!(relayed_completion.request_id, completion_request_id); + let completion_ack = guest_socket + .next() + .await + .expect("completion ack") + .expect("guest completion frame"); + let completion_ack: Envelope = + serde_json::from_str(completion_ack.to_text().expect("text completion ack")) + .expect("completion ack envelope"); + assert_eq!(completion_ack.msg_type, "match.probe.completed.relayed"); + assert_eq!(completion_ack.request_id, completion_request_id); + let (start_status, started) = request( &app, Method::POST, diff --git a/docs/IMPLEMENTATION_STATUS.md b/docs/IMPLEMENTATION_STATUS.md index d33dcae..9f85e2f 100644 --- a/docs/IMPLEMENTATION_STATUS.md +++ b/docs/IMPLEMENTATION_STATUS.md @@ -12,11 +12,15 @@ Updated 2026-08-23. acknowledgements and errors. - Rust-authoritative protocol payloads and generated TypeScript bindings with a CI drift gate. - React/Tauri login, games, lobby, challenge, room status, reconnect, local availability scan, - diagnostics, and redacted report export. + diagnostics, and redacted report export with direct-UDP frame/checksum evidence. - Safe process abstraction, canonical root checks, `PathBuf`/`OsString` arguments, process tracking, and FBNeo local detection/validation/launch. -- Deterministic mock adapter, bounded in-memory input transport, and connected direct UDP transport. -- PostgreSQL, WebSocket, lifecycle, safe-launch, mock-match, UDP, TypeScript, and MSRV checks in CI. +- Deterministic mock adapter, bounded in-memory input transport, and a nonce-bound direct-UDP match + runner wired through authenticated endpoint exchange and the desktop match screen. +- A standalone two-node probe CLI plus a real two-process localhost test that verifies identical + 60-frame transcripts and machine-readable reports. +- PostgreSQL, WebSocket, lifecycle, safe-launch, mock-match, UDP, two-process, TypeScript, and MSRV + checks in CI. ## Deliberately not claimed diff --git a/docs/alpha/LAN_TEST.md b/docs/alpha/LAN_TEST.md index a9eaca2..748619c 100644 --- a/docs/alpha/LAN_TEST.md +++ b/docs/alpha/LAN_TEST.md @@ -18,17 +18,35 @@ direct UDP frame transport; it does not claim FBNeo netplay support. 1. Register two separate users and select the same game. 2. Keep both lobby screens open until each user is visible. 3. User A sends a challenge; user B accepts it. -4. Confirm both clients reach `connecting` and exchange a correlated signaling message. -5. Run the direct-UDP proof harness on the selected LAN addresses and record connection time, - frame count, packet loss, and disconnect reason. -6. Transition the test room to `playing`, then `finished`. -7. Export the redacted report from each client and compare `room.id`, `game_id`, and final state. +4. Confirm both clients reach `connecting`. Each desktop client reserves a UDP port and exchanges a + nonce-bound `match.endpoint` candidate through the authenticated WebSocket. +5. Wait for both clients to report `Direct UDP verified`, with 60 received frames and the same + transcript checksum. The host then transitions the room to `playing` and `finished`. +6. If either side reports a firewall or timeout error, allow the advertised UDP port through the + firewall and select `Retry LAN probe` on both clients. +7. Export the redacted report from each client and compare `room.id`, `game_id`, final state, + `probe.frames_received`, and `probe.transcript_checksum`. Probe evidence deliberately omits + nonces, peer endpoints, and local user identifiers. + +For transport-only diagnosis without the desktop flow, build `openfight-match-probe` and run one +process on each host with complementary arguments: + +```bash +cargo run -p openfight-networking --bin openfight-match-probe -- \ + --local 192.168.1.10:42000 --peer 192.168.1.11:42000 \ + --room lan-test --game sfiii3 --local-user host --peer-user guest \ + --role host --session-key shared-test-key --frames 60 --timeout-ms 5000 +``` + +The other host swaps local/peer addresses and users and uses `--role guest` with the same room, +game, session key, frame count, and timeout. The command prints a machine-readable JSON report. ## Pass criteria - Both clients agree on the room and users. - No non-member can mutate or signal into the room. - The UDP transcript is ordered and identical at both endpoints. +- Both reports contain the same `transcript_checksum` and `frames_received = 60`. - The match row has `started_at` and `ended_at`. - Reports contain no session token, password, full ROM path, or emulator binary. diff --git a/packages/networking/Cargo.toml b/packages/networking/Cargo.toml index 31d0fea..d3a1e22 100644 --- a/packages/networking/Cargo.toml +++ b/packages/networking/Cargo.toml @@ -8,10 +8,10 @@ repository = "https://github.com/openfight/openfight" rust-version = "1.78" [dependencies] +openfight-emulator-sdk = { path = "../emulator-sdk" } serde = { version = "1", features = ["derive"] } serde_json = "1" thiserror = "1" tokio = { version = "1", features = ["macros", "net", "rt", "sync", "time"] } [dev-dependencies] -openfight-emulator-sdk = { path = "../emulator-sdk" } diff --git a/packages/networking/src/bin/openfight-match-probe.rs b/packages/networking/src/bin/openfight-match-probe.rs new file mode 100644 index 0000000..a88a498 --- /dev/null +++ b/packages/networking/src/bin/openfight-match-probe.rs @@ -0,0 +1,104 @@ +use openfight_emulator_sdk::{MatchDescriptor, PeerRole, TransportKind}; +use openfight_networking::{run_match_probe, MatchProbeConfig, UdpPeer}; +use std::env; +use std::net::SocketAddr; +use std::time::Duration; + +struct Arguments { + local: SocketAddr, + peer: SocketAddr, + room_id: String, + game_id: String, + local_user_id: String, + peer_user_id: String, + role: PeerRole, + session_key: String, + frames: u64, + timeout: Duration, +} + +impl Arguments { + fn parse() -> Result { + let mut args = env::args().skip(1); + let mut value = |flag: &str| -> Result { + let actual = args + .next() + .ok_or_else(|| format!("missing {flag}; {}", usage()))?; + if actual != flag { + return Err(format!("expected {flag}, found {actual}; {}", usage())); + } + args.next() + .ok_or_else(|| format!("missing value for {flag}; {}", usage())) + }; + + let local = value("--local")? + .parse() + .map_err(|_| "--local must be a socket address".to_string())?; + let peer = value("--peer")? + .parse() + .map_err(|_| "--peer must be a socket address".to_string())?; + let room_id = value("--room")?; + let game_id = value("--game")?; + let local_user_id = value("--local-user")?; + let peer_user_id = value("--peer-user")?; + let role = match value("--role")?.as_str() { + "host" => PeerRole::Host, + "guest" => PeerRole::Guest, + _ => return Err("--role must be host or guest".into()), + }; + let session_key = value("--session-key")?; + let frames = value("--frames")? + .parse() + .map_err(|_| "--frames must be an integer".to_string())?; + let timeout_ms: u64 = value("--timeout-ms")? + .parse() + .map_err(|_| "--timeout-ms must be an integer".to_string())?; + if let Some(extra) = args.next() { + return Err(format!("unexpected argument {extra}; {}", usage())); + } + Ok(Self { + local, + peer, + room_id, + game_id, + local_user_id, + peer_user_id, + role, + session_key, + frames, + timeout: Duration::from_millis(timeout_ms), + }) + } +} + +fn usage() -> &'static str { + "usage: openfight-match-probe --local IP:PORT --peer IP:PORT --room ID --game ID \ + --local-user ID --peer-user ID --role host|guest --session-key KEY --frames N --timeout-ms N" +} + +#[tokio::main(flavor = "current_thread")] +async fn main() -> Result<(), Box> { + let arguments = Arguments::parse() + .map_err(|message| std::io::Error::new(std::io::ErrorKind::InvalidInput, message))?; + let descriptor = MatchDescriptor { + room_id: arguments.room_id, + game_id: arguments.game_id, + local_user_id: arguments.local_user_id, + peer_user_id: arguments.peer_user_id, + role: arguments.role, + transport: TransportKind::DirectUdp, + local_endpoint: arguments.local, + peer_endpoint: arguments.peer, + input_delay_frames: 2, + }; + let config = MatchProbeConfig::new( + descriptor, + arguments.session_key, + arguments.frames, + arguments.timeout, + )?; + let peer = UdpPeer::bind(arguments.local, arguments.peer).await?; + let report = run_match_probe(&peer, &config).await?; + println!("{}", serde_json::to_string(&report)?); + Ok(()) +} diff --git a/packages/networking/src/lib.rs b/packages/networking/src/lib.rs index 2c77dea..ae6c14e 100644 --- a/packages/networking/src/lib.rs +++ b/packages/networking/src/lib.rs @@ -2,8 +2,13 @@ use serde::{Deserialize, Serialize}; use std::net::SocketAddr; use tokio::sync::mpsc; +mod probe; + +pub use probe::{run_match_probe, MatchProbeConfig, MatchProbeReport, MAX_PROBE_FRAMES}; + pub const MAX_INPUT_BYTES: usize = 256; pub const INPUT_QUEUE_CAPACITY: usize = 120; +const MAX_DATAGRAM_BYTES: usize = 2_048; #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct InputFrame { @@ -47,6 +52,17 @@ pub enum TransportError { Serialization(String), #[error("udp transport failed: {0}")] Io(String), + #[error("udp peer is not ready")] + PeerUnavailable, + #[error("invalid match probe configuration: {0}")] + InvalidConfiguration(String), + #[error("received a datagram for a different match or peer")] + PeerMismatch, + #[error("match probe timed out after receiving {received_frames} of {expected_frames} frames")] + Timeout { + received_frames: u64, + expected_frames: u64, + }, } /// Connected UDP transport for LAN proof runs. Authentication and endpoint negotiation remain in @@ -57,20 +73,24 @@ pub struct UdpPeer { impl UdpPeer { pub async fn bind(local: SocketAddr, peer: SocketAddr) -> Result { + let socket = Self::bind_unconnected(local).await?; + socket.connect(peer).await + } + + pub async fn bind_unconnected(local: SocketAddr) -> Result { let socket = tokio::net::UdpSocket::bind(local) .await - .map_err(|error| TransportError::Io(error.to_string()))?; - socket - .connect(peer) - .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(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> { @@ -79,24 +99,44 @@ 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; 1024]; - let received = self - .socket - .recv(&mut buffer) - .await - .map_err(|error| TransportError::Io(error.to_string()))?; + let mut buffer = [0_u8; MAX_DATAGRAM_BYTES]; + 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) } + + async fn send_packet(&self, packet: &probe::ProbePacket) -> Result<(), TransportError> { + let encoded = serde_json::to_vec(packet) + .map_err(|error| TransportError::Serialization(error.to_string()))?; + if encoded.len() > MAX_DATAGRAM_BYTES { + return Err(TransportError::Serialization( + "match probe datagram exceeds maximum size".into(), + )); + } + 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(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 error.kind() == std::io::ErrorKind::ConnectionRefused { + TransportError::PeerUnavailable + } else { + TransportError::Io(error.to_string()) + } } /// Deterministic, bounded transport used to prove the match contract without a network. @@ -139,6 +179,7 @@ mod tests { use openfight_emulator_sdk::{ EmulatorAdapter, MatchDescriptor, MockAdapter, PeerRole, TransportKind, }; + use std::time::Duration; fn descriptor(role: PeerRole) -> MatchDescriptor { let (local_user_id, peer_user_id, local_endpoint, peer_endpoint) = match role { @@ -215,6 +256,12 @@ mod tests { assert_eq!(error, Err(TransportError::FrameTooLarge(257))); } + #[test] + fn connection_refused_is_a_retryable_udp_startup_race() { + let error = std::io::Error::from(std::io::ErrorKind::ConnectionRefused); + assert_eq!(map_udp_error(error), TransportError::PeerUnavailable); + } + #[test] fn bounded_queue_applies_backpressure() { let (host, _guest) = InMemoryPeer::pair(); @@ -249,4 +296,83 @@ mod tests { .expect("udp receive"); assert_eq!(received, expected); } + + #[tokio::test] + async fn udp_match_probes_agree_on_a_deterministic_transcript() { + let reserve_host = tokio::net::UdpSocket::bind("127.0.0.1:0") + .await + .expect("reserve host"); + let reserve_guest = tokio::net::UdpSocket::bind("127.0.0.1:0") + .await + .expect("reserve guest"); + let host_address = reserve_host.local_addr().expect("host address"); + let guest_address = reserve_guest.local_addr().expect("guest address"); + drop(reserve_host); + drop(reserve_guest); + + let host_peer = UdpPeer::bind(host_address, guest_address) + .await + .expect("host peer"); + let guest_peer = UdpPeer::bind(guest_address, host_address) + .await + .expect("guest peer"); + let host_config = MatchProbeConfig::new( + direct_descriptor(PeerRole::Host, host_address, guest_address), + "shared-session", + 60, + Duration::from_secs(2), + ) + .expect("host config"); + let guest_config = MatchProbeConfig::new( + direct_descriptor(PeerRole::Guest, guest_address, host_address), + "shared-session", + 60, + Duration::from_secs(2), + ) + .expect("guest config"); + + let (host, guest) = tokio::join!( + run_match_probe(&host_peer, &host_config), + run_match_probe(&guest_peer, &guest_config) + ); + let host = host.expect("host report"); + let guest = guest.expect("guest report"); + assert_eq!(host.frames_received, 60); + assert_eq!(guest.frames_received, 60); + assert_eq!(host.transcript_checksum, guest.transcript_checksum); + } + + #[test] + fn match_probe_rejects_an_in_memory_descriptor() { + let error = MatchProbeConfig::new( + descriptor(PeerRole::Host), + "shared-session", + 60, + Duration::from_secs(2), + ) + .expect_err("in-memory descriptor must be rejected"); + assert!(matches!(error, TransportError::InvalidConfiguration(_))); + } + + fn direct_descriptor( + role: PeerRole, + local_endpoint: SocketAddr, + peer_endpoint: SocketAddr, + ) -> MatchDescriptor { + let (local_user_id, peer_user_id) = match role { + PeerRole::Host => ("host", "guest"), + PeerRole::Guest => ("guest", "host"), + }; + MatchDescriptor { + room_id: "proof-room".into(), + game_id: "sfiii3".into(), + local_user_id: local_user_id.into(), + peer_user_id: peer_user_id.into(), + role, + transport: TransportKind::DirectUdp, + local_endpoint, + peer_endpoint, + input_delay_frames: 2, + } + } } diff --git a/packages/networking/src/probe.rs b/packages/networking/src/probe.rs new file mode 100644 index 0000000..5f1bb1f --- /dev/null +++ b/packages/networking/src/probe.rs @@ -0,0 +1,361 @@ +use super::{InputFrame, TransportError, UdpPeer, MAX_INPUT_BYTES}; +use openfight_emulator_sdk::{MatchDescriptor, PeerRole, TransportKind}; +use serde::{Deserialize, Serialize}; +use std::time::{Duration, Instant}; + +pub const MAX_PROBE_FRAMES: u64 = 10_000; +const RETRY_INTERVAL: Duration = Duration::from_millis(25); +// 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)] +#[serde(tag = "type", rename_all = "snake_case")] +pub(super) enum ProbePacket { + Input { + room_id: String, + session_key: String, + frame: InputFrame, + }, +} + +#[derive(Debug, Clone)] +pub struct MatchProbeConfig { + descriptor: MatchDescriptor, + session_key: String, + frame_count: u64, + timeout: Duration, +} + +impl MatchProbeConfig { + pub fn new( + descriptor: MatchDescriptor, + session_key: impl Into, + frame_count: u64, + timeout: Duration, + ) -> Result { + descriptor + .validate() + .map_err(|error| TransportError::InvalidConfiguration(error.to_string()))?; + if descriptor.transport != TransportKind::DirectUdp { + return Err(TransportError::InvalidConfiguration( + "match probe requires direct_udp transport".into(), + )); + } + let session_key = session_key.into(); + if session_key.trim().is_empty() || session_key.len() > 128 { + return Err(TransportError::InvalidConfiguration( + "session key must contain between 1 and 128 bytes".into(), + )); + } + if !(1..=MAX_PROBE_FRAMES).contains(&frame_count) { + return Err(TransportError::InvalidConfiguration(format!( + "frame count must be between 1 and {MAX_PROBE_FRAMES}" + ))); + } + if timeout.is_zero() { + return Err(TransportError::InvalidConfiguration( + "timeout must be greater than zero".into(), + )); + } + Ok(Self { + descriptor, + session_key, + frame_count, + timeout, + }) + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct MatchProbeReport { + pub room_id: String, + pub local_user_id: String, + pub peer_user_id: String, + pub role: PeerRole, + pub transport: TransportKind, + pub frames_sent: u64, + pub frames_received: u64, + pub transcript_checksum: String, + pub elapsed_ms: u128, +} + +/// Runs a bounded deterministic frame exchange over a connected UDP peer. +/// +/// A peer repeats its current frame until the opposite frame arrives. Receiving an older frame +/// causes the matching local frame to be retransmitted, allowing either process to start first. +pub async fn run_match_probe( + peer: &UdpPeer, + config: &MatchProbeConfig, +) -> Result { + let started = Instant::now(); + let deadline = started + config.timeout; + let mut checksum = FNV_OFFSET_BASIS; + let mut frames_sent = 0; + + for frame_number in 0..config.frame_count { + let local_frame = deterministic_frame( + frame_number, + &config.descriptor.local_user_id, + config.descriptor.role, + )?; + send_probe_frame(peer, config, local_frame.clone()).await?; + frames_sent += 1; + + loop { + let now = Instant::now(); + if now >= deadline { + return Err(TransportError::Timeout { + received_frames: frame_number, + expected_frames: config.frame_count, + }); + } + let wait = RETRY_INTERVAL.min(deadline.saturating_duration_since(now)); + match tokio::time::timeout(wait, peer.receive_packet()).await { + Err(_) => { + send_probe_frame(peer, config, local_frame.clone()).await?; + frames_sent += 1; + } + 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 { + let previous = deterministic_frame( + remote.frame, + &config.descriptor.local_user_id, + config.descriptor.role, + )?; + send_probe_frame(peer, config, previous).await?; + frames_sent += 1; + continue; + } + if remote.frame > frame_number { + send_probe_frame(peer, config, local_frame.clone()).await?; + frames_sent += 1; + continue; + } + update_transcript_checksum( + &mut checksum, + config.descriptor.role, + &local_frame, + &remote, + ); + break; + } + } + } + } + + linger_for_peer(peer, config, &mut frames_sent).await?; + Ok(MatchProbeReport { + room_id: config.descriptor.room_id.clone(), + local_user_id: config.descriptor.local_user_id.clone(), + peer_user_id: config.descriptor.peer_user_id.clone(), + role: config.descriptor.role, + transport: config.descriptor.transport, + frames_sent, + frames_received: config.frame_count, + transcript_checksum: format!("{checksum:016x}"), + elapsed_ms: started.elapsed().as_millis(), + }) +} + +async fn send_probe_frame( + peer: &UdpPeer, + config: &MatchProbeConfig, + frame: InputFrame, +) -> Result<(), TransportError> { + 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( + packet: ProbePacket, + config: &MatchProbeConfig, +) -> Result { + let ProbePacket::Input { + room_id, + session_key, + frame, + } = packet; + if room_id != config.descriptor.room_id + || session_key != config.session_key + || frame.player_id != config.descriptor.peer_user_id + || frame.input.len() > MAX_INPUT_BYTES + { + return Err(TransportError::PeerMismatch); + } + Ok(frame) +} + +async fn linger_for_peer( + peer: &UdpPeer, + config: &MatchProbeConfig, + frames_sent: &mut u64, +) -> Result<(), TransportError> { + let final_frame_number = config.frame_count - 1; + let final_frame = deterministic_frame( + final_frame_number, + &config.descriptor.local_user_id, + config.descriptor.role, + )?; + let deadline = Instant::now() + COMPLETION_GRACE; + while Instant::now() < deadline { + let wait = RETRY_INTERVAL.min(deadline.saturating_duration_since(Instant::now())); + match tokio::time::timeout(wait, peer.receive_packet()).await { + Err(_) => { + send_probe_frame(peer, config, final_frame.clone()).await?; + *frames_sent += 1; + } + 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)) => { + let remote = validated_remote_frame(packet, config)?; + let response = deterministic_frame( + remote.frame.min(final_frame_number), + &config.descriptor.local_user_id, + config.descriptor.role, + )?; + send_probe_frame(peer, config, response).await?; + *frames_sent += 1; + } + } + } + Ok(()) +} + +fn deterministic_frame( + frame: u64, + player_id: &str, + role: PeerRole, +) -> Result { + let input = match role { + PeerRole::Host => (frame % 4) as u8, + PeerRole::Guest => ((frame + 2) % 4) as u8, + }; + InputFrame::new(frame, player_id, vec![input]) +} + +const FNV_OFFSET_BASIS: u64 = 0xcbf29ce484222325; +const FNV_PRIME: u64 = 0x100000001b3; + +fn update_transcript_checksum( + checksum: &mut u64, + role: PeerRole, + local: &InputFrame, + remote: &InputFrame, +) { + let (host, guest) = match role { + PeerRole::Host => (local, remote), + PeerRole::Guest => (remote, local), + }; + for byte in host.input.iter().chain(&guest.input) { + *checksum ^= u64::from(*byte); + *checksum = checksum.wrapping_mul(FNV_PRIME); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::net::SocketAddr; + + fn config() -> MatchProbeConfig { + MatchProbeConfig::new( + MatchDescriptor { + room_id: "room-1".into(), + game_id: "sfiii3".into(), + local_user_id: "host".into(), + peer_user_id: "guest".into(), + role: PeerRole::Host, + transport: TransportKind::DirectUdp, + local_endpoint: "127.0.0.1:41000" + .parse::() + .expect("local endpoint"), + peer_endpoint: "127.0.0.1:41001" + .parse::() + .expect("peer endpoint"), + input_delay_frames: 2, + }, + "session-1", + 60, + Duration::from_secs(1), + ) + .expect("probe config") + } + + #[test] + fn rejects_packets_from_another_room_session_or_player() { + let config = config(); + let valid_frame = InputFrame::new(0, "guest", vec![2]).expect("frame"); + let wrong_room = ProbePacket::Input { + room_id: "room-2".into(), + session_key: "session-1".into(), + frame: valid_frame.clone(), + }; + let wrong_session = ProbePacket::Input { + room_id: "room-1".into(), + session_key: "session-2".into(), + frame: valid_frame, + }; + let wrong_player = ProbePacket::Input { + room_id: "room-1".into(), + session_key: "session-1".into(), + frame: InputFrame::new(0, "intruder", vec![2]).expect("frame"), + }; + + assert_eq!( + validated_remote_frame(wrong_room, &config), + Err(TransportError::PeerMismatch) + ); + assert_eq!( + validated_remote_frame(wrong_session, &config), + Err(TransportError::PeerMismatch) + ); + assert_eq!( + validated_remote_frame(wrong_player, &config), + Err(TransportError::PeerMismatch) + ); + } + + #[test] + fn configuration_rejects_empty_and_oversized_probe_values() { + let descriptor = config().descriptor; + let zero_frames = + MatchProbeConfig::new(descriptor.clone(), "session-1", 0, Duration::from_secs(1)); + assert!(matches!( + zero_frames, + Err(TransportError::InvalidConfiguration(_)) + )); + + let oversized_key = + MatchProbeConfig::new(descriptor, "x".repeat(129), 1, Duration::from_secs(1)); + assert!(matches!( + oversized_key, + Err(TransportError::InvalidConfiguration(_)) + )); + } +} diff --git a/packages/networking/tests/two_process_probe.rs b/packages/networking/tests/two_process_probe.rs new file mode 100644 index 0000000..0dd515b --- /dev/null +++ b/packages/networking/tests/two_process_probe.rs @@ -0,0 +1,76 @@ +use openfight_networking::MatchProbeReport; +use std::net::{SocketAddr, UdpSocket}; +use std::process::{Child, Command, Output, Stdio}; + +fn reserve_addresses() -> (SocketAddr, SocketAddr) { + let host = UdpSocket::bind("127.0.0.1:0").expect("reserve host address"); + let guest = UdpSocket::bind("127.0.0.1:0").expect("reserve guest address"); + let host_address = host.local_addr().expect("read host address"); + let guest_address = guest.local_addr().expect("read guest address"); + (host_address, guest_address) +} + +fn spawn_probe( + local: SocketAddr, + peer: SocketAddr, + local_user: &str, + peer_user: &str, + role: &str, +) -> Child { + Command::new(env!("CARGO_BIN_EXE_openfight-match-probe")) + .args([ + "--local", + &local.to_string(), + "--peer", + &peer.to_string(), + "--room", + "two-process-room", + "--game", + "sfiii3", + "--local-user", + local_user, + "--peer-user", + peer_user, + "--role", + role, + "--session-key", + "two-process-session", + "--frames", + "60", + "--timeout-ms", + "5000", + ]) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .expect("spawn match probe") +} + +fn successful_report(output: Output) -> MatchProbeReport { + assert!( + output.status.success(), + "probe failed: {}", + String::from_utf8_lossy(&output.stderr) + ); + serde_json::from_slice(&output.stdout).expect("valid probe report") +} + +#[test] +fn independent_processes_complete_the_same_udp_transcript() { + let (host_address, guest_address) = reserve_addresses(); + assert_ne!(host_address, guest_address); + + let host = spawn_probe(host_address, guest_address, "host", "guest", "host"); + let guest = spawn_probe(guest_address, host_address, "guest", "host", "guest"); + let host_report = successful_report(host.wait_with_output().expect("host output")); + let guest_report = successful_report(guest.wait_with_output().expect("guest output")); + + assert_eq!(host_report.frames_received, 60); + assert_eq!(guest_report.frames_received, 60); + assert_eq!( + host_report.transcript_checksum, + guest_report.transcript_checksum + ); + assert_eq!(host_report.room_id, "two-process-room"); + assert_eq!(guest_report.room_id, "two-process-room"); +} diff --git a/packages/protocol/src/generated/MatchEndpointPayload.ts b/packages/protocol/src/generated/MatchEndpointPayload.ts new file mode 100644 index 0000000..2ca8c87 --- /dev/null +++ b/packages/protocol/src/generated/MatchEndpointPayload.ts @@ -0,0 +1,6 @@ +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. + +/** + * Direct-UDP endpoint candidate exchanged between authenticated room members. + */ +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 new file mode 100644 index 0000000..92fd040 --- /dev/null +++ b/packages/protocol/src/generated/MatchProbeCompletedPayload.ts @@ -0,0 +1,10 @@ +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. + +/** + * 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; +}; diff --git a/packages/protocol/src/index.ts b/packages/protocol/src/index.ts index 577667b..a0f246e 100644 --- a/packages/protocol/src/index.ts +++ b/packages/protocol/src/index.ts @@ -4,6 +4,8 @@ export { createEnvelope, validateEnvelope, parseEnvelope, serializeEnvelope } fr export type { PresencePayload, ChatPayload, + MatchEndpointPayload, + MatchProbeCompletedPayload, ChallengePayload, ChallengeState, SessionOfferPayload, diff --git a/packages/protocol/src/lib.rs b/packages/protocol/src/lib.rs index 8cada3d..4610728 100644 --- a/packages/protocol/src/lib.rs +++ b/packages/protocol/src/lib.rs @@ -157,6 +157,26 @@ pub struct SessionCandidatePayload { pub sdp_mid: Option, pub sdp_mline_index: Option, } + +/// Direct-UDP endpoint candidate exchanged between authenticated room members. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, TS)] +#[ts(export, export_to = "../src/generated/MatchEndpointPayload.ts")] +#[serde(rename_all = "snake_case")] +pub struct MatchEndpointPayload { + pub room_id: String, + pub endpoint: String, + pub nonce: String, +} + +/// Result of a completed direct-UDP proof, relayed to the other room member. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, TS)] +#[ts(export, export_to = "../src/generated/MatchProbeCompletedPayload.ts")] +#[serde(rename_all = "snake_case")] +pub struct MatchProbeCompletedPayload { + pub room_id: String, + pub frames_received: u32, + pub transcript_checksum: String, +} /// Room lifecycle states. Serialized as `snake_case` strings. /// ARCHITECTURE.md §9 DB stores WAITING/READY/PLAYING/FINISHED/CANCELLED (upper); /// AGENTS.md describes WAITING→CHALLENGING→CONNECTING→PLAYING→FINISHED|CANCELLED. @@ -274,6 +294,35 @@ mod tests { assert_eq!(candidate_back.payload, candidate); } + #[test] + fn roundtrip_match_endpoint() { + let payload = MatchEndpointPayload { + room_id: "room-1".into(), + endpoint: "192.168.1.20:42000".into(), + nonce: "8a1110d5-8dd2-4ad2-9c88-ad9768bc4905".into(), + }; + let encoded = serde_json::to_string(&Envelope::new("match.endpoint", payload.clone())) + .expect("serialize endpoint"); + let decoded: Envelope = + serde_json::from_str(&encoded).expect("deserialize endpoint"); + assert_eq!(decoded.payload, payload); + } + + #[test] + fn roundtrip_match_probe_completion() { + let payload = MatchProbeCompletedPayload { + room_id: "room-1".into(), + frames_received: 60, + transcript_checksum: "0376c2e852f4fd25".into(), + }; + let encoded = + serde_json::to_string(&Envelope::new("match.probe.completed", payload.clone())) + .expect("serialize completion"); + let decoded: Envelope = + serde_json::from_str(&encoded).expect("deserialize completion"); + assert_eq!(decoded.payload, payload); + } + #[test] fn roundtrip_room() { let payload = RoomPayload { diff --git a/packages/protocol/src/messages.ts b/packages/protocol/src/messages.ts index 19f4399..f952105 100644 --- a/packages/protocol/src/messages.ts +++ b/packages/protocol/src/messages.ts @@ -1,5 +1,7 @@ import type { ChallengePayload } from "./generated/ChallengePayload.js"; import type { ChatPayload } from "./generated/ChatPayload.js"; +import type { MatchEndpointPayload } from "./generated/MatchEndpointPayload.js"; +import type { MatchProbeCompletedPayload } from "./generated/MatchProbeCompletedPayload.js"; import type { PresencePayload } from "./generated/PresencePayload.js"; import type { RoomPayload } from "./generated/RoomPayload.js"; import type { SessionAnswerPayload } from "./generated/SessionAnswerPayload.js"; @@ -9,6 +11,8 @@ import type { SessionOfferPayload } from "./generated/SessionOfferPayload.js"; export type { ChallengePayload, ChatPayload, + MatchEndpointPayload, + MatchProbeCompletedPayload, PresencePayload, RoomPayload, SessionAnswerPayload, @@ -40,6 +44,10 @@ export type EnvelopeType = | "signaling.answer" | "signaling.candidate" | "signaling.relayed" + | "match.endpoint" + | "match.endpoint.relayed" + | "match.probe.completed" + | "match.probe.completed.relayed" | "room.state" | "connection.hello" | "error" @@ -58,6 +66,10 @@ const KNOWN_ENVELOPE_TYPES: ReadonlySet = new Set([ "signaling.answer", "signaling.candidate", "signaling.relayed", + "match.endpoint", + "match.endpoint.relayed", + "match.probe.completed", + "match.probe.completed.relayed", "room.state", "connection.hello", "error",