From f787421f82f1d0a46e99de527b61e2634f36f3ed Mon Sep 17 00:00:00 2001 From: hoteas <925970985@qq.com> Date: Sat, 11 Apr 2026 21:53:28 +0800 Subject: [PATCH] =?UTF-8?q?fix(db):=20=E4=BF=AE=E5=A4=8D=E4=BA=8B=E5=8A=A1?= =?UTF-8?q?=E4=B8=ADSQL=E9=94=99=E8=AF=AF=E6=97=B6=E7=9A=84=E5=9B=9E?= =?UTF-8?q?=E6=BB=9A=E6=9C=BA=E5=88=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 添加txFailed标志跟踪事务内的SQL错误状态 - 在事务模式下遇到SQL错误时立即标记事务失败并阻止重试 - 修改Action方法确保SQL出错时强制回滚事务 - 为事务失败场景添加完整的单元测试覆盖 - 防止在事务已回滚的情况下继续执行后续操作导致数据不一致 --- db/db.go | 5 +- db/query.go | 35 +++++++++-- db/transaction.go | 39 +++++++++--- db/transaction_test.go | 135 +++++++++++++++++++++++++++++++++++++++++ 4 files changed, 197 insertions(+), 17 deletions(-) create mode 100644 db/transaction_test.go diff --git a/db/db.go b/db/db.go index 15f99f5..95e0b58 100644 --- a/db/db.go +++ b/db/db.go @@ -31,9 +31,10 @@ type HoTimeDB struct { Mode int // mode为0生产模式,1为测试模式,2为开发模式 mu sync.RWMutex limitMu sync.Mutex - Dialect Dialect // 数据库方言适配器 - testTx *sql.Tx // 测试事务：设置后所有操作都在此事务内，测试结束回滚 + Dialect Dialect // 数据库方言适配器 + testTx *sql.Tx // 测试事务：设置后所有操作都在此事务内，测试结束回滚 testMu *sync.Mutex // 保护 testTx 单连接不被并发访问 + txFailed *bool // 事务内是否有SQL出错，出错则事务必须回滚（指针确保按值传递 HoTimeDB 时状态共享） } // SetConnect 设置数据库配置连接 diff --git a/db/query.go b/db/query.go index 052495d..5844d13 100644 --- a/db/query.go +++ b/db/query.go @@ -62,15 +62,21 @@ func (that *HoTimeDB) queryWithRetry(query string, retried bool, args ...interfa processedArgs := that.processArgs(args) if that.testTx != nil { - if that.testMu != nil { that.testMu.Lock() } + if that.testMu != nil { + that.testMu.Lock() + } resl, err = that.testTx.Query(query, processedArgs...) that.LastErr.SetError(err) if err != nil { - if that.testMu != nil { that.testMu.Unlock() } + if that.testMu != nil { + that.testMu.Unlock() + } return nil } result := that.Row(resl) - if that.testMu != nil { that.testMu.Unlock() } + if that.testMu != nil { + that.testMu.Unlock() + } return result } else if that.Tx != nil { resl, err = that.Tx.Query(query, processedArgs...) @@ -80,7 +86,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...) } @@ -128,9 +139,13 @@ func (that *HoTimeDB) execWithRetry(query string, retried bool, args ...interfac processedArgs := that.processArgs(args) if that.testTx != nil { - if that.testMu != nil { that.testMu.Lock() } + if that.testMu != nil { + that.testMu.Lock() + } resl, e = that.testTx.Exec(query, processedArgs...) - if that.testMu != nil { that.testMu.Unlock() } + if that.testMu != nil { + that.testMu.Unlock() + } } else if that.Tx != nil { resl, e = that.Tx.Exec(query, processedArgs...) } else { @@ -143,6 +158,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 65a57df..ff2aa77 100644 --- a/db/transaction.go +++ b/db/transaction.go @@ -35,21 +35,36 @@ func (that *HoTimeDB) Action(action func(db HoTimeDB) (isSuccess bool)) (isSucce testMu: that.testMu, } + 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 } @@ -65,6 +80,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) + } +}