fix: stop Request.pipe() crashing the process on request errors

README.md

@@ -1074,6 +1074,12 @@ pipeline(readableStream, transformStream, writableStream)
 request.query('select * from mytable')
 ```
 
+The readable stream is destroyed with the request's first error. Later errors are still emitted on the request, but not on the stream.
+
+If the readable stream is destroyed before the request finishes (for example because a later stream in the pipeline failed), the request still runs to the end and the rows it has left are discarded. Call `request.cancel()` to stop it early.
+
+With msnodesqlv8 before 5.6 that cancel does not stop it. The request is resumed as soon as its stream is destroyed, so that it does not keep its connection, and those versions can only cancel a paused request.
+
 
 ### pipe (stream)
 
@@ -1097,6 +1103,21 @@ stream.on('finish', () => {
 })
 ```
 
+The stream is destroyed with the request's first error, so listen for `error` on it as above. It does not emit `finish` after an error. The request still runs to the end, and its remaining rows are discarded.
+
+If the stream fails or closes before the request finishes, or was already destroyed, the request also runs to the end and its remaining rows are discarded, rather than waiting for a stream that will not read any more. Call `request.cancel()` to stop it early, bearing in mind the msnodesqlv8 caveat under [toReadableStream](#toreadablestream).
+
+`pipe()` passes errors only to the stream it is given. If you pipe through more streams after that one, use `stream.pipeline()` with [`toReadableStream`](#toreadablestream) instead, so that every error reaches you:
+
+```javascript
+const { pipeline } = require('stream')
+const request = new sql.Request()
+pipeline(request.toReadableStream(), transformStream, writableStream, err => {
+    // ...
+})
+request.query('select * from mytable')
+```
+
 ---------------------------------------
 
 ### query (command, [callback])

lib/base/request.js

@@ -10,6 +10,17 @@ const { TYPES } = require('../datatypes')
 const shared = require('../shared')
 const { CHANNELS, tracePromise, traceCallback, publish } = require('../diagnostics')
 
+// a Readable runs a single destroy hook, so one the caller passed in its stream options runs after
+// `destroy`, with the stream as `this` as the Readable would have run it. The options are spread to
+// read it, as they are for the Readable, so only an own `destroy` counts and null options are fine
+function withDestroyHook (streamOptions, destroy) {
+  const { destroy: hook } = { ...streamOptions }
+  if (typeof hook !== 'function') return destroy
+  return function (error, callback) {
+    destroy(error, () => hook.call(this, error, callback))
+  }
+}
+
 /**
  * Class Request.
  *
@@ -429,22 +440,36 @@ class Request extends EventEmitter {
   toReadableStream (streamOptions = {}) {
     this.stream = true
     this.pause()
+    let requestDone = false
+    let requestCanceled = false
     const readableStream = new Readable({
       ...streamOptions,
       objectMode: true,
       read: (/* size */) => {
         this.resume()
-      }
+      },
+      destroy: withDestroyHook(streamOptions, (error, callback) => {
+        // nothing will read from the stream again, so the request must not be left paused: it
+        // runs to the end, its remaining rows are dropped and its connection is released. A
+        // cancelled request is left to its driver, which is already stopping it: msnodesqlv8
+        // reports the cancel once the query is gone, and resuming it then reads a freed statement
+        if (!requestDone && !requestCanceled && this.paused) this.resume()
+        callback(error)
+      })
     })
     this.on('row', (row) => {
-      if (!readableStream.push(row)) {
+      if (!readableStream.push(row) && !readableStream.destroyed) {
         this.pause()
       }
     })
     this.on('error', (error) => {
-      readableStream.emit('error', error)
+      if (error.code === 'ECANCEL') requestCanceled = true
+      // the stream ends with the request's first error; destroying an already destroyed stream
+      // does nothing, so later errors stay with the request
+      readableStream.destroy(error)
     })
     this.on('done', () => {
+      requestDone = true
       readableStream.push(null)
     })
     return readableStream
@@ -453,13 +478,29 @@ class Request extends EventEmitter {
   /**
    * Wrap original request in a Readable stream that supports back pressure and pipe to the Writable stream.
    * It also sets request to `stream` mode and pulls all rows from all recordsets to a given stream.
+   * The Writable stream is destroyed with the request's first error. If the Writable stream fails or
+   * closes first, the request runs to the end and its remaining rows are discarded.
    *
    * @param {Stream} stream Stream to pipe data into.
    * @return {Stream}
    */
 
   pipe (writableStream) {
     const readableStream = this.toReadableStream()
+    // readable.pipe() leaves the source's errors to the source, but only the destination is
+    // handed back, so pass them on to it as pipeline() does
+    readableStream.on('error', (error) => {
+      writableStream.destroy(error)
+    })
+    // and when the destination fails or closes first, readable.pipe() only unpipes from it, which
+    // leaves the request paused with its connection: destroy the stream so the request runs to
+    // the end. 'unpipe' comes even from a destination that never emits 'close', and unlike
+    // finished() it adds no 'error' listener, which would stop an unhandled failure being thrown
+    writableStream.on('unpipe', (source) => {
+      if (source === readableStream && !readableStream.readableEnded) readableStream.destroy()
+    })
+    // a destination that is already destroyed will not fail or close again
+    if (writableStream.destroyed) readableStream.destroy()
     return readableStream.pipe(writableStream)
   }
 

test/common/tests.js

@@ -670,6 +670,141 @@ module.exports = (sql, driver) => {
       req.pipe(blockingTransformStream).pipe(wstream)
     },
 
+    'query with pipe and an error' (endsTheBatch, done) {
+      // pipe() returns the destination, not the stream it puts in front of it, so the
+      // destination is the only place a request's error can be handled
+      const command = endsTheBatch
+        ? 'select * from notexistingtable'
+        // the rows after this error come once the destination has been destroyed
+        : 'raiserror(N\'Something went wrong\', 16, 1); select \'asdf\' as text'
+      const pool = new sql.Request().parent
+      const before = pool.pool.numUsed()
+      const errors = []
+
+      const destination = new WritableStream()
+      destination.on('error', err => errors.push(err))
+
+      const req = new sql.Request()
+      req.on('done', () => {
+        setImmediate(() => {
+          try {
+            assert.strictEqual(errors.length, 1, 'the destination should be sent the request\'s error once')
+            assert.ok(errors[0] instanceof sql.RequestError, `the destination's error should be a RequestError, got ${errors[0].constructor.name}`)
+            assert.strictEqual(destination.errored, errors[0], 'the destination should be destroyed with the request\'s error')
+            assert.deepStrictEqual(destination.cache, [], 'the rows after the error should be dropped')
+            assert.strictEqual(pool.pool.numUsed(), before, 'the pool should have every connection back')
+            done()
+          } catch (e) {
+            done(e)
+          }
+        })
+      })
+      req.pipe(destination)
+      req.query(command)
+    },
+
+    'query with toReadableStream and a failing destination' (done) {
+      const pool = new sql.Request().parent
+      const before = pool.pool.numUsed()
+      const failure = new Error('the destination failed')
+      let pipelineError = null
+      let written = 0
+
+      const destination = new stream.Writable({
+        objectMode: true,
+        highWaterMark: 1,
+        write (row, encoding, callback) {
+          // fail once the request has filled the stream and been paused
+          callback(++written === 20 ? failure : null)
+        }
+      })
+
+      const req = new sql.Request()
+      req.on('done', () => {
+        setImmediate(() => {
+          try {
+            assert.strictEqual(pipelineError, failure, 'the pipeline should fail with the destination\'s error')
+            assert.strictEqual(pool.pool.numUsed(), before, 'the pool should have every connection back')
+            done()
+          } catch (e) {
+            done(e)
+          }
+        })
+      })
+      stream.pipeline(req.toReadableStream(), destination, err => {
+        pipelineError = err
+      })
+      req.query('select top 2000 * from streaming')
+    },
+
+    'query with pipe and a failing destination' (done) {
+      const pool = new sql.Request().parent
+      const before = pool.pool.numUsed()
+      const failure = new Error('the destination failed')
+      const errors = []
+      let heldBackAtFailure = null
+      let written = 0
+
+      const req = new sql.Request()
+      const destination = new stream.Writable({
+        objectMode: true,
+        highWaterMark: 1,
+        write (row, encoding, callback) {
+          // slow enough for the stream to fill and hold the request back, then fail
+          setTimeout(() => {
+            if (++written === 20) {
+              heldBackAtFailure = req.paused
+              return callback(failure)
+            }
+            callback()
+          }, 1)
+        }
+      })
+      destination.on('error', err => errors.push(err))
+
+      req.on('done', () => {
+        setImmediate(() => {
+          try {
+            assert.strictEqual(heldBackAtFailure, true, 'the request should be held back by the full stream when the destination fails')
+            assert.deepStrictEqual(errors, [failure], 'the destination should report its own failure')
+            assert.strictEqual(pool.pool.numUsed(), before, 'the pool should have every connection back')
+            done()
+          } catch (e) {
+            done(e)
+          }
+        })
+      })
+      req.pipe(destination)
+      req.query('select top 2000 * from streaming')
+    },
+
+    'query with pipe into a destination that is already destroyed' (done) {
+      const pool = new sql.Request().parent
+      const before = pool.pool.numUsed()
+
+      const destination = new stream.Writable({
+        objectMode: true,
+        write (row, encoding, callback) {
+          callback()
+        }
+      })
+      destination.destroy()
+
+      const req = new sql.Request()
+      req.on('done', () => {
+        setImmediate(() => {
+          try {
+            assert.strictEqual(pool.pool.numUsed(), before, 'the pool should have every connection back')
+            done()
+          } catch (e) {
+            done(e)
+          }
+        })
+      })
+      req.pipe(destination)
+      req.query('select top 2000 * from streaming')
+    },
+
     'query with duplicate output column names' (done) {
       const req = new TestRequest()
       req.arrayRowMode = true