add mongo barrier

5 years ago · e1fd9269e9
17 changed files with 392 additions and 10 deletions
--- a/dtmcli/barrier_mongo.go
+++ 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
+}
--- 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
 }
--- 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 {
--- 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
--- 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=
--- 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'
--- a/sqls/busi.mongo.js
+++ 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})
--- a/sqls/dtmcli.barrier.mongo.js
+++ 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"});
--- 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)
 	})
 }
--- 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)
--- 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)
--- 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)
--- 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()
 }
--- 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)
 }
--- a/test/msg_barrier_mongo_test.go
+++ 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")
+}
--- a/test/saga_barrier_mongo_test.go
+++ 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)
+}
--- 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")