Commit 9f12427

- **Every piped chunk reaches `write()` by property lookup.** Node 26.11's `Readable.pipe` fast path (nodejs/node#66182) calls `_write()` directly when pipe's listener is a byte-mode source's only `'data'` listener, so `compression`'s instance patch of `res.write` was skipped for most chunks of `send(...).pipe(res)`: raw bytes inside the gzip stream, or a stall once `compression` had routed pipe's `'drain'` to its idle zlib stream. On runtimes that have the fast path (`'writeKnownBuffer' in Writable`, a static it added) the response adds an inert `'data'` listener to each byte-mode core `Readable` on `'pipe'`, which fails the fast path's single-listener check, and removes it on `'unpipe'` through a listener prepended in the constructor: pipe's own cleanup resumes a source awaiting `'drain'` while any `'data'` listener remains, which would leave an unpiped source flowing into nothing. Object-mode and decoded sources, and legacy `Stream.prototype.pipe` (which never emits `'unpipe'`), are skipped. Remove it once every supported Node 26 release dispatches pipe through an overridden `write()`.

import { PassThrough, Readable, Writable } from 'node:stream';

const ignoreChunk = () => {};

const pipeSkipsWrite = 'writeKnownBuffer' in Writable;

function guardPipeSource(source: Readable) {

// object mode, a decoder, and legacy or userland pipe() never take the fast path; legacy pipe() never emits 'unpipe'

if (source instanceof Readable && !source.readableObjectMode && !source.readableEncoding)

source.on('data', ignoreChunk);

function releasePipeSource(source: unknown) {

if (source instanceof Readable) source.removeListener('data', ignoreChunk);

if (pipeSkipsWrite) {

this.on('pipe', guardPipeSource);

this.prependListener('unpipe', releasePipeSource);

}

const { PassThrough, Readable, Stream } = require('node:stream');

for (const objectMode of [false, true]) {

it(`runs a write() that middleware patched for every chunk ${objectMode ? 'an object' : 'a byte'}-mode Readable pipes in`, async function () {

const request = makeRequest();

const chunks = [];

for (let offset = 0; offset < BODY.length; offset += CHUNK_SIZE)

chunks.push(BODY.subarray(offset, offset + CHUNK_SIZE));

let next = 0;

const source = new Readable({

objectMode,

read() {

this.push(next < chunks.length ? chunks[next++] : null);

},

});

let patchedWrites = 0;

const responsePromise = request.withNodeAdapter((req, res) => {

const write = res.write;

res.write = function (...args) {

patchedWrites++;

return write.apply(this, args);

};

source.pipe(res);

});

const { body } = await withTimeout(responsePromise, 'response headers');

const received = await withTimeout(collectSlowly(body), 'the response body');

assert.deepStrictEqual(received, BODY);

assert.strictEqual(patchedWrites, chunks.length);

await waitUntil(() => source.listenerCount('data') === 0, "the source's 'data' listeners to be removed");

});

}

it('leaves a source paused with its data when its stalled pipe is unpiped or the response is destroyed, and no listener on any source', async function () {

const request = makeRequest();

const unpiped = new PassThrough();

const destroyed = new PassThrough();

let response;

const responsePromise = request.withNodeAdapter((req, res) => {

response = res;

res.writeHead(200);

unpiped.pipe(res);

destroyed.pipe(res);

});

await withTimeout(responsePromise, 'response headers');

for (let offset = 0; offset < BODY.length; offset += CHUNK_SIZE)

unpiped.write(BODY.subarray(offset, offset + CHUNK_SIZE));

await waitUntil(() => response.writableNeedDrain && unpiped.readableLength > 0, 'the pipe to stall awaiting drain');

const buffered = unpiped.readableLength;

unpiped.unpipe(response);

await sleep(1);

assert.strictEqual(unpiped.listenerCount('data'), 0);

assert.strictEqual(unpiped.readableFlowing, false);

assert.strictEqual(unpiped.readableLength, buffered);

destroyed.write(BODY.subarray(0, CHUNK_SIZE));

destroyed.write(BODY.subarray(CHUNK_SIZE, 2 * CHUNK_SIZE));

await waitUntil(() => destroyed.readableFlowing === false, 'the second pipe to stall awaiting drain');

const legacy = new Stream();

legacy.pipe(response);

legacy.emit('end');

assert.strictEqual(legacy.listenerCount('data'), 0);

response.destroy();

await waitUntil(() => destroyed.listenerCount('data') === 0, "the destroyed pipe's 'data' listeners to be removed");

await sleep(1);

assert.strictEqual(destroyed.readableFlowing, false);

assert.ok(destroyed.readableLength > 0, 'the source lost the data it had not delivered');

});

// npm still counts the dependency it would add under --dry-run, so asserting on that count

const report = response.application.npm_output;

assert.equal(report.added, 1, JSON.stringify(response.application));

// npm 11 names what it would add; npm 10's report only counts it

if (report.add)

assert.deepStrictEqual(

report.add.map(({ name }) => name),

['local-dependency']

);

// the shim is a POSIX shell script; its output and argv files travel as environment values, not as

// text in the script, so a `$` or a backtick in TMPDIR is never expanded by the shell that runs it

async function withNpmShim(stdout, callback) {

const stdoutPath = path.join(shimDir, 'stdout.txt');

fs.writeFileSync(stdoutPath, stdout);

'#!/bin/sh\nprintf \'%s\\n\' "$@" > "$HARPER_TEST_NPM_ARGV_PATH"\ncat "$HARPER_TEST_NPM_STDOUT_PATH"\n',

process.env.HARPER_TEST_NPM_ARGV_PATH = path.join(shimDir, 'argv.txt');

process.env.HARPER_TEST_NPM_STDOUT_PATH = stdoutPath;

return await callback(process.env.HARPER_TEST_NPM_ARGV_PATH);

delete process.env.HARPER_TEST_NPM_STDOUT_PATH;

}

it('runs npm with the registry audit disabled', async function () {

if (process.platform === 'win32') return this.skip();

const argv = await withNpmShim('{"added":0}\n', async (argvPath) => {

await installModules({ projects: ['application'] });

return fs.readFileSync(argvPath, 'utf8').split('\n').filter(Boolean);

});

it('parses the JSON report out of the dry-run diff that npm before 11.20 prints ahead of it', async function () {

if (process.platform === 'win32') return this.skip();

const cases = [

['add local-dependency 1.0.0\n{\n "added": 1\n}\n', { added: 1 }],

['add local-dependency 1.0.0\r\n{\r\n "added": 1\r\n}\r\n', { added: 1 }],

['{\n "added": 1\n}\n', { added: 1 }],

['{"from":"a lifecycle script"}\n{\n "added": 1\n}\n', { added: 1 }],

['not json\n', 'not json'],

];

for (const [stdout, expected] of cases) {

const response = await withNpmShim(stdout, () => installModules({ projects: ['application'], dry_run: true }));

assert.deepStrictEqual(response.application.npm_output, expected, JSON.stringify(stdout));

}

});

responseObject[project].npm_output = stdout ? parseNpmReport(stdout) : null;

function parseNpmReport(stdout: string) {

try {

return JSON.parse(stdout);

} catch {

const reportStart = stdout.lastIndexOf('\n{') + 1;

if (reportStart > 0) {

try {

return JSON.parse(stdout.slice(reportStart));

} catch {}

}

return stdout.replace('\n', '');

}