diff --git a/lib/api/api-stream.js b/lib/api/api-stream.js index 249d6774f6b..f111d70efc1 100644 --- a/lib/api/api-stream.js +++ b/lib/api/api-stream.js @@ -1,7 +1,6 @@ 'use strict' const assert = require('node:assert') -const { finished } = require('node:stream') const { AsyncResource } = require('node:async_hooks') const { InvalidArgumentError, InvalidReturnValueError } = require('../core/errors') const util = require('../core/util') @@ -9,6 +8,54 @@ const { addSignal, removeSignal } = require('./abort-signal') function noop () {} +function getWritableError (stream) { + return stream.errored ?? stream.writableErrored ?? stream._writableState?.errored +} + +function createPrematureCloseError () { + const err = new Error('Premature close') + err.code = 'ERR_STREAM_PREMATURE_CLOSE' + return err +} + +function trackWritableLifecycle (stream, callback) { + let done = false + + const cleanup = () => { + stream.removeListener('close', onClose) + stream.removeListener('error', onError) + stream.removeListener('finish', onFinish) + } + + const finish = (err, fromErrorEvent = false) => { + if (done) { + return + } + + done = true + cleanup() + callback(err, fromErrorEvent) + } + + const onClose = () => { + const err = getWritableError(stream) + finish(err ?? (!stream.writableFinished ? createPrematureCloseError() : undefined)) + } + + const onError = (err) => finish(err, true) + const onFinish = () => finish() + + stream.on('close', onClose) + stream.on('error', onError) + stream.on('finish', onFinish) + + if (stream.closed) { + process.nextTick(onClose) + } else if (stream.writableFinished) { + process.nextTick(onFinish) + } +} + class StreamHandler extends AsyncResource { constructor (opts, factory, callback) { if (!opts || typeof opts !== 'object') { @@ -117,20 +164,19 @@ class StreamHandler extends AsyncResource { throw new InvalidReturnValueError('expected Writable') } - // TODO: Avoid finished. It registers an unnecessary amount of listeners. - finished(res, { readable: false }, (err) => { + trackWritableLifecycle(res, (err, fromErrorEvent) => { const { callback, res, opaque, trailers, abort } = this this.res = null if (err || !res?.readable) { - util.destroy(res, err) + util.destroy(res, fromErrorEvent ? undefined : err) } this.callback = null this.runInAsyncScope(callback, null, err || null, { opaque, trailers }) if (err) { - abort() + abort(err) } }) diff --git a/test/client-stream.js b/test/client-stream.js index 24d64bf822c..684ccc3e1db 100644 --- a/test/client-stream.js +++ b/test/client-stream.js @@ -157,6 +157,38 @@ test('stream GET destroy res', async (t) => { await t.completed }) +test('stream GET destroy res without error', async (t) => { + t = tspl(t, { plan: 1 }) + + const server = createServer({ joinDuplicateHeaders: true }, (req, res) => { + res.write('hello') + setImmediate(() => { + res.end(' world') + }) + }) + after(() => server.close()) + + server.listen(0, () => { + const client = new Client(`http://localhost:${server.address().port}`) + after(() => client.close()) + + client.stream({ + path: '/', + method: 'GET' + }, () => { + const pt = new PassThrough() + pt.on('data', () => { + pt.destroy() + }) + return pt + }, (err) => { + t.strictEqual(err.code, 'ERR_STREAM_PREMATURE_CLOSE') + }) + }) + + await t.completed +}) + test('stream GET remote destroy', async (t) => { t = tspl(t, { plan: 4 }) @@ -283,6 +315,33 @@ test('stream waits only for writable side', async (t) => { await t.completed }) +test('stream accepts already finished writable', async (t) => { + t = tspl(t, { plan: 1 }) + + const server = createServer({ joinDuplicateHeaders: true }, (req, res) => { + res.end() + }) + after(() => server.close()) + + server.listen(0, () => { + const client = new Client(`http://localhost:${server.address().port}`) + after(() => client.close()) + + client.stream({ + path: '/', + method: 'GET' + }, () => { + const pt = new PassThrough({ autoDestroy: false }) + pt.end() + return pt + }, (err) => { + t.ifError(err) + }) + }) + + await t.completed +}) + test('stream args validation', async (t) => { t = tspl(t, { plan: 3 })