feat(server): stream chunks to clients for their subscribed region

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<ChunkPos>,
+    /// 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<ChunkPos>,
+}
+
+/// Computes the load/drop diff between a client's `previous` and `new` desired sets.
+#[must_use]
+pub fn desired_diff<S: std::hash::BuildHasher>(
+    previous: &HashSet<ChunkPos, S>,
+    new: &HashSet<ChunkPos, S>,
+) -> DesiredDiff {
+    let mut added: Vec<ChunkPos> = new.difference(previous).copied().collect();
+    let mut removed: Vec<ChunkPos> = 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<ChunkPos>,
+    /// Positions already delivered to the client as [`ChunkMessage::Chunk`].
+    sent: HashSet<ChunkPos>,
+}
+
+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<ChunkPos> {
+        &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<ChunkPos> = 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;

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<u64, ClientStream> = 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::<ServerWorld>().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::<ServerWorld>();
+        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.

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<ChunkPos> {
+    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"
+    );
+}