Go gorm mysql の構成でデッドロックに立ち向かう ~retry処理, retry-go~

今回はMySQLにおけるデッドロックが起きるパターンとそれにどう対応するかまとめていきたいと思います。

RDBにおけるデッドロックが起きるパターン

デッドロックとはRDB(InnoDB)に複数リクエストが行われ、特定の条件下で同時に貼られたトランザクションが互いに終了を待つ状態でロックがかかってしまう事です。

特定の条件下と書いたように、デッドロックが起こるパターンはいくつも存在しますが、今回は代表例としてindexが貼られたテーブルにおけるネクストキーロックによってデッドロックが発生するパターンを紹介します。

例えばuser が tweet(つぶやき)できるサービスがあったとします。以下のようなテーブルを作成します。

drop table if exists users;
create table if not exists users
(
  id      int unsigned not null primary key auto_increment,
  name    varchar(128) not null,
) character set utf8mb4 collate utf8mb4_bin;

drop table if exists tweets;
create table if not exists tweets
(
  id          int unsigned not null primary key auto_increment,
  user_id     int unsigned not null,
  text varchar(256) not null,
  CONSTRAINT fk_tweets_users FOREIGN KEY (user_id) REFERENCES users (id)
) character set utf8mb4 collate utf8mb4_bin;

userとtweetは1対Nの関係で tweetsテーブルのuser_idは外部キーとなり、MySQLでは外部キーに対して自動でインデックスが貼られます。

ここまではよくある構成ですね。

ここでtweetの作成にある制約をつけます。例えば、 userはtweetを10個までしか作成できない としましょう。

SQLを実行しながらuserがtweetを作成する流れを見て行きます。

MySQLのクライアントを開きます。

// トランザクション開始
BEGIN;

// user_idを元にtweetsをカウントする FOR UPDATEを付ける
SELECT COUNT(user_id) FROM tweets WHERE user_id = 1 FOR UPDATE;

// tweetを作成
INSERT INTO tweets (user_id, text) VALUES (1, ツイート1);

// トランザクション終了
COMMIT;

SELEC文では末尾にFOR UPDATEを付ける事で排他ロックをかけてuser_idが該当するレコードの更新を防ぎます。もしこの間に同等のuser_idでレコードが作成または削除された場合 userはtweetを10個までしか作成できないというロジックに対して整合性を担保できなくなります。

単純に上記のトランザクションが流れても特にデッドロックは発生しません。

次にデッドロックが起きるパターンを見て行きます。

まずはテーブルをtruncateして初期状態に戻し、user_id = 1 のtweetを1件作成します。*usersテーブルにはuser1(ID = 1),user2(ID = 2)を用意してください。

TRUNCATE tweets;

INSERT INTO tweets (user_id, text) VALUES (1, ツイート1);

これでuser1のtweetが1件のみある状態になりました。

ここからデッドロックを再現して行きます。 user1とuser2が同時にtweetを作成するというシナリオです。

user1

BEGIN;

SELECT COUNT(user_id) from tweets where user_id = 1 for update;

//別クライアントからuser2の処理

INSERT INTO tweets (user_id, text) VALUES (1, ツイート2);

//user2のギャップロック待ち

user2

BEGIN;

SELECT COUNT(user_id) from tweets where user_id = 2 for update;

// user1がINSERTを待機している。
INSERT INTO tweets (user_id, text) VALUES (2, ツイートA);

ERROR 1213 (40001): Deadlock found when trying to get lock; try restarting transaction

user2の方でデッドロックが起きてしまいました。

これはuser2が初めてtweetを作成すると同時にuser1もtweetを作成したという時に起きる現象です。

初めてtweetするというのがデッドロックが起きる要因となります。詳しく見て行きます。

ネクストキーロック

ネクストキーロックとはRDBにおけるファントム現象を防ぐロック方式で、インデックス行ロックとギャップロックを組み合わせたものです。

user2ではインデックスカラムであるuser_idに対し WHERE user_id = 2 で検索をしています。この時結果が0の場合tweetsテーブルに+-無限大のギャップに対してギャップロックが掛かります。

次にuser1ではUPDATE句で排他ロックをかけようとしますが、user2のギャップロックがかかっている為user1の解除待ちになります。

そしてuser2からもUPDATEを実行した時にuser1と同じく排他ロックをかけてしまいデッドロックに到達するという訳です。

もしこの時user2にuser_id = 2のtweetが1レコードでもあればデッドロックは発生しません。

以上がネクストキーロックによるデッドロックパターンの一例です。

デッドロックを掛けないようにするのは重要ですが、上記のパターンのようにどうしても発生する場面は訪れます。

次の章ではデッドロックに対してトランザクションの再試行という方法で対策していきたいと思います。

デッドロックの対策

MySQLのデッドロックに対する対処方法です。

MySQL :: MySQL 5.6 リファレンスマニュアル :: 14.2.11 デッドロックの対処方法

今回はドキュメントにもあるようにトランザクションの再試行で対策してみたいと思います。

デッドロックが原因でトランザクションに失敗した場合に、そのトランザクションを再発行できるように常に準備しておきます。デッドロックは危険ではありません。再度試してください。

他にも

分離レベルを下げる
テーブルロックをする

という方法がありますが、影響範囲を考えると デッドロックした場合はリトライ処理をする と統一した方が汎用性が高い気がします。

Go gormを使ったトランザクション

Go言語とgormを使ってまずはトランザクションを実装したいと思います。

以下、詳細は省きますがクリーンアーキテクチャなどで用いられる、interfaceを用いてトランザクションを貼るパターンです。第二引数fnにSQLとビジネスロジックが入る想定です。今回はtweetのカウントと作成などがこの引数に入ってきます。

func (t *txRepository) WithTx(ctx context.Context, fn func(context.Context) error) error {

        db := t.db.NewConn(ctx)
        tx := db.Begin()

        ctx = context.WithValue(ctx, &txKey, tx)

        // fnの中でtweetのカウント、作成などを行なっている
        err := fn(ctx)

        if err != nil {
            if dbErr := tx.Rollback().Error; dbErr != nil {
                err = dbErr
            }
        } else {
            if dbErr := tx.Commit().Error; dbErr != nil {
                err = dbErr
            }
        }

        return err
}

呼び出す際は以下のようにします。

func (us *tweetInteractor) CreateTweet(ctx context.Context, tweet *tweet.Tweet) (*tweet.Tweet, error) {


    var newTweet *tweet.Tweet
    //db transaction
    err = us.tx.WithTx(ctx, func(ctx context.Context) error {
         count, err = us.tweetRepository.CountTweetsByUserID(ctx, tweet.UserID)
                  if err != nil {
            return err
        }

                 if count >= 10 {
                   return errors.New("count over")
                 }

        newTweet, err = us.tweetRepository.Save(ctx, trs)
        if err != nil {
            return err
        }

        return nil
    })
    if err != nil {
        return nil, err
    }

  return newTweet, nil
}

以上がリトライ処理を行わないプレーンな実装です。このままだとデッドロックが起きた場合クライアントにはそのままエラーを返す事になります。

次にDBトランザクションの実装部分にリトライ処理を実装して行きたいと思います。

リトライ処理

リトライ処理は今回のようにDB間だけでなくマイクロサービスや連携APIをコールする時にも用いられます。APIを使用するサーバーでは外側にあるサービスの安全性は保証できない為です。その為、あらかじめ失敗した時を想定してリクエストを繰り返し送れるようにしておくのがリトライ処理という事です。

先ほどのトランザクション処理にシンプルにリトライ処理を実装してみます。

func (t *txRepository) WithTx(ctx context.Context, fn func(context.Context) error) error {
    txFn := func() error {
        db := t.db.NewConn(ctx)
        tx := db.Begin()

        ctx = context.WithValue(ctx, &txKey, tx)

        err := fn(ctx)
        if err != nil {
            if dbErr := tx.Rollback().Error; dbErr != nil {
                err = dbErr
            }
        } else {
            if dbErr := tx.Commit().Error; dbErr != nil {
                err = dbErr
            }
        }

        return err
    }


    // 初回の実行
    err := txFn()

    if isDeadLock(err) {
        retry := 3
        for i := 0; i < retry; i++ {
            err = txFn()
            if !isDeadLock(err) {
                break
            }
        }
    }

  


  return err
}


func isDeadLock(err error) bool {
    if err == nil {
        return false
    }

    deadlockErr := &ms.MySQLError{Number: 1213}
    if errors.As(err, &deadlockErr) {
        for err != nil {
            err = errors.Unwrap(err)
            if errors.Is(err, deadlockErr) {
                return true
            }
        }
    }

    return false
}

isDeadLock関数を用いてDBで発生したエラーがデッドロックによるものかを判定しています。 DB操作ではデッドロック以外にも様々なエラー(not foundなど)が発生する可能性がある為、この関数がないといつでもリトライしてしまいます。

for文では3回の上限を設け、デッドロック以外のエラーになった場合はbreakしています。

非常にシンプルですが一応実装することはできました。

retry-goを使う

上記でリトライ処理を実装しましたが実務レベルではもう少し複雑な要件が求められます。例えばリトライ回数を追うごとにリクエストする間隔を増やしたり、全く同じタイミングでリトライされないようにランダムに間隔をずらすなどです。すべてを自前で実装してもよいのですが、宣言的に簡潔に記述できるretry-goというパッケージを見つけたので使ってみたいと思います。

github.com

以下はretry-goを使った実装です。


func (t *txRepository) WithTx(ctx context.Context, fn func(context.Context) error) error {
    txFn := func() error {
        db := t.db.NewConn(ctx)
        tx := db.Begin()

        ctx = context.WithValue(ctx, &txKey, tx)

        err := fn(ctx)
        if err != nil {
            if dbErr := tx.Rollback().Error; dbErr != nil {
                err = dbErr
            }
        } else {
            if dbErr := tx.Commit().Error; dbErr != nil {
                err = dbErr
            }
        }

        return err
    }


    err := retry.Do(
        // トランザクション関数
        txFn,
        // デッドロックの時のみリトライ
        retry.RetryIf(func(err error) bool {
            return isDeadLock(err)
        }),
        // リトライ回数✖︎1秒づつリクエストを遅らせる
        retry.DelayType(func(n uint, err error, config *retry.Config) time.Duration {
            return time.Duration(n) * time.Second
        }),
        // 最大3回リトライ 初回1＋リトライ3
        retry.Attempts(4),
        // サーバー内のerror型で返す
        retry.LastErrorOnly(true),
    )

  return err
}


func isDeadLock(err error) bool {
// ...

上記のretry.Do()の中では以下を行なっています。

isDeadLock == trueの時のみリトライする
リトライ回数×1秒間ずつリクエスト処理までの時間を待機する
最大3回リトライする初回1＋3 = 4

LastErrorOnly(true)はパッケージ独自のエラー型に依存しないようにしています。

以上とても少ないコード量で複雑なリトライ処理を実装することができました。他にも便利なコールバック関数が用意されているのでぜひチェックしてみてください。

MySQLにおけるデッドロックとgo, gorm, retry-go, を用いたリトライ処理の実装は以上となります。

まとめ

デッドロックにまつわる仕組みからリトライ処理の実装までを行なってきましたが、いかがでしたでしょうか。単純にトランザクションを貼っただけでは整合性を担保できなという事からはじまり、ロックの仕組みを調べ、それにどう対応するか選定していくという一連の流れを行いました。

同じようなパターンで調べると、分離レベルを変更したりテーブルロックをかけるという対応が散見されましたが、そもそもデッドロックはごく稀に起こるパターンだと思うので、私はその解決の為だけにDBエンジンの設定を変更したり、特定のビジネスロジックのためにテーブルロックをかけるということをしたくありませんでした。

リトライ処理を実装しておけばDB部分の設定を意識せずデッドロックにも対応できるのではないでしょうか。

今回は以上です。最後に、このブログではweb開発について発信していくのでまたご覧頂けると嬉しいです。最後までお読み頂きありがとうございました。