refactor!: collecting a stream with an error raises an error (nushell#16738)

Sort of a follow up to nushell#16732, makes the stream-collecting error
handling universal.
This fixes similar bugs everywhere.

## Release notes summary - What our users need to know

> TODO(release-notes): Merge this with nushell#16732's notes, this just
generalizes the fix for all commands.

## Tasks after submitting
N/A

Author: Bahex

crates/nu-cmd-lang/src/core_commands/describe.rs

@@ -278,7 +278,7 @@ fn run(
                 let subtype = if options.no_collect {
                     Value::string("any", head)
                 } else {
-                    describe_value(stream.into_value(), head, engine_state)
+                    describe_value(stream.into_debug_value(), head, engine_state)
                 };
                 Value::record(
                     record! {
@@ -294,7 +294,7 @@ fn run(
             } else if options.no_collect {
                 Value::string("stream", head)
             } else {
-                let value = stream.into_value();
+                let value = stream.into_debug_value();
                 let base_description = value.get_type().to_string();
                 Value::string(format!("{base_description} (stream)"), head)
             }

crates/nu-command/src/filters/each.rs

@@ -163,16 +163,7 @@ fn each_map(value: Value, closure: &mut ClosureEval, head: Span) -> Result<Value
     let is_error = value.is_error();
     closure
         .run_with_value(value)
-        .and_then(|pipeline_data| match pipeline_data {
-            // TODO: Should collecting a stream with an error immediately raise the
-            // error by default (like we do here) be the default?
-            PipelineData::ListStream(stream, ..) => stream
-                .into_iter()
-                .map(Value::unwrap_error)
-                .collect::<Result<Vec<_>, _>>()
-                .map(|vals| Value::list(vals, head)),
-            data => data.into_value(head),
-        })
+        .and_then(|pipeline_data| pipeline_data.into_value(head))
         .map_err(|error| chain_error_with_input(error, is_error, span))
 }
 

crates/nu-command/src/path/join.rs

@@ -170,7 +170,7 @@ fn run(call: &Call, args: &Arguments, input: PipelineData) -> Result<PipelineDat
     match input {
         PipelineData::Value(val, md) => Ok(PipelineData::value(handle_value(val, args, head), md)),
         PipelineData::ListStream(stream, ..) => Ok(PipelineData::value(
-            handle_value(stream.into_value(), args, head),
+            handle_value(stream.into_value()?, args, head),
             metadata,
         )),
         PipelineData::ByteStream(stream, ..) => Ok(PipelineData::value(

crates/nu-protocol/src/pipeline/list_stream.rs

@@ -75,7 +75,20 @@ impl ListStream {
     }
 
     /// Collect the values of a [`ListStream`] into a list [`Value`].
-    pub fn into_value(self) -> Value {
+    ///
+    /// If any of the values in the stream is a [Value::Error], its inner [ShellError] is returned.
+    pub fn into_value(self) -> Result<Value, ShellError> {
+        Ok(Value::list(
+            self.stream
+                .map(Value::unwrap_error)
+                .collect::<Result<_, _>>()?,
+            self.span,
+        ))
+    }
+
+    /// Collect the values of a [`ListStream`] into a [`Value::List`], preserving [Value::Error]
+    /// items for debugging purposes.
+    pub fn into_debug_value(self) -> Value {
         Value::list(self.stream.collect(), self.span)
     }
 

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

@@ -192,7 +192,7 @@ impl PipelineData {
                     Ok(value)
                 }
             }
-            PipelineData::ListStream(stream, ..) => Ok(stream.into_value()),
+            PipelineData::ListStream(stream, ..) => stream.into_value(),
             PipelineData::ByteStream(stream, ..) => stream.into_value(),
         }
     }