io: read lines as bytes so a partial line survives an I/O error (#8400)

`Lines` accumulated into a `String`, so a partial line interrupted by an
I/O error could not be kept when it ended mid multi-byte character. The
next poll then tripped `debug_assert!(output.is_empty())`, or
underflowed `vector.len() - num_bytes_read` in `put_back_original_data`,
which panics in release builds too on the `.expect` below it.

`Lines` now holds a single `buf: Vec<u8>` and calls
`read_until_internal` directly the way `Split` does, converting to
`String` only once a whole line is available.

An `InvalidData` error now carries the utf-8 error itself rather than a
fixed string. `Lines` owns the line, so it hands over the whole
`FromUtf8Error` and the caller can still recover the bytes; `read_line`
and `read_to_string` have to put those bytes back into the caller's
`String`, so they carry `Utf8Error` instead.

`read_line_internal` has no callers outside `read_line.rs` and is now
private.

tokio/src/io/util/lines.rs

@@ -1,4 +1,4 @@
-use crate::io::util::read_line::read_line_internal;
+use crate::io::util::read_until::read_until_internal;
 use crate::io::AsyncBufRead;
 
 use pin_project_lite::pin_project;
@@ -23,9 +23,7 @@ pin_project! {
     pub struct Lines<R> {
         #[pin]
         reader: R,
-        buf: String,
-        bytes: Vec<u8>,
-        read: usize,
+        buf: Vec<u8>,
     }
 }
 
@@ -35,9 +33,7 @@ where
 {
     Lines {
         reader,
-        buf: String::new(),
-        bytes: Vec::new(),
-        read: 0,
+        buf: Vec::new(),
     }
 }
 
@@ -102,7 +98,13 @@ where
     ///  * `Poll::Pending` if the next line is not yet available.
     ///  * `Poll::Ready(Ok(Some(line)))` if the next line is available.
     ///  * `Poll::Ready(Ok(None))` if there are no more lines in this stream.
-    ///  * `Poll::Ready(Err(err))` if an IO error occurred while reading the next line.
+    ///  * `Poll::Ready(Err(err))` if an IO error occurred while reading the next line,
+    ///    or if the line was not valid UTF-8.
+    ///
+    /// A line that is not valid UTF-8 is returned as the [`FromUtf8Error`]
+    /// inside the error, and the next call reads the following line.
+    ///
+    /// [`FromUtf8Error`]: std::string::FromUtf8Error
     ///
     /// When the method returns `Poll::Pending`, the `Waker` in the provided
     /// `Context` is scheduled to receive a wakeup when more bytes become
@@ -115,22 +117,27 @@ where
     ) -> Poll<io::Result<Option<String>>> {
         let me = self.project();
 
-        let n = ready!(read_line_internal(me.reader, cx, me.buf, me.bytes, me.read))?;
-        debug_assert_eq!(*me.read, 0);
+        let mut read = 0;
+        let n = ready!(read_until_internal(me.reader, cx, b'\n', me.buf, &mut read))?;
 
         if n == 0 && me.buf.is_empty() {
             return Poll::Ready(Ok(None));
         }
 
-        if me.buf.ends_with('\n') {
-            me.buf.pop();
+        let mut bytes = mem::take(me.buf);
 
-            if me.buf.ends_with('\r') {
-                me.buf.pop();
+        if bytes.last() == Some(&b'\n') {
+            bytes.pop();
+
+            if bytes.last() == Some(&b'\r') {
+                bytes.pop();
             }
         }
 
-        Poll::Ready(Ok(Some(mem::take(me.buf))))
+        match String::from_utf8(bytes) {
+            Ok(line) => Poll::Ready(Ok(Some(line))),
+            Err(err) => Poll::Ready(Err(io::Error::new(io::ErrorKind::InvalidData, err))),
+        }
     }
 }
 

tokio/src/io/util/read_line.rs

@@ -76,12 +76,10 @@ pub(super) fn finish_string_read(
         }
         (Ok(num_bytes), Err(utf8_err)) => {
             debug_assert_eq!(read, 0);
+            let utf8_error = utf8_err.utf8_error();
             put_back_original_data(output, utf8_err.into_bytes(), num_bytes);
 
-            Poll::Ready(Err(io::Error::new(
-                io::ErrorKind::InvalidData,
-                "stream did not contain valid UTF-8",
-            )))
+            Poll::Ready(Err(io::Error::new(io::ErrorKind::InvalidData, utf8_error)))
         }
         (Err(io_err), Err(utf8_err)) => {
             put_back_original_data(output, utf8_err.into_bytes(), read);
@@ -91,7 +89,7 @@ pub(super) fn finish_string_read(
     }
 }
 
-pub(super) fn read_line_internal<R: AsyncBufRead + ?Sized>(
+fn read_line_internal<R: AsyncBufRead + ?Sized>(
     reader: Pin<&mut R>,
     cx: &mut Context<'_>,
     output: &mut String,

tokio/tests/io_lines.rs

@@ -1,7 +1,12 @@
 #![warn(rust_2018_idioms)]
 #![cfg(feature = "full")]
 
-use tokio::io::AsyncBufReadExt;
+use std::io::{Error, ErrorKind};
+use std::string::FromUtf8Error;
+use std::time::Duration;
+
+use tokio::io::{AsyncBufReadExt, BufReader};
+use tokio::time::timeout;
 use tokio_test::assert_ok;
 
 #[tokio::test]
@@ -17,3 +22,85 @@ async fn lines_inherent() {
     assert_eq!(b, "");
     assert!(assert_ok!(st.next_line().await).is_none());
 }
+
+#[tokio::test]
+async fn lines_keeps_partial_line_after_io_error() {
+    let mock = tokio_test::io::Builder::new()
+        .read(b"abc")
+        .read_error(Error::new(ErrorKind::Other, "boom"))
+        .read(b"def\nghi\n")
+        .build();
+    let mut lines = BufReader::new(mock).lines();
+
+    let err = lines.next_line().await.unwrap_err();
+    assert_eq!(err.kind(), ErrorKind::Other);
+
+    assert_eq!(lines.next_line().await.unwrap(), Some("abcdef".to_string()));
+    assert_eq!(lines.next_line().await.unwrap(), Some("ghi".to_string()));
+    assert_eq!(lines.next_line().await.unwrap(), None);
+}
+
+#[tokio::test]
+async fn lines_keeps_truncated_multibyte_char_after_io_error() {
+    let mock = tokio_test::io::Builder::new()
+        .read(b"ab\xc3")
+        .read_error(Error::new(ErrorKind::Other, "boom"))
+        .read(b"\xa9cd\nghi\n")
+        .build();
+    let mut lines = BufReader::new(mock).lines();
+
+    let err = lines.next_line().await.unwrap_err();
+    assert_eq!(err.kind(), ErrorKind::Other);
+
+    assert_eq!(lines.next_line().await.unwrap(), Some("abécd".to_string()));
+    assert_eq!(lines.next_line().await.unwrap(), Some("ghi".to_string()));
+    assert_eq!(lines.next_line().await.unwrap(), None);
+}
+
+#[tokio::test]
+async fn lines_invalid_utf8_line_errors_once_and_advances() {
+    let rd: &[u8] = b"ok\n\xff\xfe\nnext\n";
+    let mut lines = rd.lines();
+
+    assert_eq!(lines.next_line().await.unwrap(), Some("ok".to_string()));
+
+    let err = lines.next_line().await.unwrap_err();
+    assert_eq!(err.kind(), ErrorKind::InvalidData);
+
+    let inner = err.into_inner().unwrap();
+    let utf8 = inner.downcast::<FromUtf8Error>().unwrap();
+    assert_eq!(utf8.into_bytes(), b"\xff\xfe");
+
+    assert_eq!(lines.next_line().await.unwrap(), Some("next".to_string()));
+    assert_eq!(lines.next_line().await.unwrap(), None);
+}
+
+#[tokio::test]
+async fn lines_invalid_utf8_at_eof_does_not_loop_forever() {
+    let rd: &[u8] = b"ok\n\xff";
+    let mut lines = rd.lines();
+
+    assert_eq!(lines.next_line().await.unwrap(), Some("ok".to_string()));
+
+    let err = lines.next_line().await.unwrap_err();
+    assert_eq!(err.kind(), ErrorKind::InvalidData);
+
+    assert_eq!(lines.next_line().await.unwrap(), None);
+}
+
+#[tokio::test(start_paused = true)]
+async fn lines_next_line_is_cancel_safe() {
+    let mock = tokio_test::io::Builder::new()
+        .read(b"hello")
+        .wait(Duration::from_secs(1))
+        .read(b"\nworld\n")
+        .build();
+    let mut lines = BufReader::new(mock).lines();
+
+    assert!(timeout(Duration::from_millis(1), lines.next_line())
+        .await
+        .is_err());
+
+    assert_eq!(lines.next_line().await.unwrap(), Some("hello".to_string()));
+    assert_eq!(lines.next_line().await.unwrap(), Some("world".to_string()));
+}

tokio/tests/io_read_line.rs

@@ -2,6 +2,7 @@
 #![cfg(feature = "full")]
 
 use std::io::ErrorKind;
+use std::str::Utf8Error;
 use tokio::io::{AsyncBufReadExt, BufReader, Error};
 use tokio_test::{assert_ok, io::Builder};
 
@@ -69,7 +70,8 @@ async fn read_line_invalid_utf8() {
     let mut line = "Foo".to_string();
     let err = read.read_line(&mut line).await.expect_err("Should fail");
     assert_eq!(err.kind(), ErrorKind::InvalidData);
-    assert_eq!(err.to_string(), "stream did not contain valid UTF-8");
+    let utf8 = err.into_inner().unwrap().downcast::<Utf8Error>().unwrap();
+    assert_eq!(utf8.valid_up_to(), 12);
     assert_eq!(line.as_str(), "Foo");
 }
 

tokio/tests/io_read_to_string.rs

@@ -2,6 +2,7 @@
 #![cfg(feature = "full")]
 
 use std::io;
+use std::str::Utf8Error;
 use tokio::io::AsyncReadExt;
 use tokio_test::assert_ok;
 use tokio_test::io::Builder;
@@ -24,8 +25,11 @@ async fn to_string_does_not_truncate_on_utf8_error() {
 
     match AsyncReadExt::read_to_string(&mut data.as_slice(), &mut s).await {
         Ok(len) => panic!("Should fail: {len} bytes."),
-        Err(err) if err.to_string() == "stream did not contain valid UTF-8" => {}
-        Err(err) => panic!("Fail: {err}."),
+        Err(err) => {
+            assert_eq!(err.kind(), io::ErrorKind::InvalidData);
+            let utf8 = err.into_inner().unwrap().downcast::<Utf8Error>().unwrap();
+            assert_eq!(utf8.valid_up_to(), 3);
+        }
     }
 
     assert_eq!(s, "abc");