You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@seata.apache.org by te...@apache.org on 2026/04/07 13:21:14 UTC
(incubator-seata-go) branch master updated: fix: rollback auto-created AT tx on query and exec errors (#1083)
This is an automated email from the ASF dual-hosted git repository.
tew pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-seata-go.git
The following commit(s) were added to refs/heads/master by this push:
new b2bc7a29 fix: rollback auto-created AT tx on query and exec errors (#1083)
b2bc7a29 is described below
commit b2bc7a29a72cd7670e57176a9e51dcc2175c652b
Author: Zhifan C <50...@users.noreply.github.com>
AuthorDate: Tue Apr 7 21:21:09 2026 +0800
fix: rollback auto-created AT tx on query and exec errors (#1083)
Co-authored-by: Konnyaku <50...@users.noreply.github.com>
Co-authored-by: CocaElbow <sh...@outlook.com>
Co-authored-by: ThunGuo <te...@apache.org>
---
pkg/datasource/sql/conn_at.go | 19 +++++++-
pkg/datasource/sql/conn_at_test.go | 95 ++++++++++++++++++++++++++++++++++++++
2 files changed, 112 insertions(+), 2 deletions(-)
diff --git a/pkg/datasource/sql/conn_at.go b/pkg/datasource/sql/conn_at.go
index e19aa017..94697a9e 100644
--- a/pkg/datasource/sql/conn_at.go
+++ b/pkg/datasource/sql/conn_at.go
@@ -21,6 +21,7 @@ import (
"context"
gosql "database/sql"
"database/sql/driver"
+ "errors"
"seata.apache.org/seata-go/v2/pkg/datasource/sql/exec"
"seata.apache.org/seata-go/v2/pkg/datasource/sql/types"
@@ -192,7 +193,7 @@ func (c *ATConn) createTxAndExecIfNeeded(ctx context.Context, f func() (types.Ex
ret, err := f()
if err != nil {
- return nil, err
+ return nil, c.rollbackCreatedTx(tx, err)
}
// For ExecContext, commit the transaction if it was created
@@ -233,7 +234,7 @@ func (c *ATConn) createTxAndQueryIfNeeded(ctx context.Context, f func() (types.E
ret, err := f()
if err != nil {
- return nil, err
+ return nil, c.rollbackCreatedTx(tx, err)
}
// For QueryContext, wrap rows to commit on close
@@ -255,3 +256,17 @@ func (c *ATConn) createTxAndQueryIfNeeded(ctx context.Context, f func() (types.E
return ret, nil
}
+
+func (c *ATConn) rollbackCreatedTx(tx driver.Tx, execErr error) error {
+ if tx == nil || execErr == nil {
+ return execErr
+ }
+
+ rollbackErr := tx.Rollback()
+ if rollbackErr != nil {
+ log.Errorf("conn at rollback error:%v", rollbackErr)
+ return errors.Join(execErr, rollbackErr)
+ }
+
+ return execErr
+}
diff --git a/pkg/datasource/sql/conn_at_test.go b/pkg/datasource/sql/conn_at_test.go
index 1f2bad84..1b9e317b 100644
--- a/pkg/datasource/sql/conn_at_test.go
+++ b/pkg/datasource/sql/conn_at_test.go
@@ -21,6 +21,7 @@ import (
"context"
"database/sql"
"database/sql/driver"
+ "errors"
"sync/atomic"
"testing"
@@ -244,6 +245,100 @@ func TestATConn_BeginTx(t *testing.T) {
})
}
+func TestATConn_CreateTxHelpersRollbackOnError(t *testing.T) {
+ t.Run("createTxAndExecIfNeeded rolls back created tx on error", func(t *testing.T) {
+ ctrl := gomock.NewController(t)
+ defer ctrl.Finish()
+ CleanTxHooks()
+ defer CleanTxHooks()
+
+ mockTx := mock.NewMockTestDriverTx(ctrl)
+ mockTx.EXPECT().Rollback().Return(nil).Times(1)
+
+ mockConn := mock.NewMockTestDriverConn(ctrl)
+ mockConn.EXPECT().BeginTx(gomock.Any(), gomock.Any()).Return(mockTx, nil).Times(1)
+
+ atConn := &ATConn{
+ Conn: &Conn{
+ res: &DBResource{
+ dbType: types.DBTypeMySQL,
+ resourceID: "resource-id",
+ },
+ txCtx: &types.TransactionContext{
+ TransactionMode: types.ATMode,
+ },
+ targetConn: mockConn,
+ autoCommit: true,
+ },
+ }
+
+ var rollbackCnt int32
+ RegisterTxHook(&mockTxHook{
+ beforeRollback: func(tx *Tx) {
+ atomic.AddInt32(&rollbackCnt, 1)
+ },
+ })
+
+ ctx := tm.InitSeataContext(context.Background())
+ tm.SetXID(ctx, uuid.NewString())
+ expectedErr := errors.New("exec failed")
+
+ ret, err := atConn.createTxAndExecIfNeeded(ctx, func() (types.ExecResult, error) {
+ return nil, expectedErr
+ })
+
+ assert.Nil(t, ret)
+ assert.ErrorIs(t, err, expectedErr)
+ assert.Equal(t, int32(1), atomic.LoadInt32(&rollbackCnt))
+ })
+
+ t.Run("createTxAndQueryIfNeeded rolls back created tx on error", func(t *testing.T) {
+ ctrl := gomock.NewController(t)
+ defer ctrl.Finish()
+ CleanTxHooks()
+ defer CleanTxHooks()
+
+ mockTx := mock.NewMockTestDriverTx(ctrl)
+ mockTx.EXPECT().Rollback().Return(nil).Times(1)
+
+ mockConn := mock.NewMockTestDriverConn(ctrl)
+ mockConn.EXPECT().BeginTx(gomock.Any(), gomock.Any()).Return(mockTx, nil).Times(1)
+
+ atConn := &ATConn{
+ Conn: &Conn{
+ res: &DBResource{
+ dbType: types.DBTypeMySQL,
+ resourceID: "resource-id",
+ },
+ txCtx: &types.TransactionContext{
+ TransactionMode: types.ATMode,
+ },
+ targetConn: mockConn,
+ autoCommit: true,
+ },
+ }
+
+ var rollbackCnt int32
+ RegisterTxHook(&mockTxHook{
+ beforeRollback: func(tx *Tx) {
+ atomic.AddInt32(&rollbackCnt, 1)
+ },
+ })
+
+ ctx := tm.InitSeataContext(context.Background())
+ tm.SetXID(ctx, uuid.NewString())
+ expectedErr := errors.New("query failed")
+
+ ret, err := atConn.createTxAndQueryIfNeeded(ctx, func() (types.ExecResult, error) {
+ return nil, expectedErr
+ })
+
+ assert.Nil(t, ret)
+ assert.ErrorIs(t, err, expectedErr)
+ assert.Equal(t, int32(1), atomic.LoadInt32(&rollbackCnt))
+ })
+}
+
type mockTxHook struct {
beforeCommit func(tx *Tx) error
beforeRollback func(tx *Tx)
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org