Eliminate ClassifiedInputStream in favour of InputStream.

[thegedge]

This PR brings external commands closer to internal commands by returning an InputStream instead of the specialized ClassifiedInputStream. Although we now gain the ability to pipe internal commands into external commands, external | external will have significantly worse performance when lots of data is flowing over the pipe (we plan to address this in the future).

Closes #1057
Closes #928
Closes #935 (#778 has more discussion around the same topic)
Closes #887
Closes #816 (I think?)
Closes #489
Closes #412

I think I got them all 😅

What to focus on

Besides the usual, I've had to rebase and deal with a lot of merge conflicts along the way. Making sure I haven't accidentally dropped some of the other changes that have been made would be 👍

[gitpod-io]

[sophiajt]

Will try to carve out some time tomorrow to look at it. I agree: we should measure a few runs of streaming across the external boundary before and after this PR and get a sense for the hit we're taking before we land.

[sophiajt]

@thegedge - I was experimenting a bit with timing between two externals, and noticed that this fix seems to cause that scenario to break. If you're on a Linux machine and run cat Cargo.toml | wc do you see it run and never complete?

[sophiajt]

I'm not sure if related, but I noticed we also have the same behavior today in some internal->external pipelines: #1057

[thegedge]

Ah yes, I completely forgot to feed the InputStream to stdin 🤦‍♂

I'll take care of that over the next few days, and also some measurements.

I'm not sure if related, but I noticed we also have the same behavior today in some internal->external pipelines: #1057

In making this work for external → external, I'll actually kill two birds with one stone because I'll also make it work for internal → external 🎉

I think that's a big enough win that a small perf hit (for now) is acceptable.

[thegedge]

This is still a bit of a WIP, but I'm looking for some early feedback. Focal points:

• You can run tests locally, and cargo run to try it out. I've done so, and everything appears to work (including interactive commands!).
• I'm new to Rust's async. Focus on what I've written, if there are better ways to write it, if I'm doing unnecessary things, and so on.
• I haven't benchmarked this yet, but comparing the perf of this PR to directly linking the pipes (if external → external, since internal → external never worked).
• Lots of error handling needs to happen. CI is failing because I'm not doing so right now.
• There's some code duplication here, and that was intentional. There's some overlap, but ... | ^foo $it is significantly different than ... | ^foo. I'm not seeing an obvious refactor that would unify these two right now, so I chose to keep them as separate codepaths.

[sophiajt]

I noted a few things. In general, I'm +1 on getting it to the point we can benchmark and then make the call.

I'm also eager to work out exactly how $it will work in the future, but that's outside of the scope of this PR. I'd prefer to land this first, and then iterate on the design for external $it in a follow-up PR.

"but ... | ^foo $it is significantly different than ... | ^foo" <-- very much agreed. I think what you have for this is good for now. I agree we should go ahead and separate them.

[thegedge]

I couldn't really get a proper benchmark set up with the code, so I used my own profiling tool to capture a bunch of samples and emit average/stddev. Here are the results for running release builds, 50 samples:

---

^echo 1 | cat

this branch

μ = 0.2883
σ = 0.0060

master

μ = 0.3922
σ = 0.0044

^cat README.md | cat

this branch

μ = 0.2908
σ = 0.0102

master

μ = 0.3891
σ = 0.0039

^cat test.md | cat

test.md is 59 (copies of README.md)

this branch

μ = 2.6396
σ = 0.0680

master

μ = 0.3991
σ = 0.0190

---

Yeah, I double checked those, ran the script several times. Apparently this branch is faster than connecting the pipes directly when there's a small amount of stuff flowing through them. Larger files are where we take a hit.

[thegedge]

Ignoring the CI failures, @jonathandturner, what do you think about moving this branch forward? Given the huge perf hit we'd get for larger files, should I continue working on this, or focus my efforts elsewhere?

Maybe @wycats has some thoughts, since this does unify internal/external, which was one goal he mentioned he's been working on.

If we're 👍, I can take care of some of the TODO comments I scattered here and there.

[sophiajt]

@thegedge - I say let's go for it. Simplifying things seems good, and we can optimize multiple external commands in a row into a single unified external pipeline for performance later.

[thegedge]

This is ready for review. I'm just trying to understand the CI failure:

error[E0277]: `*mut (dyn std::ops::Fn() + 'static)` cannot be shared between threads safely
   --> src/commands/classified/external.rs:295:19
    |
295 |         Ok(stream.to_input_stream())
    |                   ^^^^^^^^^^^^^^^ `*mut (dyn std::ops::Fn() + 'static)` cannot be shared between threads safely
    |
    = help: within `core::fmt::Void`, the trait `std::marker::Sync` is not implemented for `*mut (dyn std::ops::Fn() + 'static)`
    = note: required because it appears within the type `std::marker::PhantomData<*mut (dyn std::ops::Fn() + 'static)>`
    = note: required because it appears within the type `core::fmt::Void`
    = note: required because of the requirements on the impl of `std::marker::Send` for `&core::fmt::Void`
    = note: required because it appears within the type `std::fmt::ArgumentV1<'_>`
    = note: required because it appears within the type `[std::fmt::ArgumentV1<'_>; 1]`
    = note: required because it appears within the type `for<'r, 's, 't0, 't1, 't2, 't3, 't4, 't5, 't6, 't7, 't8, 't9, 't10, 't11, 't12, 't13, 't14, 't15, 't16, 't17, 't18, 't19, 't20, 't21, 't22, 't23, 't24, 't25, 't26, 't27, 't28, 't29, 't30, 't31, 't32, 't33, 't34, 't35, 't36, 't37, 't38, 't39, 't40, 't41, 't42, 't43, 't44, 't45, 't46, 't47, 't48, 't49, 't50, 't51, 't52, 't53, 't54, 't55, 't56, 't57> {&'r mut stream::InputStream, stream::InputStream, futures_util::stream::next::Next<'s, stream::InputStream>, futures_util::stream::next::Next<'t0, stream::InputStream>, (), std::option::Option<nu_protocol::value::Value>, nu_protocol::value::Value, nu_protocol::value::Value, nu_protocol::value::UntaggedValue, std::string::String, std::string::String, &'t1 mut std::fs::File, std::fs::File, &'t2 std::string::String, std::string::String, &'t3 [u8], &'t4 [u8], std::result::Result<usize, std::io::Error>, std::io::Error, &'t5 mut async_stream::yielder::Sender<std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>>, async_stream::yielder::Sender<std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>>, &'t6 str, &'t7 str, [&'t8 str; 2], &'t9 [&'t10 str], &'t11 [&'t12 str; 2], std::io::Error, &'t13 std::io::Error, (&'t14 std::io::Error,), [std::fmt::ArgumentV1<'t15>; 1], &'t16 [std::fmt::ArgumentV1<'t17>], &'t18 [std::fmt::ArgumentV1<'t19>; 1], std::fmt::Arguments<'t20>, std::string::String, &'t21 str, &'t22 nu_source::meta::Tag, nu_source::meta::Tag, nu_source::meta::Tag, nu_errors::ShellError, nu_protocol::value::UntaggedValue, nu_source::meta::Tag, nu_protocol::value::Value, std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>, impl core::future::future::Future, impl core::future::future::Future, (), nu_protocol::value::UntaggedValue, &'t23 mut async_stream::yielder::Sender<std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>>, async_stream::yielder::Sender<std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>>, &'t24 str, &'t25 str, [&'t26 str; 2], &'t27 [&'t28 str], &'t29 [&'t30 str; 2], &'t31 nu_protocol::value::UntaggedValue, nu_protocol::value::UntaggedValue, &'t32 str, &'t33 &'t34 str, (&'t35 &'t36 str,), [std::fmt::ArgumentV1<'t37>; 1], &'t38 [std::fmt::ArgumentV1<'t39>], &'t40 [std::fmt::ArgumentV1<'t41>; 1], std::fmt::Arguments<'t42>, std::string::String, &'t43 str, &'t44 nu_source::meta::Tag, nu_source::meta::Tag, nu_source::meta::Tag, nu_errors::ShellError, nu_protocol::value::UntaggedValue, nu_source::meta::Tag, nu_protocol::value::Value, std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>, impl core::future::future::Future, impl core::future::future::Future, (), bool, bool, subprocess::popen::Popen, &'t45 mut std::option::Option<std::fs::File>, std::option::Option<std::fs::File>, std::option::Option<std::fs::File>, &'t46 mut async_stream::yielder::Sender<std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>>, async_stream::yielder::Sender<std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>>, &'t47 str, &'t48 str, &'t49 nu_source::meta::Tag, nu_source::meta::Tag, nu_source::meta::Tag, nu_errors::ShellError, nu_protocol::value::UntaggedValue, nu_source::meta::Tag, nu_protocol::value::Value, std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>, impl core::future::future::Future, impl core::future::future::Future, (), std::fs::File, nu_source::meta::Tag, futures_util::io::allow_std::AllowStdIo<std::fs::File>, futures_codec::framed::Framed<futures_util::io::allow_std::AllowStdIo<std::fs::File>, commands::classified::external::LinesCodec>, futures_util::stream::map::Map<futures_codec::framed::Framed<futures_util::io::allow_std::AllowStdIo<std::fs::File>, commands::classified::external::LinesCodec>, [closure@src/commands/classified/external.rs:260:45: 8:51 stream_name_tag:nu_source::meta::Tag]>, &'t50 mut futures_util::stream::map::Map<futures_codec::framed::Framed<futures_util::io::allow_std::AllowStdIo<std::fs::File>, commands::classified::external::LinesCodec>, [closure@src/commands/classified/external.rs:260:45: 8:51 stream_name_tag:nu_source::meta::Tag]>, futures_util::stream::map::Map<futures_codec::framed::Framed<futures_util::io::allow_std::AllowStdIo<std::fs::File>, commands::classified::external::LinesCodec>, [closure@src/commands/classified/external.rs:260:45: 8:51 stream_name_tag:nu_source::meta::Tag]>, futures_util::stream::next::Next<'t51, futures_util::stream::map::Map<futures_codec::framed::Framed<futures_util::io::allow_std::AllowStdIo<std::fs::File>, commands::classified::external::LinesCodec>, [closure@src/commands/classified/external.rs:260:45: 8:51 stream_name_tag:nu_source::meta::Tag]>>, futures_util::stream::next::Next<'t52, futures_util::stream::map::Map<futures_codec::framed::Framed<futures_util::io::allow_std::AllowStdIo<std::fs::File>, commands::classified::external::LinesCodec>, [closure@src/commands/classified/external.rs:260:45: 8:51 stream_name_tag:nu_source::meta::Tag]>>, (), std::option::Option<nu_protocol::value::Value>, nu_protocol::value::Value, &'t53 mut async_stream::yielder::Sender<std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>>, async_stream::yielder::Sender<std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>>, nu_protocol::value::Value, std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>, impl core::future::future::Future, impl core::future::future::Future, (), bool, bool, bool, &'t54 mut async_stream::yielder::Sender<std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>>, async_stream::yielder::Sender<std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>>, &'t55 str, &'t56 str, &'t57 nu_source::meta::Tag, nu_source::meta::Tag, nu_source::meta::Tag, nu_errors::ShellError, nu_protocol::value::UntaggedValue, nu_source::meta::Tag, nu_protocol::value::Value, std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>, impl core::future::future::Future, impl core::future::future::Future, ()}`
    = note: required because it appears within the type `[static generator@<::async_stream::stream macros>:7:10: 11:11 input:stream::InputStream, stdin_write:std::fs::File, __yield_tx:async_stream::yielder::Sender<std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>>, name_tag:nu_source::meta::Tag, is_last:bool, popen:subprocess::popen::Popen for<'r, 's, 't0, 't1, 't2, 't3, 't4, 't5, 't6, 't7, 't8, 't9, 't10, 't11, 't12, 't13, 't14, 't15, 't16, 't17, 't18, 't19, 't20, 't21, 't22, 't23, 't24, 't25, 't26, 't27, 't28, 't29, 't30, 't31, 't32, 't33, 't34, 't35, 't36, 't37, 't38, 't39, 't40, 't41, 't42, 't43, 't44, 't45, 't46, 't47, 't48, 't49, 't50, 't51, 't52, 't53, 't54, 't55, 't56, 't57> {&'r mut stream::InputStream, stream::InputStream, futures_util::stream::next::Next<'s, stream::InputStream>, futures_util::stream::next::Next<'t0, stream::InputStream>, (), std::option::Option<nu_protocol::value::Value>, nu_protocol::value::Value, nu_protocol::value::Value, nu_protocol::value::UntaggedValue, std::string::String, std::string::String, &'t1 mut std::fs::File, std::fs::File, &'t2 std::string::String, std::string::String, &'t3 [u8], &'t4 [u8], std::result::Result<usize, std::io::Error>, std::io::Error, &'t5 mut async_stream::yielder::Sender<std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>>, async_stream::yielder::Sender<std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>>, &'t6 str, &'t7 str, [&'t8 str; 2], &'t9 [&'t10 str], &'t11 [&'t12 str; 2], std::io::Error, &'t13 std::io::Error, (&'t14 std::io::Error,), [std::fmt::ArgumentV1<'t15>; 1], &'t16 [std::fmt::ArgumentV1<'t17>], &'t18 [std::fmt::ArgumentV1<'t19>; 1], std::fmt::Arguments<'t20>, std::string::String, &'t21 str, &'t22 nu_source::meta::Tag, nu_source::meta::Tag, nu_source::meta::Tag, nu_errors::ShellError, nu_protocol::value::UntaggedValue, nu_source::meta::Tag, nu_protocol::value::Value, std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>, impl core::future::future::Future, impl core::future::future::Future, (), nu_protocol::value::UntaggedValue, &'t23 mut async_stream::yielder::Sender<std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>>, async_stream::yielder::Sender<std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>>, &'t24 str, &'t25 str, [&'t26 str; 2], &'t27 [&'t28 str], &'t29 [&'t30 str; 2], &'t31 nu_protocol::value::UntaggedValue, nu_protocol::value::UntaggedValue, &'t32 str, &'t33 &'t34 str, (&'t35 &'t36 str,), [std::fmt::ArgumentV1<'t37>; 1], &'t38 [std::fmt::ArgumentV1<'t39>], &'t40 [std::fmt::ArgumentV1<'t41>; 1], std::fmt::Arguments<'t42>, std::string::String, &'t43 str, &'t44 nu_source::meta::Tag, nu_source::meta::Tag, nu_source::meta::Tag, nu_errors::ShellError, nu_protocol::value::UntaggedValue, nu_source::meta::Tag, nu_protocol::value::Value, std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>, impl core::future::future::Future, impl core::future::future::Future, (), bool, bool, subprocess::popen::Popen, &'t45 mut std::option::Option<std::fs::File>, std::option::Option<std::fs::File>, std::option::Option<std::fs::File>, &'t46 mut async_stream::yielder::Sender<std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>>, async_stream::yielder::Sender<std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>>, &'t47 str, &'t48 str, &'t49 nu_source::meta::Tag, nu_source::meta::Tag, nu_source::meta::Tag, nu_errors::ShellError, nu_protocol::value::UntaggedValue, nu_source::meta::Tag, nu_protocol::value::Value, std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>, impl core::future::future::Future, impl core::future::future::Future, (), std::fs::File, nu_source::meta::Tag, futures_util::io::allow_std::AllowStdIo<std::fs::File>, futures_codec::framed::Framed<futures_util::io::allow_std::AllowStdIo<std::fs::File>, commands::classified::external::LinesCodec>, futures_util::stream::map::Map<futures_codec::framed::Framed<futures_util::io::allow_std::AllowStdIo<std::fs::File>, commands::classified::external::LinesCodec>, [closure@src/commands/classified/external.rs:260:45: 8:51 stream_name_tag:nu_source::meta::Tag]>, &'t50 mut futures_util::stream::map::Map<futures_codec::framed::Framed<futures_util::io::allow_std::AllowStdIo<std::fs::File>, commands::classified::external::LinesCodec>, [closure@src/commands/classified/external.rs:260:45: 8:51 stream_name_tag:nu_source::meta::Tag]>, futures_util::stream::map::Map<futures_codec::framed::Framed<futures_util::io::allow_std::AllowStdIo<std::fs::File>, commands::classified::external::LinesCodec>, [closure@src/commands/classified/external.rs:260:45: 8:51 stream_name_tag:nu_source::meta::Tag]>, futures_util::stream::next::Next<'t51, futures_util::stream::map::Map<futures_codec::framed::Framed<futures_util::io::allow_std::AllowStdIo<std::fs::File>, commands::classified::external::LinesCodec>, [closure@src/commands/classified/external.rs:260:45: 8:51 stream_name_tag:nu_source::meta::Tag]>>, futures_util::stream::next::Next<'t52, futures_util::stream::map::Map<futures_codec::framed::Framed<futures_util::io::allow_std::AllowStdIo<std::fs::File>, commands::classified::external::LinesCodec>, [closure@src/commands/classified/external.rs:260:45: 8:51 stream_name_tag:nu_source::meta::Tag]>>, (), std::option::Option<nu_protocol::value::Value>, nu_protocol::value::Value, &'t53 mut async_stream::yielder::Sender<std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>>, async_stream::yielder::Sender<std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>>, nu_protocol::value::Value, std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>, impl core::future::future::Future, impl core::future::future::Future, (), bool, bool, bool, &'t54 mut async_stream::yielder::Sender<std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>>, async_stream::yielder::Sender<std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>>, &'t55 str, &'t56 str, &'t57 nu_source::meta::Tag, nu_source::meta::Tag, nu_source::meta::Tag, nu_errors::ShellError, nu_protocol::value::UntaggedValue, nu_source::meta::Tag, nu_protocol::value::Value, std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>, impl core::future::future::Future, impl core::future::future::Future, ()}]`
    = note: required because it appears within the type `std::future::GenFuture<[static generator@<::async_stream::stream macros>:7:10: 11:11 input:stream::InputStream, stdin_write:std::fs::File, __yield_tx:async_stream::yielder::Sender<std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>>, name_tag:nu_source::meta::Tag, is_last:bool, popen:subprocess::popen::Popen for<'r, 's, 't0, 't1, 't2, 't3, 't4, 't5, 't6, 't7, 't8, 't9, 't10, 't11, 't12, 't13, 't14, 't15, 't16, 't17, 't18, 't19, 't20, 't21, 't22, 't23, 't24, 't25, 't26, 't27, 't28, 't29, 't30, 't31, 't32, 't33, 't34, 't35, 't36, 't37, 't38, 't39, 't40, 't41, 't42, 't43, 't44, 't45, 't46, 't47, 't48, 't49, 't50, 't51, 't52, 't53, 't54, 't55, 't56, 't57> {&'r mut stream::InputStream, stream::InputStream, futures_util::stream::next::Next<'s, stream::InputStream>, futures_util::stream::next::Next<'t0, stream::InputStream>, (), std::option::Option<nu_protocol::value::Value>, nu_protocol::value::Value, nu_protocol::value::Value, nu_protocol::value::UntaggedValue, std::string::String, std::string::String, &'t1 mut std::fs::File, std::fs::File, &'t2 std::string::String, std::string::String, &'t3 [u8], &'t4 [u8], std::result::Result<usize, std::io::Error>, std::io::Error, &'t5 mut async_stream::yielder::Sender<std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>>, async_stream::yielder::Sender<std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>>, &'t6 str, &'t7 str, [&'t8 str; 2], &'t9 [&'t10 str], &'t11 [&'t12 str; 2], std::io::Error, &'t13 std::io::Error, (&'t14 std::io::Error,), [std::fmt::ArgumentV1<'t15>; 1], &'t16 [std::fmt::ArgumentV1<'t17>], &'t18 [std::fmt::ArgumentV1<'t19>; 1], std::fmt::Arguments<'t20>, std::string::String, &'t21 str, &'t22 nu_source::meta::Tag, nu_source::meta::Tag, nu_source::meta::Tag, nu_errors::ShellError, nu_protocol::value::UntaggedValue, nu_source::meta::Tag, nu_protocol::value::Value, std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>, impl core::future::future::Future, impl core::future::future::Future, (), nu_protocol::value::UntaggedValue, &'t23 mut async_stream::yielder::Sender<std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>>, async_stream::yielder::Sender<std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>>, &'t24 str, &'t25 str, [&'t26 str; 2], &'t27 [&'t28 str], &'t29 [&'t30 str; 2], &'t31 nu_protocol::value::UntaggedValue, nu_protocol::value::UntaggedValue, &'t32 str, &'t33 &'t34 str, (&'t35 &'t36 str,), [std::fmt::ArgumentV1<'t37>; 1], &'t38 [std::fmt::ArgumentV1<'t39>], &'t40 [std::fmt::ArgumentV1<'t41>; 1], std::fmt::Arguments<'t42>, std::string::String, &'t43 str, &'t44 nu_source::meta::Tag, nu_source::meta::Tag, nu_source::meta::Tag, nu_errors::ShellError, nu_protocol::value::UntaggedValue, nu_source::meta::Tag, nu_protocol::value::Value, std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>, impl core::future::future::Future, impl core::future::future::Future, (), bool, bool, subprocess::popen::Popen, &'t45 mut std::option::Option<std::fs::File>, std::option::Option<std::fs::File>, std::option::Option<std::fs::File>, &'t46 mut async_stream::yielder::Sender<std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>>, async_stream::yielder::Sender<std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>>, &'t47 str, &'t48 str, &'t49 nu_source::meta::Tag, nu_source::meta::Tag, nu_source::meta::Tag, nu_errors::ShellError, nu_protocol::value::UntaggedValue, nu_source::meta::Tag, nu_protocol::value::Value, std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>, impl core::future::future::Future, impl core::future::future::Future, (), std::fs::File, nu_source::meta::Tag, futures_util::io::allow_std::AllowStdIo<std::fs::File>, futures_codec::framed::Framed<futures_util::io::allow_std::AllowStdIo<std::fs::File>, commands::classified::external::LinesCodec>, futures_util::stream::map::Map<futures_codec::framed::Framed<futures_util::io::allow_std::AllowStdIo<std::fs::File>, commands::classified::external::LinesCodec>, [closure@src/commands/classified/external.rs:260:45: 8:51 stream_name_tag:nu_source::meta::Tag]>, &'t50 mut futures_util::stream::map::Map<futures_codec::framed::Framed<futures_util::io::allow_std::AllowStdIo<std::fs::File>, commands::classified::external::LinesCodec>, [closure@src/commands/classified/external.rs:260:45: 8:51 stream_name_tag:nu_source::meta::Tag]>, futures_util::stream::map::Map<futures_codec::framed::Framed<futures_util::io::allow_std::AllowStdIo<std::fs::File>, commands::classified::external::LinesCodec>, [closure@src/commands/classified/external.rs:260:45: 8:51 stream_name_tag:nu_source::meta::Tag]>, futures_util::stream::next::Next<'t51, futures_util::stream::map::Map<futures_codec::framed::Framed<futures_util::io::allow_std::AllowStdIo<std::fs::File>, commands::classified::external::LinesCodec>, [closure@src/commands/classified/external.rs:260:45: 8:51 stream_name_tag:nu_source::meta::Tag]>>, futures_util::stream::next::Next<'t52, futures_util::stream::map::Map<futures_codec::framed::Framed<futures_util::io::allow_std::AllowStdIo<std::fs::File>, commands::classified::external::LinesCodec>, [closure@src/commands/classified/external.rs:260:45: 8:51 stream_name_tag:nu_source::meta::Tag]>>, (), std::option::Option<nu_protocol::value::Value>, nu_protocol::value::Value, &'t53 mut async_stream::yielder::Sender<std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>>, async_stream::yielder::Sender<std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>>, nu_protocol::value::Value, std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>, impl core::future::future::Future, impl core::future::future::Future, (), bool, bool, bool, &'t54 mut async_stream::yielder::Sender<std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>>, async_stream::yielder::Sender<std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>>, &'t55 str, &'t56 str, &'t57 nu_source::meta::Tag, nu_source::meta::Tag, nu_source::meta::Tag, nu_errors::ShellError, nu_protocol::value::UntaggedValue, nu_source::meta::Tag, nu_protocol::value::Value, std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>, impl core::future::future::Future, impl core::future::future::Future, ()}]>`
    = note: required because it appears within the type `impl core::future::future::Future`
    = note: required because it appears within the type `async_stream::async_stream::AsyncStream<std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>, impl core::future::future::Future>`
    = note: required because of the requirements on the impl of `prelude::ToInputStream` for `async_stream::async_stream::AsyncStream<std::result::Result<nu_protocol::value::Value, nu_errors::ShellError>, impl core::future::future::Future>`

It's confusing because I can't tell why it's different from what we do in internal.rs... 🤔

[thegedge]

Looks like I may have messed up a rebase and brought back some unwraps. I'll take care of those 🙈

[sophiajt]

Looks good! Thanks