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