diff --git a/lib/msnodesqlv8/request.js b/lib/msnodesqlv8/request.js index c2a6ae71..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)) { @@ -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) }) }) } diff --git a/test/common/tests.js b/test/common/tests.js index eb542ee9..e59c73b6 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) => { @@ -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) diff --git a/test/common/unit.js b/test/common/unit.js index 3ddfe670..e2146add 100644 --- a/test/common/unit.js +++ b/test/common/unit.js @@ -3001,4 +3001,556 @@ 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') + }) + + // 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 e67f85f4..da3adf46 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,8 +305,13 @@ 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)) + 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 68bdefef..b6282879 100644 --- a/test/tedious/tedious.js +++ b/test/tedious/tedious.js @@ -334,8 +334,13 @@ 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)) + 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)) })