paste: avoid unbounded buffering for single input (#11060)

src/uu/paste/src/paste.rs

@@ -7,7 +7,7 @@ use clap::{Arg, ArgAction, Command};
 use std::cell::{OnceCell, RefCell};
 use std::ffi::OsString;
 use std::fs::File;
-use std::io::{BufRead, BufReader, Stdin, Write, stdin, stdout};
+use std::io::{BufRead, BufReader, Read, Stdin, Write, stdin, stdout};
 use std::iter::Cycle;
 use std::path::Path;
 use std::rc::Rc;
@@ -111,12 +111,24 @@ fn paste(
         input_source_vec.push(input_source);
     }
 
+    let line_ending_byte = u8::from(line_ending);
+    let input_source_vec_len = input_source_vec.len();
     let mut stdout = stdout().lock();
 
-    let line_ending_byte = u8::from(line_ending);
-    let line_ending_byte_array_ref = &[line_ending_byte];
+    if !serial && input_source_vec_len == 1 {
+        // With a single input source (no -s), `paste` output is identical to input,
+        // except that a missing final line ending must be added.
+        // Stream directly to avoid unbounded line buffering on inputs like /dev/zero.
+        return write_single_input_source(
+            &mut stdout,
+            input_source_vec
+                .pop()
+                .expect("input_source_vec_len was checked to be exactly one"),
+            line_ending_byte,
+        );
+    }
 
-    let input_source_vec_len = input_source_vec.len();
+    let line_ending_byte_array_ref = &[line_ending_byte];
 
     let mut delimiter_state = DelimiterState::new(&unescaped_and_encoded_delimiters);
 
@@ -187,6 +199,35 @@ fn paste(
     Ok(())
 }
 
+fn write_single_input_source(
+    writer: &mut impl Write,
+    mut input_source: InputSource,
+    line_ending_byte: u8,
+) -> UResult<()> {
+    let mut buffer = [0_u8; 8 * 1024];
+    let mut has_data = false;
+    let mut last_byte = line_ending_byte;
+
+    loop {
+        let bytes_read = input_source.read(&mut buffer)?;
+
+        if bytes_read == 0 {
+            break;
+        }
+
+        has_data = true;
+        last_byte = buffer[bytes_read - 1];
+
+        writer.write_all(&buffer[..bytes_read])?;
+    }
+
+    if has_data && last_byte != line_ending_byte {
+        writer.write_all(&[line_ending_byte])?;
+    }
+
+    Ok(())
+}
+
 fn parse_delimiters(delimiters: &OsString) -> UResult<Box<[Box<[u8]>]>> {
     let bytes = uucore::os_str_as_bytes(delimiters)?;
     let mut vec = Vec::<Box<[u8]>>::with_capacity(bytes.len());
@@ -341,6 +382,21 @@ enum InputSource {
 }
 
 impl InputSource {
+    fn read(&mut self, buf: &mut [u8]) -> UResult<usize> {
+        let us = match self {
+            Self::File(bu) => bu.read(buf)?,
+            Self::StandardInput(rc) => rc
+                .try_borrow()
+                .map_err(|bo| {
+                    USimpleError::new(1, translate!("paste-error-stdin-borrow", "error" => bo))
+                })?
+                .lock()
+                .read(buf)?,
+        };
+
+        Ok(us)
+    }
+
     fn read_until(&mut self, byte: u8, buf: &mut Vec<u8>) -> UResult<usize> {
         let us = match self {
             Self::File(bu) => bu.read_until(byte, buf)?,

tests/by-util/test_paste.rs

@@ -446,3 +446,46 @@ fn test_paste_non_utf8_paths() {
         .succeeds()
         .stdout_is("line1\tcol1\nline2\tcol2\n");
 }
+
+#[cfg(target_os = "linux")]
+fn make_broken_pipe() -> std::fs::File {
+    use std::os::unix::io::FromRawFd;
+
+    let mut fds: [libc::c_int; 2] = [0, 0];
+    assert_eq!(
+        unsafe { libc::pipe(fds.as_mut_ptr()) },
+        0,
+        "Failed to create pipe"
+    );
+
+    // Drop the read end so writes fail with EPIPE.
+    let _ = unsafe { std::fs::File::from_raw_fd(fds[0]) };
+
+    unsafe { std::fs::File::from_raw_fd(fds[1]) }
+}
+
+#[test]
+#[cfg(target_os = "linux")]
+fn test_dev_zero_write_error_dev_full() {
+    use std::fs::File;
+
+    let dev_full =
+        File::create("/dev/full").expect("Failed to open /dev/full - test must run on Linux");
+
+    new_ucmd!()
+        .arg("/dev/zero")
+        .set_stdout(dev_full)
+        .fails()
+        .code_is(1)
+        .stderr_contains("No space left on device");
+}
+
+#[test]
+#[cfg(target_os = "linux")]
+fn test_dev_zero_closed_pipe() {
+    new_ucmd!()
+        .arg("/dev/zero")
+        .set_stdout(make_broken_pipe())
+        .run()
+        .fails_silently();
+}