From 6c85f6fb91bfa04b1b4985a3f8eaba18d3b89bef Mon Sep 17 00:00:00 2001 From: Dan Hensby Date: Mon, 12 Oct 2026 16:19:46 +0100 Subject: [PATCH 1/3] test: read the test config afresh for each test readConfig() returned the object that require() caches, so a test that changed its config changed it for every test after it. The login failed test sets a user that does not exist, which leaves any later test that opens its own pool from readConfig() unable to log in, and the request timeout tests leave a 1 second timeout behind. Each call now parses the file again, so a test's changes stay with that test. Co-Authored-By: Claude Opus 5.5 --- test/common/tests.js | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/test/common/tests.js b/test/common/tests.js index eb542ee9..8c2df53b 100644 --- a/test/common/tests.js +++ b/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) => { From adbb02832a733e5bdd029b0a017d14c90a51cb91 Mon Sep 17 00:00:00 2001 From: Dan Hensby Date: Mon, 12 Oct 2026 16:20:07 +0100 Subject: [PATCH 2/3] fix: complete msnodesqlv8 streams when an error ends the batch msnodesqlv8 emits 'done' when a batch completes, but not when an error ends it: it emits the error, with no more to follow, then 'free' once it has released the statement. Every version from 2.x to 5.6 does this. A request that is not streamed settles on that error. A streamed request emits each error as it arrives and only completed on 'done', so after a batch-ending error (a failed statement, a conversion error, or a request timeout) it never emitted 'done', never called back or settled its promise, and never released its connection. Enough of them would leave the pool with no connections to lend and close() would never resolve. The request now completes on whichever of 'done' and 'free' comes first; msnodesqlv8's own query aggregator settles a query on 'free' too. A stream emits the error and then 'done', as it does with tedious, and the connection is released once. Co-Authored-By: Claude Opus 5.5 --- lib/msnodesqlv8/request.js | 16 ++++-- test/common/tests.js | 66 +++++++++++++++++++++++++ test/common/unit.js | 86 +++++++++++++++++++++++++++++++++ test/msnodesqlv8/msnodesqlv8.js | 14 ++---- test/tedious/tedious.js | 2 + 5 files changed, 170 insertions(+), 14 deletions(-) diff --git a/lib/msnodesqlv8/request.js b/lib/msnodesqlv8/request.js index c2a6ae71..dfe177a8 100644 --- a/lib/msnodesqlv8/request.js +++ b/lib/msnodesqlv8/request.js @@ -349,8 +349,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 @@ -844,7 +845,11 @@ 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 if (hasReturned) { @@ -973,7 +978,10 @@ class Request extends BaseRequest { delete this._cancel callback(null, recordsets, output, rowsAffected, recordsetcolumns) } - }) + } + + req.once('done', complete) + req.once('free', complete) }) }) } diff --git a/test/common/tests.js b/test/common/tests.js index 8c2df53b..2014fc3c 100644 --- a/test/common/tests.js +++ b/test/common/tests.js @@ -2985,6 +2985,72 @@ 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) + }, + 'new Table' (done) { const tvp = new MSSQLTestType() tvp.rows.add('asdf', 15) diff --git a/test/common/unit.js b/test/common/unit.js index 3ddfe670..f7a7f3a8 100644 --- a/test/common/unit.js +++ b/test/common/unit.js @@ -3001,4 +3001,90 @@ describe('msnodesqlv8 query events', () => { assert.deepStrictEqual(calls, [], 'the driver should not be paused') }) }) + + // the events a streamed request emits, in order, up to a tick after `run` settles, so that + // anything it emits after it has completed is caught too + const streamed = async (request, run) => { + const events = [] + request.stream = true + request.on('row', row => events.push(['row', row])) + request.on('error', err => events.push(['error', err.number])) + request.on('done', () => events.push(['done'])) + await run(request) + await tick() + return events + } + + // every version: a batch that completes emits 'done', then 'free' once its statement is released + it('should resolve once when the batch completes', async () => { + const { request, pool } = setup(notify => { + result('a', 1)(notify) + notify.emit('done') + notify.emit('free') + }) + const { recordset } = await request.query('select 1 as a') + assert.deepStrictEqual(recordset, [{ a: 1 }], 'the row should be returned') + await tick() + assert.strictEqual(pool.released, 1, 'the connection should be released once') + }) + + it('should complete a stream once when the batch completes', async () => { + const { request, pool } = setup(notify => { + result('a', 1)(notify) + notify.emit('done') + notify.emit('free') + }) + const events = await streamed(request, r => r.query('select 1 as a')) + assert.deepStrictEqual(events, [['row', { a: 1 }], ['done']], 'the row should be streamed, then done once') + assert.strictEqual(pool.released, 1, 'the connection should be released once') + }) + + // every version: an error that ends the batch is emitted with more=false, and no 'done' + // follows it, only 'free' once the statement is released + it('should reject once when an error ends the batch and the statement is released', async () => { + const { request, pool } = setup(notify => { + notify.fail(PRIMARY_KEY(), false) + notify.emit('free') + }) + await assert.rejects(request.query('insert into t values (1)'), err => { + assert.ok(err instanceof RequestError, 'the error should be a RequestError') + assert.strictEqual(err.number, 2627, 'the error should be the one that ended the batch') + return true + }) + await tick() + assert.strictEqual(pool.released, 1, 'the connection should be released once') + }) + + it('should stream an error that ends the batch, then complete', async () => { + const { request, pool } = setup(notify => { + result('a', 1)(notify) + notify.fail(PRIMARY_KEY(), false) + notify.emit('free') + }) + const events = await streamed(request, r => r.query('select 1 as a; insert into t values (1)')) + assert.deepStrictEqual(events, [['row', { a: 1 }], ['error', 2627], ['done']], + 'the row and the error should be streamed, then done once') + assert.strictEqual(pool.released, 1, 'the connection should be released once') + }) + + it('should stream an error that ends the batch of an execute, then complete', async () => { + const { request, pool } = setup(notify => { + notify.fail(PRIMARY_KEY(), false) + notify.emit('free') + }) + const events = await streamed(request, r => r.execute('p')) + assert.deepStrictEqual(events, [['error', 2627], ['done']], 'the error should be streamed, then done once') + assert.strictEqual(pool.released, 1, 'the connection should be released once') + }) + + it('should stream a timeout, then complete', async () => { + // the driver's query timeout ends the batch on an error like any other + const { request, pool } = setup(notify => { + notify.fail(odbcError('Query timeout expired', 0, 'HYT00'), false) + notify.emit('free') + }) + const events = await streamed(request, r => r.query('waitfor delay \'00:00:05\'')) + assert.deepStrictEqual(events, [['error', 0], ['done']], 'the timeout should be streamed, then done once') + assert.strictEqual(pool.released, 1, 'the connection should be released once') + }) }) diff --git a/test/msnodesqlv8/msnodesqlv8.js b/test/msnodesqlv8/msnodesqlv8.js index e67f85f4..476176b3 100644 --- a/test/msnodesqlv8/msnodesqlv8.js +++ b/test/msnodesqlv8/msnodesqlv8.js @@ -20,10 +20,6 @@ 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) @@ -119,12 +115,8 @@ describe('msnodesqlv8', function () { 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 back pressure across partial row batches', done => TESTS['query with pipe and back pressure across partial row batches'](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 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)) @@ -313,6 +305,8 @@ describe('msnodesqlv8', function () { it('streaming rowsaffected in stored procedure', done => TESTS['streaming rowsaffected in stored procedure'](done)) it('streaming trailing rows', done => TESTS['streaming trailing rows'](done)) it('streaming with duplicate output column names', done => TESTS['streaming with duplicate output column names'](done)) + it('streaming an error that ends the batch', done => TESTS['streaming an error that ends the batch'](done)) + it('streaming a request timeout', done => TESTS['streaming a request timeout'](done)) it('a cancelled stream emits done event', done => TESTS['a cancelled stream emits done event'](done)) it('a cancelled paused stream emits done event', done => TESTS['a cancelled paused stream emits done event'](done)) diff --git a/test/tedious/tedious.js b/test/tedious/tedious.js index 68bdefef..95cda533 100644 --- a/test/tedious/tedious.js +++ b/test/tedious/tedious.js @@ -334,6 +334,8 @@ describe('tedious', () => { it('streaming rowsaffected in stored procedure', done => TESTS['streaming rowsaffected in stored procedure'](done)) it('streaming trailing rows', done => TESTS['streaming trailing rows'](done)) it('streaming with duplicate output column names', done => TESTS['streaming with duplicate output column names'](done)) + it('streaming an error that ends the batch', done => TESTS['streaming an error that ends the batch'](done)) + it('streaming a request timeout', done => TESTS['streaming a request timeout'](done, 'tedious', /Timeout: Request failed to complete in 1000ms/)) it('a cancelled stream emits done event', done => TESTS['a cancelled stream emits done event'](done)) it('a cancelled paused stream emits done event', done => TESTS['a cancelled paused stream emits done event'](done)) From 081483fcc5c2d60c7534a7537554f557e89ddfaf Mon Sep 17 00:00:00 2001 From: Dan Hensby Date: Wed, 14 Oct 2026 09:12:56 +0100 Subject: [PATCH 3/3] fix: stop msnodesqlv8 streams resumed after they end from hanging the connection msnodesqlv8 does not check that a query is still running when it is paused, resumed or cancelled. Pausing and then resuming a query that has ended, or resuming one that ended while it was paused, puts it back at the head of its connection's work queue, where it holds up every query the connection is sent after it, so a pooled connection hangs on the next request it is lent to. From msnodesqlv8 5.1.9 resuming such a query also has it ask for its next result again and release its statement a second time, which can loop emitting rowcount events or crash the process. A stream stops pausing and resuming its query once msnodesqlv8 reports 'done' or 'free', but until then it still could after the query had ended: while the errors that end its batch were emitted, which msnodesqlv8 >= 5.1 does before it releases the statement; between such an error and 'free', as a stream paused before its first result set is when it is read on after the error; and in the few ticks between the query's reader stopping and msnodesqlv8 reporting the end. A cancel in those windows reached the ended query too. The request now leaves the query alone once msnodesqlv8 reports that it has ended, on an error that ends the batch as well as on 'done' and 'free', and takes it up again if msnodesqlv8 goes on to a result set after an error it said had nothing more to follow. A paused stream that is cancelled then still completes as cancelled. While an error that is not the last of its set is handled the query is set aside, and it is paused afterwards if the stream was paused before its first result set, unless it is being cancelled. From msnodesqlv8 3 the request also leaves the query alone once its reader has stopped. Co-Authored-By: Claude Opus 5.5 --- lib/msnodesqlv8/request.js | 70 ++++- test/common/tests.js | 69 +++++ test/common/unit.js | 466 ++++++++++++++++++++++++++++++++ test/msnodesqlv8/msnodesqlv8.js | 3 + test/tedious/tedious.js | 3 + 5 files changed, 608 insertions(+), 3 deletions(-) diff --git a/lib/msnodesqlv8/request.js b/lib/msnodesqlv8/request.js index dfe177a8..cfe5817c 100644 --- a/lib/msnodesqlv8/request.js +++ b/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)) { @@ -533,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) { @@ -584,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() } @@ -607,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())) { @@ -634,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) { @@ -813,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) @@ -832,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) @@ -851,6 +913,8 @@ class Request extends BaseRequest { // 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 diff --git a/test/common/tests.js b/test/common/tests.js index 2014fc3c..e59c73b6 100644 --- a/test/common/tests.js +++ b/test/common/tests.js @@ -3051,6 +3051,75 @@ module.exports = (sql, driver) => { }).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) diff --git a/test/common/unit.js b/test/common/unit.js index f7a7f3a8..e2146add 100644 --- a/test/common/unit.js +++ b/test/common/unit.js @@ -3087,4 +3087,470 @@ describe('msnodesqlv8 query events', () => { assert.deepStrictEqual(events, [['error', 0], ['done']], 'the timeout should be streamed, then done once') assert.strictEqual(pool.released, 1, 'the connection should be released once') }) + + // every version: pausing and then resuming a query that has ended, or resuming one that + // ended while it was paused, puts it back at the head of the connection's work queue, where it + // holds up the next query, so `calls` records the pauses, resumes and cancels that reach the query + const pausable = (calls, query = {}) => { + let paused = false + return Object.assign({ + isPaused: () => paused, + pauseQuery () { + paused = true + calls.push('pause') + }, + resumeQuery () { + paused = false + calls.push('resume') + }, + cancelQuery () { + calls.push('cancel') + } + }, query) + } + + const completion = (request, run) => new Promise(resolve => { + request.once('done', resolve) + run(request) + }) + + // a result set of `count` rows of one column + const rows = (name, count) => notify => { + notify.emit('meta', [{ name, sqlType: 'int', size: 4, nullable: false }]) + for (let i = 0; i < count; i++) { + notify.emit('row', i) + notify.emit('column', 0, i, false) + } + } + + it('should stop pausing and resuming a query once the batch completes', async () => { + const calls = [] + const { request } = setup(async notify => { + rows('a', 50)(notify) + await tick() + request.resume() + notify.emit('done') + notify.emit('free') + }, pausable(calls)) + request.stream = true + request.on('row', row => { + if (row.a === 49) request.pause() + }) + await completion(request, r => r.query('select a from t')) + request.pause() + request.resume() + assert.deepStrictEqual(calls, ['pause', 'resume'], 'only the pause at the end of the batch and its resume should reach the query') + }) + + it('should stop pausing and resuming a query once an error ends the batch', async () => { + const calls = [] + const { request } = setup(notify => { + notify.fail(PRIMARY_KEY(), false) + setImmediate(() => notify.emit('free')) + }, pausable(calls)) + request.stream = true + // as a stream that is still being read when the error arrives pauses and resumes the request, + // before msnodesqlv8 emits 'free' + request.on('error', () => { + request.pause() + request.resume() + setImmediate(() => { + request.pause() + request.resume() + }) + }) + await completion(request, r => r.query('insert into t values (1)')) + request.pause() + request.resume() + assert.deepStrictEqual(calls, [], 'no pause or resume after the error should reach the query') + }) + + // msnodesqlv8 >= 5.1 emits the errors that end a batch together, before it releases the statement + it('should not pause or resume a query while the errors that end its batch are emitted', async () => { + const calls = [] + const { request } = setup(notify => { + notify.fail(odbcError('There is already an object named \'t\' in the database.', 2714, '42S01'), true) + notify.fail(odbcError('Could not create constraint or index. See previous errors.', 1750, '42000'), false) + setImmediate(() => notify.emit('free')) + }, pausable(calls)) + request.stream = true + request.on('error', () => { + request.pause() + request.resume() + setImmediate(() => { + request.pause() + request.resume() + }) + }) + await completion(request, r => r.query('create table t (id int constraint pk primary key)')) + await tick() + assert.deepStrictEqual(calls, [], 'no pause or resume while the batch was ending, or after, should reach the query') + }) + + // msnodesqlv8 >= 5.6 + it('should pause a query on an error that does not end the batch once the error is handled', async () => { + const calls = [] + let paused + const { request } = setup(async notify => { + notify.fail(odbcError('Something went wrong', 50000, '42000'), true) + await tick() + paused = calls.slice() + request.resume() + rows('a', 1)(notify) + notify.emit('done') + notify.emit('free') + }, pausable(calls)) + request.stream = true + const received = [] + request.on('error', () => request.pause()) + request.on('row', row => received.push(row)) + await completion(request, r => r.query('raiserror(N\'Something went wrong\', 16, 1); select 0 as a')) + assert.deepStrictEqual(paused, ['pause'], 'a pause on an error that does not end the batch should reach the query') + assert.deepStrictEqual(calls, ['pause', 'resume'], 'the resume after it should reach the query too') + assert.deepStrictEqual(received, [{ a: 0 }], 'the rows after the error should be streamed') + }) + + // msnodesqlv8 >= 5.1 emits the errors that end a batch together, before it releases the statement + it('should not cancel a query while the errors that end its batch are emitted', async () => { + const calls = [] + const { request } = setup(notify => { + notify.fail(odbcError('There is already an object named \'t\' in the database.', 2714, '42S01'), true) + notify.fail(odbcError('Could not create constraint or index. See previous errors.', 1750, '42000'), false) + setImmediate(() => notify.emit('free')) + }, pausable(calls)) + request.stream = true + const errors = [] + request.on('error', err => { + errors.push(err.code) + if (errors.length === 1) request.cancel() + }) + await completion(request, r => r.query('create table t (id int constraint pk primary key)')) + assert.deepStrictEqual(errors, ['EREQUEST', 'EREQUEST'], 'only the errors should be streamed') + assert.deepStrictEqual(calls, [], 'the cancel should not reach the query, which had ended') + }) + + // msnodesqlv8 >= 5.6 + it('should cancel a query on an error that does not end the batch once the error is handled', async () => { + const calls = [] + let onError + const { request } = setup(notify => { + notify.fail(odbcError('Something went wrong', 50000, '42000'), true) + }, pausable(calls, { + cancelQuery (callback) { + calls.push('cancel') + setImmediate(() => { + this.emit('done') + callback(null) + setImmediate(() => this.emit('free')) + }) + } + })) + request.stream = true + const errors = [] + request.on('error', err => { + errors.push(err.code) + if (errors.length === 1) { + request.cancel() + onError = calls.slice() + } + }) + await completion(request, r => r.query('raiserror(N\'Something went wrong\', 16, 1); waitfor delay \'00:00:05\'')) + await tick() + assert.deepStrictEqual(onError, [], 'the cancel should wait until the error has been handled') + assert.deepStrictEqual(calls, ['cancel'], 'the cancel should then reach the query, which carried on') + assert.deepStrictEqual(errors, ['EREQUEST', 'ECANCEL'], 'the error and then the cancel should be streamed') + }) + + // msnodesqlv8 >= 5.6: a paused stream is cancelled when the query next stops, here at the start + // of the next result set + it('should leave a query that is cancelled on an error that does not end the batch to the cancel', async () => { + const calls = [] + const { request } = setup(async notify => { + notify.fail(odbcError('Something went wrong', 50000, '42000'), true) + await tick() + rows('a', 1)(notify) + }, pausable(calls, { + cancelQuery (callback) { + calls.push('cancel') + setImmediate(() => { + this.emit('done') + callback(null) + setImmediate(() => this.emit('free')) + }) + } + })) + request.stream = true + const errors = [] + request.on('error', err => { + errors.push(err.code) + if (errors.length === 1) { + request.pause() + request.cancel() + } + }) + await completion(request, r => r.query('raiserror(N\'Something went wrong\', 16, 1); select 0 as a')) + await tick() + assert.deepStrictEqual(errors, ['EREQUEST', 'ECANCEL'], 'the error and then the cancel should be streamed') + assert.deepStrictEqual(calls, ['pause', 'cancel'], 'the query should be paused only to be cancelled') + }) + + // msnodesqlv8 >= 3 stops the query's reader a few ticks before it emits 'done', and from 5.1.9 + // resuming a stopped reader releases its statement again + it('should not resume a query whose reader has stopped', async () => { + const calls = [] + const reader = { running: true } + const { request } = setup(async notify => { + notify.emit('rowcount', 1) + reader.running = false + await tick() + request.resume() + notify.emit('done') + notify.emit('free') + }, pausable(calls, { queryWorker: reader })) + request.stream = true + // as toReadableStream() pauses a request before it runs it + request.pause() + await completion(request, r => r.query('update t set a = 1')) + assert.deepStrictEqual(calls, ['pause'], 'only the pause while the reader ran should reach the query') + }) + + it('should not pause a query whose reader has stopped', async () => { + const calls = [] + const reader = { running: true } + const { request } = setup(async notify => { + notify.emit('rowcount', 1) + reader.running = false + await tick() + request.pause() + request.resume() + notify.emit('done') + notify.emit('free') + }, pausable(calls, { queryWorker: reader })) + request.stream = true + await completion(request, r => r.query('update t set a = 1')) + assert.deepStrictEqual(calls, [], 'no pause or resume after the reader stopped should reach the query') + }) + + // msnodesqlv8 2 has no reader to check, so only the error tells the request that the query has ended + it('should not resume a paused query once an error has ended it', async () => { + const calls = [] + const { request } = setup(async notify => { + notify.fail(PRIMARY_KEY(), false) + await tick() + request.resume() + await tick() + notify.emit('free') + }, pausable(calls)) + request.stream = true + request.on('error', () => {}) + // as toReadableStream() pauses a request before it runs it + request.pause() + await completion(request, r => r.query('insert into t values (1)')) + assert.deepStrictEqual(calls, ['pause'], 'only the pause made before the query ran should reach it') + }) + + // msnodesqlv8 can give the last error of a statement that fails first more=false, and carry on + // to the statements after it + it('should resume a query that goes on to a result set after an error with nothing more to follow', async () => { + const calls = [] + let resumed + const { request } = setup(async notify => { + notify.fail(PRIMARY_KEY(), false) + notify.emit('meta', [{ name: 'a', sqlType: 'int', size: 4, nullable: false }]) + await tick() + request.resume() + resumed = calls.slice() + notify.emit('row', 0) + notify.emit('column', 0, 1, false) + notify.emit('done') + notify.emit('free') + }, pausable(calls)) + request.stream = true + const received = [] + request.on('error', () => {}) + request.on('row', row => received.push(row)) + // as toReadableStream() pauses a request before it runs it + request.pause() + await completion(request, r => r.query('insert into t values (1); select 1 as a')) + assert.deepStrictEqual(resumed, ['pause', 'resume'], 'the resume should reach the query, which carried on') + assert.deepStrictEqual(received, [{ a: 1 }], 'the rows after the error should be streamed') + }) + + it('should cancel a query that goes on to a result set after it was cancelled on an error with nothing more to follow', async () => { + const calls = [] + const { request } = setup(async notify => { + notify.fail(PRIMARY_KEY(), false) + notify.emit('meta', [{ name: 'a', sqlType: 'int', size: 4, nullable: false }]) + await tick() + notify.emit('done') + notify.emit('free') + }, pausable(calls, { + cancelQuery (callback) { + calls.push('cancel') + setImmediate(() => { + this.emit('done') + callback(null) + setImmediate(() => this.emit('free')) + }) + } + })) + request.stream = true + const errors = [] + request.on('error', err => { + errors.push(err.code) + if (errors.length === 1) request.cancel() + }) + await completion(request, r => r.query('insert into t values (1); select 1 as a')) + await tick() + assert.deepStrictEqual(calls, ['cancel'], 'the cancel should reach the query once it carries on') + assert.deepStrictEqual(errors, ['EREQUEST', 'ECANCEL'], 'the error and then the cancel should be streamed') + }) + + it('should not pause a query on an error that does not end the batch unless the stream was paused', async () => { + const calls = [] + const { request } = setup(async notify => { + notify.fail(odbcError('Something went wrong', 50000, '42000'), true) + await tick() + rows('a', 1)(notify) + notify.emit('done') + notify.emit('free') + }, pausable(calls)) + request.stream = true + request.on('error', () => {}) + await completion(request, r => r.query('raiserror(N\'Something went wrong\', 16, 1); select 0 as a')) + assert.deepStrictEqual(calls, [], 'the query should not be paused') + }) + + it('should complete a paused stream as cancelled when it is cancelled after an error ends the batch', async () => { + const calls = [] + const { request } = setup(notify => { + notify.fail(PRIMARY_KEY(), false) + setImmediate(() => notify.emit('free')) + }, pausable(calls)) + request.stream = true + const errors = [] + request.on('error', err => { + errors.push(err.code) + if (errors.length === 1) { + request.pause() + request.cancel() + } + }) + await completion(request, r => r.query('insert into t values (1)')) + assert.deepStrictEqual(errors, ['EREQUEST', 'ECANCEL'], 'the error and then the cancel should be streamed') + assert.deepStrictEqual(calls, [], 'the cancel should not reach the query, which had ended') + }) + + // msnodesqlv8 2.0 ends a statement's error with more=true and the message that the statement + // was terminated, and then nothing follows + it('should stop pausing and resuming a query once its statement is terminated', async () => { + const calls = [] + const { request } = setup(notify => { + notify.fail(PRIMARY_KEY(), true) + notify.emit('info', TERMINATED(), false) + }, pausable(calls)) + request.stream = true + await new Promise(resolve => { + request.once('error', () => setImmediate(resolve)) + request.query('insert into t values (1)') + }) + request.pause() + request.resume() + assert.deepStrictEqual(calls, [], 'no pause or resume after the statement was terminated should reach the query') + }) + + // msnodesqlv8 >= 2.4 gives the errors of a set more=true until the last, and a message that the + // statement was terminated between them settles the request before the last error arrives + it('should not pause or resume a query whose request was settled by the termination of its statement', async () => { + const calls = [] + const { request } = setup(notify => { + notify.fail(PRIMARY_KEY(), true) + notify.emit('info', TERMINATED(), false) + notify.fail(odbcError('Something went wrong', 50000, '42000'), false) + }, pausable(calls)) + await assert.rejects(request.query('insert into t values (1)'), err => { + assert.strictEqual(err.number, 2627, 'the error should be the one that terminated the statement') + return true + }) + // as toReadableStream() does with a request it is given to run again + request.stream = true + request.pause() + request.resume() + assert.deepStrictEqual(calls, [], 'no pause or resume after the request was settled should reach the query') + }) + + it('should not cancel a query whose reader has stopped', async () => { + const calls = [] + const reader = { running: true } + const { request } = setup(async notify => { + notify.emit('rowcount', 1) + reader.running = false + await tick() + request.cancel() + notify.emit('done') + notify.emit('free') + }, pausable(calls, { queryWorker: reader })) + request.stream = true + const errors = [] + request.on('error', err => errors.push(err.code)) + await completion(request, r => r.query('update t set a = 1')) + assert.deepStrictEqual(errors, [], 'the stream should complete without a cancel') + assert.deepStrictEqual(calls, [], 'the cancel should not reach the query, whose reader had stopped') + }) + + it('should complete a paused stream as cancelled when it is cancelled after its reader has stopped', async () => { + const calls = [] + const reader = { running: true } + const { request } = setup(async notify => { + notify.emit('rowcount', 1) + reader.running = false + await tick() + request.cancel() + notify.emit('done') + notify.emit('free') + }, pausable(calls, { queryWorker: reader })) + request.stream = true + const errors = [] + request.on('error', err => errors.push(err.code)) + // as toReadableStream() pauses a request before it runs it + request.pause() + await completion(request, r => r.query('update t set a = 1')) + assert.deepStrictEqual(errors, ['ECANCEL'], 'the cancel should be streamed') + assert.deepStrictEqual(calls, ['pause'], 'only the pause made before the query ran should reach it') + }) + + it('should not cancel a query once an error ends the batch', async () => { + const calls = [] + const { request } = setup(notify => { + notify.fail(PRIMARY_KEY(), false) + setImmediate(() => notify.emit('free')) + }, pausable(calls)) + request.stream = true + const errors = [] + request.on('error', err => { + errors.push(err.code) + request.cancel() + }) + await completion(request, r => r.query('insert into t values (1)')) + assert.deepStrictEqual(errors, ['EREQUEST'], 'only the error should be streamed') + assert.deepStrictEqual(calls, [], 'the cancel should not reach the query, which had ended') + }) + + it('should not cancel a query for the error it is completed with', async () => { + const calls = [] + const { request } = setup(notify => { + notify.emit('info', TERMINATED(), false) + notify.emit('done') + notify.emit('free') + }, pausable(calls)) + request.stream = true + const errors = [] + request.on('error', err => { + errors.push(err.number) + request.cancel() + }) + await completion(request, r => r.query('insert into t values (1)')) + assert.deepStrictEqual(errors, [3621], 'the termination should be streamed') + assert.deepStrictEqual(calls, [], 'the cancel should not reach the query, which had ended') + }) }) diff --git a/test/msnodesqlv8/msnodesqlv8.js b/test/msnodesqlv8/msnodesqlv8.js index 476176b3..da3adf46 100644 --- a/test/msnodesqlv8/msnodesqlv8.js +++ b/test/msnodesqlv8/msnodesqlv8.js @@ -309,6 +309,9 @@ describe('msnodesqlv8', function () { it('streaming a request timeout', done => TESTS['streaming a request timeout'](done)) it('a cancelled stream emits done event', done => TESTS['a cancelled stream emits done event'](done)) it('a cancelled paused stream emits done event', done => TESTS['a cancelled paused stream emits done event'](done)) + it('pausing and resuming a stream once it has completed', done => TESTS['pausing and resuming a stream once it is over']('completed', done)) + it('pausing and resuming a stream once an error has ended it', done => TESTS['pausing and resuming a stream once it is over']('failed', done)) + it('pausing and resuming a stream once it has been cancelled', done => TESTS['pausing and resuming a stream once it is over']('cancelled', done)) after(done => sql.close(done)) }) diff --git a/test/tedious/tedious.js b/test/tedious/tedious.js index 95cda533..b6282879 100644 --- a/test/tedious/tedious.js +++ b/test/tedious/tedious.js @@ -338,6 +338,9 @@ describe('tedious', () => { it('streaming a request timeout', done => TESTS['streaming a request timeout'](done, 'tedious', /Timeout: Request failed to complete in 1000ms/)) it('a cancelled stream emits done event', done => TESTS['a cancelled stream emits done event'](done)) it('a cancelled paused stream emits done event', done => TESTS['a cancelled paused stream emits done event'](done)) + it('pausing and resuming a stream once it has completed', done => TESTS['pausing and resuming a stream once it is over']('completed', done)) + it('pausing and resuming a stream once an error has ended it', done => TESTS['pausing and resuming a stream once it is over']('failed', done)) + it('pausing and resuming a stream once it has been cancelled', done => TESTS['pausing and resuming a stream once it is over']('cancelled', done)) after(done => sql.close(done)) })