Add plugin error propagation on write/flush (#12670)

# Description
Yet another attempt to fix the `stress_internals::test_wrong_version()`
test...

This time I think it's probably because we are getting a broken pipe
write error when we try to send `Hello` or perhaps something after it,
because the pipe has already been closed by the reader when it saw the
invalid version. In that case, an error should be available in state. It
probably makes more sense to send that back to the user rather than an
unhelpful I/O error.

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

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

@@ -1037,11 +1037,21 @@ impl Interface for PluginInterface {
 
     fn write(&self, input: PluginInput) -> Result<(), ShellError> {
         log::trace!("to plugin: {:?}", input);
-        self.state.writer.write(&input)
+        self.state.writer.write(&input).map_err(|err| {
+            log::warn!("write() error: {}", err);
+            // If there's an error in the state, return that instead because it's likely more
+            // descriptive
+            self.state.error.get().cloned().unwrap_or(err)
+        })
     }
 
     fn flush(&self) -> Result<(), ShellError> {
-        self.state.writer.flush()
+        self.state.writer.flush().map_err(|err| {
+            log::warn!("flush() error: {}", err);
+            // If there's an error in the state, return that instead because it's likely more
+            // descriptive
+            self.state.error.get().cloned().unwrap_or(err)
+        })
     }
 
     fn stream_id_sequence(&self) -> &Sequence {

crates/nu_plugin_stress_internals/src/main.rs

@@ -82,6 +82,12 @@ pub fn main() -> Result<(), Box<dyn Error>> {
         std::process::exit(1)
     }
 
+    // Read `Hello` message
+    let mut de = serde_json::Deserializer::from_reader(&mut input);
+    let hello: Value = Value::deserialize(&mut de)?;
+
+    assert!(hello.get("Hello").is_some());
+
     // Send `Hello` message
     write(
         &mut output,
@@ -103,12 +109,6 @@ pub fn main() -> Result<(), Box<dyn Error>> {
         }),
     )?;
 
-    // Read `Hello` message
-    let mut de = serde_json::Deserializer::from_reader(&mut input);
-    let hello: Value = Value::deserialize(&mut de)?;
-
-    assert!(hello.get("Hello").is_some());
-
     if opts.exit_early {
         // Exit without handling anything other than Hello
         std::process::exit(0);