This is an automated email from the ASF dual-hosted git repository.
thunguo 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 976e7b94 fix: implement XA mode commitOnXA() and LockQuery() (#1061)
976e7b94 is described below
commit 976e7b9494b39ef3bf7cde2a5946080ccb3dbd03
Author: CAICAII <[email protected]>
AuthorDate: Wed Apr 22 19:18:44 2026 +0800
fix: implement XA mode commitOnXA() and LockQuery() (#1061)
* fix: implement XA mode commitOnXA() and LockQuery()
- tx_xa.go: implement commitOnXA() to execute XA END + XA PREPARE and
report status to TC
- xa_resource_manager.go: LockQuery() delegates to rmRemoting for global
lock checking
- tx.go: add xaConn field to support XA transactions
- conn_xa.go: set xaConn reference in BeginTx
* fix: comprehensive XA mode improvements and fixes
This commit addresses PR #1061 feedback and Oracle architectural review
findings:
**Reviewer Feedback (thunguo):**
- Extract anonymous xaConn interface into named XAConnection type for
better readability
**Code Quality Improvements:**
- Add comprehensive logging to commitOnXA() following existing patterns
(log.Infof/Errorf)
- Add detailed documentation comments explaining XA two-phase commit
protocol flow
- Improve test coverage with 11 new unit tests for XATx functionality
**Critical Bug Fixes:**
1. Fix BranchReport error handling - errors are now properly checked and
logged
2. Complete XA Rollback lifecycle - now executes XA END(TMFAIL) + XA
ROLLBACK + TC reporting
3. Fix SQL error reporting - createNewTxOnExecIfNeed() now reports
PhaseoneFailed to TC
4. Fix nil pointer safety - set tx.target after real transaction is created
in XAConn.BeginTx()
5. Update prepareTime after successful XA PREPARE for correct connection
hold/timeout behavior
**Files Changed:**
- pkg/datasource/sql/tx.go: Extract XAConnection interface, add xaConn field
- pkg/datasource/sql/tx_xa.go: Implement commitOnXA() with logging and
error handling, complete Rollback() lifecycle
- pkg/datasource/sql/conn_xa.go: Fix nil pointer issue, add SQL error
reporting, update prepareTime
- pkg/datasource/sql/tx_xa_test.go: Add comprehensive unit tests (11 tests
covering all scenarios)
**Test Results:**
- All existing tests pass
- 11 new tests added and passing
- Build successful with no errors
Co-authored-by: Oracle Review <architectural-review>
* fix: tighten xa branch reporting lifecycle
* fix: align xa branch tx lifecycle
* test: tighten xa review follow-ups
---------
Co-authored-by: ThunGuo <[email protected]>
---
pkg/datasource/sql/conn_xa.go | 70 +++-
pkg/datasource/sql/conn_xa_test.go | 115 +++++-
.../sql/test_helpers_test.go} | 41 +-
pkg/datasource/sql/tx.go | 21 +-
pkg/datasource/sql/tx_xa.go | 80 +++-
pkg/datasource/sql/tx_xa_test.go | 417 +++++++++++++++++++++
pkg/datasource/sql/xa_resource_manager.go | 2 +-
pkg/datasource/sql/xa_resource_manager_test.go | 95 +++++
pkg/rm/rm_cache.go | 4 +
pkg/rm/rm_cache_test.go | 16 +
10 files changed, 807 insertions(+), 54 deletions(-)
diff --git a/pkg/datasource/sql/conn_xa.go b/pkg/datasource/sql/conn_xa.go
index 5c7d1ba5..857e885b 100644
--- a/pkg/datasource/sql/conn_xa.go
+++ b/pkg/datasource/sql/conn_xa.go
@@ -33,6 +33,8 @@ import (
var xaConnTimeout time.Duration
+var errXABranchLifecycleManaged = errors.New("xa branch lifecycle is managed
by XATx or XAConn")
+
// XAConn Database connection proxy object under XA transaction model
// Conn is assumed to be stateful.
type XAConn struct {
@@ -49,6 +51,19 @@ type XAConn struct {
isConnKept bool
}
+// xaBranchTx is a sentinel driver.Tx used to satisfy database/sql wiring while
+// the real XA branch lifecycle is driven by XATx/XAConn through XA
START/END/PREPARE.
+// Any direct Commit/Rollback on this placeholder indicates the caller
bypassed the XA flow.
+type xaBranchTx struct{}
+
+func (xaBranchTx) Commit() error {
+ return errXABranchLifecycleManaged
+}
+
+func (xaBranchTx) Rollback() error {
+ return errXABranchLifecycleManaged
+}
+
func (c *XAConn) PrepareContext(ctx context.Context, query string)
(driver.Stmt, error) {
if c.createOnceTxContext(ctx) {
defer func() {
@@ -126,11 +141,20 @@ func (c *XAConn) BeginTx(ctx context.Context, opts
driver.TxOptions) (driver.Tx,
c.txCtx.XID = tm.GetXID(ctx)
c.txCtx.TransactionMode = types.XAMode
- tx, err := c.Conn.BeginTx(ctx, opts)
+ // Keep a sentinel target in Tx so any accidental fallback to the
generic
+ // driver.Tx path fails fast instead of silently masking XA lifecycle
bugs.
+ branchTx := xaBranchTx{}
+ c.tx = branchTx
+
+ tx, err := newTx(
+ withDriverConn(c.Conn),
+ withTxCtx(c.txCtx),
+ withOriginTx(branchTx),
+ withXAConn(c),
+ )
if err != nil {
return nil, err
}
- c.tx = tx
if !c.autoCommit {
if c.xaActive {
@@ -142,6 +166,8 @@ func (c *XAConn) BeginTx(ctx context.Context, opts
driver.TxOptions) (driver.Tx,
return nil, fmt.Errorf("start xa %s transaction failure
for the tx is a wrong type", c.txCtx.XID)
}
+ baseTx.xaConn = c
+
c.branchRegisterTime = time.Now()
if err := baseTx.register(c.txCtx); err != nil {
c.cleanXABranchContext()
@@ -184,12 +210,17 @@ func (c *XAConn) createNewTxOnExecIfNeed(ctx
context.Context, f func() (types.Ex
defer func() {
recoverErr := recover()
- if err != nil || recoverErr != nil {
- log.Errorf("conn at rollback error:%v or
recoverErr:%v", err, recoverErr)
+ if recoverErr != nil {
+ log.Errorf("conn xa rollback recoverErr:%v", recoverErr)
+ if tx != nil {
+ if rollbackErr := tx.Rollback(); rollbackErr !=
nil {
+ log.Errorf("conn xa rollback error:%v",
rollbackErr)
+ }
+ return
+ }
if c.tx != nil {
- rollbackErr := c.tx.Rollback()
- if rollbackErr != nil {
- log.Errorf("conn at rollback error:%v",
rollbackErr)
+ if rollbackErr := c.Rollback(ctx); rollbackErr
!= nil {
+ log.Errorf("conn xa rollback error:%v",
rollbackErr)
}
}
}
@@ -206,20 +237,23 @@ func (c *XAConn) createNewTxOnExecIfNeed(ctx
context.Context, f func() (types.Ex
// execute SQL
ret, err := f()
if err != nil {
- // XA End & Rollback
- if rollbackErr := c.Rollback(ctx); rollbackErr != nil {
- log.Errorf("failed to rollback xa branch of :%s,
err:%v", c.txCtx.XID, rollbackErr)
+ if tx != nil {
+ if rollbackErr := tx.Rollback(); rollbackErr != nil {
+ log.Errorf("failed to rollback xa branch of
:%s, err:%v", c.txCtx.XID, rollbackErr)
+ }
+ } else {
+ if rollbackErr := c.Rollback(ctx); rollbackErr != nil {
+ log.Errorf("failed to rollback xa branch of
:%s, err:%v", c.txCtx.XID, rollbackErr)
+ }
}
return nil, err
}
if tx != nil && currentAutoCommit {
- if err = c.Commit(ctx); err != nil {
+ // Commit through XATx so phase-one reporting stays coupled to
driver.Tx lifecycle.
+ if err = tx.Commit(); err != nil {
log.Errorf("xa connection proxy commit failure xid:%s,
err:%v", c.txCtx.XID, err)
- // XA End & Rollback
- if err := c.Rollback(ctx); err != nil {
- log.Errorf("xa connection proxy rollback
failure xid:%s, err:%v", c.txCtx.XID, err)
- }
+ return nil, err
}
}
@@ -321,10 +355,6 @@ func (c *XAConn) Rollback(ctx context.Context) error {
c.cleanXABranchContext()
return c.rollbackErrorHandle()
}
- if err := c.tx.Rollback(); err != nil {
- 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()
@@ -356,6 +386,8 @@ func (c *XAConn) Commit(ctx context.Context) error {
if c.xaResource.XAPrepare(ctx, c.xaBranchXid.String()) != nil {
return c.commitErrorHandle(ctx)
}
+
+ c.prepareTime = time.Now()
return nil
}
diff --git a/pkg/datasource/sql/conn_xa_test.go
b/pkg/datasource/sql/conn_xa_test.go
index 6d5e26fc..46fc412c 100644
--- a/pkg/datasource/sql/conn_xa_test.go
+++ b/pkg/datasource/sql/conn_xa_test.go
@@ -38,6 +38,7 @@ import (
"seata.apache.org/seata-go/v2/pkg/datasource/sql/types"
"seata.apache.org/seata-go/v2/pkg/datasource/sql/xa"
"seata.apache.org/seata-go/v2/pkg/protocol/branch"
+ "seata.apache.org/seata-go/v2/pkg/rm"
"seata.apache.org/seata-go/v2/pkg/tm"
)
@@ -170,6 +171,31 @@ func initXAConnTestResource(t *testing.T)
(*gomock.Controller, *sql.DB, *mockSQL
return ctrl, db, mi, ti
}
+func newMockXAConn(t *testing.T, ctrl *gomock.Controller, branchID int64)
(*XAConn, *mock.MockDataSourceManager) {
+ t.Helper()
+
+ mockMgr := mock.NewMockDataSourceManager(ctrl)
+ mockMgr.SetBranchType(branch.BranchTypeXA)
+ registerResourceManagerForTest(t, mockMgr)
+ mockMgr.EXPECT().BranchRegister(gomock.Any(),
gomock.Any()).AnyTimes().Return(branchID, nil)
+
+ mockConn := mock.NewMockTestDriverConn(ctrl)
+ baseMockConn(mockConn)
+
+ return &XAConn{
+ Conn: &Conn{
+ res: &DBResource{
+ resourceID: "jdbc:mysql://test/resource",
+ dbType: types.DBTypeMySQL,
+ },
+ txCtx: types.NewTxCtx(),
+ targetConn: mockConn,
+ autoCommit: true,
+ dbType: types.DBTypeMySQL,
+ },
+ }, mockMgr
+}
+
func TestXAConn_ExecContext(t *testing.T) {
ctrl, db, mi, ti := initXAConnTestResource(t)
@@ -206,8 +232,7 @@ func TestXAConn_ExecContext(t *testing.T) {
_, err = db.ExecContext(ctx, "SELECT 1")
assert.NoError(t, err)
- // todo fix
- assert.Equal(t, int32(0), atomic.LoadInt32(&comitCnt))
+ assert.Equal(t, int32(2), atomic.LoadInt32(&comitCnt))
})
t.Run("not xid", func(t *testing.T) {
@@ -389,7 +414,7 @@ func TestXAConn_Rollback_XAER_RMFAIL(t *testing.T) {
// 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)
+ ctrl, db, _, _ := initXAConnTestResource(t)
defer func() {
simulateExecContextError = nil
db.Close()
@@ -400,18 +425,14 @@ func TestXAConn_Rollback_HandleXAERRMFAILAlreadyEnded(t
*testing.T) {
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"}
+ 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
@@ -425,3 +446,75 @@ func TestXAConn_Rollback_HandleXAERRMFAILAlreadyEnded(t
*testing.T) {
t.Fatalf("expected error to trigger rollback path")
}
}
+
+func TestXAConn_ExecContext_AutoCommitReportsPhaseOneDone(t *testing.T) {
+ ctrl := gomock.NewController(t)
+ defer ctrl.Finish()
+ CleanTxHooks()
+ defer CleanTxHooks()
+
+ xaConn, mockMgr := newMockXAConn(t, ctrl, 123)
+ mockMgr.EXPECT().BranchReport(gomock.Any(), gomock.Any()).DoAndReturn(
+ func(ctx context.Context, param rm.BranchReportParam) error {
+ assert.Equal(t, branch.BranchTypeXA, param.BranchType)
+ assert.Equal(t, int64(123), param.BranchId)
+ assert.EqualValues(t, branch.BranchStatusPhaseoneDone,
param.Status)
+ return nil
+ },
+ ).Times(1)
+
+ var commitCnt int32
+ RegisterTxHook(&mockTxHook{
+ beforeCommit: func(tx *Tx) error {
+ atomic.AddInt32(&commitCnt, 1)
+ return nil
+ },
+ })
+
+ ctx := tm.InitSeataContext(context.Background())
+ tm.SetXID(ctx, uuid.NewString())
+
+ _, err := xaConn.ExecContext(ctx, "SELECT 1", nil)
+ assert.NoError(t, err)
+ assert.Equal(t, int32(1), atomic.LoadInt32(&commitCnt))
+}
+
+func TestXAConn_BeginTx_DoesNotStartPhysicalTx(t *testing.T) {
+ ctrl := gomock.NewController(t)
+ defer ctrl.Finish()
+
+ xaConn, mockMgr := newMockXAConn(t, ctrl, 123)
+ mockMgr.EXPECT().BranchReport(gomock.Any(), gomock.Any()).DoAndReturn(
+ func(ctx context.Context, param rm.BranchReportParam) error {
+ assert.Equal(t, branch.BranchTypeXA, param.BranchType)
+ assert.Equal(t, int64(123), param.BranchId)
+ assert.EqualValues(t,
branch.BranchStatusPhaseoneFailed, param.Status)
+ return nil
+ },
+ ).Times(1)
+
+ ctx := tm.InitSeataContext(context.Background())
+ tm.SetXID(ctx, uuid.NewString())
+
+ tx, err := xaConn.BeginTx(ctx, driver.TxOptions{})
+ assert.NoError(t, err)
+
+ xaTx, ok := tx.(*XATx)
+ if assert.True(t, ok) {
+ _, noop := xaTx.tx.target.(xaBranchTx)
+ assert.True(t, noop)
+ }
+
+ err = tx.Rollback()
+ assert.NoError(t, err)
+}
+
+func TestXABranchTx_CommitRollbackFailFast(t *testing.T) {
+ branchTx := xaBranchTx{}
+
+ err := branchTx.Commit()
+ assert.ErrorIs(t, err, errXABranchLifecycleManaged)
+
+ err = branchTx.Rollback()
+ assert.ErrorIs(t, err, errXABranchLifecycleManaged)
+}
diff --git a/pkg/rm/rm_cache_test.go b/pkg/datasource/sql/test_helpers_test.go
similarity index 54%
copy from pkg/rm/rm_cache_test.go
copy to pkg/datasource/sql/test_helpers_test.go
index e9477b80..6e9af72f 100644
--- a/pkg/rm/rm_cache_test.go
+++ b/pkg/datasource/sql/test_helpers_test.go
@@ -15,31 +15,38 @@
* limitations under the License.
*/
-package rm
+package sql
import (
"testing"
- "github.com/golang/mock/gomock"
- "github.com/stretchr/testify/assert"
-
"seata.apache.org/seata-go/v2/pkg/protocol/branch"
+ "seata.apache.org/seata-go/v2/pkg/rm"
)
-func TestGetRmCacheInstance(t *testing.T) {
- ctl := gomock.NewController(t)
-
- mockResourceManager := NewMockResourceManager(ctl)
-
mockResourceManager.EXPECT().GetBranchType().Return(branch.BranchTypeTCC)
+func registerResourceManagerForTest(t *testing.T, resourceManager
rm.ResourceManager) {
+ t.Helper()
- tests := struct {
- name string
- want *ResourceManagerCache
- }{"test1", &ResourceManagerCache{}}
+ prevMgr, hasPrevMgr :=
currentResourceManagerForTest(resourceManager.GetBranchType())
+ rm.GetRmCacheInstance().RegisterResourceManager(resourceManager)
- t.Run(tests.name, func(t *testing.T) {
-
GetRmCacheInstance().RegisterResourceManager(mockResourceManager)
- actual :=
GetRmCacheInstance().GetResourceManager(branch.BranchTypeTCC)
- assert.Equalf(t, mockResourceManager, actual,
"GetRmCacheInstance()")
+ t.Cleanup(func() {
+ if hasPrevMgr {
+ rm.GetRmCacheInstance().RegisterResourceManager(prevMgr)
+ return
+ }
+
rm.GetRmCacheInstance().UnregisterResourceManager(resourceManager.GetBranchType())
})
}
+
+func currentResourceManagerForTest(branchType branch.BranchType) (manager
rm.ResourceManager, ok bool) {
+ defer func() {
+ if recover() != nil {
+ manager = nil
+ ok = false
+ }
+ }()
+
+ manager = rm.GetRmCacheInstance().GetResourceManager(branchType)
+ return manager, true
+}
diff --git a/pkg/datasource/sql/tx.go b/pkg/datasource/sql/tx.go
index fdf4728f..c3c74c26 100644
--- a/pkg/datasource/sql/tx.go
+++ b/pkg/datasource/sql/tx.go
@@ -59,6 +59,12 @@ type (
BeforeRollback(tx *Tx)
}
+
+ // XAConnection represents an XA-capable connection that can commit or
rollback XA transactions
+ XAConnection interface {
+ Commit(ctx context.Context) error
+ Rollback(ctx context.Context) error
+ }
)
func newTx(opts ...txOption) (driver.Tx, error) {
@@ -96,11 +102,19 @@ func withTxCtx(ctx *types.TransactionContext) txOption {
}
}
+// withXAConn
+func withXAConn(xaConn XAConnection) txOption {
+ return func(t *Tx) {
+ t.xaConn = xaConn
+ }
+}
+
// Tx
type Tx struct {
conn *Conn
tranCtx *types.TransactionContext
target driver.Tx
+ xaConn XAConnection
}
// Commit do commit action
@@ -194,9 +208,10 @@ func (tx *Tx) report(success bool) error {
}
status := getStatus(success)
request := rm.BranchReportParam{
- Xid: tx.tranCtx.XID,
- BranchId: int64(tx.tranCtx.BranchID),
- Status: status,
+ BranchType: tx.tranCtx.TransactionMode.BranchType(),
+ Xid: tx.tranCtx.XID,
+ BranchId: int64(tx.tranCtx.BranchID),
+ Status: status,
}
dataSourceManager :=
datasource.GetDataSourceManager(tx.tranCtx.TransactionMode.BranchType())
if dataSourceManager == nil {
diff --git a/pkg/datasource/sql/tx_xa.go b/pkg/datasource/sql/tx_xa.go
index d3380e57..a280eea8 100644
--- a/pkg/datasource/sql/tx_xa.go
+++ b/pkg/datasource/sql/tx_xa.go
@@ -17,6 +17,13 @@
package sql
+import (
+ "context"
+ "fmt"
+
+ "seata.apache.org/seata-go/v2/pkg/util/log"
+)
+
type XATx struct {
tx *Tx
}
@@ -32,15 +39,82 @@ func (tx *XATx) Commit() error {
return tx.commitOnXA()
}
+// Rollback executes XA END(TMFAIL), XA ROLLBACK and reports to TC
func (tx *XATx) Rollback() error {
originTx := tx.tx
- if originTx.tranCtx.OpenGlobalTransaction() &&
originTx.tranCtx.IsBranchRegistered() {
- return originTx.report(false)
+
+ if !originTx.tranCtx.OpenGlobalTransaction() {
+ return nil
+ }
+
+ if originTx.xaConn == nil {
+ return fmt.Errorf("xa transaction requires xaConn")
+ }
+
+ xid := originTx.tranCtx.XID
+ branchID := originTx.tranCtx.BranchID
+
+ log.Infof("xa branch [%d/%s] executing XA rollback", branchID, xid)
+
+ if err := originTx.xaConn.Rollback(context.Background()); err != nil {
+ log.Errorf("xa branch [%d/%s] XA END(TMFAIL) + XA ROLLBACK
failed: %v", branchID, xid, err)
+ if originTx.tranCtx.IsBranchRegistered() {
+ if reportErr := originTx.report(false); reportErr !=
nil {
+ log.Errorf("xa branch [%d/%s] failed to report
rollback failure to TC: %v", branchID, xid, reportErr)
+ }
+ }
+ return err
}
+ log.Infof("xa branch [%d/%s] XA END(TMFAIL) + XA ROLLBACK succeeded",
branchID, xid)
+
+ if originTx.tranCtx.IsBranchRegistered() {
+ if err := originTx.report(false); err != nil {
+ log.Errorf("xa branch [%d/%s] failed to report rollback
to TC: %v", branchID, xid, err)
+ return err
+ }
+ log.Infof("xa branch [%d/%s] reported rollback to TC",
branchID, xid)
+ }
+
return nil
}
-// commitOnXA commit xa and register branch transaction
+// commitOnXA executes XA END, XA PREPARE and reports to TC
func (tx *XATx) commitOnXA() error {
+ originTx := tx.tx
+
+ if !originTx.tranCtx.OpenGlobalTransaction() {
+ return nil
+ }
+
+ if originTx.xaConn == nil {
+ return fmt.Errorf("xa transaction requires xaConn")
+ }
+
+ xid := originTx.tranCtx.XID
+ branchID := originTx.tranCtx.BranchID
+
+ log.Infof("xa branch [%d/%s] executing XA END + XA PREPARE", branchID,
xid)
+
+ if err := originTx.xaConn.Commit(context.Background()); err != nil {
+ log.Errorf("xa branch [%d/%s] XA END + XA PREPARE failed: %v",
branchID, xid, err)
+ if originTx.tranCtx.IsBranchRegistered() {
+ if reportErr := originTx.report(false); reportErr !=
nil {
+ log.Errorf("xa branch [%d/%s] failed to report
phase-1 failure to TC: %v", branchID, xid, reportErr)
+ return fmt.Errorf("XA PREPARE failed: %w, and
report failed: %v", err, reportErr)
+ }
+ }
+ return err
+ }
+
+ log.Infof("xa branch [%d/%s] XA END + XA PREPARE succeeded", branchID,
xid)
+
+ if originTx.tranCtx.IsBranchRegistered() {
+ if err := originTx.report(true); err != nil {
+ log.Errorf("xa branch [%d/%s] failed to report phase-1
success to TC: %v", branchID, xid, err)
+ return fmt.Errorf("XA PREPARE succeeded but report to
TC failed: %w", err)
+ }
+ log.Infof("xa branch [%d/%s] reported phase-1 success to TC",
branchID, xid)
+ }
+
return nil
}
diff --git a/pkg/datasource/sql/tx_xa_test.go b/pkg/datasource/sql/tx_xa_test.go
new file mode 100644
index 00000000..24fa4817
--- /dev/null
+++ b/pkg/datasource/sql/tx_xa_test.go
@@ -0,0 +1,417 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package sql
+
+import (
+ "context"
+ "errors"
+ "testing"
+
+ "github.com/golang/mock/gomock"
+ "github.com/stretchr/testify/assert"
+
+ "seata.apache.org/seata-go/v2/pkg/datasource/sql/mock"
+ "seata.apache.org/seata-go/v2/pkg/datasource/sql/types"
+ "seata.apache.org/seata-go/v2/pkg/protocol/branch"
+ "seata.apache.org/seata-go/v2/pkg/rm"
+)
+
+type mockXAConnection struct {
+ commitErr error
+ rollbackErr error
+ commitCalls int
+ rollbackCalls int
+}
+
+func (m *mockXAConnection) Commit(ctx context.Context) error {
+ m.commitCalls++
+ return m.commitErr
+}
+
+func (m *mockXAConnection) Rollback(ctx context.Context) error {
+ m.rollbackCalls++
+ return m.rollbackErr
+}
+
+func TestXATx_commitOnXA_NoGlobalTransaction(t *testing.T) {
+ tranCtx := types.NewTxCtx()
+ tranCtx.XID = ""
+
+ xaTx := &XATx{
+ tx: &Tx{
+ tranCtx: tranCtx,
+ },
+ }
+
+ err := xaTx.commitOnXA()
+ assert.NoError(t, err)
+}
+
+func TestXATx_commitOnXA_MissingXAConn(t *testing.T) {
+ tranCtx := types.NewTxCtx()
+ tranCtx.XID = "test-xid"
+ tranCtx.TransactionMode = types.XAMode
+
+ xaTx := &XATx{
+ tx: &Tx{
+ tranCtx: tranCtx,
+ xaConn: nil,
+ },
+ }
+
+ err := xaTx.commitOnXA()
+ assert.Error(t, err)
+ assert.Contains(t, err.Error(), "xa transaction requires xaConn")
+}
+
+func TestXATx_commitOnXA_CommitSuccess_BranchNotRegistered(t *testing.T) {
+ tranCtx := types.NewTxCtx()
+ tranCtx.XID = "test-xid"
+ tranCtx.BranchID = 0
+ tranCtx.TransactionMode = types.XAMode
+
+ mockConn := &mockXAConnection{}
+
+ xaTx := &XATx{
+ tx: &Tx{
+ tranCtx: tranCtx,
+ xaConn: mockConn,
+ },
+ }
+
+ err := xaTx.commitOnXA()
+ assert.NoError(t, err)
+ assert.Equal(t, 1, mockConn.commitCalls)
+}
+
+func TestXATx_commitOnXA_CommitFailure_BranchNotRegistered(t *testing.T) {
+ tranCtx := types.NewTxCtx()
+ tranCtx.XID = "test-xid"
+ tranCtx.BranchID = 0
+ tranCtx.TransactionMode = types.XAMode
+
+ mockConn := &mockXAConnection{
+ commitErr: errors.New("XA PREPARE failed"),
+ }
+
+ xaTx := &XATx{
+ tx: &Tx{
+ tranCtx: tranCtx,
+ xaConn: mockConn,
+ },
+ }
+
+ err := xaTx.commitOnXA()
+ assert.Error(t, err)
+ assert.Contains(t, err.Error(), "XA PREPARE failed")
+ assert.Equal(t, 1, mockConn.commitCalls)
+}
+
+func TestXATx_commitOnXA_CommitSuccess_BranchRegisteredReportsSuccess(t
*testing.T) {
+ ctrl := gomock.NewController(t)
+ defer ctrl.Finish()
+
+ mockMgr := mock.NewMockDataSourceManager(ctrl)
+ mockMgr.SetBranchType(branch.BranchTypeXA)
+ registerResourceManagerForTest(t, mockMgr)
+ mockMgr.EXPECT().BranchReport(gomock.Any(), gomock.Any()).DoAndReturn(
+ func(ctx context.Context, param rm.BranchReportParam) error {
+ assert.Equal(t, branch.BranchTypeXA, param.BranchType)
+ assert.Equal(t, int64(123), param.BranchId)
+ assert.EqualValues(t, branch.BranchStatusPhaseoneDone,
param.Status)
+ assert.Equal(t, "test-xid", param.Xid)
+ return nil
+ },
+ ).Times(1)
+
+ tranCtx := types.NewTxCtx()
+ tranCtx.XID = "test-xid"
+ tranCtx.BranchID = 123
+ tranCtx.TransactionMode = types.XAMode
+
+ mockConn := &mockXAConnection{}
+
+ xaTx := &XATx{
+ tx: &Tx{
+ tranCtx: tranCtx,
+ xaConn: mockConn,
+ },
+ }
+
+ err := xaTx.commitOnXA()
+ assert.NoError(t, err)
+ assert.Equal(t, 1, mockConn.commitCalls)
+}
+
+func TestXATx_commitOnXA_CommitFailure_BranchRegisteredReportsFailure(t
*testing.T) {
+ ctrl := gomock.NewController(t)
+ defer ctrl.Finish()
+
+ mockMgr := mock.NewMockDataSourceManager(ctrl)
+ mockMgr.SetBranchType(branch.BranchTypeXA)
+ registerResourceManagerForTest(t, mockMgr)
+ mockMgr.EXPECT().BranchReport(gomock.Any(), gomock.Any()).DoAndReturn(
+ func(ctx context.Context, param rm.BranchReportParam) error {
+ assert.Equal(t, branch.BranchTypeXA, param.BranchType)
+ assert.Equal(t, int64(123), param.BranchId)
+ assert.EqualValues(t,
branch.BranchStatusPhaseoneFailed, param.Status)
+ assert.Equal(t, "test-xid", param.Xid)
+ return nil
+ },
+ ).Times(1)
+
+ tranCtx := types.NewTxCtx()
+ tranCtx.XID = "test-xid"
+ tranCtx.BranchID = 123
+ tranCtx.TransactionMode = types.XAMode
+
+ mockConn := &mockXAConnection{
+ commitErr: errors.New("XA PREPARE failed"),
+ }
+
+ xaTx := &XATx{
+ tx: &Tx{
+ tranCtx: tranCtx,
+ xaConn: mockConn,
+ },
+ }
+
+ err := xaTx.commitOnXA()
+ assert.Error(t, err)
+ assert.Contains(t, err.Error(), "XA PREPARE failed")
+ assert.Equal(t, 1, mockConn.commitCalls)
+}
+
+func TestXATx_Rollback_NoGlobalTransaction(t *testing.T) {
+ tranCtx := types.NewTxCtx()
+ tranCtx.XID = ""
+
+ xaTx := &XATx{
+ tx: &Tx{
+ tranCtx: tranCtx,
+ },
+ }
+
+ err := xaTx.Rollback()
+ assert.NoError(t, err)
+}
+
+func TestXATx_Rollback_MissingXAConn(t *testing.T) {
+ ctrl := gomock.NewController(t)
+ defer ctrl.Finish()
+
+ mockMgr := mock.NewMockDataSourceManager(ctrl)
+ mockMgr.SetBranchType(branch.BranchTypeXA)
+ registerResourceManagerForTest(t, mockMgr)
+ mockMgr.EXPECT().BranchReport(gomock.Any(), gomock.Any()).Times(0)
+
+ tranCtx := types.NewTxCtx()
+ tranCtx.XID = "test-xid"
+ tranCtx.BranchID = 123
+ tranCtx.TransactionMode = types.XAMode
+
+ xaTx := &XATx{
+ tx: &Tx{
+ tranCtx: tranCtx,
+ },
+ }
+
+ err := xaTx.Rollback()
+ assert.Error(t, err)
+ assert.Contains(t, err.Error(), "xa transaction requires xaConn")
+}
+
+func TestXATx_Rollback_BranchNotRegistered(t *testing.T) {
+ tranCtx := types.NewTxCtx()
+ tranCtx.XID = "test-xid"
+ tranCtx.BranchID = 0
+ tranCtx.TransactionMode = types.XAMode
+
+ mockConn := &mockXAConnection{}
+
+ xaTx := &XATx{
+ tx: &Tx{
+ tranCtx: tranCtx,
+ xaConn: mockConn,
+ },
+ }
+
+ err := xaTx.Rollback()
+ assert.NoError(t, err)
+ assert.Equal(t, 1, mockConn.rollbackCalls)
+}
+
+func TestXATx_Rollback_WithXAConn(t *testing.T) {
+ tranCtx := types.NewTxCtx()
+ tranCtx.XID = "test-xid"
+ tranCtx.BranchID = 0
+ tranCtx.TransactionMode = types.XAMode
+
+ mockConn := &mockXAConnection{}
+
+ xaTx := &XATx{
+ tx: &Tx{
+ tranCtx: tranCtx,
+ xaConn: mockConn,
+ },
+ }
+
+ err := xaTx.Rollback()
+ assert.NoError(t, err)
+ assert.Equal(t, 1, mockConn.rollbackCalls)
+}
+
+func TestXATx_Rollback_XAConnError(t *testing.T) {
+ tranCtx := types.NewTxCtx()
+ tranCtx.XID = "test-xid"
+ tranCtx.BranchID = 0
+ tranCtx.TransactionMode = types.XAMode
+
+ mockConn := &mockXAConnection{
+ rollbackErr: errors.New("XA ROLLBACK failed"),
+ }
+
+ xaTx := &XATx{
+ tx: &Tx{
+ tranCtx: tranCtx,
+ xaConn: mockConn,
+ },
+ }
+
+ err := xaTx.Rollback()
+ assert.Error(t, err)
+ assert.Contains(t, err.Error(), "XA ROLLBACK failed")
+ assert.Equal(t, 1, mockConn.rollbackCalls)
+}
+
+func TestXATx_Rollback_BranchRegisteredReportsFailure(t *testing.T) {
+ ctrl := gomock.NewController(t)
+ defer ctrl.Finish()
+
+ mockMgr := mock.NewMockDataSourceManager(ctrl)
+ mockMgr.SetBranchType(branch.BranchTypeXA)
+ registerResourceManagerForTest(t, mockMgr)
+ mockMgr.EXPECT().BranchReport(gomock.Any(), gomock.Any()).DoAndReturn(
+ func(ctx context.Context, param rm.BranchReportParam) error {
+ assert.Equal(t, branch.BranchTypeXA, param.BranchType)
+ assert.Equal(t, int64(123), param.BranchId)
+ assert.EqualValues(t,
branch.BranchStatusPhaseoneFailed, param.Status)
+ assert.Equal(t, "test-xid", param.Xid)
+ return nil
+ },
+ ).Times(1)
+
+ tranCtx := types.NewTxCtx()
+ tranCtx.XID = "test-xid"
+ tranCtx.BranchID = 123
+ tranCtx.TransactionMode = types.XAMode
+
+ mockConn := &mockXAConnection{}
+
+ xaTx := &XATx{
+ tx: &Tx{
+ tranCtx: tranCtx,
+ xaConn: mockConn,
+ },
+ }
+
+ err := xaTx.Rollback()
+ assert.NoError(t, err)
+ assert.Equal(t, 1, mockConn.rollbackCalls)
+}
+
+func TestXATx_Commit_CallsCommitOnXA(t *testing.T) {
+ tranCtx := types.NewTxCtx()
+ tranCtx.XID = "test-xid"
+ tranCtx.BranchID = 0
+ tranCtx.TransactionMode = types.XAMode
+
+ mockConn := &mockXAConnection{}
+
+ xaTx := &XATx{
+ tx: &Tx{
+ tranCtx: tranCtx,
+ xaConn: mockConn,
+ },
+ }
+
+ err := xaTx.Commit()
+ assert.NoError(t, err)
+ assert.Equal(t, 1, mockConn.commitCalls)
+}
+
+func TestTx_report_NoBranchID(t *testing.T) {
+ tranCtx := types.NewTxCtx()
+ tranCtx.BranchID = 0
+
+ tx := &Tx{
+ tranCtx: tranCtx,
+ }
+
+ err := tx.report(true)
+ assert.NoError(t, err)
+}
+
+func TestXAConnection_Interface(t *testing.T) {
+ var _ XAConnection = (*mockXAConnection)(nil)
+}
+
+func TestWithXAConn(t *testing.T) {
+ mockConn := &mockXAConnection{}
+ tx := &Tx{}
+
+ opt := withXAConn(mockConn)
+ opt(tx)
+
+ assert.Equal(t, mockConn, tx.xaConn)
+}
+
+func TestXATx_Commit_BeforeCommitError(t *testing.T) {
+ tranCtx := types.NewTxCtx()
+ tranCtx.XID = "test-xid"
+ tranCtx.TransactionMode = types.XAMode
+
+ hookCalled := false
+ RegisterTxHook(&mockTxHook{
+ beforeCommit: func(tx *Tx) error {
+ hookCalled = true
+ return errors.New("hook error")
+ },
+ })
+ defer CleanTxHooks()
+
+ mockConn := &mockXAConnection{}
+
+ xaTx := &XATx{
+ tx: &Tx{
+ tranCtx: tranCtx,
+ xaConn: mockConn,
+ },
+ }
+
+ err := xaTx.Commit()
+ assert.Error(t, err)
+ assert.Contains(t, err.Error(), "hook error")
+ assert.True(t, hookCalled)
+ assert.Equal(t, 0, mockConn.commitCalls)
+}
+
+func TestGetStatus(t *testing.T) {
+ assert.EqualValues(t, branch.BranchStatusPhaseoneDone, getStatus(true))
+ assert.EqualValues(t, branch.BranchStatusPhaseoneFailed,
getStatus(false))
+}
diff --git a/pkg/datasource/sql/xa_resource_manager.go
b/pkg/datasource/sql/xa_resource_manager.go
index 0f793569..ab2446d0 100644
--- a/pkg/datasource/sql/xa_resource_manager.go
+++ b/pkg/datasource/sql/xa_resource_manager.go
@@ -197,7 +197,7 @@ func (xaManager *XAResourceManager) BranchRollback(ctx
context.Context, branchRe
}
func (xaManager *XAResourceManager) LockQuery(ctx context.Context, param
rm.LockQueryParam) (bool, error) {
- return false, nil
+ return xaManager.rmRemoting.LockQuery(param)
}
func (xaManager *XAResourceManager) BranchRegister(ctx context.Context, req
rm.BranchRegisterParam) (int64, error) {
diff --git a/pkg/datasource/sql/xa_resource_manager_test.go
b/pkg/datasource/sql/xa_resource_manager_test.go
new file mode 100644
index 00000000..e4edd0d5
--- /dev/null
+++ b/pkg/datasource/sql/xa_resource_manager_test.go
@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package sql
+
+import (
+ "context"
+ "errors"
+ "reflect"
+ "testing"
+
+ "github.com/agiledragon/gomonkey/v2"
+ "github.com/stretchr/testify/assert"
+
+ "seata.apache.org/seata-go/v2/pkg/protocol/branch"
+ "seata.apache.org/seata-go/v2/pkg/protocol/message"
+ "seata.apache.org/seata-go/v2/pkg/remoting/getty"
+ "seata.apache.org/seata-go/v2/pkg/rm"
+)
+
+func TestXAResourceManager_LockQuery(t *testing.T) {
+ tests := []struct {
+ name string
+ resp interface{}
+ respErr error
+ want bool
+ wantErr string
+ }{
+ {
+ name: "lockable",
+ resp: message.GlobalLockQueryResponse{Lockable: true},
+ want: true,
+ },
+ {
+ name: "unlockable",
+ resp: message.GlobalLockQueryResponse{Lockable: false},
+ want: false,
+ },
+ {
+ name: "remoting error",
+ respErr: errors.New("network timeout"),
+ want: false,
+ wantErr: "network timeout",
+ },
+ }
+
+ param := rm.LockQueryParam{
+ BranchType: branch.BranchTypeXA,
+ ResourceId: "jdbc:mysql://test/resource",
+ Xid: "test-xid",
+ LockKeys: "user:1",
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ patches :=
gomonkey.ApplyMethod(reflect.TypeOf(getty.GetGettyRemotingClient()),
"SendSyncRequest",
+ func(_ *getty.GettyRemotingClient, msg
interface{}) (interface{}, error) {
+ req, ok :=
msg.(message.GlobalLockQueryRequest)
+ if assert.True(t, ok) {
+ assert.Equal(t,
param.BranchType, req.BranchType)
+ assert.Equal(t,
param.ResourceId, req.ResourceId)
+ assert.Equal(t, param.Xid,
req.Xid)
+ assert.Equal(t, param.LockKeys,
req.LockKey)
+ }
+ return tt.resp, tt.respErr
+ })
+ defer patches.Reset()
+
+ xaManager := &XAResourceManager{rmRemoting:
rm.GetRMRemotingInstance()}
+
+ got, err := xaManager.LockQuery(context.Background(),
param)
+
+ assert.Equal(t, tt.want, got)
+ if tt.wantErr == "" {
+ assert.NoError(t, err)
+ } else {
+ assert.EqualError(t, err, tt.wantErr)
+ }
+ })
+ }
+}
diff --git a/pkg/rm/rm_cache.go b/pkg/rm/rm_cache.go
index 855facb3..00df6e94 100644
--- a/pkg/rm/rm_cache.go
+++ b/pkg/rm/rm_cache.go
@@ -48,6 +48,10 @@ func (d *ResourceManagerCache)
RegisterResourceManager(resourceManager ResourceM
d.resourceManagerMap.Store(resourceManager.GetBranchType(),
resourceManager)
}
+func (d *ResourceManagerCache) UnregisterResourceManager(branchType
branch.BranchType) {
+ d.resourceManagerMap.Delete(branchType)
+}
+
func (d *ResourceManagerCache) GetResourceManager(branchType
branch.BranchType) ResourceManager {
rm, ok := d.resourceManagerMap.Load(branchType)
if !ok {
diff --git a/pkg/rm/rm_cache_test.go b/pkg/rm/rm_cache_test.go
index e9477b80..80a2ba2b 100644
--- a/pkg/rm/rm_cache_test.go
+++ b/pkg/rm/rm_cache_test.go
@@ -43,3 +43,19 @@ func TestGetRmCacheInstance(t *testing.T) {
assert.Equalf(t, mockResourceManager, actual,
"GetRmCacheInstance()")
})
}
+
+func TestResourceManagerCache_UnregisterResourceManager(t *testing.T) {
+ ctl := gomock.NewController(t)
+
+ mockResourceManager := NewMockResourceManager(ctl)
+
mockResourceManager.EXPECT().GetBranchType().Return(branch.BranchTypeSAGA).AnyTimes()
+
+ cache := GetRmCacheInstance()
+ cache.RegisterResourceManager(mockResourceManager)
+
+ cache.UnregisterResourceManager(branch.BranchTypeSAGA)
+
+ assert.Panics(t, func() {
+ cache.GetResourceManager(branch.BranchTypeSAGA)
+ })
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
