Skip to content

Commit 71038b7

Browse files
committed
correct timing issues around transactions
`endSession` must wait for all the machinery behind the scenes to check out a connection and write a message before considering its job finished.
1 parent 7b1d9bd commit 71038b7

File tree

2 files changed

+36
-40
lines changed

2 files changed

+36
-40
lines changed

lib/core/sessions.js

+29-35
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@ const Transaction = require('./transactions').Transaction;
1313
const TxnState = require('./transactions').TxnState;
1414
const isPromiseLike = require('./utils').isPromiseLike;
1515
const ReadPreference = require('./topologies/read_preference');
16+
const maybePromise = require('../utils').maybePromise;
1617
const isTransactionCommand = require('./transactions').isTransactionCommand;
1718
const resolveClusterTime = require('./topologies/shared').resolveClusterTime;
1819
const isSharded = require('./wireprotocol/shared').isSharded;
@@ -125,25 +126,36 @@ class ClientSession extends EventEmitter {
125126
if (typeof options === 'function') (callback = options), (options = {});
126127
options = options || {};
127128

128-
if (this.hasEnded) {
129-
if (typeof callback === 'function') callback(null, null);
130-
return;
131-
}
129+
const session = this;
130+
return maybePromise(this, callback, done => {
131+
if (session.hasEnded) {
132+
return done();
133+
}
132134

133-
if (this.serverSession && this.inTransaction()) {
134-
this.abortTransaction(); // pass in callback?
135-
}
135+
function completeEndSession() {
136+
// release the server session back to the pool
137+
session.sessionPool.release(session.serverSession);
138+
session[kServerSession] = undefined;
136139

137-
// release the server session back to the pool
138-
this.sessionPool.release(this.serverSession);
139-
this[kServerSession] = undefined;
140+
// mark the session as ended, and emit a signal
141+
session.hasEnded = true;
142+
session.emit('ended', session);
143+
144+
// spec indicates that we should ignore all errors for `endSessions`
145+
done();
146+
}
147+
148+
if (session.serverSession && session.inTransaction()) {
149+
session.abortTransaction(err => {
150+
if (err) return done(err);
151+
completeEndSession();
152+
});
140153

141-
// mark the session as ended, and emit a signal
142-
this.hasEnded = true;
143-
this.emit('ended', this);
154+
return;
155+
}
144156

145-
// spec indicates that we should ignore all errors for `endSessions`
146-
if (typeof callback === 'function') callback(null, null);
157+
completeEndSession();
158+
});
147159
}
148160

149161
/**
@@ -227,16 +239,7 @@ class ClientSession extends EventEmitter {
227239
* @return {Promise} A promise is returned if no callback is provided
228240
*/
229241
commitTransaction(callback) {
230-
if (typeof callback === 'function') {
231-
endTransaction(this, 'commitTransaction', callback);
232-
return;
233-
}
234-
235-
return new Promise((resolve, reject) => {
236-
endTransaction(this, 'commitTransaction', (err, reply) =>
237-
err ? reject(err) : resolve(reply)
238-
);
239-
});
242+
return maybePromise(this, callback, done => endTransaction(this, 'commitTransaction', done));
240243
}
241244

242245
/**
@@ -246,16 +249,7 @@ class ClientSession extends EventEmitter {
246249
* @return {Promise} A promise is returned if no callback is provided
247250
*/
248251
abortTransaction(callback) {
249-
if (typeof callback === 'function') {
250-
endTransaction(this, 'abortTransaction', callback);
251-
return;
252-
}
253-
254-
return new Promise((resolve, reject) => {
255-
endTransaction(this, 'abortTransaction', (err, reply) =>
256-
err ? reject(err) : resolve(reply)
257-
);
258-
});
252+
return maybePromise(this, callback, done => endTransaction(this, 'abortTransaction', done));
259253
}
260254

261255
/**

test/functional/spec-runner/index.js

+7-5
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@ const expect = chai.expect;
77
const EJSON = require('mongodb-extjson');
88
const TestRunnerContext = require('./context').TestRunnerContext;
99
const resolveConnectionString = require('./utils').resolveConnectionString;
10+
const delay = require('../shared').delay;
1011

1112
chai.use(require('chai-subset'));
1213
chai.use(require('./matcher').default);
@@ -347,11 +348,12 @@ function runTestSuiteTest(configuration, spec, context) {
347348
throw err;
348349
})
349350
.then(() => {
350-
if (session0) session0.endSession();
351-
if (session1) session1.endSession();
352-
353-
return validateExpectations(context.commandEvents, spec, savedSessionData);
354-
});
351+
const promises = [];
352+
if (session0) promises.push(session0.endSession());
353+
if (session1) promises.push(session1.endSession());
354+
return Promise.all(promises);
355+
})
356+
.then(() => validateExpectations(context.commandEvents, spec, savedSessionData));
355357
});
356358
}
357359

0 commit comments

Comments
 (0)