diff --git a/crates/net/src/authority.rs b/crates/net/src/authority.rs new file mode 100644 index 0000000..828cbaa --- /dev/null +++ b/crates/net/src/authority.rs @@ -0,0 +1,97 @@ +// SPDX-License-Identifier: AGPL-3.0-only + +//! Authority-stream transport: the per-connection task that pushes server-authoritative state to a client. + +use shared::protocol::authority::AuthorityMessage; +use tokio::sync::mpsc::error::TrySendError; +use tokio::sync::mpsc::{Sender, UnboundedReceiver, UnboundedSender}; +use tracing::{debug, warn}; + +use crate::codec::{read_frame, write_frame}; + +/// Maximum accepted authority frame length, in bytes. +/// +/// Authority payloads are small fixed-shape records; the bound is generous relative to a [`ServerStats`](shared::protocol::authority::ServerStats) and exists to cap what a malformed or hostile length prefix can make the peer allocate. +pub const MAX_AUTHORITY_FRAME_LEN: usize = 64 * 1024; + +/// A synchronous handle the simulation loop uses to push [`AuthorityMessage`]s to one connection. +#[derive(Debug, Clone)] +pub struct AuthoritySink { + /// Outbound queue drained by the connection's authority task. + tx: UnboundedSender, +} + +impl AuthoritySink { + /// Wraps `tx` as an authority sink. + pub(crate) fn new(tx: UnboundedSender) -> Self { + Self { tx } + } + + /// Queues `msg` for delivery on the connection's authority 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: AuthorityMessage) { + if self.tx.send(msg).is_err() { + debug!("authority sink send failed; connection task has ended"); + } + } +} + +/// Runs the server-side authority pump for one connection until the stream or connection closes. +/// +/// Opens the unidirectional stream, then writes every [`AuthorityMessage`] handed over by the simulation loop. The loop ends when the sink is dropped (the connection is being torn down) or a write fails. +pub(crate) async fn server_authority_task( + connection: quinn::Connection, + id: u64, + mut outbound: UnboundedReceiver, +) { + let mut send = match connection.open_uni().await { + Ok(stream) => stream, + Err(error) => { + warn!(%error, id, "failed to open authority stream"); + return; + } + }; + + while let Some(message) = outbound.recv().await { + if let Err(error) = write_frame(&mut send, &message).await { + warn!(%error, id, "failed to write authority frame; ending authority stream"); + break; + } + } +} + +/// Runs the client-side authority pump for one connection until the stream or connection closes. +/// +/// Accepts the unidirectional stream the server opens, then forwards every decoded [`AuthorityMessage`] to the UI thread. The loop ends when the stream closes or the UI drops its receiver. +pub(crate) async fn client_authority_task( + connection: quinn::Connection, + inbound: Sender, +) { + let mut recv = match connection.accept_uni().await { + Ok(stream) => stream, + Err(error) => { + debug!(%error, "authority stream never opened"); + return; + } + }; + + loop { + match read_frame::(&mut recv, MAX_AUTHORITY_FRAME_LEN).await { + Ok(message) => match inbound.try_send(message) { + Ok(()) => {} + // A full channel means the UI is behind on a purely diagnostic stream; dropping the newest message is preferable to blocking the read loop. + Err(TrySendError::Full(_)) => { + debug!("authority delivery dropped; UI queue is full"); + } + // A closed channel means the UI has gone away, so there is nothing left to deliver to. + Err(TrySendError::Closed(_)) => break, + }, + Err(error) => { + // A read error is the normal end of the session (stream finished or reset). + debug!(%error, "authority stream read ended"); + break; + } + } + } +} diff --git a/crates/net/src/lib.rs b/crates/net/src/lib.rs index 9d22646..6806568 100644 --- a/crates/net/src/lib.rs +++ b/crates/net/src/lib.rs @@ -6,6 +6,7 @@ //! //! 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 authority; pub mod chunk; pub mod codec; pub mod endpoint; @@ -14,9 +15,11 @@ pub mod handshake; pub mod runtime; pub mod stats; +pub use authority::AuthoritySink; pub use chunk::{ChunkSink, ChunkSubscriber}; pub use runtime::{ - ChunkStream, ClientLink, ConnectOutcome, NetworkServer, ServerEvent, connect_in_background, + AuthorityStream, ChunkStream, ClientLink, ConnectOutcome, NetworkServer, ServerEvent, + connect_in_background, }; pub use stats::NetStats; diff --git a/crates/net/src/runtime.rs b/crates/net/src/runtime.rs index 7ffbc51..27c783c 100644 --- a/crates/net/src/runtime.rs +++ b/crates/net/src/runtime.rs @@ -6,10 +6,12 @@ use std::net::SocketAddr; use std::sync::{Arc, OnceLock}; use std::thread; +use shared::protocol::authority::AuthorityMessage; use shared::protocol::chunk::{ChunkMessage, ChunkSubscribe}; use shared::protocol::{ClientHello, HandshakeAck}; use tracing::{info, warn}; +use crate::authority::{AuthoritySink, client_authority_task, server_authority_task}; use crate::chunk::{ChunkSink, ChunkSubscriber, chunk_stream_task, client_chunk_task}; use crate::endpoint::{client_endpoint, server_endpoint}; use crate::error::NetError; @@ -26,6 +28,14 @@ pub type ChunkStream = tokio::sync::mpsc::Receiver; // TODO: revisit once meshing moves to a worker pool; the right depth follows the UI's consume rate, so this is a candidate to derive from the meshing budget / view distance in a config layer rather than a hand-set constant. const CHUNK_DELIVERY_CAPACITY: usize = 32; +/// Bounded receiver of authority-stream messages, drained by the UI thread with `try_recv`. +pub type AuthorityStream = tokio::sync::mpsc::Receiver; + +/// Capacity of the client's authority channel, in messages. +/// +/// Shallow on purpose: the server pushes roughly one message per second, so anything beyond a small backlog is stale by the time the UI would read it. The network task drops rather than blocks when this fills. +const AUTHORITY_CAPACITY: usize = 4; + /// 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. @@ -36,6 +46,8 @@ pub struct ClientLink { pub subscribe: ChunkSubscriber, /// Receives chunk deliveries from the server, drained non-blocking each frame. pub chunks: ChunkStream, + /// Receives periodic server-authoritative state, drained non-blocking each frame. + pub authority: AuthorityStream, /// The live QUIC connection, published by the network thread once the handshake completes. Held privately so the `quinn` types stay inside this crate; the UI thread reads through [`ClientLink::stats`]. connection: Arc>, /// Application-level message counters shared with the chunk task. @@ -63,6 +75,8 @@ pub enum ServerEvent { 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, + /// Outbound handle for pushing [`shared::protocol::authority::AuthorityMessage`]s to this client, retained alongside `chunks` for the same lifetime. + authority: AuthoritySink, }, /// A previously connected client's session ended. ClientDisconnected { @@ -233,26 +247,29 @@ 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. + // The outbound channels bridge the sync simulation loop to this connection's stream tasks; the sinks are handed to the loop via the connect event. let (chunk_tx, chunk_rx) = tokio::sync::mpsc::unbounded_channel(); + let (authority_tx, authority_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, chunks: ChunkSink::new(chunk_tx), + authority: AuthoritySink::new(authority_tx), }) .is_err() { return; } - // The chunk pump runs on its own task so the connection-close wait below does not block it. + // Each pump runs on its own task so the connection-close wait below does not block either. tokio::spawn(chunk_stream_task( connection.clone(), id, events.clone(), chunk_rx, )); + tokio::spawn(server_authority_task(connection.clone(), id, authority_rx)); let reason = connection.closed().await; let _ = events.send(ServerEvent::ClientDisconnected { id, @@ -283,6 +300,8 @@ pub fn connect_in_background(server_addr: SocketAddr, hello: ClientHello) -> Cli // The delivery channel is bounded so a slow (e.g. debug-build) UI thread applies backpressure to the network task instead of letting undelivered chunks accumulate without limit. let (chunks_tx, chunks_rx) = tokio::sync::mpsc::channel::(CHUNK_DELIVERY_CAPACITY); + let (authority_tx, authority_rx) = + tokio::sync::mpsc::channel::(AUTHORITY_CAPACITY); let spawned = thread::Builder::new() .name("net-client".to_owned()) @@ -315,14 +334,16 @@ pub fn connect_in_background(server_addr: SocketAddr, hello: ClientHello) -> Cli } // The cell is written exactly once, here; a failure would mean a second handshake on one link, which cannot occur. let _ = task_connection.set(connected.connection.clone()); - // 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, - &task_counters, - ) - .await; + // Pump both streams concurrently on this thread until the UI drops its handles or the server closes the connection. `join!` rather than `select!`: neither stream ending is a reason to abandon the other mid-frame. + tokio::join!( + client_chunk_task( + connected.connection.clone(), + subscribe_rx, + chunks_tx, + &task_counters, + ), + client_authority_task(connected.connection, authority_tx), + ); warn!("server connection closed"); } Err(error) => { @@ -340,6 +361,7 @@ pub fn connect_in_background(server_addr: SocketAddr, hello: ClientHello) -> Cli handshake: outcome_rx, subscribe: ChunkSubscriber::new(subscribe_tx), chunks: chunks_rx, + authority: authority_rx, connection, counters, }