Skip to content

Commit fbb806c

Browse files
martinslotabjohansebas
martinslota
and
bjohansebas
authored
Do not swallow calls to res.end() (#188)
* Add tests demonstrating scenarios under which the original res.end does not get called * Ensure stream gets closed and response gets ended when it is finished * Only call the original res.end() a single time * Avoid using the on-finished package * Use finished() helper from node:stream * Import finished() helper directly from node:stream Co-authored-by: Sebastian Beltran <bjohansebas@gmail.com> * Use finished() helper directly instead of nodeStream.finished() * Adjust stream pause / resume logic instead of messing with res.end() calls directly * Revert addition of endOnce() wrapper * Add unit test around one more alternate sequence of events that seems reasonable to test in addition to the other ones * Add test around avoiding 'write after end' errors * Avoid writing into the response after it is finished * Add a comment explaining why the compression stream gets resumed when the response finishes * chore: use isFinished of on-finished --------- Co-authored-by: Sebastian Beltran <bjohansebas@gmail.com>
1 parent f1b3e1c commit fbb806c

File tree

3 files changed

+223
-0
lines changed

3 files changed

+223
-0
lines changed

‎index.js‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,10 +14,12 @@
1414
* @private
1515
*/
1616

17+
var { finished } = require('node:stream')
1718
var Negotiator = require('negotiator')
1819
var bytes = require('bytes')
1920
var compressible = require('compressible')
2021
var debug = require('debug')('compression')
22+
const isFinished = require('on-finished').isFinished
2123
var onHeaders = require('on-headers')
2224
var vary = require('vary')
2325
var zlib = require('zlib')
@@ -215,7 +217,12 @@ function compression (options) {
215217

216218
// compression
217219
stream.on('data', function onStreamData (chunk) {
220+
if (isFinished(res)) {
221+
debug('response finished')
222+
return
223+
}
218224
if (_write.call(res, chunk) === false) {
225+
debug('pausing compression stream')
219226
stream.pause()
220227
}
221228
})
@@ -227,6 +234,15 @@ function compression (options) {
227234
_on.call(res, 'drain', function onResponseDrain () {
228235
stream.resume()
229236
})
237+
238+
// In case the stream is paused when the response finishes (e.g. because
239+
// the client cuts the connection), its `drain` event may not get emitted.
240+
// The following handler is here to ensure that the stream gets resumed so
241+
// it ends up emitting its `end` event and calling the original
242+
// `res.end()`.
243+
finished(res, function onResponseFinished () {
244+
stream.resume()
245+
})
230246
})
231247

232248
next()

‎package.json‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@
1414
"compressible": "^2.0.18",
1515
"debug": "^2.6.9",
1616
"negotiator": "^0.6.4",
17+
"on-finished": "^2.4.1",
1718
"on-headers": "^1.0.2",
1819
"vary": "^1.1.2"
1920
},

‎test/compression.js‎

Lines changed: 206 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ var assert = require('assert')
33
var bytes = require('bytes')
44
var crypto = require('crypto')
55
var http = require('http')
6+
var net = require('net')
67
var request = require('supertest')
78
var zlib = require('zlib')
89
var http2 = require('http2')
@@ -953,6 +954,198 @@ describe('compression()', function () {
953954
.expect(200, done)
954955
})
955956
})
957+
958+
describe('when the client closes the connection before consuming the response', function () {
959+
it('should call the original res.end() if connection is cut early on', function (done) {
960+
var server = http.createServer(function (req, res) {
961+
var originalResEnd = res.end
962+
var originalResEndCalledTimes = 0
963+
res.end = function () {
964+
originalResEndCalledTimes++
965+
return originalResEnd.apply(this, arguments)
966+
}
967+
968+
compression({ threshold: 0 })(req, res, function () {
969+
socket.end()
970+
971+
res.setHeader('Content-Type', 'text/plain')
972+
res.write('hello, ')
973+
setTimeout(function () {
974+
res.end('world!')
975+
976+
setTimeout(function () {
977+
server.close(function () {
978+
if (originalResEndCalledTimes === 1) {
979+
done()
980+
} else {
981+
done(new Error('The original res.end() was called ' + originalResEndCalledTimes + ' times'))
982+
}
983+
})
984+
}, 5)
985+
}, 5)
986+
})
987+
})
988+
989+
server.listen()
990+
991+
var port = server.address().port
992+
var socket = openSocketWithRequest(port)
993+
})
994+
995+
it('should call the original res.end() if connection is cut right after setting headers', function (done) {
996+
var server = http.createServer(function (req, res) {
997+
var originalResEnd = res.end
998+
var originalResEndCalledTimes = 0
999+
res.end = function () {
1000+
originalResEndCalledTimes++
1001+
return originalResEnd.apply(this, arguments)
1002+
}
1003+
1004+
compression({ threshold: 0 })(req, res, function () {
1005+
res.setHeader('Content-Type', 'text/plain')
1006+
socket.end()
1007+
1008+
res.write('hello, ')
1009+
setTimeout(function () {
1010+
res.end('world!')
1011+
1012+
setTimeout(function () {
1013+
server.close(function () {
1014+
if (originalResEndCalledTimes === 1) {
1015+
done()
1016+
} else {
1017+
done(new Error('The original res.end() was called ' + originalResEndCalledTimes + ' times'))
1018+
}
1019+
})
1020+
}, 5)
1021+
}, 5)
1022+
})
1023+
})
1024+
1025+
server.listen()
1026+
1027+
var port = server.address().port
1028+
var socket = openSocketWithRequest(port)
1029+
})
1030+
1031+
it('should call the original res.end() if connection is cut after an initial write', function (done) {
1032+
var server = http.createServer(function (req, res) {
1033+
var originalResEnd = res.end
1034+
var originalResEndCalledTimes = 0
1035+
res.end = function () {
1036+
originalResEndCalledTimes++
1037+
return originalResEnd.apply(this, arguments)
1038+
}
1039+
1040+
compression({ threshold: 0 })(req, res, function () {
1041+
res.setHeader('Content-Type', 'text/plain')
1042+
res.write('hello, ')
1043+
socket.end()
1044+
1045+
setTimeout(function () {
1046+
res.end('world!')
1047+
1048+
setTimeout(function () {
1049+
server.close(function () {
1050+
if (originalResEndCalledTimes === 1) {
1051+
done()
1052+
} else {
1053+
done(new Error('The original res.end() was called ' + originalResEndCalledTimes + ' times'))
1054+
}
1055+
})
1056+
}, 5)
1057+
}, 5)
1058+
})
1059+
})
1060+
1061+
server.listen()
1062+
1063+
var port = server.address().port
1064+
var socket = openSocketWithRequest(port)
1065+
})
1066+
1067+
it('should call the original res.end() if connection is cut just after response body was generated', function (done) {
1068+
var server = http.createServer(function (req, res) {
1069+
var originalResEnd = res.end
1070+
var originalResEndCalledTimes = 0
1071+
res.end = function () {
1072+
originalResEndCalledTimes++
1073+
return originalResEnd.apply(this, arguments)
1074+
}
1075+
1076+
compression({ threshold: 0 })(req, res, function () {
1077+
res.setHeader('Content-Type', 'text/plain')
1078+
res.write('hello, ')
1079+
res.end('world!')
1080+
socket.end()
1081+
1082+
setTimeout(function () {
1083+
server.close(function () {
1084+
if (originalResEndCalledTimes === 1) {
1085+
done()
1086+
} else {
1087+
done(new Error('The original res.end() was called ' + originalResEndCalledTimes + ' times'))
1088+
}
1089+
})
1090+
}, 5)
1091+
})
1092+
})
1093+
1094+
server.listen()
1095+
1096+
var port = server.address().port
1097+
var socket = openSocketWithRequest(port)
1098+
})
1099+
1100+
it('should not trigger write errors if connection is cut just after response body was generated', function (done) {
1101+
var requestCount = 0
1102+
1103+
var server = http.createServer(function (req, res) {
1104+
requestCount += 1
1105+
1106+
var originalWrite = res.write
1107+
var writeError = null
1108+
res.write = function (chunk, callback) {
1109+
return originalWrite.call(this, chunk, function (error) {
1110+
if (error) {
1111+
writeError = error
1112+
}
1113+
return callback?.(error)
1114+
})
1115+
}
1116+
1117+
var originalResEnd = res.end
1118+
res.end = function () {
1119+
setTimeout(function () {
1120+
if (writeError !== null) {
1121+
server.close(function () {
1122+
done(new Error(`Write error occurred: ${writeError}`))
1123+
})
1124+
} else {
1125+
if (requestCount < 50) {
1126+
socket = openSocketWithRequest(port)
1127+
} else {
1128+
server.close(done)
1129+
}
1130+
}
1131+
}, 0)
1132+
return originalResEnd.apply(this, arguments)
1133+
}
1134+
1135+
compression({ threshold: 0 })(req, res, function () {
1136+
res.setHeader('Content-Type', 'text/plain')
1137+
res.write('hello, ')
1138+
res.end('world!')
1139+
socket.end()
1140+
})
1141+
})
1142+
1143+
server.listen()
1144+
1145+
var port = server.address().port
1146+
var socket = openSocketWithRequest(port)
1147+
})
1148+
})
9561149
})
9571150

9581151
function createServer (opts, fn) {
@@ -1056,3 +1249,16 @@ function unchunk (encoding, onchunk, onend) {
10561249
stream.on('end', onend)
10571250
}
10581251
}
1252+
1253+
function openSocketWithRequest (port) {
1254+
var socket = net.connect(port, function onConnect () {
1255+
socket.write('GET / HTTP/1.1\r\n')
1256+
socket.write('Accept-Encoding: gzip\r\n')
1257+
socket.write('Host: localhost:' + port + '\r\n')
1258+
socket.write('Content-Type: text/plain\r\n')
1259+
socket.write('Content-Length: 0\r\n')
1260+
socket.write('Connection: keep-alive\r\n')
1261+
socket.write('\r\n')
1262+
})
1263+
return socket
1264+
}

0 commit comments

Comments
 (0)