You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@seata.apache.org by "Copilot (via GitHub)" <gi...@apache.org> on 2026/03/25 08:35:08 UTC
Re: [PR] feat: XA mode 'busy buffer' and 'bad connection' errors causing rollbacks [incubator-seata-go]
Copilot commented on code in PR #1052:
URL: https://github.com/apache/incubator-seata-go/pull/1052#discussion_r2986629967
##########
pkg/datasource/sql/conn_xa.go:
##########
@@ -337,6 +392,12 @@ func (c *XAConn) rollbackErrorHandle() error {
}
func (c *XAConn) Commit(ctx context.Context) error {
+ // If this XA branch was created in autoCommit mode (multi-statement transaction),
+ // don't do the actual XA commit here. The TC will handle it in Phase 2.
+ if c.txCtx.IsAutoCommitXABranch {
Review Comment:
`IsAutoCommitXABranch` causes an early return here, but the XA branch reuse registry (`xaConnsByXID`) is not cleaned up on the success path for auto-commit multi-statement transactions. Since `UnregisterXABranch` is only called in `Rollback` and the non-autoCommit prepare path, this can leak entries indefinitely. Ensure the xid->conn mapping is removed when the global transaction finishes (e.g., during phase2 `BranchCommit`/`BranchRollback`, or in `XaCommit`/`XaRollback` after the RM completes).
```suggestion
if c.txCtx.IsAutoCommitXABranch {
// The global transaction is finishing; clean up the XA branch reuse registry entry.
c.res.UnregisterXABranch(c.txCtx.XID)
```
##########
test/xa_issue904_test.go:
##########
@@ -0,0 +1,127 @@
+/*
+ * 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 test
+
+import (
+ "context"
+ "database/sql"
+ "fmt"
+ "os"
+ "path/filepath"
+ "testing"
+ "time"
+
+ _ "github.com/go-sql-driver/mysql"
+ "seata.apache.org/seata-go/v2/pkg/client"
+ "seata.apache.org/seata-go/v2/pkg/tm"
+)
+
+func init() {
+ // 设置配置文件路径
+ absPath, _ := filepath.Abs("../testdata/conf/seatago.yml")
+ os.Setenv("SEATA_GO_CONFIG_PATH", absPath)
+}
+
+// TestXAIssue904 复现 issue #904
+// XA 模式下 SELECT FOR UPDATE 后 UPDATE 报 busy buffer / bad connection
+func TestXAIssue904(t *testing.T) {
+ // MySQL 连接信息
+ const mysqlDSN = "root:1234@tcp(127.0.0.1:3306)/seata_test?charset=utf8mb4"
+
+ t.Log("=== 测试 issue #904: XA 模式 SELECT FOR UPDATE 后 UPDATE ===")
+
+ // 初始化 Seata 客户端
+ client.Init()
+ t.Log("Seata client initialized")
+
+ // 创建 XA 数据源
+ db, err := sql.Open("seata-xa-mysql", mysqlDSN)
+ if err != nil {
+ t.Fatalf("Failed to open database: %v", err)
Review Comment:
This test hard-codes a local MySQL DSN and assumes a running Seata server/config, which will make `go test ./...` fail or hang in CI and for most dev environments. Convert this into an opt-in integration test (e.g., guard with an env var and `t.Skip` when not set), and read DSN/config from env instead of embedding credentials/hostnames in the test.
##########
pkg/datasource/sql/conn_xa.go:
##########
@@ -198,30 +209,64 @@ func (c *XAConn) createNewTxOnExecIfNeed(ctx context.Context, f func() (types.Ex
}()
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)
+ }
+ // Current connection already has the XA branch, execute SQL directly
+ // Skip creating a new branch
+ } else {
+ // Create new XA branch
Review Comment:
The delegation path for XA branch reuse is incorrect: when `heldConn != c`, you call `heldConn.createNewTxOnExecIfNeed(ctx, f)` but `f` is a closure created in `QueryContext`/`ExecContext` that executes on the original receiver `c.Conn`, not on `heldConn.Conn`. This means the SQL still runs on the wrong physical connection while the XA branch is held on `heldConn`, defeating branch reuse and potentially breaking transaction semantics. Restructure this so the SQL execution is performed on `heldConn` (e.g., make `createNewTxOnExecIfNeed` accept a callback that receives the connection to use, or handle the `heldConn` selection before constructing the callback).
```suggestion
// Check if we already have an active XA branch for this transaction on the *same* connection
heldConn := c.res.GetXABranch(xid)
if heldConn != nil && heldConn == c && heldConn.xaActive && heldConn.txCtx.XID == xid {
// Current connection already has the XA branch, execute SQL directly
// Skip creating a new branch
} else {
// Create new XA branch bound to the current connection
```
##########
pkg/datasource/sql/conn_xa.go:
##########
@@ -172,23 +177,29 @@ 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 err != nil || recoverErr != nil {
- log.Errorf("conn at rollback error:%v or recoverErr:%v", err, recoverErr)
- if c.tx != nil {
+ // Check if error is ErrSkip - don't rollback for this special error
+ isErrSkip := err != nil && (err == driver.ErrSkip || err.Error() == "driver: skip fast-path; continue as if unimplemented")
Review Comment:
`driver.ErrSkip` handling relies partly on comparing `err.Error()` to a hard-coded string. This is brittle and can break if the error gets wrapped or the message changes. Prefer `errors.Is(err, driver.ErrSkip)` (or a direct equality check if you know the driver returns the sentinel) and remove the string comparison.
```suggestion
isErrSkip := err != nil && errors.Is(err, driver.ErrSkip)
```
##########
pkg/datasource/sql/tx.go:
##########
@@ -135,6 +135,12 @@ func (tx *Tx) Rollback() error {
}
}
+ // In XA mode, target might be nil (set with withOriginTx(nil))
+ // The XA transaction is managed separately, so just return nil
+ if tx.target == nil {
+ return nil
Review Comment:
`Rollback()` now silently succeeds whenever `tx.target` is nil. Since a nil target is expected only for XA mode (see `withOriginTx(nil)`), this should be guarded (e.g., only allow nil when `tx.tranCtx != nil && tx.tranCtx.TransactionMode == types.XAMode`) and otherwise return an error to avoid masking unexpected driver/initialization bugs.
```suggestion
// In XA mode, target might be nil (set with withOriginTx(nil)).
// Only allow a nil target when we are 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 transaction is managed separately.
return nil
}
return fmt.Errorf("sql.Tx Rollback: nil underlying transaction in non-XA mode")
```
##########
pkg/datasource/sql/exec/executor.go:
##########
@@ -58,6 +58,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:
The new XA-mode branch in `BuildExecutor` isn’t covered by tests (current `executor_test.go` only exercises AT mode). Add a test case for `transactionMode == types.XAMode` to ensure a `BaseExecutor` is returned and interceptors are still applied as expected.
##########
test/xa_issue904_test.go:
##########
@@ -0,0 +1,127 @@
+/*
+ * 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 test
+
+import (
+ "context"
+ "database/sql"
+ "fmt"
+ "os"
+ "path/filepath"
+ "testing"
+ "time"
+
+ _ "github.com/go-sql-driver/mysql"
+ "seata.apache.org/seata-go/v2/pkg/client"
+ "seata.apache.org/seata-go/v2/pkg/tm"
+)
+
+func init() {
+ // 设置配置文件路径
+ absPath, _ := filepath.Abs("../testdata/conf/seatago.yml")
+ os.Setenv("SEATA_GO_CONFIG_PATH", absPath)
+}
+
+// TestXAIssue904 复现 issue #904
+// XA 模式下 SELECT FOR UPDATE 后 UPDATE 报 busy buffer / bad connection
+func TestXAIssue904(t *testing.T) {
+ // MySQL 连接信息
+ const mysqlDSN = "root:1234@tcp(127.0.0.1:3306)/seata_test?charset=utf8mb4"
+
+ t.Log("=== 测试 issue #904: XA 模式 SELECT FOR UPDATE 后 UPDATE ===")
+
+ // 初始化 Seata 客户端
+ client.Init()
+ t.Log("Seata client initialized")
+
+ // 创建 XA 数据源
+ db, err := sql.Open("seata-xa-mysql", mysqlDSN)
+ if err != nil {
+ t.Fatalf("Failed to open database: %v", err)
+ }
+ defer db.Close()
+
+ t.Log("XA 数据源创建成功")
+
+ // 重置测试数据
+ resetTestData(t, mysqlDSN)
+
+ // 执行 XA 事务测试
+ t.Run("SELECT_FOR_UPDATE_then_UPDATE", func(t *testing.T) {
+ err = tm.WithGlobalTx(context.Background(), &tm.GtxConfig{
+ Name: "test-xa-issue904",
+ Timeout: 60 * time.Second,
+ }, func(ctx context.Context) error {
+ // 第一步：查询 (带 FOR UPDATE)
+ var money int
+ err := db.QueryRowContext(ctx,
+ "SELECT money FROM account_tbl WHERE user_id = ? FOR UPDATE", "user1").
+ Scan(&money)
+ if err != nil {
+ return fmt.Errorf("query failed: %w", err)
+ }
+ t.Logf("Step 1: 查询成功, money = %d", money)
+
+ // 第二步：更新
+ newBalance := money - 10
+ _, err = db.ExecContext(ctx,
+ "UPDATE account_tbl SET money = ? WHERE user_id = ?", newBalance, "user1")
+ if err != nil {
+ return fmt.Errorf("update failed: %w", err)
+ }
+ t.Logf("Step 2: 更新成功, new balance = %d", newBalance)
+ return nil
+ })
+
+ if err != nil {
+ t.Errorf("❌ 事务失败: %v", err)
+ } else {
+ t.Log("✅ 事务成功")
+ }
+ })
+}
+
+func resetTestData(t *testing.T, dsn string) {
+ nativeDB, err := sql.Open("mysql", dsn)
+ if err != nil {
+ t.Fatalf("Failed to open native db: %v", err)
+ }
+ defer nativeDB.Close()
+
+ // 创建表
+ _, err = nativeDB.Exec(`
+ CREATE TABLE IF NOT EXISTS account_tbl (
+ id int(11) NOT NULL AUTO_INCREMENT,
+ user_id varchar(255) DEFAULT NULL,
+ money int(11) DEFAULT 0,
+ PRIMARY KEY (id)
+ ) ENGINE=InnoDB DEFAULT CHARSET=utf8
+ `)
+ if err != nil {
+ t.Logf("Warning: create table failed: %v", err)
+ }
+
+ // 重置数据
+ nativeDB.Exec("DELETE FROM account_tbl WHERE user_id = 'user1'")
+ _, err = nativeDB.Exec("INSERT INTO account_tbl (user_id, money) VALUES ('user1', 100)")
+ if err != nil {
+ t.Logf("Warning: insert data failed: %v", err)
Review Comment:
`resetTestData` ignores errors from the DELETE statement, and downgrades CREATE TABLE / INSERT failures to warnings, which can lead to false positives (the test continues even when setup failed). Make the setup fail fast by checking and returning/fatal-ing on all SQL execution errors.
```suggestion
t.Fatalf("Failed to create table account_tbl: %v", err)
}
// 重置数据
_, err = nativeDB.Exec("DELETE FROM account_tbl WHERE user_id = 'user1'")
if err != nil {
t.Fatalf("Failed to delete existing test data: %v", err)
}
_, err = nativeDB.Exec("INSERT INTO account_tbl (user_id, money) VALUES ('user1', 100)")
if err != nil {
t.Fatalf("Failed to insert test data: %v", err)
```
--
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