From 9dc7bba3a92796b8746ad57bea69cff1c784d2e5 Mon Sep 17 00:00:00 2001 From: Serkyo Date: Wed, 22 Jul 2026 00:01:57 +0200 Subject: [PATCH] feat(server): stream chunks to clients for their subscribed region --- crates/server/src/client_stream.rs | 120 +++++++++++++++++++++++ crates/server/src/main.rs | 58 +++++++++-- crates/server/src/tests/client_stream.rs | 85 ++++++++++++++++ 3 files changed, 256 insertions(+), 7 deletions(-) create mode 100644 crates/server/src/client_stream.rs create mode 100644 crates/server/src/tests/client_stream.rs diff --git a/crates/server/src/client_stream.rs b/crates/server/src/client_stream.rs new file mode 100644 index 0000000..f4b78e8 --- /dev/null +++ b/crates/server/src/client_stream.rs @@ -0,0 +1,120 @@ +// SPDX-License-Identifier: AGPL-3.0-only + +//! Per-connection chunk-streaming state. + +use std::collections::HashSet; + +use net::ChunkSink; +use shared::protocol::chunk::ChunkMessage; +use shared::world::{Chunk, ChunkData, ChunkPos}; + +use crate::world_server::{ServerWorld, cylinder_chunks}; + +/// Upper bound, in chunks, on a client's requested load radius. A larger request is clamped to this, bounding the per-client resident set and the reconcile cost the server performs on the client's behalf. +// TODO: derive from server configuration and per-tier LOD limits. +pub const SERVER_MAX_RADIUS: u16 = 12; + +/// Worldgen version stamped on delivered chunk diffs. A single version exists today; this becomes the chunk's stored version once worldgen versioning lands. +const WORLDGEN_VERSION: u32 = 0; + +/// The load and drop lists produced by diffing a client's previous desired set against a new one. +#[derive(Debug, Default, PartialEq, Eq)] +pub struct DesiredDiff { + /// Positions newly wanted (present in the new set, absent from the previous). Delivered once resident. + pub added: Vec, + /// Positions no longer wanted (present in the previous set, absent from the new). The client is told to drop each it holds. + pub removed: Vec, +} + +/// Computes the load/drop diff between a client's `previous` and `new` desired sets. +#[must_use] +pub fn desired_diff( + previous: &HashSet, + new: &HashSet, +) -> DesiredDiff { + let mut added: Vec = new.difference(previous).copied().collect(); + let mut removed: Vec = previous.difference(new).copied().collect(); + added.sort_unstable(); + removed.sort_unstable(); + DesiredDiff { added, removed } +} + +/// Tracks one connected client's chunk subscription and what has been delivered to it. +pub struct ClientStream { + /// Outbound handle onto the client's chunk stream. + sink: ChunkSink, + /// The chunk positions the client currently wants resident, already clamped to [`SERVER_MAX_RADIUS`]. + desired: HashSet, + /// Positions already delivered to the client as [`ChunkMessage::Chunk`]. + sent: HashSet, +} + +impl ClientStream { + /// Creates a stream for a freshly connected client that has not yet subscribed. + #[must_use] + pub fn new(sink: ChunkSink) -> Self { + Self { + sink, + desired: HashSet::new(), + sent: HashSet::new(), + } + } + + /// Returns the client's current desired set, for folding into the world reconcile union. + #[must_use] + pub fn desired(&self) -> &HashSet { + &self.desired + } + + /// Applies a new subscription centered on `center` with load radius `radius`. + /// + /// The radius is clamped to [`SERVER_MAX_RADIUS`], the desired set is recomputed, and a [`ChunkMessage::Drop`] is emitted for every already-delivered chunk that left the set. Chunks newly entering the set are not sent here; they are delivered by [`ClientStream::flush`] once resident. Returns the number of newly-wanted positions and the number of drops emitted. + pub fn resubscribe(&mut self, center: ChunkPos, radius: u16) -> (usize, usize) { + let clamped = radius.min(SERVER_MAX_RADIUS); + let mut new_desired = HashSet::new(); + cylinder_chunks(center, i32::from(clamped), &mut new_desired); + + let diff = desired_diff(&self.desired, &new_desired); + let added = diff.added.len(); + let mut drops = 0; + for pos in diff.removed { + // Only chunks actually delivered need an explicit drop; positions that were wanted but never resident were never held by the client. + if self.sent.remove(&pos) { + self.sink.send(ChunkMessage::Drop { pos }); + drops += 1; + } + } + self.desired = new_desired; + (added, drops) + } + + /// Delivers every desired-but-undelivered chunk that has become resident in `world`. + /// + /// Each chunk is encoded as a [`ChunkData`] diff against `baseline` (an all-air chunk), making the payload self-contained. Positions still pending in the worker pool are skipped and retried on a later call. Returns the number of chunks delivered. + pub fn flush(&mut self, world: &ServerWorld, baseline: &Chunk) -> usize { + // Collected first to avoid borrowing `self.desired` while mutating `self.sent`. + let ready: Vec = self + .desired + .iter() + .filter(|pos| !self.sent.contains(pos)) + .copied() + .collect(); + + let mut delivered = 0; + for pos in ready { + let Some(chunk) = world.chunk(pos) else { + // Not resident yet; a later flush retries once the worker pool returns it. + continue; + }; + let data = ChunkData::from_diff(pos, WORLDGEN_VERSION, baseline, chunk); + self.sink.send(ChunkMessage::Chunk { pos, data }); + self.sent.insert(pos); + delivered += 1; + } + delivered + } +} + +#[cfg(test)] +#[path = "tests/client_stream.rs"] +mod tests; diff --git a/crates/server/src/main.rs b/crates/server/src/main.rs index 2c38f8f..af014be 100644 --- a/crates/server/src/main.rs +++ b/crates/server/src/main.rs @@ -6,6 +6,8 @@ /// A bounded LRU cache of regenerated chunk baselines, shared across the worker pool. pub mod chunk_cache; +/// Per-connection chunk-streaming state: desired-set tracking and delivery. +pub mod client_stream; /// Entity components describing players and other world-streaming anchors. pub mod player; /// On-disk persistence: region files and the atomic durability layer. @@ -13,7 +15,7 @@ pub mod save; /// Authoritative chunk storage and generation logic for the server. pub mod world_server; -use std::collections::HashSet; +use std::collections::{HashMap, HashSet}; use std::fs; use std::net::{Ipv4Addr, SocketAddr}; use std::time::Duration; @@ -22,10 +24,11 @@ use anyhow::Context; use bevy_ecs::prelude::{Query, ResMut, Schedule, With, World}; use glam::Vec3; use shared::generator::{VoxelGenerator, WorldGenConfig}; -use shared::world::{ChunkPos, EntityPos}; -use tracing::{debug, info}; +use shared::world::{Chunk, ChunkPos, EntityPos}; +use tracing::{debug, info, warn}; -use net::NetworkServer; +use client_stream::ClientStream; +use net::{NetworkServer, ServerEvent}; use player::{Player, Position, ViewDistance}; use world_server::{ServerWorld, cylinder_chunks}; @@ -130,12 +133,53 @@ fn main() -> anyhow::Result<()> { .context("spawning network server")?; info!(%local_addr, "network endpoint listening"); + // Chunk diffs are computed against an all-air baseline so each delivered payload is self-contained: the client renders only server-owned content and has no generator to reconstruct a worldgen baseline. Allocated once and shared across every delivery. + let empty_baseline = Chunk::default(); + + // Per-connection streaming state, keyed by the stable session id the network thread assigns. + let mut clients: HashMap = HashMap::new(); + // Authoritative simulation loop. loop { - schedule.run(&mut world); - + // Fold network events into per-client subscription state. for event in network.poll_events() { - info!(?event, "network event"); + match event { + ServerEvent::ClientConnected { id, hello, chunks } => { + info!(id, name = %hello.player_identity.display_name, "client connected"); + clients.insert(id, ClientStream::new(chunks)); + } + ServerEvent::ClientDisconnected { id, reason } => { + info!(id, %reason, "client disconnected"); + clients.remove(&id); + } + ServerEvent::ChunkSubscribe { id, request } => { + if let Some(client) = clients.get_mut(&id) { + let (added, drops) = client.resubscribe(request.center, request.radius); + info!( + id, + added, drops, "client {id}: +{added} chunks, -{drops} drops" + ); + } else { + warn!(id, "chunk subscribe from unknown session"); + } + } + } + } + + // Reconcile the resident world to the union of every client's desired set. A chunk survives as long as any connected client wants it; when no client is connected the union is empty and the world drains. + let mut desired = HashSet::new(); + for client in clients.values() { + desired.extend(client.desired().iter().copied()); + } + world.resource_mut::().reconcile(&desired); + + // Deliver newly-resident chunks to each client. Loads dispatched above may not be resident this tick; `flush` retries on later ticks until the worker pool returns them. + let server_world = world.resource::(); + for (id, client) in &mut clients { + let delivered = client.flush(server_world, &empty_baseline); + if delivered > 0 { + debug!(id, delivered, "delivered resident chunks"); + } } // Advisory ~20 Hz cadence until the real tick scheduler lands. diff --git a/crates/server/src/tests/client_stream.rs b/crates/server/src/tests/client_stream.rs new file mode 100644 index 0000000..12ca31e --- /dev/null +++ b/crates/server/src/tests/client_stream.rs @@ -0,0 +1,85 @@ +// SPDX-License-Identifier: AGPL-3.0-only + +//! Unit tests for the per-client desired-set diff. + +use super::*; +use crate::world_server::cylinder_chunks; + +/// Builds the streaming cylinder around `center` at `radius` as a set, mirroring what a subscription produces. +fn cylinder(center: ChunkPos, radius: i32) -> HashSet { + let mut set = HashSet::new(); + cylinder_chunks(center, radius, &mut set); + set +} + +#[test] +fn diff_of_equal_sets_is_empty() { + let set = cylinder(ChunkPos::new(0, 0, 0), 3); + let diff = desired_diff(&set, &set); + assert!( + diff.added.is_empty(), + "no chunks are added when the set is unchanged" + ); + assert!( + diff.removed.is_empty(), + "no chunks are removed when the set is unchanged" + ); +} + +#[test] +fn from_empty_previous_adds_all_new() { + let new = cylinder(ChunkPos::new(5, 0, -2), 2); + let diff = desired_diff(&HashSet::new(), &new); + assert_eq!( + diff.added.len(), + new.len(), + "an initial subscribe adds the whole set" + ); + assert!( + diff.removed.is_empty(), + "an initial subscribe removes nothing" + ); +} + +#[test] +fn center_move_by_one_chunk_swaps_symmetric_shells() { + let previous = cylinder(ChunkPos::new(0, 0, 0), 3); + let new = cylinder(ChunkPos::new(1, 0, 0), 3); + let diff = desired_diff(&previous, &new); + + // Every added chunk is genuinely new; every removed chunk genuinely left. + for pos in &diff.added { + assert!( + new.contains(pos) && !previous.contains(pos), + "added chunks are new-only" + ); + } + for pos in &diff.removed { + assert!( + previous.contains(pos) && !new.contains(pos), + "removed chunks are previous-only" + ); + } + + // A one-chunk shift keeps the overlap resident, so neither list is the whole set. + assert!(!diff.added.is_empty() && diff.added.len() < new.len()); + assert!(!diff.removed.is_empty()); + + // The cylinder is translation-invariant in size, so a shift moves equal counts in and out. + assert_eq!(diff.added.len(), diff.removed.len()); +} + +#[test] +fn added_and_removed_are_sorted() { + let previous = cylinder(ChunkPos::new(0, 0, 0), 2); + let new = cylinder(ChunkPos::new(2, 1, 0), 2); + let diff = desired_diff(&previous, &new); + assert!( + diff.added.windows(2).all(|w| w[0] <= w[1]), + "added list is sorted" + ); + assert!( + diff.removed.windows(2).all(|w| w[0] <= w[1]), + "removed list is sorted" + ); +}