こんにちは。KOUKIです。
Go言語でアプリケーション開発手法を紹介しています。
前回は、トランザクション処理の実装について触れました。
今日は、デッドロックについて学びたいと思います。
デッドロックについて
DB操作時にトランザクションを利用した更新があると「デッドロック」が発生する可能性があります。
DBのレコード更新時、プログラムは他のプログラムがデータを更新できないようにロックをかけます。そして更新が完了したあと、このロックを解除します。
他のプログラムはこのロックが解除されるまで、レコードにアクセスすることができません。
しかし、複数のプログラムがそれぞれのレコードに対してロックをかけ、ロックをかけたお互いのレコードも更新する必要があった場合、どうなるでしょうか？
答えは「お互いにロックが解除されるまで、処理をストップする」です。この状態をデッドロックと呼びます。
トランザクションの処理は、「複数の処理を一つにまとめる処理」です。各工程が全て完了した時にレコードのロックが解除されるのですが、複数のトランザクションが同時に走るとデッドロックが発生する確率は高くなります。
デッドロックを発生させる
説明を聞いても「?」という感じになると思うので、実際にデッドロックを発生させてみましょう。
アカウントの更新処理
前回学習したトランザクション処理に「アカウントの更新処理」を追加しましょう。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 |
// db/sqlc/store.go package db ... func (store *Store) execTx(ctx context.Context, fn func(*Queries) error) error { ... // 処理4 ---- ここから追加 account1, err := q.GetAccount(ctx, arg.FromAccountID) if err != nil { return err } result.FromAccount, err = q.UpdateAccount(ctx, UpdateAccountParams{ ID: arg.FromAccountID, Balance: account1.Balance - arg.Amount, }) if err != nil { return err } account2, err := q.GetAccount(ctx, arg.ToAccountID) if err != nil { return err } result.ToAccount, err = q.UpdateAccount(ctx, UpdateAccountParams{ ID: arg.ToAccountID, Balance: account2.Balance + arg.Amount, }) if err != nil { return err } return nil }) return result, err } |
Aさんの口座からBさんの口座へ入金する処理(Update)を実装しました。
口座確認テストの追加
続いて、前回作成したテストコードに口座確認テストを追加します。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 |
// store_test.go package db import ( "context" "fmt" "testing" "github.com/stretchr/testify/require" ) func TestTransferTx(t *testing.T) { store := NewStore(testDB) account1 := createRandomAccount(t) account2 := createRandomAccount(t) fmt.Println(">> before:", account1.Balance, account2.Balance) n := 5 amount := int64(10) errs := make(chan error) results := make(chan TransferTxResult) // check result ====== 追加 ===== existed := make(map[int]bool) for i := 0; i < n; i++ { go func() { result, err := store.TransferTx(context.Background(), TransferTxParams{ FromAccountID: account1.ID, ToAccountID: account2.ID, Amount: amount, }) errs <- err results <- result }() } for i := 0; i < n; i++ { err := <-errs require.NoError(t, err) result := <-results require.NotEmpty(t, result) // Transferのチェック transfer := result.Transfer require.NotEmpty(t, transfer) require.Equal(t, account1.ID, transfer.FromAccountID) require.Equal(t, account2.ID, transfer.ToAccountID) require.Equal(t, amount, transfer.Amount) require.NotZero(t, transfer.ID) require.NotZero(t, transfer.CreatedAt) _, err = store.GetTransfer(context.Background(), transfer.ID) require.NoError(t, err) // Entryのチェック fromEntry := result.FromEntry require.NotEmpty(t, fromEntry) require.Equal(t, account1.ID, fromEntry.AccountID) require.Equal(t, -amount, fromEntry.Amount) require.NotZero(t, fromEntry.ID) require.NotZero(t, fromEntry.CreatedAt) _, err = store.GetEntry(context.Background(), fromEntry.ID) require.NoError(t, err) toEntry := result.ToEntry require.NotEmpty(t, toEntry) require.Equal(t, account2.ID, toEntry.AccountID) require.Equal(t, amount, toEntry.Amount) require.NotZero(t, toEntry.ID) require.NotZero(t, toEntry.CreatedAt) _, err = store.GetEntry(context.Background(), toEntry.ID) require.NoError(t, err) // check accounts ====== 追加 ===== fromAccount := result.FromAccount require.NotEmpty(t, fromAccount) require.Equal(t, account1.ID, fromAccount.ID) toAccount := result.ToAccount require.NotEmpty(t, toAccount) require.Equal(t, account2.ID, toAccount.ID) // check account's balance ====== 追加 ===== fmt.Println(">> tx:", fromAccount.Balance, toAccount.Balance) diff1 := account1.Balance - fromAccount.Balance diff2 := toAccount.Balance - account2.Balance require.Equal(t, diff1, diff2) require.True(t, diff1 > 0) require.True(t, diff1%amount == 0) k := int(diff1 / amount) require.True(t, k >= 1 && k <= n) require.NotContains(t, existed, k) existed[k] = true } // check the final updated balances ====== 追加 ===== updatedAccount1, err := testQueries.GetAccount(context.Background(), account1.ID) require.NoError(t, err) updatedAccount2, err := testQueries.GetAccount(context.Background(), account2.ID) require.NoError(t, err) fmt.Println(">> after:", updatedAccount1.Balance, updatedAccount2.Balance) require.Equal(t, account1.Balance-int64(n)*amount, updatedAccount1.Balance) require.Equal(t, account2.Balance+int64(n)*amount, updatedAccount2.Balance) } |
テストを実行します。※docker-compose upでコンテナを立ち上げてください
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 |
$ make test === RUN TestTransferTx >> before: 420 938 >> tx: 410 948 store_test.go:103: Error Trace: store_test.go:103 Error: Not equal: expected: 10 actual : -10 Test: TestTransferTx --- FAIL: TestTransferTx (0.02s) FAIL coverage: 59.6% of statements FAIL golang-with-postgres/db/sqlc 0.190s ? golang-with-postgres/util [no test files] FAIL make: *** [Makefile:29: test] Error 1 |
テストが失敗しました。入出金の調整に問題がありそうです。
sqlを更新する
レコードの金額を取得するSQL文は以下のようになっています。
|
1 2 |
SELECT * FROM accounts WHERE id = $1 LIMIT 1; |
他のプログラムのトランザクション中に、このSQL文でレコードを取得すると更新前のデータが取得されます。
そのため、更新済みのレコードに対してUpdateをかけることができないので、データの更新に失敗するようです。
account.sqlに次のSQL文を追加してください。
|
1 2 3 4 |
-- name: GetAccountForUpdate :one SELECT * FROM accounts WHERE id = $1 LIMIT 1 FOR UPDATE; |
「FOR UPDATE」を追加すると、トランザクションを検知してデータの取得を待つことができます。すごく便利です。
トランザクションが完了後にデータを取得するので、先ほどのテストエラーは回避できるはずです。
次に以下のコマンドを実行して、SQLからGo言語のプログラムを作成します。
|
1 2 |
$ make sqlc sqlc generate |
sqlcの詳しい使い方については、この記事を参考にしてください。
上記のコマンドの実行により、account.sql.goには以下のメソッドが生成されました。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 |
const getAccountForUpdate = `-- name: GetAccountForUpdate :one SELECT id, owner, balance, currency, created_at FROM accounts WHERE id = $1 LIMIT 1 FOR UPDATE ` func (q *Queries) GetAccountForUpdate(ctx context.Context, id int64) (Account, error) { row := q.db.QueryRowContext(ctx, getAccountForUpdate, id) var i Account err := row.Scan( &i.ID, &i.Owner, &i.Balance, &i.Currency, &i.CreatedAt, ) return i, err } |
いよいよデッドロックへ
先ほど作成したGetAccountForUpdate関数を導入しましょう。store.goを修正します。
|
1 2 3 4 5 6 |
// 修正 GetAccount -> GetAccountForUpdateに変更 account1, err := q.GetAccountForUpdate(ctx, arg.FromAccountID) ... account2, err := q.GetAccountForUpdate(ctx, arg.ToAccountID) ... |
それでは、テストを実行します。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 |
$ make test === RUN TestTransferTx >> before: 405 671 store_test.go:51: Error Trace: store_test.go:51 Error: Received unexpected error: pq: deadlock detected <<<< Test: TestTransferTx --- FAIL: TestTransferTx (1.03s) FAIL coverage: 50.0% of statements FAIL golang-with-postgres/db/sqlc 1.108s ? golang-with-postgres/util [no test files] FAIL make: *** [Makefile:29: test] Error 1 |
「pq: deadlock detected」と表示されました！
ついに出ましたね！デッドロックが！！！
デッドロックの可視化
まず、デッドロックが発生している状況を探りましょう。
store_test.goに次の処理を実装してください。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 |
/ store_test.go ... func TestTransferTx(t *testing.T) { // 状況をわかりやすくする為、5 -> 2に変更 n := 2 amount := int64(10) ... for i := 0; i < n; i++ { txName := fmt.Sprintf("tx %d", i+1) go func() { ctx := context.WithValue(context.Background(), txKey, txName) result, err := store.TransferTx(ctx, TransferTxParams{ FromAccountID: account1.ID, ToAccountID: account2.ID, Amount: amount, }) ... } |
状況を整理する為、トランザクションの処理を5 -> 2に変更しました。そして、Contextを確認する為、contextパッケージのWithValueを使ってKeyの受け渡しを行なっています。
WithValueの使い方については、この記事を参考にしてください。
WithValueで生成したコンテキストをTransferTxメソッドに渡すことで、呼び出し先でKeyを取得できます。
store.goにKeyの取り出しとPrint文を追加しましょう。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 |
// store.go ... // 空のStructを定義 var txKey = struct{}{} func (store *Store) TransferTx(ctx context.Context, arg TransferTxParams) (TransferTxResult, error) { ... // contextからKeyを受け取る txName := ctx.Value(txKey) fmt.Println(txName, " create transfer") result.Transfer, err = q.CreateTransfer(ctx, CreateTransferParams{ ... }) ... fmt.Println(txName, " create entry 1") result.FromEntry, err = q.CreateEntry(ctx, CreateEntryParams{ ... }) fmt.Println(txName, " create entry 2") result.ToEntry, err = q.CreateEntry(ctx, CreateEntryParams{ ... }) fmt.Println(txName, "get account 1") account1, err := q.GetAccountForUpdate(ctx, arg.FromAccountID) fmt.Println(txName, "update account 1") result.FromAccount, err = q.UpdateAccount(ctx, UpdateAccountParams{ ... }) fmt.Println(txName, "get account 2") account2, err := q.GetAccountForUpdate(ctx, arg.ToAccountID) fmt.Println(txName, "update account 2") result.ToAccount, err = q.UpdateAccount(ctx, UpdateAccountParams{ }) ... }) |
テストを実行します。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 |
make test >> before: 159 654 tx 2 create transfer tx 1 create transfer tx 2 create entry 1 tx 1 create entry 1 tx 2 create entry 2 tx 1 create entry 2 tx 2 get account 1 tx 1 get account 1 tx 1 update account 1 --- FAIL: TestTransferTx (1.04s) /Users/hoge/go/src/github.com/hoge/golang-with-postgres/db/sqlc/store_test.go:46: Error Trace: store_test.go:46 Error: Received unexpected error: pq: deadlock detected Test: TestTransferTx FAIL FAIL golang-with-postgres/db/sqlc 1.058s FAIL |
「tx 1 update account 1」の出力後にデッドロックが発生していますね。
しかし、これだけでは何が起こっているかわかりません。Terminalを2つ開いて、Dockerコンテナ内に入ってみましょう。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 |
// Terminal1: コンテナ内に入る docker exec -it postgres psql -U postgres -d simplebank // Terminal2: コンテナ内に入る docker exec -it postgres psql -U postgres -d simplebank // Terminal1: Transaction開始 BEGIN; // Terminal2: Transaction開始 BEGIN; // Terminal2: tx 2 create transfer INSERT INTO transfers ( from_account_id, to_account_id, amount) VALUES (1, 2, 3) RETURNING *; // Terminal1: tx 1 create transfer INSERT INTO transfers ( from_account_id, to_account_id, amount) VALUES (1, 2, 3) RETURNING *; // Terminal2: tx 2 create entry 2 INSERT INTO entries (account_id, amount) VALUES (1, -10) RETURNING *; // Terminal1: tx 1 create entry 1 INSERT INTO entries (account_id, amount) VALUES (2, 10) RETURNING *; // Terminal2: tx 2 create entry 2 INSERT INTO entries (account_id, amount) VALUES (1, -10) RETURNING *; // Terminal1: tx 1 create entry 2 INSERT INTO entries (account_id, amount) VALUES (2, 10) RETURNING *; // Terminal2: tx 2 get account 1(ブロックが走る) SELECT * FROM accounts WHERE id = 1 LIMIT 1 FOR UPDATE; // Terminal1: tx 1 get account 1 SELECT * FROM accounts WHERE id = 1 LIMIT 1 FOR UPDATE; // デッドロック発生 INSERT 0 1 simplebank=# SELECT * FROM accounts WHERE id = 1 LIMIT 1 FOR UPDATE; ERROR: deadlock detected DETAIL: Process 125 waits for ShareLock on transaction 893; blocked by process 131. Process 131 waits for ShareLock on transaction 894; blocked by process 125. HINT: See server log for query details. CONTEXT: while locking tuple (0,1) in relation "accounts" // Terminal1: tx 1 update account 1 // 後片付け // Treminal1: ロールバック rollback; // Treminal2: ロールバック rollback; |
「SELECT * FROM accounts WHERE id = 1 LIMIT 1 FOR UPDATE;」でブロックが入り、その後、デッドロックが発生しました。Postgresはデッドロックを検知してくれるので便利ですね。
「id = 1 」はリレーションを張っているので、これが関係しそうです。
Postgresは、pg_locksを参照するとロック状態を確認することができるようです。
このサイトを参照してください。
「http://localhost:9232/」にアクセスし、次のSQL文を実行します。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 |
SELECT blocked_locks.pid AS blocked_pid, blocked_activity.usename AS blocked_user, blocking_locks.pid AS blocking_pid, blocking_activity.usename AS blocking_user, blocked_activity.query AS blocked_statement, blocking_activity.query AS current_statement_in_blocking_process FROM pg_catalog.pg_locks blocked_locks JOIN pg_catalog.pg_stat_activity blocked_activity ON blocked_activity.pid = blocked_locks.pid JOIN pg_catalog.pg_locks blocking_locks ON blocking_locks.locktype = blocked_locks.locktype AND blocking_locks.database IS NOT DISTINCT FROM blocked_locks.database AND blocking_locks.relation IS NOT DISTINCT FROM blocked_locks.relation AND blocking_locks.page IS NOT DISTINCT FROM blocked_locks.page AND blocking_locks.tuple IS NOT DISTINCT FROM blocked_locks.tuple AND blocking_locks.virtualxid IS NOT DISTINCT FROM blocked_locks.virtualxid AND blocking_locks.transactionid IS NOT DISTINCT FROM blocked_locks.transactionid AND blocking_locks.classid IS NOT DISTINCT FROM blocked_locks.classid AND blocking_locks.objid IS NOT DISTINCT FROM blocked_locks.objid AND blocking_locks.objsubid IS NOT DISTINCT FROM blocked_locks.objsubid AND blocking_locks.pid != blocked_locks.pid JOIN pg_catalog.pg_stat_activity blocking_activity ON blocking_activity.pid = blocking_locks.pid WHERE NOT blocked_locks.granted; |
実行結果のレコードの通り、「SELECT * FROM accounts WHERE id = 1 LIMIT 1 FOR UPDATE;」によってブロックされていることがわかりますね。
そして、その理由は「INSERT INTO transfers ( from_account_id, to_account_id, amount) VALUES (1, 2, 3) RETURNING *;」が実行されているからとのこと。
これだけだとよくわからないので、次のSQL文も実行してみましょう。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 |
select a.application_name, l.relation::regclass, l.transactionid, l.mode, l.locktype, l.granted, a.usename, a.query, a.pid from pg_stat_activity a join pg_locks l on l.pid = a.pid where a.application_name = 'psql' order by a.pid; |
ロックの一覧を出しました。
この中で、「ShareLock」が気になる存在です。
５レコード目の「INSERT INTO transfers ( from_account_id, to_account_id, amount) VALUES (1, 2, 3) RETURNING *;」を実行後、「SELECT * FROM accounts WHERE id = 1 LIMIT 1 FOR UPDATE;」の実行でShareLockが走っているので、やはりリレーションに問題がありそうです。
リレーションの設定は、migrationフォルダの000001_init_schema.up.sqlに記述しています。
|
1 2 3 4 5 |
ALTER TABLE "entries" ADD FOREIGN KEY ("account_id") REFERENCES "accounts" ("id"); ALTER TABLE "transfers" ADD FOREIGN KEY ("from_account_id") REFERENCES "accounts" ("id"); ALTER TABLE "transfers" ADD FOREIGN KEY ("to_account_id") REFERENCES "accounts" ("id"); |
この設定を見ると「id, from_account_id, to_account_id」でリレーションを張っていますね。Insert/Select両方ともidを必要とするので、これによりShareLockが発生したようです。
デッドロックを回避する
リレーションに問題があることがわかったので、以下のクエリを修正しましょう。
|
1 2 3 4 5 |
// account.sql -- name: GetAccountForUpdate :one SELECT * FROM accounts WHERE id = $1 LIMIT 1 FOR NO KEY UPDATE; |
ロックを回避するために、「NO KEY」を追加しました。
公式サイトによると
「獲得するロックが弱い以外はFOR UPDATEと同じように振る舞います。このロックは同じ行のロックを獲得しようとするSELECT FOR KEY SHAREコマンドをブロックしません。 このロックモードはFOR UPDATEロックを獲得しないUPDATEによっても獲得されます。」
とあるので、これで行レベルでのロックを回避できるはずです。
以下のコマンドを実行しましょう。
|
1 2 |
$ make sqlc sqlc generate |
テストを実行します。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 |
$ make test go test -v -cover ./... ? golang-with-postgres [no test files] ... === RUN TestTransferTx >> before: 386 533 tx 2 create transfer tx 2 create entry 1 tx 1 create transfer tx 2 create entry 2 tx 1 create entry 1 tx 2 get account 1 tx 1 create entry 2 tx 1 get account 1 tx 2 update account 1 tx 2 get account 2 tx 2 update account 2 tx 1 update account 1 tx 1 get account 2 tx 1 update account 2 >> tx: 376 543 >> tx: 366 553 >> after: 366 553 --- PASS: TestTransferTx (0.04s) PASS coverage: 63.5% of statements ok golang-with-postgres/db/sqlc 0.127s coverage: 63.5% of statements ? golang-with-postgres/util [no test files] |
OKですね。検証が完了したのでPrint文やコンテキストを削除しておきましょう。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 |
// store.go package db import ( "context" "database/sql" "fmt" ) type Store struct { *Queries db *sql.DB } func NewStore(db *sql.DB) *Store { return &Store{ db: db, Queries: New(db), } } func (store *Store) execTx(ctx context.Context, fn func(*Queries) error) error { tx, err := store.db.BeginTx(ctx, nil) if err != nil { return err } q := New(tx) err = fn(q) if err != nil { if rbErr := tx.Rollback(); rbErr != nil { return fmt.Errorf("tx err: %v, rb err: %v", err, rbErr) } return err } return tx.Commit() } type TransferTxParams struct { FromAccountID int64 `json:"from_account_id"` ToAccountID int64 `json:"to_account_id"` Amount int64 `json:"amount"` } type TransferTxResult struct { Transfer Transfer `json:"transfer"` FromAccount Account `json:"from_account"` ToAccount Account `json:"to_account"` FromEntry Entry `json:"from_entry"` ToEntry Entry `json:"to_entry"` } func (store *Store) TransferTx(ctx context.Context, arg TransferTxParams) (TransferTxResult, error) { var result TransferTxResult err := store.execTx(ctx, func(q *Queries) error { var err error result.Transfer, err = q.CreateTransfer(ctx, CreateTransferParams{ FromAccountID: arg.FromAccountID, ToAccountID: arg.ToAccountID, Amount: arg.Amount, }) if err != nil { return err } result.FromEntry, err = q.CreateEntry(ctx, CreateEntryParams{ AccountID: arg.FromAccountID, Amount: -arg.Amount, }) if err != nil { return err } result.ToEntry, err = q.CreateEntry(ctx, CreateEntryParams{ AccountID: arg.ToAccountID, Amount: arg.Amount, }) if err != nil { return err } account1, err := q.GetAccountForUpdate(ctx, arg.FromAccountID) if err != nil { return err } result.FromAccount, err = q.UpdateAccount(ctx, UpdateAccountParams{ ID: arg.FromAccountID, Balance: account1.Balance - arg.Amount, }) if err != nil { return err } account2, err := q.GetAccountForUpdate(ctx, arg.ToAccountID) if err != nil { return err } result.ToAccount, err = q.UpdateAccount(ctx, UpdateAccountParams{ ID: arg.ToAccountID, Balance: account2.Balance + arg.Amount, }) if err != nil { return err } return nil }) return result, err } |
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 |
// store_test.go package db import ( "context" "fmt" "testing" "github.com/stretchr/testify/require" ) func TestTransferTx(t *testing.T) { store := NewStore(testDB) account1 := createRandomAccount(t) account2 := createRandomAccount(t) fmt.Println(">> before:", account1.Balance, account2.Balance) // 5 -> 2に変更 n := 2 amount := int64(10) errs := make(chan error) results := make(chan TransferTxResult) existed := make(map[int]bool) for i := 0; i < n; i++ { go func() { ctx := context.Background() result, err := store.TransferTx(ctx, TransferTxParams{ FromAccountID: account1.ID, ToAccountID: account2.ID, Amount: amount, }) errs <- err results <- result }() } for i := 0; i < n; i++ { err := <-errs require.NoError(t, err) result := <-results require.NotEmpty(t, result) transfer := result.Transfer require.NotEmpty(t, transfer) require.Equal(t, account1.ID, transfer.FromAccountID) require.Equal(t, account2.ID, transfer.ToAccountID) require.Equal(t, amount, transfer.Amount) require.NotZero(t, transfer.ID) require.NotZero(t, transfer.CreatedAt) _, err = store.GetTransfer(context.Background(), transfer.ID) require.NoError(t, err) fromEntry := result.FromEntry require.NotEmpty(t, fromEntry) require.Equal(t, account1.ID, fromEntry.AccountID) require.Equal(t, -amount, fromEntry.Amount) require.NotZero(t, fromEntry.ID) require.NotZero(t, fromEntry.CreatedAt) _, err = store.GetEntry(context.Background(), fromEntry.ID) require.NoError(t, err) toEntry := result.ToEntry require.NotEmpty(t, toEntry) require.Equal(t, account2.ID, toEntry.AccountID) require.Equal(t, amount, toEntry.Amount) require.NotZero(t, toEntry.ID) require.NotZero(t, toEntry.CreatedAt) _, err = store.GetEntry(context.Background(), toEntry.ID) require.NoError(t, err) fromAccount := result.FromAccount require.NotEmpty(t, fromAccount) require.Equal(t, account1.ID, fromAccount.ID) toAccount := result.ToAccount require.NotEmpty(t, toAccount) require.Equal(t, account2.ID, toAccount.ID) fmt.Println(">> tx:", fromAccount.Balance, toAccount.Balance) diff1 := account1.Balance - fromAccount.Balance diff2 := toAccount.Balance - account2.Balance require.Equal(t, diff1, diff2) require.True(t, diff1 > 0) require.True(t, diff1%amount == 0) k := int(diff1 / amount) require.True(t, k >= 1 && k <= n) require.NotContains(t, existed, k) existed[k] = true } updatedAccount1, err := testQueries.GetAccount(context.Background(), account1.ID) require.NoError(t, err) updatedAccount2, err := testQueries.GetAccount(context.Background(), account2.ID) require.NoError(t, err) fmt.Println(">> after:", updatedAccount1.Balance, updatedAccount2.Balance) require.Equal(t, account1.Balance-int64(n)*amount, updatedAccount1.Balance) require.Equal(t, account2.Balance+int64(n)*amount, updatedAccount2.Balance) } |
リファクタリング – 口座残高計算クエリの作成
最後に、口座残高計算クエリを作成して終わりましょう。
|
1 2 3 4 5 |
-- name: AddAccountBalance :one UPDATE accounts SET balance = balance + sqlc.arg(amount) WHERE id = sqlc.arg(id) RETURNING *; |
このクエリにより、計算処理を実装する必要が無くなります。
次のコマンドを実行します。
|
1 2 |
$ make sqlc sqlc generate |
このコマンドの実行によって、以下のGoプログラムが作成されます。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 |
const addAccountBalance = `-- name: AddAccountBalance :one UPDATE accounts SET balance = balance + $1 WHERE id = $2 RETURNING id, owner, balance, currency, created_at ` type AddAccountBalanceParams struct { Amount int64 `json:"amount"` ID int64 `json:"id"` } func (q *Queries) AddAccountBalance(ctx context.Context, arg AddAccountBalanceParams) (Account, error) { row := q.db.QueryRowContext(ctx, addAccountBalance, arg.Amount, arg.ID) var i Account err := row.Scan( &i.ID, &i.Owner, &i.Balance, &i.Currency, &i.CreatedAt, ) return i, err } |
store.goのTransferTxメソッドを修正します。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 |
... func (store *Store) TransferTx(ctx context.Context, arg TransferTxParams) (TransferTxResult, error) { var result TransferTxResult err := store.execTx(ctx, func(q *Queries) error { var err error result.Transfer, err = q.CreateTransfer(ctx, CreateTransferParams{ FromAccountID: arg.FromAccountID, ToAccountID: arg.ToAccountID, Amount: arg.Amount, }) if err != nil { return err } result.FromEntry, err = q.CreateEntry(ctx, CreateEntryParams{ AccountID: arg.FromAccountID, Amount: -arg.Amount, }) if err != nil { return err } result.ToEntry, err = q.CreateEntry(ctx, CreateEntryParams{ AccountID: arg.ToAccountID, Amount: arg.Amount, }) if err != nil { return err } // リファクタリング result.FromAccount, err = q.AddAccountBalance(ctx, AddAccountBalanceParams{ ID: arg.FromAccountID, Amount: -arg.Amount, }) if err != nil { return err } // リファクタリング result.ToAccount, err = q.AddAccountBalance(ctx, AddAccountBalanceParams{ ID: arg.ToAccountID, Amount: arg.Amount, }) if err != nil { return err } return nil }) return result, err } |
だいぶコードがスッキリしましたね。
テストを実行します。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 |
$ make test go test -v -cover ./... ? golang-with-postgres [no test files] === RUN TestCreateAccount --- PASS: TestCreateAccount (0.01s) === RUN TestGetAccount --- PASS: TestGetAccount (0.00s) === RUN TestUpdateAccount --- PASS: TestUpdateAccount (0.00s) === RUN TestDeleteAccount --- PASS: TestDeleteAccount (0.01s) === RUN TestListAccounts --- PASS: TestListAccounts (0.02s) === RUN TestTransferTx >> before: 359 478 >> tx: 349 488 >> tx: 339 498 >> after: 339 498 --- PASS: TestTransferTx (0.03s) PASS coverage: 58.6% of statements ok golang-with-postgres/db/sqlc 0.145s coverage: 58.6% of statements ? golang-with-postgres/util [no test files] |
次回
今回の学習内容はすごく濃かったですが、ロックを発見する方法を学べたのはよかったですね^^
また、トランザクションを開始するにはBEGINを使い、ロールバックするにはROLLBACKを使うことも知ることができました(あまりPostgresを触らないんですよ)。
次回は、デットロックの回避方法について学びましょう。
にほんブログ村
コメントを残す
コメントを投稿するにはログインしてください。