From ef01396227b35f2332c162c11d79ab6bb398e196 Mon Sep 17 00:00:00 2001 From: Matteo Collina Date: Wed, 16 Sep 2026 15:41:28 +0200 Subject: [PATCH] fix: honor backpressure in decompression interceptor (#5829) * fix: honor decompression backpressure Signed-off-by: Matteo Collina * fix(decompress): disable size limit by default Signed-off-by: Matteo Collina --------- Signed-off-by: Matteo Collina (cherry picked from commit 19901d8b73cb46a48e817dab4afce790f37fd2c1) --- docs/docs/api/Dispatcher.md | 1 + lib/handler/retry-handler.js | 91 ++- lib/interceptor/decompress.js | 296 +++++++--- test/client-retry-resume-backpressure.js | 127 ++++ test/interceptors/decompress.js | 716 ++++++++++++++++++++++- test/retry-handler-controller-proxy.js | 195 ++++++ types/interceptors.d.ts | 2 +- 7 files changed, 1337 insertions(+), 91 deletions(-) create mode 100644 test/client-retry-resume-backpressure.js create mode 100644 test/retry-handler-controller-proxy.js diff --git a/docs/docs/api/Dispatcher.md b/docs/docs/api/Dispatcher.md index c9df4c63255..13b6540201d 100644 --- a/docs/docs/api/Dispatcher.md +++ b/docs/docs/api/Dispatcher.md @@ -1153,6 +1153,7 @@ The `decompress` interceptor automatically decompresses response bodies that are - `skipErrorResponses` - Whether to skip decompression for error responses (status codes >= 400). Default: `true`. - `skipStatusCodes` - Array of status codes to skip decompression for. Default: `[204, 304]`. +- `maxSize` - Maximum decompressed response size in bytes for each decompression stage. Set to `0` to disable the limit. Default: `0`. **Example - Basic Decompress Interceptor** diff --git a/lib/handler/retry-handler.js b/lib/handler/retry-handler.js index 7cc4c1ca1b6..94b27b8e666 100644 --- a/lib/handler/retry-handler.js +++ b/lib/handler/retry-handler.js @@ -35,6 +35,55 @@ function validatePartialResponseContentLength (headers, range, statusCode, retry } } +// A stable controller handed to the downstream handler for the lifetime of the +// request. Each transparent retry/resume is a separate dispatch with its own +// connection controller. The proxy always forwards to the active connection +// while preserving a downstream pause across controller replacement. +class RetryController { + #paused = false + #target = null + + set target (target) { + this.#target = target + if (this.#paused) { + target?.pause() + } + } + + get target () { return this.#target } + + pause () { + this.#paused = true + this.#target?.pause() + } + + resume () { + this.#paused = false + this.#target?.resume() + } + + abort (reason) { + this.#target?.abort(reason) + } + + get paused () { return this.#paused || (this.#target?.paused ?? false) } + get aborted () { return this.#target?.aborted ?? false } + get reason () { return this.#target?.reason ?? null } + get rawHeaders () { return this.#target?.rawHeaders ?? null } + set rawHeaders (value) { + if (this.#target) { + this.#target.rawHeaders = value + } + } + + get rawTrailers () { return this.#target?.rawTrailers ?? null } + set rawTrailers (value) { + if (this.#target) { + this.#target.rawTrailers = value + } + } +} + class RetryHandler { constructor (opts, { dispatch, handler }) { const { retryOptions, ...dispatchOpts } = opts @@ -89,6 +138,7 @@ class RetryHandler { this.start = 0 this.end = null this.etag = null + this.controllerProxy = new RetryController() } onResponseStartWithRetry (controller, statusCode, headers, statusMessage, err) { @@ -99,11 +149,11 @@ class RetryHandler { // The downstream handler already received the response from an // earlier attempt. Forwarding this response would replace the // downstream body and leave the original body pending forever. - this.handler.onResponseError?.(controller, err) + this.handler.onResponseError?.(this.controllerProxy, err) } else { this.headersSent = true this.checkpointResponseEnd(headers) - this.handler.onResponseStart?.(controller, statusCode, headers, statusMessage) + this.handler.onResponseStart?.(this.controllerProxy, statusCode, headers, statusMessage) } } else { this.error = err @@ -115,7 +165,7 @@ class RetryHandler { if (isDisturbed(this.opts.body)) { this.headersSent = true this.checkpointResponseEnd(headers) - this.handler.onResponseStart?.(controller, statusCode, headers, statusMessage) + this.handler.onResponseStart?.(this.controllerProxy, statusCode, headers, statusMessage) return } @@ -125,11 +175,11 @@ class RetryHandler { // The downstream handler already received the response from an // earlier attempt. Forwarding this response would replace the // downstream body and leave the original body pending forever. - this.handler.onResponseError?.(controller, passedErr) + this.handler.onResponseError?.(this.controllerProxy, passedErr) } else { this.headersSent = true this.checkpointResponseEnd(headers) - this.handler.onResponseStart?.(controller, statusCode, headers, statusMessage) + this.handler.onResponseStart?.(this.controllerProxy, statusCode, headers, statusMessage) } controller.resume() return @@ -165,13 +215,14 @@ class RetryHandler { } onRequestStart (controller, context) { + this.controllerProxy.target = controller if (!this.headersSent) { - this.handler.onRequestStart?.(controller, context) + this.handler.onRequestStart?.(this.controllerProxy, context) } } - onRequestUpgrade (controller, statusCode, headers, socket) { - this.handler.onRequestUpgrade?.(controller, statusCode, headers, socket) + onRequestUpgrade (_controller, statusCode, headers, socket) { + this.handler.onRequestUpgrade?.(this.controllerProxy, statusCode, headers, socket) } static [kRetryHandlerDefaultRetry] (err, { state, opts }, cb) { @@ -302,7 +353,7 @@ class RetryHandler { if (range == null) { this.headersSent = true this.handler.onResponseStart?.( - controller, + this.controllerProxy, statusCode, headers, statusMessage @@ -351,7 +402,7 @@ class RetryHandler { this.headersSent = true this.handler.onResponseStart?.( - controller, + this.controllerProxy, statusCode, headers, statusMessage @@ -364,30 +415,30 @@ class RetryHandler { } } - onResponseData (controller, chunk) { + onResponseData (_controller, chunk) { if (this.error) { return } this.start += chunk.length - this.handler.onResponseData?.(controller, chunk) + this.handler.onResponseData?.(this.controllerProxy, chunk) } - onResponseEnd (controller, trailers) { + onResponseEnd (_controller, trailers) { if (this.error && this.retryOpts.throwOnError) { throw this.error } if (!this.error) { this.retryCount = 0 - return this.handler.onResponseEnd?.(controller, trailers) + return this.handler.onResponseEnd?.(this.controllerProxy, trailers) } - this.retry(controller) + this.retry() } - retry (controller) { + retry () { if (this.start !== 0) { const headers = { range: `bytes=${this.start}-${this.end ?? ''}` } @@ -409,23 +460,23 @@ class RetryHandler { this.retryCountCheckpoint = this.retryCount this.dispatch(this.opts, this) } catch (err) { - this.handler.onResponseError?.(controller, err) + this.handler.onResponseError?.(this.controllerProxy, err) } } onResponseError (controller, err) { if (controller?.aborted || isDisturbed(this.opts.body) || (this.headersSent && !this.resume)) { - this.handler.onResponseError?.(controller, err) + this.handler.onResponseError?.(this.controllerProxy, err) return } function shouldRetry (returnedErr) { if (!returnedErr) { - this.retry(controller) + this.retry() return } - this.handler?.onResponseError?.(controller, returnedErr) + this.handler?.onResponseError?.(this.controllerProxy, returnedErr) } // We reconcile in case of a mix between network errors diff --git a/lib/interceptor/decompress.js b/lib/interceptor/decompress.js index 6c769aeff16..02d6ebc0629 100644 --- a/lib/interceptor/decompress.js +++ b/lib/interceptor/decompress.js @@ -10,6 +10,60 @@ const { runtimeFeatures } = require('../util/runtime-features') /** @typedef {import('node:stream').Transform} Controller */ /** @typedef {Transform&import('node:zlib').Zlib} DecompressorStream */ +class DecompressController { + #onPause + #onResume + #onAbort + #paused = false + + constructor (onPause, onResume, onAbort) { + this.#onPause = onPause + this.#onResume = onResume + this.#onAbort = onAbort + this.target = null + } + + pause () { + if (this.#paused) { + return + } + + this.#paused = true + this.#onPause() + } + + resume () { + if (!this.#paused) { + return + } + + this.#paused = false + this.#onResume() + } + + abort (reason) { + this.target?.abort(reason) + this.#onAbort(reason) + } + + get paused () { return this.#paused } + get aborted () { return this.target?.aborted ?? false } + get reason () { return this.target?.reason ?? null } + get rawHeaders () { return this.target?.rawHeaders ?? null } + set rawHeaders (value) { + if (this.target) { + this.target.rawHeaders = value + } + } + + get rawTrailers () { return this.target?.rawTrailers ?? null } + set rawTrailers (value) { + if (this.target) { + this.target.rawTrailers = value + } + } +} + /** @type {Record DecompressorStream>} */ const supportedEncodings = { gzip: createGunzip, @@ -22,7 +76,7 @@ const supportedEncodings = { } const defaultSkipStatusCodes = /** @type {const} */ ([204, 304]) -const defaultMaxSize = 64 * 1024 * 1024 +const defaultMaxSize = 0 /** * Limits the output of one stage in a decompression chain. @@ -54,7 +108,7 @@ let warningEmitted = /** @type {boolean} */ (false) * @typedef {Object} DecompressHandlerOptions * @property {number[]|Readonly} [skipStatusCodes=[204, 304]] - List of status codes to skip decompression for * @property {boolean} [skipErrorResponses] - Whether to skip decompression for error responses (status codes >= 400) - * @property {number} [maxSize=67108864] - Maximum decompressed response size in bytes + * @property {number} [maxSize=0] - Maximum decompressed response size in bytes. 0 disables the limit */ class DecompressHandler extends DecoratorHandler { @@ -74,16 +128,130 @@ class DecompressHandler extends DecoratorHandler { #terminated = false /** @type {boolean} */ #inputEnded = false + /** @type {boolean} */ + #inputBackpressured = false + /** @type {boolean} */ + #upstreamPaused = false + /** @type {boolean} */ + #draining = false + /** @type {boolean} */ + #drainRequested = false + /** @type {boolean} */ + #completionPending = false + /** @type {DecompressorStream | undefined} */ + #finalDecompressor + /** @type {DecompressController} */ + #controller constructor (handler, { skipStatusCodes = defaultSkipStatusCodes, skipErrorResponses = true, maxSize = defaultMaxSize } = {}) { - if (!Number.isSafeInteger(maxSize) || maxSize < 1) { - throw new InvalidArgumentError('maxSize must be a positive integer') + if (!Number.isSafeInteger(maxSize) || maxSize < 0) { + throw new InvalidArgumentError('maxSize must be a non-negative integer') } super(handler) this.#skipStatusCodes = skipStatusCodes this.#skipErrorResponses = skipErrorResponses this.#maxSize = maxSize + this.#controller = new DecompressController( + () => this.#onDownstreamPause(), + () => this.#onDownstreamResume(), + reason => { + if (this.#inputEnded && !this.#terminated) { + this.onResponseError(this.#controller, reason) + } + } + ) + } + + #onDownstreamPause () { + this.#pauseUpstream() + } + + #onDownstreamResume () { + const drainWasDeferred = this.#draining + this.#drainOutput() + if (!drainWasDeferred) { + this.#resumeUpstreamIfNeeded() + this.#finishIfReady() + } + } + + #pauseUpstream () { + if (!this.#upstreamPaused && !this.#terminated) { + this.#upstreamPaused = true + this.#controller.target?.pause() + } + } + + #resumeUpstreamIfNeeded () { + if (this.#upstreamPaused && !this.#controller.paused && !this.#inputBackpressured) { + this.#upstreamPaused = false + if (!this.#inputEnded) { + this.#controller.target?.resume() + } + } + } + + #drainOutput () { + if (this.#terminated || this.#controller.paused || !this.#finalDecompressor) { + return + } + + if (this.#draining) { + this.#drainRequested = true + return + } + + this.#draining = true + try { + do { + this.#drainRequested = false + let chunk + while (!this.#terminated && !this.#controller.paused && (chunk = this.#finalDecompressor.read()) !== null) { + if (this.#maxSize > 0) { + const decompressedSize = this.#decompressedSize + chunk.length + if (decompressedSize > this.#maxSize) { + this.#fail(new ResponseExceededMaxSizeError( + `Decompressed response size (${decompressedSize}) exceeded maxSize (${this.#maxSize})` + )) + return + } + + this.#decompressedSize = decompressedSize + } + + const result = super.onResponseData(this.#controller, chunk) + if (result === false && !this.#controller.paused) { + this.#controller.pause() + } + } + } while (this.#drainRequested && !this.#terminated && !this.#controller.paused) + } finally { + this.#draining = false + } + + this.#resumeUpstreamIfNeeded() + this.#finishIfReady() + } + + #finishIfReady () { + if (this.#terminated || !this.#completionPending || this.#controller.paused || this.#draining) { + return + } + + this.#terminated = true + this.#cleanupDecompressors() + super.onResponseEnd(this.#controller, this.#trailers) + } + + #onDecompressionEnd () { + if (this.#terminated) { + return + } + + this.#completionPending = true + this.#drainOutput() + this.#finishIfReady() } /** @@ -139,7 +307,7 @@ class DecompressHandler extends DecoratorHandler { const streams = [] for (let i = 0; i < decompressors.length; i++) { streams.push(decompressors[i]) - if (i < decompressors.length - 1) { + if (i < decompressors.length - 1 && this.#maxSize > 0) { streams.push(createMaxSizeLimiter(this.#maxSize)) } } @@ -149,11 +317,10 @@ class DecompressHandler extends DecoratorHandler { /** * Stops decompression and reports an error. - * @param {Controller} controller - The controller to coordinate with * @param {Error} error - The decompression error * @returns {void} */ - #fail (controller, error) { + #fail (error) { if (this.#terminated) { return } @@ -161,75 +328,41 @@ class DecompressHandler extends DecoratorHandler { if (this.#inputEnded) { // The request is already marked complete once the compressed input ends, // so controller.abort() can no longer propagate decoder flush errors. - this.onResponseError(controller, error) + this.onResponseError(this.#controller, error) } else { - controller.abort(error) + this.#controller.abort(error) } } /** - * Sets up event handlers for a decompressor stream using readable events + * Sets up event handlers for the final decompressor stream. * @param {DecompressorStream} decompressor - The decompressor stream - * @param {Controller} controller - The controller to coordinate with * @returns {void} */ - #setupDecompressorEvents (decompressor, controller) { - decompressor.on('readable', () => { - if (this.#terminated) { - return - } - - let chunk - while ((chunk = decompressor.read()) !== null) { - const decompressedSize = this.#decompressedSize + chunk.length - if (decompressedSize > this.#maxSize) { - this.#fail(controller, new ResponseExceededMaxSizeError( - `Decompressed response size (${decompressedSize}) exceeded maxSize (${this.#maxSize})` - )) - return - } - - this.#decompressedSize = decompressedSize - const result = super.onResponseData(controller, chunk) - if (result === false) { - break - } - } - }) - - decompressor.on('error', (error) => { - this.#fail(controller, error) - }) + #setupDecompressorEvents (decompressor) { + this.#finalDecompressor = decompressor + decompressor.on('readable', () => this.#drainOutput()) + decompressor.on('error', (error) => this.#fail(error)) } /** * Sets up event handling for a single decompressor - * @param {Controller} controller - The controller to handle events * @returns {void} */ - #setupSingleDecompressor (controller) { + #setupSingleDecompressor () { const decompressor = this.#decompressors[0] - this.#setupDecompressorEvents(decompressor, controller) + this.#setupDecompressorEvents(decompressor) - decompressor.on('end', () => { - if (this.#terminated) { - return - } - - this.#terminated = true - this.#cleanupDecompressors() - super.onResponseEnd(controller, this.#trailers) - }) + decompressor.on('end', () => this.#onDecompressionEnd()) } /** * Sets up event handling for multiple chained decompressors using pipeline - * @param {Controller} controller - The controller to handle events * @returns {void} */ - #setupMultipleDecompressors (controller) { + #setupMultipleDecompressors () { const lastDecompressor = this.#decompressors[this.#decompressors.length - 1] - this.#setupDecompressorEvents(lastDecompressor, controller) + this.#setupDecompressorEvents(lastDecompressor) pipeline(this.#decompressors, (err) => { if (this.#terminated) { @@ -237,13 +370,26 @@ class DecompressHandler extends DecoratorHandler { } if (err) { - this.#fail(controller, err) + this.#fail(err) return } - this.#terminated = true - this.#cleanupDecompressors() - super.onResponseEnd(controller, this.#trailers) + this.#onDecompressionEnd() + }) + } + + #setupInputBackpressure () { + const decompressor = this.#decompressors[0] + decompressor.on('drain', () => { + if (this.#terminated) { + return + } + + this.#inputBackpressured = false + if (!this.#controller.paused) { + this.#drainOutput() + this.#resumeUpstreamIfNeeded() + } }) } @@ -253,6 +399,16 @@ class DecompressHandler extends DecoratorHandler { */ #cleanupDecompressors () { this.#decompressors.length = 0 + this.#finalDecompressor = undefined + } + + onRequestStart (controller, context) { + this.#controller.target = controller + return super.onRequestStart(this.#controller, context) + } + + onRequestUpgrade (controller, statusCode, headers, socket) { + return super.onRequestUpgrade(this.#controller, statusCode, headers, socket) } /** @@ -267,14 +423,14 @@ class DecompressHandler extends DecoratorHandler { // If content encoding is not supported or status code is in skip list if (this.#shouldSkipDecompression(contentEncoding, statusCode)) { - return super.onResponseStart(controller, statusCode, headers, statusMessage) + return super.onResponseStart(this.#controller, statusCode, headers, statusMessage) } const decompressors = this.#createDecompressionChain(contentEncoding.toLowerCase()) if (decompressors.length === 0) { this.#cleanupDecompressors() - return super.onResponseStart(controller, statusCode, headers, statusMessage) + return super.onResponseStart(this.#controller, statusCode, headers, statusMessage) } this.#decompressors = decompressors @@ -282,8 +438,8 @@ class DecompressHandler extends DecoratorHandler { // Remove compression headers since we're decompressing const { 'content-encoding': _, 'content-length': __, ...newHeaders } = headers - if (controller?.rawHeaders) { - const rawHeaders = controller.rawHeaders + if (this.#controller.rawHeaders) { + const rawHeaders = this.#controller.rawHeaders if (Array.isArray(rawHeaders)) { const filteredHeaders = [] @@ -309,13 +465,14 @@ class DecompressHandler extends DecoratorHandler { } } + this.#setupInputBackpressure() if (this.#decompressors.length === 1) { - this.#setupSingleDecompressor(controller) + this.#setupSingleDecompressor() } else { - this.#setupMultipleDecompressors(controller) + this.#setupMultipleDecompressors() } - return super.onResponseStart(controller, statusCode, newHeaders, statusMessage) + return super.onResponseStart(this.#controller, statusCode, newHeaders, statusMessage) } /** @@ -325,10 +482,13 @@ class DecompressHandler extends DecoratorHandler { */ onResponseData (controller, chunk) { if (this.#decompressors.length > 0) { - this.#decompressors[0].write(chunk) + if (!this.#decompressors[0].write(chunk)) { + this.#inputBackpressured = true + this.#pauseUpstream() + } return } - super.onResponseData(controller, chunk) + return super.onResponseData(this.#controller, chunk) } /** @@ -343,7 +503,7 @@ class DecompressHandler extends DecoratorHandler { this.#decompressors[0].end() return } - super.onResponseEnd(controller, trailers) + return super.onResponseEnd(this.#controller, trailers) } /** @@ -361,7 +521,7 @@ class DecompressHandler extends DecoratorHandler { decompressor.destroy() } this.#cleanupDecompressors() - super.onResponseError(controller, err) + super.onResponseError(this.#controller, err) } } diff --git a/test/client-retry-resume-backpressure.js b/test/client-retry-resume-backpressure.js new file mode 100644 index 00000000000..1e3519ac9f0 --- /dev/null +++ b/test/client-retry-resume-backpressure.js @@ -0,0 +1,127 @@ +'use strict' + +const { tspl } = require('@matteo.collina/tspl') +const { test, after } = require('node:test') +const { createServer } = require('node:http') +const { Writable } = require('node:stream') +const { pipeline } = require('node:stream/promises') +const { once } = require('node:events') +const { Agent, RetryAgent, request } = require('..') + +const TOTAL = 4 * 1024 * 1024 +const PART = 1 * 1024 * 1024 +const sleep = (ms) => new Promise(resolve => setTimeout(resolve, ms)) + +// Both tests exercise the RetryHandler controller-proxy fix. A download is +// interrupted mid-body with a retryable error (ECONNRESET); RetryAgent +// transparently resumes it with a Range request on a NEW connection. Before the +// fix the downstream body kept flow-controlling the ORIGINAL (now-dead) +// connection's controller while data/pause flowed on the new one, so: +// - under backpressure the resumed body paused and was never resumed (it hung); +// - an abort hit the dead original instead of the live resumed connection. +// With the fix both follow the active connection. + +// Faithful reproduction of the reported hang. If the bug returns, pipeline() +// never resolves and the test is failed by the runner timeout (the npm scripts +// run borp with --timeout). Note: `node --test --test-timeout=0` disables that +// timeout, so run this under the npm test scripts. +test('RetryAgent resumes a backpressured body after a mid-stream connection drop', async (t) => { + t = tspl(t, { plan: 3 }) + + let requests = 0 + const server = createServer((req, res) => { + requests++ + if (!req.headers.range) { + res.writeHead(200, { 'content-length': String(TOTAL) }) + res.write(Buffer.alloc(PART)) + setTimeout(() => res.socket.resetAndDestroy(), 100) + } else { + const start = Number(/bytes=(\d+)-/.exec(req.headers.range)[1]) + res.writeHead(206, { + 'content-range': `bytes ${start}-${TOTAL - 1}/${TOTAL}`, + 'content-length': String(TOTAL - start) + }) + res.end(Buffer.alloc(TOTAL - start)) + } + }) + after(() => server.close()) + server.listen(0) + await once(server, 'listening') + + const dispatcher = new RetryAgent(new Agent(), { maxRetries: 5, minTimeout: 100, timeoutFactor: 1 }) + after(() => dispatcher.close()) + + const { statusCode, body } = await request(`http://localhost:${server.address().port}`, { dispatcher }) + t.strictEqual(statusCode, 200) + + // Slow consumer -> sustained backpressure on the resumed connection (the trigger). + let received = 0 + const slow = new Writable({ + highWaterMark: 16 * 1024, + write (chunk, enc, cb) { received += chunk.length; setTimeout(cb, 5) } + }) + + await pipeline(body, slow) // before the fix this never resolves + + t.strictEqual(received, TOTAL) + t.ok(requests >= 2, 'the download was actually resumed on a new connection') + + await t.completed +}) + +// Companion check on the abort path. A flowing consumer keeps the resumed +// connection in-flight (no backpressure pause), then aborts. The server-side +// socket of the resumed connection must close: with the fix the abort reaches +// the live connection; before it, it hit the dead original and the resumed +// socket leaked. Bounded so a regression fails fast instead of waiting. +test('RetryAgent aborts the resumed connection (not the dead original) after a drop', async (t) => { + t = tspl(t, { plan: 2 }) + + let resumeSocket = null + let onResume + const resumed = new Promise(resolve => { onResume = resolve }) + + const server = createServer((req, res) => { + if (!req.headers.range) { + res.writeHead(200, { 'content-length': String(TOTAL) }) + res.write(Buffer.alloc(PART)) + setTimeout(() => res.socket.resetAndDestroy(), 100) + } else { + resumeSocket = res.socket + const start = Number(/bytes=(\d+)-/.exec(req.headers.range)[1]) + res.writeHead(206, { + 'content-range': `bytes ${start}-${TOTAL - 1}/${TOTAL}`, + 'content-length': String(TOTAL - start) + }) + res.write(Buffer.alloc(64 * 1024)) // keep the resumed connection in-flight + onResume() + } + }) + after(() => server.close()) + server.listen(0) + await once(server, 'listening') + + const dispatcher = new RetryAgent(new Agent(), { maxRetries: 5, minTimeout: 100, timeoutFactor: 1 }) + after(() => dispatcher.destroy()) + + const ac = new AbortController() + const { statusCode, body } = await request(`http://localhost:${server.address().port}`, { dispatcher, signal: ac.signal }) + t.strictEqual(statusCode, 200) + + body.on('data', () => {}).on('error', () => {}) // flowing consumer; swallow the abort error + + await resumed // downloading on the resumed (206) connection now + await sleep(50) // let a little data flow on it + + const closed = once(resumeSocket, 'close') // attach before aborting + ac.abort() + + let timer + const bound = new Promise(resolve => { timer = setTimeout(() => resolve(false), 4000) }) + const wasClosed = await Promise.race([closed.then(() => true), bound]) + clearTimeout(timer) + + t.ok(wasClosed, 'aborting closed the resumed connection, not the dead original') + + await t.completed +}) diff --git a/test/interceptors/decompress.js b/test/interceptors/decompress.js index 2540d8d0225..2bb5d3df996 100644 --- a/test/interceptors/decompress.js +++ b/test/interceptors/decompress.js @@ -1,6 +1,7 @@ 'use strict' const { test, after } = require('node:test') +const assert = require('node:assert/strict') const { createServer } = require('node:http') const { once } = require('node:events') const { createGzip, createDeflate, createBrotliCompress, createZstdCompress, deflateSync, gzipSync } = require('node:zlib') @@ -9,6 +10,56 @@ const { tspl } = require('@matteo.collina/tspl') const { Client, errors, getGlobalDispatcher, setGlobalDispatcher, request, interceptors } = require('../..') const createDecompressInterceptor = require('../../lib/interceptor/decompress') +const immediate = () => new Promise(resolve => setImmediate(resolve)) + +function createControlledDispatch (handler, options, { forwardAbort = true } = {}) { + let sourceHandler + let abortReason = null + const events = [] + const state = { + paused: false, + aborted: false, + pauseCalls: 0, + resumeCalls: 0 + } + const controller = { + rawHeaders: ['Content-Encoding', 'gzip'], + rawTrailers: null, + pause () { + state.paused = true + state.pauseCalls++ + events.push('controller-pause') + }, + resume () { + state.paused = false + state.resumeCalls++ + events.push('controller-resume') + }, + abort (reason) { + if (state.aborted) { + return + } + state.aborted = true + abortReason = reason + if (forwardAbort) { + sourceHandler.onResponseError(controller, reason) + } + }, + get paused () { return state.paused }, + get aborted () { return state.aborted }, + get reason () { return abortReason } + } + + const dispatch = createDecompressInterceptor(options)((opts, wrappedHandler) => { + sourceHandler = wrappedHandler + return true + }) + dispatch({ method: 'GET' }, handler) + sourceHandler.onRequestStart(controller, {}) + + return { controller, events, sourceHandler, state } +} + test('should decompress gzip response', async t => { t = tspl(t, { plan: 3 }) @@ -743,6 +794,584 @@ test('should handle empty encoding values', async t => { await t.completed }) +test('decompress backpressure pauses on the first decoded chunk', { timeout: 5000 }, async () => { + const payload = Buffer.alloc(1024 * 1024, 0x61) + const compressed = gzipSync(payload) + assert(compressed.length < 16 * 1024) + + let responseController + let endController + let endTrailers + let endCalls = 0 + let errorCalls = 0 + const chunks = [] + let firstDataResolve + const firstData = new Promise(resolve => { firstDataResolve = resolve }) + let endResolve + const ended = new Promise(resolve => { endResolve = resolve }) + + const handler = { + onRequestStart (controller) { + responseController = controller + }, + onResponseStart (controller) { + assert.strictEqual(controller, responseController) + }, + onResponseData (controller, chunk) { + assert.strictEqual(controller, responseController) + chunks.push(chunk) + if (chunks.length === 1) { + controller.pause() + firstDataResolve() + } + }, + onResponseEnd (controller, trailers) { + endController = controller + endTrailers = trailers + endCalls++ + endResolve() + }, + onResponseError () { + errorCalls++ + endResolve() + } + } + + const { controller, sourceHandler, state } = createControlledDispatch(handler) + sourceHandler.onResponseStart(controller, 200, { 'content-encoding': 'gzip' }, 'OK') + sourceHandler.onResponseData(controller, compressed) + controller.rawTrailers = ['X-Trailer', 'raw-value'] + const trailers = { 'x-trailer': 'value' } + sourceHandler.onResponseEnd(controller, trailers) + + await firstData + await immediate() + assert.equal(responseController.paused, true) + assert.equal(state.pauseCalls, 1) + assert.equal(chunks.length, 1) + assert(chunks[0].length < payload.length) + assert.equal(endCalls, 0) + assert.equal(errorCalls, 0) + + responseController.resume() + await ended + assert.strictEqual(endController, responseController) + assert.deepEqual(endTrailers, trailers) + assert.deepEqual(responseController.rawTrailers, controller.rawTrailers) + assert.deepEqual(Buffer.concat(chunks), payload) + assert(chunks.length > 1) + assert.equal(endCalls, 1) + assert.equal(errorCalls, 0) +}) + +for (const { name, contentEncoding, compress } of [ + { + name: 'single decoder', + contentEncoding: 'gzip', + compress: gzipSync + }, + { + name: 'chained decoders', + contentEncoding: 'gzip, deflate', + compress: payload => deflateSync(gzipSync(payload)) + } +]) { + test(`decompress backpressure delays ${name} completion after the final chunk`, { timeout: 5000 }, async () => { + const payload = Buffer.from('the only decoded chunk') + const compressed = compress(payload) + const chunks = [] + let responseController + let endCalls = 0 + let errorCalls = 0 + let firstDataResolve + const firstData = new Promise(resolve => { firstDataResolve = resolve }) + let endResolve + const ended = new Promise(resolve => { endResolve = resolve }) + const handler = { + onRequestStart (controller) { + responseController = controller + }, + onResponseStart () {}, + onResponseData (controller, chunk) { + chunks.push(chunk) + if (chunks.length === 1) { + controller.pause() + firstDataResolve() + } + }, + onResponseEnd () { + endCalls++ + endResolve() + }, + onResponseError () { + errorCalls++ + endResolve() + } + } + + const { controller, sourceHandler } = createControlledDispatch(handler) + sourceHandler.onResponseStart(controller, 200, { 'content-encoding': contentEncoding }, 'OK') + sourceHandler.onResponseData(controller, compressed) + sourceHandler.onResponseEnd(controller, { final: 'trailer' }) + + await firstData + await immediate() + assert.equal(chunks.length, 1) + assert.deepEqual(chunks[0], payload) + assert.equal(endCalls, 0) + assert.equal(errorCalls, 0) + + responseController.resume() + await ended + assert.equal(endCalls, 1) + assert.equal(errorCalls, 0) + }) +} + +test('decompress backpressure delays empty response completion while paused', { timeout: 5000 }, async () => { + const compressed = gzipSync(Buffer.alloc(0)) + let responseController + let dataCalls = 0 + let endCalls = 0 + let errorCalls = 0 + let endResolve + const ended = new Promise(resolve => { endResolve = resolve }) + const handler = { + onRequestStart (controller) { + responseController = controller + }, + onResponseStart (controller) { + controller.pause() + }, + onResponseData () { + dataCalls++ + }, + onResponseEnd () { + endCalls++ + endResolve() + }, + onResponseError () { + errorCalls++ + endResolve() + } + } + + const { controller, sourceHandler } = createControlledDispatch(handler) + sourceHandler.onResponseStart(controller, 200, { 'content-encoding': 'gzip' }, 'OK') + sourceHandler.onResponseData(controller, compressed) + sourceHandler.onResponseEnd(controller, {}) + + for (let i = 0; i < 5; i++) { + await immediate() + } + assert.equal(dataCalls, 0) + assert.equal(endCalls, 0) + assert.equal(errorCalls, 0) + + responseController.resume() + await ended + assert.equal(dataCalls, 0) + assert.equal(endCalls, 1) + assert.equal(errorCalls, 0) +}) + +test('decompress backpressure bounds an initially unread request body', { timeout: 5000 }, async t => { + const payload = Buffer.alloc(2 * 1024 * 1024, 0x62) + const compressed = gzipSync(payload) + let finishResolve + const responseFinished = new Promise(resolve => { finishResolve = resolve }) + const server = createServer({ joinDuplicateHeaders: true }, (_req, res) => { + res.once('finish', finishResolve) + res.writeHead(200, { + 'Content-Encoding': 'gzip', + 'Content-Length': compressed.length + }) + res.end(compressed) + }) + + server.listen(0) + await once(server, 'listening') + const client = new Client(`http://localhost:${server.address().port}`) + .compose(interceptors.decompress()) + + t.after(async () => { + await client.destroy() + server.closeAllConnections?.() + if (server.listening) { + server.close() + await once(server, 'close') + } + }) + + const { body } = await client.request({ + method: 'GET', + path: '/', + highWaterMark: 32 * 1024 + }) + + if (body.readableLength === 0) { + await once(body, 'readable') + } + await responseFinished + for (let i = 0; i < 100 && body.readableLength < body.readableHighWaterMark; i++) { + await immediate() + } + + assert(body.readableLength >= body.readableHighWaterMark) + assert( + body.readableLength <= body.readableHighWaterMark + 16 * 1024, + `buffered ${body.readableLength} bytes for HWM ${body.readableHighWaterMark}` + ) + assert(body.readableLength < payload.length / 4) + assert.deepEqual(Buffer.from(await body.arrayBuffer()), payload) +}) + +test('decompress backpressure honors the first decoder write result', { timeout: 5000 }, async () => { + const payload = Buffer.allocUnsafe(256 * 1024) + let random = 0x12345678 + for (let i = 0; i < payload.length; i++) { + random ^= random << 13 + random ^= random >>> 17 + random ^= random << 5 + payload[i] = random & 0xff + } + const compressed = gzipSync(payload) + assert(compressed.length > 64 * 1024) + + const chunks = [] + let endCalls = 0 + let receivedError + let terminalResolve + const terminal = new Promise(resolve => { terminalResolve = resolve }) + const handler = { + onRequestStart () {}, + onResponseStart () {}, + onResponseData (controller, chunk) { + chunks.push(chunk) + }, + onResponseEnd () { + endCalls++ + terminalResolve() + }, + onResponseError (controller, error) { + receivedError = error + terminalResolve() + } + } + + const { controller, events, sourceHandler, state } = createControlledDispatch(handler) + sourceHandler.onResponseStart(controller, 200, { 'content-encoding': 'gzip' }, 'OK') + events.push('source-data') + sourceHandler.onResponseData(controller, compressed) + + assert.equal(state.paused, true) + assert.equal(state.pauseCalls, 1) + while (state.resumeCalls === 0) { + await immediate() + } + assert.equal(state.resumeCalls, 1) + events.push('source-end') + sourceHandler.onResponseEnd(controller, {}) + await terminal + + assert.equal(receivedError, undefined) + assert.equal(endCalls, 1) + assert.deepEqual(Buffer.concat(chunks), payload) + assert(events.indexOf('controller-pause') > events.indexOf('source-data')) + assert(events.indexOf('controller-resume') > events.indexOf('controller-pause')) + assert(events.indexOf('source-end') > events.indexOf('controller-resume')) +}) + +test('decompress backpressure keeps input and downstream pause reasons independent', { timeout: 5000 }, async () => { + const payload = Buffer.allocUnsafe(128 * 1024) + let random = 0x87654321 + for (let i = 0; i < payload.length; i++) { + random ^= random << 13 + random ^= random >>> 17 + random ^= random << 5 + payload[i] = random & 0xff + } + const compressed = gzipSync(payload) + assert(compressed.length > 64 * 1024) + + let responseController + const chunks = [] + let terminalResolve + let terminalReject + const terminal = new Promise((resolve, reject) => { + terminalResolve = resolve + terminalReject = reject + }) + const handler = { + onRequestStart (controller) { + responseController = controller + }, + onResponseStart (controller) { + controller.pause() + }, + onResponseData (controller, chunk) { + chunks.push(chunk) + }, + onResponseEnd () { + terminalResolve() + }, + onResponseError (controller, error) { + terminalReject(error) + } + } + + const { controller, sourceHandler, state } = createControlledDispatch(handler) + sourceHandler.onResponseStart(controller, 200, { 'content-encoding': 'gzip' }, 'OK') + assert.equal(state.paused, true) + assert.equal(state.pauseCalls, 1) + + sourceHandler.onResponseData(controller, compressed) + assert.equal(chunks.length, 0) + + responseController.resume() + assert.equal(state.paused, true) + assert.equal(state.resumeCalls, 0) + + while (state.resumeCalls === 0) { + await immediate() + } + sourceHandler.onResponseEnd(controller, {}) + await terminal + assert.equal(state.pauseCalls, 1) + assert.equal(state.resumeCalls, 1) + assert.deepEqual(Buffer.concat(chunks), payload) +}) + +test('decompress backpressure pauses chained output through encoded end', { timeout: 5000 }, async () => { + const payload = Buffer.alloc(512 * 1024, 0x63) + const compressed = deflateSync(gzipSync(payload)) + const chunks = [] + let responseController + let endCalls = 0 + let errorCalls = 0 + let firstDataResolve + const firstData = new Promise(resolve => { firstDataResolve = resolve }) + let endResolve + const ended = new Promise(resolve => { endResolve = resolve }) + const handler = { + onRequestStart (controller) { + responseController = controller + }, + onResponseStart () {}, + onResponseData (controller, chunk) { + chunks.push(chunk) + if (chunks.length === 1) { + controller.pause() + firstDataResolve() + } + }, + onResponseEnd () { + endCalls++ + endResolve() + }, + onResponseError () { + errorCalls++ + endResolve() + } + } + + const { controller, sourceHandler } = createControlledDispatch(handler) + sourceHandler.onResponseStart(controller, 200, { 'content-encoding': 'gzip, deflate' }, 'OK') + sourceHandler.onResponseData(controller, compressed) + sourceHandler.onResponseEnd(controller, { chained: 'trailer' }) + + await firstData + await immediate() + assert.equal(chunks.length, 1) + assert.equal(endCalls, 0) + assert.equal(errorCalls, 0) + + responseController.resume() + await ended + assert.deepEqual(Buffer.concat(chunks), payload) + assert(chunks.length > 1) + assert.equal(endCalls, 1) + assert.equal(errorCalls, 0) +}) + +test('decompress backpressure remains paused when retry replaces the transport controller', { timeout: 5000 }, async () => { + const payload = Buffer.alloc(512 * 1024, 0x65) + const compressed = gzipSync(payload) + const split = compressed.length - 8 + assert(split > 0) + assert(compressed.length < 16 * 1024) + + const attempts = [] + const baseDispatch = (opts, sourceHandler) => { + const state = { + paused: false, + pauseCalls: 0, + resumeCalls: 0 + } + let abortReason = null + const controller = { + rawHeaders: null, + rawTrailers: null, + pause () { + state.paused = true + state.pauseCalls++ + }, + resume () { + state.paused = false + state.resumeCalls++ + }, + abort (reason) { + abortReason = reason + }, + get paused () { return state.paused }, + get aborted () { return abortReason !== null }, + get reason () { return abortReason } + } + + attempts.push({ controller, opts, sourceHandler, state }) + sourceHandler.onRequestStart(controller, {}) + return true + } + + const retryDispatch = interceptors.retry({ + maxRetries: 1, + retry (_ignoredError, _context, callback) { + callback(null) + } + })(baseDispatch) + const dispatch = createDecompressInterceptor()(retryDispatch) + + let responseController + const chunks = [] + let firstDataResolve + const firstData = new Promise(resolve => { firstDataResolve = resolve }) + let terminalResolve + let terminalReject + const terminal = new Promise((resolve, reject) => { + terminalResolve = resolve + terminalReject = reject + }) + const handler = { + onRequestStart (controller) { + responseController = controller + }, + onResponseStart () {}, + onResponseData (controller, chunk) { + chunks.push(chunk) + if (chunks.length === 1) { + controller.pause() + firstDataResolve() + } + }, + onResponseEnd () { + terminalResolve() + }, + onResponseError (controller, error) { + terminalReject(error) + } + } + + dispatch({ method: 'GET', path: '/' }, handler) + assert.equal(attempts.length, 1) + const first = attempts[0] + first.controller.rawHeaders = ['Content-Encoding', 'gzip', 'Content-Length', String(compressed.length)] + first.sourceHandler.onResponseStart(first.controller, 200, { + 'content-encoding': 'gzip', + 'content-length': String(compressed.length) + }, 'OK') + first.sourceHandler.onResponseData(first.controller, compressed.subarray(0, split)) + + await firstData + assert.equal(first.state.paused, true) + assert.equal(first.state.pauseCalls, 1) + + const connectionError = new Error('connection reset during encoded response') + connectionError.code = 'ECONNRESET' + first.sourceHandler.onResponseError(first.controller, connectionError) + + assert.equal(attempts.length, 2) + const second = attempts[1] + assert.equal(second.opts.headers.range, `bytes=${split}-${compressed.length - 1}`) + assert.equal(second.state.paused, true) + assert.equal(second.state.pauseCalls, 1) + + second.controller.rawHeaders = [ + 'Content-Encoding', 'gzip', + 'Content-Range', `bytes ${split}-${compressed.length - 1}/${compressed.length}`, + 'Content-Length', String(compressed.length - split) + ] + second.sourceHandler.onResponseStart(second.controller, 206, { + 'content-encoding': 'gzip', + 'content-range': `bytes ${split}-${compressed.length - 1}/${compressed.length}`, + 'content-length': String(compressed.length - split) + }, 'Partial Content') + + responseController.resume() + assert.equal(second.state.paused, false) + assert.equal(second.state.resumeCalls, 1) + + second.sourceHandler.onResponseData(second.controller, compressed.subarray(split)) + second.sourceHandler.onResponseEnd(second.controller, { retried: 'trailer' }) + await terminal + assert.deepEqual(Buffer.concat(chunks), payload) +}) + +test('decompress abort while decoded output is paused errors exactly once', { timeout: 5000 }, async () => { + const payload = Buffer.alloc(512 * 1024, 0x64) + const compressed = gzipSync(payload) + const abortReason = new Error('abort paused decompression') + let responseController + let dataCalls = 0 + let endCalls = 0 + const errors = [] + const errorControllers = [] + let firstDataResolve + const firstData = new Promise(resolve => { firstDataResolve = resolve }) + let errorResolve + const errored = new Promise(resolve => { errorResolve = resolve }) + const handler = { + onRequestStart (controller) { + responseController = controller + }, + onResponseStart () {}, + onResponseData (controller) { + dataCalls++ + if (dataCalls === 1) { + controller.pause() + firstDataResolve() + } + }, + onResponseEnd () { + endCalls++ + }, + onResponseError (controller, error) { + errorControllers.push(controller) + errors.push(error) + errorResolve() + } + } + + const { controller, sourceHandler } = createControlledDispatch(handler, undefined, { forwardAbort: false }) + sourceHandler.onResponseStart(controller, 200, { 'content-encoding': 'gzip' }, 'OK') + sourceHandler.onResponseData(controller, compressed) + sourceHandler.onResponseEnd(controller, {}) + + await firstData + await immediate() + assert.equal(dataCalls, 1) + assert.equal(endCalls, 0) + + responseController.abort(abortReason) + await errored + await immediate() + assert.equal(dataCalls, 1) + assert.equal(endCalls, 0) + assert.deepEqual(errors, [abortReason]) + assert.deepEqual(errorControllers, [responseController]) + assert.equal(responseController.aborted, true) + assert.strictEqual(responseController.reason, abortReason) +}) + test('should handle multiple pause/resume cycles during decompression', async t => { t = tspl(t, { plan: 3 }) @@ -886,6 +1515,47 @@ test('should handle controller pause with chained decompression', async t => { await t.completed }) +test('should disable the decompressed size limit by default', async t => { + t = tspl(t, { plan: 1 }) + + const decompressedSize = 64 * 1024 * 1024 + 1 + const compressed = gzipSync(Buffer.alloc(decompressedSize, 0x61)) + const server = createServer({ joinDuplicateHeaders: true }, (_req, res) => { + res.writeHead(200, { + 'Content-Encoding': 'gzip', + 'Content-Length': compressed.length + }) + res.end(compressed) + }) + + server.listen(0) + await once(server, 'listening') + + const client = new Client( + `http://localhost:${server.address().port}` + ).compose(interceptors.decompress()) + + after(async () => { + await client.close() + server.close() + await once(server, 'close') + }) + + const response = await client.request({ + method: 'GET', + path: '/' + }) + + let received = 0 + for await (const chunk of response.body) { + received += chunk.length + } + + t.equal(received, decompressedSize) + + await t.completed +}) + test('should reject a response that exceeds maxSize after decompression', async t => { t = tspl(t, { plan: 1 }) @@ -1004,6 +1674,45 @@ test('should enforce maxSize on the final output of a decompression chain', asyn await t.completed }) +test('should apply maxSize independently to every decompression stage', async t => { + t = tspl(t, { plan: 2 }) + + const data = Buffer.from(Array.from({ length: 1024 }, (_, index) => index % 251)) + const intermediate = gzipSync(data) + const compressed = deflateSync(intermediate) + const maxSize = Math.max(data.length, intermediate.length) + t.ok(data.length + intermediate.length > maxSize) + + const server = createServer({ joinDuplicateHeaders: true }, (_req, res) => { + res.writeHead(200, { + 'Content-Encoding': 'gzip, deflate' + }) + res.end(compressed) + }) + + server.listen(0) + await once(server, 'listening') + + const client = new Client( + `http://localhost:${server.address().port}` + ).compose(interceptors.decompress({ maxSize })) + + after(async () => { + await client.close() + server.close() + await once(server, 'close') + }) + + const response = await client.request({ + method: 'GET', + path: '/' + }) + + t.deepStrictEqual(Buffer.from(await response.body.arrayBuffer()), data) + + await t.completed +}) + test('should work when composed after the retry interceptor', async t => { t = tspl(t, { plan: 1 }) @@ -1081,12 +1790,15 @@ test('should allow a decompressed response exactly equal to maxSize', async t => test('should reject invalid maxSize values', async t => { t = tspl(t, { plan: 5 }) - for (const maxSize of [0, -1, 1.5, Infinity, '1024']) { + const unlimitedDispatch = createDecompressInterceptor({ maxSize: 0 })(() => true) + t.doesNotThrow(() => unlimitedDispatch({ method: 'GET' }, {})) + + for (const maxSize of [-1, 1.5, Infinity, '1024']) { const dispatch = createDecompressInterceptor({ maxSize })(() => true) t.throws(() => dispatch({ method: 'GET' }, {}), { name: 'InvalidArgumentError', code: 'UND_ERR_INVALID_ARG', - message: 'maxSize must be a positive integer' + message: 'maxSize must be a non-negative integer' }) } diff --git a/test/retry-handler-controller-proxy.js b/test/retry-handler-controller-proxy.js new file mode 100644 index 00000000000..3e7dc3bf46d --- /dev/null +++ b/test/retry-handler-controller-proxy.js @@ -0,0 +1,195 @@ +'use strict' + +const { tspl } = require('@matteo.collina/tspl') +const { test } = require('node:test') + +const { RetryHandler } = require('..') + +// These tests pin down the RetryController proxy contract introduced to keep +// flow-control wired to the active connection across transparent retries/resumes. +// Each retry/resume is a separate dispatch with its own connection controller; +// the downstream handler is handed ONE stable proxy that always forwards to the +// controller of the currently active connection. See lib/handler/retry-handler.js. + +const baseOpts = { + method: 'GET', + path: '/', + retryOptions: {} +} + +// Stand-in for the per-dispatch RequestController of an active connection. It +// records the flow-control calls the proxy forwards to it. +function activeConnectionController () { + const calls = [] + return { + calls, + paused: true, + aborted: true, + reason: new Error('boom'), + rawHeaders: ['content-length', '2'], + rawTrailers: ['x-trailer', 'value'], + pause () { calls.push('pause') }, + resume () { calls.push('resume') }, + abort (reason) { calls.push(['abort', reason]) } + } +} + +test('controller proxy returns safe defaults and is a no-op before a connection is active', (t) => { + t = tspl(t, { plan: 6 }) + + const handler = new RetryHandler(baseOpts, { + dispatch: () => {}, + handler: {} + }) + + // No dispatch has happened yet, so the proxy has no active connection to + // forward to. Reads must fall back to safe defaults instead of throwing. + const proxy = handler.controllerProxy + t.strictEqual(proxy.paused, false) + t.strictEqual(proxy.aborted, false) + t.strictEqual(proxy.reason, null) + t.strictEqual(proxy.rawHeaders, null) + t.strictEqual(proxy.rawTrailers, null) + + // Methods must be inert (not throw) while there is nothing to forward to. + t.doesNotThrow(() => { + proxy.pause() + proxy.resume() + proxy.abort(new Error('ignored')) + proxy.rawHeaders = ['x', 'y'] + proxy.rawTrailers = ['z', '1'] + }) +}) + +test('controller proxy reapplies a persistent pause to a replacement connection', (t) => { + t = tspl(t, { plan: 6 }) + + const createController = () => { + const calls = [] + let paused = false + return { + calls, + pause () { + paused = true + calls.push('pause') + }, + resume () { + paused = false + calls.push('resume') + }, + abort () {}, + get paused () { return paused }, + get aborted () { return false }, + get reason () { return null } + } + } + + const handler = new RetryHandler(baseOpts, { + dispatch: () => {}, + handler: {} + }) + const first = createController() + handler.onRequestStart(first, {}) + const proxy = handler.controllerProxy + + proxy.pause() + t.strictEqual(proxy.paused, true) + t.deepStrictEqual(first.calls, ['pause']) + + handler.headersSent = true + const second = createController() + handler.onRequestStart(second, {}) + t.strictEqual(proxy.paused, true) + t.deepStrictEqual(second.calls, ['pause']) + + proxy.resume() + t.strictEqual(proxy.paused, false) + t.deepStrictEqual(second.calls, ['pause', 'resume']) +}) + +test('controller proxy forwards reads/writes to the active connection and stays stable across callbacks', (t) => { + t = tspl(t, { plan: 11 }) + + let downstreamController = null + let upgradeController = null + const upgradeArgs = [] + + const handler = new RetryHandler(baseOpts, { + dispatch: () => {}, + handler: { + onRequestStart (controller) { + downstreamController = controller + }, + onRequestUpgrade (controller, statusCode, headers, socket) { + upgradeController = controller + upgradeArgs.push(statusCode, headers, socket) + } + } + }) + + const connection = activeConnectionController() + + // onRequestStart is the first callback of a dispatch; it re-points the proxy + // at this connection's controller and hands the (stable) proxy downstream. + handler.onRequestStart(connection, {}) + + // The downstream handler must receive the proxy, never the raw per-connection + // controller, so flow-control survives the next resume. + t.notStrictEqual(downstreamController, connection) + + // Reads forward to the active connection's controller. + t.deepStrictEqual(downstreamController.rawHeaders, ['content-length', '2']) + t.deepStrictEqual(downstreamController.rawTrailers, ['x-trailer', 'value']) + t.strictEqual(downstreamController.paused, true) + t.strictEqual(downstreamController.aborted, true) + t.strictEqual(downstreamController.reason, connection.reason) + + // Writes forward to the active connection's controller too. + downstreamController.pause() + downstreamController.resume() + downstreamController.abort('stop') + t.deepStrictEqual(connection.calls, ['pause', 'resume', ['abort', 'stop']]) + + // Decompress (and other interceptors) rewrite rawHeaders/rawTrailers on the + // controller they were given. Those assignments must reach the active + // connection instead of throwing on a getter-only proxy. + downstreamController.rawHeaders = ['x-foo', 'bar'] + downstreamController.rawTrailers = ['x-end', '1'] + t.deepStrictEqual(connection.rawHeaders, ['x-foo', 'bar']) + t.deepStrictEqual(connection.rawTrailers, ['x-end', '1']) + + // An upgrade on the same dispatch is forwarded through the very same proxy + // instance (not the raw controller), keeping the downstream wiring stable. + handler.onRequestUpgrade(connection, 101, { upgrade: 'websocket' }, 'SOCKET') + t.strictEqual(upgradeController, downstreamController) + t.deepStrictEqual(upgradeArgs, [101, { upgrade: 'websocket' }, 'SOCKET']) +}) + +test('controller proxy carries a synchronous dispatch failure to the downstream handler', (t) => { + t = tspl(t, { plan: 2 }) + + const dispatchError = new Error('dispatch failed synchronously') + let errController = null + let receivedErr = null + + const handler = new RetryHandler(baseOpts, { + dispatch: () => { throw dispatchError }, + handler: { + onRequestStart () {}, + onResponseError (controller, err) { + errController = controller + receivedErr = err + } + } + }) + + const connection = activeConnectionController() + handler.onRequestStart(connection, {}) + + // retry() re-dispatches; when that dispatch throws synchronously the error is + // surfaced to the downstream handler through the proxy. + handler.retry() + + t.strictEqual(errController, handler.controllerProxy) + t.strictEqual(receivedErr, dispatchError) +}) diff --git a/types/interceptors.d.ts b/types/interceptors.d.ts index c534575b2a9..237c539ce1c 100644 --- a/types/interceptors.d.ts +++ b/types/interceptors.d.ts @@ -12,7 +12,7 @@ declare namespace Interceptors { export type DecompressInterceptorOpts = { skipErrorResponses?: boolean skipStatusCodes?: number[] - /** Maximum decompressed response size in bytes. @default 67108864 */ + /** Maximum decompressed response size in bytes per stage. 0 disables the limit. @default 0 */ maxSize?: number }