diff --git a/db/query.go b/db/query.go index a321cb6..8aa153b 100644 --- a/db/query.go +++ b/db/query.go @@ -95,7 +95,12 @@ func (that *HoTimeDB) queryWithRetry(query string, retried bool, args ...interfa that.LastErr.SetError(err) if err != nil { - if !retried { + if that.Tx != nil { + if that.txFailed != nil { + *that.txFailed = true + } + } + if !retried && that.Tx == nil { if pingErr := db.Ping(); pingErr == nil { return that.queryWithRetry(query, true, args...) } @@ -169,6 +174,14 @@ func (that *HoTimeDB) execWithRetry(query string, retried bool, args ...interfac if that.testTx != nil { return resl, that.LastErr } + // 事务内不做连接级重试:死锁等错误会导致 MySQL 自动回滚事务, + // 在已回滚的 Tx 上重试会以 auto-commit 模式执行,造成数据不一致 + if that.Tx != nil { + if that.txFailed != nil { + *that.txFailed = true + } + return resl, that.LastErr + } if !retried { if pingErr := that.DB.Ping(); pingErr == nil { return that.execWithRetry(query, true, args...) diff --git a/db/transaction.go b/db/transaction.go index 8d4922c..d81cbd7 100644 --- a/db/transaction.go +++ b/db/transaction.go @@ -39,21 +39,36 @@ func (that *HoTimeDB) Action(action func(db HoTimeDB) (isSuccess bool)) (isSucce testLogBuf: logBuf, } + txFailed := false + db.txFailed = &txFailed + if that.testTx != nil { spName := fmt.Sprintf("sp_%d", atomic.AddUint64(&savepointCounter, 1)) - if that.testMu != nil { that.testMu.Lock() } + if that.testMu != nil { + that.testMu.Lock() + } _, _ = that.testTx.Exec("SAVEPOINT " + spName) - if that.testMu != nil { that.testMu.Unlock() } + if that.testMu != nil { + that.testMu.Unlock() + } db.Tx = that.testTx isSuccess = action(db) - if !isSuccess { - if that.testMu != nil { that.testMu.Lock() } + if txFailed || !isSuccess { + if that.testMu != nil { + that.testMu.Lock() + } _, _ = that.testTx.Exec("ROLLBACK TO SAVEPOINT " + spName) - if that.testMu != nil { that.testMu.Unlock() } - } else { - if that.testMu != nil { that.testMu.Lock() } - _, _ = that.testTx.Exec("RELEASE SAVEPOINT " + spName) - if that.testMu != nil { that.testMu.Unlock() } + if that.testMu != nil { + that.testMu.Unlock() + } + return false + } + if that.testMu != nil { + that.testMu.Lock() + } + _, _ = that.testTx.Exec("RELEASE SAVEPOINT " + spName) + if that.testMu != nil { + that.testMu.Unlock() } return isSuccess } @@ -69,6 +84,12 @@ func (that *HoTimeDB) Action(action func(db HoTimeDB) (isSuccess bool)) (isSucce isSuccess = action(db) + // SQL 出过错 → 事务必须回滚,不管回调返回什么 + if txFailed { + _ = db.Tx.Rollback() + return false + } + if !isSuccess { err = db.Tx.Rollback() if err != nil { diff --git a/db/transaction_test.go b/db/transaction_test.go new file mode 100644 index 0000000..565efbe --- /dev/null +++ b/db/transaction_test.go @@ -0,0 +1,135 @@ +package db + +import ( + . "code.hoteas.com/golang/hotime/common" + "database/sql" + "testing" + + "github.com/sirupsen/logrus" +) + +func newTestDB(t *testing.T) *HoTimeDB { + t.Helper() + sqlDB, err := sql.Open("sqlite3", ":memory:") + if err != nil { + t.Fatal(err) + } + logger := logrus.New() + logger.SetLevel(logrus.WarnLevel) + db := &HoTimeDB{ + DB: sqlDB, + Type: "sqlite3", + Dialect: &SQLiteDialect{}, + LastErr: &Error{Logger: logger}, + Log: logger, + } + _, execErr := sqlDB.Exec("CREATE TABLE test_item (id INTEGER PRIMARY KEY, name TEXT, value INTEGER)") + if execErr != nil { + t.Fatal(execErr) + } + _, execErr = sqlDB.Exec("INSERT INTO test_item (id, name, value) VALUES (1, 'foo', 100)") + if execErr != nil { + t.Fatal(execErr) + } + return db +} + +func TestAction_TxFailedForceRollback(t *testing.T) { + db := newTestDB(t) + defer db.DB.Close() + + result := db.Action(func(tx HoTimeDB) (isSuccess bool) { + tx.Update("test_item", Map{"value": 999}, Map{"id": 1}) + + tx.Exec("THIS IS INVALID SQL THAT WILL FAIL") + + return true + }) + + if result != false { + t.Fatalf("Action should return false when SQL error occurred, got true") + } + + row := db.Get("test_item", "value", Map{"id": 1}) + if row == nil { + t.Fatal("failed to read test_item") + } + val := row.GetCeilInt64("value") + if val != 100 { + t.Fatalf("value should be rolled back to 100, got %d", val) + } +} + +func TestAction_NormalCommit(t *testing.T) { + db := newTestDB(t) + defer db.DB.Close() + + result := db.Action(func(tx HoTimeDB) (isSuccess bool) { + tx.Update("test_item", Map{"value": 200}, Map{"id": 1}) + return true + }) + + if result != true { + t.Fatalf("Action should return true on success, got false") + } + + row := db.Get("test_item", "value", Map{"id": 1}) + if row == nil { + t.Fatal("failed to read test_item") + } + val := row.GetCeilInt64("value") + if val != 200 { + t.Fatalf("value should be committed to 200, got %d", val) + } +} + +func TestAction_NormalRollback(t *testing.T) { + db := newTestDB(t) + defer db.DB.Close() + + result := db.Action(func(tx HoTimeDB) (isSuccess bool) { + tx.Update("test_item", Map{"value": 300}, Map{"id": 1}) + return false + }) + + if result != false { + t.Fatalf("Action should return false, got true") + } + + row := db.Get("test_item", "value", Map{"id": 1}) + if row == nil { + t.Fatal("failed to read test_item") + } + val := row.GetCeilInt64("value") + if val != 100 { + t.Fatalf("value should be rolled back to 100, got %d", val) + } +} + +func TestAction_SqlErrorThenReturnTrue_MustRollback(t *testing.T) { + db := newTestDB(t) + defer db.DB.Close() + + result := db.Action(func(tx HoTimeDB) (isSuccess bool) { + tx.Update("test_item", Map{"value": 500}, Map{"id": 1}) + + tx.Exec("INSERT INTO nonexistent_table (x) VALUES (1)") + + tx.Update("test_item", Map{"value": 600}, Map{"id": 1}) + + return true + }) + + if result != false { + t.Fatalf("Action should return false when SQL error occurred mid-transaction, got true") + } + + row := db.Get("test_item", "value", Map{"id": 1}) + if row == nil { + t.Fatal("failed to read test_item") + } + val := row.GetCeilInt64("value") + if val != 100 { + t.Fatalf("value should be rolled back to 100 (original), got %d", val) + } +}