98 lines
4 KiB
Rust
98 lines
4 KiB
Rust
// 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;
|
|
}
|
|
}
|
|
}
|
|
}
|