Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
2 changes: 2 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -1381,6 +1381,8 @@ __Errors__
- ENOTOPEN (`ConnectionError`) - Connection not yet open.
- EALREADYBEGUN (`TransactionError`) - Transaction has already begun.

If the transaction fails to begin once it has a connection, the error is a `TransactionError`, with the error it comes from as its `originalError`. The connection goes back to the pool, and the transaction can be begun again. A connection that may have sent the request to begin the transaction is closed rather than lent out again, as the server may have begun the transaction even though the request failed (for example, when it timed out).

---------------------------------------

### commit ([callback])
Expand Down
24 changes: 21 additions & 3 deletions lib/msnodesqlv8/transaction.js
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,15 @@ class Transaction extends BaseTransaction {
super._begin(isolationLevel, err => {
if (err) return callback(err)

// built before a connection is borrowed, so that an invalid isolation level is reported
// with nothing to give back
let statement
try {
statement = `set transaction isolation level ${isolationLevelDeclaration(this.isolationLevel)};begin tran;`
} catch (err) {
return callback(err)
}

debug('transaction(%d): begin', IDS.get(this))

this.parent.acquire(this, (err, connection, config) => {
Expand All @@ -33,13 +42,22 @@ class Transaction extends BaseTransaction {

const req = new Request(this)
req.stream = false
req.query(`set transaction isolation level ${isolationLevelDeclaration(this.isolationLevel)};begin tran;`, err => {
req.query(statement, err => {
if (err) {
this.parent.release(this._acquiredConnection)
debug('transaction(%d): failed to begin', IDS.get(this), err)

this._acquiredConnection = null
this._acquiredConfig = null

return callback(err)
// the server may have begun the transaction even though the query failed (it can
// time out after the server has run it), so the pool is left to close the connection
// rather than lend it out again with a transaction nobody will end. It is not closed
// here, as msnodesqlv8 never calls back from closing a connection that is already
// closed, and the pool would wait for it to
connection.hasError = true
this.parent.release(connection)

return callback(new TransactionError(err))
}

debug('transaction(%d): begun', IDS.get(this))
Expand Down
6 changes: 6 additions & 0 deletions lib/tedious/connection-pool.js
Original file line number Diff line number Diff line change
Expand Up @@ -167,7 +167,13 @@ class ConnectionPool extends BaseConnectionPool {
const poolId = IDS.get(this)

if (tedious.closed) {
// already closed (tedious closes a connection whose socket ends, for one), but it is
// leaving the pool all the same
debug('connection(%d): already closed', IDS.get(tedious))
publish(CHANNELS.CONNECTION_DESTROY, () => ({
connectionId,
poolId
}))
resolve()
} else {
tedious.once('end', () => {
Expand Down
43 changes: 36 additions & 7 deletions lib/tedious/transaction.js
Original file line number Diff line number Diff line change
Expand Up @@ -47,13 +47,42 @@ class Transaction extends BaseTransaction {
this._acquiredConnection.on('rollbackTransaction', this._abort)
this._acquiredConfig = config

connection.beginTransaction(err => {
if (err) err = new TransactionError(err)

debug('transaction(%d): begun', IDS.get(this))

callback(err)
}, this.name, this.isolationLevel)
// the caller hears of one outcome, whatever tedious does
let calledBack = false
const done = (err, discard) => {
if (calledBack) return
calledBack = true

if (!err) {
debug('transaction(%d): begun', IDS.get(this))
return callback(null)
}

debug('transaction(%d): failed to begin', IDS.get(this), err)

connection.removeListener('rollbackTransaction', this._abort)
this._acquiredConnection = null
this._acquiredConfig = null

// deferred so that tedious is done with the connection before it is closed, and so
// that the callback runs outside the pool's promise chain, which would pass anything
// the callback throws back to the acquire callback above as a second error
setImmediate(() => {
if (discard) connection.close()
this.parent.release(connection)
callback(new TransactionError(err))
})
}

try {
// the server may have begun the transaction even though the request failed (it can
// time out after the server has run it), so the connection is closed rather than
// lent out again with a transaction nobody will end
connection.beginTransaction(err => done(err, true), this.name, this.isolationLevel)
} catch (err) {
// tedious validates the arguments before it sends anything
done(err, false)
}
})
})
}
Expand Down
3 changes: 3 additions & 0 deletions test/cleanup.sql
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,9 @@ if exists (select * from sys.tables where name = 'prepstm_test')
if exists (select * from sys.tables where name = 'tran_test')
exec('drop table [dbo].[tran_test]')

if exists (select * from sys.tables where name = 'tran_begin_test')
exec('drop table [dbo].[tran_begin_test]')

if exists (select * from sys.tables where name = 'bulk_table')
exec('drop table [dbo].[bulk_table]')

Expand Down
14 changes: 14 additions & 0 deletions test/common/diagnostics.js
Original file line number Diff line number Diff line change
Expand Up @@ -225,6 +225,20 @@ describe('Diagnostics Channel', () => {
}
}

it('ConnectionPool._poolDestroy() emits CONNECTION_DESTROY for a tedious connection that has already closed', async () => {
const ConnectionPool = require('../../lib/tedious/connection-pool')
const pool = new ConnectionPool({})
const connection = { closed: true }

const { events, stop } = collect(CHANNELS.CONNECTION_DESTROY)
try {
await pool._poolDestroy(connection)
assert.strictEqual(events.length, 1, 'the destroy should be published once')
} finally {
stop()
}
})

it('request.query() emits TRACE_QUERY start/asyncEnd with context', async () => {
const req = new sql.Request()
req._query = (cmd, cb) => setImmediate(cb, null, [[{ x: 1 }]], {}, 1)
Expand Down
107 changes: 107 additions & 0 deletions test/common/tests.js
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
'use strict'

const assert = require('node:assert')
const net = require('node:net')
const stream = require('node:stream')
const { join } = require('node:path')
const { format } = require('node:util')
Expand Down Expand Up @@ -2094,6 +2095,112 @@ module.exports = (sql, driver) => {
}).catch(done)
},

'gives the connection back when a transaction fails to begin' (config, done) {
// with one connection in the pool, nothing more can run unless the transaction gives
// back any connection it borrowed
const pool = new sql.ConnectionPool(Object.assign({}, config, { pool: { min: 0, max: 1, acquireTimeoutMillis: 5000 } }))
pool.connect().then(() => {
const tran = new sql.Transaction(pool)
tran.isolationLevel = 99
return tran.begin().then(() => {
throw new Error('begin() should reject an isolation level that does not exist')
}, err => {
assert.ok(err instanceof sql.TransactionError, `the rejection should be a TransactionError, got ${err.name}: ${err.message}`)
})
}).then(() => {
return pool.request().query('select 1 as v')
}).then(result => {
assert.strictEqual(result.recordset[0].v, 1, 'the pool should lend the connection out again')
return pool.close()
}).then(() => done(), err => {
// a pool that is missing a connection never finishes closing
pool.close().catch(() => {})
done(err)
})
},

'closes the connection when the request to begin a transaction times out' (config, done) {
// what the server sends is held back once `delay` is set, so the server has begun the
// transaction by the time the client gives up waiting to hear that it has
let delay = 0
// where the pool would have connected: tedious takes options.port over port, and the pool
// defaults port to 1433
const port = (config.options && config.options.port) || config.port || 1433
const sockets = new Set()
const proxy = net.createServer(client => {
const server = net.connect(port, config.server)
for (const socket of [client, server]) {
sockets.add(socket)
socket.on('error', () => {})
socket.on('close', () => {
client.destroy()
server.destroy()
})
}
client.pipe(server)
// each chunk goes on `delay` after it arrives, in the order it arrived
let sent = Promise.resolve()
server.on('data', chunk => {
const due = Date.now() + delay
sent = sent.then(() => new Promise(resolve => setTimeout(resolve, Math.max(0, due - Date.now())))).then(() => client.write(chunk))
})
})

let pool
const stopProxy = () => {
for (const socket of sockets) socket.destroy()
return new Promise(resolve => proxy.close(resolve))
}

proxy.listen(0, 'localhost', () => {
pool = new sql.ConnectionPool(Object.assign({}, config, {
server: 'localhost',
port: proxy.address().port,
options: Object.assign({}, config.options, { port: undefined }),
requestTimeout: 300,
validateConnection: 'socket',
pool: { min: 0, max: 1, acquireTimeoutMillis: 5000 }
}))
pool.connect().then(() => {
// the pool's one connection is logged in before anything is held back
return pool.request().query('select 1 as v')
}).then(() => {
delay = 1500
return new sql.Transaction(pool).begin().then(() => {
throw new Error('begin() should time out')
}, err => {
assert.ok(err instanceof sql.TransactionError, `the rejection should be a TransactionError, got ${err.name}: ${err.message}`)
assert.strictEqual(err.code, 'ETIMEOUT', 'begin() should time out')
delay = 0
})
}).then(() => {
// the next transaction gets the pool's one connection, so it has to be one with
// nothing left open on it for what it commits to last
const tran = new sql.Transaction(pool)
return tran.begin()
.then(() => tran.request().query('insert into tran_begin_test values (\'committed\')'))
.then(() => tran.commit())
}).then(() => {
return pool.request().query('select @@trancount as n')
}).then(result => {
assert.strictEqual(result.recordset[0].n, 0, 'the commit should end the only transaction on the connection')
return pool.close()
}).then(() => {
// read on a connection of its own, which nothing else has set up
const reader = new sql.ConnectionPool(config)
return reader.connect()
.then(() => reader.request().query('select data from tran_begin_test'))
.finally(() => reader.close())
}).then(result => {
assert.deepStrictEqual(result.recordset, [{ data: 'committed' }], 'the committed row should outlast the connection that committed it')
}).then(() => stopProxy().then(() => done()), err => {
// a pool that is missing a connection never finishes closing
if (pool) pool.close().catch(() => {})
stopProxy().then(() => done(err))
})
})
},

'cancel request' (done, message) {
const req = new TestRequest()
req.query('waitfor delay \'00:00:05\';select 1').catch(err => {
Expand Down
Loading