stream: fix nested compose error propagation

AGENTS.md

@@ -0,0 +1,92 @@
+# Agent Guidance
+
+Non-discoverable operational guidance for Node.js development.
+
+## Non-discoverable commands
+
+### Linting
+
+- `tools/eslint/node_modules/eslint/bin/eslint.js` is the actual lint runner (not `make lint-js` which first builds it)
+- `make lint` runs: JS lint, C++ lint, addon docs lint, markdown lint, YAML lint
+- CI uses `make lint-js-ci` (different output format than regular `make lint-js`)
+
+### Testing
+
+- `make test` = `make test-only` + documentation build
+- `make test-only` = default tests without documentation build
+- `make test-tap` = TAP output format
+- `make test-only TAP=1` = subset of tests in TAP format
+
+### Test organization
+
+- `test/parallel/` = tests that can run in parallel (add new tests here when unsure)
+- `test/sequential/` = tests that must not run in parallel
+- `test/pummel/` = load tests, not run on CI
+- `test/internet/` = tests with real outbound connections, not run on CI
+- `test/known_issues/` = all tests expected to fail, run via `known_issues.status`
+
+## Landmines / do-not-touch areas
+
+### Dependencies
+
+- `deps/` and `tools/` contain bundled dependencies not part of Node.js proper
+- Do not send patches to Node.js for files in these directories
+- Send changes to their respective projects instead
+
+### Build artifacts
+
+- `out/` directory is generated (not tracked in git, but `.gitignore` has exceptions)
+- `node` is a symlink to `out/Release/node`
+- `config.gypi`, `config.status`, `config.mk`, `icu_config.gypi` are generated by `./configure`
+
+### Branch workflow
+
+- Only create branches in your GitHub fork (not `nodejs/node`)
+- `nodejs/node` branches are only for release lines
+
+## Task-specific constraints
+
+### Tests
+
+- Tests must `require('../common')` first (not just for helpers, but for: globals leak check, umask enforcement, auto-respawn with flags)
+- Test flags can be specified in comment: `// Flags: --expose-internals`
+- Use `port: 0` instead of arbitrary ports for parallel-safe tests
+- Use `common.mustCall()` to verify callbacks are invoked
+- Tests exit 0 on success, non-zero (or `process.exitCode`) on failure
+
+### Build system
+
+- `./configure` must run before `make` commands
+- `make -j4` is typical parallel build
+- Ninja build system available as faster alternative (`./configure --gyp-mode=ninja`)
+
+### Commit messages
+
+- First line: `<subsystem>: <description>` (lowercase, imperative verb)
+- Find subsystems via: `git log --oneline files/you/changed`
+- `Fixes:` and `Refs:` trailers auto-added from PR description when landing
+- Breaking changes need `semver-major` label + explanation
+
+### CI/CI workflows
+
+- `.github/workflows/test-linux.yml` is the main CI test workflow
+- `.github/workflows/linters.yml` runs lint checks
+- `test/root.status` defines test status (SLOW, SKIP, etc.) per platform
+
+## Development helpers
+
+### Speed up rebuilds
+
+- `make V=1` for verbose output
+- `--enable-coredll` loads JS from disk instead of embedding (faster rebuilds)
+- `ccache` can be used via `./configure --with-ccache`
+
+### Debug builds
+
+- `./configure --debug && make` for debug build
+- `make V=1` for build verbosity
+
+### Coverage
+
+- `make coverage` generates coverage reports
+- `COV_SKIP_TESTS` env var to skip specific tests

lib/internal/streams/duplexify.js

@@ -239,7 +239,17 @@ function fromAsyncGen(fn) {
       _resolve({ done: true, cb });
     },
     destroy(err, cb) {
-      ac.abort();
+      ac.abort(err);
+
+      // If the source async iterator is waiting for the next write/final
+      // signal, unblock it so the readable side can observe the abort and
+      // finish destroying.
+      if (resolve !== null) {
+        const _resolve = resolve;
+        resolve = null;
+        _resolve({ __proto__: null, done: true, cb() {} });
+      }
+
       cb(err);
     },
   };

lib/internal/streams/from.js

@@ -8,9 +8,12 @@ const {
 const { Buffer } = require('buffer');
 
 const {
-  ERR_INVALID_ARG_TYPE,
-  ERR_STREAM_NULL_VALUES,
-} = require('internal/errors').codes;
+  aggregateTwoErrors,
+  codes: {
+    ERR_INVALID_ARG_TYPE,
+    ERR_STREAM_NULL_VALUES,
+  },
+} = require('internal/errors');
 
 function from(Readable, iterable, opts) {
   let iterator;
@@ -43,6 +46,7 @@ function from(Readable, iterable, opts) {
     // TODO(ronag): What options should be allowed?
     ...opts,
   });
+  const originalDestroy = readable._destroy;
 
   // Flag to protect against _read
   // being called before last iteration completion.
@@ -64,11 +68,18 @@ function from(Readable, iterable, opts) {
   };
 
   readable._destroy = function(error, cb) {
-    PromisePrototypeThen(
-      close(error),
-      () => process.nextTick(cb, error), // nextTick is here in case cb throws
-      (e) => process.nextTick(cb, e || error),
-    );
+    originalDestroy.call(this, error, (destroyError) => {
+      const combinedError = destroyError || error;
+      PromisePrototypeThen(
+        close(combinedError),
+        // nextTick is here in case cb throws
+        () => process.nextTick(cb, combinedError),
+        (closeError) => process.nextTick(
+          cb,
+          aggregateTwoErrors(combinedError, closeError),
+        ),
+      );
+    });
   };
 
   async function close(error) {

test/parallel/test-stream-readable-compose.js

@@ -116,6 +116,26 @@ const assert = require('assert');
   ).then(common.mustCall());
 }
 
+{
+  // Errors from nested `.compose()` calls should propagate instead of hanging.
+  const stream = Readable.from(['hello'])
+    .compose(async function *(source) { // eslint-disable-line require-yield
+      for await (const chunk of source) {
+        throw new Error(`boom: ${chunk}`);
+      }
+    })
+    .compose(async function *(source) {
+      for await (const chunk of source) {
+        yield chunk;
+      }
+    });
+
+  assert.rejects(
+    stream.toArray(),
+    /boom: hello/,
+  ).then(common.mustCall());
+}
+
 {
   // AbortSignal
   const ac = new AbortController();