diff --git a/dtmcli/barrier.go b/dtmcli/barrier.go index 31ee7ac..4a97a55 100644 --- a/dtmcli/barrier.go +++ b/dtmcli/barrier.go @@ -98,14 +98,14 @@ func ThroughBarrierCall(db *sql.DB, transInfo *TransInfo, busiCall BusiFunc) (re currentAffected, rerr := insertBarrier(tx, ti.TransType, ti.Gid, ti.BranchID, ti.BranchType, ti.BranchType) logrus.Printf("originAffected: %d currentAffected: %d", originAffected, currentAffected) if (ti.BranchType == "cancel" || ti.BranchType == "compensate") && originAffected > 0 { // 这个是空补偿,返回成功 - res = common.MS{"dtm_result": "SUCCESS"} + res = ResultSuccess return } else if currentAffected == 0 { // 插入不成功 var result sql.NullString err := common.StxQueryRow(tx, "select result from dtm_barrier.barrier where trans_type=? and gid=? and branch_id=? and branch_type=? and reason=?", ti.TransType, ti.Gid, ti.BranchID, ti.BranchType, ti.BranchType).Scan(&result) if err == sql.ErrNoRows { // 这个是悬挂操作,返回失败,AP收到这个返回,会尽快回滚 - res = common.MS{"dtm_result": "FAILURE"} + res = ResultFailure return } if err != nil { @@ -117,7 +117,7 @@ func ThroughBarrierCall(db *sql.DB, transInfo *TransInfo, busiCall BusiFunc) (re return } // 数据库里没有上次的结果,属于重复空补偿,直接返回成功 - res = common.MS{"dtm_result": "SUCCESS"} + res = ResultSuccess return } res, rerr = busiCall(tx) diff --git a/dtmcli/message.go b/dtmcli/message.go index 9cd5b64..02df337 100644 --- a/dtmcli/message.go +++ b/dtmcli/message.go @@ -1,8 +1,6 @@ package dtmcli import ( - "fmt" - "github.com/sirupsen/logrus" "github.com/yedf/dtm/common" ) @@ -10,7 +8,7 @@ import ( // Msg reliable msg type type Msg struct { MsgData - Server string + TransBase } // MsgData msg data @@ -34,13 +32,15 @@ func NewMsg(server string, gid string) *Msg { Gid: gid, TransType: "msg", }, - Server: server, + TransBase: TransBase{ + Dtm: server, + }, } } // Add add a new step func (s *Msg) Add(action string, postData interface{}) *Msg { - logrus.Printf("msg %s Add %s %v", s.Gid, action, postData) + logrus.Printf("msg %s Add %s %v", s.MsgData.Gid, action, postData) step := MsgStep{ Action: action, Data: common.MustMarshalString(postData), @@ -49,17 +49,13 @@ func (s *Msg) Add(action string, postData interface{}) *Msg { return s } -// Submit submit the msg -func (s *Msg) Submit() error { - logrus.Printf("committing %s body: %v", s.Gid, &s.MsgData) - resp, err := common.RestyClient.R().SetBody(&s.MsgData).Post(fmt.Sprintf("%s/submit", s.Server)) - return CheckDtmResponse(resp, err) -} - // Prepare prepare the msg func (s *Msg) Prepare(queryPrepared string) error { s.QueryPrepared = common.OrString(queryPrepared, s.QueryPrepared) - logrus.Printf("preparing %s body: %v", s.Gid, &s.MsgData) - resp, err := common.RestyClient.R().SetBody(&s.MsgData).Post(fmt.Sprintf("%s/prepare", s.Server)) - return CheckDtmResponse(resp, err) + return s.CallDtm(&s.MsgData, "prepare") +} + +// Submit submit the msg +func (s *Msg) Submit() error { + return s.CallDtm(&s.MsgData, "submit") } diff --git a/dtmcli/saga.go b/dtmcli/saga.go index c92d13d..b32a340 100644 --- a/dtmcli/saga.go +++ b/dtmcli/saga.go @@ -1,8 +1,6 @@ package dtmcli import ( - "fmt" - "github.com/sirupsen/logrus" "github.com/yedf/dtm/common" ) @@ -10,7 +8,7 @@ import ( // Saga struct of saga type Saga struct { SagaData - Server string + TransBase } // SagaData sage data @@ -34,13 +32,15 @@ func NewSaga(server string, gid string) *Saga { Gid: gid, TransType: "saga", }, - Server: server, + TransBase: TransBase{ + Dtm: server, + }, } } // Add add a saga step func (s *Saga) Add(action string, compensate string, postData interface{}) *Saga { - logrus.Printf("saga %s Add %s %s %v", s.Gid, action, compensate, postData) + logrus.Printf("saga %s Add %s %s %v", s.SagaData.Gid, action, compensate, postData) step := SagaStep{ Action: action, Compensate: compensate, @@ -52,7 +52,5 @@ func (s *Saga) Add(action string, compensate string, postData interface{}) *Saga // Submit submit the saga trans func (s *Saga) Submit() error { - logrus.Printf("committing %s body: %v", s.Gid, &s.SagaData) - resp, err := common.RestyClient.R().SetBody(&s.SagaData).Post(fmt.Sprintf("%s/submit", s.Server)) - return CheckDtmResponse(resp, err) + return s.CallDtm(&s.SagaData, "submit") } diff --git a/dtmcli/tcc.go b/dtmcli/tcc.go index 08a9037..ced6910 100644 --- a/dtmcli/tcc.go +++ b/dtmcli/tcc.go @@ -10,53 +10,49 @@ import ( // Tcc struct of tcc type Tcc struct { - IDGenerator - Dtm string Gid string + TransBase } // TccGlobalFunc type of global tcc call -type TccGlobalFunc func(tcc *Tcc) (interface{}, error) +type TccGlobalFunc func(tcc *Tcc) (*resty.Response, error) // TccGlobalTransaction begin a tcc global transaction // dtm dtm服务器地址 // gid 全局事务id // tccFunc tcc事务函数,里面会定义全局事务的分支 -func TccGlobalTransaction(dtm string, gid string, tccFunc TccGlobalFunc) (ret interface{}, rerr error) { +func TccGlobalTransaction(dtm string, gid string, tccFunc TccGlobalFunc) (rerr error) { data := &M{ "gid": gid, "trans_type": "tcc", } - tcc := &Tcc{Dtm: dtm, Gid: gid} - resp, err := common.RestyClient.R().SetBody(data).Post(tcc.Dtm + "/prepare") - if IsFailure(resp, err) { - return resp, err + tcc := &Tcc{TransBase: TransBase{Dtm: dtm}, Gid: gid} + rerr = tcc.CallDtm(data, "prepare") + if rerr != nil { + return rerr } // 小概率情况下,prepare成功了,但是由于网络状况导致上面Failure,那么不执行下面defer的内容,等待超时后再回滚标记事务失败,也没有问题 defer func() { - var x interface{} - if x = recover(); x != nil || IsFailure(ret, rerr) { - resp, err = common.RestyClient.R().SetBody(data).Post(dtm + "/abort") - } else { - resp, err = common.RestyClient.R().SetBody(data).Post(dtm + "/submit") - } - if IsFailure(resp, err) { - common.RedLogf("submitting or abort global transaction error: %v resp: %s", err, resp.String()) + x := recover() + operation := common.If(x == nil && rerr == nil, "submit", "abort").(string) + err := tcc.CallDtm(data, operation) + if rerr == nil { + rerr = err } if x != nil { panic(x) } }() - ret, rerr = tccFunc(tcc) + resp, rerr := tccFunc(tcc) + rerr = CheckResponse(resp, rerr) return } // TccFromReq tcc from request info func TccFromReq(c *gin.Context) (*Tcc, error) { tcc := &Tcc{ - Dtm: c.Query("dtm"), - Gid: c.Query("gid"), - IDGenerator: IDGenerator{parentID: c.Query("branch_id")}, + TransBase: *TransBaseFromReq(c), + Gid: c.Query("gid"), } if tcc.Dtm == "" || tcc.Gid == "" { return nil, fmt.Errorf("bad tcc info. dtm: %s, gid: %s", tcc.Dtm, tcc.Gid) @@ -68,22 +64,20 @@ func TccFromReq(c *gin.Context) (*Tcc, error) { // 函数首先注册子事务的所有分支,成功后调用try分支,返回try分支的调用结果 func (t *Tcc) CallBranch(body interface{}, tryURL string, confirmURL string, cancelURL string) (*resty.Response, error) { branchID := t.NewBranchID() - resp, err := common.RestyClient.R(). - SetBody(&M{ - "gid": t.Gid, - "branch_id": branchID, - "trans_type": "tcc", - "status": "prepared", - "data": string(common.MustMarshal(body)), - "try": tryURL, - "confirm": confirmURL, - "cancel": cancelURL, - }). - Post(t.Dtm + "/registerTccBranch") - if IsFailure(resp, err) { - return resp, err + err := t.CallDtm(&M{ + "gid": t.Gid, + "branch_id": branchID, + "trans_type": "tcc", + "status": "prepared", + "data": string(common.MustMarshal(body)), + "try": tryURL, + "confirm": confirmURL, + "cancel": cancelURL, + }, "registerTccBranch") + if err != nil { + return nil, err } - return common.RestyClient.R(). + resp, err := common.RestyClient.R(). SetBody(body). SetQueryParams(common.MS{ "dtm": t.Dtm, @@ -93,4 +87,5 @@ func (t *Tcc) CallBranch(body interface{}, tryURL string, confirmURL string, can "branch_type": "try", }). Post(tryURL) + return resp, CheckResponse(resp, err) } diff --git a/dtmcli/types.go b/dtmcli/types.go index 29b8806..0aff9fa 100644 --- a/dtmcli/types.go +++ b/dtmcli/types.go @@ -1,9 +1,11 @@ package dtmcli import ( + "errors" "fmt" "strings" + "github.com/gin-gonic/gin" "github.com/go-resty/resty/v2" "github.com/yedf/dtm/common" ) @@ -18,30 +20,28 @@ func MustGenGid(server string) string { return res["gid"] } -// IsFailure 如果err非空,或者ret是http的响应且包含FAILURE,那么返回true。此时认为业务调用失败 -func IsFailure(res interface{}, err error) bool { +// CheckResponse 检查Response,返回错误 +func CheckResponse(resp *resty.Response, err error) error { + if err == nil && resp != nil { + if resp.IsError() { + return errors.New(resp.String()) + } else if strings.Contains(resp.String(), "FAILURE") { + return ErrFailure + } + } + return err +} + +// CheckResult 检查Result,返回错误 +func CheckResult(res interface{}, err error) error { resp, ok := res.(*resty.Response) - return err != nil || // 包含错误 - ok && (resp.IsError() || strings.Contains(resp.String(), "FAILURE")) || // resp包含failure - !ok && res != nil && strings.Contains(common.MustMarshalString(res), "FAILURE") // 结果中包含failure -} - -// PanicIfFailure 如果err非空,或者ret是http的响应且包含FAILURE,那么Panic。此时认为业务调用失败 -func PanicIfFailure(res interface{}, err error) { - if IsFailure(res, err) { - panic(fmt.Errorf("dtm failure ret: %v err %v", res, err)) + if ok { + return CheckResponse(resp, err) } -} - -// CheckDtmResponse check the response of dtm, if not ok ,generate error -func CheckDtmResponse(resp *resty.Response, err error) error { - if err != nil { - return err + if res != nil && strings.Contains(common.MustMarshalString(res), "FAILURE") { + return ErrFailure } - if !strings.Contains(resp.String(), "SUCCESS") || resp.IsError() { - return fmt.Errorf("dtm response failed: %s", resp.String()) - } - return nil + return err } // IDGenerator used to generate a branch id @@ -61,3 +61,52 @@ func (g *IDGenerator) NewBranchID() string { g.branchID = g.branchID + 1 return g.parentID + fmt.Sprintf("%02d", g.branchID) } + +// TransResult dtm 返回的结果 +type TransResult struct { + DtmResult string `json:"dtm_result"` + Message string +} + +// TransBase 事务的基础类 +type TransBase struct { + IDGenerator + Dtm string + // WaitResult 是否等待全局事务的最终结果 + WaitResult bool +} + +// TransBaseFromReq construct xa info from request +func TransBaseFromReq(c *gin.Context) *TransBase { + return &TransBase{ + IDGenerator: IDGenerator{parentID: c.Query("branch_id")}, + Dtm: c.Query("dtm"), + } +} + +// CallDtm 调用dtm服务器,返回事务的状态 +func (tb *TransBase) CallDtm(body interface{}, operation string) error { + params := common.MS{} + if tb.WaitResult { + params["wait_result"] = "1" + } + resp, err := common.RestyClient.R().SetQueryParams(params). + SetResult(&TransResult{}).SetBody(body).Post(fmt.Sprintf("%s/%s", tb.Dtm, operation)) + if err != nil { + return err + } + tr := resp.Result().(*TransResult) + if tr.DtmResult == "FAILURE" { + return errors.New("FAILURE: " + tr.Message) + } + return nil +} + +// ErrFailure 表示返回失败,要求回滚 +var ErrFailure = errors.New("transaction FAILURE") + +// ResultSuccess 表示返回成功,可以进行下一步 +var ResultSuccess = common.M{"dtm_result": "SUCCESS"} + +// ResultFailure 表示返回失败,要求回滚 +var ResultFailure = common.M{"dtm_result": "FAILURE"} diff --git a/dtmcli/types_test.go b/dtmcli/types_test.go index d47e232..97a3d88 100644 --- a/dtmcli/types_test.go +++ b/dtmcli/types_test.go @@ -1,7 +1,6 @@ package dtmcli import ( - "fmt" "net/url" "testing" @@ -22,8 +21,4 @@ func TestTypes(t *testing.T) { assert.Error(t, err) _, err = TransInfoFromQuery(url.Values{}) assert.Error(t, err) - - err2 := fmt.Errorf("an error") - err3 := CheckDtmResponse(nil, err2) - assert.Error(t, err2, err3) } diff --git a/dtmcli/xa.go b/dtmcli/xa.go index 39f895d..860f093 100644 --- a/dtmcli/xa.go +++ b/dtmcli/xa.go @@ -16,7 +16,7 @@ type M = map[string]interface{} var e2p = common.E2P // XaGlobalFunc type of xa global function -type XaGlobalFunc func(xa *Xa) (interface{}, error) +type XaGlobalFunc func(xa *Xa) (*resty.Response, error) // XaLocalFunc type of xa local function type XaLocalFunc func(db *sql.DB, xa *Xa) (interface{}, error) @@ -33,16 +33,13 @@ type XaClient struct { // Xa xa transaction type Xa struct { - IDGenerator Gid string + TransBase } // XaFromReq construct xa info from request func XaFromReq(c *gin.Context) *Xa { - return &Xa{ - Gid: c.Query("gid"), - IDGenerator: IDGenerator{parentID: c.Query("branch_id")}, - } + return &Xa{TransBase: *TransBaseFromReq(c), Gid: c.Query("gid")} } // NewXaClient construct a xa client @@ -66,29 +63,26 @@ func (xc *XaClient) HandleCallback(gid string, branchID string, action string) ( defer db.Close() xaID := gid + "-" + branchID _, err := common.SdbExec(db, fmt.Sprintf("xa %s '%s'", action, xaID)) - return M{"dtm_result": "SUCCESS"}, err + return ResultSuccess, err } // XaLocalTransaction start a xa local transaction func (xc *XaClient) XaLocalTransaction(c *gin.Context, xaFunc XaLocalFunc) (ret interface{}, rerr error) { xa := XaFromReq(c) + xa.Dtm = xc.Server branchID := xa.NewBranchID() xaBranch := xa.Gid + "-" + branchID db := common.SdbAlone(xc.Conf) defer func() { db.Close() }() defer func() { - var x interface{} + x := recover() _, err := common.SdbExec(db, fmt.Sprintf("XA end '%s'", xaBranch)) - if err != nil { - common.RedLogf("sql db exec error: %v", err) - } - if x = recover(); x != nil || IsFailure(ret, rerr) { - } else { + if x == nil && rerr == nil && err == nil { _, err = common.SdbExec(db, fmt.Sprintf("XA prepare '%s'", xaBranch)) } - if err != nil { - common.RedLogf("sql db exec error: %v", err) + if rerr == nil { + rerr = err } if x != nil { panic(x) @@ -99,49 +93,47 @@ func (xc *XaClient) XaLocalTransaction(c *gin.Context, xaFunc XaLocalFunc) (ret return } ret, rerr = xaFunc(db, xa) - if IsFailure(ret, rerr) { + rerr = CheckResult(ret, rerr) + if rerr != nil { return } - ret, rerr = common.RestyClient.R(). - SetBody(&M{"gid": xa.Gid, "branch_id": branchID, "trans_type": "xa", "status": "prepared", "url": xc.CallbackURL}). - Post(xc.Server + "/registerXaBranch") + rerr = xa.CallDtm(&M{"gid": xa.Gid, "branch_id": branchID, "trans_type": "xa", "status": "prepared", "url": xc.CallbackURL}, "registerXaBranch") return } // XaGlobalTransaction start a xa global transaction -func (xc *XaClient) XaGlobalTransaction(gid string, xaFunc XaGlobalFunc) (ret interface{}, rerr error) { - xa := Xa{IDGenerator: IDGenerator{}, Gid: gid} +func (xc *XaClient) XaGlobalTransaction(gid string, xaFunc XaGlobalFunc) (rerr error) { + xa := Xa{TransBase: TransBase{IDGenerator: IDGenerator{}, Dtm: xc.Server}, Gid: gid} data := &M{ "gid": gid, "trans_type": "xa", } - resp, err := common.RestyClient.R().SetBody(data).Post(xc.Server + "/prepare") - if IsFailure(resp, err) { - return resp, err + rerr = xa.CallDtm(data, "prepare") + if rerr != nil { + return } + var resp *resty.Response // 小概率情况下,prepare成功了,但是由于网络状况导致上面Failure,那么不执行下面defer的内容,等待超时后再回滚标记事务失败,也没有问题 defer func() { - var x interface{} - if x = recover(); x != nil || IsFailure(ret, rerr) { - resp, err = common.RestyClient.R().SetBody(data).Post(xc.Server + "/abort") - } else { - resp, err = common.RestyClient.R().SetBody(data).Post(xc.Server + "/submit") - } - if IsFailure(resp, err) { - common.RedLogf("submitting or abort global transaction error: %v resp: %s", err, resp.String()) + x := recover() + operation := common.If(x != nil || rerr != nil, "abort", "submit").(string) + err := xa.CallDtm(data, operation) + if rerr == nil { // 如果用户函数没有返回错误,那么返回dtm的 + rerr = err } if x != nil { panic(x) } }() - ret, rerr = xaFunc(&xa) + resp, rerr = xaFunc(&xa) + rerr = CheckResponse(resp, rerr) return } // CallBranch call a xa branch func (x *Xa) CallBranch(body interface{}, url string) (*resty.Response, error) { branchID := x.NewBranchID() - return common.RestyClient.R(). + resp, err := common.RestyClient.R(). SetBody(body). SetQueryParams(common.MS{ "gid": x.Gid, @@ -150,4 +142,5 @@ func (x *Xa) CallBranch(body interface{}, url string) (*resty.Response, error) { "branch_type": "action", }). Post(url) + return resp, CheckResponse(resp, err) } diff --git a/dtmsvr/api.go b/dtmsvr/api.go index 0ca9511..7c9af86 100644 --- a/dtmsvr/api.go +++ b/dtmsvr/api.go @@ -6,6 +6,7 @@ import ( "github.com/gin-gonic/gin" "github.com/yedf/dtm/common" + "github.com/yedf/dtm/dtmcli" "gorm.io/gorm" "gorm.io/gorm/clause" ) @@ -28,7 +29,7 @@ func prepare(c *gin.Context) (interface{}, error) { t := TransFromContext(c) t.Status = "prepared" t.saveNew(dbGet()) - return M{"dtm_result": "SUCCESS"}, nil + return dtmcli.ResultSuccess, nil } func submit(c *gin.Context) (interface{}, error) { @@ -40,8 +41,7 @@ func submit(c *gin.Context) (interface{}, error) { } t.Status = "submitted" t.saveNew(db) - go t.Process(db) - return M{"dtm_result": "SUCCESS"}, nil + return t.Process(db, c.Query("wait_result") == "true" || c.Query("wait_result") == "1"), nil } func abort(c *gin.Context) (interface{}, error) { @@ -51,8 +51,7 @@ func abort(c *gin.Context) (interface{}, error) { if t.TransType != "xa" && t.TransType != "tcc" || dbt.Status != "prepared" && dbt.Status != "aborting" { return M{"dtm_result": "FAILURE", "message": fmt.Sprintf("trans type: %s current status %s, cannot abort", dbt.TransType, dbt.Status)}, nil } - go dbt.Process(db) - return M{"dtm_result": "SUCCESS"}, nil + return dbt.Process(db, c.Query("wait_result") == "true" || c.Query("wait_result") == "1"), nil } func registerXaBranch(c *gin.Context) (interface{}, error) { @@ -73,7 +72,7 @@ func registerXaBranch(c *gin.Context) (interface{}, error) { e2p(err) global := TransGlobal{Gid: branch.Gid} global.touch(db, config.TransCronInterval) - return M{"dtm_result": "SUCCESS"}, nil + return dtmcli.ResultSuccess, nil } func registerTccBranch(c *gin.Context) (interface{}, error) { @@ -104,7 +103,7 @@ func registerTccBranch(c *gin.Context) (interface{}, error) { e2p(err) global := TransGlobal{Gid: branch.Gid} global.touch(dbGet(), config.TransCronInterval) - return M{"dtm_result": "SUCCESS"}, nil + return dtmcli.ResultSuccess, nil } func query(c *gin.Context) (interface{}, error) { diff --git a/dtmsvr/cron.go b/dtmsvr/cron.go index 64a427d..a851918 100644 --- a/dtmsvr/cron.go +++ b/dtmsvr/cron.go @@ -1,6 +1,7 @@ package dtmsvr import ( + "fmt" "math" "math/rand" "runtime/debug" @@ -12,7 +13,7 @@ import ( // CronTransOnce cron expired trans. use expireIn as expire time func CronTransOnce(expireIn time.Duration) bool { - defer handlePanic() + defer handlePanic(nil) trans := lockOneTrans(expireIn) if trans == nil { return false @@ -20,7 +21,7 @@ func CronTransOnce(expireIn time.Duration) bool { if TransProcessedTestChan != nil { defer WaitTransProcessed(trans.Gid) } - trans.Process(dbGet()) + trans.Process(dbGet(), true) return true } @@ -52,9 +53,12 @@ func lockOneTrans(expireIn time.Duration) *TransGlobal { return &trans } -func handlePanic() { +func handlePanic(perr *error) { if err := recover(); err != nil { - common.RedLogf("----panic %s handlered\n%s", err.(error).Error(), string(debug.Stack())) + common.RedLogf("----panic %v handlered\n%s", err, string(debug.Stack())) + if perr != nil { + *perr = fmt.Errorf("dtm panic: %v", err) + } } } diff --git a/dtmsvr/trans.go b/dtmsvr/trans.go index 5d2beee..d2166a5 100644 --- a/dtmsvr/trans.go +++ b/dtmsvr/trans.go @@ -7,6 +7,7 @@ import ( "github.com/gin-gonic/gin" "github.com/sirupsen/logrus" "github.com/yedf/dtm/common" + "github.com/yedf/dtm/dtmcli" "gorm.io/gorm" "gorm.io/gorm/clause" ) @@ -110,8 +111,24 @@ func (t *TransGlobal) getProcessor() transProcessor { } // Process process global transaction once -func (t *TransGlobal) Process(db *common.DB) { - defer handlePanic() +func (t *TransGlobal) Process(db *common.DB, waitResult bool) common.M { + if !waitResult { + go t.processInner(db) + return dtmcli.ResultSuccess + } + submitting := t.Status == "submitted" + err := t.processInner(db) + if err != nil { + return common.M{"dtm_result": "FAILURE", "message": err.Error()} + } + if submitting && t.Status != "succeed" { + return common.M{"dtm_result": "FAILURE", "message": "trans failed by user"} + } + return dtmcli.ResultSuccess +} + +func (t *TransGlobal) processInner(db *common.DB) (rerr error) { + defer handlePanic(&rerr) defer func() { if TransProcessedTestChan != nil { logrus.Printf("processed: %s", t.Gid) @@ -126,6 +143,7 @@ func (t *TransGlobal) Process(db *common.DB) { branches := []TransBranch{} db.Must().Where("gid=?", t.Gid).Order("id asc").Find(&branches) t.getProcessor().ProcessOnce(db, branches) + return } func (t *TransGlobal) getBranchParams(branch *TransBranch) common.MS { diff --git a/dtmsvr/trans_saga_wait_test.go b/dtmsvr/trans_saga_wait_test.go new file mode 100644 index 0000000..4472ba6 --- /dev/null +++ b/dtmsvr/trans_saga_wait_test.go @@ -0,0 +1,50 @@ +package dtmsvr + +import ( + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/yedf/dtm/examples" +) + +func TestSagaWait(t *testing.T) { + + sagaNormalWait(t) + sagaCommittedPendingWait(t) + sagaRollbackWait(t) +} + +func sagaNormalWait(t *testing.T) { + saga := genSaga("gid-noramlSagaWait", false, false) + saga.WaitResult = true + err := saga.Submit() + assert.Nil(t, err) + WaitTransProcessed(saga.Gid) + assert.Equal(t, []string{"prepared", "succeed", "prepared", "succeed"}, getBranchesStatus(saga.Gid)) + assert.Equal(t, "succeed", getTransStatus(saga.Gid)) + transQuery(t, saga.Gid) +} + +func sagaCommittedPendingWait(t *testing.T) { + saga := genSaga("gid-committedPendingWait", false, false) + examples.MainSwitch.TransOutResult.SetOnce("PENDING") + saga.WaitResult = true + err := saga.Submit() + assert.Error(t, err) + WaitTransProcessed(saga.Gid) + assert.Equal(t, []string{"prepared", "prepared", "prepared", "prepared"}, getBranchesStatus(saga.Gid)) + CronTransOnce(60 * time.Second) + assert.Equal(t, []string{"prepared", "succeed", "prepared", "succeed"}, getBranchesStatus(saga.Gid)) + assert.Equal(t, "succeed", getTransStatus(saga.Gid)) +} + +func sagaRollbackWait(t *testing.T) { + saga := genSaga("gid-rollbackSaga2Wait", false, true) + saga.WaitResult = true + err := saga.Submit() + assert.Error(t, err) + WaitTransProcessed(saga.Gid) + assert.Equal(t, "failed", getTransStatus(saga.Gid)) + assert.Equal(t, []string{"succeed", "succeed", "succeed", "failed"}, getBranchesStatus(saga.Gid)) +} diff --git a/dtmsvr/trans_tcc_barrier_test.go b/dtmsvr/trans_tcc_barrier_test.go index 5375c86..e1eb9d5 100644 --- a/dtmsvr/trans_tcc_barrier_test.go +++ b/dtmsvr/trans_tcc_barrier_test.go @@ -7,6 +7,7 @@ import ( "time" "github.com/gin-gonic/gin" + "github.com/go-resty/resty/v2" "github.com/sirupsen/logrus" "github.com/stretchr/testify/assert" "github.com/yedf/dtm/common" @@ -23,24 +24,24 @@ func TestTccBarrier(t *testing.T) { func tccBarrierRollback(t *testing.T) { gid := "tccBarrierRollback" - resp, err := dtmcli.TccGlobalTransaction(DtmServer, gid, func(tcc *dtmcli.Tcc) (interface{}, error) { - resp, err := tcc.CallBranch(&examples.TransReq{Amount: 30}, Busi+"/TccBTransOutTry", Busi+"/TccBTransOutConfirm", Busi+"/TccBTransOutCancel") - assert.True(t, !dtmcli.IsFailure(resp, err)) + err := dtmcli.TccGlobalTransaction(DtmServer, gid, func(tcc *dtmcli.Tcc) (*resty.Response, error) { + _, err := tcc.CallBranch(&examples.TransReq{Amount: 30}, Busi+"/TccBTransOutTry", Busi+"/TccBTransOutConfirm", Busi+"/TccBTransOutCancel") + assert.Nil(t, err) return tcc.CallBranch(&examples.TransReq{Amount: 30, TransInResult: "FAILURE"}, Busi+"/TccBTransInTry", Busi+"/TccBTransInConfirm", Busi+"/TccBTransInCancel") }) - assert.True(t, dtmcli.IsFailure(resp, err)) + assert.Error(t, err) WaitTransProcessed(gid) assert.Equal(t, "failed", getTransStatus(gid)) } func tccBarrierNormal(t *testing.T) { gid := "tccBarrierNormal" - resp, err := dtmcli.TccGlobalTransaction(DtmServer, gid, func(tcc *dtmcli.Tcc) (interface{}, error) { - resp, err := tcc.CallBranch(&examples.TransReq{Amount: 30}, Busi+"/TccBTransOutTry", Busi+"/TccBTransOutConfirm", Busi+"/TccBTransOutCancel") - assert.True(t, !dtmcli.IsFailure(resp, err)) + err := dtmcli.TccGlobalTransaction(DtmServer, gid, func(tcc *dtmcli.Tcc) (*resty.Response, error) { + _, err := tcc.CallBranch(&examples.TransReq{Amount: 30}, Busi+"/TccBTransOutTry", Busi+"/TccBTransOutConfirm", Busi+"/TccBTransOutCancel") + assert.Nil(t, err) return tcc.CallBranch(&examples.TransReq{Amount: 30}, Busi+"/TccBTransInTry", Busi+"/TccBTransInConfirm", Busi+"/TccBTransInCancel") }) - assert.True(t, !dtmcli.IsFailure(resp, err)) + assert.Nil(t, err) WaitTransProcessed(gid) assert.Equal(t, "succeed", getTransStatus(gid)) } @@ -49,7 +50,7 @@ func tccBarrierDisorder(t *testing.T) { timeoutChan := make(chan string, 2) finishedChan := make(chan string, 2) gid := "tccBarrierDisorder" - _, err := dtmcli.TccGlobalTransaction(DtmServer, gid, func(tcc *dtmcli.Tcc) (interface{}, error) { + err := dtmcli.TccGlobalTransaction(DtmServer, gid, func(tcc *dtmcli.Tcc) (*resty.Response, error) { body := &examples.TransReq{Amount: 30} tryURL := Busi + "/TccBTransOutTry" confirmURL := Busi + "/TccBTransOutConfirm" @@ -68,23 +69,21 @@ func tccBarrierDisorder(t *testing.T) { return res, err })) // 注册子事务 - r, err := common.RestyClient.R(). - SetBody(&M{ - "gid": tcc.Gid, - "branch_id": branchID, - "trans_type": "tcc", - "status": "prepared", - "data": string(common.MustMarshal(body)), - "try": tryURL, - "confirm": confirmURL, - "cancel": cancelURL, - }). - Post(tcc.Dtm + "/registerTccBranch") - assert.True(t, !dtmcli.IsFailure(r, err)) + err := tcc.CallDtm(M{ + "gid": tcc.Gid, + "branch_id": branchID, + "trans_type": "tcc", + "status": "prepared", + "data": string(common.MustMarshal(body)), + "try": tryURL, + "confirm": confirmURL, + "cancel": cancelURL, + }, "registerTccBranch") + assert.Nil(t, err) go func() { logrus.Printf("sleeping to wait for tcc try timeout") <-timeoutChan - r, _ = common.RestyClient.R(). + r, _ := common.RestyClient.R(). SetBody(body). SetQueryParams(common.MS{ "dtm": tcc.Dtm, diff --git a/dtmsvr/trans_tcc_test.go b/dtmsvr/trans_tcc_test.go index 1a6666f..d5383df 100644 --- a/dtmsvr/trans_tcc_test.go +++ b/dtmsvr/trans_tcc_test.go @@ -4,6 +4,7 @@ import ( "testing" "time" + "github.com/go-resty/resty/v2" "github.com/stretchr/testify/assert" "github.com/yedf/dtm/dtmcli" "github.com/yedf/dtm/examples" @@ -18,26 +19,24 @@ func TestTcc(t *testing.T) { func tccNormal(t *testing.T) { data := &examples.TransReq{Amount: 30} gid := "tccNormal" - ret, err := dtmcli.TccGlobalTransaction(examples.DtmServer, gid, func(tcc *dtmcli.Tcc) (interface{}, error) { - resp, err := tcc.CallBranch(data, Busi+"/TransOut", Busi+"/TransOutConfirm", Busi+"/TransOutRevert") - if dtmcli.IsFailure(resp, err) { - return resp, err - } + err := dtmcli.TccGlobalTransaction(examples.DtmServer, gid, func(tcc *dtmcli.Tcc) (*resty.Response, error) { + _, err := tcc.CallBranch(data, Busi+"/TransOut", Busi+"/TransOutConfirm", Busi+"/TransOutRevert") + assert.Nil(t, err) return tcc.CallBranch(data, Busi+"/TransIn", Busi+"/TransInConfirm", Busi+"/TransInRevert") }) - dtmcli.PanicIfFailure(ret, err) + assert.Nil(t, err) } func tccRollback(t *testing.T) { gid := "tccRollback" data := &examples.TransReq{Amount: 30, TransInResult: "FAILURE"} - resp, err := dtmcli.TccGlobalTransaction(examples.DtmServer, gid, func(tcc *dtmcli.Tcc) (interface{}, error) { - resp, rerr := tcc.CallBranch(data, Busi+"/TransOut", Busi+"/TransOutConfirm", Busi+"/TransOutRevert") - assert.True(t, !dtmcli.IsFailure(resp, rerr)) + err := dtmcli.TccGlobalTransaction(examples.DtmServer, gid, func(tcc *dtmcli.Tcc) (*resty.Response, error) { + _, rerr := tcc.CallBranch(data, Busi+"/TransOut", Busi+"/TransOutConfirm", Busi+"/TransOutRevert") + assert.Nil(t, rerr) examples.MainSwitch.TransOutRevertResult.SetOnce("PENDING") return tcc.CallBranch(data, Busi+"/TransIn", Busi+"/TransInConfirm", Busi+"/TransInRevert") }) - assert.True(t, dtmcli.IsFailure(resp, err)) + assert.Error(t, err) WaitTransProcessed(gid) assert.Equal(t, "aborting", getTransStatus(gid)) CronTransOnce(60 * time.Second) diff --git a/dtmsvr/trans_xa.go b/dtmsvr/trans_xa.go index 1e33353..819418c 100644 --- a/dtmsvr/trans_xa.go +++ b/dtmsvr/trans_xa.go @@ -21,7 +21,7 @@ func (t *transXaProcessor) GenBranches() []TransBranch { func (t *transXaProcessor) ExecBranch(db *common.DB, branch *TransBranch) { resp, err := common.RestyClient.R().SetQueryParams(common.MS{ "branch_id": branch.BranchID, - "action": common.If(t.Status == "prepared", "rollback", "commit").(string), + "action": common.If(t.Status == "prepared" || t.Status == "aborting", "rollback", "commit").(string), "gid": branch.Gid, }).Post(branch.URL) e2p(err) diff --git a/dtmsvr/trans_xa_test.go b/dtmsvr/trans_xa_test.go index 846ca39..5dee503 100644 --- a/dtmsvr/trans_xa_test.go +++ b/dtmsvr/trans_xa_test.go @@ -4,6 +4,7 @@ import ( "fmt" "testing" + "github.com/go-resty/resty/v2" "github.com/stretchr/testify/assert" "github.com/yedf/dtm/dtmcli" "github.com/yedf/dtm/examples" @@ -19,7 +20,8 @@ func TestXa(t *testing.T) { } func xaLocalError(t *testing.T) { - _, err := examples.XaClient.XaGlobalTransaction("xaLocalError", func(xa *dtmcli.Xa) (interface{}, error) { + xc := examples.XaClient + err := xc.XaGlobalTransaction("xaLocalError", func(xa *dtmcli.Xa) (*resty.Response, error) { return nil, fmt.Errorf("an error") }) assert.Error(t, err, fmt.Errorf("an error")) @@ -28,15 +30,15 @@ func xaLocalError(t *testing.T) { func xaNormal(t *testing.T) { xc := examples.XaClient gid := "xaNormal" - res, err := xc.XaGlobalTransaction(gid, func(xa *dtmcli.Xa) (interface{}, error) { + err := xc.XaGlobalTransaction(gid, func(xa *dtmcli.Xa) (*resty.Response, error) { req := examples.GenTransReq(30, false, false) resp, err := xa.CallBranch(req, examples.Busi+"/TransOutXa") - if dtmcli.IsFailure(resp, err) { + if err != nil { return resp, err } return xa.CallBranch(req, examples.Busi+"/TransInXa") }) - dtmcli.PanicIfFailure(res, err) + assert.Equal(t, nil, err) WaitTransProcessed(gid) assert.Equal(t, []string{"prepared", "succeed", "prepared", "succeed"}, getBranchesStatus(gid)) } @@ -44,15 +46,15 @@ func xaNormal(t *testing.T) { func xaRollback(t *testing.T) { xc := examples.XaClient gid := "xaRollback" - res, err := xc.XaGlobalTransaction(gid, func(xa *dtmcli.Xa) (interface{}, error) { + err := xc.XaGlobalTransaction(gid, func(xa *dtmcli.Xa) (*resty.Response, error) { req := &examples.TransReq{Amount: 30, TransInResult: "FAILURE"} resp, err := xa.CallBranch(req, examples.Busi+"/TransOutXa") - if dtmcli.IsFailure(resp, err) { + if err != nil { return resp, err } return xa.CallBranch(req, examples.Busi+"/TransInXa") }) - assert.True(t, dtmcli.IsFailure(res, err)) + assert.Error(t, err) WaitTransProcessed(gid) assert.Equal(t, []string{"succeed", "prepared"}, getBranchesStatus(gid)) assert.Equal(t, "failed", getTransStatus(gid)) diff --git a/examples/main_saga_barrier.go b/examples/main_saga_barrier.go index da2b61b..9c50fa8 100644 --- a/examples/main_saga_barrier.go +++ b/examples/main_saga_barrier.go @@ -33,7 +33,7 @@ func SagaBarrierAddRoute(app *gin.Engine) { func sagaBarrierAdjustBalance(sdb *sql.Tx, uid int, amount int) (interface{}, error) { _, err := common.StxExec(sdb, "update dtm_busi.user_account set balance = balance + ? where user_id = ?", amount, uid) - return common.MS{"dtm_result": "SUCCESS"}, err + return dtmcli.ResultSuccess, err } diff --git a/examples/main_tcc.go b/examples/main_tcc.go index fd7f182..7e483bd 100644 --- a/examples/main_tcc.go +++ b/examples/main_tcc.go @@ -2,6 +2,7 @@ package examples import ( "github.com/gin-gonic/gin" + "github.com/go-resty/resty/v2" "github.com/sirupsen/logrus" "github.com/yedf/dtm/common" "github.com/yedf/dtm/dtmcli" @@ -20,14 +21,14 @@ func TccSetup(app *gin.Engine) { // TccFireRequestNested 1 func TccFireRequestNested() string { gid := dtmcli.MustGenGid(DtmServer) - ret, err := dtmcli.TccGlobalTransaction(DtmServer, gid, func(tcc *dtmcli.Tcc) (interface{}, error) { + err := dtmcli.TccGlobalTransaction(DtmServer, gid, func(tcc *dtmcli.Tcc) (*resty.Response, error) { resp, err := tcc.CallBranch(&TransReq{Amount: 30}, Busi+"/TransOut", Busi+"/TransOutConfirm", Busi+"/TransOutRevert") - if dtmcli.IsFailure(resp, err) { + if err != nil { return resp, err } return tcc.CallBranch(&TransReq{Amount: 30}, Busi+"/TransInTccParent", Busi+"/TransInConfirm", Busi+"/TransInRevert") }) - dtmcli.PanicIfFailure(ret, err) + e2p(err) return gid } @@ -35,13 +36,13 @@ func TccFireRequestNested() string { func TccFireRequest() string { logrus.Printf("tcc simple transaction begin") gid := dtmcli.MustGenGid(DtmServer) - ret, err := dtmcli.TccGlobalTransaction(DtmServer, gid, func(tcc *dtmcli.Tcc) (interface{}, error) { + err := dtmcli.TccGlobalTransaction(DtmServer, gid, func(tcc *dtmcli.Tcc) (*resty.Response, error) { resp, err := tcc.CallBranch(&TransReq{Amount: 30}, Busi+"/TransOut", Busi+"/TransOutConfirm", Busi+"/TransOutRevert") - if dtmcli.IsFailure(resp, err) { + if err != nil { return resp, err } return tcc.CallBranch(&TransReq{Amount: 30}, Busi+"/TransIn", Busi+"/TransInConfirm", Busi+"/TransInRevert") }) - dtmcli.PanicIfFailure(ret, err) + e2p(err) return gid } diff --git a/examples/main_tcc_barrier.go b/examples/main_tcc_barrier.go index c7d973e..1f06d6e 100644 --- a/examples/main_tcc_barrier.go +++ b/examples/main_tcc_barrier.go @@ -5,6 +5,7 @@ import ( "fmt" "github.com/gin-gonic/gin" + "github.com/go-resty/resty/v2" "github.com/sirupsen/logrus" "github.com/yedf/dtm/common" "github.com/yedf/dtm/dtmcli" @@ -14,14 +15,14 @@ import ( func TccBarrierFireRequest() string { logrus.Printf("tcc transaction begin") gid := dtmcli.MustGenGid(DtmServer) - ret, err := dtmcli.TccGlobalTransaction(DtmServer, gid, func(tcc *dtmcli.Tcc) (interface{}, error) { + err := dtmcli.TccGlobalTransaction(DtmServer, gid, func(tcc *dtmcli.Tcc) (*resty.Response, error) { resp, err := tcc.CallBranch(&TransReq{Amount: 30}, Busi+"/TccBTransOutTry", Busi+"/TccBTransOutConfirm", Busi+"/TccBTransOutCancel") - if dtmcli.IsFailure(resp, err) { + if err != nil { return resp, err } return tcc.CallBranch(&TransReq{Amount: 30}, Busi+"/TccBTransInTry", Busi+"/TccBTransInConfirm", Busi+"/TccBTransInCancel") }) - dtmcli.PanicIfFailure(ret, err) + e2p(err) return gid } @@ -55,7 +56,7 @@ func adjustBalance(sdb *sql.Tx, uid int, amount int) (interface{}, error) { if err == nil && affected == 0 { return nil, fmt.Errorf("update 0 rows") } - return common.MS{"dtm_result": "SUCCESS"}, err + return dtmcli.ResultSuccess, err } // TCC下,转入 diff --git a/examples/main_xa.go b/examples/main_xa.go index aa0a859..b3e4a75 100644 --- a/examples/main_xa.go +++ b/examples/main_xa.go @@ -4,6 +4,7 @@ import ( "database/sql" "github.com/gin-gonic/gin" + "github.com/go-resty/resty/v2" "github.com/yedf/dtm/common" "github.com/yedf/dtm/dtmcli" ) @@ -27,33 +28,33 @@ func XaSetup(app *gin.Engine) { // XaFireRequest 注册全局XA事务,调用XA的分支 func XaFireRequest() string { gid := dtmcli.MustGenGid(DtmServer) - res, err := XaClient.XaGlobalTransaction(gid, func(xa *dtmcli.Xa) (interface{}, error) { + err := XaClient.XaGlobalTransaction(gid, func(xa *dtmcli.Xa) (*resty.Response, error) { resp, err := xa.CallBranch(&TransReq{Amount: 30}, Busi+"/TransOutXa") - if dtmcli.IsFailure(resp, err) { + if err != nil { return resp, err } return xa.CallBranch(&TransReq{Amount: 30}, Busi+"/TransInXa") }) - dtmcli.PanicIfFailure(res, err) + e2p(err) return gid } func xaTransIn(c *gin.Context) (interface{}, error) { return XaClient.XaLocalTransaction(c, func(db *sql.DB, xa *dtmcli.Xa) (interface{}, error) { if reqFrom(c).TransInResult == "FAILURE" { - return M{"dtm_result": "FAILURE"}, nil + return dtmcli.ResultFailure, nil } _, err := common.SdbExec(db, "update dtm_busi.user_account set balance=balance+? where user_id=?", reqFrom(c).Amount, 2) - return M{"dtm_result": "SUCCESS"}, err + return dtmcli.ResultSuccess, err }) } func xaTransOut(c *gin.Context) (interface{}, error) { return XaClient.XaLocalTransaction(c, func(db *sql.DB, xa *dtmcli.Xa) (interface{}, error) { if reqFrom(c).TransOutResult == "FAILURE" { - return M{"dtm_result": "FAILURE"}, nil + return dtmcli.ResultFailure, nil } _, err := common.SdbExec(db, "update dtm_busi.user_account set balance=balance-? where user_id=?", reqFrom(c).Amount, 1) - return M{"dtm_result": "SUCCESS"}, err + return dtmcli.ResultSuccess, err }) } diff --git a/examples/quick_start.go b/examples/quick_start.go index 3b3810d..409e0b3 100644 --- a/examples/quick_start.go +++ b/examples/quick_start.go @@ -44,7 +44,7 @@ func QsFireRequest() string { func qsAdjustBalance(uid int, amount int) (interface{}, error) { _, err := common.SdbExec(sdbGet(), "update dtm_busi.user_account set balance = balance + ? where user_id = ?", amount, uid) - return M{"dtm_result": "SUCCESS"}, err + return dtmcli.ResultSuccess, err } func qsAddRoute(app *gin.Engine) {