fix txn status invalid to 2.0 (#19817)

fix txn status invalid if commit failed or commit a readonly txn

Approved by: @iamlinjunhong, @sukki37

pkg/txn/client/operator.go

@@ -611,15 +611,15 @@ func (tc *txnOperator) Commit(ctx context.Context) (err error) {
 
 	tc.reset.commitCounter.addEnter()
 	defer tc.reset.commitCounter.addExit()
-	txn := tc.getTxnMeta(false)
-	util.LogTxnCommit(tc.logger, txn)
+	txnMeta := tc.getTxnMeta(false)
+	util.LogTxnCommit(tc.logger, txnMeta)
 
 	readonly := tc.reset.workspace != nil && tc.reset.workspace.Readonly()
 	if !readonly {
 		tc.reset.commitSeq = tc.NextSequence()
 		tc.reset.commitAt = time.Now()
 
-		tc.triggerEvent(newEvent(CommitEvent, txn, tc.reset.commitSeq, nil))
+		tc.triggerEvent(newEvent(CommitEvent, txnMeta, tc.reset.commitSeq, nil))
 		defer func() {
 			cost := time.Since(tc.reset.commitAt)
 			v2.TxnCNCommitDurationHistogram.Observe(cost.Seconds())
@@ -630,6 +630,7 @@ func (tc *txnOperator) Commit(ctx context.Context) (err error) {
 	if tc.opts.options.ReadOnly() {
 		tc.mu.Lock()
 		defer tc.mu.Unlock()
+		tc.mu.txn.Status = txn.TxnStatus_Committed
 		tc.closeLocked()
 		return
 	}
@@ -1257,6 +1258,9 @@ func (tc *txnOperator) needUnlockLocked() bool {
 func (tc *txnOperator) closeLocked() {
 	if !tc.mu.closed {
 		tc.mu.closed = true
+		if tc.reset.commitErr != nil {
+			tc.mu.txn.Status = txn.TxnStatus_Aborted
+		}
 		tc.triggerEventLocked(
 			TxnEvent{
 				Event: ClosedEvent,

pkg/txn/client/operator_test.go

@@ -129,6 +129,7 @@ func TestCommitWithNoWrite(t *testing.T) {
 		err := tc.Commit(ctx)
 		assert.NoError(t, err)
 		assert.Empty(t, ts.getLastRequests())
+		assert.Equal(t, txn.TxnStatus_Committed, tc.mu.txn.Status)
 	})
 }
 
@@ -137,6 +138,7 @@ func TestCommitReadOnly(t *testing.T) {
 		err := tc.Commit(ctx)
 		assert.NoError(t, err)
 		assert.Empty(t, ts.getLastRequests())
+		assert.Equal(t, txn.TxnStatus_Committed, tc.mu.txn.Status)
 	}, WithTxnReadyOnly())
 }
 
@@ -209,6 +211,7 @@ func TestCommitWithLockTablesChanged(t *testing.T) {
 				tc.mu.txn.TNShards = append(tc.mu.txn.TNShards, metadata.TNShard{TNShardRecord: metadata.TNShardRecord{ShardID: 1}})
 				err = tc.Commit(ctx)
 				assert.Error(t, err)
+				assert.Equal(t, txn.TxnStatus_Aborted, tc.mu.txn.Status)
 
 				// table 1 will be removed
 				bind, err := s.GetLockTableBind(0, tableID1)