From 89f4d133ba376834bba9dce016b4414dc8aaaf43 Mon Sep 17 00:00:00 2001 From: rifuki Date: Fri, 4 Sep 2026 18:33:54 +0700 Subject: [PATCH] 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` 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 | 37 +++++++------ tokio/src/io/util/read_line.rs | 8 ++- tokio/tests/io_lines.rs | 89 +++++++++++++++++++++++++++++++- tokio/tests/io_read_line.rs | 4 +- tokio/tests/io_read_to_string.rs | 8 ++- 5 files changed, 122 insertions(+), 24 deletions(-) diff --git a/tokio/src/io/util/lines.rs b/tokio/src/io/util/lines.rs index 8d723aaa7..6d4d201be 100644 --- a/tokio/src/io/util/lines.rs +++ b/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 { #[pin] reader: R, - buf: String, - bytes: Vec, - read: usize, + buf: Vec, } } @@ -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>> { 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))), + } } } diff --git a/tokio/src/io/util/read_line.rs b/tokio/src/io/util/read_line.rs index fdcbc417e..889eaf7d8 100644 --- a/tokio/src/io/util/read_line.rs +++ b/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( +fn read_line_internal( reader: Pin<&mut R>, cx: &mut Context<'_>, output: &mut String, diff --git a/tokio/tests/io_lines.rs b/tokio/tests/io_lines.rs index 9996d81ca..56fd27bea 100644 --- a/tokio/tests/io_lines.rs +++ b/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::().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())); +} diff --git a/tokio/tests/io_read_line.rs b/tokio/tests/io_read_line.rs index 15841c9b4..bbc984b45 100644 --- a/tokio/tests/io_read_line.rs +++ b/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::().unwrap(); + assert_eq!(utf8.valid_up_to(), 12); assert_eq!(line.as_str(), "Foo"); } diff --git a/tokio/tests/io_read_to_string.rs b/tokio/tests/io_read_to_string.rs index 80649353b..040f72692 100644 --- a/tokio/tests/io_read_to_string.rs +++ b/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::().unwrap(); + assert_eq!(utf8.valid_up_to(), 3); + } } assert_eq!(s, "abc");