fix(io): report what a compio file round sent before it failed

io/src/compio.rs

@@ -853,7 +853,8 @@ mod splice {
     }
 
     /// One round: up to a pipe's capacity from the file into the pipe, then
-    /// all of it from the pipe into the socket.
+    /// all of it from the pipe into the socket, or as much as the socket
+    /// took before an error, which is then what the round reports.
     ///
     /// The second step is not an io_uring splice: io_uring runs every splice
     /// on an io-wq worker, and with no poll support for the operation a
@@ -905,32 +906,39 @@ mod splice {
         };
 
         let mut drained = 0;
-        while drained < filled {
+        let error = loop {
+            if drained == filled {
+                stream.pipe.set(Some(pipe));
+                return BufResult(Ok(filled), file);
+            }
             match crate::sys::pipe::splice(
                 pipe.read.as_fd(),
                 stream.inner.as_fd(),
                 filled - drained,
             ) {
                 Ok(0) => {
-                    return BufResult(
-                        Err(io::Error::new(
-                            io::ErrorKind::WriteZero,
-                            "stream accepted no bytes",
-                        )),
-                        file,
-                    );
+                    break io::Error::new(io::ErrorKind::WriteZero, "stream accepted no bytes");
                 }
                 Ok(n) => drained += n,
                 Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
                     if let Err(e) = ready(stream, Interest::Writable).await {
-                        return BufResult(Err(e), file);
+                        break e;
                     }
                 }
-                Err(e) => return BufResult(Err(e), file),
+                Err(e) => break e,
             }
+        };
+        // The pipe may still hold bytes, so it goes with this round. What was
+        // drained before the error is in the socket: that is what this call
+        // sent, as `send_file` reports it, and the error is the next call's
+        // to report. Reporting the error instead would have a caller that
+        // retries it (`Interrupted`, in `send_file_all`) send those bytes
+        // again.
+        if drained > 0 {
+            BufResult(Ok(drained), file)
+        } else {
+            BufResult(Err(error), file)
         }
-        stream.pipe.set(Some(pipe));
-        BufResult(Ok(filled), file)
     }
 
     /// Makes the socket `O_NONBLOCK`, once.

io/tests/io.rs

@@ -1336,6 +1336,86 @@ mod compio {
         );
     }
 
+    /// A round that fails after part of it reached the socket reports the
+    /// part that did, as any write does, and leaves the error to the next
+    /// call: reported as the round's result, a retried error (`Interrupted`,
+    /// which `send_file_all` retries at the same offset) would send those
+    /// bytes twice.
+    ///
+    /// The peer reads nothing and both socket buffers are at their minimum,
+    /// so the round's pipe drains only partly before the socket is full;
+    /// the peer then resets the connection, which fails the rest.
+    #[cfg(any(target_os = "linux", target_os = "android"))]
+    #[test]
+    fn send_file_reports_what_it_sent_before_failing() {
+        use std::{sync::mpsc, time::Duration};
+
+        use http_kit_io::{BufResult, Listen as _, Listener as _, SendFile as _};
+
+        // Below the kernel's minimum, which each buffer is raised to
+        // (`SOCK_MIN_SNDBUF`, `SOCK_MIN_RCVBUF`), far below a pipe's 64 KiB.
+        const BUFFER: usize = 1;
+
+        let path = std::path::Path::new(env!("CARGO_TARGET_TMPDIR")).join("compio-reset-peer");
+        let contents = super::payload();
+        std::fs::write(&path, &contents).expect("write the file to send");
+        let file = std::rc::Rc::new(std::fs::File::open(&path).expect("open the file to send"));
+
+        let (reset, resets) = mpsc::channel::<()>();
+        let (addr_tx, addr_rx) = mpsc::channel();
+        let peer = std::thread::spawn(move || {
+            let addr = addr_rx.recv().expect("listener address");
+            let socket = socket2::Socket::new(socket2::Domain::IPV4, socket2::Type::STREAM, None)
+                .expect("peer socket");
+            socket
+                .set_recv_buffer_size(BUFFER)
+                .expect("peer receive buffer");
+            socket
+                .connect(&socket2::SockAddr::from(addr))
+                .expect("connect");
+            resets.recv().expect("reset");
+            // A zero linger closes with a reset.
+            socket
+                .set_linger(Some(Duration::ZERO))
+                .expect("zero linger");
+        });
+
+        let (first, next) = run(async {
+            let io = http_kit_io::compio::Compio::default();
+            let listener = io.bind_tcp(super::loopback()).expect("bind");
+            addr_tx
+                .send(listener.local_addr().expect("local address"))
+                .expect("send address");
+            let (mut stream, _) = listener.accept().await.expect("accept");
+            socket2::SockRef::from(&stream)
+                .set_send_buffer_size(BUFFER)
+                .expect("send buffer");
+            let send = stream.send_file(file.clone(), 0, contents.len());
+            let watch = async {
+                // The round fills its pipe and waits on the full socket.
+                ::compio::time::sleep(Duration::from_millis(300)).await;
+                reset.send(()).expect("reset the peer");
+            };
+            let (BufResult(first, _), ()) = futures_lite::future::zip(send, watch).await;
+            let sent = *first.as_ref().expect("the round reports what it sent");
+            let BufResult(next, _) = stream
+                .send_file(file.clone(), sent as u64, contents.len() - sent)
+                .await;
+            (sent, next)
+        });
+        peer.join().expect("peer thread");
+        std::fs::remove_file(&path).expect("remove the sent file");
+        assert!(
+            first > 0 && first < contents.len(),
+            "part of the region was sent: {first} of {} bytes",
+            contents.len()
+        );
+        assert!(
+            next.is_err(),
+            "the next call reports the reset connection: {next:?}"
+        );
+    }
+
     /// A buffer that records when it is dropped.
     #[cfg(target_os = "linux")]
     struct Tracked {