MongoDB 事务中有很多 WriteConflict 错误

Question

我目前正在 MongoDB 4.1.4 的最新可用 docker 映像中处理事务（使用 Node 8.12.0 和 Mongoose 5.3.8 作为客户端）。我用 3 个 mongo 实例制作了一个简单的副本集，一切正常，直到我WriteConflict在短时间内犯了很多错误。

我的代码如下所示：

// name, value are strings
// date is current time

const session = await createAnalyticsTransaction(); // returns 'session'

// _id is pregenerated
var stat = await Logger.findById(_id).session(session);

if (stat) {
    // do nothing if it already exists
    return true;
} 

await Logger.update({
    _id
}, {
    $setOnInsert: {
        _id,
        name,
        created: date.toDate(),
        modified: date.toDate()
    }
}, { 
    session, 
    upsert: true 
});

/*
    var period = 'month';
    var time = '2018-11'; 
    await Analytics.update({
        _id
    }, { 
        $setOnInsert: {
            _id,
            name,
            period,
            time,
            created: date.toDate()
        },
        $inc: inc
    }, {
        upsert: true,
        session: session 
    });
*/

await session.commitTransaction();
await session.endSession();

到目前为止，一切都在这里工作，直到我取消对 Analytics 集合的 upsert 的注释$inc，$setOnInsert并运行大约 1000 个同时操作。这个想法是，如果尚未创建 Analytics 集合，则应创建它。然后我开始得到很多MongoError: WriteConflict，错误的属性errorLabels有TransientTransactionError.

我认为这是因为$inc或upsert: true？有没有人经历过这个？在这种情况下，最好的解决方案是什么？

{ MongoError: WriteConflict
   at /Users/akuzmenok/Zend/MeteorLingua/lingua-analytics/node_modules/mongodb-core/lib/connection/pool.js:581:63
   at authenticateStragglers (/Users/akuzmenok/Zend/MeteorLingua/lingua-analytics/node_modules/mongodb-core/lib/connection/pool.js:504:16)
   at Connection.messageHandler (/Users/akuzmenok/Zend/MeteorLingua/lingua-analytics/node_modules/mongodb-core/lib/connection/pool.js:540:5)
   at emitMessageHandler (/Users/akuzmenok/Zend/MeteorLingua/lingua-analytics/node_modules/mongodb-core/lib/connection/connection.js:310:10)
   at Socket.<anonymous> (/Users/akuzmenok/Zend/MeteorLingua/lingua-analytics/node_modules/mongodb-core/lib/connection/connection.js:453:17)
   at emitOne (events.js:116:13)
   at Socket.emit (events.js:211:7)
   at addChunk (_stream_readable.js:263:12)
   at readableAddChunk (_stream_readable.js:250:11)
   at Socket.Readable.push (_stream_readable.js:208:10)
   at TCP.onread (net.js:597:20)
=> awaited here:
   at Function.Promise.await (/Users/akuzmenok/.meteor/packages/promise/.0.11.1.1ugu6ow.mjjhg++os+web.browser+web.browser.legacy+web.cordova/npm/node_modules/meteor-promise/promise_server.js:56:12)
   at Promise.asyncApply (imports/lib/analytics.js:97:9)
   at /Users/akuzmenok/.meteor/packages/promise/.0.11.1.1ugu6ow.mjjhg++os+web.browser+web.browser.legacy+web.cordova/npm/node_modules/meteor-promise/fiber_pool.js:43:40
=> awaited here:
   at Function.Promise.await (/Users/akuzmenok/.meteor/packages/promise/.0.11.1.1ugu6ow.mjjhg++os+web.browser+web.browser.legacy+web.cordova/npm/node_modules/meteor-promise/promise_server.js:56:12)
   at Promise.asyncApply (imports/lib/analytics.js:139:5)
   at /Users/akuzmenok/.meteor/packages/promise/.0.11.1.1ugu6ow.mjjhg++os+web.browser+web.browser.legacy+web.cordova/npm/node_modules/meteor-promise/fiber_pool.js:43:40
=> awaited here:
   at Function.Promise.await (/Users/akuzmenok/.meteor/packages/promise/.0.11.1.1ugu6ow.mjjhg++os+web.browser+web.browser.legacy+web.cordova/npm/node_modules/meteor-promise/promise_server.js:56:12)
   at Promise.asyncApply (imports/lib/analytics.js:158:5)
   at /Users/akuzmenok/.meteor/packages/promise/.0.11.1.1ugu6ow.mjjhg++os+web.browser+web.browser.legacy+web.cordova/npm/node_modules/meteor-promise/fiber_pool.js:43:40
=> awaited here:
   at Function.Promise.await (/Users/akuzmenok/.meteor/packages/promise/.0.11.1.1ugu6ow.mjjhg++os+web.browser+web.browser.legacy+web.cordova/npm/node_modules/meteor-promise/promise_server.js:56:12)
   at Promise.asyncApply (imports/lib/analytics.js:49:23)
   at /Users/akuzmenok/.meteor/packages/promise/.0.11.1.1ugu6ow.mjjhg++os+web.browser+web.browser.legacy+web.cordova/npm/node_modules/meteor-promise/fiber_pool.js:43:40
=> awaited here:
   at Function.Promise.await (/Users/akuzmenok/.meteor/packages/promise/.0.11.1.1ugu6ow.mjjhg++os+web.browser+web.browser.legacy+web.cordova/npm/node_modules/meteor-promise/promise_server.js:56:12)
   at Promise.asyncApply (imports/lib/analytics.js:14:23)
   at /Users/akuzmenok/.meteor/packages/promise/.0.11.1.1ugu6ow.mjjhg++os+web.browser+web.browser.legacy+web.cordova/npm/node_modules/meteor-promise/fiber_pool.js:43:40
 errorLabels: [ 'TransientTransactionError' ],
 operationTime: Timestamp { _bsontype: 'Timestamp', low_: 12, high_: 1541424838 },
 ok: 0,
 errmsg: 'WriteConflict',
 code: 112,
 codeName: 'WriteConflict',
 '$clusterTime':
  { clusterTime: Timestamp { _bsontype: 'Timestamp', low_: 12, high_: 1541424838 },
    signature: { hash: [Object], keyId: 0 } },
 name: 'MongoError',
 [Symbol(mongoErrorContextSymbol)]: {} }

另一个注意事项，我正在开始这样的交易：

const session = await MongoAnalytics.startSession({ causalConsistency: true });
session.startTransaction({ readConcern: { level: 'snapshot' }, writeConcern: { w: 'majority' } });

Answer 1

您从数据库读取数据，然后更新它。它看起来像：

DATA (state 0) <---
UPDATED DATA (state 1) --->

当您执行两个异步调用时：

DATA (state 0) <---
DATA (state 0) <---
UPDATED DATA (state 1) --->
UPDATED DATA (state 1') ---> ERROR

它返回一个错误，因为数据的状态改变了。这就是交易应该如何工作。

为了避免访问冲突，您可以实现自定义队列系统。或者捕获错误并使用 setTimeout 和最大尝试次数重新运行事务。

排队系统：

DATA (state 0) <---
UPDATED DATA (state 1) --->
DATA (state 1) <---
UPDATED DATA (state 2) --->

重新运行系统

DATA (state 0) <---
DATA (state 0) <---
UPDATED DATA (state 1) --->
UPDATED DATA (state 1') ---> ERROR
DATA (state 1) <---
UPDATED DATA (state 2) --->