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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
