diff --git a/README.md b/README.md
index bf8adf99..ec8e955e 100644
--- a/README.md
+++ b/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])
diff --git a/lib/base/request.js b/lib/base/request.js
index be318d9f..006e0e10 100644
--- a/lib/base/request.js
+++ b/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,6 +478,8 @@ 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}
@@ -460,6 +487,20 @@ class Request extends EventEmitter {
 
   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)
   }
 
diff --git a/test/common/tests.js b/test/common/tests.js
index bedc3421..a586acf8 100644
--- a/test/common/tests.js
+++ b/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
diff --git a/test/common/unit.js b/test/common/unit.js
index 380a2862..1cbae175 100644
--- a/test/common/unit.js
+++ b/test/common/unit.js
@@ -4,6 +4,7 @@
 
 const sql = require('../../')
 const assert = require('node:assert')
+const { Readable, Writable } = require('node:stream')
 const udt = require('../../lib/udt')
 const { assertSafeProcedureName } = require('../../lib/utils')
 const { declare } = require('../../lib/datatypes')
@@ -1404,6 +1405,215 @@ describe('connection string auth - tedious', () => {
     })
   })
 
+  describe('Request readable stream', () => {
+    const BaseRequest = require('../../lib/base/request')
+
+    it('resumes the request once the stream is destroyed', () => {
+      const req = new BaseRequest(null)
+      const readableStream = req.toReadableStream({ highWaterMark: 1 })
+      req.emit('row', { a: 1 })
+      req.emit('row', { a: 2 })
+      assert.strictEqual(req.paused, true, 'the request should be paused while nothing reads the stream')
+
+      readableStream.destroy()
+      assert.strictEqual(req.paused, false, 'destroying the stream should resume the request')
+
+      req.emit('row', { a: 3 })
+      assert.strictEqual(req.paused, false, 'the request should stay resumed while it sends the rest of its rows')
+      req.emit('done')
+    })
+
+    it('runs a destroy hook passed in the stream options', done => {
+      const failure = new Error('the destination failed')
+      const cleanupFailure = new Error('the cleanup failed')
+      const calls = []
+      const errors = []
+      const req = new BaseRequest(null)
+      const readableStream = req.toReadableStream({
+        highWaterMark: 1,
+        destroy (error, callback) {
+          calls.push({ stream: this, error, paused: req.paused })
+          callback(cleanupFailure)
+        }
+      })
+      readableStream.on('error', err => errors.push(err))
+      readableStream.on('close', () => {
+        assert.strictEqual(calls.length, 1, 'the hook should run once')
+        assert.strictEqual(calls[0].stream, readableStream, 'the hook should run with the stream as `this`')
+        assert.strictEqual(calls[0].error, failure, 'the hook should be given the error the stream was destroyed with')
+        assert.strictEqual(calls[0].paused, false, 'the request should be resumed before the hook runs')
+        assert.deepStrictEqual(errors, [cleanupFailure], 'the stream should emit the error the hook passed on')
+        done()
+      })
+
+      readableStream.destroy(failure)
+    })
+
+    it('takes null stream options as no options', () => {
+      const req = new BaseRequest(null)
+      const readableStream = req.toReadableStream(null)
+      assert.strictEqual(readableStream.readableObjectMode, true, 'the stream should be an object-mode stream')
+      readableStream.destroy()
+      assert.strictEqual(req.paused, false, 'destroying the stream should resume the request')
+    })
+
+    it('leaves a finished request alone when the stream is destroyed', () => {
+      const req = new BaseRequest(null)
+      const readableStream = req.toReadableStream({ highWaterMark: 1 })
+      req.emit('row', { a: 1 })
+      req.emit('done')
+
+      readableStream.destroy()
+      assert.strictEqual(req.paused, true, 'a request that has finished should not be resumed')
+    })
+
+    it('ends the stream with the request\'s first error', done => {
+      const first = new Error('first')
+      const errors = []
+      const req = new BaseRequest(null)
+      const readableStream = req.toReadableStream()
+      readableStream.on('error', err => errors.push(err))
+      readableStream.on('close', () => {
+        assert.deepStrictEqual(errors, [first], 'the stream should be sent the first error only')
+        assert.strictEqual(req.paused, false, 'the request should be resumed to run to the end')
+        done()
+      })
+
+      req.emit('error', first)
+      req.emit('row', { a: 1 })
+      req.emit('error', new Error('second'))
+      req.emit('done')
+    })
+
+    it('leaves a cancelled request to its driver when the stream is destroyed', () => {
+      const req = new BaseRequest(null)
+      const readableStream = req.toReadableStream({ highWaterMark: 1 })
+      const errors = []
+      readableStream.on('error', err => errors.push(err))
+      req.emit('row', { a: 1 })
+      assert.strictEqual(req.paused, true, 'the request should be paused while nothing reads the stream')
+
+      req.emit('error', new sql.RequestError('Canceled.', 'ECANCEL'))
+      assert.strictEqual(readableStream.destroyed, true, 'the cancel should destroy the stream')
+      assert.strictEqual(req.paused, true, 'a cancelled request should be left for its driver to stop')
+    })
+
+    describe('releases a piped request whose destination will not read any more', () => {
+      // the rest of a running request's rows, more than the stream buffers
+      const sendTheRest = (req) => {
+        for (let a = 100; a < 150; a++) req.emit('row', { a })
+      }
+
+      it('when the destination fails while the request is held back', done => {
+        const failure = new Error('the destination failed')
+        const errors = []
+        let finishWrite
+        const destination = new Writable({
+          objectMode: true,
+          highWaterMark: 1,
+          write (row, encoding, callback) {
+            // the first write never completes until the destination fails
+            finishWrite = callback
+          }
+        })
+        destination.on('error', err => errors.push(err))
+        const req = new BaseRequest(null)
+        req.pipe(destination)
+
+        setImmediate(() => {
+          for (let a = 0; a < 30; a++) req.emit('row', { a })
+          assert.strictEqual(req.paused, true, 'the request should be held back while the destination is full')
+
+          finishWrite(failure)
+          setImmediate(() => {
+            assert.deepStrictEqual(errors, [failure], 'the destination should report its own failure')
+            sendTheRest(req)
+            assert.strictEqual(req.paused, false, 'the request should not be held back once nothing reads its rows')
+            req.emit('done')
+            done()
+          })
+        })
+      })
+
+      it('when the destination fails without emitting close', done => {
+        const failure = new Error('the destination failed')
+        const destination = new Writable({
+          objectMode: true,
+          highWaterMark: 1,
+          autoDestroy: false,
+          emitClose: false,
+          write (row, encoding, callback) {
+            callback(failure)
+          }
+        })
+        const req = new BaseRequest(null)
+        destination.on('error', () => setImmediate(() => {
+          sendTheRest(req)
+          assert.strictEqual(req.paused, false, 'the request should not be held back once nothing reads its rows')
+          req.emit('done')
+          done()
+        }))
+
+        req.pipe(destination)
+        req.emit('row', { a: 0 })
+      })
+
+      it('when the destination is already destroyed', () => {
+        const destination = new Writable({
+          objectMode: true,
+          write (row, encoding, callback) {
+            callback()
+          }
+        })
+        destination.destroy()
+        const req = new BaseRequest(null)
+
+        req.pipe(destination)
+        sendTheRest(req)
+        assert.strictEqual(req.paused, false, 'the request should not be held back once nothing reads its rows')
+        req.emit('done')
+      })
+    })
+
+    it('leaves the errors of a pipe() destination to the caller', () => {
+      const writable = () => new Writable({
+        objectMode: true,
+        write (row, encoding, callback) {
+          callback()
+        }
+      })
+      const plain = writable()
+      new Readable({ objectMode: true, read () {} }).pipe(plain)
+      const destination = writable()
+      new BaseRequest(null).pipe(destination)
+      assert.strictEqual(destination.listenerCount('error'), plain.listenerCount('error'), 'pipe() should add no error listener of its own, so an unhandled destination error is still thrown')
+    })
+
+    it('destroys the destination of pipe() with the request\'s first error', done => {
+      const first = new Error('first')
+      const errors = []
+      const destination = new Writable({
+        objectMode: true,
+        write (row, encoding, callback) {
+          callback()
+        }
+      })
+      destination.on('error', err => errors.push(err))
+      destination.on('close', () => {
+        assert.deepStrictEqual(errors, [first], 'the destination should be sent the first error only')
+        assert.strictEqual(destination.errored, first, 'the destination should be destroyed with the first error')
+        done()
+      })
+
+      const req = new BaseRequest(null)
+      req.pipe(destination)
+      req.emit('error', first)
+      req.emit('row', { a: 1 })
+      req.emit('error', new Error('second'))
+      req.emit('done')
+    })
+  })
+
   describe('Transaction EABORT originalError', () => {
     const BaseTransaction = require('../../lib/base/transaction')
 
diff --git a/test/msnodesqlv8/msnodesqlv8.js b/test/msnodesqlv8/msnodesqlv8.js
index 3915e28a..a0ebd515 100644
--- a/test/msnodesqlv8/msnodesqlv8.js
+++ b/test/msnodesqlv8/msnodesqlv8.js
@@ -20,6 +20,10 @@ const reportsErrorsAfterAResult = major > 5 || (major === 5 && (minor > 1 || (mi
 // or polling, and lets any other run to the end
 const cancelsRunningQueries = major > 5 || (major === 5 && minor >= 6)
 
+// before 5.6 msnodesqlv8 stops a query at its first error and emits no 'done', so a stream never
+// completes after one, even an error that would not end the batch
+const continuesAfterErrors = major > 5 || (major === 5 && minor >= 6)
+
 // from 3.0 msnodesqlv8 binds a bulk column by the type of the column it loads into, and has no
 // binding for xml, sql_variant or the spatial types, which it refuses as an invalid parameter
 const itBulkLoadsAnyColumnType = major < 3 ? it : (title, fn) => it.skip(`${title} (not supported by msnodesqlv8 3 and later)`, fn)
@@ -114,6 +118,15 @@ describe('msnodesqlv8', function () {
     it('query with toReadableStream', done => TESTS['query with toReadableStream'](done))
     it('query with pipe', done => TESTS['query with pipe'](done))
     it('query with pipe and back pressure', (done) => TESTS['query with pipe and back pressure'](done))
+    it.skip('query with pipe and an error that ends the batch (msnodesqlv8 streams do not complete after one)', done => TESTS['query with pipe and an error'](true, done))
+    if (continuesAfterErrors) {
+      it('query with pipe and an error that does not end the batch', done => TESTS['query with pipe and an error'](false, done))
+    } else {
+      it.skip('query with pipe and an error that does not end the batch (msnodesqlv8 streams do not complete after an error before 5.6)', done => TESTS['query with pipe and an error'](false, done))
+    }
+    it('query with toReadableStream and a failing destination', done => TESTS['query with toReadableStream and a failing destination'](done))
+    it('query with pipe and a failing destination', done => TESTS['query with pipe and a failing destination'](done))
+    it('query with pipe into a destination that is already destroyed', done => TESTS['query with pipe into a destination that is already destroyed'](done))
     it('query with duplicate output column names', done => TESTS['query with duplicate output column names'](done))
     it('batch', done => TESTS.batch(done))
     it('batch (stream)', done => TESTS.batch(done, true))
diff --git a/test/tedious/tedious.js b/test/tedious/tedious.js
index fa3165d4..dd2ea9d7 100644
--- a/test/tedious/tedious.js
+++ b/test/tedious/tedious.js
@@ -107,6 +107,11 @@ describe('tedious', () => {
     it('query with toReadableStream', done => TESTS['query with toReadableStream'](done))
     it('query with pipe', done => TESTS['query with pipe'](done))
     it('query with pipe and back pressure', (done) => TESTS['query with pipe and back pressure'](done))
+    it('query with pipe and an error that ends the batch', done => TESTS['query with pipe and an error'](true, done))
+    it('query with pipe and an error that does not end the batch', done => TESTS['query with pipe and an error'](false, done))
+    it('query with toReadableStream and a failing destination', done => TESTS['query with toReadableStream and a failing destination'](done))
+    it('query with pipe and a failing destination', done => TESTS['query with pipe and a failing destination'](done))
+    it('query with pipe into a destination that is already destroyed', done => TESTS['query with pipe into a destination that is already destroyed'](done))
     it('query with duplicate output column names', done => TESTS['query with duplicate output column names'](done))
     it('batch', done => TESTS.batch(done))
     it('batch with output parameters and sql error', done => TESTS['batch with output parameters and sql error'](done))