Merge pull request #223 from xyctruth/delay

Msg支持延迟消息

dtmcli/msg.go

@@ -16,6 +16,7 @@ import (
 // Msg reliable msg type
 type Msg struct {
 	dtmimp.TransBase
+	delay uint64 // delay call branch, unit second
 }
 
 // NewMsg create new msg
@@ -30,6 +31,12 @@ func (s *Msg) Add(action string, postData interface{}) *Msg {
 	return s
 }
 
+// SetDelay delay call branch, unit second
+func (s *Msg) SetDelay(delay uint64) *Msg {
+	s.delay = delay
+	return s
+}
+
 // Prepare prepare the msg, msg will later be submitted
 func (s *Msg) Prepare(queryPrepared string) error {
 	s.QueryPrepared = dtmimp.OrString(queryPrepared, s.QueryPrepared)
@@ -38,6 +45,7 @@ func (s *Msg) Prepare(queryPrepared string) error {
 
 // Submit submit the msg
 func (s *Msg) Submit() error {
+	s.BuildCustomOptions()
 	return dtmimp.TransCallDtm(&s.TransBase, s, "submit")
 }
 
@@ -74,3 +82,10 @@ func (s *Msg) DoAndSubmit(queryPrepared string, busiCall func(bb *BranchBarrier)
 	}
 	return err
 }
+
+// BuildCustomOptions add custom options to the request context
+func (s *Msg) BuildCustomOptions() {
+	if s.delay > 0 {
+		s.CustomData = dtmimp.MustMarshalString(map[string]interface{}{"delay": s.delay})
+	}
+}

dtmcli/saga.go

@@ -43,12 +43,12 @@ func (s *Saga) EnableConcurrent() *Saga {
 
 // Submit submit the saga trans
 func (s *Saga) Submit() error {
-	s.AddConcurrentContext()
+	s.BuildCustomOptions()
 	return dtmimp.TransCallDtm(&s.TransBase, s, "submit")
 }
 
-// AddConcurrentContext adds concurrent options to the request context
-func (s *Saga) AddConcurrentContext() {
+// BuildCustomOptions add custom options to the request context
+func (s *Saga) BuildCustomOptions() {
 	if s.concurrent {
 		s.CustomData = dtmimp.MustMarshalString(map[string]interface{}{"orders": s.orders, "concurrent": s.concurrent})
 	}

dtmgrpc/msg.go

@@ -33,6 +33,12 @@ func (s *MsgGrpc) Add(action string, msg proto.Message) *MsgGrpc {
 	return s
 }
 
+// SetDelay delay call branch, unit second
+func (s *MsgGrpc) SetDelay(delay uint64) *MsgGrpc {
+	s.Msg.SetDelay(delay)
+	return s
+}
+
 // Prepare prepare the msg, msg will later be submitted
 func (s *MsgGrpc) Prepare(queryPrepared string) error {
 	s.QueryPrepared = dtmimp.OrString(queryPrepared, s.QueryPrepared)
@@ -41,6 +47,7 @@ func (s *MsgGrpc) Prepare(queryPrepared string) error {
 
 // Submit submit the msg
 func (s *MsgGrpc) Submit() error {
+	s.Msg.BuildCustomOptions()
 	return dtmgimp.DtmGrpcCall(&s.TransBase, "Submit")
 }
 

dtmgrpc/saga.go

@@ -43,6 +43,6 @@ func (s *SagaGrpc) EnableConcurrent() *SagaGrpc {
 
 // Submit submit the saga trans
 func (s *SagaGrpc) Submit() error {
-	s.Saga.AddConcurrentContext()
+	s.Saga.BuildCustomOptions()
 	return dtmgimp.DtmGrpcCall(&s.Saga.TransBase, "Submit")
 }

dtmsvr/storage/boltdb/boltdb.go

@@ -364,10 +364,10 @@ func (s *Store) ChangeGlobalStatus(global *storage.TransGlobalStore, newStatus s
 }
 
 // TouchCronTime updates cronTime
-func (s *Store) TouchCronTime(global *storage.TransGlobalStore, nextCronInterval int64) {
+func (s *Store) TouchCronTime(global *storage.TransGlobalStore, nextCronInterval int64, nextCronTime *time.Time) {
 	oldUnix := global.NextCronTime.Unix()
-	global.NextCronTime = dtmutil.GetNextTime(nextCronInterval)
 	global.UpdateTime = dtmutil.GetNextTime(0)
+	global.NextCronTime = nextCronTime
 	global.NextCronInterval = nextCronInterval
 	err := s.boltDb.Update(func(t *bolt.Tx) error {
 		g := tGetGlobal(t, global.Gid)

dtmsvr/storage/redis/redis.go

@@ -261,9 +261,9 @@ return gid
 }
 
 // TouchCronTime updates cronTime
-func (s *Store) TouchCronTime(global *storage.TransGlobalStore, nextCronInterval int64) {
-	global.NextCronTime = dtmutil.GetNextTime(nextCronInterval)
+func (s *Store) TouchCronTime(global *storage.TransGlobalStore, nextCronInterval int64, nextCronTime *time.Time) {
 	global.UpdateTime = dtmutil.GetNextTime(0)
+	global.NextCronTime = nextCronTime
 	global.NextCronInterval = nextCronInterval
 	args := newArgList().
 		AppendGid(global.Gid).

dtmsvr/storage/sql/sql.go

@@ -121,9 +121,9 @@ func (s *Store) ChangeGlobalStatus(global *storage.TransGlobalStore, newStatus s
 }
 
 // TouchCronTime updates cronTime
-func (s *Store) TouchCronTime(global *storage.TransGlobalStore, nextCronInterval int64) {
-	global.NextCronTime = dtmutil.GetNextTime(nextCronInterval)
+func (s *Store) TouchCronTime(global *storage.TransGlobalStore, nextCronInterval int64, nextCronTime *time.Time) {
 	global.UpdateTime = dtmutil.GetNextTime(0)
+	global.NextCronTime = nextCronTime
 	global.NextCronInterval = nextCronInterval
 	dbGet().Must().Model(global).Where("status=? and gid=?", global.Status, global.Gid).
 		Select([]string{"next_cron_time", "update_time", "next_cron_interval"}).Updates(global)

dtmsvr/storage/store.go

@@ -28,6 +28,6 @@ type Store interface {
 	LockGlobalSaveBranches(gid string, status string, branches []TransBranchStore, branchStart int)
 	MaySaveNewTrans(global *TransGlobalStore, branches []TransBranchStore) error
 	ChangeGlobalStatus(global *TransGlobalStore, newStatus string, updates []string, finished bool)
-	TouchCronTime(global *TransGlobalStore, nextCronInterval int64)
+	TouchCronTime(global *TransGlobalStore, nextCronInterval int64, nextCronTime *time.Time)
 	LockOneGlobalTrans(expireIn time.Duration) *TransGlobalStore
 }

dtmsvr/trans_status.go

@@ -17,13 +17,26 @@ import (
 	"github.com/dtm-labs/dtm/dtmcli/logger"
 	"github.com/dtm-labs/dtm/dtmgrpc"
 	"github.com/dtm-labs/dtm/dtmgrpc/dtmgimp"
+	"github.com/dtm-labs/dtm/dtmutil"
 	"github.com/dtm-labs/dtmdriver"
 	"google.golang.org/grpc/metadata"
 )
 
-func (t *TransGlobal) touchCronTime(ctype cronType) {
+// touchCronTime Based on ctype or delay set nextCronTime
+// delay = 0 ,use ctype set nextCronTime and nextCronInterval
+// delay > 0 ,use delay set nextCronTime ，use ctype set nextCronInterval
+func (t *TransGlobal) touchCronTime(ctype cronType, delay uint64) {
 	t.lastTouched = time.Now()
-	GetStore().TouchCronTime(&t.TransGlobalStore, t.getNextCronInterval(ctype))
+	nextCronInterval := t.getNextCronInterval(ctype)
+
+	var nextCronTime *time.Time
+	if delay > 0 {
+		nextCronTime = dtmutil.GetNextTime(int64(delay))
+	} else {
+		nextCronTime = dtmutil.GetNextTime(nextCronInterval)
+	}
+
+	GetStore().TouchCronTime(&t.TransGlobalStore, nextCronInterval, nextCronTime)
 	logger.Infof("TouchCronTime for: %s", t.TransGlobalStore.String())
 }
 
@@ -68,6 +81,10 @@ func (t *TransGlobal) isTimeout() bool {
 	return time.Since(*t.CreateTime)+NowForwardDuration >= time.Duration(timeout)*time.Second
 }
 
+func (t *TransGlobal) needDelay(delay uint64) bool {
+	return time.Since(*t.CreateTime)+CronForwardDuration < time.Duration(delay)*time.Second
+}
+
 func (t *TransGlobal) needProcess() bool {
 	return t.Status == dtmcli.StatusSubmitted || t.Status == dtmcli.StatusAborting || t.Status == dtmcli.StatusPrepared && t.isTimeout()
 }
@@ -135,11 +152,11 @@ func (t *TransGlobal) execBranch(branch *TransBranch, branchPos int) error {
 	// if time pass 1500ms and NextCronInterval is not default, then reset NextCronInterval
 	if err == nil && time.Since(t.lastTouched)+NowForwardDuration >= 1500*time.Millisecond ||
 		t.NextCronInterval > conf.RetryInterval && t.NextCronInterval > t.RetryInterval {
-		t.touchCronTime(cronReset)
+		t.touchCronTime(cronReset, 0)
 	} else if err == dtmimp.ErrOngoing {
-		t.touchCronTime(cronKeep)
+		t.touchCronTime(cronKeep, 0)
 	} else if err != nil {
-		t.touchCronTime(cronBackoff)
+		t.touchCronTime(cronBackoff, 0)
 	}
 	return err
 }

dtmsvr/trans_type_msg.go

@@ -11,6 +11,7 @@ import (
 	"fmt"
 
 	"github.com/dtm-labs/dtm/dtmcli"
+	"github.com/dtm-labs/dtm/dtmcli/dtmimp"
 	"github.com/dtm-labs/dtm/dtmcli/logger"
 )
 
@@ -38,6 +39,10 @@ func (t *transMsgProcessor) GenBranches() []TransBranch {
 	return branches
 }
 
+type cMsgCustom struct {
+	Delay uint64 //delay call branch, unit second
+}
+
 func (t *TransGlobal) mayQueryPrepared() {
 	if !t.needProcess() || t.Status == dtmcli.StatusSubmitted {
 		return
@@ -48,10 +53,10 @@ func (t *TransGlobal) mayQueryPrepared() {
 	} else if errors.Is(err, dtmcli.ErrFailure) {
 		t.changeStatus(dtmcli.StatusFailed)
 	} else if errors.Is(err, dtmcli.ErrOngoing) {
-		t.touchCronTime(cronReset)
+		t.touchCronTime(cronReset, 0)
 	} else {
 		logger.Errorf("getting result failed for %s. error: %v", t.QueryPrepared, err)
-		t.touchCronTime(cronBackoff)
+		t.touchCronTime(cronBackoff, 0)
 	}
 }
 
@@ -60,6 +65,17 @@ func (t *transMsgProcessor) ProcessOnce(branches []TransBranch) error {
 	if !t.needProcess() || t.Status == dtmcli.StatusPrepared {
 		return nil
 	}
+
+	cmc := cMsgCustom{Delay: 0}
+	if t.CustomData != "" {
+		dtmimp.MustUnmarshalString(t.CustomData, &cmc)
+	}
+
+	if cmc.Delay > 0 && t.needDelay(cmc.Delay) {
+		t.touchCronTime(cronKeep, cmc.Delay)
+		return nil
+	}
+
 	current := 0 // 当前正在处理的步骤
 	for ; current < len(branches); current++ {
 		branch := &branches[current]

test/msg_delay_test.go

@@ -0,0 +1,39 @@
+package test
+
+import (
+	"testing"
+
+	"github.com/dtm-labs/dtm/dtmcli"
+	"github.com/dtm-labs/dtm/dtmcli/dtmimp"
+	"github.com/dtm-labs/dtm/dtmsvr"
+	"github.com/dtm-labs/dtm/dtmutil"
+	"github.com/dtm-labs/dtm/test/busi"
+	"github.com/stretchr/testify/assert"
+)
+
+func genMsgDelay(gid string) *dtmcli.Msg {
+	req := busi.GenTransReq(30, false, false)
+	msg := dtmcli.NewMsg(dtmutil.DefaultHTTPServer, gid).
+		Add(busi.Busi+"/TransOut", &req).
+		Add(busi.Busi+"/TransIn", &req).SetDelay(10)
+	msg.QueryPrepared = busi.Busi + "/QueryPrepared"
+	return msg
+}
+
+func TestMsgDelayNormal(t *testing.T) {
+	gid := dtmimp.GetFuncName()
+	msg := genMsgDelay(gid)
+	submitForwardCron(0, func() {
+		msg.Submit()
+		waitTransProcessed(msg.Gid)
+	})
+
+	dtmsvr.NowForwardDuration = 0
+	assert.Equal(t, []string{StatusPrepared, StatusPrepared}, getBranchesStatus(msg.Gid))
+	assert.Equal(t, StatusSubmitted, getTransStatus(msg.Gid))
+	cronTransOnceForwardCron(t, "", 0)
+	cronTransOnceForwardCron(t, "", 8)
+	cronTransOnceForwardCron(t, gid, 12)
+	assert.Equal(t, []string{StatusSucceed, StatusSucceed}, getBranchesStatus(msg.Gid))
+	assert.Equal(t, StatusSucceed, getTransStatus(msg.Gid))
+}

test/store_test.go

@@ -4,11 +4,11 @@ import (
 	"testing"
 	"time"
 
-	"github.com/stretchr/testify/assert"
-
 	"github.com/dtm-labs/dtm/dtmcli/dtmimp"
 	"github.com/dtm-labs/dtm/dtmsvr/storage"
 	"github.com/dtm-labs/dtm/dtmsvr/storage/registry"
+	"github.com/dtm-labs/dtm/dtmutil"
+	"github.com/stretchr/testify/assert"
 )
 
 func initTransGlobal(gid string) (*storage.TransGlobalStore, storage.Store) {
@@ -74,11 +74,11 @@ func TestStoreLockTrans(t *testing.T) {
 	assert.NotNil(t, g2)
 	assert.Equal(t, gid, g2.Gid)
 
-	s.TouchCronTime(g, 3*conf.RetryInterval)
+	s.TouchCronTime(g, 3*conf.RetryInterval, dtmutil.GetNextTime(3*conf.RetryInterval))
 	g2 = s.LockOneGlobalTrans(2 * time.Duration(conf.RetryInterval) * time.Second)
 	assert.Nil(t, g2)
 
-	s.TouchCronTime(g, 1*conf.RetryInterval)
+	s.TouchCronTime(g, 1*conf.RetryInterval, dtmutil.GetNextTime(1*conf.RetryInterval))
 	g2 = s.LockOneGlobalTrans(2 * time.Duration(conf.RetryInterval) * time.Second)
 	assert.NotNil(t, g2)
 	assert.Equal(t, gid, g2.Gid)

test/types.go

@@ -68,6 +68,13 @@ func cronTransOnceForwardCron(t *testing.T, gid string, seconds int) {
 	dtmsvr.CronForwardDuration = old
 }
 
+func submitForwardCron(seconds int, fn func()) {
+	old := dtmsvr.CronForwardDuration
+	dtmsvr.CronForwardDuration = time.Duration(seconds) * time.Second
+	fn()
+	dtmsvr.CronForwardDuration = old
+}
+
 const (
 	// StatusPrepared status for global/branch trans status.
 	StatusPrepared = dtmcli.StatusPrepared