CryoforgeNexus/Synvael
1
0
Fork 0
Code Issues Pull requests Actions Packages Projects Releases Wiki Activity

Merge pull request #6 from Cryoforge-Nexus/feat/chunk-streaming-slice

Browse source 
feat(client): stream and render server-authoritative chunks end-to-end
This commit is contained in:
Serkyo 2026-07-22 04:01:51 +02:00 committed by GitHub
parent 2ca9ae801d d01f45ec18
commit e7c7dc5aea
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
28 changed files with 1488 additions and 465 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
Cargo.lock generated
View file

@ -551,7 +551,6 @@ dependencies = [
"net",
"raw-window-handle",
"renderer",
"serde_json",
"shared",
"tracing",
"tracing-subscriber",

5
assets/shaders/cube.vert
View file

@ -8,10 +8,13 @@ layout(location = 0) out vec3 frag_color;
layout(push_constant) uniform PushConstants {
mat4 mvp;
vec4 chunk_offset;
} push_constants;
void main() {
gl_Position = push_constants.mvp * vec4(in_position, 1.0);
// The chunk-local vertex is shifted into world space by the per-chunk offset before projection.
vec3 world_position = in_position + push_constants.chunk_offset.xyz;
gl_Position = push_constants.mvp * vec4(world_position, 1.0);
frag_color = in_color;
}

BIN
assets/shaders/cube.vert.spv
View file

Binary file not shown.

1
crates/client/Cargo.toml
View file

@ -17,6 +17,5 @@ renderer = { path = "../renderer" }
glam.workspace = true
raw-window-handle.workspace = true
ash-window.workspace = true
serde_json.workspace = true
shared = { path = "../shared" }
net = { version = "0.1.0", path = "../net" }

203
crates/client/src/chunks.rs Normal file
View file

@ -0,0 +1,203 @@
// SPDX-License-Identifier: AGPL-3.0-only
//! Client-side chunk streaming around the camera.
use std::collections::HashSet;
use shared::protocol::chunk::ChunkMessage;
use shared::world::{CHUNK_SIZE, Chunk, ChunkData, ChunkPos};
use tracing::{debug, error};
use crate::meshing;
/// Radius, in chunks, of the region kept resident around the camera center. Also the radius the client subscribes with, so the server's resident set matches the client's.
// TODO: make configurable / drive from view-distance setting.
pub const LOAD_RADIUS: i32 = 8;
/// Maximum number of chunks meshed and uploaded in a single call to [`ChunkManager::update`], bounding per-frame meshing work so the winit loop stays responsive. Deliveries beyond the budget remain queued for the next frame.
// TODO: move meshing to a worker pool.
const LOADS_PER_UPDATE: usize = 4;
/// Tracks which server-streamed chunks are currently uploaded to the renderer.
pub struct ChunkManager {
/// Positions uploaded to the renderer (whether or not they produced a non-empty mesh), so unload and drop can reconcile against the renderer.
resident: HashSet<ChunkPos>,
/// Reused all-air baseline that server [`ChunkData`] diffs are materialized against.
baseline: Chunk,
}
impl ChunkManager {
/// Creates a manager with no chunks yet resident.
#[must_use]
pub fn new() -> Self {
Self {
resident: HashSet::new(),
baseline: Chunk::default(),
}
}
/// Reconciles the resident chunk set: evicts chunks outside the load radius around `center`, then applies queued server deliveries under a per-frame meshing budget.
///
/// The client's own radius eviction runs independently of the server's authoritative `Drop`, so memory stays bounded even if the server is slow to drop chunks that leave the region.
pub fn update(
&mut self,
center: ChunkPos,
deliveries: &net::ChunkStream,
renderer: &mut renderer::Renderer,
) {
let unloaded = self.unload_outside(center, renderer);
let mut loaded = 0;
let mut dropped = 0;
// Only chunk deliveries count against the meshing budget; drops are cheap and always applied.
while loaded < LOADS_PER_UPDATE {
match deliveries.try_recv() {
Ok(ChunkMessage::Chunk { pos, data }) => {
self.apply_chunk(pos, &data, renderer);
loaded += 1;
}
Ok(ChunkMessage::Drop { pos }) => {
if self.drop_chunk(pos, renderer) {
dropped += 1;
}
}
// Empty or disconnected: nothing more to apply this frame.
Err(_) => break,
}
}
if loaded > 0 || dropped > 0 || unloaded > 0 {
debug!(
loaded,
dropped,
unloaded,
resident = self.resident.len(),
"chunk stream reconciled"
);
}
}
/// Materializes, meshes, and uploads one delivered chunk, marking its position resident.
fn apply_chunk(&mut self, pos: ChunkPos, data: &ChunkData, renderer: &mut renderer::Renderer) {
let chunk = data.materialize(&self.baseline);
let (vertices, indices) = meshing::generate_mesh(&chunk);
// Uploading a zero-length buffer is invalid, so an all-air chunk skips the renderer entirely. It is still marked resident below so a later delivery is not double-counted.
if !indices.is_empty() {
// Chunk coordinates and CHUNK_SIZE are small and represent exactly as f32.
#[expect(
clippy::cast_precision_loss,
reason = "chunk coordinates stay well within f32's exact-integer range"
)]
let world_offset = {
let size = CHUNK_SIZE as f32;
[
pos.x as f32 * size,
pos.y as f32 * size,
pos.z as f32 * size,
]
};
if let Err(e) =
renderer.insert_mesh((pos.x, pos.y, pos.z), &vertices, &indices, world_offset)
{
error!(?pos, "failed to upload chunk mesh: {e}");
}
}
self.resident.insert(pos);
}
/// Removes one chunk from the renderer on the server's authoritative instruction, returning whether it was resident.
fn drop_chunk(&mut self, pos: ChunkPos, renderer: &mut renderer::Renderer) -> bool {
if self.resident.remove(&pos) {
renderer.remove_mesh((pos.x, pos.y, pos.z));
true
} else {
false
}
}
/// Evicts every resident chunk outside the load radius around `center`, returning the number removed.
fn unload_outside(&mut self, center: ChunkPos, renderer: &mut renderer::Renderer) -> usize {
let desired = desired_chunks(center, LOAD_RADIUS);
let stale: Vec<ChunkPos> = self
.resident
.iter()
.filter(|pos| !desired.contains(pos))
.copied()
.collect();
for pos in &stale {
renderer.remove_mesh((pos.x, pos.y, pos.z));
self.resident.remove(pos);
}
stale.len()
}
}
impl Default for ChunkManager {
fn default() -> Self {
Self::new()
}
}
/// Returns the set of chunk positions within the streaming cylinder around `center`.
///
/// The region is a disc of `radius` chunks in the horizontal XZ plane and half that extent in Y, matching the flatter vertical shape of the playable world. This mirrors the server's `world_server::cylinder_chunks`.
#[must_use]
pub fn desired_chunks(center: ChunkPos, radius: i32) -> HashSet<ChunkPos> {
let mut out = HashSet::new();
for x in center.x - radius..=center.x + radius {
for z in center.z - radius..=center.z + radius {
let dx = x - center.x;
let dz = z - center.z;
// Keep only the columns whose XZ distance falls within the disc.
if dx * dx + dz * dz <= radius * radius {
for y in center.y - radius / 2..=center.y + radius / 2 {
out.insert(ChunkPos::new(x, y, z));
}
}
}
}
out
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn center_is_always_included() {
let center = ChunkPos::new(0, 0, 0);
assert!(desired_chunks(center, 4).contains(&center));
}
#[test]
fn excludes_columns_beyond_the_disc() {
let set = desired_chunks(ChunkPos::new(0, 0, 0), 4);
// One chunk past the radius along an axis: squared distance 25 > 16.
assert!(!set.contains(&ChunkPos::new(5, 0, 0)));
// The far corner: squared distance 4*4 + 4*4 = 32 > 16.
assert!(!set.contains(&ChunkPos::new(4, 0, 4)));
}
#[test]
fn vertical_extent_is_half_the_radius() {
let set = desired_chunks(ChunkPos::new(0, 0, 0), 4);
// radius / 2 == 2, so the column at the center spans y in [-2, 2].
assert!(set.contains(&ChunkPos::new(0, 2, 0)));
assert!(!set.contains(&ChunkPos::new(0, 3, 0)));
}
#[test]
fn set_is_translation_invariant() {
// Shifting the center shifts every member by the same offset; this also exercises negative coordinates on the shifted side.
let base = desired_chunks(ChunkPos::new(0, 0, 0), 3);
let shifted: HashSet<ChunkPos> = base
.iter()
.map(|p| ChunkPos::new(p.x - 10, p.y - 10, p.z - 10))
.collect();
assert_eq!(shifted, desired_chunks(ChunkPos::new(-10, -10, -10), 3));
}
}

118
crates/client/src/main.rs
View file

@ -3,7 +3,9 @@
//! Main entry point for the Synvael client.
//!
//! This crate handles window creation, input processing, and drives the renderer to display the game world.
mod camera;
mod chunks;
mod meshing;
use std::time::Instant;
@ -56,8 +58,14 @@ struct App {
input: InputState,
/// Timestamp of the previous frame, used to derive delta-time. `None` before the first frame.
last_frame: Option<Instant>,
/// Receives the outcome of the background connect and handshake, drained non-blocking from the event loop. `None` before the connection is started and once the outcome has been observed.
handshake_rx: Option<net::ConnectOutcome>,
/// Handles onto the background network connection: the handshake outcome, the chunk-subscription sender, and the chunk-delivery receiver. `None` before the connection is started.
link: Option<net::ClientLink>,
/// Whether the handshake has completed successfully. Gates chunk subscription until the connection is usable.
connected: bool,
/// The chunk position the camera last subscribed around, so a new subscription is sent only when the center chunk changes.
last_center: Option<shared::world::ChunkPos>,
/// Streams chunk meshes in and out around the camera. `None` until the renderer is initialised on resume.
chunks: Option<chunks::ChunkManager>,
}
impl Default for App {
@ -73,7 +81,10 @@ impl Default for App {
),
input: InputState::default(),
last_frame: None,
handshake_rx: None,
link: None,
connected: false,
last_center: None,
chunks: None,
}
}
}
@ -146,40 +157,9 @@ impl ApplicationHandler for App {
self.window = Some(window);
self.renderer = Some(renderer);
#[expect(
clippy::expect_used,
reason = "startup asset load; a missing worldgen config is unrecoverable at launch"
)]
let config_str = std::fs::read_to_string("assets/data/worldgen/default.json")
.expect("Failed to read worldgen config");
#[expect(
clippy::expect_used,
reason = "startup config parse; a malformed worldgen config is unrecoverable at launch"
)]
let worldgen_config: shared::generator::WorldGenConfig =
serde_json::from_str(&config_str).expect("Failed to parse worldgen config");
let seed = 4_813_530;
let generator = shared::generator::VoxelGenerator::new(worldgen_config, seed);
let chunk = generator.generate_chunk(shared::world::ChunkPos::new(0, 0, 0));
let (vertices, indices) = meshing::generate_mesh(&chunk);
tracing::info!(
"Generated Mesh with {} vertices and {} indices!",
vertices.len(),
indices.len()
);
#[expect(
clippy::expect_used,
reason = "the renderer is assigned earlier in this function"
)]
self.renderer
.as_mut()
.expect("Renderer initialized")
.update_mesh(&vertices, &indices)
.expect("Failed to upload terrain to GPU");
// The client renders only server-streamed terrain and no longer generates chunks locally.
// TODO: offline/singleplayer via an in-process server would reintroduce a local world source here.
self.chunks = Some(chunks::ChunkManager::new());
// Kick off a background connect + handshake to the local server.
let hello = shared::protocol::ClientHello {
@ -194,7 +174,7 @@ impl ApplicationHandler for App {
let server_addr =
std::net::SocketAddr::from((std::net::Ipv4Addr::LOCALHOST, net::DEFAULT_PORT));
info!("Connecting to server at {server_addr}");
self.handshake_rx = Some(net::connect_in_background(server_addr, hello));
self.link = Some(net::connect_in_background(server_addr, hello));
}
fn window_event(&mut self, event_loop: &ActiveEventLoop, _id: WindowId, event: WindowEvent) {
@ -202,6 +182,15 @@ impl ApplicationHandler for App {
WindowEvent::CloseRequested => {
event_loop.exit();
}
WindowEvent::Resized(size) => {
// Rebuild the swapchain to match the new surface size. Without this the swapchain keeps its initial extent and the compositor stretches the fixed-size image to the window, distorting the aspect ratio.
if let Some(renderer) = self.renderer.as_mut()
&& let Err(e) = renderer.recreate_swapchain(size.width, size.height)
{
error!("Failed to recreate swapchain on resize: {e}");
event_loop.exit();
}
}
WindowEvent::KeyboardInput { event, .. } => {
let pressed = event.state == ElementState::Pressed;
if let PhysicalKey::Code(code) = event.physical_key {
@ -218,28 +207,28 @@ impl ApplicationHandler for App {
}
}
WindowEvent::RedrawRequested => {
// Non-blocking check for the handshake outcome.
let mut handshake_done = false;
if let Some(rx) = self.handshake_rx.as_ref() {
match rx.try_recv() {
Ok(Ok(ack)) => {
// Non-blocking check for the handshake outcome. The link is retained after success so its chunk channels can be used; only a failure discards it.
if !self.connected {
let outcome = self
.link
.as_ref()
.and_then(|link| link.handshake.try_recv().ok());
match outcome {
Some(Ok(ack)) => {
info!(
protocol_version = ack.protocol_version,
"handshake complete"
);
handshake_done = true;
self.connected = true;
}
Ok(Err(reason)) => {
Some(Err(reason)) => {
warn!("handshake failed: {reason}");
handshake_done = true;
self.link = None;
}
// Empty: not ready yet. Disconnected: the network thread ended.
Err(_) => {}
// No outcome yet (empty), or the network thread ended (disconnected).
None => {}
}
}
if handshake_done {
self.handshake_rx = None;
}
// Derive delta-time from the previous frame so movement is framerate-independent. The first frame has no predecessor and therefore advances by zero seconds.
let now = Instant::now();
@ -252,6 +241,33 @@ impl ApplicationHandler for App {
// The accumulated motion has been applied; clear it so it is not counted twice.
self.input.mouse_delta = (0.0, 0.0);
// Reconcile streamed chunks toward the chunk the camera now occupies. `from_world` floors via `div_euclid`, so negative coordinates map to the correct chunk.
let pos = self.camera.position;
let center = shared::world::ChunkPos::from_world(
f64::from(pos.x),
f64::from(pos.y),
f64::from(pos.z),
);
// Subscribe to the region around the camera whenever the center chunk changes, so the server streams the matching set. The client subscribes with its own load radius so the server's resident set aligns with what the client keeps.
if self.connected && self.last_center != Some(center) {
if let Some(link) = self.link.as_ref() {
let radius = u16::try_from(chunks::LOAD_RADIUS).unwrap_or(u16::MAX);
link.subscribe
.send(shared::protocol::chunk::ChunkSubscribe { center, radius });
}
self.last_center = Some(center);
}
// Apply queued server deliveries and reconcile the resident set against the camera.
if let (Some(chunks), Some(link), Some(renderer)) = (
self.chunks.as_mut(),
self.link.as_ref(),
self.renderer.as_mut(),
) {
chunks.update(center, &link.chunks, renderer);
}
let view = self.camera.view_matrix();
if let Some(Err(e)) = self.renderer.as_mut().map(|r| r.draw_frame(view)) {
error!("Failed to draw frame: {e}");

176
crates/net/src/chunk.rs Normal file
View file

@ -0,0 +1,176 @@
// SPDX-License-Identifier: AGPL-3.0-only
//! Chunk-stream transport: the per-connection task that pumps chunk subscriptions and deliveries.
//!
//! After the handshake, each connection carries a dedicated bidirectional QUIC stream for chunk sync (the canonical `StreamLayout::chunk_lod0` id). The client writes [`ChunkSubscribe`] requests on it and the server writes [`ChunkMessage`] deliveries back on the same stream. This module owns the server-side pump: a single [`tokio::select`] loop that reads subscriptions off the stream and forwards them to the synchronous simulation loop, while draining outbound [`ChunkMessage`]s handed to it by that loop.
//!
//! The two channels crossing the async/sync boundary run in opposite directions and therefore use different primitives. Inbound (`ChunkSubscribe` arriving async, consumed by the sync loop) reuses the crossbeam [`ServerEvent`] channel, whose sender is non-blocking. Outbound (a `ChunkMessage` produced by the sync loop, consumed async) uses a `tokio` unbounded MPSC: its `send` is synchronous, so the non-async simulation thread can push without a runtime, while the receiver's `recv().await` composes into the pump's `select!`. A blocking `crossbeam` receiver would instead freeze the current-thread runtime and cannot appear in a `select!` arm.
use shared::protocol::chunk::{ChunkMessage, ChunkSubscribe};
use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
use tracing::{debug, warn};
use crate::codec::{MAX_CHUNK_FRAME_LEN, read_frame, write_frame};
use crate::runtime::ServerEvent;
/// A synchronous handle the simulation loop uses to hand [`ChunkMessage`]s to a connection's chunk-stream task.
///
/// The wrapped channel is a `tokio` unbounded MPSC. Its `send` is synchronous and callable from the non-async simulation thread with no runtime in scope, while the connection's task drains the receiver with `recv().await` so it composes into the task's `select!`. The `tokio` sender type is kept private so the `server` crate never names it.
#[derive(Debug, Clone)]
pub struct ChunkSink {
/// Outbound queue drained by the connection's chunk-stream task.
tx: UnboundedSender<ChunkMessage>,
}
impl ChunkSink {
/// Wraps `tx` as a chunk sink.
pub(crate) fn new(tx: UnboundedSender<ChunkMessage>) -> Self {
Self { tx }
}
/// Queues `msg` for delivery on the connection's chunk stream.
///
/// Non-blocking. A send failure means the receiving task has ended (the connection dropped); it is logged at debug and swallowed, since the simulation loop cannot act on a departed connection.
pub fn send(&self, msg: ChunkMessage) {
if self.tx.send(msg).is_err() {
debug!("chunk sink send failed; connection task has ended");
}
}
}
/// Runs the server-side chunk-stream pump for one connection until the stream or connection closes.
///
/// Accepts the connection's chunk stream, then loops: inbound [`ChunkSubscribe`] frames are forwarded to the simulation loop as [`ServerEvent::ChunkSubscribe`], and outbound [`ChunkMessage`]s taken from `outbound` are written onto the stream. The loop ends when the peer closes the stream, when the events receiver is gone (the server is shutting down), or when the outbound sink is dropped.
pub(crate) async fn chunk_stream_task(
connection: quinn::Connection,
id: u64,
events: crossbeam_channel::Sender<ServerEvent>,
mut outbound: UnboundedReceiver<ChunkMessage>,
) {
// The client opens the chunk stream after the handshake; the server accepts it here, mirroring the control-stream convention.
let (mut send, mut recv) = match connection.accept_bi().await {
Ok(stream) => stream,
Err(error) => {
warn!(%error, id, "failed to accept chunk stream");
return;
}
};
loop {
tokio::select! {
// A subscription frame arrived from the client.
frame = read_frame::<ChunkSubscribe>(&mut recv, MAX_CHUNK_FRAME_LEN) => {
match frame {
Ok(request) => {
// A closed events receiver means the simulation loop is gone; nothing more to do.
if events
.send(ServerEvent::ChunkSubscribe { id, request })
.is_err()
{
break;
}
}
Err(error) => {
// A read error is the normal end of a client session (stream finished or reset).
debug!(%error, id, "chunk stream read ended");
break;
}
}
}
// The simulation loop handed back a chunk to deliver.
msg = outbound.recv() => {
match msg {
Some(message) => {
if let Err(error) = write_frame(&mut send, &message).await {
warn!(%error, id, "failed to write chunk frame; ending chunk stream");
break;
}
}
// The sink was dropped: the connection is being torn down.
None => break,
}
}
}
}
}
/// A synchronous handle the client's UI thread uses to push [`ChunkSubscribe`] requests to its network task.
///
/// The client-side mirror of [`ChunkSink`]: `send` is synchronous and callable from the winit loop with no runtime in scope, while the client's chunk task drains the receiver with `recv().await` so it composes into that task's `select!`.
#[derive(Debug, Clone)]
pub struct ChunkSubscriber {
/// Outbound queue of subscription updates drained by the client's chunk task.
tx: UnboundedSender<ChunkSubscribe>,
}
impl ChunkSubscriber {
/// Wraps `tx` as a chunk subscriber.
pub(crate) fn new(tx: UnboundedSender<ChunkSubscribe>) -> Self {
Self { tx }
}
/// Queues a subscription update for the server.
///
/// Non-blocking. A send failure means the network task has ended (the connection dropped); it is logged at debug and swallowed, since the UI thread cannot act on a departed connection.
pub fn send(&self, request: ChunkSubscribe) {
if self.tx.send(request).is_err() {
debug!("chunk subscriber send failed; network task has ended");
}
}
}
/// Runs the client-side chunk-stream pump for one connection until the stream or connection closes.
///
/// Opens the chunk stream, then loops: [`ChunkSubscribe`] requests taken from `subscribe` are written to the server, and inbound [`ChunkMessage`] frames are forwarded to the UI thread over `deliveries`. The loop ends when the UI drops its subscriber, when the delivery receiver is gone, or when the stream closes.
pub(crate) async fn client_chunk_task(
connection: quinn::Connection,
mut subscribe: UnboundedReceiver<ChunkSubscribe>,
deliveries: crossbeam_channel::Sender<ChunkMessage>,
) {
// The client opens the chunk stream after the handshake; the server accepts it, mirroring the control-stream convention.
let (mut send, mut recv) = match connection.open_bi().await {
Ok(stream) => stream,
Err(error) => {
warn!(%error, "failed to open chunk stream");
return;
}
};
loop {
tokio::select! {
// The UI thread pushed a subscription update to forward to the server.
request = subscribe.recv() => {
match request {
Some(request) => {
if let Err(error) = write_frame(&mut send, &request).await {
warn!(%error, "failed to write chunk subscribe; ending chunk stream");
break;
}
}
// The subscriber was dropped: the UI is shutting down.
None => break,
}
}
// A chunk arrived from the server.
frame = read_frame::<ChunkMessage>(&mut recv, MAX_CHUNK_FRAME_LEN) => {
match frame {
Ok(message) => {
// A closed delivery receiver means the UI is gone; nothing more to do.
if deliveries.send(message).is_err() {
break;
}
}
Err(error) => {
// A read error is the normal end of the session (stream finished or reset).
debug!(%error, "chunk stream read ended");
break;
}
}
}
}
}
}
#[cfg(test)]
#[path = "tests/chunk.rs"]
mod tests;

3
crates/net/src/codec.rs
View file

@ -9,6 +9,9 @@ use crate::error::NetError;
/// The maximum payload length, in bytes, accepted on the control stream (64 KiB), matching the mod-payload cap. Higher-bandwidth tiers such as chunk streaming define their own caps.
pub const MAX_CONTROL_FRAME_LEN: usize = 64 * 1024;
/// The maximum payload length, in bytes, accepted on a chunk stream (1 MiB). A worst-case fully-modified 32³ chunk serializes to roughly 256 KiB as a sparse `ChunkData` (32 768 edits of a varint index plus a `u16` block), so 1 MiB clears the worst case with comfortable margin while still bounding a malicious or corrupt peer's allocation.
pub const MAX_CHUNK_FRAME_LEN: usize = 1024 * 1024;
/// The maximum number of bytes an unsigned LEB128 varint may occupy for a `u64` value (`ceil(64 / 7)`).
const MAX_VARINT_LEN: usize = 10;

6
crates/net/src/lib.rs
View file

@ -6,13 +6,17 @@
//!
//! The synchronous simulation loop (`server`) and windowing loop (`client`) never touch the async runtime directly. They exchange messages with the network over channels, so the async runtime stays confined to this crate.
pub mod chunk;
pub mod codec;
pub mod endpoint;
pub mod error;
pub mod handshake;
pub mod runtime;
pub use runtime::{ConnectOutcome, NetworkServer, ServerEvent, connect_in_background};
pub use chunk::{ChunkSink, ChunkSubscriber};
pub use runtime::{
ChunkStream, ClientLink, ConnectOutcome, NetworkServer, ServerEvent, connect_in_background,
};
/// Default UDP port the server binds and the client connects to when none is configured.
// TODO: make the bind address and port configurable through server/client configuration.

64
crates/net/src/runtime.rs
View file

@ -5,9 +5,11 @@
use std::net::SocketAddr;
use std::thread;
use shared::protocol::chunk::{ChunkMessage, ChunkSubscribe};
use shared::protocol::{ClientHello, HandshakeAck};
use tracing::{info, warn};
use crate::chunk::{ChunkSink, ChunkSubscriber, chunk_stream_task, client_chunk_task};
use crate::endpoint::{client_endpoint, server_endpoint};
use crate::error::NetError;
use crate::handshake::{ServerConnection, accept_connection, connect};
@ -15,15 +17,32 @@ use crate::handshake::{ServerConnection, accept_connection, connect};
/// Channel receiver delivering the outcome of a background client connect: the negotiated [`HandshakeAck`] on success, or a human-readable error string on failure.
pub type ConnectOutcome = crossbeam_channel::Receiver<Result<HandshakeAck, String>>;
/// Non-blocking receiver of chunks delivered by the server, drained by the UI thread with `try_recv`.
pub type ChunkStream = crossbeam_channel::Receiver<ChunkMessage>;
/// Handles a background client connection exposes to the synchronous UI thread.
///
/// The network task keeps the QUIC connection alive on its own thread; this bundle is how the winit loop observes the handshake outcome, pushes subscription updates, and drains chunk deliveries, all without touching the async runtime.
pub struct ClientLink {
/// Handshake outcome, drained once for the negotiated ack or the failure reason.
pub handshake: ConnectOutcome,
/// Sends subscription updates (center and radius) to the server as the camera moves.
pub subscribe: ChunkSubscriber,
/// Receives chunk deliveries from the server, drained non-blocking each frame.
pub chunks: ChunkStream,
}
/// An event surfaced by the network thread to the synchronous server loop.
#[derive(Debug)]
pub enum ServerEvent {
/// A client completed the Synvael handshake. Carries the stable per-session id and the `ClientHello` it presented.
/// A client completed the Synvael handshake. Carries the stable per-session id, the `ClientHello` it presented, and the sink the simulation loop uses to deliver chunks to this connection's chunk stream.
ClientConnected {
/// Stable identifier assigned to this session for the lifetime of the connection.
id: u64,
/// The identity and build parameters the client advertised.
hello: ClientHello,
/// Outbound handle for delivering [`shared::protocol::chunk::ChunkMessage`]s to this client. The simulation loop retains it, keyed by `id`, until the matching [`ServerEvent::ClientDisconnected`].
chunks: ChunkSink,
},
/// A previously connected client's session ended.
ClientDisconnected {
@ -32,6 +51,13 @@ pub enum ServerEvent {
/// Human-readable description of why the connection closed.
reason: String,
},
/// A connected client updated its chunk subscription: the initial subscribe on connect, or a later update as its center chunk moves.
ChunkSubscribe {
/// Identifier of the session that sent the subscription, matching its [`ServerEvent::ClientConnected`].
id: u64,
/// The center and radius the client wants resident.
request: ChunkSubscribe,
},
}
/// Handle to the background networking thread and its owned `tokio` runtime.
@ -187,13 +213,26 @@ async fn handle_connection(
Ok(ServerConnection {
connection, hello, ..
}) => {
// The outbound chunk channel bridges the sync simulation loop to this connection's chunk-stream task; the sink is handed to the loop via the connect event.
let (chunk_tx, chunk_rx) = tokio::sync::mpsc::unbounded_channel();
// If the receiver is gone the server is shutting down; drop the connection silently.
if events
.send(ServerEvent::ClientConnected { id, hello })
.send(ServerEvent::ClientConnected {
id,
hello,
chunks: ChunkSink::new(chunk_tx),
})
.is_err()
{
return;
}
// The chunk pump runs on its own task so the connection-close wait below does not block it.
tokio::spawn(chunk_stream_task(
connection.clone(),
id,
events.clone(),
chunk_rx,
));
let reason = connection.closed().await;
let _ = events.send(ServerEvent::ClientDisconnected {
id,
@ -206,12 +245,17 @@ async fn handle_connection(
}
}
/// Runs a one-shot connect and Synvael handshake against `server_addr` on a background `tokio` thread, reporting the outcome to the returned receiver.
/// Runs a connect and Synvael handshake against `server_addr` on a background `tokio` thread, then pumps the chunk stream, returning the handles the UI thread uses to observe and drive the connection.
///
/// The returned [`ClientLink`] is available immediately; its channels buffer until the handshake completes and the chunk task starts. A handshake failure is reported on `handshake` and leaves the subscribe and chunk channels inert.
#[must_use]
pub fn connect_in_background(server_addr: SocketAddr, hello: ClientHello) -> ConnectOutcome {
pub fn connect_in_background(server_addr: SocketAddr, hello: ClientHello) -> ClientLink {
let (outcome_tx, outcome_rx) = crossbeam_channel::bounded(1);
// Retained so a failure to spawn the thread can still be reported to the caller.
let spawn_err_tx = outcome_tx.clone();
// Subscription updates flow UI -> network (sync send, async recv); chunk deliveries flow network -> UI (async send, sync try_recv).
let (subscribe_tx, subscribe_rx) = tokio::sync::mpsc::unbounded_channel::<ChunkSubscribe>();
let (chunks_tx, chunks_rx) = crossbeam_channel::unbounded::<ChunkMessage>();
let spawned = thread::Builder::new()
.name("net-client".to_owned())
@ -242,9 +286,9 @@ pub fn connect_in_background(server_addr: SocketAddr, hello: ClientHello) -> Con
if outcome_tx.send(Ok(connected.ack.clone())).is_err() {
return;
}
// Keep the connection alive on the network thread until the server closes it. A full client session pump is a later concept.
let reason = connected.connection.closed().await;
info!(%reason, "server connection closed");
// Pump the chunk stream on this thread until the UI drops its handles or the server closes the connection.
client_chunk_task(connected.connection, subscribe_rx, chunks_tx).await;
warn!("server connection closed");
}
Err(error) => {
let _ = outcome_tx.send(Err(error.to_string()));
@ -257,5 +301,9 @@ pub fn connect_in_background(server_addr: SocketAddr, hello: ClientHello) -> Con
let _ = spawn_err_tx.send(Err(format!("failed to spawn network thread: {error}")));
}
outcome_rx
ClientLink {
handshake: outcome_rx,
subscribe: ChunkSubscriber::new(subscribe_tx),
chunks: chunks_rx,
}
}

94
crates/net/src/tests/chunk.rs Normal file
View file

@ -0,0 +1,94 @@
// SPDX-License-Identifier: AGPL-3.0-only
//! Loopback integration test for the chunk-stream transport.
//!
//! Binds a real QUIC server endpoint, completes the handshake, and drives the server-side [`chunk_stream_task`] end-to-end: a client-sent `ChunkSubscribe` must surface on the simulation-loop events channel as [`ServerEvent::ChunkSubscribe`], and a `ChunkMessage` pushed through the [`ChunkSink`] must be received by the client on the chunk stream.
use std::time::Duration;
use crate::chunk::{ChunkSink, chunk_stream_task};
use crate::codec::{MAX_CHUNK_FRAME_LEN, read_frame, write_frame};
use crate::endpoint::{client_endpoint, server_endpoint};
use crate::handshake::{accept_connection, connect};
use crate::runtime::ServerEvent;
use shared::protocol::chunk::{ChunkMessage, ChunkSubscribe};
use shared::protocol::{ClientHello, FeatureFlags, PROTOCOL_VERSION, PlayerIdentity};
use shared::world::{ChunkData, ChunkPos};
/// Builds a minimal `ClientHello` advertising the current protocol version.
fn hello(display_name: &str) -> ClientHello {
ClientHello {
protocol_version: PROTOCOL_VERSION,
client_build: "synvael-client-test".to_owned(),
player_identity: PlayerIdentity {
display_name: display_name.to_owned(),
},
installed_packs: vec![],
requested_features: FeatureFlags(0),
}
}
#[tokio::test(flavor = "multi_thread")]
async fn chunk_subscribe_and_delivery_round_trip()
-> Result<(), Box<dyn std::error::Error + Send + Sync>> {
let server = server_endpoint("127.0.0.1:0".parse()?)?;
let server_addr = server.local_addr()?;
// Stand in for the simulation loop's channels: the events channel the task forwards subscriptions to, and the outbound sink it drains deliveries from.
let (events_tx, events_rx) = crossbeam_channel::unbounded::<ServerEvent>();
let (chunk_tx, chunk_rx) = tokio::sync::mpsc::unbounded_channel::<ChunkMessage>();
let sink = ChunkSink::new(chunk_tx);
// Server side: accept one connection, complete the handshake, then run the chunk pump until the client closes.
let server_task = tokio::spawn(async move {
let incoming = server.accept().await.ok_or("server endpoint closed")?;
let conn = accept_connection(incoming, "synvael-server-test".to_owned(), 20).await?;
chunk_stream_task(conn.connection, 7, events_tx, chunk_rx).await;
Ok::<_, Box<dyn std::error::Error + Send + Sync>>(())
});
// Client side: connect, then open the chunk stream and send a subscription.
let client = client_endpoint()?;
let connected = connect(&client, server_addr, "localhost", hello("Tester")).await?;
let (mut client_send, mut client_recv) = connected.connection.open_bi().await?;
let subscribe = ChunkSubscribe {
center: ChunkPos::new(1, 2, 3),
radius: 4,
};
write_frame(&mut client_send, &subscribe).await?;
// The task must forward the subscription to the events channel. The crossbeam receiver is blocking, so it is polled on a blocking thread to avoid stalling the runtime.
let event = tokio::task::spawn_blocking(move || {
events_rx
.recv_timeout(Duration::from_secs(5))
.map(|e| (e, events_rx))
})
.await?;
let (event, events_rx) = event?;
match event {
ServerEvent::ChunkSubscribe { id, request } => {
assert_eq!(id, 7, "the subscribe must carry the session id");
assert_eq!(request, subscribe, "the subscribe must round-trip intact");
}
other => return Err(format!("expected ChunkSubscribe, got {other:?}").into()),
}
// The simulation loop hands a chunk back through the sink; the client must receive it on the stream.
let data = ChunkData::new(ChunkPos::new(1, 2, 3), 0);
let delivered = ChunkMessage::Chunk {
pos: ChunkPos::new(1, 2, 3),
data: data.clone(),
};
sink.send(delivered.clone());
let received = read_frame::<ChunkMessage>(&mut client_recv, MAX_CHUNK_FRAME_LEN).await?;
assert_eq!(received, delivered, "the chunk must round-trip intact");
// Close the client so the server task's pump ends and the endpoint winds down cleanly.
drop(events_rx);
drop(sink);
drop(connected);
server_task.await??;
Ok(())
}

22
crates/renderer/src/instance.rs
View file

@ -2,7 +2,11 @@
use crate::error::RendererError;
use ash::{Entry, Instance, ext, vk};
use std::ffi::{CStr, c_char};
use std::ffi::c_char;
// `CStr` and the tracing macros are used only by the debug-build validation callback.
#[cfg(debug_assertions)]
use std::ffi::CStr;
#[cfg(debug_assertions)]
use tracing::{debug, error, info, warn};
/// Creates a Vulkan instance and optionally a debug messenger.
@ -21,7 +25,22 @@ pub fn create_instance(
),
RendererError,
> {
// The validation extension and layer are pushed only in debug builds, so in release builds these vectors are never mutated after initialization.
#[cfg_attr(
not(debug_assertions),
expect(
unused_mut,
reason = "the debug-only block below mutates these vectors"
)
)]
let mut extensions = required_extensions.to_vec();
#[cfg_attr(
not(debug_assertions),
expect(
unused_mut,
reason = "the debug-only block below mutates these vectors"
)
)]
let mut layers = Vec::new();
#[cfg(debug_assertions)]
@ -70,6 +89,7 @@ pub fn create_instance(
/// # Safety
///
/// Invoked by the Vulkan loader, which must pass a valid `p_callback_data` pointer whose `p_message` is either null or a valid NUL-terminated C string. Not to be called directly.
#[cfg(debug_assertions)]
unsafe extern "system" fn vulkan_debug_callback(
message_severity: vk::DebugUtilsMessageSeverityFlagsEXT,
_message_type: vk::DebugUtilsMessageTypeFlagsEXT,

95
crates/renderer/src/lib.rs
View file

@ -26,9 +26,9 @@ use raw_window_handle::{RawDisplayHandle, RawWindowHandle};
use std::ffi::c_char;
pub use error::RendererError;
pub use renderer::Renderer;
pub use renderer::{MeshKey, Renderer};
use crate::mesh::Vertex;
use std::collections::HashMap;
impl Renderer {
/// Initializes the Vulkan renderer.
@ -42,7 +42,7 @@ impl Renderer {
///
/// # Panics
///
/// Panics if `MAX_FRAMES_IN_FLIGHT` or vertex data sizes exceed `u32`/`u64` limits.
/// Panics if `MAX_FRAMES_IN_FLIGHT` exceeds `u32`'s range.
// TODO: partial-construction leak. Each `?` below early-returns and leaks every Vulkan resource created so far; only a fully successful `new` reaches `Drop for Renderer`. Once the renderer grows more state, wrap each resource in an RAII guard so failure paths tear them down too.
pub fn new(
display_handle: RawDisplayHandle,
@ -125,11 +125,6 @@ impl Renderer {
let graphics_pipeline =
pipeline::create_graphics_pipeline(&device, pipeline_layout, swapchain_format)?;
let (vertex_buffer, vertex_allocation, index_buffer, index_allocation) =
create_geometry(&device, &mut allocator)?;
let index_count = 36;
let (depth_image, depth_allocation, depth_image_view) =
create_depth_resources(&device, &mut allocator, swapchain_extent)?;
@ -153,16 +148,12 @@ impl Renderer {
command_pool,
command_buffers,
allocator: Some(allocator),
index_buffer,
index_allocation: Some(index_allocation),
index_count,
chunk_meshes: HashMap::new(),
depth_image,
depth_allocation: Some(depth_allocation),
depth_image_view,
pipeline_layout,
graphics_pipeline,
vertex_buffer,
vertex_allocation: Some(vertex_allocation),
sync: Some(sync),
current_frame: 0,
})
@ -193,84 +184,6 @@ fn create_allocator(
Ok(allocator)
}
/// Creates the 3D geometry buffers (vertex and index) for a cube.
///
/// # Errors
///
/// Returns [`RendererError::AllocationError`] if GPU memory cannot be allocated, or [`RendererError::VulkanError`] if a buffer cannot be created.
fn create_geometry(
device: &ash::Device,
allocator: &mut Allocator,
) -> Result<(vk::Buffer, Allocation, vk::Buffer, Allocation), RendererError> {
let vertices = [
// Front face
Vertex {
position: [-0.5, -0.5, 0.5],
color: [1.0, 0.0, 0.0],
},
Vertex {
position: [0.5, -0.5, 0.5],
color: [0.0, 1.0, 0.0],
},
Vertex {
position: [0.5, 0.5, 0.5],
color: [0.0, 0.0, 1.0],
},
Vertex {
position: [-0.5, 0.5, 0.5],
color: [1.0, 1.0, 1.0],
},
// Back face
Vertex {
position: [-0.5, -0.5, -0.5],
color: [1.0, 0.0, 0.0],
},
Vertex {
position: [0.5, -0.5, -0.5],
color: [0.0, 1.0, 0.0],
},
Vertex {
position: [0.5, 0.5, -0.5],
color: [0.0, 0.0, 1.0],
},
Vertex {
position: [-0.5, 0.5, -0.5],
color: [1.0, 1.0, 1.0],
},
];
let indices: [u32; 36] = [
0, 1, 2, 2, 3, 0, // front
1, 5, 6, 6, 2, 1, // right
7, 6, 5, 5, 4, 7, // back
4, 0, 3, 3, 7, 4, // left
4, 5, 1, 1, 0, 4, // bottom
3, 2, 6, 6, 7, 3, // top
];
let (vertex_buffer, vertex_allocation) = create_gpu_buffer(
device,
allocator,
bytemuck::cast_slice(&vertices),
vk::BufferUsageFlags::VERTEX_BUFFER,
"Vertex Buffer",
)?;
let (index_buffer, index_allocation) = create_gpu_buffer(
device,
allocator,
bytemuck::cast_slice(&indices),
vk::BufferUsageFlags::INDEX_BUFFER,
"Index Buffer",
)?;
Ok((
vertex_buffer,
vertex_allocation,
index_buffer,
index_allocation,
))
}
/// Creates the depth buffer resources (image, memory, and view).
///
/// # Errors

7
crates/renderer/src/pipeline.rs
View file

@ -36,16 +36,17 @@ pub fn create_shader_module(
///
/// Returns [`RendererError::VulkanError`] if the device fails to create the pipeline layout.
pub fn create_pipeline_layout(device: &Device) -> Result<vk::PipelineLayout, RendererError> {
// A single push constant range is defined for the MVP matrix, allowing it to be updated for every draw call with high efficiency.
// The push-constant range covers the 64-byte MVP matrix followed by a 16-byte vec4 per-chunk world offset (80 bytes total, within the 128-byte guaranteed minimum).
#[expect(
clippy::expect_used,
reason = "size_of::<Mat4>() is 64 bytes, well within u32 range"
reason = "80 bytes (Mat4 + vec4) is well within u32 range"
)]
let push_constant_range = vk::PushConstantRange::default()
.stage_flags(vk::ShaderStageFlags::VERTEX)
.offset(0)
.size(
u32::try_from(std::mem::size_of::<glam::Mat4>()).expect("Mat4 size exceeds u32 range"),
u32::try_from(std::mem::size_of::<glam::Mat4>() + std::mem::size_of::<[f32; 4]>())
.expect("push-constant size exceeds u32 range"),
);
let layout_create_info = vk::PipelineLayoutCreateInfo::default()

294
crates/renderer/src/renderer.rs
View file

@ -1,10 +1,30 @@
// SPDX-License-Identifier: AGPL-3.0-only
use crate::create_gpu_buffer;
use crate::sync::SyncPrimitives;
use crate::{create_depth_resources, create_gpu_buffer, swapchain};
use crate::{error::RendererError, mesh::Vertex};
use ash::{Device, Instance, khr, vk};
use gpu_allocator::vulkan::{Allocation, Allocator};
use std::collections::HashMap;
/// Opaque, renderer-side identifier for one uploaded chunk mesh.
pub type MeshKey = (i32, i32, i32);
/// GPU resources for a single chunk mesh, drawn at a fixed world offset.
pub(crate) struct GpuMesh {
/// Buffer holding the chunk's vertex data.
pub(crate) vertex_buffer: vk::Buffer,
/// Backing allocation for [`GpuMesh::vertex_buffer`], freed when the mesh is removed.
pub(crate) vertex_allocation: Allocation,
/// Buffer holding the chunk's index data for indexed drawing.
pub(crate) index_buffer: vk::Buffer,
/// Backing allocation for [`GpuMesh::index_buffer`], freed when the mesh is removed.
pub(crate) index_allocation: Allocation,
/// Number of indices submitted in the mesh's `cmd_draw_indexed` call.
pub(crate) index_count: u32,
/// Chunk origin in world space (blocks); added to every vertex in the vertex shader.
pub(crate) world_offset: [f32; 3],
}
/// The core renderer structure holding the Vulkan resources.
pub struct Renderer {
@ -17,7 +37,6 @@ pub struct Renderer {
/// The debug messenger for validation layer output.
pub(crate) debug_messenger: vk::DebugUtilsMessengerEXT,
/// Handle to the selected physical device (GPU).
#[expect(dead_code, reason = "retained for later device-capability queries")]
pub(crate) physical_device: vk::PhysicalDevice,
/// The logical Vulkan device.
pub(crate) device: Device,
@ -40,7 +59,6 @@ pub struct Renderer {
/// Images acquired from the swapchain.
pub(crate) swapchain_images: Vec<vk::Image>,
/// The pixel format of the swapchain images.
#[expect(dead_code, reason = "retained for later swapchain recreation")]
pub(crate) swapchain_format: vk::Format,
/// The dimensions of the swapchain images.
pub(crate) swapchain_extent: vk::Extent2D,
@ -56,15 +74,8 @@ pub struct Renderer {
pub(crate) graphics_pipeline: vk::Pipeline,
/// Memory manager for GPU allocations.
pub(crate) allocator: Option<Allocator>,
/// Buffer containing the vertex data for the initial triangle.
pub(crate) vertex_buffer: vk::Buffer,
/// Memory allocation for the vertex buffer.
pub(crate) vertex_allocation: Option<Allocation>,
/// Buffer containing the index data for indexed drawing.
pub(crate) index_buffer: vk::Buffer,
/// Memory allocation for the index buffer.
pub(crate) index_allocation: Option<Allocation>,
pub(crate) index_count: u32,
/// Uploaded chunk meshes, keyed by an opaque renderer-side handle and drawn independently.
pub(crate) chunk_meshes: HashMap<MeshKey, GpuMesh>,
/// The depth image used for depth testing.
pub(crate) depth_image: vk::Image,
/// Image view for the depth buffer.
@ -92,22 +103,34 @@ impl Renderer {
let image_available_semaphore = sync.image_available[self.current_frame];
let cmd = self.command_buffers[self.current_frame];
// 1. Wait for the current frame's GPU work to finish
// 1. Wait for the current frame's GPU work to finish. The fence is intentionally not reset here: if the acquire below reports the swapchain is out of date, the frame is abandoned before any work is submitted, and a reset fence would then remain permanently unsignaled and deadlock the next wait.
unsafe {
self.device
.wait_for_fences(&[in_flight_fence], true, u64::MAX)?;
self.device.reset_fences(&[in_flight_fence])?;
}
// 2. Acquire an image from the swapchain
let (image_index, _is_suboptimal) = unsafe {
// 2. Acquire an image from the swapchain. An out-of-date swapchain (typically a window resize) is not a fatal error: the swapchain is rebuilt and this frame is skipped, to be retried on the next call.
let acquire = unsafe {
self.swapchain_loader.acquire_next_image(
self.swapchain,
u64::MAX,
image_available_semaphore,
vk::Fence::null(),
)?
)
};
let (image_index, _is_suboptimal) = match acquire {
Ok(pair) => pair,
Err(vk::Result::ERROR_OUT_OF_DATE_KHR) => {
self.recreate_swapchain(self.swapchain_extent.width, self.swapchain_extent.height)?;
return Ok(());
}
Err(e) => return Err(e.into()),
};
// The frame will now be submitted, so the fence is reset immediately before it is handed to the queue.
unsafe {
self.device.reset_fences(&[in_flight_fence])?;
}
// Use the semaphore tied to this specific swapchain image for rendering completion
let render_finished_semaphore = self
@ -149,17 +172,112 @@ impl Renderer {
.swapchains(std::slice::from_ref(&self.swapchain))
.image_indices(std::slice::from_ref(&image_index));
unsafe {
let present = unsafe {
self.swapchain_loader
.queue_present(self.graphics_queue, &present_info)?;
}
.queue_present(self.graphics_queue, &present_info)
};
// Advance the frame index for the next call
// Advance the frame index regardless of the present outcome; the submitted work is already in flight on `in_flight_fence`.
self.current_frame = (self.current_frame + 1) % crate::MAX_FRAMES_IN_FLIGHT;
// A suboptimal (`Ok(true)`) or out-of-date swapchain is rebuilt so the next frame targets a surface-matched swapchain. The rebuilt swapchain also corrects the projection aspect ratio, which is derived from the swapchain extent.
match present {
Ok(false) => {}
Ok(true) | Err(vk::Result::ERROR_OUT_OF_DATE_KHR) => {
self.recreate_swapchain(self.swapchain_extent.width, self.swapchain_extent.height)?;
}
Err(e) => return Err(e.into()),
}
Ok(())
}
/// Rebuilds the swapchain and every resource whose size derives from it, at the given surface dimensions.
///
/// This is called when the presentation surface has changed size (a window resize) or when Vulkan reports the swapchain is out of date. The device is drained first so no in-flight work references the resources being freed. The projection aspect ratio is derived from [`Self::swapchain_extent`], so rebuilding the swapchain at the new extent corrects a stretched or squashed image for free.
///
/// A zero-area surface (a minimized window) is a no-op: a swapchain cannot be created with a zero extent, so the previous resources are retained until a non-zero size is reported.
///
/// On platforms where the surface reports a definitive `current_extent` (typically X11), `width` and `height` are ignored in favour of that value; they are used as the fallback size only where the surface defers to the application (typically Wayland).
///
/// # Errors
///
/// Returns [`RendererError::AllocatorMissing`] if the GPU allocator has been released, or a [`RendererError`] propagated from swapchain, image-view, or depth-resource creation.
///
/// # Panics
///
/// Panics if the driver reports zero surface formats, which the Vulkan specification forbids for a supported surface.
pub fn recreate_swapchain(&mut self, width: u32, height: u32) -> Result<(), RendererError> {
// A zero extent cannot back a swapchain; defer the rebuild until the surface has area again.
if width == 0 || height == 0 {
return Ok(());
}
// The old resources may still be referenced by in-flight frames; draining the device guarantees they are free to destroy.
unsafe {
self.device.device_wait_idle()?;
}
self.destroy_swapchain_resources();
let (swapchain_loader, swapchain, swapchain_images, swapchain_format, swapchain_extent) =
swapchain::create_swapchain(
&self.instance,
self.physical_device,
&self.device,
&self.surface_loader,
self.surface,
width,
height,
)?;
let swapchain_image_views =
swapchain::create_image_views(&self.device, &swapchain_images, swapchain_format)?;
let allocator = self
.allocator
.as_mut()
.ok_or(RendererError::AllocatorMissing)?;
let (depth_image, depth_allocation, depth_image_view) =
create_depth_resources(&self.device, allocator, swapchain_extent)?;
self.swapchain_loader = swapchain_loader;
self.swapchain = swapchain;
self.swapchain_images = swapchain_images;
self.swapchain_format = swapchain_format;
self.swapchain_extent = swapchain_extent;
self.swapchain_image_views = swapchain_image_views;
self.depth_image = depth_image;
self.depth_allocation = Some(depth_allocation);
self.depth_image_view = depth_image_view;
Ok(())
}
/// Destroys the swapchain and every size-dependent resource derived from it (image views and depth buffer), leaving the fields holding stale handles until the caller overwrites them.
///
/// The device must already be idle; callers are responsible for that ordering. Only invoked from [`Self::recreate_swapchain`], which drains the device and immediately replaces every field this touches.
fn destroy_swapchain_resources(&mut self) {
unsafe {
self.device.destroy_image_view(self.depth_image_view, None);
self.device.destroy_image(self.depth_image, None);
if let Some(allocator) = self.allocator.as_mut()
&& let Some(alloc) = self.depth_allocation.take()
&& let Err(e) = allocator.free(alloc)
{
tracing::error!("Failed to free depth image allocation: {e}");
}
// Image views are destroyed before the swapchain that owns their underlying images.
for &view in &self.swapchain_image_views {
self.device.destroy_image_view(view, None);
}
self.swapchain_image_views.clear();
self.swapchain_loader
.destroy_swapchain(self.swapchain, None);
}
}
/// Records the drawing commands into the given command buffer.
///
/// # Errors
@ -293,11 +411,6 @@ impl Renderer {
};
self.device.cmd_set_scissor(cmd, 0, &[scissor]);
self.device
.cmd_bind_vertex_buffers(cmd, 0, &[self.vertex_buffer], &[0]);
self.device
.cmd_bind_index_buffer(cmd, self.index_buffer, 0, vk::IndexType::UINT32);
let aspect =
f64::from(self.swapchain_extent.width) / f64::from(self.swapchain_extent.height);
@ -315,6 +428,7 @@ impl Renderer {
// The view matrix is supplied by the caller (the client's camera); the renderer owns only the projection, which depends on the swapchain aspect ratio it manages.
let mvp = projection * camera_view;
// The MVP is identical for every chunk this frame, so it is pushed once before the loop.
let mvp_bytes = bytemuck::cast_slice(mvp.as_ref());
self.device.cmd_push_constants(
cmd,
@ -324,8 +438,36 @@ impl Renderer {
mvp_bytes,
);
self.device
.cmd_draw_indexed(cmd, self.index_count, 1, 0, 0, 0);
// The per-chunk offset occupies the push-constant range immediately after the 64-byte MVP.
#[expect(
clippy::cast_possible_truncation,
reason = "size_of::<Mat4>() is 64 bytes, well within u32 range"
)]
let chunk_offset_byte = size_of::<glam::Mat4>() as u32;
for mesh in self.chunk_meshes.values() {
// The offset is padded to a vec4 to match the std140 layout of the push-constant block; only xyz is read by the shader.
let offset = [
mesh.world_offset[0],
mesh.world_offset[1],
mesh.world_offset[2],
0.0_f32,
];
self.device.cmd_push_constants(
cmd,
self.pipeline_layout,
vk::ShaderStageFlags::VERTEX,
chunk_offset_byte,
bytemuck::cast_slice(&offset),
);
self.device
.cmd_bind_vertex_buffers(cmd, 0, &[mesh.vertex_buffer], &[0]);
self.device
.cmd_bind_index_buffer(cmd, mesh.index_buffer, 0, vk::IndexType::UINT32);
self.device
.cmd_draw_indexed(cmd, mesh.index_count, 1, 0, 0, 0);
}
}
}
@ -368,47 +510,36 @@ impl Renderer {
Ok(())
}
/// Replaces the currently rendering mesh with a new set of vertices and indices.
/// Uploads (or replaces) the mesh stored under `key`, positioned at `world_offset` (in blocks).
///
/// If a mesh already exists under `key`, its GPU resources are freed before the replacement is
/// uploaded.
///
/// # Errors
///
/// Returns [`RendererError::AllocationError`] if GPU memory cannot be allocated, or [`RendererError::VulkanError`] if the vertex or index buffers cannot be created.
/// Returns [`RendererError::AllocatorMissing`] if the GPU allocator has been torn down,
/// [`RendererError::AllocationError`] if GPU memory cannot be allocated, or
/// [`RendererError::VulkanError`] if the vertex or index buffers cannot be created.
#[expect(
clippy::cast_possible_truncation,
reason = "a chunk mesh's index count never approaches u32::MAX"
)]
pub fn update_mesh(
pub fn insert_mesh(
&mut self,
key: MeshKey,
vertices: &[Vertex],
indices: &[u32],
world_offset: [f32; 3],
) -> Result<(), RendererError> {
unsafe {
let _ = self.device.device_wait_idle();
let allocator = self
.allocator
.as_mut()
.ok_or(RendererError::AllocatorMissing)?;
if let Some(alloc) = self.vertex_allocation.take() {
let _ = allocator.free(alloc);
}
self.device.destroy_buffer(self.vertex_buffer, None);
self.vertex_buffer = vk::Buffer::null();
if let Some(alloc) = self.index_allocation.take() {
let _ = allocator.free(alloc);
}
self.device.destroy_buffer(self.index_buffer, None);
self.index_buffer = vk::Buffer::null();
}
// Free any mesh already stored under this key before uploading its replacement.
self.remove_mesh(key);
let allocator = self
.allocator
.as_mut()
.ok_or(RendererError::AllocatorMissing)?;
let (v_buf, v_alloc) = create_gpu_buffer(
let (vertex_buffer, vertex_allocation) = create_gpu_buffer(
&self.device,
allocator,
bytemuck::cast_slice(vertices),
@ -416,7 +547,7 @@ impl Renderer {
"Chunk Vertex Buffer",
)?;
let (i_buf, i_alloc) = crate::create_gpu_buffer(
let (index_buffer, index_allocation) = create_gpu_buffer(
&self.device,
allocator,
bytemuck::cast_slice(indices),
@ -424,14 +555,39 @@ impl Renderer {
"Chunk Index Buffer",
)?;
self.vertex_buffer = v_buf;
self.vertex_allocation = Some(v_alloc);
self.index_buffer = i_buf;
self.index_allocation = Some(i_alloc);
self.index_count = indices.len() as u32;
self.chunk_meshes.insert(
key,
GpuMesh {
vertex_buffer,
vertex_allocation,
index_buffer,
index_allocation,
index_count: indices.len() as u32,
world_offset,
},
);
Ok(())
}
/// Frees the GPU mesh stored under `key`. Does nothing if no mesh is present.
pub fn remove_mesh(&mut self, key: MeshKey) {
let Some(mesh) = self.chunk_meshes.remove(&key) else {
return;
};
unsafe {
// Waiting idle per removal is the simple, always-correct approach; in a bulk load/unload loop it serialises the GPU, so a single wait around the loop is preferable if this ever shows up as a measured bottleneck.
let _ = self.device.device_wait_idle();
if let Some(allocator) = self.allocator.as_mut() {
let _ = allocator.free(mesh.vertex_allocation);
let _ = allocator.free(mesh.index_allocation);
}
self.device.destroy_buffer(mesh.vertex_buffer, None);
self.device.destroy_buffer(mesh.index_buffer, None);
}
}
}
impl Drop for Renderer {
@ -443,16 +599,18 @@ impl Drop for Renderer {
self.device
.destroy_pipeline_layout(self.pipeline_layout, None);
// Drain the chunk meshes so each owned allocation can be freed and its buffers destroyed.
let meshes: Vec<GpuMesh> = self.chunk_meshes.drain().map(|(_, mesh)| mesh).collect();
if let Some(allocator) = self.allocator.as_mut() {
if let Some(alloc) = self.vertex_allocation.take()
&& let Err(e) = allocator.free(alloc)
{
tracing::error!("Failed to free vertex buffer allocation: {e}");
}
if let Some(alloc) = self.index_allocation.take()
&& let Err(e) = allocator.free(alloc)
{
tracing::error!("Failed to free index buffer allocation: {e}");
for mesh in meshes {
if let Err(e) = allocator.free(mesh.vertex_allocation) {
tracing::error!("Failed to free chunk vertex allocation: {e}");
}
if let Err(e) = allocator.free(mesh.index_allocation) {
tracing::error!("Failed to free chunk index allocation: {e}");
}
self.device.destroy_buffer(mesh.vertex_buffer, None);
self.device.destroy_buffer(mesh.index_buffer, None);
}
if let Some(alloc) = self.depth_allocation.take()
&& let Err(e) = allocator.free(alloc)
@ -460,8 +618,6 @@ impl Drop for Renderer {
tracing::error!("Failed to free depth image allocation: {e}");
}
}
self.device.destroy_buffer(self.vertex_buffer, None);
self.device.destroy_buffer(self.index_buffer, None);
self.device.destroy_image_view(self.depth_image_view, None);
self.device.destroy_image(self.depth_image, None);

120
crates/server/src/client_stream.rs Normal file
View file

@ -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;

58
crates/server/src/main.rs
View file

@ -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.

16
crates/server/src/save/region_actor.rs
View file

@ -73,7 +73,7 @@ impl SaveActor {
/// The actor's run loop: it owns the region-file map and answers requests until every sender is dropped.
fn actor_loop(region_dir: &Path, request_rx: &Receiver<SaveRequest>) {
// The actor is the sole owner of this map, so region files need no lock of their own.
let mut regions: HashMap<(i32, i32), RegionFile> = HashMap::new();
let mut regions: HashMap<(i32, i32, i32), RegionFile> = HashMap::new();
while let Ok(request) = request_rx.recv() {
match request {
SaveRequest::Read { pos, reply } => {
@ -112,7 +112,7 @@ fn actor_loop(region_dir: &Path, request_rx: &Receiver<SaveRequest>) {
/// # Errors
///
/// Returns the first [`SaveError`] produced by [`RegionFile::save`]; remaining dirty regions are still flushed.
fn flush_dirty(regions: &mut HashMap<(i32, i32), RegionFile>) -> Result<(), SaveError> {
fn flush_dirty(regions: &mut HashMap<(i32, i32, i32), RegionFile>) -> Result<(), SaveError> {
let mut result = Ok(());
for region in regions.values_mut() {
// Clean regions are skipped so a flush never rewrites an unchanged file.
@ -136,17 +136,17 @@ fn flush_dirty(regions: &mut HashMap<(i32, i32), RegionFile>) -> Result<(), Save
///
/// Returns a [`SaveError`] from [`RegionFile::open`] if the region file exists but cannot be read or decoded.
fn region_mut<'a>(
regions: &'a mut HashMap<(i32, i32), RegionFile>,
regions: &'a mut HashMap<(i32, i32, i32), RegionFile>,
region_dir: &Path,
pos: ChunkPos,
) -> Result<&'a mut RegionFile, SaveError> {
let key = region_coords(pos.x, pos.z);
let key = region_coords(pos.x, pos.y, pos.z);
// The region file is opened once on first touch; every later access hits the in-memory copy.
match regions.entry(key) {
Entry::Occupied(entry) => Ok(entry.into_mut()),
Entry::Vacant(entry) => {
Ok(entry.insert(RegionFile::open(region_path(region_dir, pos.x, pos.z))?))
}
Entry::Vacant(entry) => Ok(entry.insert(RegionFile::open(region_path(
region_dir, pos.x, pos.y, pos.z,
))?)),
}
}
@ -156,7 +156,7 @@ fn region_mut<'a>(
///
/// Returns a [`SaveError`] if the region file cannot be opened or the stored record cannot be decoded.
fn read_chunk(
regions: &mut HashMap<(i32, i32), RegionFile>,
regions: &mut HashMap<(i32, i32, i32), RegionFile>,
region_dir: &Path,
pos: ChunkPos,
) -> Result<Option<ChunkData>, SaveError> {

22
crates/server/src/save/region_file.rs
View file

@ -13,20 +13,26 @@ use shared::save::record;
use shared::save::region::{HeaderEntry, RegionIndex};
use shared::world::{ChunkData, ChunkPos};
/// The side length, in chunk columns, of the square footprint one region file covers.
/// The side length, in chunks, of the cube one region file covers on every axis.
pub const REGION_SIZE: i32 = 32;
/// Maps a chunk column `(cx, cz)` to the coordinates `(rx, rz)` of the region that contains it.
/// Maps a chunk `(cx, cy, cz)` to the coordinates `(rx, ry, rz)` of the region cube that contains it.
///
/// Every axis is floored via `div_euclid` (not truncating division) so negative chunk coordinates map to the region below rather than toward zero: chunk `-1` belongs to region `-1`, not region `0`.
#[must_use]
pub fn region_coords(cx: i32, cz: i32) -> (i32, i32) {
(cx.div_euclid(REGION_SIZE), cz.div_euclid(REGION_SIZE))
pub fn region_coords(cx: i32, cy: i32, cz: i32) -> (i32, i32, i32) {
(
cx.div_euclid(REGION_SIZE),
cy.div_euclid(REGION_SIZE),
cz.div_euclid(REGION_SIZE),
)
}
/// Builds the on-disk path of the region file containing chunk column `(cx, cz)` within `dir`.
/// Builds the on-disk path of the region file containing chunk `(cx, cy, cz)` within `dir`.
#[must_use]
pub fn region_path(dir: &Path, cx: i32, cz: i32) -> PathBuf {
let (rx, rz) = region_coords(cx, cz);
dir.join(format!("r.{rx}.{rz}.region"))
pub fn region_path(dir: &Path, cx: i32, cy: i32, cz: i32) -> PathBuf {
let (rx, ry, rz) = region_coords(cx, cy, cz);
dir.join(format!("r.{rx}.{ry}.{rz}.region"))
}
/// An open region file: its `SYNR` index, the resident chunk records, and its on-disk location.

85
crates/server/src/tests/client_stream.rs Normal file
View file

@ -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"
);
}

28
crates/server/src/tests/region_file.rs
View file

@ -13,27 +13,27 @@ fn sample(pos: ChunkPos) -> ChunkData {
}
#[test]
fn region_coords_floor_negative_columns() {
// Truncating division would map -1 to region 0; Euclidean flooring maps it to region -1.
assert_eq!(region_coords(0, 0), (0, 0));
assert_eq!(region_coords(31, 31), (0, 0));
assert_eq!(region_coords(-1, -1), (-1, -1));
assert_eq!(region_coords(-32, -33), (-1, -2));
fn region_coords_floor_negative_chunks() {
// Truncating division would map -1 to region 0; Euclidean flooring maps it to region -1. Every axis, including Y, floors identically under the 3D region grid.
assert_eq!(region_coords(0, 0, 0), (0, 0, 0));
assert_eq!(region_coords(31, 31, 31), (0, 0, 0));
assert_eq!(region_coords(-1, -1, -1), (-1, -1, -1));
assert_eq!(region_coords(-32, -33, 64), (-1, -2, 2));
}
#[test]
fn region_path_names_the_region_file() {
let dir = Path::new("/saves/world/region");
assert_eq!(
region_path(dir, -1, 5),
Path::new("/saves/world/region/r.-1.0.region")
region_path(dir, -1, 40, 5),
Path::new("/saves/world/region/r.-1.1.0.region")
);
}
#[test]
fn open_missing_file_is_empty() -> Result<(), SaveError> {
let dir = tempfile::tempdir()?;
let region = RegionFile::open(dir.path().join("r.0.0.region"))?;
let region = RegionFile::open(dir.path().join("r.0.0.0.region"))?;
assert!(region.is_empty());
assert_eq!(region.read_chunk(ChunkPos::new(0, 0, 0))?, None);
Ok(())
@ -42,7 +42,7 @@ fn open_missing_file_is_empty() -> Result<(), SaveError> {
#[test]
fn round_trips_chunks_through_disk() -> Result<(), SaveError> {
let dir = tempfile::tempdir()?;
let path = dir.path().join("r.0.0.region");
let path = dir.path().join("r.0.0.0.region");
let positions = [
ChunkPos::new(0, 0, 0),
@ -72,7 +72,7 @@ fn round_trips_chunks_through_disk() -> Result<(), SaveError> {
#[test]
fn record_offsets_are_valid_and_contiguous() -> Result<(), SaveError> {
let dir = tempfile::tempdir()?;
let path = dir.path().join("r.0.0.region");
let path = dir.path().join("r.0.0.0.region");
let mut region = RegionFile::open(path.clone())?;
for pos in [
@ -99,7 +99,7 @@ fn record_offsets_are_valid_and_contiguous() -> Result<(), SaveError> {
#[test]
fn remove_drops_only_the_named_chunk() -> Result<(), SaveError> {
let dir = tempfile::tempdir()?;
let path = dir.path().join("r.0.0.region");
let path = dir.path().join("r.0.0.0.region");
let kept = ChunkPos::new(0, 0, 0);
let dropped = ChunkPos::new(1, 1, 1);
@ -121,7 +121,7 @@ fn remove_drops_only_the_named_chunk() -> Result<(), SaveError> {
#[test]
fn stray_tmp_file_does_not_corrupt_reads() -> Result<(), SaveError> {
let dir = tempfile::tempdir()?;
let path = dir.path().join("r.0.0.region");
let path = dir.path().join("r.0.0.0.region");
let pos = ChunkPos::new(0, 0, 0);
let mut region = RegionFile::open(path.clone())?;
@ -129,7 +129,7 @@ fn stray_tmp_file_does_not_corrupt_reads() -> Result<(), SaveError> {
region.save()?;
// A leftover .tmp from an interrupted save must be ignored: only the renamed target is read.
fs::write(dir.path().join("r.0.0.region.tmp"), b"garbage")?;
fs::write(dir.path().join("r.0.0.0.region.tmp"), b"garbage")?;
let reopened = RegionFile::open(path)?;
assert_eq!(reopened.read_chunk(pos)?, Some(sample(pos)));
Ok(())

4
crates/server/src/tests/world_server.rs
View file

@ -108,7 +108,7 @@ fn saved_modification_is_applied_over_baseline() -> Result<(), SaveError> {
let mut data = ChunkData::new(pos, 0);
data.set(edited_index, edited_block);
let mut region = RegionFile::open(region_path(dir.path(), pos.x, pos.z))?;
let mut region = RegionFile::open(region_path(dir.path(), pos.x, pos.y, pos.z))?;
region.write_chunk(pos, &data, 0)?;
region.save()?;
@ -180,7 +180,7 @@ fn clean_chunk_is_not_written_back_on_eviction() -> Result<(), SaveError> {
flush(&world)?;
// No record may exist for a chunk that never diverged from its baseline.
let region = RegionFile::open(region_path(dir.path(), pos.x, pos.z))?;
let region = RegionFile::open(region_path(dir.path(), pos.x, pos.y, pos.z))?;
assert!(region.read_chunk(pos)?.is_none());
Ok(())
}

205
crates/shared/src/protocol.rs
View file

@ -1,198 +1,13 @@
// SPDX-License-Identifier: AGPL-3.0-only
//! Network protocol types and constants.
//!
//! The module is split by stream purpose: [`control`]-stream handshake and disconnect messages, and the [`chunk`]-sync request/delivery messages. Control-stream types are re-exported here so callers continue to refer to `shared::protocol::<Type>` regardless of the internal layout, while the chunk types stay namespaced under `shared::protocol::chunk` to keep the two protocols visually distinct.
use serde::{Deserialize, Serialize};
pub mod chunk;
mod control;
/// Wire-protocol version. Incremented on any breaking change to the message layout below.
pub const PROTOCOL_VERSION: u32 = 1;
/// Messages carried on the control stream (stream 0): handshake and disconnect.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub enum ControlMessage {
/// First message a client sends after the QUIC/TLS handshake.
ClientHello(ClientHello),
/// Server acceptance carrying negotiated session parameters.
HandshakeAck(HandshakeAck),
/// Server refusal with a machine-readable reason.
HandshakeReject(HandshakeReject),
/// Orderly session teardown initiated by either side.
Disconnect(Disconnect),
}
/// First message a client sends after the QUIC/TLS handshake.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct ClientHello {
/// Protocol version the client was built against; compared to `PROTOCOL_VERSION`.
pub protocol_version: u32,
/// Human-readable client build string (e.g. crate version + git hash).
pub client_build: String,
/// Identity the player presents. Minimal for M1.
pub player_identity: PlayerIdentity,
/// Content packs the client has installed. Empty in M1; validated later.
pub installed_packs: Vec<PackRef>,
/// Optional protocol feature bits the client requests. Zero in M1.
pub requested_features: FeatureFlags,
}
/// Server acceptance carrying negotiated session parameters.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct HandshakeAck {
/// Server's protocol version (equal to the client's on success).
pub protocol_version: u32,
/// Human-readable server build string.
pub server_build: String,
/// Packs the world requires, each with an optional download source. May include `PackTier::Resource` entries (a server resource pack), which are delivered one-way and applied client-side rather than strict-matched; a consumer must branch on tier (or `PackTier::requires_strict_match`) before treating an entry as a match requirement.
pub world_packs: Vec<RequiredPack>,
/// Packs the client is missing relative to the server, each with an optional download source. As with `world_packs`, `PackTier::Resource` entries are delivered, not matched.
pub missing_packs: Vec<RequiredPack>,
/// Which stream carries which purpose for this session.
pub stream_layout: StreamLayout,
/// Advisory server tick rate in Hz, for client clock setup.
pub tick_rate_hint: u16,
}
/// Server refusal with a machine-readable reason.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct HandshakeReject {
/// Machine-readable rejection category.
pub reason: RejectReason,
/// Human-readable detail for logs and UI.
pub detail: String,
/// Optional URL directing the user to a compatible build or pack, when the rejection is recoverable (e.g. `ProtocolMismatch`, `PackMismatch`).
pub upgrade_url: Option<String>,
}
/// Orderly session teardown initiated by either side.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct Disconnect {
/// Human-readable reason shown to the peer and logged.
pub reason: String,
}
/// Machine-readable categories for handshake rejection.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum RejectReason {
/// Client protocol version does not match the server's.
ProtocolMismatch,
/// Client is missing required packs or has incompatible versions.
PackMismatch,
/// Client declined or failed to fetch a server resource pack the server marked required.
ResourcePackDeclined,
/// Client failed to authenticate.
AuthFailed,
/// Client is banned from the server.
Banned,
/// Server is full.
Full,
/// Server encountered an internal error during handshake.
ServerError,
}
/// Identity presented by the player to the server.
// TODO: use authenticated identity once the Account system exists.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct PlayerIdentity {
/// Human-readable display name.
pub display_name: String,
}
/// Reference to a content pack (resource pack, data pack, or Lua mod) as it appears in a modlist exchanged during the handshake.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct PackRef {
/// Namespaced content identifier of the pack (`namespace:id`). Charset validation is deferred to the modlist-matching concept (out of M1 scope).
pub id: String,
/// Human-readable semantic version. Informational only; not the match key.
pub version: String,
/// Canonical hash of the pack contents; the authoritative match key.
// TODO: pin the canonical hashing procedure (traversal order, newline normalization) so independent builds of one pack hash identically.
pub content_hash: [u8; 32],
/// Tier the pack was classified into, which governs whether a client/server mismatch on this pack is fatal or tolerated. Inferred by the owner from the pack's folder contents (see Load order), never self-declared.
pub tier: PackTier,
}
/// Classification of a content pack, determining the handshake matching rule applied to it. Inferred from folder contents, not self-declared: `assets/`-only is a resource pack, `data/`-only is a data pack, presence of `scripts/` is a Lua mod.
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
pub enum PackTier {
/// Client-side asset overlay (`assets/` only). Never strict-matched between peers. A server may push one server resource pack of its own, delivered one-way and applied on top of the client's local pack stack; enforcement of a `required` server pack is apply-or-reject at the client, not a peer hash-match.
Resource,
/// Declarative content (`data/` only). Must match exactly between peers.
Data,
/// Lua mod (`scripts/`, optionally `data/` and `assets/`); full API access.
Mod {
/// Set when the mod ships no `data/` and every system is `scope = "client"`, so a client/server mismatch on it cannot desync authoritative state and is therefore tolerated. Not trusted blindly by the server for packs carrying data or server-scoped systems.
client_only: bool,
},
}
impl PackTier {
/// Returns whether a pack of this tier must match byte-for-byte between client and server for the connection to be accepted. Resource packs are never matched; data packs and non-`client_only` mods must match exactly. A `false` here does not imply the server never sends the pack, a server resource pack is delivered one-way despite not being part of bidirectional matching.
#[must_use]
pub fn requires_strict_match(self) -> bool {
match self {
PackTier::Resource => false,
PackTier::Data => true,
PackTier::Mod { client_only } => !client_only,
}
}
}
/// A pack the server's world requires, paired with an optional out-of-band download source. Sent server → client in the handshake; the client fetches any it lacks via the URL when present, otherwise over the QUIC asset stream.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct RequiredPack {
/// Identity and tier of the required pack.
pub pack: PackRef,
/// Optional HTTP(S) URL to fetch the pack from, bypassing the QUIC asset stream for large downloads. `None` means fetch over the asset stream.
pub download_url: Option<String>,
/// Whether the connection is rejected if the client cannot obtain and apply this pack. For data/mod tiers this is always `true` (they are mandatory for a correct session). For a `PackTier::Resource` entry (a server resource pack) it distinguishes an *optional* overlay the client may decline and keep playing (`false`) from a *required* one whose decline or fetch failure rejects the connection (`true`).
pub required: bool,
}
/// Optional protocol feature bits.
#[repr(transparent)]
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)]
pub struct FeatureFlags(pub u32);
/// Mapping of logical purposes to QUIC stream IDs.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct StreamLayout {
/// Stream ID for control messages (handshake, disconnect).
pub control: u8,
/// Stream ID for client input to server.
pub input: u8,
/// Stream ID for server authoritative state updates.
pub authority: u8,
/// Stream ID for highest detail chunk updates (LOD0).
pub chunk_lod0: u8,
/// Stream ID for chunk updates (LOD1).
pub chunk_lod1: u8,
/// Stream ID for chunk updates (LOD2).
pub chunk_lod2: u8,
/// Stream ID for chunk updates (LOD3).
pub chunk_lod3: u8,
/// Stream ID for lowest detail chunk updates (LOD4).
pub chunk_lod4: u8,
/// Stream ID for downloading assets.
pub asset: u8,
/// Stream ID for downloading mod scripts.
pub mod_data: u8,
}
impl Default for StreamLayout {
fn default() -> Self {
Self {
control: 0,
input: 1,
authority: 2,
chunk_lod0: 3,
chunk_lod1: 4,
chunk_lod2: 5,
chunk_lod3: 6,
chunk_lod4: 7,
asset: 8,
mod_data: 9,
}
}
}
#[cfg(test)]
#[path = "tests/protocol.rs"]
mod tests;
pub use control::{
ClientHello, ControlMessage, Disconnect, FeatureFlags, HandshakeAck, HandshakeReject,
PROTOCOL_VERSION, PackRef, PackTier, PlayerIdentity, RejectReason, RequiredPack, StreamLayout,
};

38
crates/shared/src/protocol/chunk.rs Normal file
View file

@ -0,0 +1,38 @@
// SPDX-License-Identifier: AGPL-3.0-only
//! Chunk-sync message types carried between server and client.
use crate::world::{ChunkData, ChunkPos};
use serde::{Deserialize, Serialize};
/// A client's request for the chunks it wants resident, expressed as a center and radius.
// TODO: per-chunk request/ack + flow control for the fuller protocol.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct ChunkSubscribe {
/// Chunk-space center the client wants chunks around (derived from its camera/player).
pub center: ChunkPos,
/// Load radius in chunks. The server clamps this to a server-side maximum.
pub radius: u16,
}
/// One chunk delivered to the client.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub enum ChunkMessage {
/// Full chunk payload at a position.
// TODO: LOD tier selection so a chunk can be delivered at a coarser stream tier.
Chunk {
/// Position of the delivered chunk.
pos: ChunkPos,
/// Serializable chunk contents (reuses the save/diff representation).
data: ChunkData,
},
/// The server has dropped this chunk from the client's set; the client should unload it. This is the server-authoritative counterpart to the client's own radius-based unload: the server can force a discard even when the chunk is still within the client's radius.
Drop {
/// Position the client should discard.
pos: ChunkPos,
},
}
#[cfg(test)]
#[path = "../tests/protocol_chunk.rs"]
mod tests;

200
crates/shared/src/protocol/control.rs Normal file
View file

@ -0,0 +1,200 @@
// SPDX-License-Identifier: AGPL-3.0-only
//! Control-stream messages: handshake negotiation and orderly disconnect.
use serde::{Deserialize, Serialize};
/// Wire-protocol version. Incremented on any breaking change to the message layout below.
pub const PROTOCOL_VERSION: u32 = 1;
/// Messages carried on the control stream (stream 0): handshake and disconnect.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub enum ControlMessage {
/// First message a client sends after the QUIC/TLS handshake.
ClientHello(ClientHello),
/// Server acceptance carrying negotiated session parameters.
HandshakeAck(HandshakeAck),
/// Server refusal with a machine-readable reason.
HandshakeReject(HandshakeReject),
/// Orderly session teardown initiated by either side.
Disconnect(Disconnect),
}
/// First message a client sends after the QUIC/TLS handshake.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct ClientHello {
/// Protocol version the client was built against; compared to `PROTOCOL_VERSION`.
pub protocol_version: u32,
/// Human-readable client build string (e.g. crate version + git hash).
pub client_build: String,
/// Identity the player presents. Minimal for M1.
pub player_identity: PlayerIdentity,
/// Content packs the client has installed. Empty in M1; validated later.
pub installed_packs: Vec<PackRef>,
/// Optional protocol feature bits the client requests. Zero in M1.
pub requested_features: FeatureFlags,
}
/// Server acceptance carrying negotiated session parameters.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct HandshakeAck {
/// Server's protocol version (equal to the client's on success).
pub protocol_version: u32,
/// Human-readable server build string.
pub server_build: String,
/// Packs the world requires, each with an optional download source. May include `PackTier::Resource` entries (a server resource pack), which are delivered one-way and applied client-side rather than strict-matched; a consumer must branch on tier (or `PackTier::requires_strict_match`) before treating an entry as a match requirement.
pub world_packs: Vec<RequiredPack>,
/// Packs the client is missing relative to the server, each with an optional download source. As with `world_packs`, `PackTier::Resource` entries are delivered, not matched.
pub missing_packs: Vec<RequiredPack>,
/// Which stream carries which purpose for this session.
pub stream_layout: StreamLayout,
/// Advisory server tick rate in Hz, for client clock setup.
pub tick_rate_hint: u16,
}
/// Server refusal with a machine-readable reason.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct HandshakeReject {
/// Machine-readable rejection category.
pub reason: RejectReason,
/// Human-readable detail for logs and UI.
pub detail: String,
/// Optional URL directing the user to a compatible build or pack, when the rejection is recoverable (e.g. `ProtocolMismatch`, `PackMismatch`).
pub upgrade_url: Option<String>,
}
/// Orderly session teardown initiated by either side.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct Disconnect {
/// Human-readable reason shown to the peer and logged.
pub reason: String,
}
/// Machine-readable categories for handshake rejection.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum RejectReason {
/// Client protocol version does not match the server's.
ProtocolMismatch,
/// Client is missing required packs or has incompatible versions.
PackMismatch,
/// Client declined or failed to fetch a server resource pack the server marked required.
ResourcePackDeclined,
/// Client failed to authenticate.
AuthFailed,
/// Client is banned from the server.
Banned,
/// Server is full.
Full,
/// Server encountered an internal error during handshake.
ServerError,
}
/// Identity presented by the player to the server.
// TODO: use authenticated identity once the Account system exists.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct PlayerIdentity {
/// Human-readable display name.
pub display_name: String,
}
/// Reference to a content pack (resource pack, data pack, or Lua mod) as it appears in a modlist exchanged during the handshake.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct PackRef {
/// Namespaced content identifier of the pack (`namespace:id`). Charset validation is deferred to the modlist-matching concept (out of M1 scope).
pub id: String,
/// Human-readable semantic version. Informational only; not the match key.
pub version: String,
/// Canonical hash of the pack contents; the authoritative match key.
// TODO: pin the canonical hashing procedure (traversal order, newline normalization) so independent builds of one pack hash identically.
pub content_hash: [u8; 32],
/// Tier the pack was classified into, which governs whether a client/server mismatch on this pack is fatal or tolerated. Inferred by the owner from the pack's folder contents (see Load order), never self-declared.
pub tier: PackTier,
}
/// Classification of a content pack, determining the handshake matching rule applied to it. Inferred from folder contents, not self-declared: `assets/`-only is a resource pack, `data/`-only is a data pack, presence of `scripts/` is a Lua mod.
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
pub enum PackTier {
/// Client-side asset overlay (`assets/` only). Never strict-matched between peers. A server may push one server resource pack of its own, delivered one-way and applied on top of the client's local pack stack; enforcement of a `required` server pack is apply-or-reject at the client, not a peer hash-match.
Resource,
/// Declarative content (`data/` only). Must match exactly between peers.
Data,
/// Lua mod (`scripts/`, optionally `data/` and `assets/`); full API access.
Mod {
/// Set when the mod ships no `data/` and every system is `scope = "client"`, so a client/server mismatch on it cannot desync authoritative state and is therefore tolerated. Not trusted blindly by the server for packs carrying data or server-scoped systems.
client_only: bool,
},
}
impl PackTier {
/// Returns whether a pack of this tier must match byte-for-byte between client and server for the connection to be accepted. Resource packs are never matched; data packs and non-`client_only` mods must match exactly. A `false` here does not imply the server never sends the pack, a server resource pack is delivered one-way despite not being part of bidirectional matching.
#[must_use]
pub fn requires_strict_match(self) -> bool {
match self {
PackTier::Resource => false,
PackTier::Data => true,
PackTier::Mod { client_only } => !client_only,
}
}
}
/// A pack the server's world requires, paired with an optional out-of-band download source. Sent server → client in the handshake; the client fetches any it lacks via the URL when present, otherwise over the QUIC asset stream.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct RequiredPack {
/// Identity and tier of the required pack.
pub pack: PackRef,
/// Optional HTTP(S) URL to fetch the pack from, bypassing the QUIC asset stream for large downloads. `None` means fetch over the asset stream.
pub download_url: Option<String>,
/// Whether the connection is rejected if the client cannot obtain and apply this pack. For data/mod tiers this is always `true` (they are mandatory for a correct session). For a `PackTier::Resource` entry (a server resource pack) it distinguishes an *optional* overlay the client may decline and keep playing (`false`) from a *required* one whose decline or fetch failure rejects the connection (`true`).
pub required: bool,
}
/// Optional protocol feature bits.
#[repr(transparent)]
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)]
pub struct FeatureFlags(pub u32);
/// Mapping of logical purposes to QUIC stream IDs.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct StreamLayout {
/// Stream ID for control messages (handshake, disconnect).
pub control: u8,
/// Stream ID for client input to server.
pub input: u8,
/// Stream ID for server authoritative state updates.
pub authority: u8,
/// Stream ID for highest detail chunk updates (LOD0).
pub chunk_lod0: u8,
/// Stream ID for chunk updates (LOD1).
pub chunk_lod1: u8,
/// Stream ID for chunk updates (LOD2).
pub chunk_lod2: u8,
/// Stream ID for chunk updates (LOD3).
pub chunk_lod3: u8,
/// Stream ID for lowest detail chunk updates (LOD4).
pub chunk_lod4: u8,
/// Stream ID for downloading assets.
pub asset: u8,
/// Stream ID for downloading mod scripts.
pub mod_data: u8,
}
impl Default for StreamLayout {
fn default() -> Self {
Self {
control: 0,
input: 1,
authority: 2,
chunk_lod0: 3,
chunk_lod1: 4,
chunk_lod2: 5,
chunk_lod3: 6,
chunk_lod4: 7,
asset: 8,
mod_data: 9,
}
}
}
#[cfg(test)]
#[path = "../tests/protocol.rs"]
mod tests;

51
crates/shared/src/tests/protocol_chunk.rs Normal file
View file

@ -0,0 +1,51 @@
// SPDX-License-Identifier: AGPL-3.0-only
use super::*;
use crate::world::BlockId;
/// Round-trips a `ChunkSubscribe` through postcard and asserts the decoded value is identical.
fn roundtrip_subscribe(msg: &ChunkSubscribe) -> Result<(), postcard::Error> {
let bytes = postcard::to_stdvec(msg)?;
let decoded: ChunkSubscribe = postcard::from_bytes(&bytes)?;
assert_eq!(msg, &decoded);
Ok(())
}
/// Round-trips a `ChunkMessage` through postcard and asserts the decoded value is identical.
fn roundtrip_message(msg: &ChunkMessage) -> Result<(), postcard::Error> {
let bytes = postcard::to_stdvec(msg)?;
let decoded: ChunkMessage = postcard::from_bytes(&bytes)?;
assert_eq!(msg, &decoded);
Ok(())
}
#[test]
fn roundtrip_chunk_subscribe() -> Result<(), postcard::Error> {
// Negative coordinates exercise the signed ChunkPos fields under serialization.
let msg = ChunkSubscribe {
center: ChunkPos::new(-4, 0, 7),
radius: 12,
};
roundtrip_subscribe(&msg)
}
#[test]
fn roundtrip_chunk_message_chunk() -> Result<(), postcard::Error> {
// A non-empty ChunkData: private fields force construction via new + set.
let mut data = ChunkData::new(ChunkPos::new(1, 2, 3), 0);
data.set(0, BlockId(1));
data.set(42, BlockId(1));
let msg = ChunkMessage::Chunk {
pos: ChunkPos::new(1, 2, 3),
data,
};
roundtrip_message(&msg)
}
#[test]
fn roundtrip_chunk_message_drop() -> Result<(), postcard::Error> {
let msg = ChunkMessage::Drop {
pos: ChunkPos::new(0, -1, 0),
};
roundtrip_message(&msg)
}

33
docs/chunk_streaming.md
View file

@ -10,7 +10,7 @@ Chunk residency is *reconciled* every tick against a **desired set**: the union
`cylinder_chunks(center, radius, out)` inserts every chunk position within the streaming cylinder around `center` into `out`. The shape is a disc in XZ (`dx² + dz² ≤ radius²`) extruded vertically to `±radius/2`, reflecting the fact that horizontal view distance exceeds vertical.
The ECS streaming system `stream_chunks` (in `main.rs`) queries every entity carrying `Player`, `Position`, and `ViewDistance`, and unions each anchor's cylinder into one `HashSet<ChunkPos>`. Because the sets are unioned, overlapping cylinders deduplicate automatically and a chunk is evicted only when *no* player wants it. See [Multiplayer](#multiplayer) below.
Two producers build desired sets. During the **startup loading gate**, the ECS streaming system `stream_chunks` (in `main.rs`) queries every entity carrying `Player`, `Position`, and `ViewDistance` and unions each anchor's cylinder, pre-warming the origin region before the network is up. During **steady-state play**, the desired set is instead the union of every connected client's subscription (each a `cylinder_chunks(center, radius)` around its camera), assembled from the `ClientStream` map in the main loop — see [Network delivery](#network-delivery-client--server). In both cases the sets are unioned, so overlapping cylinders deduplicate automatically and a chunk is evicted only when *no* subscriber wants it. See [Multiplayer](#multiplayer) below.
## The worker pool
@ -43,9 +43,38 @@ Between a chunk being dispatched and the worker returning it, the anchor may mov
Startup reuses the *same* worker pool and schedule; there is no separate synchronous loading path. Before granting player control, `main` runs the streaming schedule in a loop and polls `ServerWorld::streaming_idle()` (true when `in_flight` is empty). Once the initial region has at least one resident chunk and no work in flight, the region is ready. Waiting here is acceptable because no gameplay is running yet. During play the same reconcile runs every tick but is **never** waited on. A loading progress fraction is available as `resident / (resident + in_flight)`.
## Network delivery (client ↔ server)
Residency (above) keeps chunks in the server's memory; **delivery** streams them to each client. The two are decoupled: the reconcile pool does not know about clients, and delivery does not generate. Delivery is implemented in [`crates/net/src/chunk.rs`](../crates/net/src/chunk.rs) (transport) and [`crates/server/src/client_stream.rs`](../crates/server/src/client_stream.rs) (per-client bookkeeping), driven from `main.rs`; the client side lives in [`crates/client/src/chunks.rs`](../crates/client/src/chunks.rs).
### The chunk stream
After the handshake, the client opens one **bidirectional** QUIC stream (the canonical `StreamLayout::chunk_lod0`, stream 3) and the server accepts it, mirroring the control-stream convention. Both directions ride this one stream: client → server carries `ChunkSubscribe { center, radius }`, server → client carries `ChunkMessage::{Chunk { pos, data }, Drop { pos }}`. Frames use the existing length-prefixed `postcard` codec with a dedicated `MAX_CHUNK_FRAME_LEN` (1 MiB) cap, larger than the 64 KiB control cap.
### The async/sync bridge
The QUIC pump is async on the network thread; the simulation loop (server) and winit loop (client) are synchronous. Two channels cross the boundary per connection, in opposite directions, and use different primitives for that reason:
- **Inbound** (`ChunkSubscribe` arriving async, consumed by the sync loop) reuses the `crossbeam` `ServerEvent` channel, surfaced as `ServerEvent::ChunkSubscribe { id, request }`. The async `send` is non-blocking; the sync loop drains with `try_iter`.
- **Outbound** (a `ChunkMessage` produced by the sync loop, consumed async) uses a **`tokio` unbounded MPSC**. Its `send` is synchronous, so the non-async loop pushes without a runtime, while the pump's `recv().await` composes into its `tokio::select!`. A blocking `crossbeam` receiver would freeze the current-thread runtime and cannot appear in a `select!` arm. The tokio sender is wrapped so neither `server` nor `client` names a tokio type: `ChunkSink` (server → client deliveries) and `ChunkSubscriber` (client → server subscriptions). See [ADR-0010](adr/0010-net-crate-async-runtime.md).
The server-side pump is `chunk_stream_task`; its client mirror is `client_chunk_task`. Each is one `select!` loop over "a frame arrived to read" and "a message is queued to write." The client's `ClientLink` bundles the handshake outcome, the `ChunkSubscriber`, and a `crossbeam` `ChunkStream` receiver of deliveries.
### Per-client state and the diff
Each connected client is tracked by a `ClientStream` holding its `ChunkSink`, its current desired set (radius-clamped to `SERVER_MAX_RADIUS`), and its `sent` set. On each subscription, `desired_diff(previous, new)` yields the load list (`new − previous`) and drop list (`previous − new`); a `ChunkMessage::Drop` is emitted for every already-**sent** chunk that left the set. Newly-desired chunks are **not** sent immediately — chunk loads are async, so `ClientStream::flush` runs each tick and delivers every desired-but-unsent chunk that has since become resident, retrying on later ticks until the pool returns it.
### Self-contained payloads (all-air diff)
`ChunkMessage::Chunk` carries a `ChunkData` (the sparse, baseline-relative form; see [ADR-0009](adr/0009-baseline-relative-sparse-chunk-persistence.md)). Because the client runs **no** worldgen (the server owns world content; worldgen never runs client-side), it cannot reconstruct a worldgen baseline to diff against. So delivered chunks are diffed against an **all-air baseline** (`Chunk::default()`): the edits become the chunk's full non-air content, and the client materializes each payload against its own all-air `Chunk::default()`. This makes every delivery self-contained, at the cost of not exploiting the deterministic baseline for compression — a compression concern deferred to the LOD/compression pass.
### Client application
The client subscribes with its own `LOAD_RADIUS` (so the server's per-client resident set matches what the client keeps) whenever its center chunk changes. It drains deliveries under a per-frame meshing budget: `ChunkMessage::Chunk` → materialize → `generate_mesh` → `insert_mesh` (skipping empty meshes); `ChunkMessage::Drop` → `remove_mesh`. It **also** evicts chunks outside `LOAD_RADIUS` locally, independent of the server `Drop`, so memory stays bounded even if the server is slow.
## Multiplayer
No per-player streaming pipeline exists. Every player anchor's cylinder is unioned into one desired set, reconciled against one chunk store served by one worker pool. A player joining or leaving is simply an entity entering or leaving the ECS query; it requires no streaming-specific code. The only per-player concern is the loading gate, which for a joining player checks that player's cylinder against the resident set rather than the global set. Backpressure and fairness across players (a bounded job channel, nearest-first priority) are shared-pipeline concerns deferred for later.
Residency is a single shared pipeline: every client's subscription cylinder is unioned into one desired set, reconciled against one chunk store served by one worker pool, so a chunk is generated once no matter how many clients want it. **Delivery**, by contrast, is per-client: each `ClientStream` independently tracks what that client has been sent and diffs its own subscription (see [Network delivery](#network-delivery-client--server)). A client joining or leaving is a `ClientStream` entering or leaving the map on the connect/disconnect events. Backpressure and fairness across clients (a bounded job channel, nearest-first priority, per-chunk ack/flow-control) remain deferred.
## Level of detail

2
docs/packs.md
View file

@ -8,7 +8,7 @@ Client-side asset overlays: textures, sounds, models, fonts, language files. No
A pack is a directory tree mirroring `/assets/` that overrides files by path. The renderer/asset loader resolves logical asset IDs against a stack of pack roots (base game → installed packs by priority) and the topmost hit wins. Ownership sits with the asset pipeline (in `client`, or a sibling `assets` crate if it grows). Pack authors never touch Lua.
A client's own resource packs are a purely local choice; the server has no say over them and they are never part of gameplay modlist matching. The **one** exception is a **server resource pack**: a server may push a single cosmetic overlay of its own (a themed / total-conversion server) to connecting clients. It is a one-way server → client push, applied on top of the client's local stack, and enforced per the server's choice — *optional* packs the client may decline and keep playing, a *required* pack the client declines or fails to fetch rejects the connection. It is still `assets/`-only (no `data/`, no `scripts/`), so it can never affect authoritative state. Fetch and enforcement semantics live in the vault's `Architecture/Load order.md` § Streaming.
A client's own resource packs are a purely local choice; the server has no say over them and they are never part of gameplay modlist matching. The **one** exception is a **server resource pack**: a server may push a single cosmetic overlay of its own (a themed / total-conversion server) to connecting clients. It is a one-way server → client push, applied on top of the client's local stack, and enforced per the server's choice — *optional* packs the client may decline and keep playing, a *required* pack the client declines or fails to fetch rejects the connection. It is still `assets/`-only (no `data/`, no `scripts/`), so it can never affect authoritative state.
## Data packs