From cbcd0f44ab20e49966e0bb0ca8a880810a4a9424 Mon Sep 17 00:00:00 2001
From: Dan Hensby <git@hens.by>
Date: Mon, 12 Oct 2026 16:46:14 +0100
Subject: [PATCH 1/3] fix: resume a request when its readable stream is
 destroyed

toReadableStream() pauses the request whenever the stream's buffer is
full and resumes it only when the stream asks for more. A stream that is
destroyed never asks again. This happens, for example, when pipeline()
tears down every stream after a later one fails. The request then stayed
paused forever and kept its pooled connection.

When the stream is destroyed before the request is done, resume the
request and stop pausing it. It runs to the end, its remaining rows are
dropped, and its connection is released.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 README.md                       |  4 +++
 lib/base/request.js             | 23 ++++++++++--
 test/common/tests.js            | 34 ++++++++++++++++++
 test/common/unit.js             | 63 +++++++++++++++++++++++++++++++++
 test/msnodesqlv8/msnodesqlv8.js |  1 +
 test/tedious/tedious.js         |  1 +
 6 files changed, 124 insertions(+), 2 deletions(-)

diff --git a/README.md b/README.md
index bf8adf99..88d24575 100644
--- a/README.md
+++ b/README.md
@@ -1074,6 +1074,10 @@ pipeline(readableStream, transformStream, writableStream)
 request.query('select * from mytable')
 ```
 
+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)
 
diff --git a/lib/base/request.js b/lib/base/request.js
index be318d9f..a3612c03 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,15 +440,22 @@ class Request extends EventEmitter {
   toReadableStream (streamOptions = {}) {
     this.stream = true
     this.pause()
+    let requestDone = 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
+        if (!requestDone && this.paused) this.resume()
+        callback(error)
+      })
     })
     this.on('row', (row) => {
-      if (!readableStream.push(row)) {
+      if (!readableStream.push(row) && !readableStream.destroyed) {
         this.pause()
       }
     })
@@ -445,6 +463,7 @@ class Request extends EventEmitter {
       readableStream.emit('error', error)
     })
     this.on('done', () => {
+      requestDone = true
       readableStream.push(null)
     })
     return readableStream
diff --git a/test/common/tests.js b/test/common/tests.js
index bedc3421..101819c1 100644
--- a/test/common/tests.js
+++ b/test/common/tests.js
@@ -670,6 +670,40 @@ module.exports = (sql, driver) => {
       req.pipe(blockingTransformStream).pipe(wstream)
     },
 
+    '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 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..746047b9 100644
--- a/test/common/unit.js
+++ b/test/common/unit.js
@@ -1404,6 +1404,69 @@ 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')
+    })
+  })
+
   describe('Transaction EABORT originalError', () => {
     const BaseTransaction = require('../../lib/base/transaction')
 
diff --git a/test/msnodesqlv8/msnodesqlv8.js b/test/msnodesqlv8/msnodesqlv8.js
index 3915e28a..360df64a 100644
--- a/test/msnodesqlv8/msnodesqlv8.js
+++ b/test/msnodesqlv8/msnodesqlv8.js
@@ -114,6 +114,7 @@ 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('query with toReadableStream and a failing destination', done => TESTS['query with toReadableStream and a failing destination'](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..61bf2375 100644
--- a/test/tedious/tedious.js
+++ b/test/tedious/tedious.js
@@ -107,6 +107,7 @@ 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 toReadableStream and a failing destination', done => TESTS['query with toReadableStream and a failing destination'](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))

From fc19d959c67d2a7acfe8ab1b8fd718697a5060fd Mon Sep 17 00:00:00 2001
From: Dan Hensby <git@hens.by>
Date: Mon, 12 Oct 2026 16:49:01 +0100
Subject: [PATCH 2/3] fix: pass request errors to the destination of
 Request.pipe()

pipe() pipes an internal readable stream into the destination and returns
the destination, so nothing can listen for errors on the readable stream.
toReadableStream() re-emitted each request error on it, and
Readable.prototype.pipe() does not forward a source's errors. So every
error a piped request hit, such as a SQL error, a timeout or a cancel,
was thrown from inside the driver's event handler and crashed the
process. A listener on the destination or on the request did not help.

pipe() now destroys the destination with the request's first error, as
pipeline() does, so the destination's 'error' listener receives it.

toReadableStream() now destroys its stream with that error instead of
emitting it, which follows the stream contract. A request can emit more
than one error and then carry on to 'done'. Later errors and rows can no
longer reach a stream that has already failed: the errors stay on the
request and the rows are dropped.

A cancelled request is not resumed when its stream is destroyed. Its
driver is already stopping it. tedious resumes a cancelled request
itself, and msnodesqlv8 reports the cancel after the query is gone.
Resuming the request then would make msnodesqlv8 read the freed
statement on a connection that is already back in the pool.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 README.md                       | 15 +++++++++
 lib/base/request.js             | 18 +++++++++--
 test/common/tests.js            | 33 +++++++++++++++++++
 test/common/unit.js             | 56 +++++++++++++++++++++++++++++++++
 test/msnodesqlv8/msnodesqlv8.js | 10 ++++++
 test/tedious/tedious.js         |  2 ++
 6 files changed, 131 insertions(+), 3 deletions(-)

diff --git a/README.md b/README.md
index 88d24575..d175a8ce 100644
--- a/README.md
+++ b/README.md
@@ -1074,6 +1074,8 @@ 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.
@@ -1101,6 +1103,19 @@ 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.
+
+`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 a3612c03..cdece02c 100644
--- a/lib/base/request.js
+++ b/lib/base/request.js
@@ -441,6 +441,7 @@ class Request extends EventEmitter {
     this.stream = true
     this.pause()
     let requestDone = false
+    let requestCanceled = false
     const readableStream = new Readable({
       ...streamOptions,
       objectMode: true,
@@ -449,8 +450,10 @@ class Request extends EventEmitter {
       },
       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
-        if (!requestDone && this.paused) this.resume()
+        // 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)
       })
     })
@@ -460,7 +463,10 @@ class Request extends EventEmitter {
       }
     })
     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
@@ -472,6 +478,7 @@ 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.
    *
    * @param {Stream} stream Stream to pipe data into.
    * @return {Stream}
@@ -479,6 +486,11 @@ 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)
+    })
     return readableStream.pipe(writableStream)
   }
 
diff --git a/test/common/tests.js b/test/common/tests.js
index 101819c1..61bed927 100644
--- a/test/common/tests.js
+++ b/test/common/tests.js
@@ -670,6 +670,39 @@ 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()
diff --git a/test/common/unit.js b/test/common/unit.js
index 746047b9..31ebb213 100644
--- a/test/common/unit.js
+++ b/test/common/unit.js
@@ -4,6 +4,7 @@
 
 const sql = require('../../')
 const assert = require('node:assert')
+const { Writable } = require('node:stream')
 const udt = require('../../lib/udt')
 const { assertSafeProcedureName } = require('../../lib/utils')
 const { declare } = require('../../lib/datatypes')
@@ -1465,6 +1466,61 @@ describe('connection string auth - tedious', () => {
       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')
+    })
+
+    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', () => {
diff --git a/test/msnodesqlv8/msnodesqlv8.js b/test/msnodesqlv8/msnodesqlv8.js
index 360df64a..65681aaa 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,12 @@ 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 duplicate output column names', done => TESTS['query with duplicate output column names'](done))
     it('batch', done => TESTS.batch(done))
diff --git a/test/tedious/tedious.js b/test/tedious/tedious.js
index 61bf2375..39bc7b9c 100644
--- a/test/tedious/tedious.js
+++ b/test/tedious/tedious.js
@@ -107,6 +107,8 @@ 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 duplicate output column names', done => TESTS['query with duplicate output column names'](done))
     it('batch', done => TESTS.batch(done))

From d039a07671cc6d7fab64aa8bb828ba631f6f766a Mon Sep 17 00:00:00 2001
From: Dan Hensby <git@hens.by>
Date: Tue, 13 Oct 2026 18:51:15 +0100
Subject: [PATCH 3/3] fix: release a piped request when its destination fails

When the destination of Request.pipe() fails or closes before the request
finishes, Readable.prototype.pipe() only unpipes from it. The readable
stream that pipe() puts in front of the destination is internal, so
nothing destroys it. It stays full, the request stays paused, and the
request keeps its pooled connection for good. Piping into a destination
that has already been destroyed does the same.

pipe() now destroys that stream when the destination unpipes before the
stream has ended, or is already destroyed when pipe() is called, as
pipeline() would. The request then runs to the end, its remaining rows
are dropped, and its connection is released.

pipe() listens for 'unpipe' rather than 'close'. A destination created
with autoDestroy or emitClose turned off never emits 'close'. It also
does not use finished(), because finished() adds an 'error' listener,
which would stop an unhandled destination error from being thrown.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 README.md                       |  2 +
 lib/base/request.js             | 12 ++++-
 test/common/tests.js            | 68 ++++++++++++++++++++++++
 test/common/unit.js             | 93 ++++++++++++++++++++++++++++++++-
 test/msnodesqlv8/msnodesqlv8.js |  2 +
 test/tedious/tedious.js         |  2 +
 6 files changed, 177 insertions(+), 2 deletions(-)

diff --git a/README.md b/README.md
index d175a8ce..ec8e955e 100644
--- a/README.md
+++ b/README.md
@@ -1105,6 +1105,8 @@ 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
diff --git a/lib/base/request.js b/lib/base/request.js
index cdece02c..006e0e10 100644
--- a/lib/base/request.js
+++ b/lib/base/request.js
@@ -478,7 +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.
+   * 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}
@@ -491,6 +492,15 @@ class Request extends EventEmitter {
     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 61bed927..a586acf8 100644
--- a/test/common/tests.js
+++ b/test/common/tests.js
@@ -737,6 +737,74 @@ module.exports = (sql, driver) => {
       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 31ebb213..1cbae175 100644
--- a/test/common/unit.js
+++ b/test/common/unit.js
@@ -4,7 +4,7 @@
 
 const sql = require('../../')
 const assert = require('node:assert')
-const { Writable } = require('node:stream')
+const { Readable, Writable } = require('node:stream')
 const udt = require('../../lib/udt')
 const { assertSafeProcedureName } = require('../../lib/utils')
 const { declare } = require('../../lib/datatypes')
@@ -1498,6 +1498,97 @@ describe('connection string auth - tedious', () => {
       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 = []
diff --git a/test/msnodesqlv8/msnodesqlv8.js b/test/msnodesqlv8/msnodesqlv8.js
index 65681aaa..a0ebd515 100644
--- a/test/msnodesqlv8/msnodesqlv8.js
+++ b/test/msnodesqlv8/msnodesqlv8.js
@@ -125,6 +125,8 @@ describe('msnodesqlv8', function () {
       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 39bc7b9c..dd2ea9d7 100644
--- a/test/tedious/tedious.js
+++ b/test/tedious/tedious.js
@@ -110,6 +110,8 @@ describe('tedious', () => {
     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))