fix: stop msnodesqlv8 streams failing when resumed while their query executes

.github/workflows/nodejs.yml

@@ -162,7 +162,9 @@ jobs:
       # a combo without a prebuild falls back to a node-gyp source build, which is
       # brittle on the GitHub Windows images (e.g. the VS 2026 rollout on
       # windows-2025 that node-gyp cannot detect). Coverage:
-      #   v2: Node 18    v3: Node 18-20    v4: Node 18-24    v5: Node 20-24
+      #   v2: Node 18    v3: Node 18-20    v4: Node 18-24    v5: Node 20-24    v6: Node 22-24
+      #
+      # msnodesqlv8 6 requires Node 22 or later.
       #
       # v5 is split: msnodesqlv8 5.2.2 dropped the Node 20 (ABI 115) win32-x64
       # prebuild, so `@^5` is gated to Node 22-24 and Node 20 is pinned to 5.2.1,
@@ -201,6 +203,12 @@ jobs:
       - name: Run msnodesqlv8 v5 tests
         if: ${{ matrix.node == '22.x' || matrix.node == '24.x' }}
         run: npm run test-msnodesqlv8
+      - name: Install msnodesqlv8 v6
+        if: ${{ matrix.node == '22.x' || matrix.node == '24.x' }}
+        run: npm install --no-save msnodesqlv8@^6
+      - name: Run msnodesqlv8 v6 tests
+        if: ${{ matrix.node == '22.x' || matrix.node == '24.x' }}
+        run: npm run test-msnodesqlv8
   release:
     name: Release
     concurrency: release

README.md

@@ -1281,7 +1281,11 @@ request.cancel()
 
 A cancelled stream emits an `error` event with code `ECANCEL`, followed by `done`.
 
-With the `msnodesqlv8` driver, a request that is already running can only be cancelled from msnodesqlv8 5.6. Earlier versions can cancel a paused stream, but let any other running request run to the end and complete as if `cancel()` had not been called. A paused stream is cancelled once the driver stops reading, at the end of a batch of 50 rows or the start of the next result set; a query that ends first, or times out, completes as cancelled.
+With the `msnodesqlv8` driver, a request that is already running can only be cancelled from msnodesqlv8 5.6. Earlier versions can cancel a paused stream, but let any other running request run to the end and complete as if `cancel()` had not been called. A paused stream is cancelled once the driver stops reading, at the end of a batch of 50 rows or the start of the next result set.
+
+Before msnodesqlv8 6, that includes a stream paused before its query reaches its first result set, as `toReadableStream()` leaves one until it is read. Cancelling it while msnodesqlv8 is still working through the statements ahead of that result set can crash the process, so the cancel waits for them: they run to the end and keep their effects, however long they take. The request timeout ends the wait if it is at least a second. With `requestTimeout: 0`, or anything under a second, which msnodesqlv8 takes as no timeout, nothing ends it. msnodesqlv8 6 cancels such a stream at once.
+
+A query that ends before the cancel takes effect, or times out, completes as cancelled, after emitting any error it ended with.
 
 ## Transaction
 

lib/msnodesqlv8/request.js

@@ -10,6 +10,18 @@ const { PARSERS: UDT } = require('../udt')
 const Table = require('../table')
 const { valueHandler } = require('../shared')
 
+// msnodesqlv8 6 emits nothing for a paused query until it is resumed, 'meta' and 'done' included,
+// and ends a paused query it cancels once nothing is under way, without freeing its statement
+// first. What the request does for earlier versions while a query is held back would wait for
+// those events forever
+const driverHoldsPausedQueries = () => {
+  try {
+    return Number(require('msnodesqlv8/package.json').version.split('.')[0]) >= 6
+  } catch (err) {
+    return false
+  }
+}
+
 const JSON_COLUMN_ID = 'JSON_F52E2B61-18A1-11d1-B105-00805F49916B'
 const XML_COLUMN_ID = 'XML_F52E2B61-18A1-11d1-B105-00805F49916B'
 const EMPTY_BUFFER = Buffer.alloc(0)
@@ -403,6 +415,8 @@ class Request extends BaseRequest {
       let driverPaused = false
       // a cancel of a paused stream waits for the driver to reach a point where it can be paused
       let cancelDeferred = false
+      // the cancel waits for a query held back before its first result set
+      let cancelBeforeResults = false
       // msnodesqlv8 has freed the statement, which it does without 'done' after an error ends the query
       let freed = false
       let resuming = false
@@ -626,13 +640,18 @@ class Request extends BaseRequest {
         const pauseAndCancelDriver = () => {
           cancelDeferred = false
           // msnodesqlv8 5.0 to 5.5 send a query one cancel, which a cancel made before the query
-          // started has used up and they refused, so the query runs on
+          // started has used up and they refused, so the query runs on. A query held back before its
+          // first result set was paused when that cancel went out, as it is when msnodesqlv8 cancels
+          // it itself on a timeout, so there the cancel went through and the query is ending
           if (typeof req.getCancelSent === 'function' && req.getCancelSent()) {
+            if (cancelBeforeResults) return true
             cancelRequested = false
             return false
           }
-          driverPaused = true
-          req.pauseQuery()
+          if (!driverPaused) {
+            driverPaused = true
+            req.pauseQuery()
+          }
           cancelDriver()
           return true
         }
@@ -648,16 +667,27 @@ 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
+        // or before its query starts, holds it back at once. But msnodesqlv8 < 6's resume() fetches
+        // at once, whatever the query is doing, and once the query has started ('submitted') it may
+        // still be executing, or moving on through statements with no result, when the fetch fails
+        // with a 'Function sequence error' or the result set is fetched twice. So a query held back
+        // then is resumed only once it has stopped at a result set, a tick after its 'meta', and not
+        // at all if it ends first. Until msnodesqlv8 starts it, a resume only stops it starting paused.
+        // msnodesqlv8 6 resumes only what it has held back, so there it is resumed at once
+        let heldBeforeResults = false
+        let submitted = false
+        let atResultSet = false
+        req.once('submitted', () => { submitted = true })
         this._pauseStream = () => {
           if (!columns && !driverPaused && !finished && reachable()) {
             driverPaused = true
+            heldBeforeResults = true
             req.pauseQuery()
           }
         }
         if (this.stream && this.paused) this._pauseStream()
 
-        this._resumeStream = () => {
+        const resumeStream = () => {
           // a listener that resumes the stream while it is emitting what was waiting leaves the
           // rest to this loop, so that every listener gets the events in order
           if (resuming) return
@@ -673,11 +703,13 @@ 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 && reachable()) {
+          if (!pending.length && !this.paused && driverPaused && !finished && !cancelRequested && reachable() &&
+            (!heldBeforeResults || !submitted || atResultSet || driverHoldsPausedQueries())) {
             driverPaused = false
             req.resumeQuery()
           }
         }
+        this._resumeStream = resumeStream
 
         const cancel = () => {
           debug('request(%d): cancel', IDS.get(this))
@@ -696,6 +728,18 @@ class Request extends BaseRequest {
           }
           cancelRequested = true
           cancelAttempts++
+          // A query held back before its first result set still runs the statements ahead of it,
+          // which have none. msnodesqlv8 < 6 cancels a paused query by freeing its statement (5.6 not
+          // while the first statement is still executing), and a free that lands while msnodesqlv8
+          // reads on to the next result can crash the process. Nothing tells the request when the
+          // query moves on, so a held query is cancelled once it stops at its first result set, and
+          // one that ends first is reported as cancelled. The statements ahead run to their end, or
+          // to the request timeout. msnodesqlv8 6 cancels it at once.
+          if (driverPaused && !columns && !driverHoldsPausedQueries()) {
+            cancelDeferred = true
+            cancelBeforeResults = true
+            return
+          }
           if (!this.stream || !this.paused || driverPaused) {
             return cancelDriver()
           }
@@ -725,6 +769,14 @@ class Request extends BaseRequest {
             // later, once it has stopped at the result set
             if (!this.paused && driverPaused && !cancelRequested) setImmediate(this._resumeStream)
           }
+          // msnodesqlv8 calls dispatch() once 'meta' is handled, which returns while the query is
+          // paused, so a tick later a query held back before its first result set has stopped there
+          if (heldBeforeResults && !atResultSet) {
+            setImmediate(() => {
+              atResultSet = true
+              resumeStream()
+            })
+          }
 
           if (row) {
             if (isChunkedRecordset) {

test/common/tests.js

@@ -3219,6 +3219,136 @@ module.exports = (sql, driver) => {
       }).catch(done)
     },
 
+    'cancelling a stream held back while its query moves on to its first result set' (done) {
+      // a stream paused before its query runs, as toReadableStream() leaves it, and cancelled once
+      // msnodesqlv8 has moved on from a statement with no result set: the server has sent that
+      // statement's row count, so msnodesqlv8 is reading on when the cancel arrives, and freeing
+      // the statement then can crash the process. 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 }
+      let timer
+      let held = false
+
+      const pool = new sql.ConnectionPool(config)
+      pool.connect().then(() => new Promise((resolve, reject) => {
+        const errors = []
+        const rows = []
+        const req = new sql.Request(pool)
+        req.stream = true
+        req.on('error', err => errors.push(err.code))
+        req.on('row', row => rows.push(row))
+        timer = setTimeout(() => {
+          held = true
+          reject(new Error('the cancelled stream should complete'))
+        }, 10000)
+        req.once('done', () => {
+          clearTimeout(timer)
+          resolve({ errors, rows })
+        })
+        req.pause()
+        req.query('set nocount off; declare @t table (a int); insert @t values (1); raiserror(\'flush\', 0, 1) with nowait; waitfor delay \'00:00:02\'; select 1 as a')
+        setTimeout(() => req.cancel(), 500)
+      })).then(({ errors, rows }) => {
+        assert.deepStrictEqual(errors, ['ECANCEL'], 'the stream should end as cancelled')
+        assert.deepStrictEqual(rows, [], 'no row should be streamed')
+        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))
+      })
+    },
+
+    'resuming a stream held back while its query executes' (failing, done) {
+      // a stream paused before its query runs, as toReadableStream() leaves it, and read while
+      // the query still executes: msnodesqlv8 fetches from the query at once, which can fail with a
+      // 'Function sequence error' (HY010), lose the error the query raises, or crash the process
+      const rows = []
+      const errors = []
+      const req = new sql.Request()
+      req.stream = true
+      req.on('row', row => rows.push(row))
+      req.on('error', err => errors.push(err))
+      req.once('done', () => {
+        try {
+          if (failing) {
+            assert.deepStrictEqual(errors.map(e => e.number), [208], `only the error the query raised should be streamed, not: ${errors.map(e => e.message).join(' / ')}`)
+            assert.deepStrictEqual(rows, [], 'no row should be streamed')
+          } else {
+            assert.deepStrictEqual(errors.map(e => e.message), [], 'no error should be streamed')
+            assert.deepStrictEqual(rows, [{ a: 1 }], 'the row should be streamed once')
+          }
+          done()
+        } catch (e) {
+          done(e)
+        }
+      })
+      req.pause()
+      req.query(failing
+        ? 'waitfor delay \'00:00:01\'; select * from notexistingtable'
+        : 'waitfor delay \'00:00:01\'; select 1 as a')
+      setTimeout(() => req.resume(), 300)
+    },
+
+    'cancelling a stream held back while its query executes' (done) {
+      // a stream paused before its query runs, as toReadableStream() leaves it, read while the query
+      // executes and then cancelled: the query stays held back until its first result set, and is
+      // cancelled there. 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 }
+      let timer
+      let held = false
+
+      const pool = new sql.ConnectionPool(config)
+      pool.connect().then(() => new Promise(resolve => {
+        const errors = []
+        const rows = []
+        const req = new sql.Request(pool)
+        req.stream = true
+        req.on('error', err => errors.push(err.code))
+        req.on('row', row => rows.push(row))
+        req.once('done', () => resolve({ errors, rows }))
+        req.pause()
+        req.query('waitfor delay \'00:00:01\'; select 1 as a')
+        setTimeout(() => req.resume(), 200)
+        setTimeout(() => req.cancel(), 400)
+      })).then(({ errors, rows }) => {
+        assert.deepStrictEqual(errors, ['ECANCEL'], 'the stream should end as cancelled')
+        assert.deepStrictEqual(rows, [], 'no row should be streamed')
+        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))
+      })
+    },
+
     '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