トランザクション処理について

概要

このページでは、MongoDB のトランザクションの概要と、CIRCUS API でそれをラップしたものである TransactionManager の使い方について解説する。特にデータの削除が起きる場面では、よく考慮してコードを記述しないとデータの整合性が破綻してしまう。

多くの場合は TransactionManager で提供される機能を単に使うことでデータの整合性が保たれるが、外部参照を持つドキュメントを操作する場合に、明示的なロック処理が必要になることがあるため、その条件についても解説する。（この点については MongoDB 公式ブログの記事も参照。）

前提知識

1. CIRCUS ではデータベース操作の際に MongoDB NodeJS Driver の操作をラップした独自のアクセッサ CollectionAccessor を用いている。その目的はバリデーションをかけてドキュメントの整合性を担保することである。
2. 各コレクションに対してそれぞれ存在する CollectionAccessor の集合を Model と呼ぶ。
3. Model は API Server 起動時に作成されて、CIRCUS 独自の DI コンテナである ServiceLoader を通じて各サービスに注入される。

MongoDB のトランザクション

準備：レプリカセットの設定

MongoDB のトランザクション機能を使うにはレプリカセットの設定が必須である（機能を有効にするだけでよく、実際にサーバを 2 台用意する必要はない）。レプリカセット設定後、 mongosh にて rs.initiate() で初期化を行う。

レプリカセット設定例

1. /etc/mongod.confに下記設定を追記

replication:
  replSetName: "rs0"

2. mongod を再起動

sudo service mongod restart

3. mongosh にて初期化

$ mongo shell
>
>rs.initiate()
rs0:SECONDARY> ⏎
rs0:PRIMARY>

Session の使用

トランザクションの管理には ClientSession を用いる。コレクションを操作するメソッドに引数として ClientSession を渡すことで、ClientSession が渡された一連の操作をグループ化できる。

以下は、ひとつのドキュメントを取得し、その中のフィールド count に 1 を足して更新する、いわゆるカウンタである。この操作が複数同時に行われたとき、それぞれが取得する count の値は同じものになってしまうため、取得した値に 1 を足した結果も同じになる。そのため、複数の操作の更新内容がすべて同じになってしまう。求める結果としては、操作の回数分の count の増加であるが、結果として得られるのは、初期値から 1 増えた値になる。

const countUp = async () => {
  const doc = await db.collection('count').findOne({ id: '1' });
  const newCount = doc.count + 1;
  await db.collection('count').updateOne({ id: '1' }, { count: newCount });
};

トランザクションを導入することでドキュメントの取得と値の更新がグループ化され、取得したドキュメントが他の操作により変化していた場合(*1)、更新が失敗してトランザクションが再試行される。(*2) 再試行時にドキュメントの変化が無ければ書き込み操作が実行されるため、求める結果が得られる。

const countUp = async () => {
  const session = client.startSession();
  await session.withTransaction(async () => {
    const doc = await db.collection('count').findOne({ id: '1' }, { session });
    const newCount = doc.count + 1;
    await database
      .collection('count')
      .updateOne({ id: '1' }, { count: newCount }, { session });
  });
};

CIRCUS 内でのトランザクション実装方法

ここまでは NodeJS MongoDB Driver の機能を直接使った例だったが、そのようにすると CIRCUS API 内では以下のような問題が生じる。

1. Model で使っていたバリデーションなどの機能が使えなくなる
2. DB 操作ごとに { session } を書くのが大変で、書き忘れると発見しづらいバグの原因になる

そこで、トランザクション内で自動的に { session } を渡しつつバリデーションも機能させるための Session 付き Model が必要になる。

Session 付き Model は FunctionService のひとつである TransactionManager で提供される。これは公式の session.withTransaction() をラップし、Session 付き Model をトランザクションごとに作成し、コールバックに渡すものである。

TransactionManager の使用例を以下に示す。

// シリーズの images を書き換える例
const handleUpdate = ({ transactionManager }) => {
  return async (ctx, next) => {
    const { seriesUid } = ctx.params;
    await transactionManager.withTransaction(async models => {
      // ここでの models は「session 付き Model」である
      const { images } = await models.series.findById(seriesUid);
      const mr = multirange(images).append('7');
      await models.series.modifyOne(seriesUid, { images: mr.toString() });
    });
  };
};

const handleUpdate = ({ transactionManager }) => {
  // TransactionManager の外で series を find している
  const { images } = await db
    .collection('series')
    .findOne({ seriesUid: '111.222.333' });
  return async (ctx, next) => {
    await transactionManager.withTransaction(async models => {
      // find した series はこの時点で他の操作によって書き換えられているおそれがある
      const mr = multirange(images).append('7');
      await models.series.modifyOne(seriesUid, { images: mr.toString() });
    });
  };
};

ひとつのトランザクション内に一連の「ドキュメントからのデータ読み込み」「同一ドキュメントへのデータ書き込み」がまとまっている限りは、SELECT FOR UPDATE 相当のロック処理を手動で行う必要はない。

ただし「ドキュメントからのデータ読み込み」「そのデータを別のドキュメントへ書き込み」という組み合わせの場合には問題が生じるので後述する。

外部ドキュメントへの参照を持つドキュメントでのトランザクション

TransactionManager をそのまま使っても解決できない問題の例

MongoDB のトランザクションでは、同一ドキュメントに対して複数のトランザクションが同時に書き込みをしようとした場合はその競合を検出できるが、読み取りだけの場合には競合を検出できない。

以下は、外部参照を持つドキュメントを作成しようとしているが、その瞬間に参照先ドキュメントが削除されてしまう、という例である。（CIRCUS での例：series を参照している job や case の作成、case / job / series を参照している mylist の作成）

参照先の確認時には存在していたドキュメント（この例では Series）が、トランザクションの途中で削除されているのだが、それに気づいていないため、存在しない参照先をもつドキュメント（この例では Case）が作成されてしまっている。

このような状況を防ぐために、明示的なドキュメントのロックが必要となる。

ドキュメントのロックについて

上記の問題の解決策として、参照先のドキュメントを検索した際にロック用のフィールドの追加とその即時削除を行い(*3)、ドキュメントの内容を実質的には変更せずにロックする方法を採っている。ロックという MongoDB の操作が実際にあるわけではなく、ダミーの書き込みをすることでロック相当の操作を実現している。

ドキュメントのロックは Model の findById(id, { withLock: true }) で行う。

const handleCreateCase = ({ transactionManager }) => {
  return async (ctx, next) => {
    await transactionManager.withTransaction(async models => {
      // シリーズの存在確認をしつつロックする（withLockが重要!!）
      const series = await models.series.findById(seriesUid, {
        withLock: true
      });
      // 外部データを含む新しいドキュメントを作成
      const newCase = { revisions: [{ series: [{ seriesUid: series.uid }] }] };
      await models.clinicalCase.insert(newCase);
    });
  };
};