diff --git a/CLONE-HIGHWATERMARK.md b/CLONE-HIGHWATERMARK.md new file mode 100644 index 000000000..d3153390a --- /dev/null +++ b/CLONE-HIGHWATERMARK.md @@ -0,0 +1,109 @@ +# Increasing clone highWaterMark + +## The Problem + +When using `res.clone` method it might happen that you want process either the original body or the cloned body first: + +```js +fetch(url) + .then(res => cache.put(url, res.clone())) + .then(res => res.json()) + ... +``` + +The original response waits for the cloned response to be completed. That means the whole response is buffered in a memory during the first `than` statement. With big response sizes that might lead to consuming too much of the precious server resources. + +To keep app allocated memory low, Node.js provides [`highWaterMark` limit of stream internal buffer][hwm]. It defaults to 16kB for all streams but can be overridden explicitly. + +The problem is that the code above freezes and times out for larger response sizes. Nobody consumes the original response stream leading to accumulation of data in it's buffers. When `highWaterMark` limits are hit a mechanism called [backpressure] kicks in. As a result, data will stop flowing, endlessly waiting for the original response to signal it can take more. + +1. At the beginning 6 packets are ready to be transmitted. + + ``` + Data Original + +-------------+ +-----------+ + | O O O O O O +-----+------>+ | X + +-------------+ | +-----------+ + | + | Cloned + | +-----------+ + +------>+ +----> + +-----------+ + ``` + +2. 2 chunks passed to both streams. + + ``` + Data Original + +-------------+ +-----------+ + | O O O O +-----+------>+ O O | X + +-------------+ | +-----------+ + | + | Cloned + | +-----------+ + +------>+ O O +----> + +-----------+ + ``` + +3. 5 chunks passed to both streams. The original one triggers backpressure. Source of data stops until notification that it can send more. + + ``` + Data Original + +-------------+ +-----------+ + | O +-----+------>+ O O O O O | X + +-------------+ | +-----------+ + | + | Cloned + | +-----------+ + +------>+ O O O O O +----> + +-----------+ + ``` + +4. Chunks in the cloned stream reaches their destination. But the flow stopped. The last chunk won't be transmitted. + + ``` + Data Original + +-------------+ +-----------+ + | O +-----+------>+ O O O O O | X + +-------------+ | +-----------+ + | + | Cloned + | +-----------+ + +------>+ +----> + +-----------+ + ``` + +There are few inaccuracies in diagrams above for the sake of simplification. + +[hwm]: https://nodejs.org/api/stream.html#stream_buffering +[backpressure]: https://nodejs.org/en/docs/guides/backpressuring-in-streams/ + +## The Solution + +Set bigger `highWaterMark` limit by passing a value to `clone` method: + +```js +res.clone(40 * 1024) +``` + +Use `expected_maximal_request_size / 2 + 1` as the value. + +Don't forget that the whole response still goes into memory. Calculate carefully not to deplete all your server memory with few requests. + + +### Why + +The cloned body is in fact a Node.js [*PassThrough*][passthrough] stream. *PassThrough* streams have two buffers with the same `highWaterMark`, one for [*Writable*][writable] stream on the input and on for [*Readable*][readable] stream on output. It can contain **double the value of `highWaterMark`**. + +When `highWateMark` of the *Writeable* stream is reached, the stream writing data stops. Increasing the value by **a single byte** is sufficient to avoid that. + +In fact, streams can take much more data. `highWaterMark` is [not a limit][] really. It is rather just a mark as the name suggests. Backpressure kicks in when data written [reaches **or overflows**][highwatermark-check] `highWaterMark` value. With two buffers of *PassThrough*, the **first chunk can have any size**. Well, almost any size. A TCP packet maximum size is [64kB], so this is the most common chunk size when dealing with large HTTP responses. + +But to avoid the backpressure, when chunks fill in the second buffer without any overflow by chance, we need to make sure the first buffer won't get to the `highWaterMark`. Hence the ½ + 1 limit. + +[passthrough]: https://nodejs.org/api/stream.html#stream_class_stream_passthrough +[writable]: https://nodejs.org/api/stream.html#stream_writable_streams +[readable]: https://nodejs.org/api/stream.html#stream_readable_streams +[not a limit]: https://stackoverflow.com/a/45905930/5763764 +[highwatermark-check]: https://github.com/nodejs/node/blob/master/lib/_stream_writable.js#L378 +[64kB]: https://stackoverflow.com/a/2614188/5763764 diff --git a/LIMITS.md b/LIMITS.md index 9c4b8c0c8..9345f775e 100644 --- a/LIMITS.md +++ b/LIMITS.md @@ -12,7 +12,7 @@ Known differences - `res.url` contains the final url when following redirects. -- For convenience, `res.body` is a Node.js [Readable stream][readable-stream], so decoding can be handled independently. +- For convenience, `res.body` is a Node.js [Readable stream][], so decoding can be handled independently. - Similarly, `req.body` can either be `null`, a string, a buffer or a Readable stream. @@ -24,9 +24,12 @@ Known differences - Current implementation lacks server-side cookie store, you will need to extract `Set-Cookie` headers manually. -- If you are using `res.clone()` and writing an isomorphic app, note that stream on Node.js have a smaller internal buffer size (16Kb, aka `highWaterMark`) from client-side browsers (>1Mb, not consistent across browsers). +- If you are using `res.clone()` and writing an isomorphic app, note that stream on Node.js has a smaller default internal buffer size (16kB, aka [`highWaterMark`][]) from client-side browsers (>1MB, not consistent across browsers). You can override the default value by passing a custom `highWaterMark` value to `clone` method. This parameter is taken into account only by `node-fetch`. See [CLONE-HIGHWATERMARK.md][] for more details. -- Because node.js stream doesn't expose a [*disturbed*](https://fetch.spec.whatwg.org/#concept-readablestream-disturbed) property like Stream spec, using a consumed stream for `new Response(body)` will not set `bodyUsed` flag correctly. +- Because node.js stream doesn't expose a [*disturbed*][] property like Stream spec, using a consumed stream for `new Response(body)` will not set `bodyUsed` flag correctly. -[readable-stream]: https://nodejs.org/api/stream.html#stream_readable_streams +[Readable stream]: https://nodejs.org/api/stream.html#stream_readable_streams [ERROR-HANDLING.md]: https://github.com/bitinn/node-fetch/blob/master/ERROR-HANDLING.md +[`highWaterMark`]: https://nodejs.org/api/stream.html#stream_buffering +[CLONE-HIGHWATERMARK.md]: https://github.com/bitinn/node-fetch/blob/master/CLONE-HIGHWATERMARK.md +[*disturbed*]: https://fetch.spec.whatwg.org/#concept-readablestream-disturbed diff --git a/src/body.js b/src/body.js index 90cbcabfa..17742d132 100644 --- a/src/body.js +++ b/src/body.js @@ -358,10 +358,11 @@ function isURLSearchParams(obj) { /** * Clone body given Res/Req instance * - * @param Mixed instance Response or Request instance + * @param Mixed instance Response or Request instance + * @param String highWaterMark highWaterMark for both PassThrough body streams * @return Mixed */ -export function clone(instance) { +export function clone(instance, highWaterMark) { let p1, p2; let body = instance.body; @@ -374,8 +375,8 @@ export function clone(instance) { // note: we can't clone the form-data object without having it as a dependency if ((body instanceof Stream) && (typeof body.getBoundary !== 'function')) { // tee instance body - p1 = new PassThrough(); - p2 = new PassThrough(); + p1 = new PassThrough({ highWaterMark }); + p2 = new PassThrough({ highWaterMark }); body.pipe(p1); body.pipe(p2); // set instance body to teed body and return the other teed body diff --git a/src/response.js b/src/response.js index f29bfe296..03789908f 100644 --- a/src/response.js +++ b/src/response.js @@ -70,10 +70,11 @@ export default class Response { /** * Clone this response * + * @param String highWaterMark highWaterMark for both PassThrough body streams * @return Response */ - clone() { - return new Response(clone(this), { + clone(highWaterMark) { + return new Response(clone(this, highWaterMark), { url: this.url, status: this.status, statusText: this.statusText, diff --git a/test/chai-timeout.js b/test/chai-timeout.js new file mode 100644 index 000000000..cd02bcd17 --- /dev/null +++ b/test/chai-timeout.js @@ -0,0 +1,17 @@ +module.exports = (chai, utils) => { + utils.addProperty(chai.Assertion.prototype, 'timeout', function () { + return new Promise(resolve => { + const timer = setTimeout(() => resolve(true), 150); + this._obj.then(() => { + clearTimeout(timer); + resolve(false); + }); + }).then(timeouted => { + this.assert( + timeouted, + 'expected promise to timeout but it was resolved', + 'expected promise not to timeout but it timed out' + ); + }) + }); +}; diff --git a/test/server.js b/test/server.js index 4028f0cc4..7a0efcc5c 100644 --- a/test/server.js +++ b/test/server.js @@ -31,9 +31,23 @@ export default class TestServer { this.server.close(cb); } + mockResponse(responseHandler) { + this.server.nextResponseHandler = responseHandler; + return `http://${this.hostname}:${this.port}/mocked` + } + router(req, res) { let p = parse(req.url).pathname; + if (p === '/mocked') { + if (this.nextResponseHandler) { + this.nextResponseHandler(res); + this.nextResponseHandler = undefined; + } else { + throw new Error('No mocked response. Use \'TestServer.mockResponse()\'.'); + } + } + if (p === '/hello') { res.statusCode = 200; res.setHeader('Content-Type', 'text/plain'); diff --git a/test/test.js b/test/test.js index 78301a4b7..96306bbdf 100644 --- a/test/test.js +++ b/test/test.js @@ -12,6 +12,7 @@ import URLSearchParams_Polyfill from 'url-search-params'; import { URL } from 'whatwg-url'; import { AbortController } from 'abortcontroller-polyfill/dist/abortcontroller'; import AbortController2 from 'abort-controller'; +import crypto from 'crypto'; const { spawn } = require('child_process'); const http = require('http'); @@ -30,9 +31,12 @@ const { let convert; try { convert = require('encoding').convert; } catch(e) { } +import chaiTimeout from './chai-timeout'; + chai.use(chaiPromised); chai.use(chaiIterator); chai.use(chaiString); +chai.use(chaiTimeout); const expect = chai.expect; import TestServer from './server'; @@ -1698,6 +1702,69 @@ describe('node-fetch', () => { ); }); + it('should timeout on cloning response without consuming one of the streams when the second packet size is equal default highWaterMark', function () { + this.timeout(300); + const url = local.mockResponse(res => { + // Observed behavior of TCP packets splitting: + // - response body size <= 65438 → single packet sent + // - response body size > 65438 → multiple packets sent + // Max TCP packet size is 64kB (https://stackoverflow.com/a/2614188/5763764), + // but first packet probably transfers more than the response body. + const firstPacketMaxSize = 65438; + const secondPacketSize = 16 * 1024; // = defaultHighWaterMark + res.end(crypto.randomBytes(firstPacketMaxSize + secondPacketSize)); + }); + return expect( + fetch(url).then(res => res.clone().buffer()) + ).to.timeout; + }); + + it('should timeout on cloning response without consuming one of the streams when the second packet size is equal custom highWaterMark', function () { + this.timeout(300); + const url = local.mockResponse(res => { + const firstPacketMaxSize = 65438; + const secondPacketSize = 10; + res.end(crypto.randomBytes(firstPacketMaxSize + secondPacketSize)); + }); + return expect( + fetch(url).then(res => res.clone(10).buffer()) + ).to.timeout; + }); + + it('should not timeout on cloning response without consuming one of the streams when the second packet size is less than default highWaterMark', function () { + this.timeout(300); + const url = local.mockResponse(res => { + const firstPacketMaxSize = 65438; + const secondPacketSize = 16 * 1024; // = defaultHighWaterMark + res.end(crypto.randomBytes(firstPacketMaxSize + secondPacketSize - 1)); + }); + return expect( + fetch(url).then(res => res.clone().buffer()) + ).not.to.timeout; + }); + + it('should not timeout on cloning response without consuming one of the streams when the second packet size is less than custom highWaterMark', function () { + this.timeout(300); + const url = local.mockResponse(res => { + const firstPacketMaxSize = 65438; + const secondPacketSize = 10; + res.end(crypto.randomBytes(firstPacketMaxSize + secondPacketSize - 1)); + }); + return expect( + fetch(url).then(res => res.clone(10).buffer()) + ).not.to.timeout; + }); + + it('should not timeout on cloning response without consuming one of the streams when the response size is double the custom large highWaterMark - 1', function () { + this.timeout(300); + const url = local.mockResponse(res => { + res.end(crypto.randomBytes(2 * 512 * 1024 - 1)); + }); + return expect( + fetch(url).then(res => res.clone(512 * 1024).buffer()) + ).not.to.timeout; + }); + it('should allow get all responses of a header', function() { const url = `${base}cookie`; return fetch(url).then(res => {