Replace panics with errors in thread spawning (#12040)

# Description
Replace panics with errors in thread spawning.

Also adds `IntoSpanned` trait for easily constructing `Spanned`, and an
implementation of `From<Spanned<std::io::Error>>` for `ShellError`,
which is used to provide context for the error wherever there was a span
conveniently available. In general this should make it more convenient
to do the right thing with `std::io::Error` and always add a span to it
when it's possible to do so.

# User-Facing Changes
Fewer panics!

# Tests + Formatting
- 🟢 `toolkit fmt`
- 🟢 `toolkit clippy`
- 🟢 `toolkit test`
- 🟢 `toolkit test stdlib`

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

@@ -4,8 +4,8 @@ use nu_engine::{eval_block_with_early_return, redirect_env, CallExt};
 use nu_protocol::ast::Call;
 use nu_protocol::engine::{Closure, Command, EngineState, Stack};
 use nu_protocol::{
-    Category, Example, ListStream, PipelineData, RawStream, ShellError, Signature, SyntaxShape,
+    Category, Example, IntoSpanned, ListStream, PipelineData, RawStream, ShellError, Signature,
-    Type, Value,
+    SyntaxShape, Type, Value,
 };
 
 #[derive(Clone)]
@@ -147,7 +147,8 @@ impl Command for Do {
                 // consumes the first 65535 bytes
                 // So we need a thread to receive stdout message, then the current thread can continue to consume
                 // stderr messages.
-                let stdout_handler = stdout.map(|stdout_stream| {
+                let stdout_handler = stdout
+                    .map(|stdout_stream| {
                         thread::Builder::new()
                             .name("stderr redirector".to_string())
                             .spawn(move || {
@@ -162,8 +163,9 @@ impl Command for Do {
                                     None,
                                 )
                             })
-                        .expect("Failed to create thread")
+                            .map_err(|e| e.into_spanned(call.head))
-                });
+                    })
+                    .transpose()?;
 
                 // Intercept stderr so we can return it in the error if the exit code is non-zero.
                 // The threading issues mentioned above dictate why we also need to intercept stdout.

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

@@ -3,6 +3,7 @@ use nu_engine::CallExt;
 use nu_path::expand_path_with;
 use nu_protocol::ast::{Call, Expr, Expression};
 use nu_protocol::engine::{Command, EngineState, Stack};
+use nu_protocol::IntoSpanned;
 use nu_protocol::{
     Category, DataSource, Example, PipelineData, PipelineMetadata, RawStream, ShellError,
     Signature, Span, Spanned, SyntaxShape, Type, Value,
@@ -123,19 +124,22 @@ impl Command for Save {
                 )?;
 
                 // delegate a thread to redirect stderr to result.
-                let handler = stderr.map(|stderr_stream| match stderr_file {
+                let handler = stderr
+                    .map(|stderr_stream| match stderr_file {
                         Some(stderr_file) => thread::Builder::new()
                             .name("stderr redirector".to_string())
-                        .spawn(move || stream_to_file(stderr_stream, stderr_file, span, progress))
+                            .spawn(move || {
-                        .expect("Failed to create thread"),
+                                stream_to_file(stderr_stream, stderr_file, span, progress)
+                            }),
                         None => thread::Builder::new()
                             .name("stderr redirector".to_string())
                             .spawn(move || {
                                 let _ = stderr_stream.into_bytes();
                                 Ok(PipelineData::empty())
+                            }),
                     })
-                        .expect("Failed to create thread"),
+                    .transpose()
-                });
+                    .map_err(|e| e.into_spanned(span))?;
 
                 let res = stream_to_file(stream, file, span, progress);
                 if let Some(h) = handler {

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

@@ -4,8 +4,8 @@ use nu_engine::{eval_block_with_early_return, CallExt};
 use nu_protocol::{
     ast::Call,
     engine::{Closure, Command, EngineState, Stack},
-    Category, Example, IntoInterruptiblePipelineData, PipelineData, RawStream, ShellError,
+    Category, Example, IntoInterruptiblePipelineData, IntoSpanned, PipelineData, RawStream,
-    Signature, Spanned, SyntaxShape, Type, Value,
+    ShellError, Signature, Spanned, SyntaxShape, Type, Value,
 };
 
 #[derive(Clone)]
@@ -128,8 +128,10 @@ use it in your pipeline."#
 
                 if use_stderr {
                     if let Some(stderr) = stderr {
+                        let iter = tee(stderr.stream, with_stream)
+                            .map_err(|e| e.into_spanned(call.head))?;
                         let raw_stream = RawStream::new(
-                            Box::new(tee(stderr.stream, with_stream).map(flatten_result)),
+                            Box::new(iter.map(flatten_result)),
                             stderr.ctrlc,
                             stderr.span,
                             stderr.known_size,
@@ -158,14 +160,18 @@ use it in your pipeline."#
                         })
                     }
                 } else {
-                    let stdout = stdout.map(|stdout| {
+                    let stdout = stdout
-                        RawStream::new(
+                        .map(|stdout| {
-                            Box::new(tee(stdout.stream, with_stream).map(flatten_result)),
+                            let iter = tee(stdout.stream, with_stream)
+                                .map_err(|e| e.into_spanned(call.head))?;
+                            Ok::<_, ShellError>(RawStream::new(
+                                Box::new(iter.map(flatten_result)),
                                 stdout.ctrlc,
                                 stdout.span,
                                 stdout.known_size,
-                        )
+                            ))
-                    });
+                        })
+                        .transpose()?;
                     Ok(PipelineData::ExternalStream {
                         stdout,
                         stderr,
@@ -201,6 +207,7 @@ use it in your pipeline."#
                     // Make sure to drain any iterator produced to avoid unexpected behavior
                     result.and_then(|data| data.drain())
                 })
+                .map_err(|e| e.into_spanned(call.head))?
                 .map(move |result| result.unwrap_or_else(|err| Value::error(err, closure_span)))
                 .into_pipeline_data_with_metadata(metadata, engine_state.ctrlc.clone());
 
@@ -227,7 +234,7 @@ fn flatten_result<T, E>(result: Result<Result<T, E>, E>) -> Result<T, E> {
 fn tee<T>(
     input: impl Iterator<Item = T>,
     with_cloned_stream: impl FnOnce(mpsc::Receiver<T>) -> Result<(), ShellError> + Send + 'static,
-) -> impl Iterator<Item = Result<T, ShellError>>
+) -> Result<impl Iterator<Item = Result<T, ShellError>>, std::io::Error>
 where
     T: Clone + Send + 'static,
 {
@@ -237,14 +244,13 @@ where
     let mut thread = Some(
         thread::Builder::new()
             .name("stderr consumer".into())
-            .spawn(move || with_cloned_stream(rx))
+            .spawn(move || with_cloned_stream(rx))?,
-            .expect("could not create thread"),
     );
 
     let mut iter = input.into_iter();
     let mut tx = Some(tx);
 
-    std::iter::from_fn(move || {
+    Ok(std::iter::from_fn(move || {
         if thread.as_ref().is_some_and(|t| t.is_finished()) {
             // Check for an error from the other thread
             let result = thread
@@ -274,7 +280,7 @@ where
                     .map(Err)
             })
         }
-    })
+    }))
 }
 
 #[test]
@@ -289,6 +295,7 @@ fn tee_copies_values_to_other_thread_and_passes_them_through() {
         }
         Ok(())
     })
+    .expect("io error")
     .collect::<Result<Vec<i32>, ShellError>>()
     .expect("should not produce error");
 
@@ -305,7 +312,8 @@ fn tee_forwards_errors_back_immediately() {
     let slow_input = (0..100).inspect(|_| std::thread::sleep(Duration::from_millis(1)));
     let iter = tee(slow_input, |_| {
         Err(ShellError::IOError { msg: "test".into() })
-    });
+    })
+    .expect("io error");
     for result in iter {
         if let Ok(val) = result {
             // should not make it to the end
@@ -331,7 +339,8 @@ fn tee_waits_for_the_other_thread() {
         std::thread::sleep(Duration::from_millis(10));
         waited_clone.store(true, Ordering::Relaxed);
         Err(ShellError::IOError { msg: "test".into() })
-    });
+    })
+    .expect("io error");
     let last = iter.last();
     assert!(waited.load(Ordering::Relaxed), "failed to wait");
     assert!(

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

@@ -283,7 +283,7 @@ fn send_cancellable_request(
             let ret = request_fn();
             let _ = tx.send(ret); // may fail if the user has cancelled the operation
         })
-        .expect("Failed to create thread");
+        .map_err(ShellError::from)?;
 
     // ...and poll the channel for responses
     loop {

crates/nu-command/src/system/complete.rs

@@ -1,7 +1,8 @@
 use nu_protocol::{
     ast::Call,
     engine::{Command, EngineState, Stack},
-    Category, Example, IntoPipelineData, PipelineData, Record, ShellError, Signature, Type, Value,
+    Category, Example, IntoPipelineData, IntoSpanned, PipelineData, Record, ShellError, Signature,
+    Type, Value,
 };
 
 use std::thread;
@@ -52,9 +53,9 @@ impl Command for Complete {
                 // consumes the first 65535 bytes
                 // So we need a thread to receive stderr message, then the current thread can continue to consume
                 // stdout messages.
-                let stderr_handler = stderr.map(|stderr| {
+                let stderr_handler = stderr
+                    .map(|stderr| {
                         let stderr_span = stderr.span;
-                    (
                         thread::Builder::new()
                             .name("stderr consumer".to_string())
                             .spawn(move || {
@@ -65,10 +66,10 @@ impl Command for Complete {
                                     Ok::<_, ShellError>(Value::binary(stderr.item, stderr.span))
                                 }
                             })
-                            .expect("failed to create thread"),
+                            .map(|handle| (handle, stderr_span))
-                        stderr_span,
+                            .map_err(|err| err.into_spanned(call.head))
-                    )
+                    })
-                });
+                    .transpose()?;
 
                 if let Some(stdout) = stdout {
                     let stdout = stdout.into_bytes()?;

crates/nu-command/src/system/run_external.rs

@@ -2,6 +2,7 @@ use nu_cmd_base::hook::eval_hook;
 use nu_engine::env_to_strings;
 use nu_engine::eval_expression;
 use nu_engine::CallExt;
+use nu_protocol::IntoSpanned;
 use nu_protocol::NuGlob;
 use nu_protocol::{
     ast::{Call, Expr},
@@ -438,7 +439,7 @@ impl ExternalCommand {
 
                                 Ok(())
                             })
-                            .expect("Failed to create thread");
+                            .map_err(|e| e.into_spanned(head))?;
                     }
                 }
 
@@ -526,7 +527,7 @@ impl ExternalCommand {
                             Ok(())
                         }
                     }
-                }).expect("Failed to create thread");
+                }).map_err(|e| e.into_spanned(head))?;
 
                 let (stderr_tx, stderr_rx) = mpsc::sync_channel(OUTPUT_BUFFERS_IN_FLIGHT);
                 if redirect_stderr {
@@ -543,7 +544,7 @@ impl ExternalCommand {
                             read_and_redirect_message(stderr, stderr_tx, stderr_ctrlc);
                             Ok::<(), ShellError>(())
                         })
-                        .expect("Failed to create thread");
+                        .map_err(|e| e.into_spanned(head))?;
                 }
 
                 let stdout_receiver = ChannelReceiver::new(stdout_rx);

crates/nu-engine/src/eval.rs

@@ -7,8 +7,8 @@ use nu_protocol::{
     },
     engine::{Closure, EngineState, Stack},
     eval_base::Eval,
-    Config, DeclId, IntoPipelineData, PipelineData, RawStream, ShellError, Span, Spanned, Type,
+    Config, DeclId, IntoPipelineData, IntoSpanned, PipelineData, RawStream, ShellError, Span,
-    Value, VarId, ENV_VARIABLE_ID,
+    Spanned, Type, Value, VarId, ENV_VARIABLE_ID,
 };
 use std::thread::{self, JoinHandle};
 use std::{borrow::Cow, collections::HashMap};
@@ -542,7 +542,7 @@ fn eval_element_with_input(
                                     stderr_stack,
                                     save_call,
                                     input,
-                                ));
+                                )?);
                                 let (result_out_stream, result_err_stream) = if result_is_out {
                                     (result_out_stream, None)
                                 } else {
@@ -1090,8 +1090,9 @@ impl DataSaveJob {
         mut stack: Stack,
         save_call: Call,
         input: PipelineData,
-    ) -> Self {
+    ) -> Result<Self, ShellError> {
-        Self {
+        let span = save_call.head;
+        Ok(Self {
             inner: thread::Builder::new()
                 .name("stderr saver".to_string())
                 .spawn(move || {
@@ -1100,8 +1101,8 @@ impl DataSaveJob {
                         eprintln!("WARNING: error occurred when redirect to stderr: {:?}", err);
                     }
                 })
-                .expect("Failed to create thread"),
+                .map_err(|e| e.into_spanned(span))?,
-        }
+        })
     }
 
     pub fn join(self) -> thread::Result<()> {

crates/nu-plugin/src/plugin/interface.rs

@@ -369,18 +369,22 @@ where
                 exit_code,
             } => {
                 thread::scope(|scope| {
-                    let stderr_thread = stderr.map(|(mut writer, stream)| {
+                    let stderr_thread = stderr
+                        .map(|(mut writer, stream)| {
                             thread::Builder::new()
                                 .name("plugin stderr writer".into())
-                            .spawn_scoped(scope, move || writer.write_all(raw_stream_iter(stream)))
+                                .spawn_scoped(scope, move || {
-                            .expect("failed to spawn thread")
+                                    writer.write_all(raw_stream_iter(stream))
-                    });
+                                })
-                    let exit_code_thread = exit_code.map(|(mut writer, stream)| {
+                        })
+                        .transpose()?;
+                    let exit_code_thread = exit_code
+                        .map(|(mut writer, stream)| {
                             thread::Builder::new()
                                 .name("plugin exit_code writer".into())
                                 .spawn_scoped(scope, move || writer.write_all(stream))
-                            .expect("failed to spawn thread")
+                        })
-                    });
+                        .transpose()?;
                     // Optimize for stdout: if only stdout is present, don't spawn any other
                     // threads.
                     if let Some((mut writer, stream)) = stdout {
@@ -407,10 +411,12 @@ where
 
     /// Write all of the data in each of the streams. This method returns immediately; any necessary
     /// write will happen in the background. If a thread was spawned, its handle is returned.
-    pub(crate) fn write_background(self) -> Option<thread::JoinHandle<Result<(), ShellError>>> {
+    pub(crate) fn write_background(
+        self,
+    ) -> Result<Option<thread::JoinHandle<Result<(), ShellError>>>, ShellError> {
         match self {
-            PipelineDataWriter::None => None,
+            PipelineDataWriter::None => Ok(None),
-            _ => Some(
+            _ => Ok(Some(
                 thread::Builder::new()
                     .name("plugin stream background writer".into())
                     .spawn(move || {
@@ -421,9 +427,8 @@ where
                             log::warn!("Error while writing pipeline in background: {err}");
                         }
                         result
-                    })
+                    })?,
-                    .expect("failed to spawn thread"),
+            )),
-            ),
         }
     }
 }

crates/nu-plugin/src/plugin/interface/plugin.rs

@@ -419,7 +419,7 @@ impl PluginInterface {
         let (writer, rx) = self.write_plugin_call(call, context.clone())?;
 
         // Finish writing stream in the background
-        writer.write_background();
+        writer.write_background()?;
 
         self.receive_plugin_call_response(rx)
     }

crates/nu-plugin/src/plugin/mod.rs

@@ -126,7 +126,7 @@ fn make_plugin_interface(
         .stdin
         .take()
         .ok_or_else(|| ShellError::PluginFailedToLoad {
-            msg: "plugin missing stdin writer".into(),
+            msg: "Plugin missing stdin writer".into(),
         })?;
 
     let mut stdout = child
@@ -158,7 +158,9 @@ fn make_plugin_interface(
             drop(manager);
             let _ = child.wait();
         })
-        .expect("failed to spawn thread");
+        .map_err(|err| ShellError::PluginFailedToLoad {
+            msg: format!("Failed to spawn thread for plugin: {err}"),
+        })?;
 
     Ok(interface)
 }
@@ -422,37 +424,7 @@ pub fn serve_plugin(plugin: &mut impl StreamingPlugin, encoder: impl PluginEncod
     // We need to hold on to the interface to keep the manager alive. We can drop it at the end
     let interface = manager.get_interface();
 
-    // Try an operation that could result in ShellError. Exit if an I/O error is encountered.
+    // Determine the plugin name, for errors
-    // Try to report the error to nushell otherwise, and failing that, panic.
-    macro_rules! try_or_report {
-        ($interface:expr, $expr:expr) => (match $expr {
-            Ok(val) => val,
-            // Just exit if there is an I/O error. Most likely this just means that nushell
-            // interrupted us. If not, the error probably happened on the other side too, so we
-            // don't need to also report it.
-            Err(ShellError::IOError { .. }) => std::process::exit(1),
-            // If there is another error, try to send it to nushell and then exit.
-            Err(err) => {
-                let _ = $interface.write_response(Err(err.clone())).unwrap_or_else(|_| {
-                    // If we can't send it to nushell, panic with it so at least we get the output
-                    panic!("{}", err)
-                });
-                std::process::exit(1)
-            }
-        })
-    }
-
-    // Send Hello message
-    try_or_report!(interface, interface.hello());
-
-    // Spawn the reader thread
-    std::thread::Builder::new()
-        .name("engine interface reader".into())
-        .spawn(move || {
-            if let Err(err) = manager.consume_all((std::io::stdin().lock(), encoder)) {
-                // Do our best to report the read error. Most likely there is some kind of
-                // incompatibility between the plugin and nushell, so it makes more sense to try to
-                // report it on stderr than to send something.
     let exe = std::env::current_exe().ok();
 
     let plugin_name: String = exe
@@ -466,11 +438,49 @@ pub fn serve_plugin(plugin: &mut impl StreamingPlugin, encoder: impl PluginEncod
         })
         .unwrap_or_else(|| "(unknown)".into());
 
-                eprintln!("Plugin `{plugin_name}` read error: {err}");
+    // Try an operation that could result in ShellError. Exit if an I/O error is encountered.
+    // Try to report the error to nushell otherwise, and failing that, panic.
+    macro_rules! try_or_report {
+        ($interface:expr, $expr:expr) => (match $expr {
+            Ok(val) => val,
+            // Just exit if there is an I/O error. Most likely this just means that nushell
+            // interrupted us. If not, the error probably happened on the other side too, so we
+            // don't need to also report it.
+            Err(ShellError::IOError { .. }) => std::process::exit(1),
+            // If there is another error, try to send it to nushell and then exit.
+            Err(err) => {
+                let _ = $interface.write_response(Err(err.clone())).unwrap_or_else(|_| {
+                    // If we can't send it to nushell, panic with it so at least we get the output
+                    panic!("Plugin `{plugin_name}`: {}", err)
+                });
+                std::process::exit(1)
+            }
+        })
+    }
+
+    // Send Hello message
+    try_or_report!(interface, interface.hello());
+
+    let plugin_name_clone = plugin_name.clone();
+
+    // Spawn the reader thread
+    std::thread::Builder::new()
+        .name("engine interface reader".into())
+        .spawn(move || {
+            if let Err(err) = manager.consume_all((std::io::stdin().lock(), encoder)) {
+                // Do our best to report the read error. Most likely there is some kind of
+                // incompatibility between the plugin and nushell, so it makes more sense to try to
+                // report it on stderr than to send something.
+
+                eprintln!("Plugin `{plugin_name_clone}` read error: {err}");
                 std::process::exit(1);
             }
         })
-        .expect("failed to spawn thread");
+        .unwrap_or_else(|err| {
+            // If we fail to spawn the reader thread, we should exit
+            eprintln!("Plugin `{plugin_name}` failed to launch: {err}");
+            std::process::exit(1);
+        });
 
     for plugin_call in call_receiver {
         match plugin_call {
@@ -492,7 +502,7 @@ pub fn serve_plugin(plugin: &mut impl StreamingPlugin, encoder: impl PluginEncod
                 let result = plugin.run(&name, &config, &call, input);
                 let write_result = engine
                     .write_response(result)
-                    .map(|writer| writer.write_background());
+                    .and_then(|writer| writer.write_background());
                 try_or_report!(engine, write_result);
             }
             // Do an operation on a custom value
@@ -514,7 +524,7 @@ pub fn serve_plugin(plugin: &mut impl StreamingPlugin, encoder: impl PluginEncod
                             .map(|value| PipelineData::Value(value, None));
                         let write_result = engine
                             .write_response(result)
-                            .map(|writer| writer.write_background());
+                            .and_then(|writer| writer.write_background());
                         try_or_report!(engine, write_result);
                     }
                 }

crates/nu-protocol/src/pipeline_data.rs

@@ -868,8 +868,7 @@ pub fn print_if_stream(
                         let _ = stderr.write_all(&bytes);
                     }
                 }
-            })
+            })?;
-            .expect("could not create thread");
     }
 
     if let Some(stream) = stream {

crates/nu-protocol/src/shell_error.rs

@@ -2,7 +2,9 @@ use miette::Diagnostic;
 use serde::{Deserialize, Serialize};
 use thiserror::Error;
 
-use crate::{ast::Operator, engine::StateWorkingSet, format_error, ParseError, Span, Value};
+use crate::{
+    ast::Operator, engine::StateWorkingSet, format_error, ParseError, Span, Spanned, Value,
+};
 
 /// The fundamental error type for the evaluation engine. These cases represent different kinds of errors
 /// the evaluator might face, along with helpful spans to label. An error renderer will take this error value
@@ -1361,6 +1363,15 @@ impl From<std::io::Error> for ShellError {
     }
 }
 
+impl From<Spanned<std::io::Error>> for ShellError {
+    fn from(error: Spanned<std::io::Error>) -> Self {
+        ShellError::IOErrorSpanned {
+            msg: error.item.to_string(),
+            span: error.span,
+        }
+    }
+}
+
 impl std::convert::From<Box<dyn std::error::Error>> for ShellError {
     fn from(input: Box<dyn std::error::Error>) -> ShellError {
         ShellError::IOError {

crates/nu-protocol/src/span.rs

@@ -3,14 +3,33 @@ use serde::{Deserialize, Serialize};
 
 /// A spanned area of interest, generic over what kind of thing is of interest
 #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
-pub struct Spanned<T>
+pub struct Spanned<T> {
-where
-    T: Clone + std::fmt::Debug,
-{
     pub item: T,
     pub span: Span,
 }
 
+/// Helper trait to create [`Spanned`] more ergonomically.
+pub trait IntoSpanned: Sized {
+    /// Wrap items together with a span into [`Spanned`].
+    ///
+    /// # Example
+    ///
+    /// ```
+    /// # use nu_protocol::{Span, IntoSpanned};
+    /// # let span = Span::test_data();
+    /// let spanned = "Hello, world!".into_spanned(span);
+    /// assert_eq!("Hello, world!", spanned.item);
+    /// assert_eq!(span, spanned.span);
+    /// ```
+    fn into_spanned(self, span: Span) -> Spanned<Self>;
+}
+
+impl<T> IntoSpanned for T {
+    fn into_spanned(self, span: Span) -> Spanned<Self> {
+        Spanned { item: self, span }
+    }
+}
+
 /// Spans are a global offset across all seen files, which are cached in the engine's state. The start and
 /// end offset together make the inclusive start/exclusive end pair for where to underline to highlight
 /// a given point of interest.