fix: stop msnodesqlv8 streams resumed after they end from hanging the connection

lib/msnodesqlv8/request.js

@@ -30,6 +30,12 @@ const thrownMessage = function (value) {
   }
 }
 
+// msnodesqlv8 >= 3 keeps the reader of a query on it, and the reader stops running as soon as
+// the query ends, a few ticks before msnodesqlv8 reports the end with 'done' or 'free'
+const stopped = function (query) {
+  return query.queryWorker?.running === false
+}
+
 const castParameter = function (value, type) {
   if (value == null) {
     if ((type === TYPES.Binary) || (type === TYPES.VarBinary) || (type === TYPES.Image)) {
@@ -349,8 +355,9 @@ class Request extends BaseRequest {
       // set while the error just collected is being dispatched, along with whatever came with it
       let failing = false
       let cancelRequested = false
-      // 'done' ends the request. msnodesqlv8 >= 5.6 emits it as soon as a cancel is through, and
-      // the query can go on emitting after it, so the listeners ignore anything that follows
+      // 'done' or 'free' ends the request. msnodesqlv8 >= 5.6 emits 'done' as soon as a cancel is
+      // through, and the query can go on emitting after it, so the listeners ignore anything that
+      // follows
       let finished = false
 
       // A stream goes on receiving rows after it is paused, until the driver reaches the end of a
@@ -532,6 +539,22 @@ class Request extends BaseRequest {
 
         this._setCurrentRequest(req)
 
+        // msnodesqlv8 does not check that a query is still running when it is paused or resumed.
+        // Pausing and resuming one that has ended puts it back on its connection's work queue,
+        // where it holds up whatever that connection is sent next, and resuming it can set it
+        // reading again from a statement that is gone, or crash the process. So once msnodesqlv8
+        // reports that the query has ended, the request no longer pauses, resumes or cancels it,
+        // before anything the request emits from then on is handled. Before then it leaves the
+        // query alone while an error that may end the batch is being handled and, from msnodesqlv8
+        // 3, once the query's reader has stopped
+        let detached = false
+        let settingAside = false
+        let cancelHeld = false
+        const detach = () => {
+          detached = true
+        }
+        const reachable = () => !detached && !settingAside && !stopped(req)
+
         const cancelDriver = (attempt = cancelAttempts) => {
           req.cancelQuery(err => {
             if (err) {
@@ -583,7 +606,7 @@ class Request extends BaseRequest {
         // until the first result set the driver has read nothing, so a stream paused before then,
         // or before its query starts, holds it back at once
         this._pauseStream = () => {
-          if (!columns && !driverPaused && !finished) {
+          if (!columns && !driverPaused && !finished && reachable()) {
             driverPaused = true
             req.pauseQuery()
           }
@@ -606,14 +629,22 @@ class Request extends BaseRequest {
           // stream completes as cancelled straight away, as it does when it is paused first
           if (finished && cancelRequested && this.paused && pending.length) emitPendingOnCancel()
           // a query that is being cancelled stays paused, so that the cancel can go through
-          if (!pending.length && !this.paused && driverPaused && !finished && !cancelRequested) {
+          if (!pending.length && !this.paused && driverPaused && !finished && !cancelRequested && reachable()) {
             driverPaused = false
             req.resumeQuery()
           }
         }
 
-        this._cancel = () => {
+        const cancel = () => {
           debug('request(%d): cancel', IDS.get(this))
+          // a query that has ended has nothing to cancel, nor one whose reader has stopped. The
+          // cancel is held: it is made if msnodesqlv8 goes on to a result set after all, and once
+          // the query is through a paused stream completes as cancelled, without waiting to be
+          // resumed
+          if (detached || stopped(req) || settingAside) {
+            cancelHeld = true
+            return
+          }
           // newer msnodesqlv8 versions drop a repeat cancel without calling back, so the
           // outcome of the first one stands
           if (cancelDeferred || (typeof req.getCancelSent === 'function' && req.getCancelSent())) {
@@ -633,9 +664,19 @@ class Request extends BaseRequest {
           if (freed) return req.emit('done')
           cancelDeferred = true
         }
+        this._cancel = cancel
 
         req.on('meta', metadata => {
           if (finished) return
+          // a query that goes on to a result set has not ended: msnodesqlv8 can give the last error
+          // of a statement that fails first more=false, and carry on to the statements after it
+          if (detached) {
+            detached = false
+            if (cancelHeld) {
+              cancelHeld = false
+              cancel()
+            }
+          }
 
           if (row) {
             if (isChunkedRecordset) {
@@ -812,6 +853,7 @@ class Request extends BaseRequest {
               // msnodesqlv8 2.0 ends a statement's error with more=true and this message, and then
               // nothing follows: the error just collected is the one that terminated the statement.
               // A stream has already emitted it
+              detach()
               if (!this.stream) {
                 const error = errors.pop()
                 handleError(req, connection, error.originalError || error, false)
@@ -831,6 +873,27 @@ class Request extends BaseRequest {
             connection.hasError = true
           }
 
+          if (!moreErrors) {
+            // an error with nothing more to follow ends the batch
+            detach()
+          } else if (!detached && !settingAside) {
+            // the errors that end a batch are emitted together, all but the last with more=true,
+            // and msnodesqlv8 >= 5.1 releases the statement as soon as they are, so the query is
+            // set aside while they are handled, and a cancel made meanwhile waits until they have
+            // been. If none of them ended the batch it is taken up again and, if the stream was
+            // paused meanwhile before its first result set, paused, unless it is being cancelled.
+            // A stream that is paused holds its events back, so none is resumed meanwhile
+            settingAside = true
+            queueMicrotask(() => {
+              settingAside = false
+              if (cancelHeld) {
+                cancelHeld = false
+                cancel()
+              }
+              if (!detached && !cancelRequested && this.paused && this._pauseStream) this._pauseStream()
+            })
+          }
+
           if (finished) return
 
           handleError(req, connection, info, moreErrors)
@@ -844,8 +907,14 @@ class Request extends BaseRequest {
           }
         })
 
-        req.once('done', () => {
+        // msnodesqlv8 emits 'done' when a batch completes, but not when an error ends it; either
+        // way it emits 'free' once it has released the statement. A stream reports its errors as
+        // they arrive rather than settling on them, so without 'free' it would never complete
+        // after such an error, nor give its connection back
+        const complete = () => {
           finished = true
+          detach()
+          if (cancelHeld && this.stream && this.paused) cancelRequested = true
 
           if (hasReturned) {
             return
@@ -973,7 +1042,10 @@ class Request extends BaseRequest {
             delete this._cancel
             callback(null, recordsets, output, rowsAffected, recordsetcolumns)
           }
-        })
+        }
+
+        req.once('done', complete)
+        req.once('free', complete)
       })
     })
   }

test/common/tests.js

@@ -3,6 +3,7 @@
 const assert = require('node:assert')
 const stream = require('node:stream')
 const { join } = require('node:path')
+const { readFileSync } = require('node:fs')
 const { format } = require('node:util')
 const ISOLATION_LEVELS = require('../../lib/isolationlevel')
 const BaseTransaction = require('../../lib/base/transaction')
@@ -29,8 +30,10 @@ class WritableStream extends stream.Writable {
   }
 }
 
+// read afresh for each test: tests change the config they are given, and a copy that
+// require() caches would carry the change into every test that comes after
 const readConfig = () => {
-  return require(join(__dirname, '../.mssql.json'))
+  return JSON.parse(readFileSync(join(__dirname, '../.mssql.json'), 'utf8'))
 }
 
 module.exports = (sql, driver) => {
@@ -2982,6 +2985,141 @@ module.exports = (sql, driver) => {
       })
     },
 
+    'streaming an error that ends the batch' (done) {
+      // msnodesqlv8 emits no 'done' after an error that ends the batch: the stream still has
+      // to report the error, then complete once and give its connection back. The pool is the
+      // test's own, so that no other test's connection is out while it is checked
+      const config = readConfig()
+      config.driver = driver
+
+      new sql.ConnectionPool(config).connect().then(conn => {
+        const errors = []
+        let completed = 0
+
+        const req = new sql.Request(conn)
+        req.stream = true
+        req.on('error', err => errors.push(err))
+        req.on('done', () => {
+          if (completed++) return
+          setImmediate(() => {
+            try {
+              assert.strictEqual(completed, 1, 'the stream should complete once')
+              assert.strictEqual(errors.length, 1, 'the error should be streamed once')
+              assert.ok(errors[0] instanceof sql.RequestError, `the error should be a RequestError, got ${errors[0].constructor.name}`)
+              assert.strictEqual(errors[0].code, 'EREQUEST', 'the error should be a request error')
+              assert.strictEqual(conn.pool.numUsed(), 0, 'the pool should have every connection back')
+            } catch (e) {
+              return conn.close().then(() => done(e), () => done(e))
+            }
+            conn.close().then(() => done(), done)
+          })
+        })
+        req.query('select * from notexistingtable')
+      }).catch(done)
+    },
+
+    'streaming a request timeout' (done, driver, message) {
+      // a timeout ends the batch on an error like any other, which is when a stream is most
+      // likely to be left waiting for a 'done' that never comes
+      const config = readConfig()
+      config.driver = driver
+      config.requestTimeout = 1000 // note: msnodesqlv8 doesn't support timeouts less than 1 second
+
+      new sql.ConnectionPool(config).connect().then(conn => {
+        const errors = []
+        let completed = 0
+
+        const req = new sql.Request(conn)
+        req.stream = true
+        req.on('error', err => errors.push(err))
+        req.on('done', () => {
+          if (completed++) return
+          setImmediate(() => {
+            try {
+              assert.strictEqual(completed, 1, 'the stream should complete once')
+              assert.strictEqual(errors.length, 1, 'the timeout should be streamed once')
+              assert.ok(message ? message.test(errors[0].message) : errors[0] instanceof sql.RequestError,
+                `the error should be the timeout, got ${errors[0].constructor.name}: ${errors[0].message}`)
+              assert.strictEqual(conn.pool.numUsed(), 0, 'the pool should have every connection back')
+            } catch (e) {
+              return conn.close().then(() => done(e), () => done(e))
+            }
+            conn.close().then(() => done(), done)
+          })
+        })
+        req.query('waitfor delay \'00:00:05\';select 1')
+      }).catch(done)
+    },
+
+    'pausing and resuming a stream once it is over' (ending, done) {
+      // msnodesqlv8 does not check that a query is still running when it is paused or resumed,
+      // and can put one that has ended back on its connection's work queue, where it holds up
+      // whatever that connection runs next. With one connection in the pool, the query that
+      // follows the stream runs on the connection the stream had
+      const config = readConfig()
+      config.driver = driver
+      config.pool = { max: 1, min: 0 }
+      const queries = {
+        completed: 'select 1 as a union all select 2',
+        failed: 'select * from notexistingtable',
+        cancelled: 'select * from streaming'
+      }
+      const endings = { completed: [], failed: ['EREQUEST'], cancelled: ['ECANCEL'] }
+      let timer
+      let held = false
+
+      const pool = new sql.ConnectionPool(config)
+      pool.connect().then(() => new Promise(resolve => {
+        const errors = []
+        let rows = 0
+        const req = new sql.Request(pool)
+        req.stream = true
+        // as a stream that is still being read when the error arrives resumes the request
+        req.on('error', err => {
+          errors.push(err.code)
+          req.pause()
+          req.resume()
+        })
+        req.on('row', () => {
+          if (ending === 'cancelled' && ++rows === 10) {
+            req.pause()
+            req.cancel()
+          }
+        })
+        req.once('done', () => {
+          req.pause()
+          req.resume()
+          // and again once the driver has released the statement: msnodesqlv8 can take a query
+          // that is put back on the queue before then off it again as it releases the statement.
+          // Were the release to take longer, the test would only stop telling the bug apart
+          setTimeout(() => {
+            req.pause()
+            req.resume()
+            resolve(errors)
+          }, 200)
+        })
+        req.query(queries[ending])
+      })).then(errors => {
+        assert.deepStrictEqual(errors, endings[ending], `the stream should end as ${ending}`)
+        return new Promise((resolve, reject) => {
+          timer = setTimeout(() => {
+            held = true
+            reject(new Error('the next query on the connection should run'))
+          }, 5000)
+          pool.request().query('select 1 as a').then(resolve, reject)
+        })
+      }).then(result => {
+        clearTimeout(timer)
+        assert.deepStrictEqual(result.recordset, [{ a: 1 }], 'the next query should return its result')
+        return pool.close()
+      }).then(() => done(), err => {
+        clearTimeout(timer)
+        // a connection that is held up cannot be closed either, so the pool is left as it is
+        if (held) return done(err)
+        pool.close().then(() => done(err), () => done(err))
+      })
+    },
+
     'new Table' (done) {
       const tvp = new MSSQLTestType()
       tvp.rows.add('asdf', 15)