synvael/crates/net/src/chunk.rs

// SPDX-License-Identifier: AGPL-3.0-only

//! Chunk-stream transport: the per-connection task that pumps chunk subscriptions and deliveries.

use shared::protocol::chunk::{ChunkMessage, ChunkSubscribe};
use tokio::sync::mpsc::{Sender, 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;
        }
    };

    // Inbound subscriptions are read on their own future so `read_frame` is never cancelled mid-frame by an outbound write becoming ready.
    let reader = async {
        loop {
            match read_frame::<ChunkSubscribe>(&mut recv, MAX_CHUNK_FRAME_LEN).await {
                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;
                }
            }
        }
    };

    // Outbound chunks handed back by the simulation loop are written on their own future. The loop ends when the sink is dropped (the connection is being torn down).
    let writer = async {
        while let Some(message) = outbound.recv().await {
            if let Err(error) = write_frame(&mut send, &message).await {
                warn!(%error, id, "failed to write chunk frame; ending chunk stream");
                break;
            }
        }
    };

    // The task ends as soon as either direction closes; the other future is then dropped, abandoning the stream that is already being torn down.
    tokio::select! {
        () = reader => {}
        () = writer => {}
    }
}

/// 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: 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;
        }
    };

    // Inbound chunks are read on their own future so `read_frame` is never cancelled mid-frame by an outbound subscribe becoming ready.
    let reader = async {
        loop {
            match read_frame::<ChunkMessage>(&mut recv, MAX_CHUNK_FRAME_LEN).await {
                Ok(message) => {
                    // `send` awaits when the delivery channel is full: the task suspends (yielding the runtime thread so the connection keeps ACKing) until the UI drains a slot, and until then reads no further frames, which backpressures the server via QUIC stream flow control. An error means the UI dropped its receiver, so the session ends.
                    if deliveries.send(message).await.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;
                }
            }
        }
    };

    // Outbound subscription updates from the UI thread are written on their own future. The loop ends when the subscriber is dropped (the UI is shutting down).
    let writer = async {
        while let Some(request) = subscribe.recv().await {
            if let Err(error) = write_frame(&mut send, &request).await {
                warn!(%error, "failed to write chunk subscribe; ending chunk stream");
                break;
            }
        }
    };

    // The task ends as soon as either direction closes; the other future is then dropped, abandoning the stream that is already being torn down.
    tokio::select! {
        () = reader => {}
        () = writer => {}
    }
}

#[cfg(test)]
#[path = "tests/chunk.rs"]
mod tests;