Prevent server hangs on unconsumed streams.

package.json

@@ -65,9 +65,9 @@
     "build:clean": "rm -r lib; mkdir lib",
     "build:mjs": "BABEL_ESM=1 babel src -d lib --keep-file-extension",
     "build:js": "babel src -d lib",
-    "build:prettier": "prettier 'lib/**/*.{mjs,js}' --write",
+    "build:prettier": "prettier 'src/**/*.{mjs,js}' --write && prettier '*.{json,md}' --write",
     "lint:eslint": "eslint . --ext mjs,js",
-    "lint:prettier": "prettier '**/*.{json,md}' -l",
+    "lint:prettier": "prettier '*.{json,md}' -l",
     "tap:mjs": "node --experimental-modules --no-warnings lib/test | tap-mocha-reporter spec",
     "tap:js": "node lib/test | tap-mocha-reporter spec",
     "watch": "watch 'npm test --silent' src --interval 1",

src/middleware.mjs

@@ -22,6 +22,10 @@ class Upload {
           this.done = true
         })
 
+        // Prevent a crash if the stream disconnects before the consumers’ error
+        // listeners can attach.
+        file.stream.on('error', () => {})
+
         // Monkey patch busboy to emit an error when a file is too big.
         file.stream.once('limit', () =>
           file.stream.emit(
@@ -94,7 +98,7 @@ export const processRequest = (
               operationsPath.set(path, map.get(fieldName).promise)
           }
 
-          resolve(operations)
+          resolve({ operations, map })
         }
       }
     })
@@ -138,6 +142,8 @@ export const processRequest = (
             )
     })
 
+    parser.on('error', () => null)
+
     request.on('close', () => {
       if (map)
         for (const upload of map.values())
@@ -162,16 +168,39 @@ export const processRequest = (
   })
 
 export const apolloUploadKoa = options => async (ctx, next) => {
-  if (ctx.request.is('multipart/form-data'))
-    ctx.request.body = await processRequest(ctx.req, options)
-  await next()
+  if (!ctx.request.is('multipart/form-data')) return next()
+
+  // add uploads to the request
+  const { operations, map } = await processRequest(ctx.req, options)
+  ctx.request.body = operations
+
+  try {
+    ctx.respond = false
+    await next()
+  } finally {
+    await Promise.all(
+      [...map.values()].map(async upload => {
+        if (upload.done) return
+
+        await upload.promise
+        return new Promise(resolve => {
+          upload.file.stream.on('end', resolve)
+          upload.file.stream.on('error', resolve)
+          if (!upload.file.stream.readableFlowing) upload.file.stream.resume()
+        })
+      })
+    )
+
+    ctx.respond = true
+    if (ctx.body) ctx.body = ctx.body
+  }
 }
 
 export const apolloUploadExpress = options => (request, response, next) => {
   if (!request.is('multipart/form-data')) return next()
   processRequest(request, options)
-    .then(body => {
-      request.body = body
+    .then(({ operations }) => {
+      request.body = operations
       next()
     })
     .catch(error => {

src/test.mjs

@@ -1,4 +1,5 @@
 import fs from 'fs'
+import { Readable } from 'stream'
 import t from 'tap'
 import Koa from 'koa'
 import express from 'express'
@@ -101,6 +102,66 @@ t.test('Single file.', async t => {
   })
 })
 
+t.test('Early response.', async t => {
+  t.jobs = 1
+
+  const testRequest = async (port, stream) => {
+    const body = new FormData()
+
+    body.append(
+      'operations',
+      JSON.stringify({
+        variables: {
+          file: null
+        }
+      })
+    )
+
+    body.append('map', JSON.stringify({ 1: ['variables.file'] }))
+    body.append(1, stream)
+
+    await fetch(`http://localhost:${port}`, { method: 'POST', body })
+  }
+
+  await t.test('Koa middleware.', async t => {
+    t.plan(1)
+
+    const data = fs.readFileSync(TEST_FILE_PATH)
+
+    var requestHasFinished = false
+    const stream = new Readable()
+    stream.path = TEST_FILE_PATH
+    stream._read = () => {}
+    stream.on('end', () => {
+      requestHasFinished = true
+    })
+
+    const app = new Koa().use(apolloUploadKoa()).use(ctx => {
+      ctx.body = 'EARLY RETURN VALUE'
+    })
+    const port = await startServer(t, app)
+    const promise = testRequest(port, stream).then(
+      () => {
+        t.equals(
+          requestHasFinished,
+          true,
+          'The server should not respond before the request has finished'
+        )
+      },
+      err => {
+        throw err
+      }
+    )
+
+    setTimeout(() => {
+      stream.push(data)
+      stream.push(null)
+    }, 10)
+
+    return promise
+  })
+})
+
 t.test('Deduped files.', async t => {
   t.jobs = 2