Using duplex() with fd3: it must be a writable stream

[matpen]

I am trying to use execa within a pipeline, which results in very elegant code.

However, I am facing a problem when using an input file descriptor different from stdin. The following snippet contains a

minimal reproducer

// Importing dependencies
import fs from 'node:fs';
import dedent from 'dedent';
import { execa } from 'execa';
import { pipeline } from 'node:stream/promises';
import { Readable } from 'node:stream';

// Run execa as duplex stream in a pipeline
// The pipeline source and sink, as well as the subprocess (Python here) are not important,
// instead the focus is on the "to" option for duplex()
await pipeline(
    Readable.from('This is a test'),
    execa(
        'python3',
        {
            input: dedent(`
                import os
                with os.fdopen(3, 'r') as file:
                    content = json.load(file)
                print(content)
            `),
            stdio: ['pipe', 'pipe', 'pipe', 'pipe'],
        },
    ).duplex({ to: 'fd3' }),
    process.stdout,
);

Running the code on GNU/Linux 6.14.0 with Node.js v22.15.0 and execa 9.6.0 causes the following

error message

file:///tmp/test/node_modules/execa/lib/arguments/fd-options.js:66
                throw new TypeError(`"${getOptionName(isWritable)}" must not be ${fdName}. It must be a writable stream, not readable.`);
                      ^

TypeError: "to" must not be fd3. It must be a writable stream, not readable.
    at validateFdNumber (file:///tmp/test/node_modules/execa/lib/arguments/fd-options.js:66:9)
    at getFdNumber (file:///tmp/test/node_modules/execa/lib/arguments/fd-options.js:36:2)
    at getToStream (file:///tmp/test/node_modules/execa/lib/arguments/fd-options.js:7:19)
    at getSubprocessStdin (file:///tmp/test/node_modules/execa/lib/convert/writable.js:32:26)
    at createDuplex (file:///tmp/test/node_modules/execa/lib/convert/duplex.js:22:68)
    at main (file:///tmp/test/[eval1]:29:11)
    at file:///tmp/test/[eval1]:37:2
    at ModuleJob.run (node:internal/modules/esm/module_job:274:25)
    at async onImport.tracePromise.__proto__ (node:internal/modules/esm/loader:268:26)
    at async ModuleLoader.executeModuleJob (node:internal/modules/esm/loader:264:20)

I tracked the source, and noticed that by simply removing the validation rules, the code runs and behaves as expected. Unfortunately, I am not familiar enough with the source code to suggest a solution, so I would appreciate any help by the maintainers or more experienced users.

Thank you for a wonderful library, and its extensive documentation.

[ehmicky]

Thanks @matpen for the nice words :)

Execa does a lot of stream redirection / piping under the hood, so it needs to know whether each file descriptor is input or output. We do this here.

For stdin, stdout, stderr, it's easy. For fd3 or other, we can guess based on the value passed to the stdio option. For example, passing a writable stream means you intend the file descriptor to used for output.

Unfortunately, when using 'pipe', this is ambiguous and we cannot know. You might think that using .duplex({to: 'fd3'}) should indicate the intent. However, .duplex() is executed after execa() has already resolved and spawned the subprocess.

So at the moment, we default to making those ambiguous file descriptors output, not input, because I was assuming this might be the most common use case.

The real solution here would be to allow the direction to be overridden in the stdio option.

{stdio: ['pipe', 'pipe', 'pipe', {value: 'pipe', direction: 'input'}]}}

But this would actually require quite lots of implementation due to the multiple shapes of this option, and also to the fact it's used in many places. So at the moment, the following workaround should work for you:

{stdio: ['pipe', 'pipe', 'pipe', ['pipe', []]]}}

The reason behind this weird/ugly syntax is because the stdio option items can be array of values. And one of those possible values is an iterable of values, passed as input, which therefore interprets the file descriptor as input. This passes an empty array, so nothing is actually passed as input, it just hints that the file descriptor is meant as input.

So putting it together:

// Importing dependencies
import fs from 'node:fs';
import dedent from 'dedent';
import { execa } from 'execa';
import { pipeline } from 'node:stream/promises';
import { Readable } from 'node:stream';

// Run execa as duplex stream in a pipeline
// The pipeline source and sink, as well as the subprocess (Python here) are not important,
// instead the focus is on the "to" option for duplex()
await pipeline(
    Readable.from('This is a test'),
    execa(
        'python3',
        {
            input: dedent(`
                import os
                with os.fdopen(3, 'r') as file:
                    content = json.load(file)
                print(content)
            `),
-            stdio: ['pipe', 'pipe', 'pipe', 'pipe'],
+            stdio: ['pipe', 'pipe', 'pipe', ['pipe', []]],
        },
    ).duplex({ to: 'fd3' }),
    process.stdout,
);

Please let me know if that works.

[ehmicky]

Note: this specific example can also be simplified by not using streams nor .duplex(), and passing strings to the stdio option instead:

// Double array needed for syntax reasons unfortunately
const inputStdin = [[`import os
with os.fdopen(3, 'r') as file:
  content = json.load(file)
print(content)`]]
const inputFd3 = [['This is a test']]
await execa('python3', { stdio: [inputStdin, 'inherit', 'pipe', inputFd3] })

But I'm assuming you might want streams for other reasons.

[matpen]

So at the moment, the following workaround should work for you:

@ehmicky this is perfect!

Thank you so much for the prompt reply, and the extensive explanation regarding the logic. While I would have never guessed the solution, it makes perfect sense based on your clarification and pointers to the code.

Please let me know if that works.

I tested it both in the reproducer and in the larger code I am dealing with, and it works as expected.

But I'm assuming you might want streams for other reasons.

Correct, the actual code from which I extracted the reproducer is a bit more complex, both upstream and downstream, and involves async generators, too. However, thank you for taking the time to add more ideas: they could be useful to other future readers!

Thanks @matpen for the nice words :)

It looks like you deserve more of those! ;)

[ehmicky]

Thank you @matpen!

Should we close this issue?

[matpen]

Should we close this issue?

No objection, since the solution you offered more than solves the OP. Thank you again for your time!

[matpen]

I am sorry, @ehmicky, but I encountered another problem with your workaround: it looks like it works for very short input streams, but fails on anything a bit longer.

To illustrate, I have prepared another

reproducer

import dedent from 'dedent';
import { execa } from 'execa';
import { pipeline } from 'node:stream/promises';
import { Readable } from 'node:stream';

await pipeline(
    Readable.from(async function* generate() {
        for (let i = 0; i < 2000; ++i) {
            if (i % 100 === 0) {
                await new Promise(r => setTimeout(r, 1000));
            }
            yield 'This is a test';
        }
    }()),
    execa(
        'python3',
        {
            input: dedent(`
                import os
                with os.fdopen(3, 'r') as file:
                    for line in file:
                        print(line)
            `),
            stdio: ['pipe', 'pipe', 'pipe', ['pipe', []]],
        },
    ).duplex({ to: 'fd3' }),
    process.stdout,
);

which fails with the following

error

node:internal/process/promises:394
    triggerUncaughtException(err, true /* fromPromise */);
    ^

Error [ERR_STREAM_PREMATURE_CLOSE]: Premature close
    at Duplex.<anonymous> (node:internal/streams/pipeline:415:29)
    at Duplex.emit (node:events:530:35)
    at emitCloseNT (node:internal/streams/destroy:148:10)
    at process.processTicksAndRejections (node:internal/process/task_queues:89:21) {
  code: 'ERR_STREAM_PREMATURE_CLOSE'
}

My current interpretation is that execa is closing both input streams connected to fd3, i.e. the pipe and the [], when any one of them finishes. Since the second is an empty array, it finishes immediately, and this closes the pipeline prematurely.

I tried tracking this behavior by starting from the pointer you gave me, but failed to identify the source of the issue. Would you mind re-opening this? Thank you in advance.

[ehmicky]

Hi again @matpen,

Yes, what you're thinking is correct. That is, the workaround I gave about passing an empty array does not work since that closes fd3 right away.

The core of the problem is that the default direction of fd3 is output. I assumed reading from fd3 would be an edge case, but you're encountering a case where this is needed.

I tried changing the default direction of fd3 to input in the code, and it did solve the problem. So I think one possible solution would be this: introduce a new '{value: 'pipe', input: true} value for the stdio[*] option. It would work exactly like 'pipe' except for fd3+, where it would assume the file descriptor to receive input, as opposed to write output.

[matpen]

Hi there, @ehmicky, and thank you for reopening this.

So I think one possible solution would be this: introduce a new 'pipeInput' value for the stdio[*] option. It would work exactly like 'pipe' except for fd3+, where it would assume the file descriptor to receive input, as opposed to write output.

This sounds like a reasonable approach. It would provide a "direction hint" to the code, so that it can do the right thing without guesswork.

That would perfectly solve the issue, and I would gladly test the solution, if you are willing to proceed with it. I just want to add a few information and thoughts, which can maybe provide additional help.

passing an empty array does not work since that closes fd3 right away.

That still does not resonate with me. I thought that providing multiple streams would pipe them all into the destination, so it seems more appropriate to close the destination when the last of the sources ends, not the first. Is this something worth looking into? Perhaps the option { end: false } was simply forgotten somewhere.

I tracked the source, and noticed that by simply removing the validation rules, the code runs and behaves as expected.

Also, I wanted to remind about my experiment from the OP. The only thing preventing the code from working (at least with the minimal reproducer I offered) is actually the validation check. If that is confirmed, would it be thinkable to simply remove the validation? I do recognize its value in some other scenarios, of course.

[ehmicky]

Also, I wanted to remind about my experiment from the OP. The only thing preventing the code from working (at least with the minimal reproducer I offered) is actually the validation check. If that is confirmed, would it be thinkable to simply remove the validation? I do recognize its value in some other scenarios, of course.

On the bright side, that failure did highlight there was a bug there, which was actually quite helpful to find a proper fix for it. I think this validation is helpful in general, as it tells users when they are using a file descriptor as output instead of input, or vice-versa, which most likely won't work.

In your case, you were unknowingly using a fd3 as output instead of input, and that error helped us find out right away this was the problem. While it might appear that it worked, Execa does use that "direction" hint in quite a few places, so it's not impossible that something less obvious about the data piping would not have worked correctly.

Also, by introducing a new syntax for users to specify the direction, this error message can now be even more helpful by suggesting that new syntax as a fix.

That still does not resonate with me. I thought that providing multiple streams would pipe them all into the destination, so it seems more appropriate to close the destination when the last of the sources ends, not the first. Is this something worth looking into? Perhaps the option { end: false } was simply forgotten somewhere.

Yes, you're correct that it does indicate something is off (different from the other issue). When piping multiple readable streams to a single output stream, we're actually very mindful about making sure the output stream is ended only when the last readable stream ends. It turns out it's quite hard to achieve this correctly in Node.js, so we created a module for this very purpose, that Execa uses: merge-streams. For example, one needs to handle every possible way a stream might terminate, streams being dynamically added or removed, unpiping, and more.

https://github.com/sindresorhus/merge-streams/blob/2f7dae33c8aed2e9be92bc8d64a38883695a59de/index.js#L69

It's thoroughly tested, even on edge cases, so my guess is that the bug might be in Execa itself there, but that's definitely something to look into.

I would gladly test the solution, if you are willing to proceed with it.

Thanks!