MongoDB - トランザクション

概要

トランザクションとは、複数のデータベース操作を1つの作業単位としてまとめる機能である。
トランザクション内の全ての操作が成功した場合のみ変更がコミットされ、1つでも失敗した場合は全ての変更がロールバックされる。

MongoDBは、ACID特性 (Atomicity, Consistency, Isolation, Durability) を保証するトランザクション機能を提供している。

MongoDB 4.0以降でマルチドキュメントトランザクションがサポートされ、MongoDB 4.2以降ではシャードクラスタでの分散トランザクションもサポートされている。
トランザクション機能を使用するには、レプリカセットまたはシャードクラスタ環境が必要であり、スタンドアロン構成では使用できない。

トランザクションの種類

MongoDBは2種類のトランザクション機能を提供している。

単一ドキュメントのアトミック性

MongoDBは、単一ドキュメントに対する操作を自動的にアトミックに実行する。
これには、埋め込みドキュメントや配列を含むドキュメント全体の更新が含まれる。

単一ドキュメントのアトミック操作は、特別なコードや設定なしで常に有効であり、パフォーマンスが高い。
MongoDBの設計では、80〜90[%]のユースケースが単一ドキュメントのアトミック性で対応可能である。

マルチドキュメントトランザクション

複数のドキュメントや複数のコレクションにまたがる操作を1つのトランザクションとして実行する機能である。

サポートバージョン:

• MongoDB 4.0以降

  レプリカセットでのマルチドキュメントトランザクション

• MongoDB 4.2以降

  シャードクラスタでの分散トランザクション

マルチドキュメントトランザクションは、複数の操作をまとめて実行し、全てが成功した場合のみコミットする必要がある場合に使用する。

単一ドキュメントのアトミック性

MongoDBでは、単一ドキュメントへの書き込み操作は自動的にアトミックである。

埋め込みドキュメントの更新

埋め込みドキュメントや配列を含むドキュメント全体の更新は、アトミックに実行される。

以下の例では、balanceフィールドとlast_transaction埋め込みドキュメント全体が同時に更新される。

 db.accounts.updateOne(
    { _id: 'account_123' },
    {
       $set: {
          'balance': 1500,
          'last_transaction': {
             'date': new Date(),
             'amount': 500,
             'type': 'deposit'
          }
       }
    }
 )

使用を推奨する場面

単一ドキュメントのアトミック性を活用することで、マルチドキュメントトランザクションの必要性を減らすことができる。

推奨される設計パターン:

• 関連データを埋め込みドキュメントとして格納
• 配列を使用して関連する複数の値を1つのドキュメントに格納
• 非正規化によるデータの重複を許容

このアプローチにより、パフォーマンスが向上し、トランザクションの複雑さを回避できる。

マルチドキュメントトランザクション

複数のドキュメントや複数のコレクションにまたがる操作を、1つのトランザクションとして実行する機能である。

レプリカセットでのトランザクション

MongoDB 4.0以降では、レプリカセット環境でマルチドキュメントトランザクションがサポートされている。

前提条件:

• レプリカセット構成であること (スタンドアロンでは使用不可)
• MongoDB 4.0以降のバージョン
• WiredTigerストレージエンジンを使用

シャードクラスタでの分散トランザクション

MongoDB 4.2以降では、シャードクラスタ環境でも分散トランザクションがサポートされている。

前提条件を以下に示す。

• 全てのシャードがレプリカセット構成であること
• MongoDB 4.2以降のバージョン
• configサーバがレプリカセット構成であること

シャードクラスタでのトランザクションは、複数のシャードにまたがるデータに対して実行できる。

トランザクション処理

MongoDBでトランザクションを実行する基本的な手順を説明する。

セッションの開始

トランザクションを実行するには、まずセッションを開始する必要がある。

• mongoshでのセッション開始

   const session = db.getMongo().startSession({
      readPreference: { mode: 'primary' }
   });
  

トランザクションの開始

セッションを使用してトランザクションを開始する。

トランザクションオプションを指定できる。

• readConcern

  読み取り関心レベル

• writeConcern

  書き込み関心レベル

• readPreference

  読み取り設定

 session.startTransaction({
    readConcern: { level: 'snapshot' },
    writeConcern: { w: 'majority' }
 });

コミット

トランザクション内の全ての操作が成功した場合、commitTransaction メソッドを呼び出して変更を確定する。

 session.commitTransaction();

中止

エラーが発生した場合やトランザクションをキャンセルする場合は、abortTransaction メソッドを呼び出す。

 session.abortTransaction();

mongoshでの完全なコード例

銀行口座間の送金を例とした完全なトランザクション処理の例を示す。

以下の例では、Patの口座から100を引き出し、Sashaの口座に100を入金している。
両方の操作が成功した場合のみコミットされ、いずれかが失敗した場合はロールバックされる。

 const session = db.getMongo().startSession({
    readPreference: { mode: 'primary' }
 });
 
 session.startTransaction({
    readConcern: { level: 'snapshot' },
    writeConcern: { w: 'majority' }
 });
 
 try {
    const accountsCollection = session.getDatabase('bank').accounts;
 
    accountsCollection.updateOne(
       { 'customer': 'Pat' },
       { $inc: { 'balance': -100 } },
       { session: session }
    );
 
    accountsCollection.updateOne(
       { 'customer': 'Sasha' },
       { $inc: { 'balance': 100 } },
       { session: session }
    );
 
    session.commitTransaction();
    print("トランザクション成功");
 }
 catch (error) {
    print("トランザクション失敗: " + error);
    session.abortTransaction();
 }
 finally {
    session.endSession();
 }

withTransaction API

MongoDBは、トランザクション処理を簡素化するための withTransaction コールバックAPIを提供している。

自動リトライ機能

withTransaction メソッドは、以下に示す機能を自動的に提供する。

• トランザクションの開始
• コールバック関数の実行
• 一時的なエラー発生時の自動リトライ (TransientTransactionError)
• コミット操作の自動リトライ (UnknownTransactionCommitResult)
• エラー時の自動中止

これにより、手動でエラーハンドリングやリトライロジックを実装する必要がなくなる。

mongoshでのコード例

withTransaction を使用した送金処理の例を示す。

 const session = db.getMongo().startSession();
 
 function transferFunds(session) {
    const accountsCollection = session.getDatabase('bank').accounts;
 
    accountsCollection.updateOne(
       { 'customer': 'Pat' },
       { $inc: { 'balance': -100 } },
       { session: session }
    );
 
    accountsCollection.updateOne(
       { 'customer': 'Sasha' },
       { $inc: { 'balance': 100 } },
       { session: session }
    );
 }
 
 try {
    session.withTransaction(transferFunds, {
       readConcern: { level: 'snapshot' },
       writeConcern: { w: 'majority' }
    });
    print("トランザクション成功");
 }
 catch (error) {
    print("トランザクション失敗: " + error);
 }
 finally {
    session.endSession();
 }

withTransaction メソッドは、コールバック関数内の処理を自動的にトランザクションとして実行し、エラー処理とリトライを管理する。

読み取り関心レベル

読み取り関心レベル (Read Concern) は、トランザクション内でのデータの読み取り一貫性を制御する設定である。

local

ローカルノードで利用可能な最新のデータを読み込む。

• 最も高速
• 一貫性の保証なし
• ロールバックされる可能性のあるデータを読み込む場合がある

• 設定例

   session.startTransaction({
      readConcern: { level: 'local' }
   });
  

majority

レプリカセットの過半数のノードで確認されたデータのみを読み込む。

• 耐久性が保証される
• ロールバックされることがない
• localより遅いが、一貫性が高い

• 設定例

   session.startTransaction({
      readConcern: { level: 'majority' }
   });
  

snapshot

特定時点の一貫したスナップショットからデータを読み込む。

• トランザクション内で推奨されるレベル
• 読み取り中の他の書き込み操作の影響を受けない
• 完全な分離性を保証

トランザクション内では、snapshot レベルが最も推奨される。

• 設定例

   session.startTransaction({
      readConcern: { level: 'snapshot' }
   });
  

書き込み関心レベル

書き込み関心レベル (Write Concern) は、書き込み操作の確認レベルを制御する設定である。

w: 1

プライマリノードのみで書き込みを確認する。

• デフォルト設定
• 最も高速
• プライマリノード障害時にデータ損失の可能性がある

• 設定例

   session.startTransaction({
      writeConcern: { w: 1 }
   });
  

w: "majority"

レプリカセットの過半数のノードで書き込みを確認する。

• 高い耐久性
• ノード障害時でもデータが保持される
• w: 1より遅い

• 設定例

   session.startTransaction({
      writeConcern: { w: 'majority' }
   });
  

j: true

ジャーナルへの永続化を待機する。

• 最も安全
• ディスクへの書き込み完了を確認
• パフォーマンスへの影響が大きい

• 設定例

   session.startTransaction({
      writeConcern: { w: 'majority', j: true }
   });
  

wtimeout

書き込み確認の待機時間をミリ秒単位で指定する。

以下の例では、5秒以内に過半数のノードで確認されない場合、エラーが発生する。

• 設定例

   session.startTransaction({
      writeConcern: { w: 'majority', wtimeout: 5000 }
   });
  

読み取り設定

読み取り設定 (Read Preference) は、レプリカセット内のどのメンバーから読み取りを行うかを指定する設定である。

primary

常にプライマリノードから読み取る。

• デフォルト設定
• 最新のデータを保証
• トランザクション内で推奨される設定

• 設定例

   const session = db.getMongo().startSession({
      readPreference: { mode: 'primary' }
   });
  

secondary

セカンダリノードから読み取る。

• プライマリノードの負荷を軽減
• トランザクション内では制限がある

※注意
マルチドキュメントトランザクション内では、読み取り設定は primary である必要がある。
secondary や他の設定を使用すると、エラーが発生する。

タイムアウトと制限事項

MongoDBトランザクションには、いくつかのタイムアウト設定と制限事項がある。

transactionLifetimeLimitSeconds

トランザクションの最大実行時間を秒単位で指定する。
この時間を超えると、トランザクションは自動的に中止される。

• デフォルト値

  60秒

• 設定変更の方法

   db.adminCommand({
      setParameter: 1,
      transactionLifetimeLimitSeconds: 120
   })
  

maxTransactionLockRequestTimeoutMillis

トランザクションがロック取得を待機する最大時間をミリ秒単位で指定する。
この時間内にロックを取得できない場合、トランザクションは中止される。

• デフォルト値

  5ミリ秒

maxTimeMS

個々の操作の最大実行時間をミリ秒単位で指定する。

• 設定例

   db.collection.find().maxTimeMS(5000);
  

シャードクラスタでの制限

シャードクラスタでトランザクションを使用する場合、以下に示す制限がある。

• トランザクション内の全ての操作のデータサイズは16[MB]以内
• トランザクションの実行時間はデフォルト60秒以内
• 全てのシャードがレプリカセット構成である必要がある。
• Read Concernは snapshot または majority が推奨される。

コレクション作成の制限

トランザクション内で新しいコレクションを作成することはできない。

トランザクション実行前にコレクションを作成しておく必要がある。

 db.createCollection('newCollection');
 
 const session = db.getMongo().startSession();
 session.startTransaction();
 session.getDatabase('testDB').newCollection.insertOne({ a: 1 }, { session: session });
 session.commitTransaction();

Pythonでのトランザクション

PyMongoを使用したトランザクション処理の例を示す。

with_transactionコールバックAPIの使用例

PyMongoは、with_transaction メソッドを提供しており、自動的にエラーハンドリングとリトライを行う。

 from pymongo import MongoClient
 from pymongo.errors import OperationFailure
 
 client = MongoClient('mongodb://localhost:27017/')
 db = client['bank_db']
 session = client.start_session()
 
 def transfer_funds(session):
    accounts = db.accounts
 
    accounts.update_one(
       {'_id': 'account_a'},
       {'$inc': {'balance': -100}},
       session=session
    )
 
    accounts.update_one(
       {'_id': 'account_b'},
       {'$inc': {'balance': 100}},
       session=session
    )
 
 try:
    session.with_transaction(transfer_funds)
    print("トランザクション成功")
 except OperationFailure as e:
    print(f"トランザクション失敗: {e}")
 finally:
    session.end_session()

手動トランザクション管理の使用例

手動でトランザクションの開始、コミット、中止を管理する例を示す。

手動管理では、トランザクションオプションをより詳細に制御することができる。

 from pymongo import MongoClient
 
 client = MongoClient('mongodb://localhost:27017/')
 db = client['bank_db']
 session = client.start_session()
 
 try:
    session.start_transaction(
       read_concern={'level': 'snapshot'},
       write_concern={'w': 'majority', 'j': True}
    )
 
    accounts = db.accounts
    accounts.update_one(
       {'_id': 'account_a'},
       {'$inc': {'balance': -100}},
       session=session
    )
 
    accounts.update_one(
       {'_id': 'account_b'},
       {'$inc': {'balance': 100}},
       session=session
    )
 
    session.commit_transaction()
    print("トランザクションコミット成功")
 
 except Exception as e:
    session.abort_transaction()
    print(f"トランザクション中止: {e}")
 finally:
    session.end_session()

Node.jsでのトランザクション

MongoDB Node.jsドライバを使用したトランザクション処理の例を示す。

withTransactionコールバックAPI

Node.jsドライバも withTransaction メソッドを提供している。

 const { MongoClient } = require('mongodb');
 
 async function main() {
    const client = new MongoClient('mongodb://localhost:27017/');
 
    try {
       await client.connect();
       const db = client.db('bank_db');
       const session = client.startSession();
 
       const transferCallback = async () => {
          const accounts = db.collection('accounts');
 
          await accounts.updateOne(
             { _id: 'account_a' },
             { $inc: { balance: -100 } },
             { session }
          );
 
          await accounts.updateOne(
             { _id: 'account_b' },
             { $inc: { balance: 100 } },
             { session }
          );
       };
 
       await session.withTransaction(transferCallback, {
          readConcern: { level: 'snapshot' },
          writeConcern: { w: 'majority' }
       });
 
       console.log('トランザクション成功');
       await session.endSession();
    }
    finally {
       await client.close();
    }
 }
 
 main().catch(console.error);

手動トランザクション管理

手動でトランザクションを管理する例を示す。

 const { MongoClient } = require('mongodb');
 
 async function main() {
    const client = new MongoClient('mongodb://localhost:27017/');
 
    try {
       await client.connect();
       const db = client.db('bank_db');
       const session = client.startSession();
 
       try {
          session.startTransaction({
             readConcern: { level: 'snapshot' },
             writeConcern: { w: 'majority' }
          });
 
          const accounts = db.collection('accounts');
 
          await accounts.updateOne(
             { _id: 'account_a' },
             { $inc: { balance: -100 } },
             { session }
          );
 
          await accounts.updateOne(
             { _id: 'account_b' },
             { $inc: { balance: 100 } },
             { session }
          );
 
          await session.commitTransaction();
          console.log('トランザクションコミット成功');
       }
       catch (error) {
          await session.abortTransaction();
          console.error('トランザクション中止:', error);
       }
       finally {
          await session.endSession();
       }
    }
    finally {
       await client.close();
    }
 }
 
 main().catch(console.error);

推奨される事柄

単一ドキュメント優先の設計

可能な限り、単一ドキュメントのアトミック性を活用する設計を優先する。

80〜90[%]のユースケースは、単一ドキュメントのアトミック性で対応可能である。

• 関連データを埋め込みドキュメントとして格納する。
• 非正規化を活用してデータの重複を許容する。
• 配列を使用して関連する複数の値を1つのドキュメントに格納する。

再試行ロジック

トランザクションは、一時的なネットワークエラーやリソース競合により失敗する可能性がある。

withTransaction メソッドを使用すると、これらの再試行が自動的に処理される。

推奨される再試行戦略を以下に示す。

• TransientTransactionError

  トランザクション全体を再試行

• UnknownTransactionCommitResult

  コミット操作のみを再試行

トランザクション実行時間の最小化

トランザクションの実行時間はできるだけ短く保つ。

推奨事項を以下に示す。

• トランザクション内で重い計算処理を避ける。
• トランザクション外でデータの検証やビジネスロジックを実行する。。
• トランザクション内では、必要最小限の操作のみを実行する。

デフォルトのタイムアウトは60秒であるため、長時間実行されるトランザクションは避ける。

適切な関心レベルの選択

トランザクションの要件に応じて、適切な読み取り関心レベルと書き込み関心レベルを選択する。

推奨設定を以下に示す。

• Read Concern

  snapshot (トランザクション内で推奨)

• Write Concern

  { w: 'majority' } (高い耐久性が必要な場合)

パフォーマンスと一貫性のトレードオフを考慮して選択する。

パフォーマンス考慮事項

トランザクションは、通常の操作よりもオーバーヘッドが大きい。

パフォーマンスの最適化方法を以下に示す。

• インデックスを適切に設定してクエリを高速化
• トランザクション内の操作数を最小限に抑える。
• バッチ処理が可能な場合は、バルク操作を検討する。
• モニタリングを実施して、トランザクションのボトルネックを特定する。

エラーハンドリング

トランザクション実行時のエラーを適切に処理する。

• トランザクション開始前に入力データを検証する。
• try-catch-finallyブロックでエラーを捕捉する。
• エラー発生時は必ず abortTransaction を呼び出す。
• セッションは必ず endSession で終了する。

リソースリークを防ぐため、finallyブロックでセッションを確実に終了する。