opensource/dtm
3
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

test pass

Browse Source
This commit is contained in:
yedongfu 2021-05-27 23:25:28 +08:00
parent 12d8331def
commit 291cf0bb28
2 changed files with 22 additions and 80 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

85
dtmsvr/trans.go
View File

@ -3,7 +3,6 @@ package dtmsvr
import ( import (
"fmt" "fmt"
"strings" "strings"
"time"
"github.com/yedf/dtm/common" "github.com/yedf/dtm/common"
) )
@ -62,7 +61,7 @@ func (t *TransSagaProcessor) ProcessOnce(db *common.MyDb, branches []TransBranch
} }
} }
if current == len(branches) { // saga 事务完成 if current == len(branches) { // saga 事务完成
t.saveStatus(db.Must(), "finished") t.changeStatus(db.Must(), "finished")
return nil return nil
} }
for current = current - 1; current >= 0; current-- { for current = current - 1; current >= 0; current-- {
@ -84,7 +83,7 @@ func (t *TransSagaProcessor) ProcessOnce(db *common.MyDb, branches []TransBranch
if current != -1 { if current != -1 {
return fmt.Errorf("saga current not -1") return fmt.Errorf("saga current not -1")
} }
t.saveStatus(db.Must(), "rollbacked") t.changeStatus(db.Must(), "rollbacked")
return nil return nil
} }
@ -112,7 +111,6 @@ func (t *TransTccProcessor) GenBranches() []TransBranch {
} }
func (t *TransTccProcessor) ProcessOnce(db *common.MyDb, branches []TransBranch) error { func (t *TransTccProcessor) ProcessOnce(db *common.MyDb, branches []TransBranch) error {
gid := t.Gid
current := 0 // 当前正在处理的步骤 current := 0 // 当前正在处理的步骤
for ; current < len(branches); current++ { for ; current < len(branches); current++ {
step := branches[current] step := branches[current]
@ -125,21 +123,11 @@ func (t *TransTccProcessor) ProcessOnce(db *common.MyDb, branches []TransBranch)
return err return err
} }
body := resp.String() body := resp.String()
db.Must().Model(&TransGlobal{}).Where("gid=?", gid).Update("gid", gid) // 更新update_time，避免被定时任务再次 t.touch(db)
if strings.Contains(body, "SUCCESS") { if strings.Contains(body, "SUCCESS") {
writeTransLog(gid, "step finished", "finished", step.Branch, "") step.changeStatus(db, "finished")
dbr := db.Must().Model(&step).Where("status=?", "prepared").Updates(M{
"status": "finished",
"finish_time": time.Now(),
})
checkAffected(dbr)
} else if strings.Contains(body, "FAIL") { } else if strings.Contains(body, "FAIL") {
writeTransLog(gid, "step rollbacked", "rollbacked", step.Branch, "") step.changeStatus(db, "rollbacked")
dbr := db.Must().Model(&step).Where("status=?", "prepared").Updates(M{
"status": "rollbacked",
"rollback_time": time.Now(),
})
checkAffected(dbr)
break break
} else { } else {
return fmt.Errorf("unknown response: %s, will be retried", body) return fmt.Errorf("unknown response: %s, will be retried", body)
@ -148,44 +136,9 @@ func (t *TransTccProcessor) ProcessOnce(db *common.MyDb, branches []TransBranch)
} }
////////////////////////////////////////////////// //////////////////////////////////////////////////
if current == len(branches) { // tcc 事务完成 if current == len(branches) { // tcc 事务完成
writeTransLog(gid, "saga finished", "finished", "", "") t.changeStatus(db, "finished")
dbr := db.Must().Model(&TransGlobal{}).Where("gid=? and status=?", gid, "committed").Updates(M{
"status": "finished",
"finish_time": time.Now(),
})
checkAffected(dbr)
return nil return nil
} }
for current = current - 1; current >= 0; current-- {
step := branches[current]
if step.BranchType != "compensate" || step.Status != "prepared" {
continue
}
resp, err := common.RestyClient.R().SetBody(step.Data).SetQueryParam("gid", step.Gid).Post(step.Url)
if err != nil {
return err
}
body := resp.String()
if strings.Contains(body, "SUCCESS") {
writeTransLog(gid, "step rollbacked", "rollbacked", step.Branch, "")
dbr := db.Must().Model(&step).Where("status=?", step.Status).Updates(M{
"status": "rollbacked",
"rollback_time": time.Now(),
})
checkAffected(dbr)
} else {
return fmt.Errorf("expect compensate return SUCCESS")
}
}
if current != -1 {
return fmt.Errorf("saga current not -1")
}
writeTransLog(gid, "saga rollbacked", "rollbacked", "", "")
dbr := db.Must().Model(&TransGlobal{}).Where("status=? and gid=?", "committed", gid).Updates(M{
"status": "rollbacked",
"rollback_time": time.Now(),
})
checkAffected(dbr)
return nil return nil
} }
@ -207,7 +160,7 @@ func (t *TransXaProcessor) ProcessOnce(db *common.MyDb, branches []TransBranch)
if branch.Status == "finished" { if branch.Status == "finished" {
continue continue
} }
db.Must().Model(&TransGlobal{}).Where("gid=?", gid).Update("gid", gid) // 更新update_time，避免被定时任务再次 t.touch(db) // 更新update_time，避免被定时任务再次
resp, err := common.RestyClient.R().SetBody(M{ resp, err := common.RestyClient.R().SetBody(M{
"branch": branch.Branch, "branch": branch.Branch,
"action": "commit", "action": "commit",
@ -220,17 +173,9 @@ func (t *TransXaProcessor) ProcessOnce(db *common.MyDb, branches []TransBranch)
if !strings.Contains(body, "SUCCESS") { if !strings.Contains(body, "SUCCESS") {
return fmt.Errorf("bad response: %s", body) return fmt.Errorf("bad response: %s", body)
} }
writeTransLog(gid, "step finished", "finished", branch.Branch, "") branch.changeStatus(db, "finished")
db.Must().Model(&branch).Where("status=?", "prepared").Updates(M{
"status": "finished",
"finish_time": time.Now(),
})
} }
writeTransLog(gid, "xa finished", "finished", "", "") t.changeStatus(db, "finished")
db.Must().Model(&TransGlobal{}).Where("gid=? and status=?", gid, "committed").Updates(M{
"status": "finished",
"finish_time": time.Now(),
})
} else if t.Status == "prepared" { // 未commit直接处理的情况为回滚场景 } else if t.Status == "prepared" { // 未commit直接处理的情况为回滚场景
for _, branch := range branches { for _, branch := range branches {
if branch.Status == "rollbacked" { if branch.Status == "rollbacked" {
@ -249,17 +194,9 @@ func (t *TransXaProcessor) ProcessOnce(db *common.MyDb, branches []TransBranch)
if !strings.Contains(body, "SUCCESS") { if !strings.Contains(body, "SUCCESS") {
return fmt.Errorf("bad response: %s", body) return fmt.Errorf("bad response: %s", body)
} }
writeTransLog(gid, "step rollbacked", "rollbacked", branch.Branch, "") branch.changeStatus(db, "rollbacked")
db.Must().Model(&branch).Where("status=?", "prepared").Updates(M{
"status": "rollbacked",
"finish_time": time.Now(),
})
} }
writeTransLog(gid, "xa rollbacked", "rollbacked", "", "") t.changeStatus(db, "rollbacked")
db.Must().Model(&TransGlobal{}).Where("gid=? and status=?", gid, "prepared").Updates(M{
"status": "rollbacked",
"finish_time": time.Now(),
})
} else { } else {
return fmt.Errorf("bad trans status: %s", t.Status) return fmt.Errorf("bad trans status: %s", t.Status)
} }

15
dtmsvr/types.go
View File

@ -37,12 +37,17 @@ func (t *TransGlobal) touch(db *common.MyDb) *gorm.DB {
return db.Model(&TransGlobal{}).Where("gid=?", t.Gid).Update("gid", t.Gid) // 更新update_time，避免被定时任务再次 return db.Model(&TransGlobal{}).Where("gid=?", t.Gid).Update("gid", t.Gid) // 更新update_time，避免被定时任务再次
} }
func (t *TransGlobal) saveStatus(db *common.MyDb, status string) *gorm.DB { func (t *TransGlobal) changeStatus(db *common.MyDb, status string) *gorm.DB {
writeTransLog(t.Gid, "step change", status, "", "") writeTransLog(t.Gid, "change status", status, "", "")
dbr := db.Must().Model(t).Where("status=?", t.Status).Updates(M{ updates := M{
"status": status, "status": status,
"finish_time": time.Now(), }
}) if status == "finished" {
updates["finish_time"] = time.Now()
} else if status == "rollbacked" {
updates["rollback_time"] = time.Now()
}
dbr := db.Must().Model(t).Where("status=?", t.Status).Updates(updates)
checkAffected(dbr) checkAffected(dbr)
t.Status = status t.Status = status
return dbr return dbr