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: [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]
