MsgPack deserializer: improve handling of EOF (#12183)

# Description

`rmp_serde` has two kinds of errors that contain I/O errors, and an EOF
can occur inside either of them, but we were only treating an EOF inside
an `InvalidMarkerRead` as an EOF, which would make sense for the
beginning of a message.

However, we should also treat an incomplete message + EOF as an EOF.
There isn't really any point in reporting that an EOF was received
mid-message.

This should fix the issue where the
`seq_describe_no_collect_succeeds_without_error` test would sometimes
fail, as doing a `describe --no-collect` followed by nushell exiting
could (but was not guaranteed to) cause this exact scenario.

# User-Facing Changes
Will probably remove useless `read error` messages from plugins after
exit of `nu`

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

# After Submitting

crates/nu-plugin/src/serializers/msgpack.rs

@@ -82,17 +82,16 @@ fn rmp_encode_err(err: rmp_serde::encode::Error) -> ShellError {
 fn rmp_decode_err<T>(err: rmp_serde::decode::Error) -> Result<Option<T>, ShellError> {
     match err {
         rmp_serde::decode::Error::InvalidMarkerRead(err)
-            if matches!(err.kind(), ErrorKind::UnexpectedEof) =>
-        {
-            // EOF
-            Ok(None)
-        }
-        rmp_serde::decode::Error::InvalidMarkerRead(_)
-        | rmp_serde::decode::Error::InvalidDataRead(_) => {
-            // I/O error
-            Err(ShellError::IOError {
-                msg: err.to_string(),
-            })
+        | rmp_serde::decode::Error::InvalidDataRead(err) => {
+            if matches!(err.kind(), ErrorKind::UnexpectedEof) {
+                // EOF
+                Ok(None)
+            } else {
+                // I/O error
+                Err(ShellError::IOError {
+                    msg: err.to_string(),
+                })
+            }
         }
         _ => {
             // Something else

tests/plugins/stream.rs

@@ -15,14 +15,18 @@ fn seq_produces_stream() {
 #[test]
 fn seq_describe_no_collect_succeeds_without_error() {
     // This tests to ensure that there's no error if the stream is suddenly closed
-    let actual = nu_with_plugins!(
-        cwd: "tests/fixtures/formats",
-        plugin: ("nu_plugin_stream_example"),
-        "stream_example seq 1 5 | describe --no-collect"
-    );
+    // Test several times, because this can cause different errors depending on what is written
+    // when the engine stops running, especially if there's partial output
+    for _ in 0..10 {
+        let actual = nu_with_plugins!(
+            cwd: "tests/fixtures/formats",
+            plugin: ("nu_plugin_stream_example"),
+            "stream_example seq 1 5 | describe --no-collect"
+        );
 
-    assert_eq!(actual.out, "stream");
-    assert_eq!(actual.err, "");
+        assert_eq!(actual.out, "stream");
+        assert_eq!(actual.err, "");
+    }
 }
 
 #[test]