diff --git a/Cargo.lock b/Cargo.lock index cff621a6..d8b5c6fa 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4759,10 +4759,12 @@ dependencies = [ name = "skate-net" version = "0.1.0" dependencies = [ + "blake3", "half", "serde", "serde_json", "skate-core", + "socket2", ] [[package]] @@ -4857,6 +4859,16 @@ dependencies = [ "serde", ] +[[package]] +name = "socket2" +version = "0.6.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4" +dependencies = [ + "libc", + "windows-sys 0.61.2", +] + [[package]] name = "spade" version = "2.15.1" diff --git a/crates/skate-game/src/custom_models.rs b/crates/skate-game/src/custom_models.rs index b005d7ad..89c87f7c 100644 --- a/crates/skate-game/src/custom_models.rs +++ b/crates/skate-game/src/custom_models.rs @@ -101,9 +101,6 @@ pub(crate) struct CustomModels { dirty: bool, } impl CustomModels { - pub(crate) fn online_imports(&self) -> Vec<(String,PathBuf)> { - self.entries.iter().filter(|e|e.native.is_none()).map(|e|(e.asset_path("character.glb"),self.directory.join("entries").join(&e.id).join("character.glb"))).collect() - } pub(crate) fn online_selection(&self) -> Option<(Option, PathBuf)> { let e = self.entries.iter().find(|e| Some(&e.id)==self.active.as_ref())?; let directory = if e.asset_prefix.is_empty() { &self.directory } else { &self.native_directory }; diff --git a/crates/skate-game/src/multiplayer/appearance.rs b/crates/skate-game/src/multiplayer/appearance.rs index 158327b5..9d60cb4c 100644 --- a/crates/skate-game/src/multiplayer/appearance.rs +++ b/crates/skate-game/src/multiplayer/appearance.rs @@ -3,10 +3,7 @@ use super::{ Multiplayer, appearance_transfer::{Exchange, MAX_BLOB}, }; -use bevy::{ - prelude::*, - tasks::{AsyncComputeTaskPool, Task, block_on, poll_once}, -}; +use bevy::prelude::*; use serde::{Deserialize, Serialize}; use serde_json::Value; use std::{ @@ -33,32 +30,6 @@ pub(super) struct Appearances { cached: BTreeMap<[u8; 32], Look>, pub status: String, pub progress: String, - indexed: Vec, - indexing: Option>>, -} -fn index_imports( - imports: Vec<(String, PathBuf)>, -) -> Vec<(super::appearance_transfer::Identity, Look)> { - imports - .into_iter() - .filter_map(|(asset, path)| { - if std::fs::metadata(&path).ok()?.len() as usize >= MAX_BLOB { - return None; - } - let bytes = std::fs::read(path).ok()?; - validate_glb(&bytes).ok()?; - let mut hash = blake3::Hasher::new(); - hash.update(&[0]); - hash.update(&bytes); - Some(( - super::appearance_transfer::Identity { - hash: *hash.finalize().as_bytes(), - size: bytes.len() + 1, - }, - Look::Imported(asset), - )) - }) - .collect() } pub(super) fn cleanup(mut exit: MessageReader) { // Generated process-private directory, never the personal import library. @@ -90,24 +61,6 @@ pub(super) fn sync( ..default() }; } - let imports = models.online_imports(); - let keys: Vec<_> = imports.iter().map(|(key, _)| key.clone()).collect(); - if state.indexed != keys { - state.indexed = keys; - state.indexing = - Some(AsyncComputeTaskPool::get().spawn(async move { index_imports(imports) })); - } - if let Some(found) = state - .indexing - .as_mut() - .and_then(|task| block_on(poll_once(task))) - { - for (id, look) in found { - state.cached.insert(id.hash, look); - state.exchange.remember(id); - } - state.indexing = None; - } let selection = models.online_selection(); let signature = models .active @@ -160,9 +113,7 @@ pub(super) fn sync( } let now = net.started.elapsed().as_millis() as u64; let lobby = net.lobby.as_mut().unwrap(); - if state.indexing.is_none() { - state.exchange.tick(lobby, now); - } + state.exchange.tick(lobby, now); state.progress = state.exchange.progress(); state.looks.retain(|id, _| lobby.actors.contains_key(id)); state.seen.retain(|id, _| lobby.actors.contains_key(id)); @@ -413,64 +364,3 @@ mod tests { validate_glb(&std::fs::read(path).unwrap()).unwrap(); } } -#[cfg(test)] -mod local_reuse_tests { - use super::*; - #[test] - #[ignore = "requires SKATE_ONLINE_TEST_GLB"] - fn online_appearance_existing_import_needs_no_model_upload() { - let path = PathBuf::from(std::env::var("SKATE_ONLINE_TEST_GLB").unwrap()); - let found = index_imports(vec![( - "characters://owned/character.glb".into(), - path.clone(), - )]); - assert_eq!(found.len(), 1); - let (identity, look) = found[0].clone(); - assert!(matches!(look, Look::Imported(_))); - let mut bytes = vec![0]; - bytes.extend(std::fs::read(path).unwrap()); - assert_eq!(identity.hash, *blake3::hash(&bytes).as_bytes()); - let mut sessions: Vec<_> = (0..2) - .map(|i| { - skate_net::lobby::Session::new( - 3, - skate_net::lobby::Info { - id: 10 + i, - map: 1, - rig: 1, - physics: 1, - appearance: 1, - }, - if i == 0 { None } else { Some(1) }, - ) - }) - .collect(); - let mut transfer = [Exchange::default(), Exchange::default()]; - transfer[0].publish(bytes); - transfer[1].publish(vec![1]); - transfer[1].remember(identity.clone()); - for step in 0..200 { - let now = step * 10; - let mut wire = vec![]; - for i in 0..2 { - transfer[i].tick(&mut sessions[i], now); - for packet in sessions[i].service(now) { - assert!( - packet.data.len() < 200, - "local content must not send model chunks" - ); - wire.push((i + 1, packet)); - } - } - for (from, p) in wire { - sessions[p.peer as usize - 1].receive(from as u64, &p.data, now); - } - } - let (received, payload) = transfer[1] - .ready - .get(&10) - .expect("local appearance must be ready after handshake"); - assert_eq!(received, &identity); - assert!(payload.is_empty()); - } -} diff --git a/crates/skate-game/src/multiplayer/appearance_transfer.rs b/crates/skate-game/src/multiplayer/appearance_transfer.rs index e64f0b00..c678a62e 100644 --- a/crates/skate-game/src/multiplayer/appearance_transfer.rs +++ b/crates/skate-game/src/multiplayer/appearance_transfer.rs @@ -1,222 +1,191 @@ -//! Bounded appearance transfer over the lobby's authenticated, budgeted record stream. -//! Fixed keys keep late joins/retries reliable without growing lobby state per file. +//! Appearance content uses the same authenticated bulk stream on every transport. +pub use skate_net::blob::{Identity, MAX_BLOB}; use skate_net::lobby::Session; -use std::collections::BTreeMap; -const META: &str = "@look/meta"; -const CHUNK: usize = 980; -const WINDOW: usize = 32; -pub const MAX_BLOB: usize = 64 * 1024 * 1024; -const SESSION_BUDGET: usize = 256 * 1024 * 1024; -#[derive(Clone, Debug, PartialEq, Eq)] -pub struct Identity { - pub hash: [u8; 32], - pub size: usize, -} -impl Identity { - fn read(b: &[u8]) -> Option { - if b.len() != 36 { - return None; - } - let size = u32::from_le_bytes(b[32..].try_into().ok()?) as usize; - (size > 0 && size <= MAX_BLOB).then(|| Self { - hash: b[..32].try_into().unwrap(), - size, - }) - } - fn bytes(&self) -> Vec { - let mut b = self.hash.to_vec(); - b.extend((self.size as u32).to_le_bytes()); - b - } -} -struct Incoming { - id: Identity, - bytes: Vec, - received: Vec, -} +use std::collections::{BTreeMap, BTreeSet}; #[derive(Default)] pub struct Exchange { - local: Option<(Identity, Vec)>, - incoming: BTreeMap, + pending: Option>, pub ready: BTreeMap)>, - spent: usize, - known: BTreeMap<[u8; 32], Identity>, - round: usize, - next: u64, - serving: Option<([u8; 32], usize, u64)>, + known: BTreeSet<[u8; 32]>, + progress: (usize, usize), } impl Exchange { pub fn remember(&mut self, id: Identity) { - self.known.insert(id.hash, id); - } - pub fn progress(&self) -> String { - let total: usize = self.incoming.values().map(|p| p.received.len()).sum(); - let done: usize = self - .incoming - .values() - .map(|p| p.received.iter().filter(|v| **v).count()) - .sum(); - if total == 0 { - format!("Characters: {} synced", self.ready.len()) - } else { - format!( - "Characters: {} synced | downloading {} ({}%)", - self.ready.len(), - self.incoming.len(), - done * 100 / total - ) - } + self.known.insert(id.hash); } pub fn publish(&mut self, bytes: Vec) -> bool { if bytes.is_empty() || bytes.len() > MAX_BLOB { return false; } - let id = Identity { - hash: *blake3::hash(&bytes).as_bytes(), - size: bytes.len(), - }; - self.local = Some((id, bytes)); + self.pending = Some(bytes); true } - pub fn tick(&mut self, session: &mut Session, now: u64) { - if now < self.next { - return; - } - self.next = now + 25; - self.incoming - .retain(|id, _| session.actors.contains_key(id)); - self.ready.retain(|id, _| session.actors.contains_key(id)); - let Some((local, bytes)) = &self.local else { - return; - }; - session.publish_application(META, local.bytes(), now); - let mut records = vec![]; - let mut requests = vec![]; - for (slot, (&actor, a)) in session.actors.iter().enumerate() { - if actor == session.local { - continue; - } - for r in a - .application - .iter() - .filter(|(key, _)| key.starts_with("@look/r/")) - .map(|(_, r)| r) - { - if r.value.len() == 44 - && u64::from_le_bytes(r.value[..8].try_into().unwrap()) == session.local - && r.value[8..40] == local.hash - { - let page = u32::from_le_bytes(r.value[40..].try_into().unwrap()) as usize; - if page < bytes.len().div_ceil(CHUNK * WINDOW) { - requests.push(page); - } - } - } - let Some(id) = a - .application - .get(META) - .and_then(|r| Identity::read(&r.value)) - else { - continue; - }; - if self.ready.get(&actor).is_some_and(|(old, _)| old == &id) { - self.incoming.remove(&actor); - continue; - } - if self.known.get(&id.hash) == Some(&id) { - self.incoming.remove(&actor); - self.ready.insert(actor, (id, vec![])); - continue; - } - if self.incoming.get(&actor).is_none_or(|p| p.id != id) { - if self.spent.saturating_add(id.size) > SESSION_BUDGET { - continue; - } - self.spent += id.size; - self.incoming.insert( - actor, - Incoming { - received: vec![false; id.size.div_ceil(CHUNK)], - bytes: vec![0; id.size], - id, - }, - ); - } - let p = self.incoming.get_mut(&actor).unwrap(); - for i in 0..WINDOW { - let Some(r) = a.application.get(&format!("@look/c/{i}")) else { - continue; - }; - let b = &r.value; - if b.len() < 37 || b[..32] != p.id.hash { - continue; - } - let offset = u32::from_le_bytes(b[32..36].try_into().unwrap()) as usize; - if offset % CHUNK != 0 - || offset >= p.bytes.len() - || b.len() - 36 != CHUNK.min(p.bytes.len() - offset) - { - continue; - } - p.bytes[offset..offset + b.len() - 36].copy_from_slice(&b[36..]); - p.received[offset / CHUNK] = true; - } - if let Some(missing) = p.received.iter().position(|done| !*done) { - let mut request = actor.to_le_bytes().to_vec(); - request.extend(p.id.hash); - request.extend(((missing / WINDOW) as u32).to_le_bytes()); - records.push((format!("@look/r/{slot}"), request)); - } else { - let p = self.incoming.remove(&actor).unwrap(); - if blake3::hash(&p.bytes).as_bytes() == &p.id.hash { - self.ready.insert(actor, (p.id, p.bytes)); - } - // Clear the request after completion, including corrupted transfers. - records.push((format!("@look/r/{slot}"), vec![])); - } - } - // Requests occupy stable bounded slots, but actor ordering can change. - // Explicit tombstones stop transfers after completion/cache hits/departures. - for slot in 0..skate_net::lobby::MAX_PLAYERS { - let key = format!("@look/r/{slot}"); - if !records.iter().any(|(k, _)| k == &key) { - records.push((key, vec![])); - } - } - if requests.is_empty() { - for slot in 0..WINDOW { - records.push((format!("@look/c/{slot}"), vec![])); - } - } - if !requests.is_empty() { - requests.sort_unstable(); - requests.dedup(); - let page = if let Some((hash, page, at)) = self.serving.filter(|(hash, page, at)| { - *hash == local.hash && requests.contains(page) && now.saturating_sub(*at) < 150 - }) { - let _ = (hash, at); - page - } else { - let page = requests[self.round % requests.len()]; - self.round = self.round.wrapping_add(1); - self.serving = Some((local.hash, page, now)); - page - }; - for slot in 0..WINDOW { - let offset = (page * WINDOW + slot) * CHUNK; - if offset >= bytes.len() { - break; - } - let mut b = local.hash.to_vec(); - b.extend((offset as u32).to_le_bytes()); - b.extend(&bytes[offset..(offset + CHUNK).min(bytes.len())]); - records.push((format!("@look/c/{slot}"), b)); - } - } - for (key, value) in records { - session.publish_application(&key, value, now); + pub fn progress(&self) -> String { + let (done, total) = self.progress; + if total == 0 { + format!("Characters: {} synced", self.ready.len()) + } else { + format!( + "Characters: {} synced | downloading ({}%)", + self.ready.len(), + done * 100 / total + ) } } + pub fn tick(&mut self, session: &mut Session, _now: u64) { + if let Some(bytes) = self.pending.take() { + session.blobs.publish(session.local, bytes); + } + self.ready.retain(|id, _| session.actors.contains_key(id)); + for &actor in session.actors.keys().filter(|id| **id != session.local) { + let Some((id, bytes)) = session.blobs.ready(actor) else { + continue; + }; + if self.ready.get(&actor).is_some_and(|(old, _)| old == id) { + continue; + } + let bytes = if self.known.contains(&id.hash) { + vec![] + } else { + bytes.to_vec() + }; + self.ready.insert(actor, (id.clone(), bytes)); + } + self.progress = session.blobs.progress(session.local); + } +} +/// Exercise actual sockets, host forwarding and a receiver with an empty cache. +/// Also used by the owned-asset test before loading the received GLB in Bevy. +#[cfg(test)] +pub(super) fn transfer_over_udp(model: Vec) -> Vec { + use std::{ + net::UdpSocket, + time::{Duration, Instant}, + }; + let sockets: Vec<_> = (0..3) + .map(|_| { + let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); + skate_net::socket::configure(&socket).unwrap(); + socket + }) + .collect(); + let addresses: Vec<_> = sockets.iter().map(|s| s.local_addr().unwrap()).collect(); + let mut sessions: Vec<_> = (0..3) + .map(|i| { + Session::new( + 91, + skate_net::lobby::Info { + id: 10 + i, + map: 1, + rig: 1, + physics: 1, + appearance: 1, + }, + if i == 0 { None } else { Some(1) }, + ) + }) + .collect(); + let mut exchanges: Vec<_> = (0..3).map(|_| Exchange::default()).collect(); + for e in &mut exchanges { + e.publish(vec![5]); + } + exchanges[1].publish(model.clone()); + let started = Instant::now(); + let mut frame = 0; + let mut buffer = [0u8; 1500]; + let mut result = None; + let mut completed_at = None; + let mut last_payload = 0; + loop { + let now = started.elapsed().as_millis() as u64; + assert!( + now < 25000, + "actual UDP transfer stalled: {}", + exchanges[2].progress() + ); + for i in 0..3 { + for _ in 0..4096 { + match sockets[i].recv_from(&mut buffer) { + Ok((n, from)) => { + let peer = addresses.iter().position(|a| *a == from).unwrap() + 1; + sessions[i].receive(peer as u64, &buffer[..n], now); + } + Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => break, + Err(e) => panic!("UDP receive: {e}"), + } + } + let pose = skate_net::Pose { + p: [frame as f32 * 0.001, 0., 0.], + q: [0., 0., 0., 1.], + }; + let body = skate_net::Body { + pose, + velocity: [0.; 3], + angular: [0.; 3], + }; + sessions[i].publish( + skate_net::packed::BODY, + skate_net::packed::Packed::body(&skate_net::packed::BodyState { + root: pose, + enabled: (1u64 << 33) - 1, + bodies: vec![body; 33], + }) + .unwrap(), + now, + ); + exchanges[i].tick(&mut sessions[i], now); + for p in sessions[i].service(now) { + if p.data[24] == skate_net::blob::DATA { + last_payload = now; + } + sockets[i] + .send_to(&p.data, addresses[p.peer as usize - 1]) + .unwrap(); + } + } + if result.is_none() + && exchanges[2] + .ready + .get(&11) + .is_some_and(|(_, b)| b == &model) + { + eprintln!( + "ONLINE_UDP_COMPLETE bytes={} elapsed_ms={now} fps=15 route=client-host-client", + model.len() + ); + assert!( + now < 15000, + "32 MB should not take minutes even at 15 FPS: {now} ms" + ); + result = Some(std::mem::take( + &mut exchanges[2].ready.get_mut(&11).unwrap().1, + )); + completed_at = Some(now); + } + if completed_at.is_some_and(|at| now - at > 1500) { + assert!( + sessions[2].actors[&11] + .body + .latest() + .is_some_and(|b| now - b.received < 300), + "model upload starved the remote player's movement" + ); + assert!( + last_payload < completed_at.unwrap() + 500, + "completed upload keeps sending data" + ); + return result.unwrap(); + } + frame += 1; + if let Some(wait) = Duration::from_millis(frame * 67).checked_sub(started.elapsed()) { + std::thread::sleep(wait); + } + } +} +#[cfg(test)] +#[test] +fn online_appearance_real_udp_large_model_at_low_frame_rate() { + transfer_over_udp((0..32_067_793).map(|i| (i * 17) as u8).collect()); } #[cfg(test)] mod tests { @@ -251,6 +220,7 @@ mod tests { assert!(e.publish(b.clone())); } let mut completed = false; + let mut delayed = vec![]; for step in 0..16000 { let now = step * 25; let active = if step < 200 { 3 } else { 10 }; @@ -289,6 +259,13 @@ mod tests { } wire.reverse(); for (from, p) in wire { + let due = now + 50 + ((from as u64 * 17 + step * 13) % 70); + delayed.push((due, from, p)); + } + let (arrived, pending): (Vec<_>, Vec<_>) = + delayed.into_iter().partition(|(due, _, _)| *due <= now); + delayed = pending; + for (_, from, p) in arrived { let to = p.peer as usize - 1; if to < active { sessions[to].receive(from as u64, &p.data, now); @@ -330,12 +307,11 @@ mod tests { } #[test] fn online_appearance_rejects_unbounded_metadata_and_chunks() { - assert!(Identity::read(&[]).is_none()); let id = Identity { hash: [1; 32], size: MAX_BLOB + 1, }; - assert!(Identity::read(&id.bytes()).is_none()); + assert!(id.size > MAX_BLOB); let mut e = Exchange::default(); assert!(!e.publish(vec![])); } @@ -360,9 +336,7 @@ mod delivery_tests { ) }) .collect(); - for s in &mut sessions { - s.set_loopback(true); - } + let mut exchanges = [Exchange::default(), Exchange::default()]; let model: Vec = (0..32_067_793).map(|i| (i * 17) as u8).collect(); exchanges[0].publish(model.clone()); @@ -372,13 +346,12 @@ mod delivery_tests { let mut last_model_packet = 0; let mut reused = false; for step in 0..18000 { - let now = step * 10; + let now = step * 67; let mut wire = vec![]; for i in 0..2 { exchanges[i].tick(&mut sessions[i], now); for p in sessions[i].service(now) { - if p.data.get(24) == Some(&skate_net::lobby::APPLICATION) && p.data.len() > 100 - { + if p.data.get(24) == Some(&skate_net::blob::DATA) && p.data.len() > 100 { last_model_packet = now; assert!(phase < 3, "completed model was retransmitted"); } @@ -396,7 +369,10 @@ mod delivery_tests { .get(&10) .is_some_and(|(_, b)| b == &model) { - assert!(now < 90000, "local model transfer is too slow: {now} ms"); + assert!( + now < 15000, + "network model transfer at 15 FPS is too slow: {now} ms" + ); let identity = exchanges[1].ready[&10].0.clone(); exchanges[1].remember(identity); exchanges[1].ready.get_mut(&10).unwrap().1.clear(); diff --git a/crates/skate-game/src/multiplayer/render.rs b/crates/skate-game/src/multiplayer/render.rs index a98db69b..16cb074a 100644 --- a/crates/skate-game/src/multiplayer/render.rs +++ b/crates/skate-game/src/multiplayer/render.rs @@ -373,6 +373,8 @@ fn bind( commands.entity(old).despawn(); } commands.entity(p.root).insert(Visibility::Inherited); + info!("ONLINE_CHARACTER_VISIBLE peer={id} joints={} kind={}", bindings.len(), + if matches!(p.look, Look::Imported(_)) { "import" } else { "retail" }); skin.bindings = bindings; skin.poses = Buffer::default(); skin.pending = None; @@ -574,7 +576,12 @@ mod online_owned_tests { super::super::appearance::validate_glb(&bytes).unwrap(); let cache = super::super::appearance::cache_directory(); std::fs::create_dir_all(cache).unwrap(); - std::fs::write(cache.join("received-test.glb"), bytes).unwrap(); + let mut payload = vec![0]; + payload.extend(bytes); + let received = super::super::appearance_transfer::transfer_over_udp(payload); + assert_eq!(received[0], 0); + super::super::appearance::validate_glb(&received[1..]).unwrap(); + std::fs::write(cache.join("received-test.glb"), &received[1..]).unwrap(); let handle = app .world() .resource::() @@ -604,14 +611,24 @@ mod online_owned_tests { Query<&ChildOf>, )> = SystemState::new(app.world_mut()); let (meshes, nodes, parents) = queries.get(app.world()); - assert!( - !crate::animation::AnimationStatus::for_scene( - root, &names, &meshes, &nodes, &parents - ) - .unwrap() - .online_bindings() - .is_empty() - ); + let animation = crate::animation::AnimationStatus::for_scene( + root, &names, &meshes, &nodes, &parents + ).unwrap(); + let joints=animation.online_bindings(); + assert!(!joints.is_empty()); + // A received model must accept poses before display, then keep moving. + let first:Vec<_>=(0..names.len()).map(|i|Mat4::from_translation(Vec3::new(i as f32*0.01,1.,0.))).collect(); + for (joint,pose) in animation.pose_transforms(&first) { + assert!(pose.to_matrix().is_finite()); + *app.world_mut().get_mut::(joint).unwrap()=pose; + } + *app.world_mut().get_mut::(root).unwrap()=Visibility::Inherited; + let second:Vec<_>=first.iter().enumerate().map(|(i,m)|*m*Mat4::from_rotation_z(0.01*(i+1) as f32)).collect(); + for (joint,pose) in animation.pose_transforms(&second) { + assert!(pose.to_matrix().is_finite()); + assert_ne!(*app.world().get::(joint).unwrap(),pose); + *app.world_mut().get_mut::(joint).unwrap()=pose; + } app.world_mut().entity_mut(root).despawn(); app.update(); std::fs::remove_file(cache.join("received-test.glb")).unwrap(); diff --git a/crates/skate-game/src/multiplayer/transport.rs b/crates/skate-game/src/multiplayer/transport.rs index a18f3f92..e1a3d5ea 100644 --- a/crates/skate-game/src/multiplayer/transport.rs +++ b/crates/skate-game/src/multiplayer/transport.rs @@ -40,7 +40,7 @@ impl Direct { pub fn new(bind: SocketAddr) -> io::Result { endpoint(bind)?; let socket = UdpSocket::bind(bind)?; - socket.set_nonblocking(true)?; + skate_net::socket::configure(&socket)?; Ok(Self { socket }) } } @@ -56,7 +56,7 @@ impl Transport for Direct { fn receive(&mut self) -> io::Result)>> { let mut output = Vec::new(); let mut buffer = [0; 1500]; - for _ in 0..256 { + for _ in 0..4096 { match self.socket.recv_from(&mut buffer) { Ok((n, from)) if n <= skate_net::packed::MTU => { output.push((endpoint(from)?, buffer[..n].to_vec())) @@ -99,7 +99,7 @@ pub(super) struct Steam { impl Steam { pub fn new(peer: u64, session: u64) -> Result { let socket = UdpSocket::bind("127.0.0.1:0").map_err(|e| e.to_string())?; - socket.set_nonblocking(true).map_err(|e| e.to_string())?; + skate_net::socket::configure(&socket).map_err(|e| e.to_string())?; let dir = std::env::current_exe() .map_err(|e| e.to_string())? .parent() @@ -200,7 +200,7 @@ impl Transport for Steam { let mut packets = vec![]; let mut buffer = [0; 1500]; let prefix = format!("SK8RELAY {} ", self.cookie); - for _ in 0..256 { + for _ in 0..4096 { match self.socket.recv_from(&mut buffer) { Ok((n, from)) => { if !from.ip().is_loopback() || self.peer.is_some_and(|p| p != from) { diff --git a/crates/skate-net/Cargo.toml b/crates/skate-net/Cargo.toml index 3ea6c094..03a989f4 100644 --- a/crates/skate-net/Cargo.toml +++ b/crates/skate-net/Cargo.toml @@ -7,7 +7,9 @@ publish.workspace = true [dependencies] serde.workspace = true serde_json.workspace = true +blake3 = "=1.8.7" half = "=2.7.1" +socket2 = "0.6" [dev-dependencies] skate-core.workspace = true diff --git a/crates/skate-net/src/blob.rs b/crates/skate-net/src/blob.rs new file mode 100644 index 00000000..71a2fda1 --- /dev/null +++ b/crates/skate-net/src/blob.rs @@ -0,0 +1,455 @@ +//! Reliable, bounded appearance streams. Movement is serviced first by the lobby. +//! Selective acknowledgements slide continuously; no page-sized round trips. +use crate::{lobby::Outgoing, packed}; +use std::{ + collections::{BTreeMap, BTreeSet}, + sync::Arc, +}; + +pub const META: u8 = 12; +pub const DATA: u8 = 13; +pub const ACK: u8 = 14; +pub const MAX_BLOB: usize = 64 * 1024 * 1024; +const MEMORY: usize = (crate::lobby::MAX_PLAYERS + 2) * MAX_BLOB; +const CHUNK: usize = 1024; +const WINDOW: usize = 1024; +const MASK: usize = WINDOW / 8; + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct Identity { + pub hash: [u8; 32], + pub size: usize, +} + +#[cfg(test)] +mod tests { + use super::*; + fn manifest(bytes: &[u8]) -> Vec { + let mut b = blake3::hash(bytes).as_bytes().to_vec(); + b.extend((bytes.len() as u32).to_le_bytes()); + b + } + #[test] + fn blob_bounds_hash_validation_and_obsolete_chunks() { + let mut b = Blobs::default(); + let mut oversized = vec![0; 32]; + oversized.extend(((MAX_BLOB + 1) as u32).to_le_bytes()); + b.receive(1, 10, META, 1, &oversized); + assert!(b.cache.is_empty()); + b.receive(1, 10, META, 1, &manifest(&[1, 2])); + let mut wrong = 0u32.to_le_bytes().to_vec(); + wrong.extend([3, 4]); + b.receive(1, 10, DATA, 1, &wrong); + assert!( + b.ready(10).is_none(), + "corrupt content must never be loaded" + ); + b.receive(1, 10, META, 2, &manifest(&[5, 6])); + b.receive(1, 10, DATA, 1, &wrong); + b.receive(1, 10, META, 1, &manifest(&[1, 2])); + assert_eq!(b.actors[&10].0, 2); + let mut valid = 0u32.to_le_bytes().to_vec(); + valid.extend([5, 6]); + b.receive(1, 10, DATA, 2, &valid); + assert_eq!(b.ready(10).unwrap().1, &[5, 6]); + // A later peer can consume verified cached bytes without data retransmission. + b.receive(2, 11, META, 1, &manifest(&[5, 6])); + assert_eq!(b.ready(11).unwrap().1, &[5, 6]); + } + #[test] + fn lobby_rejects_forged_blob_origins_and_unregistered_senders() { + use crate::lobby::{Info, Session}; + let info = |id| Info { + id, + map: 1, + rig: 1, + physics: 1, + appearance: 1, + }; + let mut host = Session::new(88, info(10), None); + let mut client = Session::new(88, info(11), Some(1)); + for p in client.service(1) { + host.receive(2, &p.data, 1); + } + let mut forged = packed::header(88, 12, META, 1); + forged.extend(manifest(&[9])); + host.receive(2, &forged, 2); + assert!(host.blobs.actors.is_empty()); + let mut valid = packed::header(88, 11, META, 1); + valid.extend(manifest(&[9])); + host.receive(3, &valid, 3); + assert!(host.blobs.actors.is_empty()); + host.receive(2, &valid, 4); + assert!(host.blobs.actors.contains_key(&11)); + } +} +struct Blob { + id: Identity, + bytes: Arc>, + received: Vec, + prefix: usize, + complete: bool, + failed: bool, +} +impl Blob { + fn count(&self) -> usize { + self.id.size.div_ceil(CHUNK) + } +} +struct Sending { + seq: u32, + meta_at: Option, + accepted: bool, + prefix: usize, + mask: [u8; MASK], + sent: BTreeMap, + window: usize, + last_loss: u64, +} +impl Sending { + fn new(seq: u32) -> Self { + Self { + seq, + meta_at: None, + accepted: false, + prefix: 0, + mask: [0; MASK], + sent: BTreeMap::new(), + window: 64, + last_loss: 0, + } + } + fn has(&self, i: usize) -> bool { + i < self.prefix + || (i - self.prefix < WINDOW + && self.mask[(i - self.prefix) / 8] & (1 << ((i - self.prefix) % 8)) != 0) + } +} +#[derive(Default)] +pub struct Blobs { + actors: BTreeMap, + cache: BTreeMap<[u8; 32], Blob>, + sending: BTreeMap<(u64, u64), Sending>, + acknowledgements: BTreeSet<(u64, u64)>, + credits: BTreeMap, + last: u64, + round: usize, + pub congested: bool, +} +impl Blobs { + fn reserve(&mut self, size: usize) -> bool { + let mut used: usize = self.cache.values().map(|b| b.id.size).sum(); + let unused: Vec<_> = self + .cache + .keys() + .filter(|hash| !self.actors.values().any(|(_, h)| h == *hash)) + .copied() + .collect(); + for hash in unused { + if used + size <= MEMORY { + break; + } + used -= self.cache.remove(&hash).unwrap().id.size; + } + used + size <= MEMORY + } + pub fn publish(&mut self, actor: u64, bytes: Vec) -> bool { + if bytes.is_empty() || bytes.len() > MAX_BLOB { + return false; + } + let hash = *blake3::hash(&bytes).as_bytes(); + if self.actors.get(&actor).is_some_and(|(_, h)| *h == hash) { + return true; + } + if self.cache.get(&hash).is_none_or(|b| !b.complete) { + if !self.cache.contains_key(&hash) && !self.reserve(bytes.len()) { + return false; + } + let count = bytes.len().div_ceil(CHUNK); + self.cache.insert( + hash, + Blob { + id: Identity { + hash, + size: bytes.len(), + }, + bytes: Arc::new(bytes), + received: vec![], + prefix: count, + complete: true, + failed: false, + }, + ); + } + let seq = self + .actors + .get(&actor) + .map_or(1, |(s, _)| s.saturating_add(1)); + self.actors.insert(actor, (seq, hash)); + true + } + pub fn ready(&self, actor: u64) -> Option<(&Identity, &[u8])> { + let (_, hash) = self.actors.get(&actor)?; + let b = self.cache.get(hash)?; + b.complete.then_some((&b.id, b.bytes.as_slice())) + } + pub fn progress(&self, local: u64) -> (usize, usize) { + self.actors + .iter() + .filter(|(id, _)| **id != local) + .filter_map(|(_, (_, h))| self.cache.get(h)) + .filter(|b| !b.complete) + .fold((0, 0), |(done, total), b| { + ( + done + b.received.iter().filter(|v| **v).count(), + total + b.count(), + ) + }) + } + // The lobby authenticates origin/peer and bounds membership before calling this. + pub(crate) fn receive(&mut self, peer: u64, actor: u64, kind: u8, seq: u32, data: &[u8]) { + if kind == META { + if data.len() != 36 || seq == 0 { + return; + } + let hash: [u8; 32] = data[..32].try_into().unwrap(); + let size = u32::from_le_bytes(data[32..].try_into().unwrap()) as usize; + if size == 0 || size > MAX_BLOB { + return; + } + if self + .actors + .get(&actor) + .is_some_and(|(old, h)| seq < *old || (seq == *old && hash != *h)) + { + return; + } + if let Some(b) = self.cache.get(&hash) { + if b.id.size != size { + return; + } + } else { + if !self.reserve(size) { + return; + } + self.cache.insert( + hash, + Blob { + id: Identity { hash, size }, + bytes: Arc::new(vec![0; size]), + received: vec![false; size.div_ceil(CHUNK)], + prefix: 0, + complete: false, + failed: false, + }, + ); + } + self.actors.insert(actor, (seq, hash)); + self.acknowledgements.insert((peer, actor)); + } else if kind == DATA { + let Some((current, hash)) = self.actors.get(&actor) else { + return; + }; + if seq != *current || data.len() < 5 { + return; + } + let b = self.cache.get_mut(hash).unwrap(); + let index = u32::from_le_bytes(data[..4].try_into().unwrap()) as usize; + if index >= b.count() || data.len() - 4 != CHUNK.min(b.id.size - index * CHUNK) { + return; + } + self.acknowledgements.insert((peer, actor)); + if b.complete || b.failed || b.received[index] { + return; + } + Arc::get_mut(&mut b.bytes).unwrap()[index * CHUNK..index * CHUNK + data.len() - 4] + .copy_from_slice(&data[4..]); + b.received[index] = true; + while b.prefix < b.count() && b.received[b.prefix] { + b.prefix += 1; + } + if b.prefix == b.count() { + b.complete = blake3::hash(&b.bytes).as_bytes() == &b.id.hash; + b.failed = !b.complete; + // Bad bytes never reach the asset loader or trigger an endless retry. + } + } else if kind == ACK { + if data.len() != 12 + MASK { + return; + } + let origin = u64::from_le_bytes(data[..8].try_into().unwrap()); + let prefix = u32::from_le_bytes(data[8..12].try_into().unwrap()) as usize; + let Some((current, hash)) = self.actors.get(&origin) else { + return; + }; + let count = self.cache[hash].count(); + let Some(tx) = self.sending.get_mut(&(peer, origin)) else { + return; + }; + if seq != *current || seq != tx.seq || prefix > count || prefix < tx.prefix { + return; + } + if prefix == tx.prefix { + for (a, b) in tx.mask.iter_mut().zip(&data[12..]) { + *a |= *b; + } + } else { + tx.mask.copy_from_slice(&data[12..]); + tx.prefix = prefix; + } + tx.accepted = true; + let remove: Vec<_> = tx.sent.keys().filter(|i| tx.has(**i)).copied().collect(); + tx.window = (tx.window + remove.len()).min(WINDOW); + for i in remove { + tx.sent.remove(&i); + } + } + } + pub(crate) fn service( + &mut self, + session: u64, + local: u64, + host: bool, + members: &BTreeSet, + peers: &[(u64, u64, u64)], + now: u64, + ) -> Vec { + self.actors.retain(|id, _| members.contains(id)); + self.sending.retain(|(peer, id), _| { + members.contains(id) && peers.iter().any(|(p, _, _)| p == peer) + }); + self.credits + .retain(|peer, _| peers.iter().any(|(p, _, _)| p == peer)); + let mut out = vec![]; + for (peer, origin) in std::mem::take(&mut self.acknowledgements) { + if !peers.iter().any(|(p, _, _)| *p == peer) { + continue; + } + let Some((seq, hash)) = self.actors.get(&origin) else { + continue; + }; + let b = &self.cache[hash]; + let mut data = packed::header(session, local, ACK, *seq); + data.extend(origin.to_le_bytes()); + data.extend((b.prefix as u32).to_le_bytes()); + let mut mask = [0u8; MASK]; + for i in b.prefix..(b.prefix + WINDOW).min(b.count()) { + if b.complete || b.received[i] { + mask[(i - b.prefix) / 8] |= 1 << ((i - b.prefix) % 8); + } + } + data.extend(mask); + out.push(Outgoing { peer, data }); + } + let dt = now.saturating_sub(self.last).min(1000) as f64 / 1000.; + self.last = now; + // Same rates for direct and Steam. Bursts cover slow render frames; ACKs + // bound outstanding data and relay congestion reduces bulk throughput. + let rate = if self.congested { 128_000. } else { 4_000_000. }; + let mut global = 2_000_000usize; + let mut peers = peers.to_vec(); + if !peers.is_empty() { + let n = peers.len(); + peers.rotate_left(self.round % n); + } + self.round = self.round.wrapping_add(1); + 'peers: for (peer, recipient, rtt) in peers { + if recipient == 0 { + continue; + } + let credit = self.credits.entry(peer).or_default(); + *credit = (*credit + dt * rate).min(512_000.); + let mut origins: Vec<_> = self + .actors + .iter() + .filter(|(id, _)| **id != recipient && (host || **id == local)) + .map(|(&id, _)| id) + .collect(); + if !origins.is_empty() { + let n = origins.len(); + origins.rotate_left(self.round % n); + } + for origin in &origins { + let (seq, hash) = self.actors[origin]; + let b = &self.cache[&hash]; + let tx = self + .sending + .entry((peer, *origin)) + .or_insert_with(|| Sending::new(seq)); + if tx.seq != seq { + *tx = Sending::new(seq); + } + if !tx.accepted && tx.meta_at.is_none_or(|at| now.saturating_sub(at) >= 500) { + let mut data = packed::header(session, *origin, META, seq); + data.extend(hash); + data.extend((b.id.size as u32).to_le_bytes()); + out.push(Outgoing { peer, data }); + tx.meta_at = Some(now); + } + } + // Round robin chunks across actors, with a sliding selective-ACK window. + let mut candidates = vec![]; + for origin in origins { + let (_, hash) = self.actors[&origin]; + let b = &self.cache[&hash]; + let tx = self.sending.get_mut(&(peer, origin)).unwrap(); + if !tx.accepted || b.failed { + continue; + } + let retry = (rtt * 3 + 200).clamp(400, 2000); + if now.saturating_sub(tx.last_loss) >= retry + && tx.sent.values().any(|at| now.saturating_sub(*at) >= retry) + { + tx.window = (tx.window / 2).max(32); + tx.last_loss = now; + } + let chunks: Vec<_> = (tx.prefix..(tx.prefix + tx.window).min(b.count())) + .filter(|&i| { + !tx.has(i) + && (b.complete || b.received[i]) + && tx + .sent + .get(&i) + .is_none_or(|at| now.saturating_sub(*at) >= retry) + }) + .collect(); + candidates.push((origin, chunks.into_iter())); + } + loop { + let mut sent = false; + for (origin, indices) in &mut candidates { + let Some(i) = indices.next() else { + continue; + }; + let (seq, hash) = self.actors[origin]; + let b = &self.cache[&hash]; + let start = i * CHUNK; + let end = (start + CHUNK).min(b.id.size); + let size = packed::HEADER + 4 + end - start; + if global < size { + return out; + } + if *credit < size as f64 { + continue 'peers; + } + let mut data = packed::header(session, *origin, DATA, seq); + data.extend((i as u32).to_le_bytes()); + data.extend(&b.bytes[start..end]); + self.sending + .get_mut(&(peer, *origin)) + .unwrap() + .sent + .insert(i, now); + *credit -= size as f64; + global -= size; + out.push(Outgoing { peer, data }); + sent = true; + } + if !sent { + break; + } + } + } + out + } +} diff --git a/crates/skate-net/src/lib.rs b/crates/skate-net/src/lib.rs index 84051ffd..761ce2f4 100644 --- a/crates/skate-net/src/lib.rs +++ b/crates/skate-net/src/lib.rs @@ -3,6 +3,8 @@ use serde::{Deserialize, Serialize}; use std::time::{Duration, Instant}; mod codec; pub mod lobby; +pub mod blob; +pub mod socket; pub mod directory; pub mod packed; pub mod interpolation; diff --git a/crates/skate-net/src/lobby.rs b/crates/skate-net/src/lobby.rs index 9d574edd..91112f2e 100644 --- a/crates/skate-net/src/lobby.rs +++ b/crates/skate-net/src/lobby.rs @@ -158,6 +158,7 @@ pub struct Session { link_budget: f64, round: usize, loopback: bool, + pub blobs: crate::blob::Blobs, } impl Session { pub fn new(session: u64, info: Info, host: Option) -> Self { @@ -190,6 +191,7 @@ impl Session { link_budget: DEFAULT_LINK_BUDGET, round: 0, loopback: false, + blobs: Default::default(), } } /// An external membership authority chooses the new endpoint. Preserve our @@ -210,7 +212,7 @@ impl Session { self.budget = if host.is_none() { DEFAULT_HOST_BUDGET } else { - if self.loopback {2_000_000.} else {DEFAULT_LINK_BUDGET} + DEFAULT_LINK_BUDGET }; self.credits = 12_000.; self.notice = if host.is_none() { @@ -221,8 +223,6 @@ impl Session { } pub fn set_loopback(&mut self, enabled: bool) { self.loopback = enabled; - self.budget=if enabled {12_000_000.} else {DEFAULT_HOST_BUDGET}; - self.set_congested(false); } pub fn is_host(&self) -> bool { self.host.is_none() @@ -231,10 +231,11 @@ impl Session { self.is_host() || self.received_roster != 0 } pub fn set_congested(&mut self, congested: bool) { + self.blobs.congested = congested; self.link_budget = if congested { - if self.loopback {1_000_000.} else {DEFAULT_LINK_BUDGET * 0.5} + DEFAULT_LINK_BUDGET * 0.5 } else { - if self.loopback {2_000_000.} else {DEFAULT_LINK_BUDGET} + DEFAULT_LINK_BUDGET }; } pub fn publish(&mut self, kind: u8, mut state: Packed, now: u64) { @@ -381,6 +382,19 @@ impl Session { .into(); return; } + if matches!(kind, crate::blob::META | crate::blob::DATA | crate::blob::ACK) { + let valid = if kind == crate::blob::ACK { + self.links[&peer].actor == actor + } else { + actor != self.local && self.actors.contains_key(&actor) + && (!self.is_host() || self.links[&peer].actor == actor) + }; + if valid { + self.blobs.receive(peer, actor, kind, seq, &data[HEADER..]); + self.links.get_mut(&peer).unwrap().seen = now; + } + return; + } if self.links[&peer].actor != actor && !matches!(kind, BODY | POSE | APPLICATION) { return; } @@ -513,9 +527,9 @@ impl Session { pub fn service(&mut self, now: u64) -> Vec { let dt = now.saturating_sub(self.last_service).min(1000) as f64 / 1000.; self.last_service = now; - self.credits = (self.credits + dt * self.budget).min(if self.loopback {128_000.} else {12_000.}); + self.credits = (self.credits + dt * self.budget).min(12_000.); for l in self.links.values_mut() { - l.credits = (l.credits + dt * self.link_budget).min(if self.loopback {64_000.} else {3600.}); + l.credits = (l.credits + dt * self.link_budget).min(3600.); } if self.is_host() { let expired: Vec<_> = self @@ -699,6 +713,9 @@ impl Session { output.push(Outgoing {peer, data}); } } + let members = self.actors.keys().copied().collect(); + let peers: Vec<_> = self.links.iter().map(|(&peer, link)| (peer, link.actor, link.rtt)).collect(); + output.extend(self.blobs.service(self.session, self.local, self.host.is_none(), &members, &peers, now)); self.stats.rtt_ms = self.links.values().map(|l| l.rtt).max().unwrap_or(0); output } diff --git a/crates/skate-net/src/socket.rs b/crates/skate-net/src/socket.rs new file mode 100644 index 00000000..729a5c92 --- /dev/null +++ b/crates/skate-net/src/socket.rs @@ -0,0 +1,8 @@ +//! Datagram queues must hold a render frame's bulk traffic, including relay IPC. +use std::{io, net::UdpSocket}; +pub fn configure(socket: &UdpSocket) -> io::Result<()> { + let socket = socket2::SockRef::from(socket); + socket.set_recv_buffer_size(4 * 1024 * 1024)?; + socket.set_send_buffer_size(4 * 1024 * 1024)?; + socket.set_nonblocking(true) +} diff --git a/crates/skate-steam-relay/src/main.rs b/crates/skate-steam-relay/src/main.rs index fbdf4451..b9743687 100644 --- a/crates/skate-steam-relay/src/main.rs +++ b/crates/skate-steam-relay/src/main.rs @@ -32,7 +32,7 @@ fn run() -> Result<(), String> { let cookie = &args[3]; let socket = UdpSocket::bind("127.0.0.1:0").map_err(|e| e.to_string())?; socket.connect(parent).map_err(|e| e.to_string())?; - socket.set_nonblocking(true).map_err(|e| e.to_string())?; + skate_net::socket::configure(&socket).map_err(|e| e.to_string())?; let status = |s: &str| { let _ = socket.send(format!("SK8RELAY {cookie} {s}").as_bytes()); }; @@ -52,6 +52,19 @@ fn run() -> Result<(), String> { return Err("Same Steam identity".into()); } client.networking_utils().init_relay_network_access(); + // SDK defaults cap model uploads at 256 KB/s. Raise only the ceiling; + // retain Steam's congestion control and the lobby's bulk pacing/backoff. + // SAFETY: Steam is initialized and owns this interface for `client`'s lifetime. + unsafe { + let utils = steamworks::sys::SteamAPI_SteamNetworkingUtils_SteamAPI_v004(); + if utils.is_null() || !steamworks::sys::SteamAPI_ISteamNetworkingUtils_SetGlobalConfigValueInt32( + utils, + steamworks::sys::ESteamNetworkingConfigValue::k_ESteamNetworkingConfig_SendRateMax, + 4_500_000, + ) { + return Err("Could not configure Steam model transfer bandwidth".into()); + } + } let messages = client.networking_messages(); let mut attempts = 0; let mut window = Instant::now(); @@ -144,7 +157,7 @@ fn run() -> Result<(), String> { )); heartbeat = Instant::now(); } - for _ in 0..256 { + for _ in 0..4096 { match socket.recv(&mut buffer) { Ok(n) => { if buffer[..n] == *ping.as_bytes() { @@ -187,7 +200,7 @@ fn run() -> Result<(), String> { } parent_seen = Instant::now(); let state = matches!(kind, skate_net::packed::BODY | skate_net::packed::POSE); - if state && peers.get(&target).is_some_and(|(_, queue)| *queue > 50) { + if (state || kind == skate_net::blob::DATA) && peers.get(&target).is_some_and(|(_, queue)| *queue > 50) { dropped += 1; continue; } @@ -213,7 +226,7 @@ fn run() -> Result<(), String> { Err(e) => return Err(e.to_string()), } } - for message in messages.receive_messages_on_channel(CHANNEL, 256) { + for message in messages.receive_messages_on_channel(CHANNEL, 4096) { let id = message .identity_peer() .steam_id() diff --git a/docs/multiplayer-test.md b/docs/multiplayer-test.md index ecb02d02..a6c3a187 100644 --- a/docs/multiplayer-test.md +++ b/docs/multiplayer-test.md @@ -10,16 +10,18 @@ prepared character library. The same setup update that supplies the local customiser supplies these online assets. Imported characters transfer their converted, self-contained GLB automatically; -other players do not need to import the model. Installed imports are indexed on a -background worker by content hash. When a matching file is already available, the -receiver loads that local asset and requests no model data, including when a -different character is currently equipped. Completed appearances are cached for -the connection: switching back sends only the identity, and acknowledged model data -stops sending. Departed receivers and completed requests are explicitly cleared. -Updates and late joins use bounded, -reliable chunks through the existing host relay. Movement/collision packets keep -priority and existing internet bandwidth budgets. Loopback test connections use -a larger local bandwidth budget. The HUD shows character transfer progress. +other players do not need to import the model. There is no installed-library scan +or local-file bypass. Direct and Steam use the same authenticated bulk stream. +The host forwards available chunks while downloading. A sliding selective-ACK +window replaces the old 32-chunk request/wait cycle. Each peer receives the current +model once; completed content is cached for later swaps, and a late join receives +the current selection. New selections cancel obsolete transfers. +Movement/collision packets are serviced first. Bulk traffic has its own bounded +4 MB/s per-link ceiling, grows its window as chunks are acknowledged, backs off +on loss/relay congestion, and retries missing +chunks. Datagram queues hold slow render frames' traffic on both direct sockets +and Steam IPC. The Steam send-rate ceiling is raised from its small-message +default. The HUD shows character transfer progress. Each remote keeps its visible character until the new scenes, textures and every clothing rig are ready. A replacement is posed before visibility is published; paused local swaps use the current skeleton and remote swaps use the latest @@ -32,7 +34,8 @@ All players need this build to see appearances; older builds continue to display their stock fallback. Missing retail content or rejected imports also retain a fallback. Online imports are limited to 64 MiB, standard embedded PNG/JPEG glTF content, 8192 pixels per texture dimension and 64 million total texture pixels. -A connection accepts at most 256 MiB of new appearance payloads. Temporary imports +The network cache is bounded to twelve maximum-sized blobs (768 MiB), with inactive +content evicted when needed; active actor content is retained for forwarding. Temporary imports are separate from the personal character library and cleaned up on leaving/exiting. Manual regression: use two PCs/accounts or local clients with separate settings. @@ -131,6 +134,25 @@ cache. Supply your prepared asset view through `-AssetRoot`. If using a separate session-marker overlay, set `SKATE3_SESSION_MARKER_OVERLAY` before launching. The merged game and Steam relay were compiled without launching either. Actual -Steam connectivity still needs two PCs with separate Steam accounts. Remote clothing -replication remains the implementation's appearance-ID/fallback mechanism; merging -the character customiser does not add transmission of its full clothing selections. +Steam connectivity still needs two PCs with separate Steam accounts. Current +appearance replication is described in the Online characters section above. + +### Network stream regression (September 9) + +The previous page exchange took approximately two minutes for the reported 32 MB +model. The installed-file shortcut has been removed. Production UDP sockets now +transfer 32,067,793 bytes through client -> host -> client in approximately 8.4 s +at a deliberately limited 15 FPS, with movement continuing and no model chunks +sent after acknowledgement. This is measured transfer time, not a claim of +instantaneous delivery or a Steam/WAN measurement. + +Nine headless character tests pass, including the real owned import transferred +through UDP before Bevy loads it from the receiver's private source, binds the +rig and applies successive poses. Other checks cover male/female clothing rigs, +material isolation, swapped rigs, invalid GLBs, and ten peers with delay/loss, +reordering, late joins and changed selections. The transport adds explicit +origin-authentication, stale-generation and corrupt-content regression tests. +No game or Steam client was launched. Manual tests should check both screens, +including skating/bailing and switching back to stock. Logs distinguish download +completion (`ONLINE_CHARACTER_READY`) from a posed scene becoming visible +(`ONLINE_CHARACTER_VISIBLE`).