Skip to content
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

Jump to bottom

Use MongoDB transactions #9916

Open
wants to merge 7 commits into
base: master
Choose a base branch
from
Open

Use MongoDB transactions #9916

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
d18f9cb
feat: basic transaction support in MongoDB
oplik0 Oct 16, 2021
d886e02
style: fix linting issues
oplik0 Oct 16, 2021
7f34eb5
fix: use session properly
oplik0 Oct 16, 2021
44d99b3
fix: promisify transaction
oplik0 Oct 16, 2021
08ecf14
chore: remove TODO from mongo transaction
oplik0 Oct 16, 2021
eeb435f
fix: don't run transactions on single instances
oplik0 Oct 16, 2021
7cc4045
fix: move getting value out of transaction
oplik0 Oct 17, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
3 changes: 2 additions & 1 deletion src/database/mongo.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,7 @@ mongoModule.questions = [

mongoModule.init = async function () {
client = await connection.connect(nconf.get('mongo'));
mongoModule.mongoClient = client;
mongoModule.client = client.db();
};

Expand Down Expand Up @@ -185,4 +186,4 @@ require('./mongo/sorted')(mongoModule);
require('./mongo/list')(mongoModule);
require('./mongo/transaction')(mongoModule);

require('../promisify')(mongoModule, ['client', 'sessionStore']);
require('../promisify')(mongoModule, ['client', 'sessionStore', 'transaction']);
46 changes: 27 additions & 19 deletions src/database/mongo/hash.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,9 +19,11 @@ module.exports = function (module) {
}
try {
if (isArray) {
const bulk = module.client.collection('objects').initializeUnorderedBulkOp();
key.forEach(key => bulk.find({ _key: key }).upsert().updateOne({ $set: writeData }));
await bulk.execute();
await module.transaction(async (session) => {
const bulk = module.client.collection('objects').initializeUnorderedBulkOp({ session });
key.forEach(key => bulk.find({ _key: key }).upsert().updateOne({ $set: writeData }));
await bulk.execute();
});
} else {
await module.client.collection('objects').updateOne({ _key: key }, { $set: writeData }, { upsert: true });
}
Expand All @@ -42,18 +44,20 @@ module.exports = function (module) {

const writeData = data.map(helpers.serializeData);
try {
let bulk;
keys.forEach((key, i) => {
if (Object.keys(writeData[i]).length) {
if (!bulk) {
bulk = module.client.collection('objects').initializeUnorderedBulkOp();
await module.transaction(async (session) => {
let bulk;
keys.forEach((key, i) => {
if (Object.keys(writeData[i]).length) {
if (!bulk) {
bulk = module.client.collection('objects').initializeUnorderedBulkOp({ session });
}
bulk.find({ _key: key }).upsert().updateOne({ $set: writeData[i] });
}
bulk.find({ _key: key }).upsert().updateOne({ $set: writeData[i] });
});
if (bulk) {
await bulk.execute();
}
});
if (bulk) {
await bulk.execute();
}
} catch (err) {
if (err && err.message.startsWith('E11000 duplicate key error')) {
return await module.setObjectBulk(keys, data);
Expand Down Expand Up @@ -196,12 +200,14 @@ module.exports = function (module) {
field = helpers.fieldToString(field);
data[field] = '';
});

if (Array.isArray(key)) {
await module.client.collection('objects').updateMany({ _key: { $in: key } }, { $unset: data });
await module.transaction(async (session) => {
await module.client.collection('objects').updateMany({ _key: { $in: key } }, { $unset: data }, { session });
});
} else {
await module.client.collection('objects').updateOne({ _key: key }, { $unset: data });
}

cache.del(key);
};

Expand All @@ -224,13 +230,15 @@ module.exports = function (module) {
increment[field] = value;

if (Array.isArray(key)) {
const bulk = module.client.collection('objects').initializeUnorderedBulkOp();
key.forEach((key) => {
bulk.find({ _key: key }).upsert().update({ $inc: increment });
await module.transaction(async (session) => {
const bulk = module.client.collection('objects').initializeUnorderedBulkOp({ session });
key.forEach((key) => {
bulk.find({ _key: key }).upsert().update({ $inc: increment });
});
await bulk.execute();
});
await bulk.execute();
cache.del(key);
const result = await module.getObjectsFields(key, [field]);
cache.del(key);
return result.map(data => data && data[field]);
}

Expand Down
16 changes: 12 additions & 4 deletions src/database/mongo/main.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,9 @@ module.exports = function (module) {
};

module.emptydb = async function () {
await module.client.collection('objects').deleteMany({});
await module.transaction(async (session) => {
await module.client.collection('objects').deleteMany({}, { session });
});
module.objectCache.reset();
};

Expand Down Expand Up @@ -46,15 +48,19 @@ module.exports = function (module) {
if (!key) {
return;
}
await module.client.collection('objects').deleteMany({ _key: key });
await module.transaction(async (session) => {
await module.client.collection('objects').deleteMany({ _key: key }, { session });
});
module.objectCache.del(key);
};

module.deleteAll = async function (keys) {
if (!Array.isArray(keys) || !keys.length) {
return;
}
await module.client.collection('objects').deleteMany({ _key: { $in: keys } });
await module.transaction(async (session) => {
await module.client.collection('objects').deleteMany({ _key: { $in: keys } }, { session });
});
module.objectCache.del(keys);
};

Expand Down Expand Up @@ -100,7 +106,9 @@ module.exports = function (module) {
};

module.rename = async function (oldKey, newKey) {
await module.client.collection('objects').updateMany({ _key: oldKey }, { $set: { _key: newKey } });
await module.transaction(async (session) => {
await module.client.collection('objects').updateMany({ _key: oldKey }, { $set: { _key: newKey } }, { session });
});
module.objectCache.del([oldKey, newKey]);
};

Expand Down
48 changes: 25 additions & 23 deletions src/database/mongo/sets.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,6 @@ module.exports = function (module) {
return;
}
value = value.map(v => helpers.valueToString(v));

await module.client.collection('objects').updateOne({
_key: key,
}, {
Expand All @@ -35,26 +34,27 @@ module.exports = function (module) {
}

value = value.map(v => helpers.valueToString(v));

const bulk = module.client.collection('objects').initializeUnorderedBulkOp();

for (let i = 0; i < keys.length; i += 1) {
bulk.find({ _key: keys[i] }).upsert().updateOne({
$addToSet: {
members: {
$each: value,
await module.transaction(async (session) => {
const bulk = module.client.collection('objects').initializeUnorderedBulkOp({ session });

for (let i = 0; i < keys.length; i += 1) {
bulk.find({ _key: keys[i] }).upsert().updateOne({
$addToSet: {
members: {
$each: value,
},
},
},
});
}
try {
await bulk.execute();
} catch (err) {
if (err && err.message.startsWith('E11000 duplicate key error')) {
return await module.setsAdd(keys, value);
});
}
throw err;
}
try {
await bulk.execute();
} catch (err) {
if (err && err.message.startsWith('E11000 duplicate key error')) {
return await module.setsAdd(keys, value);
}
throw err;
}
});
};

module.setRemove = async function (key, value) {
Expand All @@ -63,17 +63,19 @@ module.exports = function (module) {
}

value = value.map(v => helpers.valueToString(v));

await module.client.collection('objects').updateMany({ _key: Array.isArray(key) ? { $in: key } : key }, { $pullAll: { members: value } });
await module.transaction(async (session) => {
await module.client.collection('objects').updateMany({ _key: Array.isArray(key) ? { $in: key } : key }, { $pullAll: { members: value } }, { session });
});
};

module.setsRemove = async function (keys, value) {
if (!Array.isArray(keys) || !keys.length) {
return;
}
value = helpers.valueToString(value);

await module.client.collection('objects').updateMany({ _key: { $in: keys } }, { $pull: { members: value } });
await module.transaction(async (session) => {
await module.client.collection('objects').updateMany({ _key: { $in: keys } }, { $pull: { members: value } }, { session });
});
};

module.isSetMember = async function (key, value) {
Expand Down
5 changes: 3 additions & 2 deletions src/database/mongo/sorted.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -453,8 +453,9 @@ module.exports = function (module) {
module.sortedSetRemoveRangeByLex = async function (key, min, max) {
const query = { _key: key };
buildLexQuery(query, min, max);

await module.client.collection('objects').deleteMany(query);
await module.transaction(async (session) => {
await module.client.collection('objects').deleteMany(query, { session });
});
};

function buildLexQuery(query, min, max) {
Expand Down
52 changes: 30 additions & 22 deletions src/database/mongo/sorted/add.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,12 +39,15 @@ module.exports = function (module) {
}
}
values = values.map(helpers.valueToString);

const bulk = module.client.collection('objects').initializeUnorderedBulkOp();
for (let i = 0; i < scores.length; i += 1) {
bulk.find({ _key: key, value: values[i] }).upsert().updateOne({ $set: { score: parseFloat(scores[i]) } });
}
await bulk.execute();
await module.transaction(async (session) => {
const bulk = module.client.collection('objects').initializeUnorderedBulkOp({ session });
for (let i = 0; i < scores.length; i += 1) {
bulk.find({ _key: key, value: values[i] }).upsert().updateOne(
{ $set: { score: parseFloat(scores[i]) } },
);
}
await bulk.execute();
});
}

module.sortedSetsAdd = async function (keys, scores, value) {
Expand All @@ -62,28 +65,33 @@ module.exports = function (module) {
}

value = helpers.valueToString(value);

const bulk = module.client.collection('objects').initializeUnorderedBulkOp();
for (let i = 0; i < keys.length; i += 1) {
bulk
.find({ _key: keys[i], value: value })
.upsert()
.updateOne({ $set: { score: parseFloat(isArrayOfScores ? scores[i] : scores) } });
}
await bulk.execute();
await module.transaction(async (session) => {
const bulk = module.client.collection('objects').initializeUnorderedBulkOp({ session });
for (let i = 0; i < keys.length; i += 1) {
bulk
.find({ _key: keys[i], value: value })
.upsert()
.updateOne({ $set: { score: parseFloat(isArrayOfScores ? scores[i] : scores) } });
}
await bulk.execute();
});
};

module.sortedSetAddBulk = async function (data) {
if (!Array.isArray(data) || !data.length) {
return;
}
const bulk = module.client.collection('objects').initializeUnorderedBulkOp();
data.forEach((item) => {
if (!utils.isNumber(item[1])) {
throw new Error(`[[error:invalid-score, ${item[1]}]]`);
}
bulk.find({ _key: item[0], value: String(item[2]) }).upsert().updateOne({ $set: { score: parseFloat(item[1]) } });
await module.transaction(async (session) => {
const bulk = module.client.collection('objects').initializeUnorderedBulkOp({ session });
data.forEach((item) => {
if (!utils.isNumber(item[1])) {
throw new Error(`[[error:invalid-score, ${item[1]}]]`);
}
bulk.find({ _key: item[0], value: String(item[2]) }).upsert().updateOne({
$set: { score: parseFloat(item[1]) },
});
});
await bulk.execute();
});
await bulk.execute();
};
};
27 changes: 16 additions & 11 deletions src/database/mongo/sorted/remove.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,10 +17,11 @@ module.exports = function (module) {
} else {
value = helpers.valueToString(value);
}

await module.client.collection('objects').deleteMany({
_key: Array.isArray(key) ? { $in: key } : key,
value: isValueArray ? { $in: value } : value,
await module.transaction(async (session) => {
await module.client.collection('objects').deleteMany({
_key: Array.isArray(key) ? { $in: key } : key,
value: isValueArray ? { $in: value } : value,
}, { session });
});
};

Expand All @@ -29,8 +30,9 @@ module.exports = function (module) {
return;
}
value = helpers.valueToString(value);

await module.client.collection('objects').deleteMany({ _key: { $in: keys }, value: value });
await module.transaction(async (session) => {
await module.client.collection('objects').deleteMany({ _key: { $in: keys }, value: value }, { session });
});
};

module.sortedSetsRemoveRangeByScore = async function (keys, min, max) {
Expand All @@ -48,16 +50,19 @@ module.exports = function (module) {
query.score = query.score || {};
query.score.$lte = parseFloat(max);
}

await module.client.collection('objects').deleteMany(query);
await module.transaction(async (session) => {
await module.client.collection('objects').deleteMany(query, { session });
});
};

module.sortedSetRemoveBulk = async function (data) {
if (!Array.isArray(data) || !data.length) {
return;
}
const bulk = module.client.collection('objects').initializeUnorderedBulkOp();
data.forEach(item => bulk.find({ _key: item[0], value: String(item[1]) }).delete());
await bulk.execute();
await module.transaction(async (session) => {
const bulk = module.client.collection('objects').initializeUnorderedBulkOp({ session });
data.forEach(item => bulk.find({ _key: item[0], value: String(item[1]) }).delete());
await bulk.execute();
});
};
};
13 changes: 10 additions & 3 deletions src/database/mongo/transaction.js
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,8 +1,15 @@
'use strict';

module.exports = function (module) {
// TODO
module.transaction = function (perform, callback) {
perform(module.client, callback);
module.transaction = async function (perform) {
if (module.mongoClient.topology.description.type === 'Single' || !module.mongoClient.topology.hasSessionSupport()) {
return await perform();
}
const session = module.mongoClient.startSession();
try {
return await session.withTransaction(perform);
} finally {
await session.endSession();
}
};
};