thunguo commented on code in PR #1052: URL: https://github.com/apache/incubator-seata-go/pull/1052#discussion_r2870692292
########## 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 Review Comment: We need to consider whether the registry is truly necessary. My understanding is that as long as there is an active XA branch on the current connection, it can be reused. Perhaps it would be better to check this situation? ########## pkg/datasource/sql/conn_xa.go: ########## @@ -127,9 +139,15 @@ func (c *XAConn) BeginTx(ctx context.Context, opts driver.TxOptions) (driver.Tx, c.txCtx.ResourceID = c.res.resourceID c.txCtx.XID = tm.GetXID(ctx) c.txCtx.TransactionMode = types.XAMode + // Store the original autoCommit state for later use in commit logic + c.txCtx.GlobalLockRequire = wasAutoCommit Review Comment: Is there any semantic confusion here? Would defining a new field like `IsAutoCommitXABranch` be better? ########## pkg/datasource/sql/exec/executor.go: ########## @@ -58,6 +59,29 @@ 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 (like undo log generation) + // XA transactions don't need undo logs - they use the two-phase commit protocol managed by TC + if transactionMode == types.XAMode { Review Comment: Could you double-check if this change is necessary? The `undo_log_hook.go` file handles this conditional branch, meaning the hook should occur earlier than the logic in the executor. It's likely that this part won't actually be executed. I understand that the current changes in `undo_log_hook.go` are sufficient. ########## pkg/datasource/sql/conn_xa.go: ########## @@ -184,45 +202,110 @@ func (c *XAConn) createNewTxOnExecIfNeed(ctx context.Context, f func() (types.Ex err error ) + xid := tm.GetXID(ctx) + log.Infof("[XA-DEBUG] createNewTxOnExecIfNeed called, xid: %s, autoCommit: %v, txCtx.TransactionMode: %v, txCtx.XID: %s", + xid, c.autoCommit, c.txCtx.TransactionMode, c.txCtx.XID) + defer func() { recoverErr := recover() - if err != nil || recoverErr != nil { - log.Errorf("conn at rollback error:%v or recoverErr:%v", err, recoverErr) + // 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") + + if (err != nil && !isErrSkip) || recoverErr != nil { + log.Errorf("[XA-DEBUG] defer triggered, err: %v, recoverErr: %v", err, recoverErr) + // Don't try to rollback if tx is nil if c.tx != nil { + log.Infof("[XA-DEBUG] calling c.tx.Rollback() in defer") rollbackErr := c.tx.Rollback() if rollbackErr != nil { log.Errorf("conn at rollback error:%v", rollbackErr) } } + } else { + log.Infof("[XA-DEBUG] defer: no error or ErrSkip, skipping rollback") } }() currentAutoCommit := c.autoCommit + registry := getXARegistry() + + // For global transactions in autoCommit mode, create/reuse XA branch if c.txCtx.TransactionMode != types.Local && tm.IsGlobalTx(ctx) && c.autoCommit { + // Check if we already have an active XA branch for this transaction + if registry.canReuse(xid) { + // Reuse existing XA branch - execute SQL directly + log.Infof("Reusing existing XA branch for xid: %s", xid) + ret, err := f() + if err != nil { + // On error (but not ErrSkip), rollback the entire branch + isErrSkip := err == driver.ErrSkip || err.Error() == "driver: skip fast-path; continue as if unimplemented" + if !isErrSkip && c.xaActive { + if rollbackErr := c.Rollback(ctx); rollbackErr != nil { Review Comment: `c.Rollback(ctx)` performs an XA-level rollback (XA END + XA ROLLBACK), but does not set `c.tx` to nil. This causes the `defer` statement at the top of the function to subsequently call `c.tx.Rollback()` again, resulting in a duplicate rollback of the already terminated underlying connection. Could this potentially lead to a `bad connection`? ########## pkg/datasource/sql/conn_xa.go: ########## @@ -346,6 +458,11 @@ func (c *XAConn) Commit(ctx context.Context) error { } now := time.Now() + + // Update registry state to ENDED before XA END + registry := getXARegistry() + registry.setState(c.txCtx.XID, xaStateEnded) Review Comment: Should the status modification operation be performed after the XA END action? -- 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]
