feat(net): add the authority stream transport

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<AuthorityMessage>,
+}
+
+impl AuthoritySink {
+    /// Wraps `tx` as an authority sink.
+    pub(crate) fn new(tx: UnboundedSender<AuthorityMessage>) -> 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<AuthorityMessage>,
+) {
+    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<AuthorityMessage>,
+) {
+    let mut recv = match connection.accept_uni().await {
+        Ok(stream) => stream,
+        Err(error) => {
+            debug!(%error, "authority stream never opened");
+            return;
+        }
+    };
+
+    loop {
+        match read_frame::<AuthorityMessage>(&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;
+            }
+        }
+    }
+}

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;
 

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<ChunkMessage>;
 // 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<AuthorityMessage>;
+
+/// 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<OnceLock<quinn::Connection>>,
     /// 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::<ChunkMessage>(CHUNK_DELIVERY_CAPACITY);
+    let (authority_tx, authority_rx) =
+        tokio::sync::mpsc::channel::<AuthorityMessage>(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.
+                        // 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,
+                                connected.connection.clone(),
                                 subscribe_rx,
                                 chunks_tx,
                                 &task_counters,
-                        )
+                            ),
-                        .await;
+                            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,
     }