Fix active xa rollback failure and added error message judgment (apache#875)

* Add exception judgment apache#708

* Add exception judgment apache#708

* update const

* update test

* Update pkg/datasource/sql/conn_xa.go

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* Update pkg/datasource/sql/conn_xa.go

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

---------

Co-authored-by: FengZhang <zfcode@qq.com>
Co-authored-by: 吴孝宇 <wuxiaoyu2@xiaohongshu.com>
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Author: 4 people

pkg/datasource/sql/conn_xa.go

@@ -23,8 +23,10 @@ import (
 	"database/sql/driver"
 	"errors"
 	"fmt"
+	"strings"
 	"time"
 
+	"github.com/go-sql-driver/mysql"
 	"seata.apache.org/seata-go/pkg/datasource/sql/types"
 	"seata.apache.org/seata-go/pkg/datasource/sql/xa"
 	"seata.apache.org/seata-go/pkg/tm"
@@ -302,9 +304,19 @@ func (c *XAConn) Rollback(ctx context.Context) error {
 	}
 
 	if !c.rollBacked {
-		if c.xaResource.End(ctx, c.xaBranchXid.String(), xa.TMFail) != nil {
-			return c.rollbackErrorHandle()
+		// First end the XA branch with TMFail
+		if err := c.xaResource.End(ctx, c.xaBranchXid.String(), xa.TMFail); err != nil {
+			// Handle XAER_RMFAIL exception - check if it's already ended
+			//expected error: Error 1399 (XAE07): XAER_RMFAIL: The command cannot be executed when global transaction is in the  IDLE state
+			if isXAER_RMFAILAlreadyEnded(err) {
+				// If already ended, continue with rollback
+				log.Infof("XA branch already ended, continuing with rollback for xid: %s", c.txCtx.XID)
+			} else {
+				return c.rollbackErrorHandle()
+			}
 		}
+
+		// Then perform XA rollback
 		if c.XaRollback(ctx, c.xaBranchXid) != nil {
 			c.cleanXABranchContext()
 			return c.rollbackErrorHandle()
@@ -313,6 +325,7 @@ func (c *XAConn) Rollback(ctx context.Context) error {
 			c.cleanXABranchContext()
 			return fmt.Errorf("failed to report XA branch commit-failure on xid:%s err:%w", c.txCtx.XID, err)
 		}
+		c.rollBacked = true
 	}
 	c.cleanXABranchContext()
 	return nil
@@ -404,3 +417,19 @@ func (c *XAConn) XaRollback(ctx context.Context, xaXid XAXid) error {
 	c.releaseIfNecessary()
 	return err
 }
+
+// isXAER_RMFAILAlreadyEnded checks if the XAER_RMFAIL error indicates the XA branch is already ended
+// expected error: Error 1399 (XAE07): XAER_RMFAIL: The command cannot be executed when global transaction is in the IDLE state
+func isXAER_RMFAILAlreadyEnded(err error) bool {
+	if err == nil {
+		return false
+	}
+	if mysqlErr, ok := err.(*mysql.MySQLError); ok {
+		if mysqlErr.Number == types.ErrCodeXAER_RMFAIL_IDLE {
+			return strings.Contains(mysqlErr.Message, "IDLE state") || strings.Contains(mysqlErr.Message, "already ended")
+		}
+	}
+	// TODO: handle other DB errors
+
+	return false
+}

pkg/datasource/sql/conn_xa_test.go

@@ -22,11 +22,13 @@ import (
 	"database/sql"
 	"database/sql/driver"
 	"io"
+	"strings"
 	"sync/atomic"
 	"testing"
 	"time"
 
 	"github.com/bluele/gcache"
+	"github.com/go-sql-driver/mysql"
 	"github.com/golang/mock/gomock"
 	"github.com/google/uuid"
 	"github.com/stretchr/testify/assert"
@@ -118,10 +120,23 @@ func (mi *mockTxHook) BeforeRollback(tx *Tx) {
 	}
 }
 
+// simulateExecContextError allows tests to inject driver errors for certain SQL strings.
+// When set, baseMockConn will call this hook for each ExecContext.
+var simulateExecContextError func(query string) error
+
 func baseMockConn(mockConn *mock.MockTestDriverConn) {
 	branchStatusCache = gcache.New(1024).LRU().Expiration(time.Minute * 10).Build()
 
-	mockConn.EXPECT().ExecContext(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().Return(&driver.ResultNoRows, nil)
+	mockConn.EXPECT().ExecContext(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().DoAndReturn(
+		func(ctx context.Context, query string, args []driver.NamedValue) (driver.Result, error) {
+			if simulateExecContextError != nil {
+				if err := simulateExecContextError(query); err != nil {
+					return &driver.ResultNoRows, err
+				}
+			}
+			return &driver.ResultNoRows, nil
+		},
+	)
 	mockConn.EXPECT().Exec(gomock.Any(), gomock.Any()).AnyTimes().Return(&driver.ResultNoRows, nil)
 	mockConn.EXPECT().ResetSession(gomock.Any()).AnyTimes().Return(nil)
 	mockConn.EXPECT().Close().AnyTimes().Return(nil)
@@ -329,3 +344,96 @@ func TestXAConn_BeginTx(t *testing.T) {
 	})
 
 }
+
+func TestXAConn_Rollback_XAER_RMFAIL(t *testing.T) {
+	tests := []struct {
+		name string
+		err  error
+		want bool
+	}{
+		{
+			name: "no error case",
+			err:  nil,
+			want: false,
+		},
+		{
+			name: "matching XAER_RMFAIL error with IDLE state",
+			err: &mysql.MySQLError{
+				Number:  1399,
+				Message: "Error 1399 (XAE07): XAER_RMFAIL: The command cannot be executed when global transaction is in the IDLE state",
+			},
+			want: true,
+		},
+		{
+			name: "matching XAER_RMFAIL error with already ended",
+			err: &mysql.MySQLError{
+				Number:  1399,
+				Message: "Error 1399 (XAE07): XAER_RMFAIL: The command cannot be executed when global transaction has already ended",
+			},
+			want: true,
+		},
+		{
+			name: "matching error code but mismatched message",
+			err: &mysql.MySQLError{
+				Number:  1399,
+				Message: "Error 1399 (XAE07): XAER_RMFAIL: Other error message",
+			},
+			want: false,
+		},
+		{
+			name: "mismatched error code but matching message",
+			err: &mysql.MySQLError{
+				Number:  1234,
+				Message: "The command cannot be executed when global transaction is in the IDLE state",
+			},
+			want: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := isXAER_RMFAILAlreadyEnded(tt.err); got != tt.want {
+				t.Errorf("isXAER_RMFAILAlreadyEnded() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+// Covers the XA rollback flow when End() returns XAER_RMFAIL (IDLE/already ended)
+func TestXAConn_Rollback_HandleXAERRMFAILAlreadyEnded(t *testing.T) {
+	ctrl, db, _, ti := initXAConnTestResource(t)
+	defer func() {
+		simulateExecContextError = nil
+		db.Close()
+		ctrl.Finish()
+		CleanTxHooks()
+	}()
+
+	ctx := tm.InitSeataContext(context.Background())
+	tm.SetXID(ctx, uuid.New().String())
+
+	// Ensure Tx.Rollback has a non-nil underlying target to avoid nil-deref when test triggers rollback
+	ti.beforeRollback = func(tx *Tx) {
+		mtx := mock.NewMockTestDriverTx(ctrl)
+		mtx.EXPECT().Rollback().AnyTimes().Return(nil)
+		tx.target = mtx
+	}
+
+	// Inject: XA END returns XAER_RMFAIL(IDLE), normal SQL returns an error to trigger rollback
+	simulateExecContextError = func(query string) error {
+		upper := strings.ToUpper(query)
+		if strings.HasPrefix(upper, "XA END") {
+			return &mysql.MySQLError{Number: types.ErrCodeXAER_RMFAIL_IDLE, Message: "Error 1399 (XAE07): XAER_RMFAIL: The command cannot be executed when global transaction is in the IDLE state"}
+		}
+		if !strings.HasPrefix(upper, "XA ") {
+			return io.EOF
+		}
+		return nil
+	}
+
+	// Execute to enter XA flow; the user SQL fails, but rollback should proceed without panicking
+	_, err := db.ExecContext(ctx, "SELECT 1")
+	if err == nil {
+		t.Fatalf("expected error to trigger rollback path")
+	}
+}

pkg/datasource/sql/types/const.go

@@ -354,3 +354,14 @@ func MySQLStrToJavaType(mysqlType string) JDBCType {
 		return JDBCTypeOther
 	}
 }
+
+// XA transaction related error code constants (based on MySQL/MariaDB specifications)
+const (
+	// ErrCodeXAER_RMFAIL_IDLE 1399: XAER_RMFAIL - The command cannot be executed when global transaction is in the IDLE state
+	// Typically occurs when trying to perform operations on an XA transaction that's in idle state
+	ErrCodeXAER_RMFAIL_IDLE = 1399
+
+	// ErrCodeXAER_INVAL 1400: XAER_INVAL - Invalid XA transaction ID format
+	// Triggered by malformed XID (e.g., invalid gtrid/branchid format or excessive length)
+	ErrCodeXAER_INVAL = 1400
+)