From 061048ec3455611803c001d30c8966c8539c3263 Mon Sep 17 00:00:00 2001 From: Dan Hensby Date: Tue, 13 Oct 2026 11:00:16 +0100 Subject: [PATCH 1/2] fix: stop msnodesqlv8 streams losing rows under back pressure msnodesqlv8 fetches a result set's rows 50 at a time. Paused part way through a result set's last, short batch, it moves on without the rest of that batch. On the last result set it ends the query, and the rows still to come are never emitted. On an earlier one it starts the next result set, and on resume emits the rest of the batch under the next result set's columns; from 5.0 it then skips the next result set as well. 5.6 waits for the rest of a batch, but when paused on the only row of a result set's last batch it still skips the next result set. Back pressure from pipe() and toReadableStream() pauses a stream on almost every row, so a piped result set whose row count is not a multiple of 50 silently lost up to 49 rows. The driver is now paused only once it has handed over a whole batch, or before it has read anything. Rows that arrive in the meantime are held, along with anything else the stream emits after them, and emitted in order once the stream is resumed, and the stream completes once it has emitted them. Nothing is emitted while a stream is paused; before, a row that arrived while it was paused was dropped. msnodesqlv8 cancels a query cleanly only once it has stopped reading: paused or cancelled part way through a fetch, or as it moves to the next result set or ends the query, it can free the statement twice and break the queue of the connection. So a paused stream whose driver is still reading is cancelled when the driver next stops, at the end of a batch or the start of a result set, and a query that ends first, or times out, completes as cancelled. A paused stream that is cancelled drops the rows still waiting and reports ECANCEL, also when the query had already ended. Co-Authored-By: Claude Opus 5.5 --- README.md | 8 +- lib/msnodesqlv8/request.js | 221 ++++++++++-- test/common/tests.js | 33 ++ test/common/unit.js | 607 +++++++++++++++++++++++++++++++- test/msnodesqlv8/msnodesqlv8.js | 1 + test/tedious/tedious.js | 1 + 6 files changed, 839 insertions(+), 32 deletions(-) diff --git a/README.md b/README.md index ec8e955e..4ca77c87 100644 --- a/README.md +++ b/README.md @@ -430,6 +430,12 @@ function processRows() { } ``` +Nothing is emitted while the request is paused: `row`, `recordset`, `info`, `error`, `rowsaffected` and `done` + events wait until it is resumed. The msnodesqlv8 driver reads rows 50 at a time and only stops reading at the end + of a full batch, so the rows it reads before then are held until the request is resumed. A paused request that is + cancelled does not wait to be resumed: it drops the rows it holds, emits the other events it holds, and completes + with an `error` (`ECANCEL`) and `done` (see [cancel](#cancel)). + ## Connection Pools An important concept to understand when using this library is [Connection Pooling](https://en.wikipedia.org/wiki/Connection_pool) as this library uses connection pooling extensively. As one Node JS process is able to handle multiple requests at once, we can take advantage of this long running process to create a pool of database connections for reuse; this saves overhead of connecting to the database for each request @@ -1271,7 +1277,7 @@ 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. +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. ## Transaction diff --git a/lib/msnodesqlv8/request.js b/lib/msnodesqlv8/request.js index c6711c1f..c2a6ae71 100644 --- a/lib/msnodesqlv8/request.js +++ b/lib/msnodesqlv8/request.js @@ -14,6 +14,12 @@ 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) +// msnodesqlv8 fetches a result set's rows this many at a time. It is only safe to pause once +// it has handed over every row of the batch it fetched: paused part way through a result set's +// last, short batch, it moves on to the next result set, or ends the query, with the rest of +// that batch still to come. +const DRIVER_ROW_BATCH_SIZE = 50 + // String() converts a Symbol, which only fails to convert implicitly; an object without a // working toString falls back to its tag const thrownMessage = function (value) { @@ -347,6 +353,39 @@ class Request extends BaseRequest { // 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 + // full batch (see DRIVER_ROW_BATCH_SIZE). What the stream emits waits here while it is + // paused, and is emitted in the order it arrived once it is resumed. + const pending = [] + let rowsInRecordset = 0 + 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 + // msnodesqlv8 has freed the statement, which it does without 'done' after an error ends the query + let freed = false + let resuming = false + // a refused cancel undoes only the latest one + let cancelAttempts = 0 + + const emitInOrder = (isRow, emit) => { + if (this.stream && (this.paused || pending.length)) { + pending.push({ isRow, emit }) + } else { + emit() + } + } + + // once a cancel is through, what is waiting is emitted at once, but for the rows of a + // paused stream, which are dropped + const emitPendingOnCancel = () => { + // decided once, so that a listener that resumes the stream part way through cannot let + // the rows of the cancelled stream through + const dropRows = this.paused + for (const { isRow, emit } of pending.splice(0)) { + if (!isRow || !dropRows) emit() + } + } + const handleError = (req, connection, info, moreErrors) => { // a cancel is reported from 'done'. msnodesqlv8 >= 5.6 also sends an 'Operation canceled' // error after a cancel, and one that arrives before 'done' is dropped here @@ -359,7 +398,7 @@ class Request extends BaseRequest { err.code = 'EREQUEST' if (this.stream) { - this.emit('error', err) + emitInOrder(false, () => this.emit('error', err)) } else { if (doReturn && !hasReturned) { if (req) { @@ -493,30 +532,108 @@ class Request extends BaseRequest { this._setCurrentRequest(req) - this._cancel = () => { - debug('request(%d): cancel', IDS.get(this)) - // newer msnodesqlv8 versions drop a repeat cancel without calling back, so the - // outcome of the first one stands - if (typeof req.getCancelSent === 'function' && req.getCancelSent()) { - return - } - cancelRequested = true + const cancelDriver = (attempt = cancelAttempts) => { req.cancelQuery(err => { if (err) { debug('request(%d): failed to cancel', IDS.get(this), err) // msnodesqlv8 < 5.6 refuses to cancel a query that is neither paused nor polling, - // and the query runs to the end - if (!hasReturned) cancelRequested = false + // and the query runs to the end; if it reached the end of a batch meanwhile, a + // stream that is no longer paused resumes it + if (!hasReturned && attempt === cancelAttempts) { + cancelRequested = false + this._resumeStream() + } } - // this fixes an issue where paused connections don't emit a done event + // this fixes an issue where paused connections don't emit a done event; a query whose + // cancel was refused goes on, though it may have paused at the end of a batch since try { - if (req.isPaused()) req.emit('done') + if (!err && req.isPaused()) req.emit('done') } catch (err) { // do nothing } }) } + // every version of msnodesqlv8 cancels a paused query; it is paused here only where it stops + // for a pause, with nothing of its own under way + 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 + if (typeof req.getCancelSent === 'function' && req.getCancelSent()) { + cancelRequested = false + return false + } + driverPaused = true + req.pauseQuery() + cancelDriver() + return true + } + + // called with each row once it is complete + const pauseDriverAtEndOfBatch = () => { + if (driverPaused || rowsInRecordset % DRIVER_ROW_BATCH_SIZE !== 0) return + if (cancelDeferred && pauseAndCancelDriver()) return + if (this.stream && this.paused) { + driverPaused = true + req.pauseQuery() + } + } + + // 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) { + driverPaused = true + req.pauseQuery() + } + } + if (this.stream && this.paused) this._pauseStream() + + this._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 + resuming = true + try { + while (pending.length && !this.paused) { + pending.shift().emit() + } + } finally { + resuming = false + } + // cancelled while it emitted what was waiting after the query ended, and then paused, a + // 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) { + driverPaused = false + req.resumeQuery() + } + } + + this._cancel = () => { + debug('request(%d): cancel', IDS.get(this)) + // 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())) { + return + } + cancelRequested = true + cancelAttempts++ + if (!this.stream || !this.paused || driverPaused) { + return cancelDriver() + } + // A paused stream holds the driver back only once it reaches the end of a batch. Until + // then the driver goes on reading, and pausing or cancelling it part way through a fetch, + // or as it moves to the next result set or ends the query, can leave its queue broken. + // So it is paused and cancelled when it next stops, at the end of a batch or the start + // of a result set, and a query that ends first, at the latest on the request timeout, + // is reported as cancelled. + if (freed) return req.emit('done') + cancelDeferred = true + } + req.on('meta', metadata => { if (finished) return @@ -536,7 +653,7 @@ class Request extends BaseRequest { const ex2 = new RequestError(`Failed to parse incoming JSON. ${ex.message}`, 'EJSON') if (this.stream) { - this.emit('error', ex2) + emitInOrder(false, () => this.emit('error', ex2)) } else { console.error(ex2) } @@ -548,12 +665,14 @@ class Request extends BaseRequest { chunksBuffer = null if (row && row.___return___ == null) { // row with ___return___ col is the last row - if (this.stream && !this.paused) this.emit('row', row) + const last = row + if (this.stream) emitInOrder(true, () => this.emit('row', last)) } } } row = null + rowsInRecordset = 0 columns = metadata recordset = [] @@ -588,16 +707,23 @@ class Request extends BaseRequest { } if (this.stream) { if (!hasReturnColumn) { - this.emit('recordset', recordset.columns) + const recordsetColumns = recordset.columns + emitInOrder(false, () => this.emit('recordset', recordsetColumns)) } } else { recordsets.push(recordset) } if (this.arrayRowMode) recordsetcolumns.push(recordset.columns) + + // the driver fetches nothing of a result set it is paused at the start of + if (cancelDeferred) pauseAndCancelDriver() }) req.on('row', rownumber => { - if (finished || (row && isChunkedRecordset)) return + if (finished) return + + rowsInRecordset++ + if (row && isChunkedRecordset) return if (this.arrayRowMode) { row = [] @@ -642,11 +768,14 @@ class Request extends BaseRequest { } } if (!hasReturnColumn) { - if (this.stream && !this.paused && idx === columns.length - 1) { - this.emit('row', row) + if (this.stream && idx === columns.length - 1) { + const complete = row + emitInOrder(true, () => this.emit('row', complete)) } } } + + if (idx === columns.length - 1) pauseDriverAtEndOfBatch() }) req.on('rowcount', rowCount => { @@ -654,7 +783,7 @@ class Request extends BaseRequest { rowsAffected.push(rowCount) if (this.stream) { - this.emit('rowsaffected', rowCount) + emitInOrder(false, () => this.emit('rowsaffected', rowCount)) } }) @@ -666,7 +795,7 @@ class Request extends BaseRequest { msg.message = parsedMessage[1] } - this.emit('info', { + const info = { message: msg.message, number: msg.code, state: msg.sqlstate, @@ -674,7 +803,8 @@ class Request extends BaseRequest { lineNumber: msg.lineNumber || 0, serverName: msg.serverName, procName: msg.procName - }) + } + emitInOrder(false, () => this.emit('info', info)) // query terminated if (msg.code === 3621 && !hasReturned) { @@ -706,6 +836,14 @@ class Request extends BaseRequest { handleError(req, connection, info, moreErrors) }) + req.on('free', () => { + freed = true + if (cancelDeferred && !finished) { + cancelDeferred = false + req.emit('done') + } + }) + req.once('done', () => { finished = true @@ -729,6 +867,9 @@ class Request extends BaseRequest { const err = new RequestError('Canceled.', 'ECANCEL') if (this.stream) { + delete this._resumeStream + delete this._pauseStream + emitPendingOnCancel() // as with tedious, a cancelled stream reports the cancel as an error and still // completes with what was read up to it this.emit('error', err) @@ -770,7 +911,7 @@ class Request extends BaseRequest { const ex2 = new RequestError(`Failed to parse incoming JSON. ${ex.message}`, 'EJSON') if (this.stream) { - this.emit('error', ex2) + emitInOrder(false, () => this.emit('error', ex2)) } else { console.error(ex2) } @@ -782,7 +923,8 @@ class Request extends BaseRequest { chunksBuffer = null if (row && row.___return___ == null) { // row with ___return___ col is the last row - if (this.stream && !this.paused) { this.emit('row', row) } + const last = row + if (this.stream) emitInOrder(true, () => this.emit('row', last)) } } } @@ -803,14 +945,32 @@ class Request extends BaseRequest { } } - delete this._cancel this.parent.release(connection) debug('request(%d): completed', IDS.get(this)) if (this.stream) { - callback(null, this._nested ? row : null, output, rowsAffected, recordsetcolumns) + // the driver is through, but a paused stream completes once it has emitted what is + // still waiting. Cancelled before then, it completes as cancelled: at once if it is + // paused, dropping its rows, or else once it has emitted the rest + this._cancel = () => { + cancelRequested = true + if (this.paused) emitPendingOnCancel() + } + const last = this._nested ? row : null + emitInOrder(false, () => { + delete this._cancel + delete this._resumeStream + delete this._pauseStream + if (cancelRequested) { + debug('request(%d): canceled', IDS.get(this)) + this.emit('error', new RequestError('Canceled.', 'ECANCEL')) + return callback(null, null, output, rowsAffected, recordsetcolumns) + } + callback(null, last, output, rowsAffected, recordsetcolumns) + }) } else { + delete this._cancel callback(null, recordsets, output, rowsAffected, recordsetcolumns) } }) @@ -964,17 +1124,18 @@ class Request extends BaseRequest { }) } + // the query pauses the driver itself, mostly at the end of a batch (see DRIVER_ROW_BATCH_SIZE) _pause () { super._pause() - if (this._currentRequest) { - this._currentRequest.pauseQuery() + if (this._pauseStream) { + this._pauseStream() } } _resume () { super._resume() - if (this._currentRequest) { - this._currentRequest.resumeQuery() + if (this._resumeStream) { + this._resumeStream() } } } diff --git a/test/common/tests.js b/test/common/tests.js index a586acf8..eb542ee9 100644 --- a/test/common/tests.js +++ b/test/common/tests.js @@ -670,6 +670,39 @@ module.exports = (sql, driver) => { req.pipe(blockingTransformStream).pipe(wstream) }, + 'query with pipe and back pressure across partial row batches' (done) { + // msnodesqlv8 fetches rows 50 at a time, so each of these result sets ends with a short + // batch, or a batch of one row, that the stream is paused part way through or at the end of + const recordsets = [['a', 75], ['b', 51], ['c', 120]] + const expected = recordsets.flatMap(([column, count]) => Array.from({ length: count }, (_, i) => `${column}=${i + 1}`)) + const rows = [] + // takes one row at a time, and more slowly than the driver fetches a batch, so the request + // is paused throughout, including when a batch of one row arrives + const wstream = new stream.Writable({ + objectMode: true, + highWaterMark: 1, + write (row, encoding, cb) { + rows.push(Object.entries(row).map(([column, value]) => `${column}=${value}`).join()) + setTimeout(cb, 10) + } + }) + wstream.on('finish', () => { + try { + assert.deepStrictEqual(rows, expected, 'every row should arrive once, in order, under its own column') + done() + } catch (err) { + done(err) + } + }) + wstream.on('error', done) + + const req = new sql.Request() + req.query(recordsets.map(([column, count]) => + `select top (${count}) cast(row_number() over (order by object_id, column_id) as int) as ${column} from sys.all_columns order by object_id, column_id` + ).join(';')) + req.pipe(wstream) + }, + 'query with pipe and an error' (endsTheBatch, done) { // pipe() returns the destination, not the stream it puts in front of it, so the // destination is the only place a request's error can be handled diff --git a/test/common/unit.js b/test/common/unit.js index 1cbae175..6751717d 100644 --- a/test/common/unit.js +++ b/test/common/unit.js @@ -2301,13 +2301,16 @@ describe('msnodesqlv8 query events', () => { assert.deepStrictEqual(recordset, [{ a: 1 }], 'the query should complete') }) - // every msnodesqlv8 version cancels a paused query, and ends it with 'done' before calling back + // every msnodesqlv8 version cancels a paused query, and ends it with 'done' before calling back; + // a stream paused part way through a result set is cancelled once the driver has handed over + // the batch, so this query, whose batch is its last, ends first it('should stream a cancel of a paused query once, then complete with what was read', async () => { let paused = false const { request, pool } = setup(notify => { result('a', 1)(notify) notify.emit('rowcount', 1) result('b', 2)(notify) + setImmediate(() => notify.emit('done')) }, { isPaused: () => paused, pauseQuery () { paused = true }, @@ -2375,4 +2378,606 @@ describe('msnodesqlv8 query events', () => { await tick() assert.strictEqual(connection.hasError, true, 'a link failure after a cancel should mark the connection as broken') }) + + describe('stream back pressure', () => { + const { Writable } = require('node:stream') + const BATCH = 50 + + // Plays `recordsets` ([column, row count] each) the way msnodesqlv8's reader does: it + // fetches a result set's rows BATCH at a time and hands them over until it is paused. What + // it does after a pause part way through a batch is what differs: up to 4.x it goes on as if + // the batch was handed over; 5.0 to 5.5 do too, and when resumed after the last batch of a + // result set, move on to the next one a second time; 5.6 waits for the rest of the batch, + // unless it was paused on the batch's last row. `timeOutAtFetch` names a fetch that does not + // come back before the request times out, and `failAtEnd` ends the query on an error; a query + // that ends on an error emits 'free' and no 'done', as msnodesqlv8 does. `beginDelay` holds + // back the end of the first statement, and `stallBegin` keeps it from ending. + const reader = (version, recordsets, { timeOutAtFetch, failAtEnd, beginDelay, stallBegin } = {}) => { + const pausedAt = [] + const calls = [] + let notify = null + let queried = false + let handle = false + let token = null + let cancelSent = false + let fetches = 0 + let set = 0 + let read = 0 + let batch = null + let index = 0 + let paused = false + let running = true + + const meta = () => [{ name: recordsets[set][0], sqlType: 'int', size: 4, nullable: false }] + + const fetch = () => { + const data = [] + while (data.length < BATCH && read < recordsets[set][1]) data.push(++read) + // a fetch that comes back short has reached the end of the result set + return { data, endOfRows: data.length < BATCH } + } + + const isPaused = () => paused + + // handing over a row can pause the query, which stops the loop + const dispatchRows = () => { + if (!batch) return + while (!isPaused() && index < batch.data.length) { + notify.emit('row', index) + notify.emit('column', 0, batch.data[index++], false) + } + } + + const nextResult = () => setImmediate(() => { + if (!running) return + set++ + read = 0 + if (set < recordsets.length) { + notify.emit('meta', meta()) + dispatch() + } else if (failAtEnd) { + fail(Object.assign(new Error('Something went wrong'), { code: 50000, sqlstate: '42000' })) + } else { + running = false + notify.emit('done') + notify.emit('free') + } + }) + + const fail = err => { + running = false + notify.fail(err, false) + setImmediate(() => notify.emit('free')) + } + + const dispatch = () => { + if (!running || paused) return + if (++fetches === timeOutAtFetch) { + return setTimeout(fail, 10, Object.assign(new Error('Query timeout expired'), { code: 0, sqlstate: 'HYT00' })) + } + setImmediate(() => { + if (!running) return + batch = fetch() + index = 0 + dispatchRows() + if (version === '5.6' && paused && index < batch.data.length) return + if (batch.endOfRows) { + nextResult() + } else { + dispatch() + } + }) + } + + // every version cancels a paused query, and ends it with 'done' before calling back; + // before 5.6 a query that is neither paused nor polling is not cancelled + const sendCancel = callback => { + if (version !== '5.6' && !paused) { + calls.push('cancel refused') + return setImmediate(callback, new Error('Error: [msnodesql] cancel only supported for statements where polling is enabled.')) + } + calls.push('cancel') + running = false + setImmediate(() => { + notify.emit('done') + callback(null) + }) + } + + const query = { + isPaused, + pauseQuery () { + calls.push('pause') + pausedAt.push(batch ? `${index} of ${batch.data.length}` : 'before the first batch') + paused = true + }, + resumeQuery () { + calls.push('resume') + paused = false + dispatchRows() + if (paused) return + if (version !== '2.x to 4.x' && batch && batch.endOfRows && index >= batch.data.length) { + nextResult() + } else { + dispatch() + } + }, + // from 5.0 a query is sent one cancel: one made before the query has a statement waits + // for it, and a repeat is dropped without calling back + cancelQuery (callback) { + if (version !== '2.x to 4.x') { + if (cancelSent) return calls.push('cancel dropped') + if (!handle) { + token = callback + return + } + cancelSent = true + } + sendCancel(callback) + } + } + if (version !== '2.x to 4.x') query.getCancelSent = () => cancelSent + + return { + pausedAt, + calls, + fetches: () => fetches, + queried: () => queried, + query, + play (emitter) { + notify = emitter + queried = true + const begin = () => { + handle = true + if (token) { + const callback = token + token = null + cancelSent = true + sendCancel(callback) + } + if (stallBegin || !running) return + notify.emit('meta', meta()) + dispatch() + } + if (beginDelay) { + setTimeout(begin, beginDelay) + } else { + begin() + } + } + } + } + + const setupReader = (version, recordsets, options) => { + const query = reader(version, recordsets, options) + return { ...setup(query.play, query.query), pausedAt: query.pausedAt, calls: query.calls, fetches: query.fetches, queried: query.queried } + } + + const rowsOf = recordsets => recordsets.flatMap(([column, count]) => Array.from({ length: count }, (_, i) => `${column}=${i + 1}`)) + + // the rows the stream hands to a writable that takes one at a time, so that it is paused + // throughout + const pipeToSlowWritable = request => new Promise((resolve, reject) => { + const rows = [] + const writable = new Writable({ + objectMode: true, + highWaterMark: 1, + write (row, encoding, callback) { + rows.push(Object.entries(row).map(([column, value]) => `${column}=${value}`).join()) + setImmediate(callback) + } + }) + writable.on('finish', () => resolve(rows)) + request.on('error', reject) + request.pipe(writable) + }) + + // a short last batch in the first result set, a last batch of one row in the next, and a + // short last batch in the last + const RECORDSETS = [['a', 75], ['b', 51], ['c', 120]] + + for (const version of ['2.x to 4.x', '5.0 to 5.5', '5.6']) { + it(`should pipe every row of every result set with msnodesqlv8 ${version}`, async () => { + const { request, pausedAt } = setupReader(version, RECORDSETS) + const completed = request.query('select a; select b; select c') + const rows = await pipeToSlowWritable(request) + await completed + assert.deepStrictEqual(rows, rowsOf(RECORDSETS), 'every row should arrive once, in order, under its own column') + assert.ok(pausedAt.length > 0, 'the driver should have been paused') + assert.deepStrictEqual(pausedAt.filter(at => at !== `${BATCH} of ${BATCH}`), [], 'the driver should only be paused once it has handed over a whole batch') + }) + } + + it('should emit nothing while a stream is paused, and complete once it is resumed', async () => { + const { request, notify } = setupReader('5.6', [['a', 75]]) + request.stream = true + const rows = [] + let completed = false + request.on('row', row => { + rows.push(row.a) + if (rows.length === 60) request.pause() + }) + request.on('done', () => { completed = true }) + const finished = new Promise(resolve => notify.once('done', resolve)) + const result = request.query('select a') + + await finished + await tick() + assert.strictEqual(rows.length, 60, 'no row should be emitted while the stream is paused') + assert.strictEqual(completed, false, 'the stream should not complete while rows are waiting') + + request.resume() + await result + assert.deepStrictEqual(rows, rowsOf([['a', 75]]).map(row => Number(row.slice(2))), 'the waiting rows should be emitted once the stream is resumed') + assert.strictEqual(completed, true, 'the stream should complete once it has emitted the waiting rows') + }) + + it('should stream a cancel of a paused stream without emitting the rows still waiting', async () => { + const { request, notify, pool } = setupReader('5.6', [['a', 75]]) + request.stream = true + const rows = [] + const errors = [] + request.on('error', err => errors.push(err.code)) + request.on('row', row => { + rows.push(row.a) + if (rows.length === 10) request.pause() + }) + const completed = new Promise(resolve => request.on('done', resolve)) + request.query('select a') + + // the driver goes on to the end of its batch + while (!notify.isPaused()) await tick() + request.cancel() + await completed + assert.strictEqual(rows.length, 10, 'no row should be emitted after the stream was paused') + assert.deepStrictEqual(errors, ['ECANCEL'], 'the cancel should be streamed as an error') + assert.strictEqual(pool.released, 1, 'the connection should be released once') + }) + + it('should cancel a paused stream once the driver has handed over its batch', async () => { + const { request, pausedAt, calls } = setupReader('5.6', [['a', 75]]) + request.stream = true + const rows = [] + const errors = [] + request.on('error', err => errors.push(err.code)) + request.on('row', row => { + rows.push(row.a) + if (rows.length === 10) { + request.pause() + request.cancel() + } + }) + const completed = new Promise(resolve => request.on('done', resolve)) + request.query('select a') + + await completed + assert.strictEqual(rows.length, 10, 'no row should be emitted after the stream was paused') + assert.deepStrictEqual(errors, ['ECANCEL'], 'the cancel should be streamed as an error') + assert.deepStrictEqual(pausedAt, ['50 of 50'], 'the driver should be paused at the end of its batch') + assert.deepStrictEqual(calls, ['pause', 'cancel'], 'the driver should be cancelled once it is paused') + }) + + it('should cancel a paused stream at the start of the next result set', async () => { + const { request, calls } = setupReader('5.6', [['a', 30], ['b', 30]]) + request.stream = true + const rows = [] + const errors = [] + request.on('error', err => errors.push(err.code)) + request.on('row', row => { + rows.push(row) + if (rows.length === 10) { + request.pause() + request.cancel() + } + }) + const completed = new Promise(resolve => request.on('done', resolve)) + request.query('select a; select b') + + await completed + assert.strictEqual(rows.length, 10, 'no row should be emitted after the stream was paused') + assert.deepStrictEqual(errors, ['ECANCEL'], 'the cancel should be streamed as an error') + assert.deepStrictEqual(calls, ['pause', 'cancel'], 'the driver should be paused and cancelled once, before it reads the next result set') + }) + + it('should not pause a driver that is ending the query when a paused stream is cancelled', async () => { + const { request, notify, calls } = setupReader('5.6', [['a', 75]]) + request.stream = true + const rows = [] + const errors = [] + request.on('error', err => errors.push(err.code)) + request.on('row', row => { + rows.push(row.a) + if (rows.length === 60) request.pause() + }) + // once the driver has handed over its last, short batch it goes on to end the query + notify.on('column', (index, value) => { + if (value === 75) setImmediate(() => request.cancel()) + }) + const completed = new Promise(resolve => request.on('done', resolve)) + request.query('select a') + + await completed + assert.strictEqual(rows.length, 60, 'no row should be emitted after the stream was paused') + assert.deepStrictEqual(errors, ['ECANCEL'], 'a stream that drops rows should report the cancel') + assert.deepStrictEqual(calls, [], 'the driver should be left to end the query') + }) + + it('should stream a cancel of a paused stream whose query has ended with rows still waiting', async () => { + const { request, notify, pool } = setupReader('5.6', [['a', 75]]) + request.stream = true + const rows = [] + const errors = [] + request.on('error', err => errors.push(err.code)) + request.on('row', row => { + rows.push(row.a) + if (rows.length === 60) request.pause() + }) + const completed = new Promise(resolve => request.on('done', resolve)) + const finished = new Promise(resolve => notify.once('done', resolve)) + request.query('select a') + + await finished + await tick() + request.cancel() + await completed + assert.strictEqual(rows.length, 60, 'no row should be emitted after the stream was paused') + assert.deepStrictEqual(errors, ['ECANCEL'], 'a stream that drops rows should report the cancel') + assert.strictEqual(pool.released, 1, 'the connection should be released once') + }) + + it('should keep the driver paused for a cancel made while a stream emits its waiting rows', async () => { + const { request, notify, calls } = setupReader('5.0 to 5.5', [['a', 200]]) + request.stream = true + const rows = [] + const errors = [] + let completedAfter = null + request.on('error', err => errors.push(err.code)) + request.on('row', row => { + if (row.a === 10) request.pause() + if (row.a === 30) request.cancel() + }) + // a listener after the one that cancels + request.on('row', row => rows.push(row.a)) + const completed = new Promise(resolve => request.on('done', () => { + completedAfter = rows.length + resolve() + })) + request.query('select a') + + // the driver goes on to the end of its batch, then the stream is resumed + while (!notify.isPaused()) await tick() + request.resume() + await completed + assert.deepStrictEqual(calls, ['pause', 'cancel'], 'the driver should not be resumed under the cancel') + assert.deepStrictEqual(errors, ['ECANCEL'], 'the cancel should be the only error') + assert.deepStrictEqual(rows, Array.from({ length: 50 }, (_, i) => i + 1), 'the rows handed over before the cancel was through should be emitted in order') + assert.strictEqual(completedAfter, 50, 'the stream should complete after its last row') + }) + + it('should go on piping every row when msnodesqlv8 refuses a cancel', async () => { + const { request, calls } = setupReader('2.x to 4.x', [['a', 75], ['b', 51]]) + request.on('row', row => { + if (row.a === 5) request.cancel() + }) + const completed = request.query('select a; select b') + const rows = await pipeToSlowWritable(request) + await completed + assert.strictEqual(calls[0], 'cancel refused', 'the cancel should have been refused') + assert.deepStrictEqual(rows, rowsOf([['a', 75], ['b', 51]]), 'a query whose cancel is refused should run to the end') + }) + + it('should complete a cancelled paused stream once a fetch under way times out', async () => { + const { request, calls } = setupReader('5.0 to 5.5', [['a', 200]], { timeOutAtFetch: 3 }) + request.stream = true + const rows = [] + const errors = [] + request.on('error', err => errors.push(err.code)) + request.on('row', row => rows.push(row.a)) + const completed = new Promise(resolve => request.on('done', resolve)) + request.query('select a') + + // the third fetch does not come back + while (rows.length < 100) await tick() + await tick() + request.pause() + request.cancel() + await completed + assert.deepStrictEqual(calls, [], 'the driver should not be paused or cancelled part way through a fetch') + assert.deepStrictEqual(errors, ['EREQUEST', 'ECANCEL'], 'the timeout and then the cancel should be streamed') + assert.strictEqual(rows.length, 100, 'no row should be emitted after the stream was paused') + }) + + it('should report a cancel of a paused stream although an earlier cancel was refused', async () => { + const { request, calls } = setupReader('2.x to 4.x', [['a', 75]]) + request.stream = true + const rows = [] + const errors = [] + request.on('error', err => errors.push(err.code)) + request.on('row', row => { + rows.push(row.a) + // refused, as the driver is not paused, but only once the next cancel is under way + if (rows.length === 10) request.cancel() + if (rows.length === 13) { + request.pause() + request.cancel() + } + }) + const completed = new Promise(resolve => request.on('done', resolve)) + request.query('select a') + + await completed + assert.deepStrictEqual(calls, ['cancel refused', 'pause', 'cancel'], 'the second cancel should go through at the end of the batch') + assert.deepStrictEqual(errors, ['ECANCEL'], 'the cancel that went through should be streamed') + assert.strictEqual(rows.length, 13, 'no row should be emitted after the stream was paused') + }) + + it('should hold the driver back from the start for a stream paused before its query', async () => { + const { request, notify, calls, fetches } = setupReader('5.0 to 5.5', [['a', 75]]) + request.stream = true + request.pause() + const rows = [] + const errors = [] + request.on('error', err => errors.push(err.code)) + request.on('row', row => rows.push(row.a)) + const started = new Promise(resolve => notify.once('meta', resolve)) + const completed = new Promise(resolve => request.on('done', resolve)) + request.query('select a') + + await started + await tick() + await tick() + assert.strictEqual(fetches(), 0, 'the driver should fetch nothing while the stream is paused') + request.cancel() + await completed + assert.deepStrictEqual(calls, ['pause', 'cancel'], 'the driver should be paused before it fetches, and cancelled there') + assert.deepStrictEqual(errors, ['ECANCEL'], 'the cancel should be streamed as an error') + assert.strictEqual(rows.length, 0, 'no row should be emitted') + }) + + it('should complete a cancelled paused stream whose query ends on an error', async () => { + const { request, notify, pool } = setupReader('5.6', [['a', 75]], { failAtEnd: true }) + request.stream = true + const rows = [] + const errors = [] + request.on('error', err => errors.push(err.code)) + request.on('row', row => { + rows.push(row.a) + if (rows.length === 60) request.pause() + }) + notify.on('column', (index, value) => { + if (value === 75) setImmediate(() => request.cancel()) + }) + const completed = new Promise(resolve => request.on('done', resolve)) + request.query('select a') + + await completed + assert.strictEqual(rows.length, 60, 'no row should be emitted after the stream was paused') + assert.deepStrictEqual(errors, ['EREQUEST', 'ECANCEL'], 'the error and then the cancel should be streamed') + assert.strictEqual(pool.released, 1, 'the connection should be released once') + }) + + it('should emit what was waiting in order to every listener when one resumes the stream', async () => { + const { request } = setupReader('5.6', [['a', 120]]) + request.stream = true + const rows = [] + request.on('row', row => { + if (row.a === 10) { + request.pause() + setTimeout(() => request.resume(), 5) + } + // a stream that is not paused can be resumed all the same + if (row.a > 10 && row.a < 15) request.resume() + }) + // a listener after the one that resumes + request.on('row', row => rows.push(row.a)) + await request.query('select a') + assert.deepStrictEqual(rows, Array.from({ length: 120 }, (_, i) => i + 1), 'every listener should get the rows in order') + }) + + it('should complete as cancelled a stream cancelled and paused again while it emits what was waiting', async () => { + const { request, notify } = setupReader('5.6', [['a', 30]]) + request.stream = true + const rows = [] + const errors = [] + request.on('error', err => errors.push(err.code)) + request.on('row', row => { + rows.push(row.a) + if (rows.length === 5) request.pause() + if (rows.length === 10) { + request.cancel() + request.pause() + } + }) + const completed = new Promise(resolve => request.on('done', resolve)) + const finished = new Promise(resolve => notify.once('done', resolve)) + request.query('select a') + + // the query ends while the stream is paused, and the stream is resumed after it + await finished + await tick() + request.resume() + await completed + assert.deepStrictEqual(errors, ['ECANCEL'], 'the cancel should be streamed as an error') + assert.strictEqual(rows.length, 10, 'no row should be emitted after the stream was paused') + }) + + it('should cancel a paused stream whose first statement does not end', async () => { + const { request, calls, queried } = setupReader('5.6', [['a', 75]], { stallBegin: true }) + request.stream = true + const errors = [] + request.on('error', err => errors.push(err.code)) + const completed = new Promise(resolve => request.on('done', resolve)) + request.query('select a') + + while (!queried()) await tick() + request.pause() + request.cancel() + await completed + assert.deepStrictEqual(calls, ['pause', 'cancel'], 'the driver should be paused and cancelled at once') + assert.deepStrictEqual(errors, ['ECANCEL'], 'the cancel should be streamed as an error') + }) + + it('should cancel a stream paused before its first statement ends, after a cancel made while it was flowing', async () => { + const { request, calls, queried } = setupReader('5.0 to 5.5', [['a', 75]], { beginDelay: 10 }) + request.stream = true + const errors = [] + request.on('error', err => errors.push(err.code)) + const completed = new Promise(resolve => request.on('done', resolve)) + request.query('select a') + + while (!queried()) await tick() + // msnodesqlv8 holds this cancel until the query has a statement, and refuses it then + request.cancel() + request.pause() + request.cancel() + await completed + assert.deepStrictEqual(calls, ['pause', 'cancel'], 'the driver should be paused at once, so that the cancel is accepted') + assert.deepStrictEqual(errors, ['ECANCEL'], 'the cancel should be streamed as an error') + }) + + it('should not emit the rows of a cancelled paused stream that a listener resumes', async () => { + const { request, notify } = setupReader('5.6', [['a', 75], ['b', 30]]) + request.stream = true + const rows = [] + const errors = [] + request.on('error', err => errors.push(err.code)) + request.on('row', row => { + rows.push(row) + if (rows.length === 60) request.pause() + }) + // the next result set's 'recordset' waits ahead of its rows, and its listener resumes the stream + request.on('recordset', () => request.resume()) + const completed = new Promise(resolve => request.on('done', resolve)) + const finished = new Promise(resolve => notify.once('done', resolve)) + request.query('select a; select b') + + await finished + await tick() + request.cancel() + await completed + assert.strictEqual(rows.length, 60, 'no row should be emitted after the stream was paused and cancelled') + assert.deepStrictEqual(errors, ['ECANCEL'], 'the cancel should be streamed as an error') + }) + + it('should leave a cancel of a request that is no longer streaming to msnodesqlv8', async () => { + const { request, notify, calls } = setupReader('2.x to 4.x', [['a', 120]]) + request.stream = true + request.pause() + request.stream = false + notify.once('column', () => setImmediate(() => request.cancel())) + const result = await request.query('select a') + assert.deepStrictEqual(calls, ['cancel refused'], 'the driver should not be paused for the cancel') + assert.strictEqual(result.recordset.length, 120, 'a query whose cancel is refused should run to the end') + }) + + it('should not pause the driver for a request that is no longer streaming', async () => { + const { request, calls } = setupReader('5.6', [['a', 75]]) + request.stream = true + request.pause() + request.stream = false + const result = await request.query('select a') + assert.strictEqual(result.recordset.length, 75, 'every row should be read') + assert.deepStrictEqual(calls, [], 'the driver should not be paused') + }) + }) }) diff --git a/test/msnodesqlv8/msnodesqlv8.js b/test/msnodesqlv8/msnodesqlv8.js index a0ebd515..e67f85f4 100644 --- a/test/msnodesqlv8/msnodesqlv8.js +++ b/test/msnodesqlv8/msnodesqlv8.js @@ -118,6 +118,7 @@ describe('msnodesqlv8', function () { it('query with toReadableStream', done => TESTS['query with toReadableStream'](done)) it('query with pipe', done => TESTS['query with pipe'](done)) it('query with pipe and back pressure', (done) => TESTS['query with pipe and back pressure'](done)) + it('query with 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)) diff --git a/test/tedious/tedious.js b/test/tedious/tedious.js index dd2ea9d7..68bdefef 100644 --- a/test/tedious/tedious.js +++ b/test/tedious/tedious.js @@ -107,6 +107,7 @@ describe('tedious', () => { it('query with toReadableStream', done => TESTS['query with toReadableStream'](done)) it('query with pipe', done => TESTS['query with pipe'](done)) it('query with pipe and back pressure', (done) => TESTS['query with pipe and back pressure'](done)) + it('query with pipe and back pressure across partial row batches', done => TESTS['query with pipe and back pressure across partial row batches'](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)) From 5e7322f2150fa503e88514389758b16022a8a16a Mon Sep 17 00:00:00 2001 From: Dan Hensby Date: Tue, 13 Oct 2026 21:22:04 +0100 Subject: [PATCH 2/2] test: run a held-back msnodesqlv8 stream to the end once nothing reads it Co-Authored-By: Claude Opus 5.5 --- test/common/unit.js | 23 ++++++++++++++++++++++- 1 file changed, 22 insertions(+), 1 deletion(-) diff --git a/test/common/unit.js b/test/common/unit.js index 6751717d..3ddfe670 100644 --- a/test/common/unit.js +++ b/test/common/unit.js @@ -2380,7 +2380,7 @@ describe('msnodesqlv8 query events', () => { }) describe('stream back pressure', () => { - const { Writable } = require('node:stream') + const { Writable, pipeline } = require('node:stream') const BATCH = 50 // Plays `recordsets` ([column, row count] each) the way msnodesqlv8's reader does: it @@ -2959,6 +2959,27 @@ describe('msnodesqlv8 query events', () => { assert.deepStrictEqual(errors, ['ECANCEL'], 'the cancel should be streamed as an error') }) + it('should run a stream that is held back to the end once nothing reads it', async () => { + const { request, pool, calls } = setupReader('5.6', [['a', 75], ['b', 120]]) + const failure = new Error('the destination failed') + let written = 0 + const destination = new Writable({ + objectMode: true, + highWaterMark: 1, + write (row, encoding, callback) { + setImmediate(callback, ++written === 20 ? failure : null) + } + }) + const completed = new Promise(resolve => request.on('done', resolve)) + const failed = new Promise(resolve => pipeline(request.toReadableStream(), destination, resolve)) + request.query('select a; select b') + + assert.strictEqual(await failed, failure, 'the pipeline should fail with the destination\'s error') + await completed + assert.deepStrictEqual(calls, ['pause', 'resume'], 'the driver, held back at the end of its batch, should be resumed to run to the end') + assert.strictEqual(pool.released, 1, 'the connection should be released once') + }) + it('should leave a cancel of a request that is no longer streaming to msnodesqlv8', async () => { const { request, notify, calls } = setupReader('2.x to 4.x', [['a', 120]]) request.stream = true