Транзакции в Mongoose

Транзакции появились в MongoDB 4.0 и Mongoose 5.2.0. Транзакции позволяют выполнять несколько операций изолированно и потенциально отменить все операции, если одна из них завершится неудачно. Это руководство поможет вам начать использовать транзакции с Mongoose.

Начало работы с транзакциями

Если вы ещё этого не сделали, импортируйте mongoose:

import mongoose from 'mongoose';

Чтобы создать транзакцию, сначала необходимо создать сеанс, используя Mongoose#startSession или Connection#startSession().

// Using Mongoose's default connection
const session = await mongoose.startSession();

// Using custom connection
const db = await mongoose.createConnection(mongodbUri).asPromise();
const session = await db.startSession();

На практике вы должны использовать либо помощник session.withTransaction(), либо функцию Mongoose Connection#transaction() для выполнения транзакции. Помощник session.withTransaction() обрабатывает:

• Создание транзакции
• Подтверждение транзакции, если она выполняется успешно
• Прерывание транзакции, если ваша операция завершается ошибкой
• Повторную попытку в случае временной ошибки транзакции.

let session = null;
return Customer.createCollection().
  then(() => Customer.startSession()).
  // The `withTransaction()` function's first parameter is a function
  // that returns a promise.
  then(_session => {
    session = _session;
    return session.withTransaction(() => {
      return Customer.create([{ name: 'Test' }], { session: session });
    });
  }).
  then(() => Customer.countDocuments()).
  then(count => assert.strictEqual(count, 1)).
  then(() => session.endSession());

Дополнительную информацию о функции ClientSession#withTransaction() см. в документации драйвера MongoDB Node.js.

Функция Mongoose Connection#transaction() — это оболочка вокруг withTransaction(), которая интегрирует отслеживание изменений Mongoose с транзакциями. Например, предположим, что вы save() документ в транзакции, которая впоследствии завершается ошибкой. Изменения в этом документе не сохраняются в MongoDB. Функция Connection#transaction() сообщает отслеживанию изменений Mongoose, что транзакция была отменена, и помечает все поля, которые были изменены в транзакции, как изменённые.

const doc = new Person({ name: 'Will Riker' });

await db.transaction(async function setRank(session) {
  doc.name = 'Captain';
  await doc.save({ session });
  doc.isNew; // false

  // Throw an error to abort the transaction
  throw new Error('Oops!');
}, { readPreference: 'primary' }).catch(() => {});

// true, `transaction()` reset the document's state because the
// transaction was aborted.
doc.isNew;

С документами Mongoose и save()

Если вы получаете документ Mongoose из findOne() или find() с использованием сеанса, документ будет хранить ссылку на сеанс и использовать этот сеанс для save().

Чтобы получить/установить сеанс, связанный с данным документом, используйте doc.$session().

const User = db.model('User', new Schema({ name: String }));

let session = null;
return User.createCollection().
  then(() => db.startSession()).
  then(_session => {
    session = _session;
    return User.create({ name: 'foo' });
  }).
  then(() => {
    session.startTransaction();
    return User.findOne({ name: 'foo' }).session(session);
  }).
  then(user => {
    // Getter/setter for the session associated with this document.
    assert.ok(user.$session());
    user.name = 'bar';
    // By default, `save()` uses the associated session
    return user.save();
  }).
  then(() => User.findOne({ name: 'bar' })).
  // Won't find the doc because `save()` is part of an uncommitted transaction
  then(doc => assert.ok(!doc)).
  then(() => session.commitTransaction()).
  then(() => session.endSession()).
  then(() => User.findOne({ name: 'bar' })).
  then(doc => assert.ok(doc));

С фреймворком агрегации

Функция Model.aggregate() также поддерживает транзакции. Агрегации Mongoose имеют помощник session(), который устанавливает параметр session. Ниже приведён пример выполнения агрегации в рамках транзакции.

const Event = db.model('Event', new Schema({ createdAt: Date }), 'Event');

let session = null;
return Event.createCollection().
  then(() => db.startSession()).
  then(_session => {
    session = _session;
    session.startTransaction();
    return Event.insertMany([
      { createdAt: new Date('2018-06-01') },
      { createdAt: new Date('2018-06-02') },
      { createdAt: new Date('2017-06-01') },
      { createdAt: new Date('2017-05-31') }
    ], { session: session });
  }).
  then(() => Event.aggregate([
    {
      $group: {
        _id: {
          month: { $month: '$createdAt' },
          year: { $year: '$createdAt' }
        },
        count: { $sum: 1 }
      }
    },
    { $sort: { count: -1, '_id.year': -1, '_id.month': -1 } }
  ]).session(session)).
  then(res => assert.deepEqual(res, [
    { _id: { month: 6, year: 2018 }, count: 2 },
    { _id: { month: 6, year: 2017 }, count: 1 },
    { _id: { month: 5, year: 2017 }, count: 1 }
  ])).
  then(() => session.commitTransaction()).
  then(() => session.endSession());

Расширенное использование

Пользователи, которые хотят иметь более точный контроль над моментом подтверждения или прерывания транзакций, могут использовать session.startTransaction() для запуска транзакции:

const Customer = db.model('Customer', new Schema({ name: String }));

let session = null;
return Customer.createCollection().
  then(() => db.startSession()).
  then(_session => {
    session = _session;
    // Start a transaction
    session.startTransaction();
    // This `create()` is part of the transaction because of the `session`
    // option.
    return Customer.create([{ name: 'Test' }], { session: session });
  }).
  // Transactions execute in isolation, so unless you pass a `session`
  // to `findOne()` you won't see the document until the transaction
  // is committed.
  then(() => Customer.findOne({ name: 'Test' })).
  then(doc => assert.ok(!doc)).
  // This `findOne()` will return the doc, because passing the `session`
  // means this `findOne()` will run as part of the transaction.
  then(() => Customer.findOne({ name: 'Test' }).session(session)).
  then(doc => assert.ok(doc)).
  // Once the transaction is committed, the write operation becomes
  // visible outside of the transaction.
  then(() => session.commitTransaction()).
  then(() => Customer.findOne({ name: 'Test' })).
  then(doc => assert.ok(doc)).
  then(() => session.endSession());

Также можно использовать session.abortTransaction() для прерывания транзакции:

let session = null;
return Customer.createCollection().
  then(() => Customer.startSession()).
  then(_session => {
    session = _session;
    session.startTransaction();
    return Customer.create([{ name: 'Test' }], { session: session });
  }).
  then(() => Customer.create([{ name: 'Test2' }], { session: session })).
  then(() => session.abortTransaction()).
  then(() => Customer.countDocuments()).
  then(count => assert.strictEqual(count, 0)).
  then(() => session.endSession());