You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@seata.apache.org by "YuZhangLarry (via GitHub)" <gi...@apache.org> on 2026/07/08 11:35:14 UTC
[PR] feat: support XA branch reuse in autoCommit mode [incubator-seata-go]
YuZhangLarry opened a new pull request, #1137:
URL: https://github.com/apache/incubator-seata-go/pull/1137
## What this PR does
Support XA branch reuse when `autoCommit=true`, so that multiple SQL
statements can reuse the same XA branch instead of opening a new branch
for every statement.
Fixes #904 (busy buffer error when a `SELECT ... FOR UPDATE` is followed
by an `UPDATE` in XA mode with `autoCommit=true`).
## Changes
- 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`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org
Re: [PR] feat: support XA branch reuse in autoCommit mode [incubator-seata-go]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #1137:
URL: https://github.com/apache/incubator-seata-go/pull/1137#issuecomment-4914432717
## [Codecov](https://app.codecov.io/gh/apache/incubator-seata-go/pull/1137?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `25.14620%` with `128 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 56.64%. Comparing base ([`c34171e`](https://app.codecov.io/gh/apache/incubator-seata-go/commit/c34171eabcf68fd5a34a7f8cd21e58f75f3ce6fc?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`e4d1d44`](https://app.codecov.io/gh/apache/incubator-seata-go/commit/e4d1d44d9ed1950dcdee674e17c9c0731f4ac894?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
| [Files with missing lines](https://app.codecov.io/gh/apache/incubator-seata-go/pull/1137?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [pkg/datasource/sql/xa\_registry.go](https://app.codecov.io/gh/apache/incubator-seata-go/pull/1137?src=pr&el=tree&filepath=pkg%2Fdatasource%2Fsql%2Fxa_registry.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGtnL2RhdGFzb3VyY2Uvc3FsL3hhX3JlZ2lzdHJ5Lmdv) | 0.00% | [76 Missing :warning: ](https://app.codecov.io/gh/apache/incubator-seata-go/pull/1137?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [pkg/datasource/sql/conn\_xa.go](https://app.codecov.io/gh/apache/incubator-seata-go/pull/1137?src=pr&el=tree&filepath=pkg%2Fdatasource%2Fsql%2Fconn_xa.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGtnL2RhdGFzb3VyY2Uvc3FsL2Nvbm5feGEuZ28=) | 36.66% | [28 Missing and 10 partials :warning: ](https://app.codecov.io/gh/apache/incubator-seata-go/pull/1137?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [pkg/datasource/sql/tx\_xa.go](https://app.codecov.io/gh/apache/incubator-seata-go/pull/1137?src=pr&el=tree&filepath=pkg%2Fdatasource%2Fsql%2Ftx_xa.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGtnL2RhdGFzb3VyY2Uvc3FsL3R4X3hhLmdv) | 62.50% | [4 Missing and 2 partials :warning: ](https://app.codecov.io/gh/apache/incubator-seata-go/pull/1137?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [pkg/datasource/sql/tx.go](https://app.codecov.io/gh/apache/incubator-seata-go/pull/1137?src=pr&el=tree&filepath=pkg%2Fdatasource%2Fsql%2Ftx.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGtnL2RhdGFzb3VyY2Uvc3FsL3R4Lmdv) | 0.00% | [3 Missing and 1 partial :warning: ](https://app.codecov.io/gh/apache/incubator-seata-go/pull/1137?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [pkg/datasource/sql/db.go](https://app.codecov.io/gh/apache/incubator-seata-go/pull/1137?src=pr&el=tree&filepath=pkg%2Fdatasource%2Fsql%2Fdb.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGtnL2RhdGFzb3VyY2Uvc3FsL2RiLmdv) | 77.77% | [2 Missing :warning: ](https://app.codecov.io/gh/apache/incubator-seata-go/pull/1137?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [pkg/datasource/sql/hook/undo\_log\_hook.go](https://app.codecov.io/gh/apache/incubator-seata-go/pull/1137?src=pr&el=tree&filepath=pkg%2Fdatasource%2Fsql%2Fhook%2Fundo_log_hook.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGtnL2RhdGFzb3VyY2Uvc3FsL2hvb2svdW5kb19sb2dfaG9vay5nbw==) | 0.00% | [1 Missing and 1 partial :warning: ](https://app.codecov.io/gh/apache/incubator-seata-go/pull/1137?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #1137 +/- ##
==========================================
- Coverage 57.07% 56.64% -0.44%
==========================================
Files 299 300 +1
Lines 20495 20628 +133
==========================================
- Hits 11697 11684 -13
- Misses 7793 7949 +156
+ Partials 1005 995 -10
```
</details>
[:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/incubator-seata-go/pull/1137?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org
Re: [PR] feat: support XA branch reuse in autoCommit mode [incubator-seata-go]
Posted by "thunguo (via GitHub)" <gi...@apache.org>.
thunguo commented on code in PR #1137:
URL: https://github.com/apache/incubator-seata-go/pull/1137#discussion_r3620230167
##########
pkg/datasource/sql/conn_xa.go:
##########
@@ -237,22 +244,51 @@ func (c *XAConn) createNewTxOnExecIfNeed(ctx context.Context, f func() (types.Ex
// execute SQL
ret, err := f()
if err != nil {
+ // driver.ErrSkip is not a real failure: it asks database/sql to retry this
+ // statement through the Prepare+Exec fallback path.
+ if errors.Is(err, driver.ErrSkip) {
+ // If we already opened an XA branch for this statement (autoCommit mode),
+ // the fallback retry would run OUTSIDE the branch - autoCommit is now false
+ // and txCtx has been reset - leaving this branch registered-but-never-
+ // prepared (a leak) and the retried write escaping the global transaction.
+ // Roll the branch back and surface a real (non-ErrSkip) error so the caller
+ // fails fast instead of silently corrupting the global transaction.
+ if tx != nil {
+ if rollbackErr := tx.Rollback(); rollbackErr != nil {
+ log.Errorf("failed to rollback xa branch of :%s after ErrSkip, err:%v", c.txCtx.XID, rollbackErr)
+ }
+ xaRollbacked = true
+ return nil, fmt.Errorf("xa branch %s cannot fall back to non-XA execution after XA START: %v", c.txCtx.XID, err)
+ }
+ // No branch opened yet - safe to let database/sql use its fallback path.
+ return nil, err
+ }
+ // 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.
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 {
Review Comment:
验证一下这里的 Rollback 会不会出现这种情况：
tx.Commit() 进行DB层的XA END + PREPARE，然后向TC上报phase1失败了，tx.Commit返回失败之后这里的Rollback是不是会对一个已经 PREPARE 的 branch 重复调用 XA END，对照下后面 `end()` 方法里
```go
if c.xaErrorClassifier.IsAlreadyEnded(err) {
```
这部分，确认这个场景行为正确，不然对应分支会一直持有锁没法释放
##########
pkg/datasource/sql/exec/executor.go:
##########
Review Comment:
这里可以顺带修复一下，之前不会走到 `e.ex == nil` 这个路径，现在 `newXAExecutor` 是有可能构造出这种case 的，后面会出现参数数量不匹配的问题，改成下面这样即可。
```go
nvargs := make([]driver.NamedValue, 0, len(execCtx.Values))
```
然后补充一个e.ex == nil 的测试 case 验证下，以及修正 TestBaseExecutor_ExecWithValue 现在只判断非空，长度不对也能pass
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org
Re: [PR] feat: support XA branch reuse in autoCommit mode [incubator-seata-go]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #1137:
URL: https://github.com/apache/incubator-seata-go/pull/1137#discussion_r3543880403
##########
pkg/datasource/sql/xa_registry.go:
##########
@@ -0,0 +1,192 @@
+/*
+ * 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 (
+ "sync"
+ "time"
+
+ "seata.apache.org/seata-go/v2/pkg/util/log"
+)
+
+// xaState represents the state of an XA transaction branch
+type xaState int
+
+const (
+ xaStateIdle xaState = iota // No XA transaction
+ xaStateStarted // XA START executed
+ xaStateEnded // XA END executed
+ xaStatePrepared // XA PREPARE executed
+)
+
+// xaEntry represents an active XA transaction branch
+type xaEntry struct {
+ conn *XAConn
+ xid string
+ branchID string
+ resourceID string
+ state xaState
+ createTime time.Time
+ lastAccessTime time.Time
+ statementCount int
+}
+
+// xaRegistry manages XA connections for global transactions
+// It ensures multiple SQL operations in the same global transaction
+// reuse the same XA branch, preventing "busy buffer" errors
+type xaRegistry struct {
+ mu sync.RWMutex
+ entries map[string]*xaEntry // key: xid
+}
+
+var (
+ globalRegistry *xaRegistry
+ registryOnce sync.Once
+)
+
+// getXARegistry returns the global XA registry (singleton)
+func getXARegistry() *xaRegistry {
+ registryOnce.Do(func() {
+ globalRegistry = &xaRegistry{
+ entries: make(map[string]*xaEntry),
+ }
+ })
+ return globalRegistry
+}
+
+// register registers or retrieves an XA connection for the given xid
+// Returns (isNew, entry, error)
+func (r *xaRegistry) register(xid, branchID, resourceID string, conn *XAConn) (bool, *xaEntry) {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+
+ now := time.Now()
+
+ // Check if entry already exists
+ if entry, ok := r.entries[xid]; ok {
+ entry.lastAccessTime = now
+ entry.statementCount++
+ log.Infof("Reusing existing XA branch for xid: %s, branchID: %s, statementCount: %d, state: %v",
+ xid, entry.branchID, entry.statementCount, entry.state)
+ return false, entry
+ }
+
+ // Create new entry
+ entry := &xaEntry{
+ conn: conn,
+ xid: xid,
+ branchID: branchID,
+ resourceID: resourceID,
+ state: xaStateStarted,
+ createTime: now,
+ lastAccessTime: now,
+ statementCount: 1,
+ }
+
+ r.entries[xid] = entry
+ log.Infof("Registered new XA branch, xid: %s, branchID: %s, resourceID: %s", xid, branchID, resourceID)
+
+ return true, entry
+}
+
+// get retrieves an XA entry by xid
+func (r *xaRegistry) get(xid string) (*xaEntry, bool) {
+ r.mu.RLock()
+ defer r.mu.RUnlock()
+
+ entry, ok := r.entries[xid]
+ if ok {
+ entry.lastAccessTime = time.Now()
+ }
+ return entry, ok
+}
Review Comment:
`xaRegistry.get` acquires an `RLock` but still mutates `entry.lastAccessTime`, which violates the RWMutex contract and can cause data races under concurrent access (reads are not exclusive). Use an exclusive lock for the read+touch, or avoid mutating shared state in `get`.
##########
pkg/datasource/sql/xa_registry.go:
##########
@@ -0,0 +1,192 @@
+/*
+ * 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 (
+ "sync"
+ "time"
+
+ "seata.apache.org/seata-go/v2/pkg/util/log"
+)
+
+// xaState represents the state of an XA transaction branch
+type xaState int
+
+const (
+ xaStateIdle xaState = iota // No XA transaction
+ xaStateStarted // XA START executed
+ xaStateEnded // XA END executed
+ xaStatePrepared // XA PREPARE executed
+)
+
+// xaEntry represents an active XA transaction branch
+type xaEntry struct {
+ conn *XAConn
+ xid string
+ branchID string
+ resourceID string
+ state xaState
+ createTime time.Time
+ lastAccessTime time.Time
+ statementCount int
+}
+
+// xaRegistry manages XA connections for global transactions
+// It ensures multiple SQL operations in the same global transaction
+// reuse the same XA branch, preventing "busy buffer" errors
+type xaRegistry struct {
+ mu sync.RWMutex
+ entries map[string]*xaEntry // key: xid
+}
+
+var (
+ globalRegistry *xaRegistry
+ registryOnce sync.Once
+)
+
+// getXARegistry returns the global XA registry (singleton)
+func getXARegistry() *xaRegistry {
+ registryOnce.Do(func() {
+ globalRegistry = &xaRegistry{
+ entries: make(map[string]*xaEntry),
+ }
+ })
+ return globalRegistry
+}
+
+// register registers or retrieves an XA connection for the given xid
+// Returns (isNew, entry, error)
Review Comment:
The comment for `xaRegistry.register` says it returns `(isNew, entry, error)`, but the function signature returns only `(bool, *xaEntry)`. This is misleading for maintainers.
##########
pkg/datasource/sql/conn_xa.go:
##########
@@ -196,63 +205,96 @@ func (c *XAConn) createOnceTxContext(ctx context.Context) bool {
c.txCtx.ResourceID = c.res.resourceID
c.txCtx.XID = tm.GetXID(ctx)
c.txCtx.TransactionMode = types.XAMode
- c.txCtx.GlobalLockRequire = true
+ c.txCtx.IsAutoCommitXABranch = true
}
return onceTx
}
func (c *XAConn) createNewTxOnExecIfNeed(ctx context.Context, 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
)
+ xid := tm.GetXID(ctx)
+
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 {
+ // For XA transactions, use the connection's rollback which handles XA END + ROLLBACK
+ if !xaRollbacked && c.xaActive {
+ rollbackErr := c.Rollback(ctx)
+ if rollbackErr != nil {
+ log.Errorf("defer rollback xa branch error:%v", rollbackErr)
}
+ xaRollbacked = true
}
}
}()
currentAutoCommit := c.autoCommit
+
+ // For global transactions in autoCommit mode, create/reuse 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 {
- return nil, err
+ // Check if we already have an active XA branch for this transaction
+ heldConn := c.res.GetXABranch(xid)
+ if heldConn != nil && heldConn.xaActive && heldConn.txCtx.XID == xid {
+ if heldConn != c {
+ // Delegate to the connection that holds the XA branch
+ return heldConn.createNewTxOnExecIfNeed(ctx, f)
Review Comment:
In autoCommit branch-reuse, the delegation `return heldConn.createNewTxOnExecIfNeed(ctx, f)` passes a closure `f` that is created in `QueryContext`/`ExecContext` and captures the *original* receiver connection. If `heldConn != c`, `f()` will still execute on `c.Conn`, not on `heldConn`, so the SQL runs on the wrong physical connection while the XA branch lives on `heldConn`. This breaks branch reuse and can reintroduce connection-state errors.
##########
pkg/datasource/sql/conn_xa.go:
##########
@@ -418,10 +487,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()
}
Review Comment:
Phase-2 XA commit/rollback paths (`BranchCommit`/`BranchRollback`) call `XaCommit`/`XaRollbackByBranchId`, but the new `DBResource.xaConnsByXID` map is only cleaned up in `XAConn.Commit`/`XAConn.Rollback` (phase-1). For autoCommit branch reuse this can leave `xid -> *XAConn` entries behind indefinitely, causing an unbounded memory leak. Consider unregistering by global XID during phase-2 as well.
##########
pkg/datasource/sql/tx_xa.go:
##########
@@ -93,6 +106,19 @@ func (tx *XATx) commitOnXA() error {
xid := originTx.tranCtx.XID
branchID := originTx.tranCtx.BranchID
+ // For autoCommit mode (branch reuse), skip XA END/PREPARE but report success
+ if originTx.tranCtx.IsAutoCommitXABranch {
+ log.Infof("xa branch [%d/%s] in autoCommit mode, skipping XA END/PREPARE for branch reuse", 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 err
+ }
+ log.Infof("xa branch [%d/%s] reported phase-1 success to TC (autoCommit mode)", branchID, xid)
+ }
+ return nil
Review Comment:
For `IsAutoCommitXABranch`, `commitOnXA` reports `BranchStatusPhaseoneDone` to TC but explicitly skips `XA END` + `XA PREPARE`. TC will later execute phase-2 commit with `onePhase=false`, which assumes the branch is prepared; reporting success without preparing can put TC/RM out of sync and cause phase-2 failures.
##########
pkg/datasource/sql/tx_xa.go:
##########
@@ -54,6 +54,19 @@ func (tx *XATx) Rollback() error {
xid := originTx.tranCtx.XID
branchID := originTx.tranCtx.BranchID
+ // For autoCommit mode (branch reuse), skip XA END/ROLLBACK but report failure
+ if originTx.tranCtx.IsAutoCommitXABranch {
+ log.Infof("xa branch [%d/%s] in autoCommit mode, skipping XA END/ROLLBACK for branch reuse", branchID, xid)
+ if originTx.tranCtx.IsBranchRegistered() {
+ if err := originTx.report(false); err != nil {
+ log.Errorf("xa branch [%d/%s] failed to report rollback failure to TC: %v", branchID, xid, err)
+ return err
+ }
+ log.Infof("xa branch [%d/%s] reported rollback to TC (autoCommit mode)", branchID, xid)
+ }
+ return nil
+ }
Review Comment:
For `IsAutoCommitXABranch`, `Rollback` reports `BranchStatusPhaseoneFailed` to TC while skipping `XA END(TMFAIL)` + `XA ROLLBACK`. This can leave an in-doubt XA branch on the RM while TC believes phase-1 is complete, leading to inconsistent recovery/cleanup behavior.
##########
pkg/datasource/sql/conn_xa.go:
##########
@@ -132,6 +132,8 @@ func (c *XAConn) BeginTx(ctx context.Context, opts driver.TxOptions) (driver.Tx,
return tx, err
}
+ // Save the original autoCommit state before modifying it
+ wasAutoCommit := c.autoCommit
c.autoCommit = false
Review Comment:
`BeginTx` persists `c.autoCommit = false` but does not restore the prior value on any error path (e.g., `newTx` failure, branch register failure, start failure). If the driver doesn’t reliably call `ResetSession`, this can leave the connection in a non-autoCommit state and affect subsequent operations on the same connection.
##########
pkg/datasource/sql/exec/executor.go:
##########
@@ -67,6 +67,15 @@ 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 use the two-phase commit protocol managed by TC
+ // Note: undo_log_hook.go already handles skipping undo log generation for XA mode
+ if transactionMode == types.XAMode {
+ e := &BaseExecutor{}
+ e.Interceptors(hooks)
+ return e, nil
+ }
Review Comment:
`BuildExecutor` returns `BaseExecutor` for XA mode, but `BaseExecutor` ignores errors returned by hook `Before(...)` calls. This contradicts the `SQLHook` contract in `exec/hook.go` (“Before hook errors will prevent SQL execution”) and changes behavior for any custom hooks in XA mode.
##########
pkg/datasource/sql/conn_xa.go:
##########
@@ -388,6 +448,15 @@ func (c *XAConn) Commit(ctx context.Context) error {
}
c.prepareTime = time.Now()
+
+ // Update registry state to PREPARED and unregister
+ registry := getXARegistry()
+ registry.setState(c.txCtx.XID, xaStatePrepared)
+ registry.unregister(c.txCtx.XID)
+
Review Comment:
`getXARegistry()` introduces a process-wide singleton registry, but the PR description says the XA branch registry is DBResource-scoped. Also, the registry is never populated (`(*xaRegistry).register` is not called anywhere), so the state updates below are effectively no-ops and add dead code paths.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org
Re: [PR] feat: support XA branch reuse in autoCommit mode [incubator-seata-go]
Posted by "YuZhangLarry (via GitHub)" <gi...@apache.org>.
YuZhangLarry commented on PR #1137:
URL: https://github.com/apache/incubator-seata-go/pull/1137#issuecomment-4915964498
@thunguo I have fixed it, please check.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org
Re: [PR] feat: XA branch enrollment for autoCommit statements" [incubator-seata-go]
Posted by "thunguo (via GitHub)" <gi...@apache.org>.
thunguo merged PR #1137:
URL: https://github.com/apache/incubator-seata-go/pull/1137
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org
Re: [PR] feat: support XA branch reuse in autoCommit mode [incubator-seata-go]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #1137:
URL: https://github.com/apache/incubator-seata-go/pull/1137#discussion_r3549919775
##########
pkg/datasource/sql/exec/executor.go:
##########
@@ -67,6 +67,15 @@ 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 use the two-phase commit protocol managed by TC
+ // Note: undo_log_hook.go already handles skipping undo log generation for XA mode
+ if transactionMode == types.XAMode {
+ e := &BaseExecutor{}
+ e.Interceptors(hooks)
+ return e, nil
+ }
Review Comment:
BuildExecutor’s XA fast-path is bypassed when parser.DoParser fails for PostgreSQL: the function returns newATExecutor(dbType, nil) before reaching the XA branch. That means XA mode may still construct an AT executor (and require AT executor registration) for PostgreSQL-specific SQL that the shared parser can’t parse.
##########
pkg/datasource/sql/conn_xa.go:
##########
@@ -204,29 +202,38 @@ func (c *XAConn) createOnceTxContext(ctx context.Context) bool {
func (c *XAConn) createNewTxOnExecIfNeed(ctx context.Context, 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 {
Review Comment:
The new driver.ErrSkip handling returns ErrSkip without rolling back, but XAConn.ExecContext/QueryContext defer-reset txCtx when autoCommit=true. That means the subsequent database/sql fallback (prepare+exec on the same Conn) will likely run with txCtx reset to Local while an XA branch may already be active/registered, causing inconsistent state and potential branch leaks or protocol errors.
##########
pkg/datasource/sql/hook/undo_log_hook.go:
##########
@@ -44,6 +44,12 @@ func (h *undoLogSQLHook) Before(ctx context.Context, execCtx *types.ExecContext)
return nil
}
+ // Skip undo log generation for XA mode - XA transactions don't need undo logs
+ // Undo logs are only needed for AT mode's automatic compensation
+ if execCtx.TxCtx != nil && execCtx.TxCtx.TransactionMode == types.XAMode {
+ return nil
+ }
Review Comment:
The XA-mode skip added here is likely never reached in normal execution because undoLogSQLHook.Type() returns SQLTypeUnknown, and exec.RegisterHook ignores SQLTypeUnknown hooks (so the undo log hook is never registered into hookSolts). As a result, this XA skip (and undo log generation in general) may not run at all.
##########
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")
+ }
Review Comment:
Returning nil when tx.target is nil in XA mode will silently treat rollback as successful even if this Tx was created without an xaConn (e.g., Conn.BeginTx sets withOriginTx(nil) and does not set withXAConn). That can mask incorrect transaction wiring; consider delegating to xaConn when present, and otherwise returning an explicit error in XA mode too.
##########
changes/dev.md:
##########
@@ -26,6 +26,7 @@
### feature：
- [[#123](https://github.com/apache/incubator-seata-go/pull/123)] add two phase and dubbo
+ - support XA branch enrollment for autoCommit statements in global transaction
Review Comment:
This changelog entry claims support for XA branch enrollment/reuse for autoCommit statements, but the code changes in this PR don’t appear to introduce the advertised mechanisms (e.g., IsAutoCommitXABranch / XA branch reuse registry) and conn_xa.go still commits a branch per statement in autoCommit mode.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org
Re: [PR] feat: support XA branch reuse in autoCommit mode [incubator-seata-go]
Posted by "thunguo (via GitHub)" <gi...@apache.org>.
thunguo commented on PR #1137:
URL: https://github.com/apache/incubator-seata-go/pull/1137#issuecomment-4958474856
Check if the comments given by copilot should be resolved.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org
Re: [PR] feat: XA branch enrollment for autoCommit statements" [incubator-seata-go]
Posted by "YuZhangLarry (via GitHub)" <gi...@apache.org>.
YuZhangLarry commented on PR #1137:
URL: https://github.com/apache/incubator-seata-go/pull/1137#issuecomment-5191991594
<img width="1248" height="280" alt="PixPin_2026-08-05_20-55-28" src="https://github.com/user-attachments/assets/d4c74843-82a4-4f7e-8312-93e7c7f18e31" />
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org
Re: [PR] feat: support XA branch reuse in autoCommit mode [incubator-seata-go]
Posted by "Ethan-Xingyue (via GitHub)" <gi...@apache.org>.
Ethan-Xingyue commented on PR #1137:
URL: https://github.com/apache/incubator-seata-go/pull/1137#issuecomment-4914842201
Please fix the CI issue
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org
Re: [PR] feat: support XA branch reuse in autoCommit mode [incubator-seata-go]
Posted by "thunguo (via GitHub)" <gi...@apache.org>.
thunguo commented on code in PR #1137:
URL: https://github.com/apache/incubator-seata-go/pull/1137#discussion_r3543972615
##########
pkg/datasource/sql/conn_xa.go:
##########
@@ -156,27 +161,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")
- }
+ // Create XA branch for both explicit transactions and autoCommit mode (branch reuse)
+ // In autoCommit mode, we register the branch but keep it open for multiple SQL statements
+ 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()
+ // For autoCommit mode (branch reuse), skip XA START here
+ // It will be done when needed for the actual execution
+ if !wasAutoCommit {
if err = c.start(ctx); err != nil {
Review Comment:
这里语义有问题？autoCommit下跳过了 XA_START，但是看起来后续的时机没有补充调用。autoCommit下完全没有拉起 XA Branch。
##########
pkg/datasource/sql/conn_xa.go:
##########
@@ -358,6 +407,10 @@ func (c *XAConn) Rollback(ctx context.Context) error {
c.rollBacked = true
}
c.cleanXABranchContext()
+
+ // Clean up resource holder on rollback
+ c.res.UnregisterXABranch(c.txCtx.XID)
Review Comment:
是不是应该先Unregister
##########
pkg/datasource/sql/xa_registry.go:
##########
@@ -0,0 +1,192 @@
+/*
+ * 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 (
+ "sync"
+ "time"
+
+ "seata.apache.org/seata-go/v2/pkg/util/log"
+)
+
+// xaState represents the state of an XA transaction branch
+type xaState int
+
+const (
+ xaStateIdle xaState = iota // No XA transaction
+ xaStateStarted // XA START executed
+ xaStateEnded // XA END executed
+ xaStatePrepared // XA PREPARE executed
+)
+
+// xaEntry represents an active XA transaction branch
+type xaEntry struct {
+ conn *XAConn
+ xid string
+ branchID string
+ resourceID string
+ state xaState
+ createTime time.Time
+ lastAccessTime time.Time
+ statementCount int
+}
+
+// xaRegistry manages XA connections for global transactions
+// It ensures multiple SQL operations in the same global transaction
+// reuse the same XA branch, preventing "busy buffer" errors
+type xaRegistry struct {
+ mu sync.RWMutex
+ entries map[string]*xaEntry // key: xid
+}
+
+var (
+ globalRegistry *xaRegistry
+ registryOnce sync.Once
+)
+
+// getXARegistry returns the global XA registry (singleton)
+func getXARegistry() *xaRegistry {
+ registryOnce.Do(func() {
+ globalRegistry = &xaRegistry{
+ entries: make(map[string]*xaEntry),
+ }
+ })
+ return globalRegistry
+}
+
+// register registers or retrieves an XA connection for the given xid
+// Returns (isNew, entry, error)
+func (r *xaRegistry) register(xid, branchID, resourceID string, conn *XAConn) (bool, *xaEntry) {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+
+ now := time.Now()
+
+ // Check if entry already exists
+ if entry, ok := r.entries[xid]; ok {
+ entry.lastAccessTime = now
+ entry.statementCount++
+ log.Infof("Reusing existing XA branch for xid: %s, branchID: %s, statementCount: %d, state: %v",
+ xid, entry.branchID, entry.statementCount, entry.state)
+ return false, entry
+ }
+
+ // Create new entry
+ entry := &xaEntry{
+ conn: conn,
+ xid: xid,
+ branchID: branchID,
+ resourceID: resourceID,
+ state: xaStateStarted,
+ createTime: now,
+ lastAccessTime: now,
+ statementCount: 1,
+ }
+
+ r.entries[xid] = entry
+ log.Infof("Registered new XA branch, xid: %s, branchID: %s, resourceID: %s", xid, branchID, resourceID)
+
+ return true, entry
+}
+
+// get retrieves an XA entry by xid
+func (r *xaRegistry) get(xid string) (*xaEntry, bool) {
+ r.mu.RLock()
+ defer r.mu.RUnlock()
+
+ entry, ok := r.entries[xid]
+ if ok {
+ entry.lastAccessTime = time.Now()
+ }
+ return entry, ok
+}
+
+// canReuse checks if an XA connection can be reused for the given xid
+func (r *xaRegistry) canReuse(xid string) bool {
+ entry, ok := r.get(xid)
+ if !ok {
+ return false
+ }
+
+ // Can reuse if: entry exists, connection is active, and not yet ended/prepared
+ // Also check that the connection is the same (for connection pooling)
+ if entry.conn == nil {
+ return false
+ }
+ return entry.state == xaStateStarted && entry.conn.xaActive && entry.branchID != "pending"
+}
+
+// setState updates the state of an XA entry
+func (r *xaRegistry) setState(xid string, state xaState) {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+
+ if entry, ok := r.entries[xid]; ok {
+ oldState := entry.state
+ entry.state = state
+ log.Infof("XA state changed, xid: %s, branchID: %s, %v -> %v",
+ xid, entry.branchID, oldState, state)
+ }
+}
+
+// unregister removes an XA entry from the registry
+func (r *xaRegistry) unregister(xid string) {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+
+ if entry, ok := r.entries[xid]; ok {
+ log.Infof("Unregistered XA branch, xid: %s, branchID: %s, statementCount: %d",
+ xid, entry.branchID, entry.statementCount)
+ delete(r.entries, xid)
+ }
+}
+
+// cleanup removes stale entries (called periodically)
+func (r *xaRegistry) cleanup() {
Review Comment:
同上，没有使用
##########
pkg/datasource/sql/conn_xa.go:
##########
@@ -289,12 +331,19 @@ func (c *XAConn) start(ctx context.Context) error {
return fmt.Errorf("xa xid %s resource connection start err:%w", c.txCtx.XID, err)
}
+ // For multi-statement XA transactions (originally in autoCommit mode),
+ // skip the termination check. The check will be done during Phase 2 commit/rollback.
+ if c.txCtx.IsAutoCommitXABranch {
Review Comment:
这里是不不太对，`c.start(ctx)` 调用前已经 check `wasAutoCommit` 了
##########
pkg/datasource/sql/xa_registry.go:
##########
@@ -0,0 +1,192 @@
+/*
+ * 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 (
+ "sync"
+ "time"
+
+ "seata.apache.org/seata-go/v2/pkg/util/log"
+)
+
+// xaState represents the state of an XA transaction branch
+type xaState int
+
+const (
+ xaStateIdle xaState = iota // No XA transaction
+ xaStateStarted // XA START executed
+ xaStateEnded // XA END executed
+ xaStatePrepared // XA PREPARE executed
+)
+
+// xaEntry represents an active XA transaction branch
+type xaEntry struct {
+ conn *XAConn
+ xid string
+ branchID string
+ resourceID string
+ state xaState
+ createTime time.Time
+ lastAccessTime time.Time
+ statementCount int
+}
+
+// xaRegistry manages XA connections for global transactions
+// It ensures multiple SQL operations in the same global transaction
+// reuse the same XA branch, preventing "busy buffer" errors
+type xaRegistry struct {
+ mu sync.RWMutex
+ entries map[string]*xaEntry // key: xid
+}
+
+var (
+ globalRegistry *xaRegistry
+ registryOnce sync.Once
+)
+
+// getXARegistry returns the global XA registry (singleton)
+func getXARegistry() *xaRegistry {
+ registryOnce.Do(func() {
+ globalRegistry = &xaRegistry{
+ entries: make(map[string]*xaEntry),
+ }
+ })
+ return globalRegistry
+}
+
+// register registers or retrieves an XA connection for the given xid
+// Returns (isNew, entry, error)
+func (r *xaRegistry) register(xid, branchID, resourceID string, conn *XAConn) (bool, *xaEntry) {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+
+ now := time.Now()
+
+ // Check if entry already exists
+ if entry, ok := r.entries[xid]; ok {
+ entry.lastAccessTime = now
+ entry.statementCount++
+ log.Infof("Reusing existing XA branch for xid: %s, branchID: %s, statementCount: %d, state: %v",
+ xid, entry.branchID, entry.statementCount, entry.state)
+ return false, entry
+ }
+
+ // Create new entry
+ entry := &xaEntry{
+ conn: conn,
+ xid: xid,
+ branchID: branchID,
+ resourceID: resourceID,
+ state: xaStateStarted,
+ createTime: now,
+ lastAccessTime: now,
+ statementCount: 1,
+ }
+
+ r.entries[xid] = entry
+ log.Infof("Registered new XA branch, xid: %s, branchID: %s, resourceID: %s", xid, branchID, resourceID)
+
+ return true, entry
+}
+
+// get retrieves an XA entry by xid
+func (r *xaRegistry) get(xid string) (*xaEntry, bool) {
+ r.mu.RLock()
+ defer r.mu.RUnlock()
+
+ entry, ok := r.entries[xid]
+ if ok {
+ entry.lastAccessTime = time.Now()
Review Comment:
为什么在读锁下做写操作？这里会出现竞态，可以调整下：
```go
type xaEntry struct {
// ...
lastAccessTime atomic.Int64
}
func (r *xaRegistry) get(xid string) (*xaEntry, bool) {
// ...
if ok {
entry.lastAccessTime.Store(time.Now().UnixNano())
}
// ...
}
```
##########
pkg/datasource/sql/xa_registry.go:
##########
@@ -0,0 +1,192 @@
+/*
+ * 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 (
+ "sync"
+ "time"
+
+ "seata.apache.org/seata-go/v2/pkg/util/log"
+)
+
+// xaState represents the state of an XA transaction branch
+type xaState int
+
+const (
+ xaStateIdle xaState = iota // No XA transaction
+ xaStateStarted // XA START executed
+ xaStateEnded // XA END executed
+ xaStatePrepared // XA PREPARE executed
+)
+
+// xaEntry represents an active XA transaction branch
+type xaEntry struct {
+ conn *XAConn
+ xid string
+ branchID string
+ resourceID string
+ state xaState
+ createTime time.Time
+ lastAccessTime time.Time
+ statementCount int
+}
+
+// xaRegistry manages XA connections for global transactions
+// It ensures multiple SQL operations in the same global transaction
+// reuse the same XA branch, preventing "busy buffer" errors
+type xaRegistry struct {
+ mu sync.RWMutex
+ entries map[string]*xaEntry // key: xid
+}
+
+var (
+ globalRegistry *xaRegistry
+ registryOnce sync.Once
+)
+
+// getXARegistry returns the global XA registry (singleton)
+func getXARegistry() *xaRegistry {
+ registryOnce.Do(func() {
+ globalRegistry = &xaRegistry{
+ entries: make(map[string]*xaEntry),
+ }
+ })
+ return globalRegistry
+}
+
+// register registers or retrieves an XA connection for the given xid
+// Returns (isNew, entry, error)
+func (r *xaRegistry) register(xid, branchID, resourceID string, conn *XAConn) (bool, *xaEntry) {
Review Comment:
xa_registry感觉是非必要的，这里完全没有使用
##########
pkg/datasource/sql/conn_xa.go:
##########
@@ -196,63 +205,96 @@ func (c *XAConn) createOnceTxContext(ctx context.Context) bool {
c.txCtx.ResourceID = c.res.resourceID
c.txCtx.XID = tm.GetXID(ctx)
c.txCtx.TransactionMode = types.XAMode
- c.txCtx.GlobalLockRequire = true
Review Comment:
这里为什么删除？
##########
pkg/datasource/sql/db.go:
##########
@@ -117,7 +117,8 @@ type DBResource struct {
// for xa
metaCache datasource.TableMetaCache
shouldBeHeld bool
- keeper sync.Map
+ keeper sync.Map // xaBranchID -> *XAConn
+ xaConnsByXID sync.Map // xid -> *XAConn (for XA branch reuse in autoCommit mode)
Review Comment:
如果异常导致Unregistry行为没有执行这里是不是会出现泄漏的情况？
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org
Re: [PR] feat: support XA branch reuse in autoCommit mode [incubator-seata-go]
Posted by "YuZhangLarry (via GitHub)" <gi...@apache.org>.
YuZhangLarry commented on code in PR #1137:
URL: https://github.com/apache/incubator-seata-go/pull/1137#discussion_r3620270072
##########
pkg/datasource/sql/conn_xa.go:
##########
@@ -237,22 +244,51 @@ func (c *XAConn) createNewTxOnExecIfNeed(ctx context.Context, f func() (types.Ex
// execute SQL
ret, err := f()
if err != nil {
+ // driver.ErrSkip is not a real failure: it asks database/sql to retry this
+ // statement through the Prepare+Exec fallback path.
+ if errors.Is(err, driver.ErrSkip) {
+ // If we already opened an XA branch for this statement (autoCommit mode),
+ // the fallback retry would run OUTSIDE the branch - autoCommit is now false
+ // and txCtx has been reset - leaving this branch registered-but-never-
+ // prepared (a leak) and the retried write escaping the global transaction.
+ // Roll the branch back and surface a real (non-ErrSkip) error so the caller
+ // fails fast instead of silently corrupting the global transaction.
+ if tx != nil {
+ if rollbackErr := tx.Rollback(); rollbackErr != nil {
+ log.Errorf("failed to rollback xa branch of :%s after ErrSkip, err:%v", c.txCtx.XID, rollbackErr)
+ }
+ xaRollbacked = true
+ return nil, fmt.Errorf("xa branch %s cannot fall back to non-XA execution after XA START: %v", c.txCtx.XID, err)
+ }
+ // No branch opened yet - safe to let database/sql use its fallback path.
+ return nil, err
+ }
+ // 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.
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 {
Review Comment:
好的，我改一下
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org