Skip to content

Event contract corpus; fix the pause crash, #480 and #481 (v6 groundwork) #482

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
TimelordUK merged 4 commits into from
Oct 17, 2026
Merged

Event contract corpus; fix the pause crash, #480 and #481 (v6 groundwork) #482

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
0078734
test: event contract corpus, and handover of the 5.6.0 event regressions
TimelordUK Oct 15, 2026
b45a83a
fix: pausing on the last row of a batch could run two native calls at…
TimelordUK Oct 17, 2026
ca8e222
fix: the work queue finishes each item exactly once, and a paused que…
TimelordUK Oct 17, 2026
055711b
feat!: refuse new work on a connection while a query is paused
TimelordUK Oct 17, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
158 changes: 96 additions & 62 deletions lib/driver.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,6 @@ const driverModule = ((() => {
CLOSE: 17,
UNBIND: 18
}
const executingStates = new Set(['STATEMENT_SUBMITTED', 'STATEMENT_POLLING'])

class DriverMgr {
constructor (sql, version) {
Expand All @@ -37,6 +36,13 @@ const driverModule = ((() => {
}

close (callback) {
// a paused query holds the connection, and would hold it forever: cancel it, so it
// ends as cancelled and frees its statement before the close runs.
const head = this.workQueue.peek()
if (head?.paused && head.commandType === driverCommandEnum.QUERY) {
logger.debugLazy(() => 'close: cancel the paused query at the head of the queue', 'DriverMgr.close')
this.cancel(head.args[0], () => {})
}
this.workQueue.enqueue(driverCommandEnum.CLOSE, () => {
logger.debugLazy(() => 'cppDriver.close ', 'DriverMgr.close')
this.cppDriver.close(() => {
Expand Down Expand Up @@ -66,7 +72,7 @@ const driverModule = ((() => {
const qid = notify.getQueryId()
const handle = notify.getHandle()

if (queueItem.operationId === peek.operationId) {
if (queueItem === peek) {
logger.debugLazy(() => `send ${queueItem.operationId} cancel to the driver`)
this.cppDriver.cancelQuery(qid, handle, (e) => {
this.forwardCancel(e, callback)
Expand Down Expand Up @@ -97,32 +103,19 @@ const driverModule = ((() => {
return qid === currentQueryId
}

// the native statement is still running (SQLExecute not yet returned) - for a
// polling statement a cancel only flags the poll loop, so the statement is busy
// for a while after the cancel request has returned.
isExecuting (notify) {
const state = notify.getLastStateChange()
return state ? executingStates.has(state.newState) : false
}

drop (qid, first, notify, callback) {
const executing = this.isExecuting(notify)
logger.debugLazy(() => `drop called for query id ${qid}, paused: ${first.paused}, executing: ${executing}`, 'DriverMgr.drop')
if (first.paused && executing) {
// paused before execution finished: the statement must not be freed here while
// the native side still uses it. un-park it and cancel it as the active query -
// the reader then ends and frees the statement as normal.
logger.debugLazy(() => `un-park executing query id ${qid} to cancel`, 'DriverMgr.drop')
this.workQueue.resume(first)
this.execCancel(notify, first, callback)
} else if (first.paused) {
logger.debugLazy(() => `executing cancel for paused query id ${qid}`, 'DriverMgr.drop')
this.execCancel(notify, first, () => {
this.freeStatement(notify, () => {
logger.debugLazy(() => `freed statement and dropped paused query id ${qid}`, 'DriverMgr.drop')
this.workQueue.dropItem(first)
callback(null)
})
logger.debugLazy(() => `drop called for query id ${qid}, paused: ${first.paused}`, 'DriverMgr.drop')
if (first.paused) {
// a paused query still owns its statement and the head of the queue. cancel the
// native statement, then the reader ends the query as cancelled and frees the
// statement as any other query does.
logger.debugLazy(() => `cancel paused query id ${qid}`, 'DriverMgr.drop')
this.execCancel(notify, first, (e) => {
// once cancelled the query is ending, and no longer holds the connection: work
// submitted from the cancel callback queues behind it
if (!e) this.workQueue.resume(first)
callback(e)
if (!e && notify.queryWorker) notify.queryWorker.cancelPaused()
})
} else {
logger.debugLazy(() => `executing cancel for active query id ${qid}`, 'DriverMgr.drop')
Expand Down Expand Up @@ -196,12 +189,14 @@ const driverModule = ((() => {
: []
}

raiseFree (queryId, notify, callback) {
// op: the free statement's own queue item, finished once 'free' is raised (none
// when there was no statement to free)
raiseFree (queryId, notify, callback, op) {
setImmediate(() => {
callback(null, queryId)
setImmediate(() => {
notify.emit('free', queryId)
this.workQueue.nextOp()
this.workQueue.finish(op)
})
})
}
Expand All @@ -210,16 +205,16 @@ const driverModule = ((() => {
const handle = notify.getHandle()
const queryId = notify.getQueryId()
if (handle) {
this.workQueue.enqueue(driverCommandEnum.FREE_STATEMENT, () => {
const op = this.workQueue.enqueueFollowUp(driverCommandEnum.FREE_STATEMENT, () => {
logger.debugLazy(() => `free statement ${queryId}`, 'DriverMgr.freeStatement')
this.cppDriver.releaseStatement(queryId, handle,
() => {
logger.debugLazy(() => `statement ${queryId} free - raise event`, 'DriverMgr.freeStatement')
this.raiseFree(queryId, notify, callback)
this.raiseFree(queryId, notify, callback, op)
})
}, [])
} else {
this.raiseFree(queryId, notify, callback)
this.raiseFree(queryId, notify, callback, null)
}
}

Expand All @@ -241,51 +236,106 @@ const driverModule = ((() => {
// only be unbound when rest of query completes. The output params
// will now be ready to fetch out of the statement.

next (callback, err) {
next (callback, err, op) {
setImmediate(() => {
callback(err || null, false)
setImmediate(() => {
this.workQueue.nextOp()
this.workQueue.finish(op)
})
})
}

// the native calls below always complete asynchronously, so op is assigned by then
// a paused query holds the connection: new work is refused until it is resumed and
// finished, or cancelled. work queued before the pause still waits its turn.
pausedQuery () {
const head = this.workQueue.peek()
return head?.paused && head.commandType === driverCommandEnum.QUERY ? head : null
}

pausedError (head) {
const qid = head.args[0].getQueryId()
const err = new Error(`[msnodesql] query ${qid} on this connection is paused - resume or cancel it before submitting more work`)
err.sqlstate = 'HY010' // function sequence error
err.code = 0
return err
}

// true when refused: callback gets the error
refuseWhilePaused (callback) {
const head = this.pausedQuery()
if (!head) return false
const err = this.pausedError(head)
setImmediate(() => callback(err))
return true
}

// a refused query ends as any failed query does: the error, then done and free.
// it was never submitted, so there is no 'submitted'.
refuseQueryWhilePaused (notify, cb) {
const head = this.pausedQuery()
if (!head) return false
const err = this.pausedError(head)
logger.debugLazy(() => `refuse query ${notify.getQueryId()}: ${err.message}`, 'DriverMgr.refuseQueryWhilePaused')
setImmediate(() => {
if (cb) {
cb(err, null, false)
} else if (notify.listenerCount('error') > 0) {
notify.emit('error', err, false)
}
// a tick on: the connection defers the user's callback by one
setImmediate(() => {
notify.emit('done', notify.getQueryId())
notify.emit('free', notify.getQueryId())
})
})
return true
}

beginTransaction (callback) {
this.workQueue.enqueue(driverCommandEnum.BEGIN_TRANSACTION, () => {
this.cppDriver.beginTransaction(err => { this.next(callback, err) })
if (this.refuseWhilePaused(callback)) return
const op = this.workQueue.enqueue(driverCommandEnum.BEGIN_TRANSACTION, () => {
this.cppDriver.beginTransaction(err => { this.next(callback, err, op) })
}, [])
}

rollback (callback) {
this.workQueue.enqueue(driverCommandEnum.ROLLBACK, () => {
this.cppDriver.rollback(err => { this.next(callback, err) })
if (this.refuseWhilePaused(callback)) return
const op = this.workQueue.enqueue(driverCommandEnum.ROLLBACK, () => {
this.cppDriver.rollback(err => { this.next(callback, err, op) })
}, [])
}

commit (callback) {
this.workQueue.enqueue(driverCommandEnum.COMMIT, () => {
this.cppDriver.commit(err => { this.next(callback, err) })
if (this.refuseWhilePaused(callback)) return
const op = this.workQueue.enqueue(driverCommandEnum.COMMIT, () => {
this.cppDriver.commit(err => { this.next(callback, err, op) })
}, [])
}

prepare (notify, queryOrObj, callback) {
this.workQueue.enqueue(driverCommandEnum.PREPARE, () => {
if (this.refuseWhilePaused(callback)) return
const op = this.workQueue.enqueue(driverCommandEnum.PREPARE, () => {
this.cppDriver.prepare(notify.getQueryId(), queryOrObj, (err, meta) => {
callback(err, meta)
this.workQueue.nextOp()
this.workQueue.finish(op)
})
}, [])
}

// the query keeps its own queue item: a prepared statement shares one notify across
// its executions, so the notify's operation is only ever the latest one queued
readOperation (notify, queryObj, params, factory, cb) {
notify.setOperation(this.workQueue.enqueue(driverCommandEnum.QUERY,
if (this.refuseQueryWhilePaused(notify, cb)) return
const op = this.workQueue.enqueue(driverCommandEnum.QUERY,
(notify, query, params, callback) => {
setImmediate(() => {
const q = this.reader.getQuery(notify, query, params, factory(), callback)
const q = this.reader.getQuery(notify, query, params, factory(), callback, op)
notify.setQueryWorker(q)
q.begin()
})
}, [notify, queryObj, params, cb]))
}, [notify, queryObj, params, cb])
notify.setOperation(op)
}

readAllPrepared (notify, queryObj, params, cb) {
Expand All @@ -303,24 +353,8 @@ const driverModule = ((() => {
() => new NativeQueryHandler(this.cppDriver, this), cb)
}

headPaused (notify, queryObj, params, cb) {
const peek = this.workQueue.peek()
const paused = peek?.paused
if (paused) {
const pausedNotify = peek.args[0]
this.freeStatement(pausedNotify, () => {
this.workQueue.dropItem(peek)
this.read(notify, queryObj, params, cb)
})
}
return paused
}

readAllQuery (notify, queryObj, params, cb) {
// if paused at head of q then kill this statement to allow driver to set up this one
if (!this.headPaused(notify, queryObj, params, cb)) {
this.read(notify, queryObj, params, cb)
}
this.read(notify, queryObj, params, cb)
}
}

Expand Down
12 changes: 6 additions & 6 deletions lib/query-handler.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,13 +65,13 @@ class NativeProcedureQueryHandler extends QueryHandler {
// for a stored procedure with multiple statements, only unbind after all
// statements are completed

unbind (more, not, qid, results, callback) {
unbind (more, not, qid, results, callback, op) {
// Check if the query was cancelled before attempting unbind
if (not.getCancelSent && not.getCancelSent()) {
logger.debugLazy(() => `Skipping parameter unbind for cancelled statement ${qid}`, 'NativeProcedureQueryHandler.unbind')
this.onStatementCompleteHandler.onStatementComplete(not, null, callback, results, more)
if (!more) {
this.workQueue.nextOp()
this.workQueue.finish(op)
}
return
}
Expand All @@ -81,7 +81,7 @@ class NativeProcedureQueryHandler extends QueryHandler {
logger.debugLazy(() => `Skipping parameter unbind for cancelled query worker ${qid}`, 'NativeProcedureQueryHandler.unbind')
this.onStatementCompleteHandler.onStatementComplete(not, null, callback, results, more)
if (!more) {
this.workQueue.nextOp()
this.workQueue.finish(op)
}
return
}
Expand All @@ -93,16 +93,16 @@ class NativeProcedureQueryHandler extends QueryHandler {
not.emit('output', outputVector)
this.onStatementCompleteHandler.onStatementComplete(not, outputVector, callback, results, more)
if (!more) {
this.workQueue.nextOp()
this.workQueue.finish(op)
}
})
}

end (not, outputParams, callback, results, endMore) {
if (!endMore) {
const qid = not.getQueryId()
this.workQueue.enqueue(this.unbindEnum, (a) =>
setImmediate(() => { this.unbind(a, not, qid, results, callback) }), [endMore])
const op = this.workQueue.enqueueFollowUp(this.unbindEnum, (a) =>
setImmediate(() => { this.unbind(a, not, qid, results, callback, op) }), [endMore])
} else {
this.onStatementCompleteHandler.onStatementComplete(not, null, callback, results, endMore)
}
Expand Down
Loading
Loading