saga http failed rollback reason

dtmsvr/storage/trans.go

@@ -55,14 +55,15 @@ func (g *TransGlobalStore) String() string {
 // TransBranchStore branch transaction
 type TransBranchStore struct {
 	dtmutil.ModelBase
-	Gid          string `json:"gid,omitempty"`
-	URL          string `json:"url,omitempty"`
-	BinData      []byte
-	BranchID     string     `json:"branch_id,omitempty"`
-	Op           string     `json:"op,omitempty"`
-	Status       string     `json:"status,omitempty"`
-	FinishTime   *time.Time `json:"finish_time,omitempty"`
-	RollbackTime *time.Time `json:"rollback_time,omitempty"`
+	Gid            string `json:"gid,omitempty"`
+	URL            string `json:"url,omitempty"`
+	BinData        []byte
+	BranchID       string     `json:"branch_id,omitempty"`
+	Op             string     `json:"op,omitempty"`
+	Status         string     `json:"status,omitempty"`
+	FinishTime     *time.Time `json:"finish_time,omitempty"`
+	RollbackTime   *time.Time `json:"rollback_time,omitempty"`
+	RollbackReason string     `json:"-" gorm:"-"`
 }
 
 // TableName TableName

dtmsvr/trans_status.go

@@ -190,6 +190,7 @@ func (t *TransGlobal) getBranchResult(branch *TransBranch) (string, error) {
 	if err == nil {
 		return dtmcli.StatusSucceed, nil
 	} else if t.TransType == "saga" && branch.Op == dtmimp.OpAction && errors.Is(err, dtmcli.ErrFailure) {
+		branch.RollbackReason = fmt.Sprintf("url:%s return failed: %s", branch.URL, err.Error())
 		return dtmcli.StatusFailed, nil
 	} else if errors.Is(err, dtmcli.ErrOngoing) {
 		return "", dtmcli.ErrOngoing

dtmsvr/trans_type_saga.go

@@ -45,10 +45,11 @@ type cSagaCustom struct {
 }
 
 type branchResult struct {
-	index   int
-	status  string
-	started bool
-	op      string
+	index          int
+	status         string
+	started        bool
+	op             string
+	rollbackReason string
 }
 
 func (t *transSagaProcessor) ProcessOnce(branches []TransBranch) error {
@@ -73,6 +74,7 @@ func (t *transSagaProcessor) ProcessOnce(branches []TransBranch) error {
 	}
 	// resultStats
 	var rsAToStart, rsAStarted, rsADone, rsAFailed, rsASucceed, rsCToStart, rsCDone, rsCSucceed int
+	var rollbackReason string
 	branchResults := make([]branchResult, n) // save the branch result
 	for i := 0; i < n; i++ {
 		b := branches[i]
@@ -125,7 +127,7 @@ func (t *transSagaProcessor) ProcessOnce(branches []TransBranch) error {
 			if x := recover(); x != nil {
 				err = dtmimp.AsError(x)
 			}
-			resultChan <- branchResult{index: i, status: branches[i].Status, op: branches[i].Op}
+			resultChan <- branchResult{index: i, status: branches[i].Status, op: branches[i].Op, rollbackReason: branches[i].RollbackReason}
 			if err != nil && !errors.Is(err, dtmcli.ErrOngoing) {
 				logger.Errorf("exec branch %s %s %s error: %v", branches[i].BranchID, branches[i].Op, branches[i].URL, err)
 			}
@@ -172,6 +174,7 @@ func (t *transSagaProcessor) ProcessOnce(branches []TransBranch) error {
 				rsADone++
 				if r.status == dtmcli.StatusFailed {
 					rsAFailed++
+					rollbackReason = r.rollbackReason
 				} else if r.status == dtmcli.StatusSucceed {
 					rsASucceed++
 				}
@@ -220,7 +223,7 @@ func (t *transSagaProcessor) ProcessOnce(branches []TransBranch) error {
 		return nil
 	}
 	if t.Status == dtmcli.StatusSubmitted && rsAFailed > 0 {
-		t.changeStatus(dtmcli.StatusAborting, withRollbackReason("Transaction branch execution failed"))
+		t.changeStatus(dtmcli.StatusAborting, withRollbackReason(rollbackReason))
 	}
 	if t.Status == dtmcli.StatusSubmitted && t.isTimeout() {
 		t.changeStatus(dtmcli.StatusAborting, withRollbackReason(fmt.Sprintf("Timeout after %d seconds", t.TimeoutToFail)))

test/saga_test.go

@@ -27,12 +27,13 @@ func TestSagaNormal(t *testing.T) {
 
 func TestSagaRollback(t *testing.T) {
 	saga := genSaga(dtmimp.GetFuncName(), false, true)
+	saga.Concurrent = false
 	err := saga.Submit()
 	assert.Nil(t, err)
 	waitTransProcessed(saga.Gid)
 	assert.Equal(t, []string{StatusSucceed, StatusSucceed, StatusSucceed, StatusFailed}, getBranchesStatus(saga.Gid))
 	assert.Equal(t, StatusFailed, getTransStatus(saga.Gid))
-	assert.Equal(t, "Transaction branch execution failed", getTrans(saga.Gid).RollbackReason)
+	assert.Equal(t, "url:http://localhost:8081/api/busi/TransIn return failed: {\"error\":\"FAILURE\"}. FAILURE", getTrans(saga.Gid).RollbackReason)
 }
 
 func TestSagaOngoingSucceed(t *testing.T) {

test/types.go

@@ -33,7 +33,7 @@ func waitTransProcessed(gid string) {
 	case id := <-dtmsvr.TransProcessedTestChan:
 		logger.FatalfIf(id != gid, "------- expecting: %s but %s found", gid, id)
 		logger.Debugf("finish for gid %s", gid)
-	case <-time.After(time.Duration(time.Second * 4)):
+	case <-time.After(time.Duration(time.Second * 10000)):
 		logger.FatalfIf(true, "Wait Trans timeout")
 	}
 }