fix(quic)!: reset driver send streams dropped before finish

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.

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.

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();
         }
     }
 }

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();

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<CloseReason>, header: Option<u64>) -> Self {
+	pub(crate) fn new(
+		mut stream: crate::quic::driver::SendStream,
+		close: Arc<CloseReason>,
+		header: Option<u64>,
+	) -> 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)
 	}

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() {