feat(net): carry chunk subscribe and delivery on the chunk stream

crates/net/src/chunk.rs

@@ -0,0 +1,99 @@
+// 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,
+                }
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+#[path = "tests/chunk.rs"]
+mod tests;

crates/net/src/codec.rs

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

crates/net/src/lib.rs

@@ -6,12 +6,14 @@
 //!
 //! 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 chunk::ChunkSink;
 pub use runtime::{ConnectOutcome, NetworkServer, ServerEvent, connect_in_background};
 
 /// Default UDP port the server binds and the client connects to when none is configured.

crates/net/src/runtime.rs

@@ -5,9 +5,11 @@
 use std::net::SocketAddr;
 use std::thread;
 
+use shared::protocol::chunk::ChunkSubscribe;
 use shared::protocol::{ClientHello, HandshakeAck};
 use tracing::{info, warn};
 
+use crate::chunk::{ChunkSink, chunk_stream_task};
 use crate::endpoint::{client_endpoint, server_endpoint};
 use crate::error::NetError;
 use crate::handshake::{ServerConnection, accept_connection, connect};
@@ -18,12 +20,14 @@ pub type ConnectOutcome = crossbeam_channel::Receiver<Result<HandshakeAck, Strin
 /// 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 +36,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 +198,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,

crates/net/src/tests/chunk.rs

@@ -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(())
+}