From bb4e591a0923eebfe8229e5c53cb300ee6f6a129 Mon Sep 17 00:00:00 2001 From: Serkyo Date: Wed, 22 Jul 2026 14:45:41 +0200 Subject: [PATCH] fix(net): split chunk stream reader and writer to avoid frame desync --- crates/net/src/chunk.rs | 118 ++++++++++++++++++++-------------------- 1 file changed, 58 insertions(+), 60 deletions(-) diff --git a/crates/net/src/chunk.rs b/crates/net/src/chunk.rs index dea63b9..8ab581f 100644 --- a/crates/net/src/chunk.rs +++ b/crates/net/src/chunk.rs @@ -1,10 +1,6 @@ // 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::{Sender, UnboundedReceiver, UnboundedSender}; @@ -56,41 +52,42 @@ pub(crate) async fn chunk_stream_task( } }; - loop { - tokio::select! { - // A subscription frame arrived from the client. - frame = read_frame::(&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"); + // 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::(&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; } } - } - // 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, + 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 => {} } } @@ -136,38 +133,39 @@ pub(crate) async fn client_chunk_task( } }; - loop { - tokio::select! { - // The UI thread pushed a subscription update to forward to the server. - request = subscribe.recv() => { - match request { - Some(request) => { - if let Err(error) = write_frame(&mut send, &request).await { - warn!(%error, "failed to write chunk subscribe; ending chunk stream"); - break; - } - } - // The subscriber was dropped: the UI is shutting down. - None => break, - } - } - // A chunk arrived from the server. - frame = read_frame::(&mut recv, MAX_CHUNK_FRAME_LEN) => { - match frame { - 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"); + // 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::(&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 => {} } }