Release 1.0.1

README.md

@@ -12,7 +12,7 @@ Make sure your have go1.11+ already installed
 ### Install 
 ```bash 
 $ make install 
-```  
+```
 
 ### Run-delivery 
 ```bash 

bridge/setu/broadcaster/broadcaster.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"sync"
+	"time"
 
 	"github.com/cosmos/cosmos-sdk/client"
 	cliContext "github.com/cosmos/cosmos-sdk/client/context"
@@ -56,6 +57,16 @@ func NewTxBroadcaster(cdc *codec.Codec) *TxBroadcaster {
 	return &txBroadcaster
 }
 
+func (tb *TxBroadcaster) BroadcastToHeimdallWithDelay(msg sdk.Msg, delay time.Duration) error {
+	eta := time.Now().Add(delay)
+	timeAfterTrigger := time.After(delay)
+	tb.logger.Info("Tx sent on heimdall ", "msg", msg.Type(), "currentTime", time.Now(), "delayTime", eta)
+	select {
+	case <-timeAfterTrigger:
+		return tb.BroadcastToHeimdall(msg)
+	}
+}
+
 // BroadcastToHeimdall broadcast to heimdall
 func (tb *TxBroadcaster) BroadcastToHeimdall(msg sdk.Msg) error {
 	tb.heimdallMutex.Lock()

bridge/setu/listener/tron.go

@@ -214,7 +214,7 @@ func (tl *TronListener) queryAndBroadcastEvents(chainManagerParams *chainmanager
 						// topup has to be processed first before validator join. so adding delay.
 						delay := util.TaskDelayBetweenEachVal
 						tl.sendTaskWithDelay("sendValidatorJoinToHeimdall", selectedEvent.Name, logBytes, delay)
-					} else if isCurrentValidator, delay := util.CalculateTaskDelay(tl.cliCtx); isCurrentValidator {
+					} else if isCurrentValidator, delay := util.CalculateTaskDelayWithOffset(tl.cliCtx,1); isCurrentValidator {
 						// topup has to be processed first before validator join. so adding delay.
 						delay = delay + util.TaskDelayBetweenEachVal
 						tl.sendTaskWithDelay("sendValidatorJoinToHeimdall", selectedEvent.Name, logBytes, delay)
@@ -237,7 +237,7 @@ func (tl *TronListener) queryAndBroadcastEvents(chainManagerParams *chainmanager
 					}
 					if bytes.Equal(event.SignerPubkey, pubkeyBytes) {
 						tl.sendTaskWithDelay("sendSignerChangeToHeimdall", selectedEvent.Name, logBytes, 0)
-					} else if isCurrentValidator, delay := util.CalculateTaskDelay(tl.cliCtx); isCurrentValidator {
+					} else if isCurrentValidator, delay := util.CalculateTaskDelayWithOffset(tl.cliCtx,1); isCurrentValidator {
 						tl.sendTaskWithDelay("sendSignerChangeToHeimdall", selectedEvent.Name, logBytes, delay)
 					}
 
@@ -248,7 +248,7 @@ func (tl *TronListener) queryAndBroadcastEvents(chainManagerParams *chainmanager
 					}
 					if util.IsEventSender(tl.cliCtx, event.ValidatorId.Uint64()) {
 						tl.sendTaskWithDelay("sendUnstakeInitToHeimdall", selectedEvent.Name, logBytes, 0)
-					} else if isCurrentValidator, delay := util.CalculateTaskDelay(tl.cliCtx); isCurrentValidator {
+					} else if isCurrentValidator, delay := util.CalculateTaskDelayWithOffset(tl.cliCtx,1); isCurrentValidator {
 						tl.sendTaskWithDelay("sendUnstakeInitToHeimdall", selectedEvent.Name, logBytes, delay)
 					}
 
@@ -264,7 +264,7 @@ func (tl *TronListener) queryAndBroadcastEvents(chainManagerParams *chainmanager
 					}
 					if bytes.Equal(event.User.Bytes(), helper.GetAddress()) {
 						tl.sendTaskWithDelay("sendTopUpFeeToHeimdall", selectedEvent.Name, logBytes, 0)
-					} else if isCurrentValidator, delay := util.CalculateTaskDelay(tl.cliCtx); isCurrentValidator {
+					} else if isCurrentValidator, delay := util.CalculateTaskDelayWithOffset(tl.cliCtx,1); isCurrentValidator {
 						tl.sendTaskWithDelay("sendTopUpFeeToHeimdall", selectedEvent.Name, logBytes, delay)
 					}
 
@@ -280,7 +280,7 @@ func (tl *TronListener) queryAndBroadcastEvents(chainManagerParams *chainmanager
 					}
 					if util.IsEventSender(tl.cliCtx, event.ValidatorId.Uint64()) {
 						tl.sendTaskWithDelay("sendUnjailToHeimdall", selectedEvent.Name, logBytes, 0)
-					} else if isCurrentValidator, delay := util.CalculateTaskDelay(tl.cliCtx); isCurrentValidator {
+					} else if isCurrentValidator, delay := util.CalculateTaskDelayWithOffset(tl.cliCtx,1); isCurrentValidator {
 						tl.sendTaskWithDelay("sendUnjailToHeimdall", selectedEvent.Name, logBytes, delay)
 					}
 

bridge/setu/processor/checkpointsync.go

@@ -61,13 +61,15 @@ func (cp *CheckpointProcessor) handleCheckpointSync() {
 			msg := checkpointTypes.NewMsgCheckpointSync(hmTypes.BytesToHeimdallAddress(helper.GetAddress()),
 				proposer, nextCheckpointNumber, start, end, rootChain)
 			// return broadcast to heimdall
-			if err := cp.txBroadcaster.BroadcastToHeimdall(msg); err != nil {
-				cp.Logger.Error("Error while broadcasting checkpoint-sync to heimdall",
-					"root", rootChain, "number", nextCheckpointNumber, "start", start, "end", end, "error", err)
-				continue
+			if isCurrentValidator, delay := util.CalculateTaskDelay(cp.cliCtx); isCurrentValidator {
+				if err := cp.txBroadcaster.BroadcastToHeimdallWithDelay(msg, delay); err != nil {
+					cp.Logger.Error("Error while broadcasting checkpoint-sync to heimdall",
+						"root", rootChain, "number", nextCheckpointNumber, "start", start, "end", end, "error", err)
+					continue
+				}
+				cp.Logger.Info("checkpoint sync transaction sent successfully",
+					"root", rootChain, "number", nextCheckpointNumber, "start", start, "end", end)
 			}
-			cp.Logger.Info("checkpoint sync transaction sent successfully",
-				"root", rootChain, "number", nextCheckpointNumber, "start", start, "end", end)
 		}
 	}
 }
@@ -183,13 +185,17 @@ func (cp *CheckpointProcessor) sendCheckpointSyncAckToHeimdall(eventName string,
 		)
 		// fetch checkpoint sync buffer
 		bufferedCheckpoint, err := util.GetBufferedCheckpointSync(cp.cliCtx, checkpointChain)
-		if err == nil {
-			bufferedTime := time.Unix(int64(bufferedCheckpoint.TimeStamp), 0)
-			currentTime := time.Now().UTC()
-			if currentTime.Sub(bufferedTime).Seconds() > checkpointParams.CheckpointBufferTime.Seconds()/5 {
-				cp.Logger.Debug("checkpoint sync buffer has expired, ignore this ack")
-				return nil
-			}
+
+		if err != nil {
+			cp.Logger.Debug("checkpoint sync buffer has been cleared,  this ack already submitted.")
+			return nil
+		}
+
+		bufferedTime := time.Unix(int64(bufferedCheckpoint.TimeStamp), 0)
+		currentTime := time.Now().UTC()
+		if currentTime.Sub(bufferedTime).Seconds() > checkpointParams.CheckpointBufferTime.Seconds()/5 {
+			cp.Logger.Debug("checkpoint sync buffer has expired, ignore this ack")
+			return nil
 		}
 
 		// create msg checkpoint ack message

bridge/setu/processor/staking.go

@@ -3,6 +3,7 @@ package processor
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"time"
 
 	checkpointTypes "github.com/maticnetwork/heimdall/checkpoint/types"
@@ -22,6 +23,10 @@ import (
 	hmTypes "github.com/maticnetwork/heimdall/types"
 )
 
+const (
+	defaultDelayDuration = 10 * time.Second
+)
+
 // StakingProcessor - process staking related events
 type StakingProcessor struct {
 	BaseProcessor
@@ -200,6 +205,17 @@ func (sp *StakingProcessor) sendUnstakeInitToHeimdall(eventName string, logBytes
 			return nil
 		}
 
+		validNonce, nonceDelay, err := sp.checkValidNonce(event.ValidatorId.Uint64(), event.Nonce.Uint64())
+		if err != nil {
+			sp.Logger.Error("Error while validating nonce for the validator", "error", err)
+			return err
+		}
+
+		if !validNonce {
+			sp.Logger.Info("Ignoring task to send unstake-init to heimdall as nonce is out of order")
+			return tasks.NewErrRetryTaskLater("Nonce out of order", defaultDelayDuration*time.Duration(nonceDelay))
+		}
+
 		sp.Logger.Info(
 			"✅ Received task to send unstake-init to heimdall",
 			"event", eventName,
@@ -327,6 +343,18 @@ func (sp *StakingProcessor) sendSignerChangeToHeimdall(eventName string, logByte
 			)
 			return nil
 		}
+
+		validNonce, nonceDelay, err := sp.checkValidNonce(event.ValidatorId.Uint64(), event.Nonce.Uint64())
+		if err != nil {
+			sp.Logger.Error("Error while validating nonce for the validator", "error", err)
+			return err
+		}
+
+		if !validNonce {
+			sp.Logger.Info("Ignoring task to send signer-change to heimdall as nonce is out of order")
+			return tasks.NewErrRetryTaskLater("Nonce out of order", defaultDelayDuration*time.Duration(nonceDelay))
+		}
+
 		sp.Logger.Info(
 			"✅ Received task to send signer-change to heimdall",
 			"event", eventName,
@@ -681,3 +709,61 @@ func (sp *StakingProcessor) getStakingContext(rootChain string) (*StakingContext
 		ChainmanagerParams: chainmanagerParams,
 	}, nil
 }
+
+func (sp *StakingProcessor) checkValidNonce(validatorId uint64, txnNonce uint64) (bool, uint64, error) {
+	currentNonce, currentHeight, err := util.GetValidatorNonce(sp.cliCtx, validatorId)
+	if err != nil {
+		sp.Logger.Error("Failed to fetch validator nonce and height data from API", "validatorId", validatorId)
+		return false, 0, err
+	}
+
+	if currentNonce+1 != txnNonce {
+		sp.Logger.Error("Nonce for the given event not in order", "validatorId", validatorId, "currentNonce", currentNonce, "txnNonce", txnNonce)
+		return false, txnNonce - currentNonce, nil
+	}
+
+	stakingTxnCount, err := queryTxCount(sp.cliCtx, validatorId, currentHeight)
+	if err != nil {
+		sp.Logger.Error("Failed to query stake txns by txquery for the given validator", "validatorId", validatorId)
+		return false, 0, err
+	}
+
+	if stakingTxnCount != 0 {
+		sp.Logger.Info("Recent staking txn count for the given validator is not zero", "validatorId", validatorId, "currentNonce", currentNonce, "txnNonce", txnNonce, "currentHeight", currentHeight)
+		return false, 1, nil
+	}
+
+	return true, 0, nil
+}
+
+func queryTxCount(cliCtx cliContext.CLIContext, validatorId uint64, currentHeight int64) (int, error) {
+	const (
+		defaultPage  = 1
+		defaultLimit = 30 // should be consistent with tendermint/tendermint/rpc/core/pipe.go:19
+	)
+
+	stakingTxnMsgMap := map[string]string{
+		"validator-stake-update": "stake-update",
+		"validator-join":         "validator-join",
+		"signer-update":          "signer-update",
+		"validator-exit":         "validator-exit",
+	}
+
+	for msg, action := range stakingTxnMsgMap {
+		events := []string{
+			fmt.Sprintf("%s.%s='%s'", sdk.EventTypeMessage, sdk.AttributeKeyAction, msg),
+			fmt.Sprintf("%s.%s=%d", action, "validator-id", validatorId),
+			fmt.Sprintf("%s.%s>%d", "tx", "height", currentHeight-3),
+		}
+
+		searchResult, err := helper.QueryTxsByEvents(cliCtx, events, defaultPage, defaultLimit)
+		if err != nil {
+			return 0, err
+		}
+
+		if searchResult.TotalCount != 0 {
+			return searchResult.TotalCount, nil
+		}
+	}
+	return 0, nil
+}