Do not swallow calls to res.end() (#188)

martinslota · bjohansebas · web-flow · commit fbb806c86363 · 2025-04-16T12:28:55.000-05:00
* 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 &lt;bjohansebas@gmail.com&gt;

* 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 &lt;bjohansebas@gmail.com&gt;
diff --git a/index.js b/index.js
@@ -14,10 +14,12 @@
  * @private
  */
 
+var { finished } = require('node:stream')
 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')
@@ -215,7 +217,12 @@ 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()
         }
       })
@@ -227,6 +234,15 @@ function compression (options) {
       _on.call(res, 'drain', function onResponseDrain () {
         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()
+      })
     })
 
     next()
diff --git 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"
   },
diff --git 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,198 @@ 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 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
+        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)
+    })
+
+    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) {
+              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)
+    })
+  })
 })
 
 function createServer (opts, fn) {
@@ -1056,3 +1249,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
+}
