Implicitly finish send streams on drop

quinn/src/connection.rs

@@ -503,11 +503,11 @@ impl ConnectionInner {
                     stream,
                     stop_reason,
                 } => {
-                    let _ = self
-                        .finishing
-                        .remove(&stream)
-                        .unwrap()
-                        .send(stop_reason.map(|e| WriteError::Stopped { error_code: e }));
+                    if let Some(finishing) = self.finishing.remove(&stream) {
+                        // If the finishing stream was already dropped, there's nothing more to do.
+                        let _ = finishing
+                            .send(stop_reason.map(|e| WriteError::Stopped { error_code: e }));
+                    }
                 }
             }
         }

quinn/src/streams.rs

@@ -20,6 +20,9 @@ pub enum NewStream {
 }
 
 /// A stream that can only be used to send data
+///
+/// If dropped, streams that haven't been explicitly `reset` will continue to (re)transmit
+/// previously written data until it has been fully acknowledged or the connection is closed.
 pub struct SendStream {
     conn: ConnectionRef,
     stream: StreamId,
@@ -150,9 +153,11 @@ impl Drop for SendStream {
         if conn.error.is_some() || (self.is_0rtt && conn.check_0rtt().is_err()) {
             return;
         }
-        if !self.finished {
-            conn.inner.reset(self.stream, 0);
-            conn.notify();
+        if !self.finished && self.finishing.is_none() {
+            // Errors indicate that the stream was already reset, which is fine.
+            if conn.inner.finish(self.stream).is_ok() {
+                conn.notify();
+            }
         }
     }
 }

quinn/src/tests.rs

@@ -221,11 +221,8 @@ fn run_echo(client_addr: SocketAddr, server_addr: SocketAddr) {
                             .and_then(move |(send, recv)| {
                                 tokio::io::write_all(send, b"foo".to_vec())
                                     .map_err(|e| panic!("write: {}", e))
-                                    .and_then(|(send, _)| {
-                                        tokio::io::shutdown(send)
-                                            .map_err(|e| panic!("finish: {}", e))
-                                    })
                                     .and_then(move |_| {
+                                        // Rely on send being implicitly finished when we drop it
                                         recv.read_to_end(usize::max_value())
                                             .unwrap()
                                             .map_err(|e| panic!("read: {}", e))