From 53d771d3a2b3329b17bf8df9fd58ff77c67e1c62 Mon Sep 17 00:00:00 2001 From: Martin Slota Date: Tue, 13 Aug 2024 21:53:53 +0200 Subject: [PATCH 01/14] Add tests demonstrating scenarios under which the original res.end does not get called --- test/compression.js | 121 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 121 insertions(+) diff --git a/test/compression.js b/test/compression.js index 449c6198..15fe46b9 100644 --- a/test/compression.js +++ b/test/compression.js @@ -3,6 +3,7 @@ var assert = require('assert') var bytes = require('bytes') var crypto = require('crypto') var http = require('http') +var net = require('net') var request = require('supertest') var zlib = require('zlib') var http2 = require('http2') @@ -953,6 +954,113 @@ describe('compression()', function () { .expect(200, done) }) }) + + describe('when the client closes the connection before consuming the response', function () { + it('should call the original res.end() if connection is cut early on', function (done) { + var server = http.createServer(function (req, res) { + var originalResEnd = res.end + var originalResEndCalledTimes = 0 + res.end = function () { + originalResEndCalledTimes++ + return originalResEnd.apply(this, arguments) + } + + compression({ threshold: 0 })(req, res, function () { + socket.end() + + res.setHeader('Content-Type', 'text/plain') + res.write('hello, ') + setTimeout(function () { + res.end('world!') + + setTimeout(function () { + server.close(function () { + if (originalResEndCalledTimes === 1) { + done() + } else { + done(new Error('The original res.end() was called ' + originalResEndCalledTimes + ' times')) + } + }) + }, 5) + }, 5) + }) + }) + + server.listen() + + var port = server.address().port + var socket = openSocketWithRequest(port) + }) + + it('should call the original res.end() if connection is cut after an initial write', function (done) { + var server = http.createServer(function (req, res) { + var originalResEnd = res.end + var originalResEndCalledTimes = 0 + res.end = function () { + originalResEndCalledTimes++ + return originalResEnd.apply(this, arguments) + } + + compression({ threshold: 0 })(req, res, function () { + res.setHeader('Content-Type', 'text/plain') + res.write('hello, ') + socket.end() + + setTimeout(function () { + res.end('world!') + + setTimeout(function () { + server.close(function () { + if (originalResEndCalledTimes === 1) { + done() + } else { + done(new Error('The original res.end() was called ' + originalResEndCalledTimes + ' times')) + } + }) + }, 5) + }, 5) + }) + }) + + server.listen() + + var port = server.address().port + var socket = openSocketWithRequest(port) + }) + + it('should call the original res.end() if connection is cut just after response body was generated', function (done) { + var server = http.createServer(function (req, res) { + var originalResEnd = res.end + var originalResEndCalledTimes = 0 + res.end = function () { + originalResEndCalledTimes++ + return originalResEnd.apply(this, arguments) + } + + compression({ threshold: 0 })(req, res, function () { + res.setHeader('Content-Type', 'text/plain') + res.write('hello, ') + res.end('world!') + socket.end() + + setTimeout(function () { + server.close(function () { + if (originalResEndCalledTimes === 1) { + done() + } else { + done(new Error('The original res.end() was called ' + originalResEndCalledTimes + ' times')) + } + }) + }, 5) + }) + }) + + server.listen() + + var port = server.address().port + var socket = openSocketWithRequest(port) + }) + }) }) function createServer (opts, fn) { @@ -1056,3 +1164,16 @@ function unchunk (encoding, onchunk, onend) { stream.on('end', onend) } } + +function openSocketWithRequest (port) { + var socket = net.connect(port, function onConnect () { + socket.write('GET / HTTP/1.1\r\n') + socket.write('Accept-Encoding: gzip\r\n') + socket.write('Host: localhost:' + port + '\r\n') + socket.write('Content-Type: text/plain\r\n') + socket.write('Content-Length: 0\r\n') + socket.write('Connection: keep-alive\r\n') + socket.write('\r\n') + }) + return socket +} From a9bb7b24c5301937bfe014a56a2c6c31f2073f73 Mon Sep 17 00:00:00 2001 From: Martin Slota Date: Tue, 13 Aug 2024 14:25:43 +0200 Subject: [PATCH 02/14] Ensure stream gets closed and response gets ended when it is finished --- index.js | 11 +++++++++++ package.json | 1 + 2 files changed, 12 insertions(+) diff --git a/index.js b/index.js index bd6d5ea9..c445834d 100644 --- a/index.js +++ b/index.js @@ -18,6 +18,7 @@ var Negotiator = require('negotiator') var bytes = require('bytes') var compressible = require('compressible') var debug = require('debug')('compression') +var onFinished = require('on-finished') var onHeaders = require('on-headers') var vary = require('vary') var zlib = require('zlib') @@ -120,6 +121,10 @@ function compression (options) { // mark ended ended = true + if (onFinished.isFinished(this)) { + return _end.call(this) + } + // write Buffer for Node.js 0.8 return chunk ? stream.end(toBuffer(chunk, encoding)) @@ -227,6 +232,12 @@ function compression (options) { _on.call(res, 'drain', function onResponseDrain () { stream.resume() }) + + onFinished(res, function onFinished () { + if (ended) { + _end.call(res) + } + }) }) next() diff --git a/package.json b/package.json index f14e3418..4f15f7d0 100644 --- a/package.json +++ b/package.json @@ -14,6 +14,7 @@ "compressible": "~2.0.18", "debug": "2.6.9", "negotiator": "~0.6.4", + "on-finished": "~2.3.0", "on-headers": "~1.0.2", "vary": "~1.1.2" }, From 176847dd3b3426d71c542ab14708ccd99d65b792 Mon Sep 17 00:00:00 2001 From: Martin Slota Date: Tue, 13 Aug 2024 14:26:31 +0200 Subject: [PATCH 03/14] Only call the original res.end() a single time --- index.js | 16 ++++++++++++---- 1 file changed, 12 insertions(+), 4 deletions(-) diff --git a/index.js b/index.js index c445834d..095e1977 100644 --- a/index.js +++ b/index.js @@ -77,6 +77,14 @@ function compression (options) { var _on = res.on var _write = res.write + var endCalled = false + function endOnce () { + if (!endCalled) { + endCalled = true + _end.apply(this, arguments) + } + } + // flush res.flush = function flush () { if (stream) { @@ -115,14 +123,14 @@ function compression (options) { } if (!stream) { - return _end.call(this, chunk, encoding) + return endOnce.call(this, chunk, encoding) } // mark ended ended = true if (onFinished.isFinished(this)) { - return _end.call(this) + return endOnce.call(this) } // write Buffer for Node.js 0.8 @@ -226,7 +234,7 @@ function compression (options) { }) stream.on('end', function onStreamEnd () { - _end.call(res) + endOnce.call(res) }) _on.call(res, 'drain', function onResponseDrain () { @@ -235,7 +243,7 @@ function compression (options) { onFinished(res, function onFinished () { if (ended) { - _end.call(res) + endOnce.call(res) } }) }) From 0ea9e823c75a9e94a7d82abe82866d738c8c1be1 Mon Sep 17 00:00:00 2001 From: Martin Slota Date: Mon, 7 Apr 2025 21:49:17 +0200 Subject: [PATCH 04/14] Avoid using the on-finished package --- index.js | 5 ++--- package.json | 1 - 2 files changed, 2 insertions(+), 4 deletions(-) diff --git a/index.js b/index.js index 095e1977..933ceeb0 100644 --- a/index.js +++ b/index.js @@ -18,7 +18,6 @@ var Negotiator = require('negotiator') var bytes = require('bytes') var compressible = require('compressible') var debug = require('debug')('compression') -var onFinished = require('on-finished') var onHeaders = require('on-headers') var vary = require('vary') var zlib = require('zlib') @@ -129,7 +128,7 @@ function compression (options) { // mark ended ended = true - if (onFinished.isFinished(this)) { + if (this.writableFinished || !this.socket.writable) { return endOnce.call(this) } @@ -241,7 +240,7 @@ function compression (options) { stream.resume() }) - onFinished(res, function onFinished () { + res.socket.on('close', function onSocketClose () { if (ended) { endOnce.call(res) } diff --git a/package.json b/package.json index 4f15f7d0..f14e3418 100644 --- a/package.json +++ b/package.json @@ -14,7 +14,6 @@ "compressible": "~2.0.18", "debug": "2.6.9", "negotiator": "~0.6.4", - "on-finished": "~2.3.0", "on-headers": "~1.0.2", "vary": "~1.1.2" }, From e9309d1220482e3967423a639aacc5f471ae19e6 Mon Sep 17 00:00:00 2001 From: Martin Slota Date: Tue, 8 Apr 2025 15:40:50 +0200 Subject: [PATCH 05/14] Use finished() helper from node:stream --- index.js | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/index.js b/index.js index 933ceeb0..a19565e9 100644 --- a/index.js +++ b/index.js @@ -14,6 +14,7 @@ * @private */ +var nodeStream = require('node:stream') var Negotiator = require('negotiator') var bytes = require('bytes') var compressible = require('compressible') @@ -240,7 +241,7 @@ function compression (options) { stream.resume() }) - res.socket.on('close', function onSocketClose () { + nodeStream.finished(res, function onResponseFinished () { if (ended) { endOnce.call(res) } From 4631d241140f3081268927ac8cc99e7c55d5e377 Mon Sep 17 00:00:00 2001 From: Martin Slota <46676886+martinslota@users.noreply.github.com> Date: Tue, 8 Apr 2025 16:06:22 +0200 Subject: [PATCH 06/14] Import finished() helper directly from node:stream Co-authored-by: Sebastian Beltran --- index.js | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/index.js b/index.js index a19565e9..0c0f0ac7 100644 --- a/index.js +++ b/index.js @@ -14,7 +14,7 @@ * @private */ -var nodeStream = require('node:stream') +var { finished } = require('node:stream') var Negotiator = require('negotiator') var bytes = require('bytes') var compressible = require('compressible') From 7ff0e90025dff30c9fdf5722411cc26405c41cdf Mon Sep 17 00:00:00 2001 From: Sebastian Beltran Date: Tue, 8 Apr 2025 09:08:19 -0500 Subject: [PATCH 07/14] Use finished() helper directly instead of nodeStream.finished() --- index.js | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/index.js b/index.js index 0c0f0ac7..a5036a27 100644 --- a/index.js +++ b/index.js @@ -241,7 +241,7 @@ function compression (options) { stream.resume() }) - nodeStream.finished(res, function onResponseFinished () { + finished(res, function onResponseFinished () { if (ended) { endOnce.call(res) } From 3ae0d9b6c691c70c8a1d6213452dce5a46b28719 Mon Sep 17 00:00:00 2001 From: Martin Slota Date: Tue, 15 Apr 2025 11:53:54 +0200 Subject: [PATCH 08/14] Adjust stream pause / resume logic instead of messing with res.end() calls directly --- index.js | 21 +++++++++++++-------- 1 file changed, 13 insertions(+), 8 deletions(-) diff --git a/index.js b/index.js index a5036a27..06510a8c 100644 --- a/index.js +++ b/index.js @@ -129,10 +129,6 @@ function compression (options) { // mark ended ended = true - if (this.writableFinished || !this.socket.writable) { - return endOnce.call(this) - } - // write Buffer for Node.js 0.8 return chunk ? stream.end(toBuffer(chunk, encoding)) @@ -228,7 +224,7 @@ function compression (options) { // compression stream.on('data', function onStreamData (chunk) { - if (_write.call(res, chunk) === false) { + if (_write.call(res, chunk) === false && !isFinished(res)) { stream.pause() } }) @@ -242,9 +238,7 @@ function compression (options) { }) finished(res, function onResponseFinished () { - if (ended) { - endOnce.call(res) - } + stream.resume() }) }) @@ -331,3 +325,14 @@ function headersSent (res) { ? Boolean(res._header) : res.headersSent } + +/** + * Determine if the response is finished. + * + * @param {object} res + * @returns {boolean} + * @private + */ +function isFinished (res) { + return res.writableFinished || !res.socket.writable +} From 57d735a01c7abb0ec5450eb5eee5b4de3a573d61 Mon Sep 17 00:00:00 2001 From: Martin Slota Date: Tue, 15 Apr 2025 11:55:22 +0200 Subject: [PATCH 09/14] Revert addition of endOnce() wrapper --- index.js | 12 ++---------- 1 file changed, 2 insertions(+), 10 deletions(-) diff --git a/index.js b/index.js index 06510a8c..f48fb6c0 100644 --- a/index.js +++ b/index.js @@ -77,14 +77,6 @@ function compression (options) { var _on = res.on var _write = res.write - var endCalled = false - function endOnce () { - if (!endCalled) { - endCalled = true - _end.apply(this, arguments) - } - } - // flush res.flush = function flush () { if (stream) { @@ -123,7 +115,7 @@ function compression (options) { } if (!stream) { - return endOnce.call(this, chunk, encoding) + return _end.call(this, chunk, encoding) } // mark ended @@ -230,7 +222,7 @@ function compression (options) { }) stream.on('end', function onStreamEnd () { - endOnce.call(res) + _end.call(res) }) _on.call(res, 'drain', function onResponseDrain () { From 73576d9d62996ead243e7fd7bcbf07245edca97c Mon Sep 17 00:00:00 2001 From: Martin Slota Date: Tue, 15 Apr 2025 21:40:21 +0200 Subject: [PATCH 10/14] Add unit test around one more alternate sequence of events that seems reasonable to test in addition to the other ones --- test/compression.js | 36 ++++++++++++++++++++++++++++++++++++ 1 file changed, 36 insertions(+) diff --git a/test/compression.js b/test/compression.js index 15fe46b9..d4c322a0 100644 --- a/test/compression.js +++ b/test/compression.js @@ -992,6 +992,42 @@ describe('compression()', function () { var socket = openSocketWithRequest(port) }) + it('should call the original res.end() if connection is cut right after setting headers', function (done) { + var server = http.createServer(function (req, res) { + var originalResEnd = res.end + var originalResEndCalledTimes = 0 + res.end = function () { + originalResEndCalledTimes++ + return originalResEnd.apply(this, arguments) + } + + compression({ threshold: 0 })(req, res, function () { + res.setHeader('Content-Type', 'text/plain') + socket.end() + + res.write('hello, ') + setTimeout(function () { + res.end('world!') + + setTimeout(function () { + server.close(function () { + if (originalResEndCalledTimes === 1) { + done() + } else { + done(new Error('The original res.end() was called ' + originalResEndCalledTimes + ' times')) + } + }) + }, 5) + }, 5) + }) + }) + + server.listen() + + var port = server.address().port + var socket = openSocketWithRequest(port) + }) + it('should call the original res.end() if connection is cut after an initial write', function (done) { var server = http.createServer(function (req, res) { var originalResEnd = res.end From fd1500bf96c2ee751329efe3b7e7c496863c9bc0 Mon Sep 17 00:00:00 2001 From: Martin Slota Date: Tue, 15 Apr 2025 21:41:24 +0200 Subject: [PATCH 11/14] Add test around avoiding 'write after end' errors --- test/compression.js | 49 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 49 insertions(+) diff --git a/test/compression.js b/test/compression.js index d4c322a0..c1406ebd 100644 --- a/test/compression.js +++ b/test/compression.js @@ -1096,6 +1096,55 @@ describe('compression()', function () { var port = server.address().port var socket = openSocketWithRequest(port) }) + + it('should not trigger write errors if connection is cut just after response body was generated', function (done) { + var requestCount = 0 + + var server = http.createServer(function (req, res) { + requestCount += 1 + + var originalWrite = res.write + var writeError = null + res.write = function (chunk, callback) { + return originalWrite.call(this, chunk, function (error) { + if (error?.code === 'ERR_STREAM_WRITE_AFTER_END') { + writeError = error + } + return callback?.(error) + }) + } + + var originalResEnd = res.end + res.end = function () { + setTimeout(function () { + if (writeError !== null) { + server.close(function () { + done(new Error(`Write error occurred: ${writeError}`)) + }) + } else { + if (requestCount < 50) { + socket = openSocketWithRequest(port) + } else { + server.close(done) + } + } + }, 0) + return originalResEnd.apply(this, arguments) + } + + compression({ threshold: 0 })(req, res, function () { + res.setHeader('Content-Type', 'text/plain') + res.write('hello, ') + res.end('world!') + socket.end() + }) + }) + + server.listen() + + var port = server.address().port + var socket = openSocketWithRequest(port) + }) }) }) From 175a50d4684692bcbee07c8700d6bc3985d2ac94 Mon Sep 17 00:00:00 2001 From: Martin Slota Date: Tue, 15 Apr 2025 21:42:46 +0200 Subject: [PATCH 12/14] Avoid writing into the response after it is finished --- index.js | 5 ++++- test/compression.js | 2 +- 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/index.js b/index.js index c2816e5a..82d84e98 100644 --- a/index.js +++ b/index.js @@ -216,7 +216,10 @@ function compression (options) { // compression stream.on('data', function onStreamData (chunk) { - if (_write.call(res, chunk) === false && !isFinished(res)) { + if (isFinished(res)) { + return + } + if (_write.call(res, chunk) === false) { stream.pause() } }) diff --git a/test/compression.js b/test/compression.js index c1406ebd..9ba4838d 100644 --- a/test/compression.js +++ b/test/compression.js @@ -1107,7 +1107,7 @@ describe('compression()', function () { var writeError = null res.write = function (chunk, callback) { return originalWrite.call(this, chunk, function (error) { - if (error?.code === 'ERR_STREAM_WRITE_AFTER_END') { + if (error) { writeError = error } return callback?.(error) From 11f8f24f7d78eeb7d157a361aaf71dff3c8321d7 Mon Sep 17 00:00:00 2001 From: Martin Slota Date: Wed, 16 Apr 2025 17:22:09 +0200 Subject: [PATCH 13/14] Add a comment explaining why the compression stream gets resumed when the response finishes --- index.js | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/index.js b/index.js index 82d84e98..a31c52ca 100644 --- a/index.js +++ b/index.js @@ -232,6 +232,11 @@ function compression (options) { stream.resume() }) + // In case the stream is paused when the response finishes (e.g. because + // the client cuts the connection), its `drain` event may not get emitted. + // The following handler is here to ensure that the stream gets resumed so + // it ends up emitting its `end` event and calling the original + // `res.end()`. finished(res, function onResponseFinished () { stream.resume() }) From 9cbf9926746e0ff036915a440be23287722a2b4d Mon Sep 17 00:00:00 2001 From: Sebastian Beltran Date: Wed, 16 Apr 2025 12:06:59 -0500 Subject: [PATCH 14/14] chore: use isFinished of on-finished --- index.js | 14 +++----------- package.json | 1 + 2 files changed, 4 insertions(+), 11 deletions(-) diff --git a/index.js b/index.js index a31c52ca..3a0c18e3 100644 --- a/index.js +++ b/index.js @@ -19,6 +19,7 @@ var Negotiator = require('negotiator') var bytes = require('bytes') var compressible = require('compressible') var debug = require('debug')('compression') +const isFinished = require('on-finished').isFinished var onHeaders = require('on-headers') var vary = require('vary') var zlib = require('zlib') @@ -217,9 +218,11 @@ function compression (options) { // compression stream.on('data', function onStreamData (chunk) { if (isFinished(res)) { + debug('response finished') return } if (_write.call(res, chunk) === false) { + debug('pausing compression stream') stream.pause() } }) @@ -311,14 +314,3 @@ function toBuffer (chunk, encoding) { ? chunk : Buffer.from(chunk, encoding) } - -/** - * Determine if the response is finished. - * - * @param {object} res - * @returns {boolean} - * @private - */ -function isFinished (res) { - return res.writableFinished || !res.socket.writable -} diff --git a/package.json b/package.json index 707c4aa7..449cbe23 100644 --- a/package.json +++ b/package.json @@ -14,6 +14,7 @@ "compressible": "^2.0.18", "debug": "^2.6.9", "negotiator": "^0.6.4", + "on-finished": "^2.4.1", "on-headers": "^1.0.2", "vary": "^1.1.2" },