Merge pull request #37 from InjectiveLabs/fix_non_contiguous_eventnonce

Fix Non contiguous event nonce issue

orchestrator/cosmos/broadcast.go

@@ -260,10 +260,15 @@ func (s *peggyBroadcastClient) sendDepositClaims(
 		Orchestrator:   s.broadcastClient.FromAddress().String(),
 	}
 
-	if err := s.broadcastClient.QueueBroadcastMsg(msg); err != nil {
+	if txResponse, err := s.broadcastClient.SyncBroadcastMsg(msg); err != nil {
 		metrics.ReportFuncError(s.svcTags)
 		log.WithError(err).Errorln("broadcasting MsgDepositClaim failed")
 		return err
+	} else {
+		log.WithFields(log.Fields{
+			"event_nonce": deposit.EventNonce.String(),
+			"txHash":      txResponse.TxHash,
+		}).Infoln("Oracle sent deposit event succesfully")
 	}
 
 	return nil
@@ -289,10 +294,16 @@ func (s *peggyBroadcastClient) sendWithdrawClaims(
 		TokenContract: withdraw.Token.Hex(),
 		Orchestrator:  s.AccFromAddress().String(),
 	}
-	if err := s.broadcastClient.QueueBroadcastMsg(msg); err != nil {
+
+	if txResponse, err := s.broadcastClient.SyncBroadcastMsg(msg); err != nil {
 		metrics.ReportFuncError(s.svcTags)
 		log.WithError(err).Errorln("broadcasting MsgWithdrawClaim failed")
 		return err
+	} else {
+		log.WithFields(log.Fields{
+			"event_nonce": withdraw.EventNonce.String(),
+			"txHash":      txResponse.TxHash,
+		}).Infoln("Oracle sent Withdraw event succesfully")
 	}
 
 	return nil
@@ -330,10 +341,15 @@ func (s *peggyBroadcastClient) sendValsetUpdateClaims(
 		Orchestrator: s.AccFromAddress().String(),
 	}
 
-	if err := s.broadcastClient.QueueBroadcastMsg(msg); err != nil {
+	if txResponse, err := s.broadcastClient.SyncBroadcastMsg(msg); err != nil {
 		metrics.ReportFuncError(s.svcTags)
 		log.WithError(err).Errorln("broadcasting MsgValsetUpdatedClaim failed")
 		return err
+	} else {
+		log.WithFields(log.Fields{
+			"event_nonce": valsetUpdate.EventNonce.String(),
+			"txHash":      txResponse.TxHash,
+		}).Infoln("Oracle sent ValsetUpdate event succesfully")
 	}
 
 	return nil
@@ -355,7 +371,6 @@ func (s *peggyBroadcastClient) SendEthereumClaims(
 	// Individual arrays (deposits, withdraws, valsetUpdates) are sorted.
 	// Broadcast claim events sequentially starting with eventNonce = lastClaimEvent + 1.
 	for count < totalClaimEvents {
-		time.Sleep(100 * time.Millisecond)
 		if i < len(deposits) && deposits[i].EventNonce.Uint64() == lastClaimEvent+1 {
 			// send deposit
 			if err := s.sendDepositClaims(ctx, deposits[i]); err != nil {
@@ -383,6 +398,11 @@ func (s *peggyBroadcastClient) SendEthereumClaims(
 		}
 		count = count + 1
 		lastClaimEvent = lastClaimEvent + 1
+
+		// Considering blockTime=2.8s on Injective chain, Adding Sleep to make sure new event is
+		// sent only after previous event is executed successfully.
+		// Otherwise it will through `non contiguous event nonce` failing CheckTx.
+		time.Sleep(3 * time.Second)
 	}
 	return nil
 }

orchestrator/eth_event_watcher.go

@@ -11,6 +11,10 @@ import (
 	wrappers "github.com/InjectiveLabs/peggo/solidity/wrappers/Peggy.sol"
 )
 
+// Considering blocktime=3s approx on injective-chain, and oracle loop duration = 1 minute, we can broadcast only 20 events in each iteration.
+// So better to search only 20 blocks to ensure all the events gets broadcasted to injective chain without missing.
+const defaultBlocksToSearch = 20
+
 // CheckForEvents checks for events such as a deposit to the Peggy Ethereum contract or a validator set update
 // or a transaction batch update. It then responds to these events by performing actions on the Cosmos chain if required
 func (s *peggyOrchestrator) CheckForEvents(

orchestrator/main_loops.go

@@ -94,7 +94,7 @@ func (s *peggyOrchestrator) EthOracleMainLoop(ctx context.Context) (err error) {
 				2. if validator was in UnBonding state, the claims broadcasted in last iteration are failed.
 				3. if infura call failed while filtering events, the peggo missed to broadcast claim events occured in last iteration.
 		**/
-		if time.Since(lastResync) >= 6*time.Hour {
+		if time.Since(lastResync) >= 10*time.Hour {
 			if err := retry.Do(func() (err error) {
 				lastCheckedBlock, err = s.GetLastCheckedBlock(ctx)
 				return

orchestrator/oracle_resync.go

@@ -4,8 +4,6 @@ import (
 	"context"
 )
 
-const defaultBlocksToSearch = 2000
-
 // GetLastCheckedBlock retrieves the last claim event this oracle has relayed to Cosmos.
 func (s *peggyOrchestrator) GetLastCheckedBlock(ctx context.Context) (uint64, error) {
 	lastClaimEvent, err := s.cosmosQueryClient.LastClaimEventByAddr(ctx, s.peggyBroadcastClient.AccFromAddress())

orchestrator/relayer/main_loop.go

@@ -10,7 +10,7 @@ import (
 	"github.com/InjectiveLabs/peggo/orchestrator/loops"
 )
 
-const defaultLoopDur = 1 * time.Minute
+const defaultLoopDur = 5 * time.Minute
 
 func (s *peggyRelayer) Start(ctx context.Context) error {
 	logger := log.WithField("loop", "RelayerMainLoop")