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 <[email protected]>
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 <[email protected]>
    Co-authored-by: CocaElbow <[email protected]>
    Co-authored-by: ThunGuo <[email protected]>
---
 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: [email protected]
For additional commands, e-mail: [email protected]

Reply via email to