diff --git a/rs/moq-quic/README.md b/rs/moq-quic/README.md index 10a2a8ffc4..4f8de3b237 100644 --- a/rs/moq-quic/README.md +++ b/rs/moq-quic/README.md @@ -81,6 +81,7 @@ Changes on top of the upstream commit, besides the renames: - **Draining endpoint event**: `EndpointEvent::is_draining` fires once per connection when it stops being active, before the drained event, so the driver's `Endpoint::wait_all_draining` can end a shutdown without sitting out every closing period. - **Late stream limits**: peer transport parameters that raise a stream limit after an open was refused emit `StreamEvent::Available`, as `MAX_STREAMS` does. A server accepted before the rest of a multi-packet ClientHello (a post-quantum key share is enough) otherwise leaves a 0.5-RTT `open_uni` parked until the idle timeout. Not offered upstream yet. - **Stream credit on stop**: `RecvStream::stop` on a stream whose final size is already known (finished or reset) frees it and queues `MAX_STREAMS` at once, as reading the reset does, instead of leaving the credit until the next packet arrives. Not offered upstream yet; quinn `main` has the same gap. +- **Drop resets** (`quic::driver::SendStream`): a send stream dropped before `finish` is reset with code 0, or with the peer's `STOP_SENDING` code, instead of quinn's implicit `finish`, so a cancelled writer never ends a truncated stream with a FIN. This matches moq-uring's driver and moq-net's transport contract. `quic::web` resets after the WebTransport header with the mapped code. quinn's `stream_stopped_2` finishes its stream before dropping it. Not offered upstream: it reverses a documented quinn behaviour. - **Connection limits** (`Limits`, `Connection::set_limits`, and the drivers' `set_limits`): one validated call changes a live connection's stream limit and windows, refusing values a peer would reject and changing nothing then. moq-tokio and moq-uring re-export the type, so both runtimes check limits the same way. Not offered upstream: it only bundles quinn's existing setters. Recheck the BBR3 changes when cherry-picking quinn#2481 updates, or if it merges. diff --git a/rs/moq-tokio/MIGRATE.md b/rs/moq-tokio/MIGRATE.md index e76df24141..ab41045e50 100644 --- a/rs/moq-tokio/MIGRATE.md +++ b/rs/moq-tokio/MIGRATE.md @@ -4,6 +4,11 @@ How to update your code for each breaking change, newest first. ## Unreleased +### Dropping an unfinished send stream resets it (#5909) + +`quic::driver::SendStream` and `quic::web::SendStream` send `RESET_STREAM` with code 0 when dropped before `finish()`, instead of finishing implicitly. +Call `finish()` before dropping a stream whose data should arrive whole. + ### `jemalloc::tikv_jemallocator` is 0.7 (#6053) The re-exported allocator crate moved from tikv-jemallocator 0.6 to 0.7. diff --git a/rs/moq-tokio/src/quic/driver/send_stream.rs b/rs/moq-tokio/src/quic/driver/send_stream.rs index 44010e1e1d..7e09c65b47 100644 --- a/rs/moq-tokio/src/quic/driver/send_stream.rs +++ b/rs/moq-tokio/src/quic/driver/send_stream.rs @@ -16,9 +16,10 @@ use crate::quic::driver::{ /// A stream that can only be used to send data /// -/// If dropped, streams that haven't been explicitly [`reset()`] will be implicitly [`finish()`]ed, -/// continuing to (re)transmit previously written data until it has been fully acknowledged or the -/// connection is closed. +/// If dropped, streams that haven't been explicitly [`finish()`]ed are [`reset()`], with code 0 +/// unless the peer stopped them, so the peer can't mistake a truncated stream for a complete one. +/// A finished stream keeps (re)transmitting previously written data until it has been fully +/// acknowledged or the connection is closed. /// /// # Cancellation /// @@ -35,6 +36,9 @@ pub struct SendStream { conn: ConnectionRef, stream: StreamId, is_0rtt: bool, + finished: bool, + /// The offset and code a drop before `finish` resets at + drop_reset: (u64, VarInt), } impl SendStream { @@ -43,6 +47,8 @@ impl SendStream { conn, stream, is_0rtt, + finished: false, + drop_reset: (0, VarInt::from_u32(0)), } } @@ -185,6 +191,7 @@ impl SendStream { let mut conn = self.conn.state.lock("finish"); match conn.inner.send_stream(self.stream).finish() { Ok(()) => { + self.finished = true; conn.wake(); Ok(()) } @@ -312,6 +319,11 @@ impl SendStream { self.stream } + /// Reset at `offset` with `code`, instead of at 0 with code 0, if dropped before `finish` + pub(crate) fn set_drop_reset(&mut self, offset: u64, code: VarInt) { + self.drop_reset = (offset, code); + } + /// Attempt to write bytes from buf into the stream. /// /// On success, returns Poll::Ready(Ok(num_bytes_written)). @@ -375,15 +387,21 @@ impl Drop for SendStream { if conn.error.is_some() || (self.is_0rtt && conn.check_0rtt().is_err()) { return; } - match conn.inner.send_stream(self.stream).finish() { - Ok(()) => conn.wake(), - Err(FinishError::Stopped(reason)) => { - if conn.inner.send_stream(self.stream).reset(reason).is_ok() { - conn.wake(); - } + // Fails harmlessly when already reset. + let mut stream = conn.inner.send_stream(self.stream); + let reset = match stream.stopped() { + // Echo the code, as RFC 9000 section 3.5 asks, even after `finish`: the peer reads + // nothing more, so retransmitting the rest is wasted. + Ok(Some(code)) => stream.reset(code), + _ if self.finished => return, + _ => { + let (offset, code) = self.drop_reset; + let offset = offset.min(stream.offset()); + stream.reset_at(offset, code) } - // Already finished or reset, which is fine. - Err(FinishError::ClosedStream) => {} + }; + if reset.is_ok() { + conn.wake(); } } } diff --git a/rs/moq-tokio/src/quic/driver/tests.rs b/rs/moq-tokio/src/quic/driver/tests.rs index 8338ad0671..96aac54e64 100755 --- a/rs/moq-tokio/src/quic/driver/tests.rs +++ b/rs/moq-tokio/src/quic/driver/tests.rs @@ -44,7 +44,8 @@ use tracing::{error_span, info}; use tracing_subscriber::EnvFilter; use super::{ - ClientConfig, Credit, Endpoint, EndpointConfig, RecvStream, SendStream, TransportConfig, + ClientConfig, Credit, Endpoint, EndpointConfig, ReadError, ReadToEndError, RecvStream, + SendStream, TransportConfig, VarInt, }; #[test] @@ -1712,7 +1713,7 @@ async fn stream_stopped_2() { async { endpoint.accept().await.unwrap().await } ) .unwrap(); - let send_stream = conn.open_uni(Default::default()).await.unwrap(); + let mut send_stream = conn.open_uni(Default::default()).await.unwrap(); let stopped = timeout(Duration::from_millis(100), send_stream.stopped()) .instrument(error_span!("stopped")); tokio::pin!(stopped); @@ -1722,13 +1723,63 @@ async fn stream_stopped_2() { _x = &mut stopped => {}, _x = std::future::ready(()) => {} } - // drop the send stream + // finish and drop the send stream, since dropping it unfinished would reset it + send_stream.finish().unwrap(); drop(send_stream); // make sure the stopped future still resolves let res = stopped.await; assert_eq!(res, Ok(Ok(None))); } +/// A send stream dropped before `finish` is reset, so the peer can't mistake the truncated data +/// for the whole stream. +#[tokio::test] +async fn dropped_send_stream_resets() { + let _guard = subscribe(); + let endpoint = endpoint(); + + let (conn, server_conn) = tokio::try_join!( + endpoint + .connect(endpoint.local_addr().unwrap(), "localhost") + .unwrap(), + async { endpoint.accept().await.unwrap().await } + ) + .unwrap(); + let mut send_stream = conn.open_uni(Default::default()).await.unwrap(); + send_stream.write_all(b"truncated").await.unwrap(); + drop(send_stream); + + let mut recv_stream = server_conn.accept_uni().await.unwrap(); + let res = recv_stream.read_to_end(usize::MAX).await; + assert!( + matches!(res, Err(ReadToEndError::Read(ReadError::Reset(code))) if code == VarInt::from_u32(0)), + "{res:?}" + ); +} + +/// A send stream finished before it drops still delivers its data and FIN. +#[tokio::test] +async fn finished_send_stream_survives_drop() { + let _guard = subscribe(); + let endpoint = endpoint(); + + let (conn, server_conn) = tokio::try_join!( + endpoint + .connect(endpoint.local_addr().unwrap(), "localhost") + .unwrap(), + async { endpoint.accept().await.unwrap().await } + ) + .unwrap(); + let mut send_stream = conn.open_uni(Default::default()).await.unwrap(); + send_stream.write_all(b"complete").await.unwrap(); + send_stream.finish().unwrap(); + drop(send_stream); + + let mut recv_stream = server_conn.accept_uni().await.unwrap(); + let data = recv_stream.read_to_end(usize::MAX).await.unwrap(); + assert_eq!(data, b"complete"); +} + #[tokio::test] async fn stream_drop_removes_blocked_reader() { let _guard = subscribe(); diff --git a/rs/moq-tokio/src/quic/web/send.rs b/rs/moq-tokio/src/quic/web/send.rs index 30aae19ba5..5fecef3597 100644 --- a/rs/moq-tokio/src/quic/web/send.rs +++ b/rs/moq-tokio/src/quic/web/send.rs @@ -11,6 +11,9 @@ use crate::quic::web::{CloseReason, ClosedStream, SessionError, WriteError}; /// A stream that can be used to send bytes. See [`crate::quic::driver::SendStream`]. /// +/// Dropping it before [`finish`](Self::finish) resets it with code 0, still delivering the +/// WebTransport header. +/// /// This wrapper is mainly needed for error codes, which is unfortunate. /// WebTransport uses u32 error codes and they're mapped in a reserved HTTP/3 error space. #[derive(Debug)] @@ -22,7 +25,12 @@ pub struct SendStream { } impl SendStream { - pub(crate) fn new(stream: crate::quic::driver::SendStream, close: Arc, header: Option) -> Self { + pub(crate) fn new( + mut stream: crate::quic::driver::SendStream, + close: Arc, + header: Option, + ) -> Self { + stream.set_drop_reset(header.unwrap_or(0), close.encode_stream_code(0)); Self { stream, close, header } } @@ -93,9 +101,6 @@ impl SendStream { /// Mark the stream as finished, such that no more data can be written. See /// [`crate::quic::driver::SendStream::finish`]. - /// - /// WARNING: This is implicitly called on Drop, but it's a common footgun in QUIC libraries. - /// If you cancel futures by dropping them you'll get incomplete writes. pub fn finish(&mut self) -> Result<(), ClosedStream> { self.stream.finish().map_err(Into::into) } diff --git a/rs/moq-tokio/src/quic/web/session.rs b/rs/moq-tokio/src/quic/web/session.rs index b65d732209..5215933272 100644 --- a/rs/moq-tokio/src/quic/web/session.rs +++ b/rs/moq-tokio/src/quic/web/session.rs @@ -1325,6 +1325,44 @@ mod tests { assert!(matches!(bi, crate::quic::web::ReadError::Reset(8)), "{bi:?}"); } + /// A stream dropped before `finish` is reset with WebTransport code 0, still after its + /// header, so the peer neither reads the truncated data as complete nor loses the reset. + #[tokio::test] + async fn drop_resets_after_header() { + let (client_conn, server_conn, _endpoints) = pair().await; + let (client, server) = tokio::join!( + Session::connect(client_conn, url::Url::parse("https://localhost/").unwrap()), + async { crate::quic::web::Request::accept(server_conn).await.unwrap().ok().await }, + ); + let (client, server) = (client.unwrap(), server.unwrap()); + + let mut uni = client.open_uni(Default::default()).await.unwrap(); + uni.write_all(b"truncated").await.unwrap(); + drop(uni); + let (mut send, mut client_bi) = client.open_bi(Default::default()).await.unwrap(); + send.write_all(b"request").await.unwrap(); + send.finish().unwrap(); + + let read = async { + let mut uni = server.accept_uni().await.unwrap(); + // The accepted half carries no header of its own, but still needs the code mapped. + let (send, mut bi) = server.accept_bi().await.unwrap(); + assert_eq!(bi.read_to_end(usize::MAX).await.unwrap(), b"request"); + drop(send); + let uni = loop { + match uni.read_chunk(usize::MAX).await { + Ok(Some(_)) => {} + res => break res, + } + }; + (uni, client_bi.read_chunk(usize::MAX).await) + }; + // Without the header the server never surfaces the stream, so bound the wait. + let (uni, bi) = tokio::time::timeout(Duration::from_secs(10), read).await.unwrap(); + assert!(matches!(uni, Err(crate::quic::web::ReadError::Reset(0))), "{uni:?}"); + assert!(matches!(bi, Err(crate::quic::web::ReadError::Reset(0))), "{bi:?}"); + } + /// A stream that names another session is still reported as `UnknownSession`. #[tokio::test] async fn uni_other_session() {