do not emit None mid-stream during parse (#9925)

<!--
if this PR closes one or more issues, you can automatically link the PR
with
them by using one of the [*linking
keywords*](https://docs.github.com/en/issues/tracking-your-work-with-issues/linking-a-pull-request-to-an-issue#linking-a-pull-request-to-an-issue-using-a-keyword),
e.g.
- this PR should close #issue 

you can also mention related issues, PRs or discussions!
-->

# Description
<!--
Thank you for improving Nushell. Please, check our [contributing
guide](../CONTRIBUTING.md) and talk to the core team before making major
changes.

Description of your pull request goes here. **Provide examples and/or
screenshots** if your changes affect the user experience.
-->
Currently `parse` acts like a `.filter` over an iterator, except that it
emits `None` for elements that can't be parsed. This causes consumers of
the adapted iterator to stop iterating too early. The correct behaviour
is to keep pulling the inner iterator until either the end of it is
reached or an element can be parsed.

- this PR should close #9906 

# User-Facing Changes
<!-- List of all changes that impact the user experience here. This
helps us keep track of breaking changes. -->
List streams won't be truncated anymore after the first parse failure.

# Tests + Formatting
<!--
Don't forget to add tests that cover your changes.

Make sure you've run and fixed any issues with these commands:

- [x] `cargo fmt --all -- --check` to check standard code formatting
(`cargo fmt --all` applies these changes)
- [x] `cargo clippy --workspace -- -D warnings -D clippy::unwrap_used -A
clippy::needless_collect -A clippy::result_large_err` to check that
you're using the standard code style
- [-] `cargo test --workspace` to check that all tests pass

- `cargo run -- -c "use std testing; testing run-tests --path
crates/nu-std"` to run the tests for the standard library

> **Note**
> from `nushell` you can also use the `toolkit` as follows
> ```bash
> use toolkit.nu # or use an `env_change` hook to activate it
automatically
> toolkit check pr
> ```
-->
- [x] `cargo fmt --all -- --check` to check standard code formatting
(`cargo fmt --all` applies these changes)
- [x] `cargo clippy --workspace -- -D warnings -D clippy::unwrap_used -A
clippy::needless_collect -A clippy::result_large_err` to check that
you're using the standard code style
- [x] `cargo test --workspace` to check that all tests pass
  - 11 tests fail, but the same 11 tests fail on main as well
- [x] `cargo run -- -c "use std testing; testing run-tests --path
crates/nu-std"` to run the tests for the standard library

# After Submitting
<!-- If your PR had any user-facing changes, update [the
documentation](https://github.com/nushell/nushell.github.io) after the
PR is merged, if necessary. This will help us keep the docs up to date.
-->

crates/nu-command/src/strings/parse.rs

@@ -1,3 +1,6 @@
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::Arc;
+
 use fancy_regex::Regex;
 use nu_engine::CallExt;
 use nu_protocol::ast::Call;
@@ -230,6 +233,7 @@ fn operate(
                     regex: regex_pattern,
                     columns,
                     stream: stream.stream,
+                    ctrlc: ctrlc.clone(),
                 },
                 ctrlc,
             ),
@@ -329,6 +333,7 @@ pub struct ParseStreamer {
     regex: Regex,
     columns: Vec<String>,
     stream: Box<dyn Iterator<Item = Value> + Send + 'static>,
+    ctrlc: Option<Arc<AtomicBool>>,
 }
 
 impl Iterator for ParseStreamer {
@@ -338,27 +343,38 @@ impl Iterator for ParseStreamer {
             return Some(self.excess.remove(0));
         }
 
-        let v = self.stream.next();
+        loop {
+            if let Some(ctrlc) = &self.ctrlc {
+                if ctrlc.load(Ordering::SeqCst) {
+                    break None;
+                }
+            }
 
-        if let Some(v) = v {
-            match v.as_string() {
-                Ok(s) => stream_helper(
-                    self.regex.clone(),
-                    v.span().unwrap_or(self.span),
-                    s,
-                    self.columns.clone(),
-                    &mut self.excess,
-                ),
-                Err(_) => Some(Value::Error {
+            let Some(v) = self.stream.next() else { return None };
+
+            let Ok(s) = v.as_string() else {
+                return Some(Value::Error {
                     error: Box::new(ShellError::PipelineMismatch {
                         exp_input_type: "string".into(),
                         dst_span: self.span,
                         src_span: v.span().unwrap_or(self.span),
                     }),
-                }),
-            }
-        } else {
-            None
+                })
+            };
+
+            let parsed = stream_helper(
+                self.regex.clone(),
+                v.span().unwrap_or(self.span),
+                s,
+                self.columns.clone(),
+                &mut self.excess,
+            );
+
+            if parsed.is_none() {
+                continue;
+            };
+
+            return parsed;
         }
     }
 }

crates/nu-command/tests/commands/parse.rs

@@ -197,4 +197,18 @@ mod regex {
 
         assert_eq!(actual.out, "table<letter: string, a: string> (stream)")
     }
+
+    #[test]
+    fn parse_does_not_truncate_list_streams() {
+        let actual = nu!(pipeline(
+            r#"
+                [a b c]
+                | each {|x| $x}
+                | parse --regex "[ac]"
+                | length
+            "#
+        ));
+
+        assert_eq!(actual.out, "2");
+    }
 }