enter batch model if fall behind

config/config.go

@@ -54,6 +54,7 @@ type BBCConfig struct {
 	Mnemonic                                   string `json:"mnemonic"`
 	SleepMillisecondForWaitBlock               int64  `json:"sleep_millisecond_for_wait_block"`
 	BlockIntervalForCleanUpUndeliveredPackages uint64 `json:"block_interval_for_clean_up_undelivered_packages"`
+	BehindBlockThreshold                       uint64 `json:"behind_block_threshold"`
 }
 
 func (cfg *BBCConfig) Validate() {
@@ -78,6 +79,9 @@ func (cfg *BBCConfig) Validate() {
 	if cfg.BlockIntervalForCleanUpUndeliveredPackages == 0 {
 		panic("block interval for cleanup undelivered packages must not be zero")
 	}
+	if cfg.BehindBlockThreshold == 0 {
+		panic("behind block threshold should be be positive")
+	}
 }
 
 type BSCConfig struct {

config/config.json

@@ -11,7 +11,8 @@
     "aws_secret_name": "",
     "mnemonic": "",
     "sleep_millisecond_for_wait_block": 500,
-    "block_interval_for_clean_up_undelivered_packages": 5500
+    "block_interval_for_clean_up_undelivered_packages": 5500,
+    "behind_block_threshold": 100
   },
   "bsc_config": {
     "key_type": "local_private_key",

model/model.go

@@ -62,6 +62,8 @@ func InitTables(db *gorm.DB) {
 		db.CreateTable(&RelayTransaction{})
 		db.Model(&RelayTransaction{}).AddIndex("idx_tx_height", "tx_height")
 		db.Model(&RelayTransaction{}).AddIndex("idx_tx_status", "tx_status")
+		db.Model(&RelayTransaction{}).AddIndex("idx_tx_channel_id", "channel_id")
+		db.Model(&RelayTransaction{}).AddIndex("idx_tx_sequence", "sequence")
 		db.Model(&RelayTransaction{}).AddIndex("idx_tx_create_time", "create_time")
 	}
 

relayer/cleanup.go

@@ -2,6 +2,7 @@ package relayer
 
 import (
 	"github.com/binance-chain/bsc-relayer/common"
+	"github.com/binance-chain/bsc-relayer/model"
 )
 
 func (r *Relayer) cleanPreviousPackages(height uint64) error {
@@ -16,6 +17,10 @@ func (r *Relayer) cleanPreviousPackages(height uint64) error {
 		if err != nil {
 			return err
 		}
+		latestDeliveredSeq := r.getLatestDeliveredSequence(channelId)
+		if nextDeliverSequence < latestDeliveredSeq {
+			nextDeliverSequence = latestDeliveredSeq
+		}
 		common.Logger.Infof("channelID: %d, next deliver sequence %d on BSC, next sequence %d on BC",
 			channelId, nextDeliverSequence, nextSequence)
 		if nextSequence > nextDeliverSequence {
@@ -36,3 +41,12 @@ func (r *Relayer) cleanPreviousPackages(height uint64) error {
 	}
 	return nil
 }
+
+func (r *Relayer) getLatestDeliveredSequence(channelID uint8) uint64 {
+	if r.db == nil {
+		return 0
+	}
+	tx := model.RelayTransaction{}
+	r.db.Where("channel_id = ? and tx_status != ?", channelID, model.Failure).Order("sequence desc").First(&tx)
+	return tx.Sequence
+}

relayer/daemon.go

@@ -17,12 +17,31 @@ const (
 	WaitSecondForTrackTx = 10
 )
 
+func (r *Relayer) getLatestHeight() uint64 {
+	abciInfo, err := r.bbcExecutor.RpcClient.ABCIInfo()
+	if err != nil {
+		common.Logger.Errorf("Query latest height error: %s", err.Error())
+		return 0
+	}
+	return uint64(abciInfo.Response.LastBlockHeight)
+}
+
 func (r *Relayer) relayerDaemon(startHeight uint64, curValidatorsHash cmn.HexBytes) {
 	var tashSet *common.TaskSet
 	var err error
 	height := startHeight
 	common.Logger.Info("Start relayer daemon")
 	for {
+		latestHeight := r.getLatestHeight() - 1
+		if latestHeight > height+r.bbcExecutor.Config.BBCConfig.BehindBlockThreshold {
+			err := r.cleanPreviousPackages(latestHeight)
+			if err != nil {
+				common.Logger.Error(err.Error())
+			}
+			height = latestHeight + 1
+			continue // packages have been delivered on cleanup
+		}
+
 		tashSet, curValidatorsHash, err = r.bbcExecutor.MonitorCrossChainPackage(int64(height), curValidatorsHash)
 		if err != nil {
 			sleepTime := time.Duration(r.bbcExecutor.Config.BBCConfig.SleepMillisecondForWaitBlock * int64(time.Millisecond))