Fix inconsistent print behavior (#12675)

# Description

I found a bunch of issues relating to the specialized reimplementation
of `print()` that's done in `nu-cli` and it just didn't seem necessary.
So I tried to unify the behavior reasonably. `PipelineData::print()`
already handles the call to `table` and it even has a `no_newline`
option.

One of the most major issues before was that we were using the value
iterator, and then converting to string, and then printing each with
newlines. This doesn't work well for an external stream, because its
iterator ends up creating `Value::binary()` with each buffer... so we
were doing lossy UTF-8 conversion on those and then printing them with
newlines, which was very weird:


![Screenshot_2024-04-26_02-02-29](https://github.com/nushell/nushell/assets/10729/131c2224-08ee-4582-8617-6ecbb3ce8da5)

You can see the random newline inserted in a break between buffers, but
this would be even worse if it were on a multibyte UTF-8 character. You
can produce this by writing a large amount of text to a text file, and
then doing `nu -c 'open file.txt'` - in my case I just wrote `^find .`;
it just has to be large enough to trigger a buffer break.

Using `print()` instead led to a new issue though, because it doesn't
abort on errors. This is so that certain commands can produce a stream
of errors and have those all printed. There are tests for e.g. `rm` that
depend on this behavior. I assume we want to keep that, so instead I
made my target `BufferedReader`, and had that fuse closed if an error
was encountered. I can't imagine we want to keep reading from a wrapped
I/O stream if an error occurs; more often than not the error isn't going
to magically resolve itself, it's not going to be a different error each
time, and it's just going to lead to an infinite stream of the same
error.

The test that broke without that was `open . | lines`, because `lines`
doesn't fuse closed on error. But I don't know if it's expected or not
for it to do that, so I didn't target that.

I think this PR makes things better but I'll keep looking for ways to
improve on how errors and streams interact, especially trying to
eliminate cases where infinite error loops can happen.

# User-Facing Changes
- **Breaking**: `BufferedReader` changes + no more public fields
- A raw I/O stream from e.g. `open` won't produce infinite errors
anymore, but I consider that to be a plus
- the implicit `print` on script output is the same as the normal one
now

# Tests + Formatting
Everything passes but I didn't add anything specific.

crates/nu-cli/src/eval_file.rs

@@ -5,13 +5,11 @@ use nu_engine::{convert_env_values, current_dir, eval_block};
 use nu_parser::parse;
 use nu_path::canonicalize_with;
 use nu_protocol::{
-    ast::Call,
     debugger::WithoutDebug,
     engine::{EngineState, Stack, StateWorkingSet},
     report_error, Config, PipelineData, ShellError, Span, Value,
 };
-use nu_utils::stdout_write_all_and_flush;
-use std::sync::Arc;
+use std::{io::Write, sync::Arc};
 
 /// Entry point for evaluating a file.
 ///
@@ -210,29 +208,8 @@ pub(crate) fn print_table_or_error(
         std::process::exit(1);
     }
 
-    if let Some(decl_id) = engine_state.find_decl("table".as_bytes(), &[]) {
-        let command = engine_state.get_decl(decl_id);
-        if command.get_block_id().is_some() {
-            print_or_exit(pipeline_data, engine_state, config, no_newline);
-        } else {
-            // The final call on table command, it's ok to set redirect_output to false.
-            let call = Call::new(Span::new(0, 0));
-            let table = command.run(engine_state, stack, &call, pipeline_data);
-
-            match table {
-                Ok(table) => {
-                    print_or_exit(table, engine_state, config, no_newline);
-                }
-                Err(error) => {
-                    let working_set = StateWorkingSet::new(engine_state);
-                    report_error(&working_set, &error);
-                    std::process::exit(1);
-                }
-            }
-        }
-    } else {
-        print_or_exit(pipeline_data, engine_state, config, no_newline);
-    }
+    // We don't need to do anything special to print a table because print() handles it
+    print_or_exit(pipeline_data, engine_state, stack, no_newline);
 
     // Make sure everything has finished
     if let Some(exit_code) = exit_code {
@@ -250,23 +227,19 @@ pub(crate) fn print_table_or_error(
 
 fn print_or_exit(
     pipeline_data: PipelineData,
-    engine_state: &mut EngineState,
-    config: &Config,
+    engine_state: &EngineState,
+    stack: &mut Stack,
     no_newline: bool,
 ) {
-    for item in pipeline_data {
-        if let Value::Error { error, .. } = item {
-            let working_set = StateWorkingSet::new(engine_state);
+    let result = pipeline_data.print(engine_state, stack, no_newline, false);
 
-            report_error(&working_set, &*error);
+    let _ = std::io::stdout().flush();
+    let _ = std::io::stderr().flush();
 
-            std::process::exit(1);
-        }
-
-        let mut out = item.to_expanded_string("\n", config);
-        if !no_newline {
-            out.push('\n');
-        }
-        let _ = stdout_write_all_and_flush(out).map_err(|err| eprintln!("{err}"));
+    if let Err(error) = result {
+        let working_set = StateWorkingSet::new(engine_state);
+        report_error(&working_set, &error);
+        let _ = std::io::stderr().flush();
+        std::process::exit(1);
     }
 }

crates/nu-command/src/filesystem/open.rs

@@ -145,7 +145,7 @@ impl Command for Open {
 
                     let file_contents = PipelineData::ExternalStream {
                         stdout: Some(RawStream::new(
-                            Box::new(BufferedReader { input: buf_reader }),
+                            Box::new(BufferedReader::new(buf_reader)),
                             ctrlc.clone(),
                             call_span,
                             None,

crates/nu-command/src/network/http/client.rs

@@ -123,9 +123,7 @@ pub fn response_to_buffer(
 
     PipelineData::ExternalStream {
         stdout: Some(RawStream::new(
-            Box::new(BufferedReader {
-                input: buffered_input,
-            }),
+            Box::new(BufferedReader::new(buffered_input)),
             engine_state.ctrlc.clone(),
             span,
             buffer_size,

crates/nu-protocol/src/pipeline_data/mod.rs

@@ -998,8 +998,16 @@ pub fn print_if_stream(
                     if nu_utils::ctrl_c::was_pressed(&ctrlc) {
                         break;
                     }
-                    if let Ok(bytes) = bytes {
-                        let _ = stderr.write_all(&bytes);
+                    match bytes {
+                        Ok(bytes) => {
+                            let _ = stderr.write_all(&bytes);
+                        }
+                        Err(err) => {
+                            // we don't have access to EngineState, but maybe logging the debug
+                            // impl is better than nothing
+                            eprintln!("Error in stderr stream: {err:?}");
+                            break;
+                        }
                     }
                 }
             })?;

crates/nu-protocol/src/util.rs

@@ -2,12 +2,20 @@ use crate::ShellError;
 use std::io::{BufRead, BufReader, Read};
 
 pub struct BufferedReader<R: Read> {
-    pub input: BufReader<R>,
+    input: BufReader<R>,
+    error: bool,
 }
 
 impl<R: Read> BufferedReader<R> {
     pub fn new(input: BufReader<R>) -> Self {
-        Self { input }
+        Self {
+            input,
+            error: false,
+        }
+    }
+
+    pub fn into_inner(self) -> BufReader<R> {
+        self.input
     }
 }
 
@@ -15,6 +23,11 @@ impl<R: Read> Iterator for BufferedReader<R> {
     type Item = Result<Vec<u8>, ShellError>;
 
     fn next(&mut self) -> Option<Self::Item> {
+        // Don't try to read more data if an error occurs
+        if self.error {
+            return None;
+        }
+
         let buffer = self.input.fill_buf();
         match buffer {
             Ok(s) => {
@@ -30,7 +43,10 @@ impl<R: Read> Iterator for BufferedReader<R> {
                     Some(Ok(result))
                 }
             }
-            Err(e) => Some(Err(ShellError::IOError { msg: e.to_string() })),
+            Err(e) => {
+                self.error = true;
+                Some(Err(ShellError::IOError { msg: e.to_string() }))
+            }
         }
     }
 }