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 9992f1b7 feat: XA branch enrollment for autoCommit statements" (#1137)
9992f1b7 is described below
commit 9992f1b7ae8fafb8fd6e8850e6977c0545c28c97
Author: Larry <[email protected]>
AuthorDate: Thu Aug 6 17:42:21 2026 +0800
feat: XA branch enrollment for autoCommit statements" (#1137)
* feat: support XA branch reuse in autoCommit mode
Fix #904: busy buffer error when a SELECT ... FOR UPDATE is followed by
an UPDATE in XA mode with autoCommit=true.
When autoCommit=true, allow multiple SQL statements to reuse the same XA
branch instead of opening a new branch for every statement.
- Add IsAutoCommitXABranch to distinguish autoCommit-originated XA branches
- Introduce a DBResource-scoped XA branch registry for lifecycle and reuse
- Track a rollback flag to avoid duplicate XA branch rollback
- Skip undo log generation and AT-specific hooks for XA mode in
BuildExecutor
- Handle nil target in Tx.Rollback for non-XA modes
- Add XA mode test cases for BuildExecutor
* refactor: make each autoCommit statement a complete XA branch
Address review comments on the autoCommit XA path. The previous branch-reuse
machinery left the XA branch never actually started in autoCommit mode
(XA START was skipped in BeginTx and never issued later), so no data was
enrolled and phase-2 commit ran against an unprepared branch.
Instead, drive a full per-statement XA lifecycle (START -> exec -> END ->
PREPARE -> report) for each autoCommit statement, matching the connection
pool's ResetSession semantics where every checkout is a fresh branch.
- BeginTx: always XA START and set xaActive, for both explicit and
autoCommit transactions.
- createNewTxOnExecIfNeed: drop the GetXABranch/RegisterXABranch reuse and
cross-connection delegation; keep ErrSkip handling and the
xaActive-guarded
defer rollback.
- createOnceTxContext: restore GlobalLockRequire instead of the reuse flag.
- start/Commit/Rollback: remove the autoCommit branch-reuse special cases.
- Delete the unused xaRegistry and the DBResource.xaConnsByXID map plus its
Register/Get/UnregisterXABranch helpers (potential leak on abnormal exit).
- Remove TransactionContext.IsAutoCommitXABranch.
* fix: report phase-1 failure to TC when an autoCommit XA statement fails
On a real SQL error (non-ErrSkip), roll back through XATx.Rollback when a
branch was started, so the already-registered branch reports phase-1 failure
to the TC, instead of only doing a local XA END(FAIL) + ROLLBACK on the raw
connection. Fall back to the connection rollback for non-autoCommit paths.
Applied to both the exec error path and the defer safety net.
* fix: harden XA autoCommit executor routing, ErrSkip cleanup, and drop
dead undo-hook code
- executor: route XA mode to a pass-through executor before the parser
fallback so unparseable SQL is never sent to an AT executor
- conn_xa: on driver.ErrSkip after an XA branch is opened, roll the branch
back and return a real error instead of letting database/sql fall back to
non-XA execution (prevents branch leak and writes escaping the global tx)
- undo_log_hook: remove the unreachable XA skip; the hook is never
registered
(SQLTypeUnknown is dropped by RegisterHook), so the code was dead
* test: stabilize mysql trigger tests via injection seams
The trigger tests patched getColumnMetas/getIndexes with gomonkey
ApplyPrivateMethod, which rewrites machine code and silently fails to
apply once the compiler inlines those methods on a default build,
making the tests flaky (pass under -race / -gcflags=all=-l, fail
otherwise).
Switch to the getColumnMetasFn / getIndexesFn injection seams already
present on mysqlTrigger: the stubs now set those function fields
directly instead of monkey-patching, and the gomonkey dependency is
dropped from the test. Verified with default go test, -race and
-shuffle=on.
* fix: preallocate NamedValue slice with zero length in ExecWithValue
The pass-through branch built the converted args with
make([]driver.NamedValue, len(execCtx.Values)) and then appended,
producing a slice of double length whose leading half was empty
NamedValue{} elements with zero Ordinals. Any driver consuming those
args saw phantom leading parameters.
Use make(..., 0, len(...)) so the appended values are the only
elements, with 0-based Ordinals. The executor test now asserts the
exact converted []driver.NamedValue (length and every element),
covers the empty-values case, and the e.ex != nil delegation path.
* fix: harden XA autoCommit branch lifecycle
Three related fixes found while reviewing the autoCommit branch-reuse
path, all on the XA connection's per-statement branch lifecycle:
* Reuse: XAConn.Commit's phase-1 success path never cleared xaActive
(only the rollback/cleanup path did), and ResetSession lives on the
embedded *Conn so it could not reach the flag either. A second
autoCommit statement on the same pooled connection then tripped
BeginTx's "xa branch is active" guard and failed. Clear xaActive
after phase-1 (keeping prepareTime and the held xaBranchXid for
phase-2) and override XAConn.ResetSession as a backstop.
* Deferred query commit: committing the branch (XA END + XA PREPARE)
inline while a query's rows are still open issues a command on top of
an unread result set, which go-sql-driver rejects as a busy buffer /
commands out of sync and database/sql surfaces as "bad connection"
(issue #904, e.g. SELECT ... FOR UPDATE). Defer the branch commit
until the rows are closed via RowsCommitOnClose, mirroring AT mode.
* PREPARED-state rollback: IsAlreadyEnded only matched the IDLE-state
XAER_RMFAIL message, so when a branch was already prepared (XA END +
XA PREPARE ran but the phase-1 report to the TC failed) Rollback
bailed out before XA ROLLBACK, leaving the branch holding locks.
Match the PREPARED-state message too so XA ROLLBACK still runs.
Adds regression tests for each: same-XAConn reuse across autoCommit
statements, deferred commit until rows close, and PREPARED-branch
rollback still issuing XA ROLLBACK.
* test: cover the #904 SELECT FOR UPDATE + UPDATE sequence
Add TestXAConn_AutoCommit_SelectForUpdateThenUpdate, an end-to-end
regression for the exact #904 scenario: under an autoCommit global
transaction a "SELECT ... FOR UPDATE" is followed by an "UPDATE" on the
same pooled connection. It drives query -> drain/close -> commit branch
1 -> ResetSession -> UPDATE as branch 2, and asserts each statement
forms its own complete XA branch (two XA END + XA PREPARE + phase-1
reports) with no busy-buffer error. Verified it fails without the
xaActive reset fix and passes with it.
Also document the ErrSkip tradeoff at its source: because the XA branch
is opened before the statement runs, a statement that needs the
Prepare+Exec fallback (answers driver.ErrSkip) hard-errors under XA
autoCommit instead of retrying. Note lazy branch-open as the follow-up
that would remove the tradeoff.
* docs: changelog for XA autoCommit branch enrollment and #904 fix
Correct the feature note to describe the implemented behavior (each
autoCommit statement is registered as its own complete XA branch; N
statements create N branches at the TC) and add the #904 bugfix entry
for the deferred branch commit that avoids the busy-buffer error. Mirror
in dev_zh.md.
* fix: run in-branch Prepare+Exec fallback for XA autoCommit ErrSkip
Under an XA autoCommit global transaction the branch is opened (XA START)
before the statement runs, so a driver.ErrSkip retry could not be handed
back to database/sql without leaking the branch and letting the write
escape the global transaction. driver.ErrSkip is the common case: with the
default go-sql-driver DSN (interpolateParams=false) the direct
Execer/Queryer answers ErrSkip for any statement carrying bind parameters.
Instead of hard-erroring, createNewTxOnExecIfNeed now runs the Prepare+Exec
fallback itself on the same physical connection (still holding XA START)
via execPreparedInBranch / queryPreparedInBranch, so the retried statement
stays inside the branch and the normal XA END + XA PREPARE commit applies
unchanged. The query variant wraps the driver rows in rowsWithStmt so the
prepared statement outlives the result set (composing with
RowsCommitOnClose). Mirrors AT mode's existing prepared-statement fallback.
* test: cover query-path ErrSkip fallback and in-branch fallback error
rollback
Add two XA autoCommit cases around the in-branch ErrSkip fallback:
- ParameterizedSelectForUpdateErrSkipDefersBranchCommit: the real #904
shape,
a parameterized `SELECT ... FOR UPDATE WHERE id = ?` whose direct Queryer
answers driver.ErrSkip, exercises queryPreparedInBranch AND the
busy-buffer
guard together - the fallback rows are still wrapped in RowsCommitOnClose
so
XA END + XA PREPARE are deferred until the rows close, then fire exactly
once.
- InBranchFallbackErrorRollsBackBranch: when the in-branch Prepare+Exec
fallback
fails with a real (non-ErrSkip) error, the branch rolls back, reports
phase-1
FAILED to the TC, and surfaces the concrete error (never raw ErrSkip).
Adds test seams simulateQueryContextError and simulatePreparedExecError,
and a
query field on fakePreparedStmt so the prepared exec can be targeted by
query.
* fix: guard nil xaBranchXid in XAConn keep/release to prevent sweep panic
The two-phase timeout checker force-closes committed XA connections after
the hold time elapses. CloseForce -> cleanXABranchContext nils xaBranchXid
once the branch is no longer kept, then calls releaseIfNecessary, which
dereferenced the now-nil *XABranchXid via String() and panicked (SIGSEGV).
Add nil-guards to keepIfNecessary and releaseIfNecessary so the background
sweep of a committed connection no longer crashes. Surfaced by the #904
GORM SELECT ... FOR UPDATE + UPDATE end-to-end run.
---------
Co-authored-by: Ethan <[email protected]>
---
changes/dev.md | 2 +
changes/dev_zh.md | 2 +
pkg/datasource/sql/conn_xa.go | 270 +++++++++--
pkg/datasource/sql/conn_xa_test.go | 524 ++++++++++++++++++++-
.../sql/datasource/mysql/trigger_test.go | 97 ++--
pkg/datasource/sql/db.go | 2 +-
pkg/datasource/sql/exec/executor.go | 24 +-
pkg/datasource/sql/exec/executor_test.go | 110 ++++-
pkg/datasource/sql/tx.go | 11 +
pkg/datasource/sql/types/types.go | 2 +-
pkg/datasource/sql/xa/mysql_xa_connection.go | 21 +-
11 files changed, 945 insertions(+), 120 deletions(-)
diff --git a/changes/dev.md b/changes/dev.md
index e837c1a0..bf2d9ab0 100755
--- a/changes/dev.md
+++ b/changes/dev.md
@@ -26,11 +26,13 @@
### feature：
- [[#123](https://github.com/apache/incubator-seata-go/pull/123)] add two
phase and dubbo
+ - support XA branch enrollment for autoCommit statements in a global
transaction: each autoCommit statement is registered and prepared as its own
complete XA branch (note: N autoCommit statements create N branches at the TC);
parameterized statements (which the default go-sql-driver DSN answers with
`driver.ErrSkip`) are executed via an in-branch Prepare+Exec fallback so they
stay inside the branch
- support PostgreSQL XA via pgx driver
- [[#1130](https://github.com/apache/incubator-seata-go/issues/1130)]
support MySQL multi-value INSERT in AT mode for composite and mixed primary keys
### bugfix：
+ - [[#904](https://github.com/apache/incubator-seata-go/issues/904)] fix
"busy buffer" / "driver: bad connection" when a `SELECT ... FOR UPDATE` is
followed by another statement under XA autoCommit, by deferring the branch
commit (XA END + XA PREPARE) until the query rows are closed
- [[#130](https://github.com/apache/incubator-seata-go/pull/130)] getty
session auto close bug
- [[#991](https://github.com/apache/incubator-seata-go/issues/991)] fix
connection leaks and prevent nil pointer panic in async worker
- [[#887](https://github.com/apache/incubator-seata-go/issues/887)] make
DayValue serialization timezone-stable
diff --git a/changes/dev_zh.md b/changes/dev_zh.md
index 53615803..23ff3746 100644
--- a/changes/dev_zh.md
+++ b/changes/dev_zh.md
@@ -27,11 +27,13 @@ Seata-go 是一款开源的分布式事务解决方案，提供高性能和简
### feature：
- [[#123](https://github.com/apache/incubator-seata-go/pull/123)]
添加二阶段事务接口，以及dubbo集成
+- 支持全局事务中 autoCommit 语句的 XA 分支注册：每条 autoCommit 语句都作为一个完整的 XA 分支单独注册并
prepare（注意：N 条 autoCommit 语句会在 TC 侧产生 N 个分支）；带参数的语句（默认 go-sql-driver DSN 会返回
`driver.ErrSkip`）通过分支内 Prepare+Exec 回退执行，从而保证仍在分支内完成
- 支持基于 pgx 驱动的 PostgreSQL XA
- [[#1130](https://github.com/apache/incubator-seata-go/issues/1130)] 支持 AT
模式下 MySQL 多值 INSERT 的复合主键与混合主键场景
### bugfix：
+- [[#904](https://github.com/apache/incubator-seata-go/issues/904)] 修复 XA
autoCommit 下 `SELECT ... FOR UPDATE` 后紧接其他语句导致的 "busy buffer" / "driver: bad
connection"：将分支提交（XA END + XA PREPARE）延迟到查询结果集关闭之后再执行
- [[#130](https://github.com/apache/incubator-seata-go/pull/130)] 修复getty
session自动关闭的bug
### optimize：
diff --git a/pkg/datasource/sql/conn_xa.go b/pkg/datasource/sql/conn_xa.go
index 857e885b..a7c73cf6 100644
--- a/pkg/datasource/sql/conn_xa.go
+++ b/pkg/datasource/sql/conn_xa.go
@@ -90,7 +90,7 @@ func (c *XAConn) QueryContext(ctx context.Context, query
string, args []driver.N
}()
}
- ret, err := c.createNewTxOnExecIfNeed(ctx, func() (types.ExecResult,
error) {
+ ret, err := c.createNewTxOnExecIfNeed(ctx, true, query, args, func()
(types.ExecResult, error) {
ret, err := c.Conn.QueryContext(ctx, query, args)
if err != nil {
return nil, err
@@ -110,7 +110,7 @@ func (c *XAConn) ExecContext(ctx context.Context, query
string, args []driver.Na
}()
}
- ret, err := c.createNewTxOnExecIfNeed(ctx, func() (types.ExecResult,
error) {
+ ret, err := c.createNewTxOnExecIfNeed(ctx, false, query, args, func()
(types.ExecResult, error) {
ret, err := c.Conn.ExecContext(ctx, query, args)
if err != nil {
return nil, err
@@ -156,33 +156,31 @@ func (c *XAConn) BeginTx(ctx context.Context, opts
driver.TxOptions) (driver.Tx,
return nil, err
}
- if !c.autoCommit {
- if c.xaActive {
- return nil, errors.New("should NEVER happen:
setAutoCommit from true to false while xa branch is active")
- }
+ if c.xaActive {
+ return nil, errors.New("should NEVER happen: setAutoCommit from
true to false while xa branch is active")
+ }
- baseTx, ok := tx.(*Tx)
- if !ok {
- return nil, fmt.Errorf("start xa %s transaction failure
for the tx is a wrong type", c.txCtx.XID)
- }
+ baseTx, ok := tx.(*Tx)
+ if !ok {
+ return nil, fmt.Errorf("start xa %s transaction failure for the
tx is a wrong type", c.txCtx.XID)
+ }
- baseTx.xaConn = c
+ baseTx.xaConn = c
- c.branchRegisterTime = time.Now()
- if err := baseTx.register(c.txCtx); err != nil {
- c.cleanXABranchContext()
- return nil, fmt.Errorf("failed to register xa branch
%s, err:%w", c.txCtx.XID, err)
- }
+ c.branchRegisterTime = time.Now()
+ if err := baseTx.register(c.txCtx); err != nil {
+ c.cleanXABranchContext()
+ return nil, fmt.Errorf("failed to register xa branch %s,
err:%w", c.txCtx.XID, err)
+ }
- c.xaBranchXid = XaIdBuild(c.txCtx.XID, c.txCtx.BranchID)
- c.keepIfNecessary()
+ c.xaBranchXid = XaIdBuild(c.txCtx.XID, c.txCtx.BranchID)
+ c.keepIfNecessary()
- if err = c.start(ctx); err != nil {
- c.cleanXABranchContext()
- return nil, fmt.Errorf("failed to start xa branch
xid:%s err:%w", c.txCtx.XID, err)
- }
- c.xaActive = true
+ if err = c.start(ctx); err != nil {
+ c.cleanXABranchContext()
+ return nil, fmt.Errorf("failed to start xa branch xid:%s
err:%w", c.txCtx.XID, err)
}
+ c.xaActive = true
return &XATx{tx: tx.(*Tx)}, nil
}
@@ -202,31 +200,40 @@ func (c *XAConn) createOnceTxContext(ctx context.Context)
bool {
return onceTx
}
-func (c *XAConn) createNewTxOnExecIfNeed(ctx context.Context, f func()
(types.ExecResult, error)) (types.ExecResult, error) {
+func (c *XAConn) createNewTxOnExecIfNeed(ctx context.Context, isQuery bool,
query string, args []driver.NamedValue, f func() (types.ExecResult, error))
(types.ExecResult, error) {
var (
- tx driver.Tx
- err error
+ tx driver.Tx
+ err error
+ xaRollbacked bool // Track if XA rollback was already done to
avoid duplicate rollback
)
defer func() {
recoverErr := recover()
- 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 {
- if rollbackErr := c.Rollback(ctx); rollbackErr
!= nil {
- log.Errorf("conn xa rollback error:%v",
rollbackErr)
+ // Check if error is ErrSkip - don't rollback for this special
error
+ isErrSkip := err != nil && errors.Is(err, driver.ErrSkip)
+
+ if (err != nil && !isErrSkip) || recoverErr != nil {
+ // Prefer XATx.Rollback so a registered branch reports
phase-1 failure to
+ // the TC; fall back to the raw connection rollback for
non-autoCommit paths.
+ if !xaRollbacked {
+ if tx != nil {
+ if rollbackErr := tx.Rollback();
rollbackErr != nil {
+ log.Errorf("defer rollback xa
branch error:%v", rollbackErr)
+ }
+ xaRollbacked = true
+ } else if c.xaActive {
+ if rollbackErr := c.Rollback(ctx);
rollbackErr != nil {
+ log.Errorf("defer rollback xa
branch error:%v", rollbackErr)
+ }
+ xaRollbacked = true
}
}
}
}()
currentAutoCommit := c.autoCommit
+
+ // For global transactions in autoCommit mode, each statement is a
complete XA branch
if c.txCtx.TransactionMode != types.Local && tm.IsGlobalTx(ctx) &&
c.autoCommit {
tx, err = c.BeginTx(ctx, driver.TxOptions{Isolation:
driver.IsolationLevel(gosql.LevelDefault)})
if err != nil {
@@ -236,23 +243,77 @@ func (c *XAConn) createNewTxOnExecIfNeed(ctx
context.Context, f func() (types.Ex
// execute SQL
ret, err := f()
+ if err != nil && errors.Is(err, driver.ErrSkip) {
+ // driver.ErrSkip is not a real failure: with the default
go-sql-driver DSN
+ // (interpolateParams=false) the direct Execer/Queryer answers
ErrSkip for any
+ // statement carrying bind parameters, asking database/sql to
retry it through
+ // the Prepare+Exec fallback path.
+ if tx == nil {
+ // No XA branch opened for this statement - safe to
hand the retry back to
+ // database/sql and let it run its own Prepare+Exec
fallback.
+ return nil, err
+ }
+ // We already opened an XA branch (XA START) for this
statement. We cannot hand
+ // the retry back to database/sql: it would run on a
*different* pooled
+ // connection (autoCommit is now false and txCtx has been
reset), leaving this
+ // branch registered-but-never-prepared (a leak) and letting
the retried write
+ // escape the global transaction. Instead run the Prepare+Exec
fallback
+ // OURSELVES on this same physical connection, which still
holds XA START open,
+ // so the retried statement stays inside the branch and the
normal XA END +
+ // XA PREPARE commit below applies unchanged. These helpers
never return ErrSkip.
+ if isQuery {
+ ret, err = c.queryPreparedInBranch(ctx, query, args)
+ } else {
+ ret, err = c.execPreparedInBranch(ctx, query, args)
+ }
+ }
if err != nil {
+ // On real error, rollback the entire branch. Prefer
XATx.Rollback so the
+ // already-registered branch reports phase-1 failure to the TC;
fall back to
+ // the raw connection rollback for non-autoCommit paths.
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 {
+ } else if c.xaActive {
if rollbackErr := c.Rollback(ctx); rollbackErr != nil {
log.Errorf("failed to rollback xa branch of
:%s, err:%v", c.txCtx.XID, rollbackErr)
}
}
+ xaRollbacked = true // Mark that rollback was handled
return nil, err
}
+ // For autoCommit mode with global transaction, commit the branch now:
+ // XA END + XA PREPARE + report phase-1 success to TC.
if tx != nil && currentAutoCommit {
- // Commit through XATx so phase-one reporting stays coupled to
driver.Tx lifecycle.
+ // A query statement returns an open result set that still
occupies this
+ // connection's read buffer. Running XA END + XA PREPARE here -
before the
+ // caller has drained/closed the rows - issues a new command on
top of that
+ // unread result set, which go-sql-driver rejects as a "busy
buffer" /
+ // "commands out of sync" error and database/sql surfaces as
+ // "driver: bad connection", forcing the whole transaction to
roll back
+ // (issue #904, e.g. SELECT ... FOR UPDATE followed by UPDATE).
Defer the
+ // branch commit until the caller closes the rows, mirroring AT
mode's
+ // RowsCommitOnClose handling. xaDeferredCommitTx keeps the
inline path's
+ // rollback-on-commit-failure semantics so a failed deferred
commit never
+ // leaves a prepared branch holding locks. (GetRows must only
be called on
+ // a query result - it panics on a write result - hence the
isQuery gate.)
+ if isQuery {
+ if dr := ret.GetRows(); dr != nil {
+ return
types.NewResult(types.WithRows(&RowsCommitOnClose{
+ rows: dr,
+ tx: xaDeferredCommitTx{tx: tx},
+ })), nil
+ }
+ }
if err = tx.Commit(); err != nil {
- log.Errorf("xa connection proxy commit failure xid:%s,
err:%v", c.txCtx.XID, err)
+ log.Errorf("xa transaction commit failure xid:%s,
err:%v", c.txCtx.XID, err)
+ // XA End & Rollback
+ if rollbackErr := tx.Rollback(); rollbackErr != nil {
+ log.Errorf("xa transaction rollback failure
xid:%s, err:%v", c.txCtx.XID, rollbackErr)
+ }
+ xaRollbacked = true
return nil, err
}
}
@@ -260,7 +321,104 @@ func (c *XAConn) createNewTxOnExecIfNeed(ctx
context.Context, f func() (types.Ex
return ret, nil
}
+// execPreparedInBranch runs an ExecContext statement through the driver's
+// Prepare+Exec path on the XAConn's own physical connection, which is still
inside
+// the open XA branch (XA START has been issued and not yet ended). It exists
so a
+// statement that answers driver.ErrSkip on the direct Execer path - the
default
+// go-sql-driver behavior for parameterized statements - can still be executed
+// without handing the retry back to database/sql, which would run it on a
different
+// connection outside the branch. Unlike the direct path this never returns
+// driver.ErrSkip: it either produces a concrete result or a concrete error.
+func (c *XAConn) execPreparedInBranch(ctx context.Context, query string, args
[]driver.NamedValue) (types.ExecResult, error) {
+ preparer, ok := c.Conn.targetConn.(driver.ConnPrepareContext)
+ if !ok {
+ return nil, fmt.Errorf("xa branch %s: driver connection does
not support PrepareContext, cannot recover from ErrSkip", c.txCtx.XID)
+ }
+ stmt, err := preparer.PrepareContext(ctx, query)
+ if err != nil {
+ return nil, err
+ }
+ defer stmt.Close()
+
+ execer, ok := stmt.(driver.StmtExecContext)
+ if !ok {
+ return nil, fmt.Errorf("xa branch %s: prepared statement does
not support ExecContext, cannot recover from ErrSkip", c.txCtx.XID)
+ }
+ res, err := execer.ExecContext(ctx, args)
+ if err != nil {
+ return nil, err
+ }
+ return types.NewResult(types.WithResult(res)), nil
+}
+
+// queryPreparedInBranch is the QueryContext counterpart of
execPreparedInBranch.
+// The prepared statement must outlive the result set, so it is wrapped in
+// rowsWithStmt, which closes the statement when the rows are closed (this
composes
+// with RowsCommitOnClose: draining the rows closes both the driver rows and
the
+// statement and then runs the deferred XA END + XA PREPARE).
+func (c *XAConn) queryPreparedInBranch(ctx context.Context, query string, args
[]driver.NamedValue) (types.ExecResult, error) {
+ preparer, ok := c.Conn.targetConn.(driver.ConnPrepareContext)
+ if !ok {
+ return nil, fmt.Errorf("xa branch %s: driver connection does
not support PrepareContext, cannot recover from ErrSkip", c.txCtx.XID)
+ }
+ stmt, err := preparer.PrepareContext(ctx, query)
+ if err != nil {
+ return nil, err
+ }
+ queryer, ok := stmt.(driver.StmtQueryContext)
+ if !ok {
+ _ = stmt.Close()
+ return nil, fmt.Errorf("xa branch %s: prepared statement does
not support QueryContext, cannot recover from ErrSkip", c.txCtx.XID)
+ }
+ rows, err := queryer.QueryContext(ctx, args)
+ if err != nil {
+ _ = stmt.Close()
+ return nil, err
+ }
+ return types.NewResult(types.WithRows(&rowsWithStmt{Rows: rows, stmt:
stmt})), nil
+}
+
+// xaDeferredCommitTx wraps an XA branch tx whose commit is deferred until the
+// query's rows are closed (see createNewTxOnExecIfNeed / RowsCommitOnClose).
+// It mirrors the inline exec path: if the deferred XA END + XA PREPARE (or the
+// phase-1 report to the TC) fails, the branch is rolled back so it does not
stay
+// prepared and hold locks.
+type xaDeferredCommitTx struct {
+ tx driver.Tx
+}
+
+func (t xaDeferredCommitTx) Commit() error {
+ if err := t.tx.Commit(); err != nil {
+ log.Errorf("deferred xa branch commit failed, rolling back
branch: %v", err)
+ if rollbackErr := t.tx.Rollback(); rollbackErr != nil {
+ log.Errorf("deferred xa branch rollback failed: %v",
rollbackErr)
+ }
+ return err
+ }
+ return nil
+}
+
+func (t xaDeferredCommitTx) Rollback() error {
+ return t.tx.Rollback()
+}
+
+// ResetSession is called by database/sql before reusing a pooled connection.
+// XAConn.xaActive lives on the XA wrapper, so the embedded *Conn.ResetSession
+// cannot clear it; without this override a connection whose previous
autoCommit
+// branch already completed phase-1 would keep xaActive=true and the next
+// statement's BeginTx would fail the "xa branch is active" guard. Clearing it
+// here (in addition to XAConn.Commit) is a defensive backstop for any path
that
+// leaves a stale flag. xaBranchXid is intentionally left untouched so a held
+// branch remains available for phase-2.
+func (c *XAConn) ResetSession(ctx context.Context) error {
+ c.xaActive = false
+ return c.Conn.ResetSession(ctx)
+}
+
func (c *XAConn) keepIfNecessary() {
+ if c.xaBranchXid == nil {
+ return
+ }
if c.ShouldBeHeld() {
if err := c.res.Hold(c.xaBranchXid.String(), c); err == nil {
c.isConnKept = true
@@ -269,6 +427,14 @@ func (c *XAConn) keepIfNecessary() {
}
func (c *XAConn) releaseIfNecessary() {
+ // cleanXABranchContext nils xaBranchXid once a branch is no longer
kept, and
+ // the two-phase timeout checker force-closes committed connections
after the
+ // hold time elapses. Guard against the nil branch xid so that sweep
(which
+ // calls CloseForce -> cleanXABranchContext -> releaseIfNecessary) does
not
+ // dereference a nil *XABranchXid via String().
+ if c.xaBranchXid == nil {
+ return
+ }
if c.ShouldBeHeld() && c.xaBranchXid.String() != "" {
if c.isConnKept {
c.res.Release(c.xaBranchXid.String())
@@ -294,7 +460,7 @@ func (c *XAConn) start(ctx context.Context) error {
c.XaRollback(ctx, c.xaBranchXid)
return err
}
- return err
+ return nil
}
func (c *XAConn) end(ctx context.Context, flags int) error {
@@ -341,10 +507,9 @@ func (c *XAConn) Rollback(ctx context.Context) error {
// 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 c.xaErrorClassifier.IsAlreadyEnded(err) {
- // If already ended, continue with rollback
log.Infof("XA branch already ended, continuing
with rollback for xid: %s", c.txCtx.XID)
+ // Already ended, continue with rollback
} else {
return c.rollbackErrorHandle()
}
@@ -358,6 +523,7 @@ func (c *XAConn) Rollback(ctx context.Context) error {
c.rollBacked = true
}
c.cleanXABranchContext()
+
return nil
}
@@ -375,6 +541,7 @@ func (c *XAConn) Commit(ctx context.Context) error {
}
now := time.Now()
+
if c.end(ctx, xa.TMSuccess) != nil {
return c.commitErrorHandle(ctx)
}
@@ -388,6 +555,16 @@ func (c *XAConn) Commit(ctx context.Context) error {
}
c.prepareTime = time.Now()
+
+ // Phase-1 is done: this session no longer has an in-flight XA branch.
Clear
+ // only the session-active flag so a subsequent autoCommit statement on
the
+ // same (possibly pooled/reused) connection can open a fresh branch
instead of
+ // tripping the "xa branch is active" guard in BeginTx. The branch
itself is
+ // still prepared and, when held, retrievable for phase-2 via
xaBranchXid, so
+ // we must NOT call cleanXABranchContext here (that would reset
prepareTime and
+ // drop xaBranchXid). Phase-2 XaCommit/XaRollback do not depend on
xaActive.
+ c.xaActive = false
+
return nil
}
@@ -418,10 +595,11 @@ func (c *XAConn) Close() error {
return nil
}
c.cleanXABranchContext()
- if err := c.Conn.Close(); err != nil {
- return err
+ // Check if Conn is nil before calling Close
+ if c.Conn == nil {
+ return nil
}
- return nil
+ return c.Conn.Close()
}
func (c *XAConn) CloseForce() error {
diff --git a/pkg/datasource/sql/conn_xa_test.go
b/pkg/datasource/sql/conn_xa_test.go
index 46fc412c..4174b5a0 100644
--- a/pkg/datasource/sql/conn_xa_test.go
+++ b/pkg/datasource/sql/conn_xa_test.go
@@ -21,6 +21,7 @@ import (
"context"
"database/sql"
"database/sql/driver"
+ "errors"
"io"
"strings"
"sync/atomic"
@@ -53,8 +54,7 @@ func (m *mysqlMockRows) Columns() []string {
}
func (m *mysqlMockRows) Close() error {
- //TODO implement me
- panic("implement me")
+ return nil
}
func (m *mysqlMockRows) Next(dest []driver.Value) error {
@@ -104,9 +104,60 @@ func (mi *mockSQLInterceptor) After(ctx context.Context,
execCtx *types.ExecCont
}
// simulateExecContextError allows tests to inject driver errors for certain
SQL strings.
-// When set, baseMockConn will call this hook for each ExecContext.
+// When set, baseMockConn will call this hook for each direct ExecContext.
var simulateExecContextError func(query string) error
+// simulateQueryContextError injects driver errors for certain SQL strings on
the
+// direct QueryContext path (e.g. returning driver.ErrSkip for a parameterized
+// SELECT, as the default go-sql-driver DSN does). When set, baseMockConn
calls it
+// for each direct QueryContext.
+var simulateQueryContextError func(query string) error
+
+// simulatePreparedExecError injects an error from the prepared statement's
+// ExecContext, keyed by the query it was prepared with. It lets tests drive
the
+// case where the in-branch Prepare+Exec fallback itself fails with a real (non
+// ErrSkip) error, so the branch must roll back and report phase-1 failure.
+var simulatePreparedExecError func(query string) error
+
+// fakePreparedStmt models a driver prepared statement. It is what the driver
+// returns from PrepareContext, and its ExecContext/QueryContext succeed by
default -
+// mirroring the real go-sql-driver, where the direct Execer answers
driver.ErrSkip
+// for parameterized statements but the Prepare+Exec path executes fine. This
lets
+// tests exercise XAConn's in-branch ErrSkip fallback (execPreparedInBranch /
+// queryPreparedInBranch). simulatePreparedExecError can force the prepared
exec to
+// fail for the fallback-error rollback path.
+type fakePreparedStmt struct {
+ query string
+}
+
+func (s *fakePreparedStmt) Close() error { return nil }
+func (s *fakePreparedStmt) NumInput() int { return -1 }
+
+func (s *fakePreparedStmt) Exec(args []driver.Value) (driver.Result, error) {
+ return &driver.ResultNoRows, nil
+}
+
+func (s *fakePreparedStmt) Query(args []driver.Value) (driver.Rows, error) {
+ rows := &mysqlMockRows{}
+ rows.data = [][]interface{}{{"8.0.29"}}
+ return rows, nil
+}
+
+func (s *fakePreparedStmt) ExecContext(ctx context.Context, args
[]driver.NamedValue) (driver.Result, error) {
+ if simulatePreparedExecError != nil {
+ if err := simulatePreparedExecError(s.query); err != nil {
+ return nil, err
+ }
+ }
+ return &driver.ResultNoRows, nil
+}
+
+func (s *fakePreparedStmt) QueryContext(ctx context.Context, args
[]driver.NamedValue) (driver.Rows, error) {
+ rows := &mysqlMockRows{}
+ rows.data = [][]interface{}{{"8.0.29"}}
+ return rows, nil
+}
+
func baseMockConn(mockConn *mock.MockTestDriverConn) {
branchStatusCache = gcache.New(1024).LRU().Expiration(time.Minute *
10).Build()
@@ -124,8 +175,22 @@ func baseMockConn(mockConn *mock.MockTestDriverConn) {
mockConn.EXPECT().ResetSession(gomock.Any()).AnyTimes().Return(nil)
mockConn.EXPECT().Close().AnyTimes().Return(nil)
+ // The Prepare+Exec fallback path (used when the direct ExecContext
answers
+ // driver.ErrSkip) prepares on the same physical connection and runs the
+ // statement through the prepared stmt, which succeeds by default. The
prepared
+ // stmt keeps the query so simulatePreparedExecError can target it.
+ mockConn.EXPECT().PrepareContext(gomock.Any(),
gomock.Any()).AnyTimes().DoAndReturn(
+ func(ctx context.Context, query string) (driver.Stmt, error) {
+ return &fakePreparedStmt{query: query}, nil
+ })
+
mockConn.EXPECT().QueryContext(gomock.Any(), gomock.Any(),
gomock.Any()).AnyTimes().DoAndReturn(
func(ctx context.Context, query string, args
[]driver.NamedValue) (driver.Rows, error) {
+ if simulateQueryContextError != nil {
+ if err := simulateQueryContextError(query); err
!= nil {
+ return nil, err
+ }
+ }
rows := &mysqlMockRows{}
rows.data = [][]interface{}{
{"8.0.29"},
@@ -376,6 +441,14 @@ func TestXAConn_Rollback_XAER_RMFAIL(t *testing.T) {
},
want: true,
},
+ {
+ name: "matching XAER_RMFAIL error with PREPARED state",
+ err: &mysql.MySQLError{
+ Number: 1399,
+ Message: "Error 1399 (XAE07): XAER_RMFAIL: The
command cannot be executed when global transaction is in the PREPARED state",
+ },
+ want: true,
+ },
{
name: "matching XAER_RMFAIL error with already ended",
err: &mysql.MySQLError{
@@ -447,6 +520,51 @@ func TestXAConn_Rollback_HandleXAERRMFAILAlreadyEnded(t
*testing.T) {
}
}
+// Reproduces the review scenario where the branch is already PREPARED when
Rollback runs:
+// during autoCommit Commit the DB executed XA END + XA PREPARE, but the
phase-1 report to
+// the TC failed, so the branch is left in the PREPARED state. The follow-up
rollback issues
+// XA END(TMFAIL), which MySQL rejects with XAER_RMFAIL "...PREPARED state".
Before the fix
+// IsAlreadyEnded only recognized the IDLE-state message, so Rollback bailed
out via
+// rollbackErrorHandle() BEFORE running XA ROLLBACK, leaving the branch
holding locks forever.
+// This asserts XA ROLLBACK is still issued so the prepared branch releases
its locks.
+func TestXAConn_Rollback_PreparedBranchStillRollsBack(t *testing.T) {
+ ctrl, db, _, _ := initXAConnTestResource(t)
+ defer func() {
+ simulateExecContextError = nil
+ db.Close()
+ ctrl.Finish()
+ CleanTxHooks()
+ }()
+
+ ctx := tm.InitSeataContext(context.Background())
+ tm.SetXID(ctx, uuid.New().String())
+
+ var rollbackSeen int32
+ // Inject: XA END returns XAER_RMFAIL with the PREPARED-state message;
user SQL fails to
+ // trigger the rollback path; record whether XA ROLLBACK is
subsequently issued.
+ simulateExecContextError = func(query string) error {
+ upper := strings.ToUpper(strings.TrimSpace(query))
+ switch {
+ case 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 PREPARED state",
+ }
+ case strings.HasPrefix(upper, "XA ROLLBACK"):
+ atomic.StoreInt32(&rollbackSeen, 1)
+ return nil
+ case !strings.HasPrefix(upper, "XA "):
+ return io.EOF
+ }
+ return nil
+ }
+
+ _, err := db.ExecContext(ctx, "UPDATE user SET age = 1 WHERE id = 1")
+ assert.Error(t, err, "expected error to trigger rollback path")
+ assert.Equal(t, int32(1), atomic.LoadInt32(&rollbackSeen),
+ "XA ROLLBACK must run so a PREPARED branch releases its locks")
+}
+
func TestXAConn_ExecContext_AutoCommitReportsPhaseOneDone(t *testing.T) {
ctrl := gomock.NewController(t)
defer ctrl.Finish()
@@ -479,6 +597,406 @@ func
TestXAConn_ExecContext_AutoCommitReportsPhaseOneDone(t *testing.T) {
assert.Equal(t, int32(1), atomic.LoadInt32(&commitCnt))
}
+// Regression for the autoCommit branch-reuse bug: after a statement's XA
branch
+// completes phase-1 (XA END + XA PREPARE + report), the session must no
longer be
+// marked as having an active branch, otherwise the next autoCommit statement
on the
+// SAME physical connection (which database/sql reuses via the pool, calling
+// ResetSession in between) trips BeginTx's "should NEVER happen:
setAutoCommit from
+// true to false while xa branch is active" guard. Before the fix,
XAConn.Commit's
+// success path never cleared xaActive (only the rollback/cleanup path did) and
+// ResetSession - living on the embedded *Conn - could not reach it, so the
second
+// statement always failed.
+func TestXAConn_ExecContext_ReuseAfterAutoCommitBranch(t *testing.T) {
+ ctrl := gomock.NewController(t)
+ defer ctrl.Finish()
+ CleanTxHooks()
+ defer CleanTxHooks()
+
+ // XAConn.Commit -> checkTimeout compares branchRegisterTime against
xaConnTimeout,
+ // which is 0 unless InitXA runs. Give the branch a real budget so
phase-1 prepares.
+ prevTimeout := xaConnTimeout
+ xaConnTimeout = time.Minute
+ defer func() { xaConnTimeout = prevTimeout }()
+
+ xaConn, mockMgr := newMockXAConn(t, ctrl, 123)
+ mockMgr.EXPECT().BranchReport(gomock.Any(),
gomock.Any()).AnyTimes().Return(nil)
+
+ 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())
+
+ // First autoCommit statement: opens and completes a full XA branch.
+ _, err := xaConn.ExecContext(ctx, "SELECT 1", nil)
+ assert.NoError(t, err)
+ // The Commit success path must clear the session-active flag on its
own, so the
+ // fix holds even for paths where database/sql does not call
ResetSession.
+ assert.False(t, xaConn.xaActive, "xaActive must be cleared after
phase-1 completes")
+
+ // Simulate database/sql returning the connection to the pool and
reusing it:
+ // ResetSession restores autoCommit=true (and, via the XAConn override,
clears the
+ // XA session flag as a backstop).
+ assert.NoError(t, xaConn.ResetSession(ctx))
+ assert.True(t, xaConn.autoCommit, "ResetSession must restore autoCommit
for pooled reuse")
+ assert.False(t, xaConn.xaActive, "ResetSession must leave no active XA
branch")
+
+ // Second autoCommit statement on the SAME XAConn must open a fresh
branch instead
+ // of failing the "xa branch is active" guard.
+ _, err = xaConn.ExecContext(ctx, "SELECT 2", nil)
+ assert.NoError(t, err, "second autoCommit statement on a reused XAConn
must succeed")
+
+ assert.Equal(t, int32(2), atomic.LoadInt32(&commitCnt))
+}
+
+// Reproduces the #904 "busy buffer" scenario on the query path: a SELECT ...
FOR
+// UPDATE opens a result set that still occupies the connection's read buffer.
If the
+// autoCommit branch were committed inline (XA END + XA PREPARE) while those
rows are
+// open, go-sql-driver would reject the new command with a "busy buffer" /
+// "commands out of sync" error surfacing as "driver: bad connection". This
asserts the
+// branch commit is deferred: XA END / XA PREPARE / the phase-1 report only
run once the
+// caller closes the rows, so the busy-buffer collision never happens.
+func TestXAConn_QueryContext_DefersBranchCommitUntilRowsClose(t *testing.T) {
+ ctrl := gomock.NewController(t)
+ defer ctrl.Finish()
+ CleanTxHooks()
+ defer func() {
+ simulateExecContextError = nil
+ CleanTxHooks()
+ }()
+
+ // checkTimeout compares against xaConnTimeout, which is only set by
InitXA in a
+ // running server. Give the branch a real budget so the deferred commit
prepares
+ // instead of aborting as timed-out.
+ prevTimeout := xaConnTimeout
+ xaConnTimeout = time.Minute
+ defer func() { xaConnTimeout = prevTimeout }()
+
+ xaConn, mockMgr := newMockXAConn(t, ctrl, 123)
+
+ var reported int32
+ mockMgr.EXPECT().BranchReport(gomock.Any(), gomock.Any()).DoAndReturn(
+ func(ctx context.Context, param rm.BranchReportParam) error {
+ assert.EqualValues(t, branch.BranchStatusPhaseoneDone,
param.Status)
+ atomic.StoreInt32(&reported, 1)
+ return nil
+ },
+ ).Times(1)
+
+ // Record when the branch-commit statements run on the physical
connection.
+ var endSeen, prepareSeen int32
+ simulateExecContextError = func(query string) error {
+ upper := strings.ToUpper(strings.TrimSpace(query))
+ switch {
+ case strings.HasPrefix(upper, "XA END"):
+ atomic.StoreInt32(&endSeen, 1)
+ case strings.HasPrefix(upper, "XA PREPARE"):
+ atomic.StoreInt32(&prepareSeen, 1)
+ }
+ return nil
+ }
+
+ ctx := tm.InitSeataContext(context.Background())
+ tm.SetXID(ctx, uuid.NewString())
+
+ rows, err := xaConn.QueryContext(ctx, "SELECT * FROM user WHERE id = 1
FOR UPDATE", nil)
+ assert.NoError(t, err)
+
+ // While the result set is still open, the branch must NOT have been
committed -
+ // issuing XA END / XA PREPARE here is exactly the #904 busy-buffer
trigger.
+ assert.Equal(t, int32(0), atomic.LoadInt32(&endSeen), "XA END must be
deferred until rows close")
+ assert.Equal(t, int32(0), atomic.LoadInt32(&prepareSeen), "XA PREPARE
must be deferred until rows close")
+ assert.Equal(t, int32(0), atomic.LoadInt32(&reported), "phase-1 report
must be deferred until rows close")
+
+ // The returned rows must be the deferred-commit wrapper.
+ _, ok := rows.(*RowsCommitOnClose)
+ assert.True(t, ok, "XA query rows must be wrapped in RowsCommitOnClose
to defer the branch commit")
+
+ // Closing the rows drains the connection first, then runs XA END + XA
PREPARE + report.
+ assert.NoError(t, rows.Close())
+
+ assert.Equal(t, int32(1), atomic.LoadInt32(&endSeen), "XA END must run
once rows are closed")
+ assert.Equal(t, int32(1), atomic.LoadInt32(&prepareSeen), "XA PREPARE
must run once rows are closed")
+ assert.Equal(t, int32(1), atomic.LoadInt32(&reported), "phase-1 report
must run once rows are closed")
+}
+
+// End-to-end regression for the exact #904 sequence: under an autoCommit
global
+// transaction, a "SELECT ... FOR UPDATE" is immediately followed by an
"UPDATE" on the
+// SAME physical connection. The SELECT's open result set occupies the
connection's read
+// buffer; the busy-buffer error struck because the first branch used to be
committed
+// inline (XA END + XA PREPARE) while those rows were still open, then the
second
+// statement could not open its own branch. This drives the full flow - query,
drain,
+// commit branch 1, pool reuse (ResetSession), then the UPDATE as branch 2 -
and asserts
+// each statement forms its own complete branch (two XA END + XA PREPARE +
phase-1
+// reports) with no error, so the busy-buffer collision cannot recur.
+func TestXAConn_AutoCommit_SelectForUpdateThenUpdate(t *testing.T) {
+ ctrl := gomock.NewController(t)
+ defer ctrl.Finish()
+ CleanTxHooks()
+ defer func() {
+ simulateExecContextError = nil
+ CleanTxHooks()
+ }()
+
+ prevTimeout := xaConnTimeout
+ xaConnTimeout = time.Minute
+ defer func() { xaConnTimeout = prevTimeout }()
+
+ xaConn, mockMgr := newMockXAConn(t, ctrl, 123)
+
+ var reportCnt int32
+ mockMgr.EXPECT().BranchReport(gomock.Any(), gomock.Any()).DoAndReturn(
+ func(ctx context.Context, param rm.BranchReportParam) error {
+ assert.EqualValues(t, branch.BranchStatusPhaseoneDone,
param.Status)
+ atomic.AddInt32(&reportCnt, 1)
+ return nil
+ },
+ ).AnyTimes()
+
+ var endCnt, prepareCnt int32
+ simulateExecContextError = func(query string) error {
+ upper := strings.ToUpper(strings.TrimSpace(query))
+ switch {
+ case strings.HasPrefix(upper, "XA END"):
+ atomic.AddInt32(&endCnt, 1)
+ case strings.HasPrefix(upper, "XA PREPARE"):
+ atomic.AddInt32(&prepareCnt, 1)
+ }
+ return nil
+ }
+
+ ctx := tm.InitSeataContext(context.Background())
+ tm.SetXID(ctx, uuid.NewString())
+
+ // Statement 1: SELECT ... FOR UPDATE. The branch commit is deferred
while the rows
+ // are open, so no XA END / XA PREPARE fires yet - that would be the
busy-buffer bug.
+ rows, err := xaConn.QueryContext(ctx, "SELECT * FROM user WHERE id = 1
FOR UPDATE", nil)
+ assert.NoError(t, err)
+ assert.Equal(t, int32(0), atomic.LoadInt32(&endCnt), "branch 1 must not
commit while its rows are open")
+
+ // Draining/closing the rows completes branch 1 (XA END + XA PREPARE +
report).
+ assert.NoError(t, rows.Close())
+ assert.Equal(t, int32(1), atomic.LoadInt32(&endCnt), "branch 1 commits
once its rows close")
+ assert.Equal(t, int32(1), atomic.LoadInt32(&prepareCnt))
+ assert.False(t, xaConn.xaActive, "branch 1 must leave no active branch
on the session")
+
+ // database/sql returns the connection to the pool and resets it before
reuse.
+ assert.NoError(t, xaConn.ResetSession(ctx))
+
+ // Statement 2: the follow-up UPDATE on the SAME connection must form
its own branch.
+ _, err = xaConn.ExecContext(ctx, "UPDATE user SET age = age + 1 WHERE
id = 1", nil)
+ assert.NoError(t, err, "UPDATE after SELECT ... FOR UPDATE must succeed
(no busy buffer)")
+
+ assert.Equal(t, int32(2), atomic.LoadInt32(&endCnt), "each statement
forms one complete XA branch")
+ assert.Equal(t, int32(2), atomic.LoadInt32(&prepareCnt))
+ assert.Equal(t, int32(2), atomic.LoadInt32(&reportCnt), "each branch
reports phase-1 done to the TC")
+}
+
+// ErrSkip in-branch fallback under XA autoCommit.
+//
+// go-sql-driver returns driver.ErrSkip from Exec/Query whenever a statement
carries
+// bind arguments and the DSN does NOT set interpolateParams=true (the
default). See
+// go-sql-driver/[email protected] connection.go: `if len(args) != 0 { if
!cfg.InterpolateParams
+// { return nil, driver.ErrSkip } }`. database/sql normally answers ErrSkip by
retrying the
+// statement through the Prepare+Exec path.
+//
+// Under XA autoCommit + a global transaction, createNewTxOnExecIfNeed opens
the XA branch
+// (XA START) BEFORE running the statement, so it cannot hand the retry back
to database/sql
+// (that retry would run on another connection, outside the branch). Instead
XAConn runs the
+// Prepare+Exec fallback ITSELF on the same physical connection - which still
holds XA START
+// open - so a perfectly ordinary parameterized statement (`UPDATE ... WHERE
id = ?` with the
+// default MySQL DSN) completes inside the branch and the branch commits
normally.
+//
+// The mock models the driver faithfully: the direct ExecContext answers
ErrSkip for the
+// business UPDATE, while PrepareContext + the prepared stmt's ExecContext
succeed.
+func TestXAConn_AutoCommit_ParameterizedStmtErrSkipFallsBackInBranch(t
*testing.T) {
+ ctrl := gomock.NewController(t)
+ defer ctrl.Finish()
+ CleanTxHooks()
+ defer func() {
+ simulateExecContextError = nil
+ CleanTxHooks()
+ }()
+
+ prevTimeout := xaConnTimeout
+ xaConnTimeout = time.Minute
+ defer func() { xaConnTimeout = prevTimeout }()
+
+ xaConn, mockMgr := newMockXAConn(t, ctrl, 123)
+ // The branch prepares and reports phase-1 success once the in-branch
fallback succeeds.
+ var reportCnt int32
+ mockMgr.EXPECT().BranchReport(gomock.Any(),
gomock.Any()).AnyTimes().DoAndReturn(
+ func(_ context.Context, _ interface{}) error {
+ atomic.AddInt32(&reportCnt, 1)
+ return nil
+ })
+
+ // Model go-sql-driver's default behavior: a parameterized business
statement answers
+ // ErrSkip on the direct Execer path; the XA control statements (XA
START/END/PREPARE)
+ // succeed. PrepareContext + prepared ExecContext (wired in
baseMockConn) succeed.
+ simulateExecContextError = func(query string) error {
+ if strings.HasPrefix(strings.ToUpper(strings.TrimSpace(query)),
"UPDATE") {
+ return driver.ErrSkip
+ }
+ return nil
+ }
+
+ ctx := tm.InitSeataContext(context.Background())
+ tm.SetXID(ctx, uuid.NewString())
+
+ _, err := xaConn.ExecContext(ctx, "UPDATE user SET age = age + 1 WHERE
id = ?",
+ []driver.NamedValue{{Ordinal: 1, Value: int64(1)}})
+
+ // The fix: the parameterized UPDATE completes inside the branch via
the in-branch
+ // Prepare+Exec fallback, and the branch commits (phase-1 reported to
the TC).
+ assert.NoError(t, err, "parameterized UPDATE should complete via the
in-branch Prepare+Exec fallback")
+ assert.Equal(t, int32(1), atomic.LoadInt32(&reportCnt),
+ "the branch prepares and reports phase-1 success after the
in-branch fallback")
+}
+
+// The real #904 scenario is a PARAMETERIZED `SELECT ... FOR UPDATE WHERE id =
?`
+// (the samples all bind parameters). Under the default MySQL DSN the direct
Queryer
+// answers driver.ErrSkip for it, so this exercises the query-path in-branch
fallback
+// (queryPreparedInBranch) AND the #904 busy-buffer guard together: the
fallback rows
+// must still be wrapped in RowsCommitOnClose so the branch commit (XA END + XA
+// PREPARE) is deferred until the caller drains/closes the rows, never issued
on top
+// of the still-open result set. Closing the rows also closes the prepared stmt
+// (rowsWithStmt) and then runs XA END + XA PREPARE + the phase-1 report
exactly once.
+func
TestXAConn_AutoCommit_ParameterizedSelectForUpdateErrSkipDefersBranchCommit(t
*testing.T) {
+ ctrl := gomock.NewController(t)
+ defer ctrl.Finish()
+ CleanTxHooks()
+ defer func() {
+ simulateExecContextError = nil
+ simulateQueryContextError = nil
+ CleanTxHooks()
+ }()
+
+ prevTimeout := xaConnTimeout
+ xaConnTimeout = time.Minute
+ defer func() { xaConnTimeout = prevTimeout }()
+
+ xaConn, mockMgr := newMockXAConn(t, ctrl, 123)
+
+ var reportCnt int32
+ mockMgr.EXPECT().BranchReport(gomock.Any(),
gomock.Any()).AnyTimes().DoAndReturn(
+ func(_ context.Context, param rm.BranchReportParam) error {
+ assert.EqualValues(t, branch.BranchStatusPhaseoneDone,
param.Status)
+ atomic.AddInt32(&reportCnt, 1)
+ return nil
+ })
+
+ // The XA control statements run on the direct Execer path and succeed;
count when
+ // the deferred branch commit fires.
+ var endCnt, prepareCnt int32
+ simulateExecContextError = func(query string) error {
+ upper := strings.ToUpper(strings.TrimSpace(query))
+ switch {
+ case strings.HasPrefix(upper, "XA END"):
+ atomic.AddInt32(&endCnt, 1)
+ case strings.HasPrefix(upper, "XA PREPARE"):
+ atomic.AddInt32(&prepareCnt, 1)
+ }
+ return nil
+ }
+ // Model the default go-sql-driver DSN: the direct Queryer answers
ErrSkip for the
+ // parameterized business SELECT, forcing the in-branch prepared-query
fallback.
+ simulateQueryContextError = func(query string) error {
+ if strings.HasPrefix(strings.ToUpper(strings.TrimSpace(query)),
"SELECT") {
+ return driver.ErrSkip
+ }
+ return nil
+ }
+
+ ctx := tm.InitSeataContext(context.Background())
+ tm.SetXID(ctx, uuid.NewString())
+
+ rows, err := xaConn.QueryContext(ctx, "SELECT * FROM user WHERE id = ?
FOR UPDATE",
+ []driver.NamedValue{{Ordinal: 1, Value: int64(1)}})
+ assert.NoError(t, err, "parameterized SELECT ... FOR UPDATE must
complete via the in-branch prepared-query fallback")
+
+ // Even though we fell back to queryPreparedInBranch, the branch commit
must still be
+ // deferred while the rows are open - issuing XA END / XA PREPARE now
is the #904 bug.
+ _, ok := rows.(*RowsCommitOnClose)
+ assert.True(t, ok, "the in-branch query fallback must still wrap rows
in RowsCommitOnClose to defer the branch commit")
+ assert.Equal(t, int32(0), atomic.LoadInt32(&endCnt), "XA END must be
deferred until the fallback rows close")
+ assert.Equal(t, int32(0), atomic.LoadInt32(&prepareCnt), "XA PREPARE
must be deferred until the fallback rows close")
+ assert.Equal(t, int32(0), atomic.LoadInt32(&reportCnt), "phase-1 report
must be deferred until the fallback rows close")
+
+ // Closing the rows closes both the driver rows and the prepared stmt
(rowsWithStmt),
+ // then runs the deferred XA END + XA PREPARE + phase-1 report exactly
once.
+ assert.NoError(t, rows.Close())
+ assert.Equal(t, int32(1), atomic.LoadInt32(&endCnt), "XA END runs once
the fallback rows close")
+ assert.Equal(t, int32(1), atomic.LoadInt32(&prepareCnt), "XA PREPARE
runs once the fallback rows close")
+ assert.Equal(t, int32(1), atomic.LoadInt32(&reportCnt), "the branch
reports phase-1 done once the fallback rows close")
+}
+
+// When the in-branch Prepare+Exec fallback itself fails with a real
(non-ErrSkip)
+// error, the branch must not leak: createNewTxOnExecIfNeed rolls it back and
reports
+// phase-1 FAILED to the TC, and surfaces the concrete error (never
driver.ErrSkip) to
+// the caller. This guards the post-fallback error path added with the fix.
+func TestXAConn_AutoCommit_InBranchFallbackErrorRollsBackBranch(t *testing.T) {
+ ctrl := gomock.NewController(t)
+ defer ctrl.Finish()
+ CleanTxHooks()
+ defer func() {
+ simulateExecContextError = nil
+ simulatePreparedExecError = nil
+ CleanTxHooks()
+ }()
+
+ prevTimeout := xaConnTimeout
+ xaConnTimeout = time.Minute
+ defer func() { xaConnTimeout = prevTimeout }()
+
+ xaConn, mockMgr := newMockXAConn(t, ctrl, 123)
+
+ var failedReportCnt int32
+ mockMgr.EXPECT().BranchReport(gomock.Any(),
gomock.Any()).AnyTimes().DoAndReturn(
+ func(_ context.Context, param rm.BranchReportParam) error {
+ if param.Status == branch.BranchStatusPhaseoneFailed {
+ atomic.AddInt32(&failedReportCnt, 1)
+ }
+ return nil
+ })
+
+ // Direct Execer answers ErrSkip for the business UPDATE (default DSN
behavior); the
+ // XA control statements succeed.
+ simulateExecContextError = func(query string) error {
+ if strings.HasPrefix(strings.ToUpper(strings.TrimSpace(query)),
"UPDATE") {
+ return driver.ErrSkip
+ }
+ return nil
+ }
+ // The in-branch prepared exec then fails with a real error (e.g. a
constraint
+ // violation) - this is NOT ErrSkip, so it must abort and roll back the
branch.
+ prepErr := errors.New("Error 1062: Duplicate entry for key 'PRIMARY'")
+ simulatePreparedExecError = func(query string) error {
+ if strings.HasPrefix(strings.ToUpper(strings.TrimSpace(query)),
"UPDATE") {
+ return prepErr
+ }
+ return nil
+ }
+
+ ctx := tm.InitSeataContext(context.Background())
+ tm.SetXID(ctx, uuid.NewString())
+
+ _, err := xaConn.ExecContext(ctx, "UPDATE user SET age = age + 1 WHERE
id = ?",
+ []driver.NamedValue{{Ordinal: 1, Value: int64(1)}})
+
+ assert.Error(t, err, "a real error from the in-branch fallback must
surface")
+ assert.False(t, errors.Is(err, driver.ErrSkip), "the caller must never
see raw driver.ErrSkip - the fix converts it into a concrete result or error")
+ assert.ErrorIs(t, err, prepErr, "the concrete fallback error must be
surfaced to the caller")
+ assert.Equal(t, int32(1), atomic.LoadInt32(&failedReportCnt),
+ "the failed branch must report phase-1 FAILED to the TC so it
does not leak")
+ assert.False(t, xaConn.xaActive, "the rolled-back branch must leave no
active branch on the session")
+}
+
func TestXAConn_BeginTx_DoesNotStartPhysicalTx(t *testing.T) {
ctrl := gomock.NewController(t)
defer ctrl.Finish()
diff --git a/pkg/datasource/sql/datasource/mysql/trigger_test.go
b/pkg/datasource/sql/datasource/mysql/trigger_test.go
index 49737522..ab9037c6 100644
--- a/pkg/datasource/sql/datasource/mysql/trigger_test.go
+++ b/pkg/datasource/sql/datasource/mysql/trigger_test.go
@@ -24,7 +24,6 @@ import (
"testing"
"github.com/DATA-DOG/go-sqlmock"
- "github.com/agiledragon/gomonkey/v2"
"github.com/golang/mock/gomock"
"github.com/stretchr/testify/assert"
@@ -62,20 +61,23 @@ func initMockColumnMeta() []types.ColumnMeta {
}
}
-func initGetIndexesStub(m *mysqlTrigger, indexMeta []types.IndexMeta)
*gomonkey.Patches {
- getIndexesStub := gomonkey.ApplyPrivateMethod(m, "getIndexes",
- func(_ *mysqlTrigger, ctx context.Context, dbName string,
tableName string, conn *sql.Conn) ([]types.IndexMeta, error) {
- return indexMeta, nil
- })
- return getIndexesStub
+// initGetIndexesStub injects a getIndexes implementation via the
mysqlTrigger's
+// built-in getIndexesFn seam. This replaces gomonkey.ApplyPrivateMethod, which
+// rewrites the method's machine code and silently no-ops once the compiler
inlines
+// getIndexes (default builds without -race/-gcflags=all=-l), making the
LoadOne/
+// LoadAll tests order-dependent.
+func initGetIndexesStub(m *mysqlTrigger, indexMeta []types.IndexMeta) {
+ m.getIndexesFn = func(ctx context.Context, dbName string, tableName
string, conn *sql.Conn) ([]types.IndexMeta, error) {
+ return indexMeta, nil
+ }
}
-func initGetColumnMetasStub(m *mysqlTrigger, columnMeta []types.ColumnMeta)
*gomonkey.Patches {
- getColumnMetasStub := gomonkey.ApplyPrivateMethod(m, "getColumnMetas",
- func(_ *mysqlTrigger, ctx context.Context, dbName string, table
string, conn *sql.Conn) ([]types.ColumnMeta, error) {
- return columnMeta, nil
- })
- return getColumnMetasStub
+// initGetColumnMetasStub injects a getColumnMetas implementation via the
+// mysqlTrigger's built-in getColumnMetasFn seam (see initGetIndexesStub).
+func initGetColumnMetasStub(m *mysqlTrigger, columnMeta []types.ColumnMeta) {
+ m.getColumnMetasFn = func(ctx context.Context, dbName string, table
string, conn *sql.Conn) ([]types.ColumnMeta, error) {
+ return columnMeta, nil
+ }
}
func Test_mysqlTrigger_LoadOne(t *testing.T) {
@@ -105,11 +107,8 @@ func Test_mysqlTrigger_LoadOne(t *testing.T) {
t.Run(tt.name, func(t *testing.T) {
m := &mysqlTrigger{}
- getColumnMetasStub := initGetColumnMetasStub(m,
tt.columnMeta)
- defer getColumnMetasStub.Reset()
-
- getIndexesStub := initGetIndexesStub(m, tt.indexMeta)
- defer getIndexesStub.Reset()
+ initGetColumnMetasStub(m, tt.columnMeta)
+ initGetIndexesStub(m, tt.indexMeta)
got, err := m.LoadOne(tt.args.ctx, tt.args.dbName,
tt.args.tableName, tt.args.conn)
if err != nil {
@@ -175,11 +174,8 @@ func Test_mysqlTrigger_LoadAll(t *testing.T) {
t.Run(tt.name, func(t *testing.T) {
m := &mysqlTrigger{}
- getColumnMetasStub := initGetColumnMetasStub(m,
tt.columnMeta)
- defer getColumnMetasStub.Reset()
-
- getIndexesStub := initGetIndexesStub(m, tt.indexMeta)
- defer getIndexesStub.Reset()
+ initGetColumnMetasStub(m, tt.columnMeta)
+ initGetIndexesStub(m, tt.indexMeta)
got, err := m.LoadAll(tt.args.ctx, tt.args.dbName,
tt.args.conn, tt.args.tables...)
if err != nil {
@@ -234,25 +230,19 @@ func Test_mysqlTrigger_LoadOne_ErrorCases(t *testing.T) {
m := &mysqlTrigger{}
if tt.columnMetaErr != nil {
- getColumnMetasStub :=
gomonkey.ApplyPrivateMethod(m, "getColumnMetas",
- func(_ *mysqlTrigger, ctx
context.Context, dbName string, table string, conn *sql.Conn)
([]types.ColumnMeta, error) {
- return nil, tt.columnMetaErr
- })
- defer getColumnMetasStub.Reset()
+ m.getColumnMetasFn = func(ctx context.Context,
dbName string, table string, conn *sql.Conn) ([]types.ColumnMeta, error) {
+ return nil, tt.columnMetaErr
+ }
} else {
- getColumnMetasStub := initGetColumnMetasStub(m,
tt.columnMeta)
- defer getColumnMetasStub.Reset()
+ initGetColumnMetasStub(m, tt.columnMeta)
}
if tt.indexMetaErr != nil {
- getIndexesStub :=
gomonkey.ApplyPrivateMethod(m, "getIndexes",
- func(_ *mysqlTrigger, ctx
context.Context, dbName string, tableName string, conn *sql.Conn)
([]types.IndexMeta, error) {
- return nil, tt.indexMetaErr
- })
- defer getIndexesStub.Reset()
+ m.getIndexesFn = func(ctx context.Context,
dbName string, tableName string, conn *sql.Conn) ([]types.IndexMeta, error) {
+ return nil, tt.indexMetaErr
+ }
} else {
- getIndexesStub := initGetIndexesStub(m,
tt.indexMeta)
- defer getIndexesStub.Reset()
+ initGetIndexesStub(m, tt.indexMeta)
}
_, err := m.LoadOne(context.Background(), "testdb",
"testtable", nil)
@@ -300,11 +290,8 @@ func Test_mysqlTrigger_LoadOne_ComplexIndexes(t
*testing.T) {
},
}
- getColumnMetasStub := initGetColumnMetasStub(m, columnMeta)
- defer getColumnMetasStub.Reset()
-
- getIndexesStub := initGetIndexesStub(m, indexMeta)
- defer getIndexesStub.Reset()
+ initGetColumnMetasStub(m, columnMeta)
+ initGetIndexesStub(m, indexMeta)
tableMeta, err := m.LoadOne(context.Background(), "testdb",
"testtable", nil)
@@ -604,18 +591,15 @@ func Test_mysqlTrigger_LoadAll_ErrorHandling(t
*testing.T) {
indexMeta := initMockIndexMeta()
callCount := 0
- getColumnMetasStub := gomonkey.ApplyPrivateMethod(m, "getColumnMetas",
- func(_ *mysqlTrigger, ctx context.Context, dbName string, table
string, conn *sql.Conn) ([]types.ColumnMeta, error) {
- callCount++
- if callCount == 2 {
- return nil, errors.New("column error")
- }
- return columnMeta, nil
- })
- defer getColumnMetasStub.Reset()
+ m.getColumnMetasFn = func(ctx context.Context, dbName string, table
string, conn *sql.Conn) ([]types.ColumnMeta, error) {
+ callCount++
+ if callCount == 2 {
+ return nil, errors.New("column error")
+ }
+ return columnMeta, nil
+ }
- getIndexesStub := initGetIndexesStub(m, indexMeta)
- defer getIndexesStub.Reset()
+ initGetIndexesStub(m, indexMeta)
// LoadAll should continue even if one table fails
result, err := m.LoadAll(context.Background(), "testdb", nil, "table1",
"table2", "table3")
@@ -657,11 +641,8 @@ func
Test_mysqlTrigger_LoadOne_MultipleIndexesOnSameColumn(t *testing.T) {
},
}
- getColumnMetasStub := initGetColumnMetasStub(m, columnMeta)
- defer getColumnMetasStub.Reset()
-
- getIndexesStub := initGetIndexesStub(m, indexMeta)
- defer getIndexesStub.Reset()
+ initGetColumnMetasStub(m, columnMeta)
+ initGetIndexesStub(m, indexMeta)
tableMeta, err := m.LoadOne(context.Background(), "testdb",
"testtable", nil)
diff --git a/pkg/datasource/sql/db.go b/pkg/datasource/sql/db.go
index cc019283..10fdb156 100644
--- a/pkg/datasource/sql/db.go
+++ b/pkg/datasource/sql/db.go
@@ -117,7 +117,7 @@ type DBResource struct {
// for xa
metaCache datasource.TableMetaCache
shouldBeHeld bool
- keeper sync.Map
+ keeper sync.Map // xaBranchID -> *XAConn
}
func (db *DBResource) GetResourceGroupId() string {
diff --git a/pkg/datasource/sql/exec/executor.go
b/pkg/datasource/sql/exec/executor.go
index 5a0b9868..57f7b25c 100644
--- a/pkg/datasource/sql/exec/executor.go
+++ b/pkg/datasource/sql/exec/executor.go
@@ -55,6 +55,14 @@ type (
func BuildExecutor(dbType types.DBType, transactionMode types.TransactionMode,
query string) (SQLExecutor, error) {
parseContext, err := parser.DoParser(query)
if err != nil {
+ // XA mode uses a pass-through executor and never needs the
parsed statement:
+ // two-phase commit is managed by the TC and no undo log is
generated, so SQL
+ // the shared parser cannot understand must still execute as-is
instead of
+ // erroring or falling back to an AT executor. This check must
come before the
+ // PostgreSQL fallback so XA mode is never routed to an AT
executor.
+ if transactionMode == types.XAMode {
+ return newXAExecutor(commonHook), nil
+ }
if dbType == types.DBTypePostgreSQL {
// PostgreSQL local execution must remain pass-through
even when the
// shared parser cannot understand PostgreSQL-only
syntax yet.
@@ -67,9 +75,23 @@ func BuildExecutor(dbType types.DBType, transactionMode
types.TransactionMode, q
hooks = append(hooks, commonHook...)
hooks = append(hooks, hookSolts[parseContext.SQLType]...)
+ // For XA mode, use a plain executor without AT-specific hooks: XA
transactions
+ // don't need undo logs, they rely on the two-phase commit protocol
managed by TC.
+ if transactionMode == types.XAMode {
+ return newXAExecutor(hooks), nil
+ }
+
return newATExecutor(dbType, hooks)
}
+// newXAExecutor builds a pass-through executor for XA mode. It applies the
given
+// hooks but never wraps an AT executor, so no undo log is generated.
+func newXAExecutor(hooks []SQLHook) SQLExecutor {
+ e := &BaseExecutor{}
+ e.Interceptors(hooks)
+ return e
+}
+
func newATExecutor(dbType types.DBType, hooks []SQLHook) (SQLExecutor, error) {
builder, ok := atExecutors[dbType]
if !ok || builder == nil {
@@ -127,7 +149,7 @@ func (e *BaseExecutor) ExecWithValue(ctx context.Context,
execCtx *types.ExecCon
return e.ex.ExecWithValue(ctx, execCtx, f)
}
- nvargs := make([]driver.NamedValue, len(execCtx.Values))
+ nvargs := make([]driver.NamedValue, 0, len(execCtx.Values))
for i, value := range execCtx.Values {
nvargs = append(nvargs, driver.NamedValue{
Value: value,
diff --git a/pkg/datasource/sql/exec/executor_test.go
b/pkg/datasource/sql/exec/executor_test.go
index 5870785e..0ac9846c 100644
--- a/pkg/datasource/sql/exec/executor_test.go
+++ b/pkg/datasource/sql/exec/executor_test.go
@@ -150,6 +150,34 @@ func TestBuildExecutor(t *testing.T) {
query: "INVALID SQL QUERY",
wantErr: true,
},
+ {
+ name: "XA mode INSERT statement",
+ dbType: types.DBTypeMySQL,
+ transactionMode: types.XAMode,
+ query: "INSERT INTO users (name, age) VALUES
('Alice', 30)",
+ wantErr: false,
+ },
+ {
+ name: "XA mode UPDATE statement",
+ dbType: types.DBTypeMySQL,
+ transactionMode: types.XAMode,
+ query: "UPDATE users SET age = 31 WHERE name
= 'Alice'",
+ wantErr: false,
+ },
+ {
+ name: "XA mode DELETE statement",
+ dbType: types.DBTypeMySQL,
+ transactionMode: types.XAMode,
+ query: "DELETE FROM users WHERE name =
'Alice'",
+ wantErr: false,
+ },
+ {
+ name: "XA mode SELECT statement",
+ dbType: types.DBTypeMySQL,
+ transactionMode: types.XAMode,
+ query: "SELECT * FROM users WHERE name =
'Alice'",
+ wantErr: false,
+ },
}
for _, tt := range tests {
@@ -165,8 +193,14 @@ func TestBuildExecutor(t *testing.T) {
assert.NoError(t, err, "should not return error
for valid query")
assert.NotNil(t, executor, "executor should not
be nil")
// Verify that the mock executor received the
interceptors
- assert.True(t, mockExecutor.interceptorsCalled,
"Interceptors should be called")
- assert.NotEmpty(t, mockExecutor.hooks, "hooks
should be set")
+ if tt.transactionMode == types.XAMode {
+ assert.IsType(t, &BaseExecutor{},
executor, "XA mode should return BaseExecutor")
+ baseExec := executor.(*BaseExecutor)
+ assert.NotNil(t, baseExec.hooks,
"BaseExecutor should have hooks set")
+ } else {
+ assert.True(t,
mockExecutor.interceptorsCalled, "Interceptors should be called")
+ assert.NotEmpty(t, mockExecutor.hooks,
"hooks should be set")
+ }
}
})
}
@@ -324,23 +358,44 @@ func TestBaseExecutor_ExecWithValue(t *testing.T) {
tests := []struct {
name string
setupHooks []SQLHook
+ innerExecutor SQLExecutor
callback CallbackWithNamedValue
execCtx *types.ExecContext
wantErr bool
wantBeforeCount int
wantAfterCount int
+ // wantArgs is the exact []driver.NamedValue the callback must
receive on the
+ // e.ex == nil pass-through path (values converted from
execCtx.Values). Length
+ // and every element are asserted, so an off-by-N conversion
bug cannot pass.
+ wantArgs []driver.NamedValue
}{
{
- name: "execute with values - converts to NamedValues",
+ name: "e.ex == nil - values are converted to
NamedValues without extra empty slots",
callback: func(ctx context.Context, query string, args
[]driver.NamedValue) (types.ExecResult, error) {
- // Verify that values were converted to
NamedValues
- assert.NotEmpty(t, args, "args should not be
empty")
return newMockExecResult(1, 1), nil
},
execCtx: &types.ExecContext{
Query: "UPDATE users SET age = ? WHERE name =
?",
Values: []driver.Value{31, "Alice"},
},
+ wantArgs: []driver.NamedValue{
+ {Ordinal: 0, Value: 31},
+ {Ordinal: 1, Value: "Alice"},
+ },
+ wantErr: false,
+ wantBeforeCount: 0,
+ wantAfterCount: 0,
+ },
+ {
+ name: "e.ex == nil - empty values produce empty args",
+ callback: func(ctx context.Context, query string, args
[]driver.NamedValue) (types.ExecResult, error) {
+ return newMockExecResult(1, 1), nil
+ },
+ execCtx: &types.ExecContext{
+ Query: "DELETE FROM users",
+ Values: []driver.Value{},
+ },
+ wantArgs: []driver.NamedValue{},
wantErr: false,
wantBeforeCount: 0,
wantAfterCount: 0,
@@ -357,6 +412,10 @@ func TestBaseExecutor_ExecWithValue(t *testing.T) {
Query: "UPDATE users SET age = ? WHERE name =
?",
Values: []driver.Value{31, "Alice"},
},
+ wantArgs: []driver.NamedValue{
+ {Ordinal: 0, Value: 31},
+ {Ordinal: 1, Value: "Alice"},
+ },
wantErr: false,
wantBeforeCount: 1,
wantAfterCount: 1,
@@ -373,19 +432,58 @@ func TestBaseExecutor_ExecWithValue(t *testing.T) {
Query: "UPDATE users SET age = ? WHERE name =
?",
Values: []driver.Value{31, "Alice"},
},
+ wantArgs: []driver.NamedValue{
+ {Ordinal: 0, Value: 31},
+ {Ordinal: 1, Value: "Alice"},
+ },
wantErr: true,
wantBeforeCount: 1,
wantAfterCount: 1,
},
+ {
+ name: "e.ex != nil - delegates to the inner executor",
+ setupHooks: []SQLHook{
+ &mockSQLHook{sqlType: types.SQLTypeUpdate},
+ },
+ innerExecutor: &mockSQLExecutor{
+ execWithValueFunc: func(ctx context.Context,
execCtx *types.ExecContext, f CallbackWithNamedValue) (types.ExecResult, error)
{
+ return newMockExecResult(2, 2), nil
+ },
+ },
+ callback: func(ctx context.Context, query string, args
[]driver.NamedValue) (types.ExecResult, error) {
+ // The inner executor short-circuits the
pass-through conversion, so the
+ // base executor must never invoke this
callback itself.
+ panic("callback should not be called when inner
executor is set")
+ },
+ execCtx: &types.ExecContext{
+ Query: "UPDATE users SET age = ? WHERE name =
?",
+ Values: []driver.Value{31, "Alice"},
+ },
+ wantErr: false,
+ wantBeforeCount: 1,
+ wantAfterCount: 1,
+ },
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
+ // Wrap the callback to assert the exact converted args
on the pass-through path.
+ callback := tt.callback
+ if tt.wantArgs != nil {
+ inner := tt.callback
+ callback = func(ctx context.Context, query
string, args []driver.NamedValue) (types.ExecResult, error) {
+ assert.Equal(t, len(tt.wantArgs),
len(args), "converted args length should match values length")
+ assert.Equal(t, tt.wantArgs, args,
"converted args should match expected NamedValues")
+ return inner(ctx, query, args)
+ }
+ }
+
executor := &BaseExecutor{
hooks: tt.setupHooks,
+ ex: tt.innerExecutor,
}
- result, err :=
executor.ExecWithValue(context.Background(), tt.execCtx, tt.callback)
+ result, err :=
executor.ExecWithValue(context.Background(), tt.execCtx, callback)
if tt.wantErr {
assert.Error(t, err, "should return error")
diff --git a/pkg/datasource/sql/tx.go b/pkg/datasource/sql/tx.go
index c3c74c26..46351ab6 100644
--- a/pkg/datasource/sql/tx.go
+++ b/pkg/datasource/sql/tx.go
@@ -149,6 +149,17 @@ func (tx *Tx) Rollback() error {
}
}
+ // In XA mode, target might be nil (set with withOriginTx(nil))
+ // Only allow nil target when explicitly in XA mode; otherwise,
+ // treat it as an error to avoid masking unexpected
driver/initialization bugs
+ if tx.target == nil {
+ if tx.tranCtx != nil && tx.tranCtx.TransactionMode ==
types.XAMode {
+ // XA transactions are managed separately
+ return nil
+ }
+ return fmt.Errorf("sql.Tx Rollback: underlying transaction is
nil in non-XA mode")
+ }
+
return tx.target.Rollback()
}
diff --git a/pkg/datasource/sql/types/types.go
b/pkg/datasource/sql/types/types.go
index 74598ce1..0893139a 100644
--- a/pkg/datasource/sql/types/types.go
+++ b/pkg/datasource/sql/types/types.go
@@ -147,7 +147,7 @@ type TransactionContext struct {
BranchID uint64
// XID global transaction id
XID string
- // GlobalLockRequire
+ // GlobalLockRequire indicates whether global lock is required (used in
AT mode)
GlobalLockRequire bool
// RoundImages when run in AT mode, record before and after Row image
RoundImages *RoundRecordImage
diff --git a/pkg/datasource/sql/xa/mysql_xa_connection.go
b/pkg/datasource/sql/xa/mysql_xa_connection.go
index 601a27dd..181874eb 100644
--- a/pkg/datasource/sql/xa/mysql_xa_connection.go
+++ b/pkg/datasource/sql/xa/mysql_xa_connection.go
@@ -50,9 +50,20 @@ func (f *mysqlXAResourceFactory) CreateErrorClassifier()
XAErrorClassifier {
// MysqlXAErrorClassifier classifies MySQL-specific XA errors.
type MysqlXAErrorClassifier struct{}
-// IsAlreadyEnded 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
+// IsAlreadyEnded reports whether the XAER_RMFAIL error means the XA branch has
+// already left the ACTIVE state, so a subsequent XA END(TMFAIL) is a no-op
and the
+// caller should proceed straight to XA ROLLBACK.
+//
+// Two states qualify, both raised as Error 1399 (XAE07) XAER_RMFAIL:
+// - IDLE: "...cannot be executed when global transaction is in the IDLE
state"
+// (branch already ended once, e.g. Commit ran XA END then failed later).
+// - PREPARED: "...cannot be executed when global transaction is in the
PREPARED state"
+// (branch already ended AND prepared, e.g. Commit did XA END + XA PREPARE
at the DB
+// but the phase-1 report to the TC failed, and Rollback now needs to
release locks).
+//
+// Treating the PREPARED case as "already ended" is required so that Rollback
does not
+// bail out before XA ROLLBACK - otherwise a prepared branch would keep
holding locks.
+// XA ROLLBACK is a legal transition out of the PREPARED state, so it still
releases them.
func (c *MysqlXAErrorClassifier) IsAlreadyEnded(err error) bool {
if err == nil {
return false
@@ -60,7 +71,9 @@ func (c *MysqlXAErrorClassifier) IsAlreadyEnded(err error)
bool {
var mysqlErr *mysql.MySQLError
if errors.As(err, &mysqlErr) {
if mysqlErr.Number == types.ErrCodeXAER_RMFAIL_IDLE {
- return strings.Contains(mysqlErr.Message, "IDLE state")
|| strings.Contains(mysqlErr.Message, "already ended")
+ return strings.Contains(mysqlErr.Message, "IDLE state")
||
+ strings.Contains(mysqlErr.Message, "PREPARED
state") ||
+ strings.Contains(mysqlErr.Message, "already
ended")
}
}
return false
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
