diff --git a/dtmcli/barrier_mongo.go b/dtmcli/barrier_mongo.go new file mode 100644 index 0000000..c4902ac --- /dev/null +++ b/dtmcli/barrier_mongo.go @@ -0,0 +1,102 @@ +package dtmcli + +import ( + "context" + "fmt" + "strings" + + "github.com/dtm-labs/dtm/dtmcli/dtmimp" + "github.com/dtm-labs/dtm/dtmcli/logger" + "go.mongodb.org/mongo-driver/bson" + "go.mongodb.org/mongo-driver/mongo" +) + +// MongoCall sub-trans barrier for mongo. see http://dtm.pub/practice/barrier +// experimental +func (bb *BranchBarrier) MongoCall(mc *mongo.Client, busiCall func(mongo.SessionContext) error) (rerr error) { + bb.BarrierID = bb.BarrierID + 1 + bid := fmt.Sprintf("%02d", bb.BarrierID) + return mc.UseSession(context.Background(), func(sc mongo.SessionContext) (rerr error) { + rerr = sc.StartTransaction() + if rerr != nil { + return nil + } + defer dtmimp.DeferDo(&rerr, func() error { + return sc.CommitTransaction(sc) + }, func() error { + return sc.AbortTransaction(sc) + }) + originOp := map[string]string{ + BranchCancel: BranchTry, + BranchCompensate: BranchAction, + }[bb.Op] + + originAffected, oerr := mongoInsertBarrier(sc, mc, bb.TransType, bb.Gid, bb.BranchID, originOp, bid, bb.Op) + currentAffected, rerr := mongoInsertBarrier(sc, mc, bb.TransType, bb.Gid, bb.BranchID, bb.Op, bid, bb.Op) + logger.Debugf("originAffected: %d currentAffected: %d", originAffected, currentAffected) + if rerr == nil { + rerr = oerr + } + if (bb.Op == BranchCancel || bb.Op == BranchCompensate) && originAffected > 0 || // null compensate + currentAffected == 0 { // repeated request or dangled request + return + } + if rerr == nil { + rerr = busiCall(sc) + } + return + }) +} + +// MongoQueryPrepared query prepared for redis +// experimental +func (bb *BranchBarrier) MongoQueryPrepared(mc *mongo.Client) error { + _, err := mongoInsertBarrier(context.Background(), mc, bb.TransType, bb.Gid, "00", "msg", "01", "rollback") + var result bson.M + if err == nil { + fs := strings.Split(dtmimp.BarrierTableName, ".") + barrier := mc.Database(fs[0]).Collection(fs[1]) + err = barrier.FindOne(context.Background(), bson.D{ + {Key: "gid", Value: bb.Gid}, + {Key: "branch_id", Value: "00"}, + {Key: "op", Value: "msg"}, + {Key: "barrier_id", Value: "01"}, + }).Decode(&result) + } + var reason string + if err == nil { + reason, _ = result["reason"].(string) + } + if err == nil && reason == "rollback" { + return ErrFailure + } + return err +} + +func mongoInsertBarrier(sc context.Context, mc *mongo.Client, transType string, gid string, branchID string, op string, barrierID string, reason string) (int64, error) { + if op == "" { + return 0, nil + } + fs := strings.Split(dtmimp.BarrierTableName, ".") + barrier := mc.Database(fs[0]).Collection(fs[1]) + r := barrier.FindOne(sc, bson.D{ + {Key: "gid", Value: gid}, + {Key: "branch_id", Value: branchID}, + {Key: "op", Value: op}, + {Key: "barrier_id", Value: barrierID}, + }) + err := r.Err() + if err == mongo.ErrNoDocuments { + _, err = barrier.InsertOne(sc, + bson.D{ + {Key: "trans_type", Value: transType}, + {Key: "gid", Value: gid}, + {Key: "branch_id", Value: branchID}, + {Key: "op", Value: op}, + {Key: "barrier_id", Value: barrierID}, + {Key: "reason", Value: reason}, + }) + return 1, err + } + return 0, err +} diff --git a/dtmcli/dtmimp/trans_xa_base.go b/dtmcli/dtmimp/trans_xa_base.go index bea23e9..d1eb352 100644 --- a/dtmcli/dtmimp/trans_xa_base.go +++ b/dtmcli/dtmimp/trans_xa_base.go @@ -54,7 +54,9 @@ func (xc *XaClientBase) HandleLocalTrans(xa *TransBase, cb func(*sql.DB) error) if rerr != nil { return } - defer DBExec(db, GetDBSpecial().GetXaSQL("end", xaBranch)) + defer func() { + _, _ = DBExec(db, GetDBSpecial().GetXaSQL("end", xaBranch)) + }() rerr = cb(db) return } diff --git a/dtmcli/dtmimp/utils.go b/dtmcli/dtmimp/utils.go index ff28f9e..cdfe466 100644 --- a/dtmcli/dtmimp/utils.go +++ b/dtmcli/dtmimp/utils.go @@ -220,6 +220,7 @@ func RespAsErrorCompatible(resp *resty.Response) error { return nil } +// DeferDo a common defer do used in dtmcli/dtmgrpc func DeferDo(rerr *error, success func() error, fail func() error) { defer func() { if x := recover(); x != nil { diff --git a/go.mod b/go.mod index 856f0a7..7d2519b 100644 --- a/go.mod +++ b/go.mod @@ -15,11 +15,12 @@ require ( github.com/lib/pq v1.10.3 github.com/lithammer/shortuuid v2.0.3+incompatible github.com/lithammer/shortuuid/v3 v3.0.7 - github.com/natefinch/lumberjack v2.0.0+incompatible // indirect + github.com/natefinch/lumberjack v2.0.0+incompatible github.com/onsi/gomega v1.16.0 github.com/prometheus/client_golang v1.11.0 github.com/stretchr/testify v1.7.0 go.etcd.io/bbolt v1.3.6 + go.mongodb.org/mongo-driver v1.8.3 go.uber.org/atomic v1.9.0 // indirect go.uber.org/automaxprocs v1.4.1-0.20210525221652-0180b04c18a7 go.uber.org/multierr v1.7.0 // indirect diff --git a/go.sum b/go.sum index 70b2d7f..ada06ee 100644 --- a/go.sum +++ b/go.sum @@ -167,6 +167,7 @@ github.com/go-sql-driver/mysql v1.4.0/go.mod h1:zAC/RDZ24gD3HViQzih4MyKcchzm+sOG github.com/go-sql-driver/mysql v1.5.0/go.mod h1:DCzpHaOWr8IXmIStZouvnhqoel9Qv2LBy8hT2VhHyBg= github.com/go-sql-driver/mysql v1.6.0 h1:BCTh4TKNUYmOmMUcQ3IipzF5prigylS7XXjEkfCHuOE= github.com/go-sql-driver/mysql v1.6.0/go.mod h1:DCzpHaOWr8IXmIStZouvnhqoel9Qv2LBy8hT2VhHyBg= +github.com/go-stack/stack v1.8.0 h1:5SgMzNM5HxrEjV0ww2lTmX6E2Izsfxas4+YHWRs3Lsk= github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY= github.com/go-task/slim-sprig v0.0.0-20210107165309-348f09dbbbc0/go.mod h1:fyg7847qk6SyHyPtNmDHnmrv/HOrqktSC+C9fM+CJOE= github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA= @@ -204,6 +205,8 @@ github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaS github.com/golang/protobuf v1.5.2 h1:ROPKBNFfQgOUMifHyP+KYbvpjbdoFNs+aK7DXlji0Tw= github.com/golang/protobuf v1.5.2/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY= github.com/golang/snappy v0.0.0-20180518054509-2e65f85255db/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q= +github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q= +github.com/golang/snappy v0.0.4 h1:yAGX7huGHXlcLOEtBnF4w7FQwA26wojNCwOYAEhLjQM= github.com/golang/snappy v0.0.4/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q= github.com/gonum/blas v0.0.0-20181208220705-f22b278b28ac/go.mod h1:P32wAyui1PQ58Oce/KYkOqQv8cVw1zAapXOl+dRFGbc= github.com/gonum/floats v0.0.0-20181209220543-c233463c7e82/go.mod h1:PxC8OnwL11+aosOB5+iEPoV3picfs8tUpkVd0pDo+Kg= @@ -340,6 +343,7 @@ github.com/julienschmidt/httprouter v1.3.0/go.mod h1:JR6WtHb+2LUe8TCKY3cZOxFyyO8 github.com/justinas/alice v1.2.0/go.mod h1:fN5HRH/reO/zrUflLfTN43t3vXvKzvZIENsNEe7i7qA= github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8= github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck= +github.com/klauspost/compress v1.13.6 h1:P76CopJELS0TiO2mebmnzgWaajssP/EszplttgQxcgc= github.com/klauspost/compress v1.13.6/go.mod h1:/3/Vjq9QcHkK5uEr5lBEmyoZ1iFhe47etQ6QUkpK6sk= github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ= github.com/konsorten/go-windows-terminal-sequences v1.0.2/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ= @@ -392,6 +396,7 @@ github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJ github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0= github.com/modern-go/reflect2 v1.0.1 h1:9f412s+6RmYXLWZSEzVVgPGK7C2PphHj5RJrvfx9AWI= github.com/modern-go/reflect2 v1.0.1/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0= +github.com/montanaflynn/stats v0.0.0-20171201202039-1bf9dbcd8cbe/go.mod h1:wL8QJuTMNUDYhXwkmfOly8iTdp5TEcJFWZD2D7SIkUc= github.com/munnerz/goautoneg v0.0.0-20120707110453-a547fc61f48d/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U= github.com/mwitkow/go-conntrack v0.0.0-20190716064945-2f068394615f/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U= @@ -498,15 +503,22 @@ github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5Cc github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/tal-tech/go-zero v1.2.4 h1:S0Cj69rHfIbZBVgQHnblR2dGkhz5Ij3bm1lM19kkig4= github.com/tal-tech/go-zero v1.2.4/go.mod h1:EHOQsRClBD4svg5mJl8iqkQn5bTMZx+AeHwkQpSjJXg= +github.com/tidwall/pretty v1.0.0 h1:HsD+QiTn7sK6flMKIvNmpqz1qrpP3Ps6jOKIKMooyg4= +github.com/tidwall/pretty v1.0.0/go.mod h1:XNkn88O1ChpSDQmQeStsy+sBenx6DDtFZJxhVysOjyk= github.com/ugorji/go v1.1.7 h1:/68gy2h+1mWMrwZFeD1kQialdSzAb432dtpeJ42ovdo= github.com/ugorji/go v1.1.7/go.mod h1:kZn38zHttfInRq0xu/PH0az30d+z6vm202qpg1oXVMw= github.com/ugorji/go/codec v1.1.7 h1:2SvQaVZ1ouYrrKKwoSk2pzd4A9evlKJb9oTL+OaLUSs= github.com/ugorji/go/codec v1.1.7/go.mod h1:Ax+UKWsSmolVDwsd+7N3ZtXu+yMGCf907BLYF3GoBXY= github.com/urfave/cli v1.22.5/go.mod h1:Gos4lmkARVdJ6EkW0WaNv/tZAAMe9V7XWyB60NtXRu0= github.com/urfave/cli/v2 v2.3.0/go.mod h1:LJmUH05zAU44vOAcrfzZQKsZbVcdbOG8rtL3/XcUArI= +github.com/xdg-go/pbkdf2 v1.0.0 h1:Su7DPu48wXMwC3bs7MCNG+z4FhcyEuz5dlvchbq0B0c= github.com/xdg-go/pbkdf2 v1.0.0/go.mod h1:jrpuAogTd400dnrH08LKmI/xc1MbPOebTwRqcT5RDeI= +github.com/xdg-go/scram v1.0.2 h1:akYIkZ28e6A96dkWNJQu3nmCzH3YfwMPQExUYDaRv7w= github.com/xdg-go/scram v1.0.2/go.mod h1:1WAq6h33pAW+iRreB34OORO2Nf7qel3VV3fjBj+hCSs= +github.com/xdg-go/stringprep v1.0.2 h1:6iq84/ryjjeRmMJwxutI51F2GIPlP5BfTvXHeYjyhBc= github.com/xdg-go/stringprep v1.0.2/go.mod h1:8F9zXuvzgwmyT5DUm4GUfZGDdT3W+LCvS6+da4O5kxM= +github.com/youmark/pkcs8 v0.0.0-20181117223130-1be2e3e5546d h1:splanxYIlg+5LfHAM6xpdFEAYOk8iySO56hMFq6uLyA= +github.com/youmark/pkcs8 v0.0.0-20181117223130-1be2e3e5546d/go.mod h1:rHwXgn7JulP+udvsHwJoVG1YGAP6VLg4y9I5dyZdqmA= github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= github.com/yuin/goldmark v1.3.5/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k= @@ -522,6 +534,8 @@ go.etcd.io/etcd/client/pkg/v3 v3.5.1 h1:XIQcHCFSG53bJETYeRJtIxdLv2EWRGxcfzR8lSnT go.etcd.io/etcd/client/pkg/v3 v3.5.1/go.mod h1:IJHfcCEKxYu1Os13ZdwCwIUTUVGYTSAM3YSwc9/Ac1g= go.etcd.io/etcd/client/v3 v3.5.1 h1:oImGuV5LGKjCqXdjkMHCyWa5OO1gYKCnC/1sgdfj1Uk= go.etcd.io/etcd/client/v3 v3.5.1/go.mod h1:OnjH4M8OnAotwaB2l9bVgZzRFKru7/ZMoS46OtKyd3Q= +go.mongodb.org/mongo-driver v1.8.3 h1:TDKlTkGDKm9kkJVUOAXDK5/fkqKHJVwYQSpoRfB43R4= +go.mongodb.org/mongo-driver v1.8.3/go.mod h1:0sQWfOeY63QTntERDJJ/0SuKK0T1uVSgKCuAROlKEPY= go.opencensus.io v0.21.0/go.mod h1:mSImk1erAIZhrmZN+AvHh14ztQfjbGwt4TtuofqLduU= go.opencensus.io v0.22.0/go.mod h1:+kGneAE2xo2IficOXnaByMWTGM9T73dGwxeWcUqIpI8= go.opencensus.io v0.22.2/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw= @@ -572,6 +586,7 @@ golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPh golang.org/x/crypto v0.0.0-20201002170205-7f63de1d35b0/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= golang.org/x/crypto v0.0.0-20201112155050-0c6587e931a9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= golang.org/x/crypto v0.0.0-20201203163018-be400aefbc4c/go.mod h1:jdWPYTVW3xRLrWPugEBEK3UY2ZEsg3UU495nc5E+M+I= +golang.org/x/crypto v0.0.0-20201216223049-8b5274cf687f/go.mod h1:jdWPYTVW3xRLrWPugEBEK3UY2ZEsg3UU495nc5E+M+I= golang.org/x/crypto v0.0.0-20210616213533-5ff15b29337e/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc= golang.org/x/crypto v0.0.0-20210711020723-a769d52b0f97/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc= golang.org/x/crypto v0.0.0-20210920023735-84f357641f63/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc= @@ -658,6 +673,7 @@ golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJ golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20201207232520-09787c993a3a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20210220032951-036812b2e83c h1:5KslGYwFpkhGh+Q16bwMP3cOontH8FOep7tGV86Y7SQ= golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= @@ -738,6 +754,7 @@ golang.org/x/tools v0.0.0-20190425150028-36563e24a262/go.mod h1:RgjU9mgBXZiqYHBn golang.org/x/tools v0.0.0-20190425163242-31fd60d6bfdc/go.mod h1:RgjU9mgBXZiqYHBnxXauZ1Gv1EHHAz9KjViQ78xBX0Q= golang.org/x/tools v0.0.0-20190506145303-2d16b83fe98c/go.mod h1:RgjU9mgBXZiqYHBnxXauZ1Gv1EHHAz9KjViQ78xBX0Q= golang.org/x/tools v0.0.0-20190524140312-2c0ae7006135/go.mod h1:RgjU9mgBXZiqYHBnxXauZ1Gv1EHHAz9KjViQ78xBX0Q= +golang.org/x/tools v0.0.0-20190531172133-b3315ee88b7d/go.mod h1:/rFqwRUd4F7ZHNgwSSTFct+R/Kf4OFW1sUzUTQQTgfc= golang.org/x/tools v0.0.0-20190606124116-d0a3d012864b/go.mod h1:/rFqwRUd4F7ZHNgwSSTFct+R/Kf4OFW1sUzUTQQTgfc= golang.org/x/tools v0.0.0-20190614205625-5aca471b1d59/go.mod h1:/rFqwRUd4F7ZHNgwSSTFct+R/Kf4OFW1sUzUTQQTgfc= golang.org/x/tools v0.0.0-20190621195816-6e04913cbbac/go.mod h1:/rFqwRUd4F7ZHNgwSSTFct+R/Kf4OFW1sUzUTQQTgfc= diff --git a/helper/compose.store.yml b/helper/compose.store.yml index 48eaff1..4053807 100644 --- a/helper/compose.store.yml +++ b/helper/compose.store.yml @@ -32,3 +32,10 @@ services: - /etc/timezone:/etc/timezone:ro ports: - '6379:6379' + mongo: + image: yedf/mongo-rs + volumes: + - /etc/localtime:/etc/localtime:ro + - /etc/timezone:/etc/timezone:ro + ports: + - '27017:27017' diff --git a/sqls/busi.mongo.js b/sqls/busi.mongo.js new file mode 100644 index 0000000..c9fd6ae --- /dev/null +++ b/sqls/busi.mongo.js @@ -0,0 +1,3 @@ +use busi +db.busi.insert({user_id: 1, balance: 10000}) +db.busi.insert({user_id: 2, balance: 10000}) diff --git a/sqls/dtmcli.barrier.mongo.js b/sqls/dtmcli.barrier.mongo.js new file mode 100644 index 0000000..47afde2 --- /dev/null +++ b/sqls/dtmcli.barrier.mongo.js @@ -0,0 +1,4 @@ +use dtm_barrier +db.barrier.drop() +db.barrier.createIndex({gid:1, branch_id:1, op: 1, barrier_id: 1}, {unique: true}) +//db.barrier.insert({gid:"123", branch_id:"01", op:"action", barrier_id:"01", reason:"action"}); diff --git a/test/busi/barrier.go b/test/busi/barrier.go index 556e6aa..d0c29ad 100644 --- a/test/busi/barrier.go +++ b/test/busi/barrier.go @@ -14,6 +14,7 @@ import ( "github.com/dtm-labs/dtm/dtmgrpc" "github.com/dtm-labs/dtm/dtmutil" "github.com/gin-gonic/gin" + "go.mongodb.org/mongo-driver/mongo" emptypb "google.golang.org/protobuf/types/known/emptypb" ) @@ -83,21 +84,46 @@ func init() { app.POST(BusiAPI+"/SagaRedisTransOutCom", dtmutil.WrapHandler2(func(c *gin.Context) interface{} { return MustBarrierFromGin(c).RedisCheckAdjustAmount(RedisGet(), GetRedisAccountKey(TransOutUID), reqFrom(c).Amount, 7*86400) })) + app.POST(BusiAPI+"/SagaMongoTransIn", dtmutil.WrapHandler2(func(c *gin.Context) interface{} { + return MustBarrierFromGin(c).MongoCall(MongoGet(), func(sc mongo.SessionContext) error { + return SagaMongoAdjustBalance(sc, sc.Client(), TransInUID, reqFrom(c).Amount, reqFrom(c).TransInResult) + }) + })) + app.POST(BusiAPI+"/SagaMongoTransInCom", dtmutil.WrapHandler2(func(c *gin.Context) interface{} { + return MustBarrierFromGin(c).MongoCall(MongoGet(), func(sc mongo.SessionContext) error { + return SagaMongoAdjustBalance(sc, sc.Client(), TransInUID, -reqFrom(c).Amount, "") + }) + })) + app.POST(BusiAPI+"/SagaMongoTransOut", dtmutil.WrapHandler2(func(c *gin.Context) interface{} { + return MustBarrierFromGin(c).MongoCall(MongoGet(), func(sc mongo.SessionContext) error { + return SagaMongoAdjustBalance(sc, sc.Client(), TransOutUID, -reqFrom(c).Amount, reqFrom(c).TransOutResult) + }) + })) + app.POST(BusiAPI+"/SagaMongoTransOutCom", dtmutil.WrapHandler2(func(c *gin.Context) interface{} { + return MustBarrierFromGin(c).MongoCall(MongoGet(), func(sc mongo.SessionContext) error { + return SagaMongoAdjustBalance(sc, sc.Client(), TransOutUID, reqFrom(c).Amount, "") + }) + })) app.POST(BusiAPI+"/TccBTransOutTry", dtmutil.WrapHandler2(func(c *gin.Context) interface{} { req := reqFrom(c) if req.TransOutResult != "" { return dtmcli.String2DtmError(req.TransOutResult) } + bb := MustBarrierFromGin(c) if req.Store == Redis { - return MustBarrierFromGin(c).RedisCheckAdjustAmount(RedisGet(), GetRedisAccountKey(TransOutUID), req.Amount, 7*86400) + return bb.RedisCheckAdjustAmount(RedisGet(), GetRedisAccountKey(TransOutUID), req.Amount, 7*86400) + } else if req.Store == Mongo { + return bb.MongoCall(MongoGet(), func(sc mongo.SessionContext) error { + return SagaMongoAdjustBalance(sc, sc.Client(), TransOutUID, -req.Amount, "") + }) } - return MustBarrierFromGin(c).Call(txGet(), func(tx *sql.Tx) error { + return bb.Call(txGet(), func(tx *sql.Tx) error { return tccAdjustTrading(tx, TransOutUID, -req.Amount) }) })) app.POST(BusiAPI+"/TccBTransOutConfirm", dtmutil.WrapHandler2(func(c *gin.Context) interface{} { - if reqFrom(c).Store == Redis { + if reqFrom(c).Store == Redis || reqFrom(c).Store == Mongo { return nil } return MustBarrierFromGin(c).Call(txGet(), func(tx *sql.Tx) error { @@ -111,10 +137,16 @@ func init() { // TccBarrierTransOutCancel will be use in test func TccBarrierTransOutCancel(c *gin.Context) interface{} { req := reqFrom(c) + bb := MustBarrierFromGin(c) if req.Store == Redis { - return MustBarrierFromGin(c).RedisCheckAdjustAmount(RedisGet(), GetRedisAccountKey(TransOutUID), -req.Amount, 7*86400) + return bb.RedisCheckAdjustAmount(RedisGet(), GetRedisAccountKey(TransOutUID), -req.Amount, 7*86400) + } + if req.Store == Mongo { + return bb.MongoCall(MongoGet(), func(sc mongo.SessionContext) error { + return SagaMongoAdjustBalance(sc, sc.Client(), TransOutUID, reqFrom(c).Amount, "") + }) } - return MustBarrierFromGin(c).Call(txGet(), func(tx *sql.Tx) error { + return bb.Call(txGet(), func(tx *sql.Tx) error { return tccAdjustTrading(tx, TransOutUID, reqFrom(c).Amount) }) } diff --git a/test/busi/base_http.go b/test/busi/base_http.go index 5822aef..f5c9250 100644 --- a/test/busi/base_http.go +++ b/test/busi/base_http.go @@ -137,6 +137,11 @@ func BaseAddRoute(app *gin.Engine) { bb := MustBarrierFromGin(c) return bb.RedisQueryPrepared(RedisGet(), 86400) })) + app.GET(BusiAPI+"/MongoQueryPrepared", dtmutil.WrapHandler2(func(c *gin.Context) interface{} { + logger.Debugf("%s MongoQueryPrepared", c.Query("gid")) + bb := MustBarrierFromGin(c) + return bb.MongoQueryPrepared(MongoGet()) + })) app.POST(BusiAPI+"/TransInXa", dtmutil.WrapHandler2(func(c *gin.Context) interface{} { return XaClient.XaLocalTransaction(c.Request.URL.Query(), func(db *sql.DB, xa *dtmcli.Xa) error { return SagaAdjustBalance(db, TransInUID, reqFrom(c).Amount, reqFrom(c).TransInResult) diff --git a/test/busi/base_types.go b/test/busi/base_types.go index c20ba89..ced598a 100644 --- a/test/busi/base_types.go +++ b/test/busi/base_types.go @@ -7,12 +7,14 @@ package busi import ( + "context" "fmt" "github.com/dtm-labs/dtm/dtmcli" "github.com/dtm-labs/dtm/dtmcli/dtmimp" "github.com/dtm-labs/dtm/dtmcli/logger" "github.com/gin-gonic/gin" + "go.mongodb.org/mongo-driver/bson" ) // BusiConf 1 @@ -42,6 +44,13 @@ func GetBalanceByUID(uid int, store string) int { accA, err := rd.Get(rd.Context(), GetRedisAccountKey(uid)).Result() dtmimp.E2P(err) return dtmimp.MustAtoi(accA) + } else if store == "mongo" { + mg := MongoGet() + account := mg.Database("dtm_busi").Collection("user_account") + var result bson.M + err := account.FindOne(context.Background(), bson.D{{Key: "user_id", Value: uid}}).Decode(&result) + dtmimp.E2P(err) + return int(result["balance"].(int32)) } ua := UserAccount{} _ = dbGet().Must().Model(&ua).Where("user_id=?", uid).First(&ua) diff --git a/test/busi/busi.go b/test/busi/busi.go index c39dc27..3697723 100644 --- a/test/busi/busi.go +++ b/test/busi/busi.go @@ -1,6 +1,7 @@ package busi import ( + "context" "errors" "fmt" "strings" @@ -9,6 +10,8 @@ import ( "github.com/dtm-labs/dtm/dtmcli/dtmimp" "github.com/dtm-labs/dtm/dtmcli/logger" "github.com/gin-gonic/gin" + "go.mongodb.org/mongo-driver/bson" + "go.mongodb.org/mongo-driver/mongo" codes "google.golang.org/grpc/codes" status "google.golang.org/grpc/status" ) @@ -22,6 +25,9 @@ const TransInUID = 2 // Redis 1 const Redis = "redis" +// Mongo 1 +const Mongo = "mongo" + func handleGrpcBusiness(in *BusiReq, result1 string, result2 string, busi string) error { res := dtmimp.OrString(result1, result2, dtmcli.ResultSuccess) logger.Debugf("grpc busi %s %v %s %s result: %s", busi, in, result1, result2, res) @@ -72,6 +78,19 @@ func SagaAdjustBalance(db dtmcli.DB, uid int, amount int, result string) error { _, err := dtmimp.DBExec(db, "update dtm_busi.user_account set balance = balance + ? where user_id = ?", amount, uid) return err } + +// SagaMongoAdjustBalance 1 +func SagaMongoAdjustBalance(ctx context.Context, mc *mongo.Client, uid int, amount int, result string) error { + if strings.Contains(result, dtmcli.ResultFailure) { + return dtmcli.ErrFailure + } + _, err := mc.Database("dtm_busi").Collection("user_account").UpdateOne(ctx, + bson.D{{Key: "user_id", Value: uid}}, + bson.D{{Key: "$inc", Value: bson.D{{Key: "balance", Value: amount}}}}) + logger.Debugf("dtm_busi.user_account $inc balance of %d by %d err: %v", uid, amount, err) + return err +} + func tccAdjustTrading(db dtmcli.DB, uid int, amount int) error { affected, err := dtmimp.DBExec(db, `update dtm_busi.user_account set trading_balance=trading_balance+? where user_id=? and trading_balance + ? + balance >= 0`, amount, uid, amount) diff --git a/test/busi/startup.go b/test/busi/startup.go index 7e02949..3181787 100644 --- a/test/busi/startup.go +++ b/test/busi/startup.go @@ -25,4 +25,5 @@ func PopulateDB(skipDrop bool) { _, err := RedisGet().FlushAll(context.Background()).Result() // redis barrier need clear dtmimp.E2P(err) SetRedisBothAccount(10000, 10000) + SetupMongoBarrierAndBusi() } diff --git a/test/busi/utils.go b/test/busi/utils.go index 86615e4..b144e70 100644 --- a/test/busi/utils.go +++ b/test/busi/utils.go @@ -19,6 +19,9 @@ import ( "github.com/gin-gonic/gin" "github.com/go-redis/redis/v8" "github.com/go-resty/resty/v2" + "go.mongodb.org/mongo-driver/bson" + "go.mongodb.org/mongo-driver/mongo" + "go.mongodb.org/mongo-driver/mongo/options" grpc "google.golang.org/grpc" "google.golang.org/grpc/metadata" ) @@ -135,11 +138,59 @@ func RedisGet() *redis.Client { return rdb } +var ( + mongoOnce sync.Once + mongoc *mongo.Client +) + +// MongoGet get mongo client +func MongoGet() *mongo.Client { + mongoOnce.Do(func() { + ctx := context.Background() + client, err := mongo.Connect(ctx, options.Client().ApplyURI("mongodb://localhost:27017/?retryWrites=false")) + dtmimp.E2P(err) + mongoc = client + }) + return mongoc +} + // SetRedisBothAccount 1 -func SetRedisBothAccount(accountA int, accountB int) { +func SetRedisBothAccount(amountA int, ammountB int) { rd := RedisGet() - _, err := rd.Set(rd.Context(), GetRedisAccountKey(TransOutUID), accountA, 0).Result() + _, err := rd.Set(rd.Context(), GetRedisAccountKey(TransOutUID), amountA, 0).Result() + dtmimp.E2P(err) + _, err = rd.Set(rd.Context(), GetRedisAccountKey(TransInUID), ammountB, 0).Result() dtmimp.E2P(err) - _, err = rd.Set(rd.Context(), GetRedisAccountKey(TransInUID), accountB, 0).Result() +} + +// SetMongoBothAccount 1 +func SetMongoBothAccount(amountA int, amountB int) { + mc := MongoGet() + col := mc.Database("dtm_busi").Collection("user_account") + _, err := col.InsertOne(context.Background(), bson.D{{Key: "user_id", Value: TransOutUID}, {Key: "balance", Value: amountA}}) + dtmimp.E2P(err) + _, err = col.InsertOne(context.Background(), bson.D{{Key: "user_id", Value: TransInUID}, {Key: "balance", Value: amountB}}) + dtmimp.E2P(err) + +} + +// SetupMongoBarrierAndBusi 1 +func SetupMongoBarrierAndBusi() { + mc := MongoGet() + err := mc.Database("dtm_busi").Drop(context.Background()) + dtmimp.E2P(err) + err = mc.Database("dtm_barrier").Drop(context.Background()) + dtmimp.E2P(err) + col := mc.Database("dtm_barrier").Collection("barrier") + _, err = col.Indexes().CreateOne(context.Background(), mongo.IndexModel{ + Keys: bson.D{ + {Key: "gid", Value: 1}, + {Key: "branch_id", Value: 1}, + {Key: "op", Value: 1}, + {Key: "barrier_id", Value: 1}, + }, + Options: options.Index().SetUnique(true), + }) dtmimp.E2P(err) + SetMongoBothAccount(10000, 10000) } diff --git a/test/msg_barrier_mongo_test.go b/test/msg_barrier_mongo_test.go new file mode 100644 index 0000000..1f36f6f --- /dev/null +++ b/test/msg_barrier_mongo_test.go @@ -0,0 +1,78 @@ +package test + +import ( + "errors" + "testing" + + "github.com/dtm-labs/dtm/dtmcli" + "github.com/dtm-labs/dtm/dtmcli/dtmimp" + "github.com/dtm-labs/dtm/test/busi" + "github.com/stretchr/testify/assert" + "go.mongodb.org/mongo-driver/mongo" +) + +func TestMsgMongoDoSucceed(t *testing.T) { + before := getBeforeBalances("mongo") + gid := dtmimp.GetFuncName() + req := busi.GenTransReq(30, false, false) + msg := dtmcli.NewMsg(DtmServer, gid). + Add(busi.Busi+"/SagaMongoTransIn", req) + err := msg.DoAndSubmit(Busi+"/MongoQueryPrepared", func(bb *dtmcli.BranchBarrier) error { + return bb.MongoCall(busi.MongoGet(), func(sc mongo.SessionContext) error { + return busi.SagaMongoAdjustBalance(sc, sc.Client(), busi.TransOutUID, -30, "") + }) + }) + assert.Nil(t, err) + waitTransProcessed(msg.Gid) + assert.Equal(t, []string{StatusSucceed}, getBranchesStatus(msg.Gid)) + assert.Equal(t, StatusSucceed, getTransStatus(msg.Gid)) + assertNotSameBalance(t, before, "mongo") +} + +func TestMsgMongoDoBusiFailed(t *testing.T) { + before := getBeforeBalances("mongo") + gid := dtmimp.GetFuncName() + req := busi.GenTransReq(30, false, false) + msg := dtmcli.NewMsg(DtmServer, gid). + Add(busi.Busi+"/SagaMongoTransIn", req) + err := msg.DoAndSubmit(Busi+"/MongoQueryPrepared", func(bb *dtmcli.BranchBarrier) error { + return errors.New("an error") + }) + assert.Error(t, err) + assertSameBalance(t, before, "mongo") +} + +func TestMsgMongoDoCommitFailed(t *testing.T) { + before := getBeforeBalances("mongo") + gid := dtmimp.GetFuncName() + req := busi.GenTransReq(30, false, false) + msg := dtmcli.NewMsg(DtmServer, gid). + Add(busi.Busi+"/SagaMongoTransIn", req) + err := msg.DoAndSubmit(Busi+"/MongoQueryPrepared", func(bb *dtmcli.BranchBarrier) error { + return bb.MongoCall(busi.MongoGet(), func(sc mongo.SessionContext) error { + err := busi.SagaMongoAdjustBalance(sc, sc.Client(), busi.TransOutUID, -30, "") + assert.Nil(t, err) + return errors.New("commit failed") + }) + }) + assert.Error(t, err) + assertSameBalance(t, before, "mongo") +} + +func TestMsgMongoDoCommitAfterFailed(t *testing.T) { + before := getBeforeBalances("mongo") + gid := dtmimp.GetFuncName() + req := busi.GenTransReq(30, false, false) + msg := dtmcli.NewMsg(DtmServer, gid). + Add(busi.Busi+"/SagaMongoTransIn", req) + err := msg.DoAndSubmit(Busi+"/MongoQueryPrepared", func(bb *dtmcli.BranchBarrier) error { + err := bb.MongoCall(busi.MongoGet(), func(sc mongo.SessionContext) error { + return busi.SagaMongoAdjustBalance(sc, sc.Client(), busi.TransOutUID, -30, "") + }) + assert.Nil(t, err) + return errors.New("an error") + }) + assert.Error(t, err) + waitTransProcessed(gid) + assertNotSameBalance(t, before, "mongo") +} diff --git a/test/saga_barrier_mongo_test.go b/test/saga_barrier_mongo_test.go new file mode 100644 index 0000000..3937475 --- /dev/null +++ b/test/saga_barrier_mongo_test.go @@ -0,0 +1,46 @@ +/* + * Copyright (c) 2021 yedf. All rights reserved. + * Use of this source code is governed by a BSD-style + * license that can be found in the LICENSE file. + */ + +package test + +import ( + "testing" + + "github.com/dtm-labs/dtm/dtmcli" + "github.com/dtm-labs/dtm/dtmcli/dtmimp" + "github.com/dtm-labs/dtm/test/busi" + "github.com/stretchr/testify/assert" +) + +func TestSagaBarrierMongoNormal(t *testing.T) { + before := getBeforeBalances("mongo") + saga := genSagaBarrierMongo(dtmimp.GetFuncName(), false) + err := saga.Submit() + assert.Nil(t, err) + waitTransProcessed(saga.Gid) + assert.Equal(t, []string{StatusPrepared, StatusSucceed, StatusPrepared, StatusSucceed}, getBranchesStatus(saga.Gid)) + assert.Equal(t, StatusSucceed, getTransStatus(saga.Gid)) + assertNotSameBalance(t, before, "mongo") +} + +func TestSagaBarrierMongoRollback(t *testing.T) { + before := getBeforeBalances("mongo") + saga := genSagaBarrierMongo(dtmimp.GetFuncName(), true) + err := saga.Submit() + assert.Nil(t, err) + waitTransProcessed(saga.Gid) + assert.Equal(t, StatusFailed, getTransStatus(saga.Gid)) + assert.Equal(t, []string{StatusSucceed, StatusSucceed, StatusSucceed, StatusFailed}, getBranchesStatus(saga.Gid)) + assertSameBalance(t, before, "mongo") +} + +func genSagaBarrierMongo(gid string, transInFailed bool) *dtmcli.Saga { + req := busi.GenTransReq(30, false, transInFailed) + req.Store = "mongo" + return dtmcli.NewSaga(DtmServer, gid). + Add(Busi+"/SagaMongoTransOut", Busi+"/SagaMongoTransOutCom", req). + Add(Busi+"/SagaMongoTransIn", Busi+"/SagaMongoTransInCom", req) +} diff --git a/test/tcc_barrier_test.go b/test/tcc_barrier_test.go index 57692b3..c55308d 100644 --- a/test/tcc_barrier_test.go +++ b/test/tcc_barrier_test.go @@ -54,6 +54,10 @@ func TestTccBarrierDisorderMysql(t *testing.T) { runTestTccBarrierDisorder(t, "mysql") } +func TestTccBarrierDisorderMongo(t *testing.T) { + runTestTccBarrierDisorder(t, "mongo") +} + func TestTccBarrierDisorderRedis(t *testing.T) { busi.SetRedisBothAccount(200, 200) runTestTccBarrierDisorder(t, "redis")