Merge pull request #13 from strangelove-ventures/bd21/ethereum-nonce-…

…sequence-map

Bd21/ethereum nonce sequence map

cmd/circle/attestation.go

@@ -38,6 +38,7 @@ func CheckAttestation(cfg config.Config, logger log.Logger, irisLookupId string)
 		logger.Debug("unable to unmarshal response")
 		return nil
 	}
+	logger.Info(fmt.Sprintf("Attestation found for %s%s%s", cfg.Circle.AttestationBaseUrl, "0x", irisLookupId))
 
 	return &response
 }

cmd/ethereum/broadcast.go

@@ -40,21 +40,15 @@ func Broadcast(
 	}
 
 	for attempt := 0; attempt <= cfg.Networks.Destination.Ethereum.BroadcastRetries; attempt++ {
-		logger.Debug(fmt.Sprintf(
+		logger.Info(fmt.Sprintf(
 			"Broadcasting %s message from %d to %d: with source tx hash %s",
 			msg.Type,
 			msg.SourceDomain,
 			msg.DestDomain,
 			msg.SourceTxHash))
 
-		// TODO Account sequence lock is implemented but gets out of sync with remote.
-		// accountSequence := sequenceMap.Next(cfg.Networks.Destination.Noble.DomainId)
-		_, err := GetEthereumAccountNonce(cfg.Networks.Destination.Ethereum.RPC, ethereumAddress)
-		if err != nil {
-			logger.Error("unable to retrieve ethereum account nonce")
-			continue
-		}
-		//auth.Nonce = big.NewInt(nonce)
+		nonce := sequenceMap.Next(cfg.Networks.Destination.Ethereum.DomainId)
+		auth.Nonce = big.NewInt(nonce)
 
 		// broadcast txn
 		tx, err := messageTransmitter.ReceiveMessage(
@@ -66,12 +60,16 @@ func Broadcast(
 			msg.Status = types.Complete
 			return tx, nil
 		} else {
-
 			logger.Error(fmt.Sprintf("error during broadcast: %s", err.Error()))
 			if parsedErr, ok := err.(JsonError); ok {
 				if parsedErr.ErrorCode() == 3 && parsedErr.Error() == "execution reverted: Nonce already used" {
-					msg.Status = types.Failed
-					return nil, errors.New(fmt.Sprintf("Nonce already used"))
+					nonce, err = GetEthereumAccountNonce(cfg.Networks.Destination.Ethereum.RPC, ethereumAddress)
+					if err != nil {
+						logger.Error("unable to retrieve account number")
+					}
+					logger.Debug(fmt.Sprintf("retrying with new nonce: %d", nonce))
+					sequenceMap.Put(cfg.Networks.Destination.Ethereum.DomainId, nonce)
+
 				}
 			}
 

cmd/ethereum/listener.go

@@ -72,6 +72,10 @@ func StartListener(cfg config.Config, logger log.Logger, processingQueue chan *t
 		logger.Info(fmt.Sprintf("New historical msg from source domain %d with tx hash %s", parsedMsg.SourceDomain, parsedMsg.SourceTxHash))
 
 		processingQueue <- parsedMsg
+
+		// It might help to wait a small amount of time between sending messages into the processing queue
+		// so that account sequences / nonces are set correctly
+		// time.Sleep(10 * time.Millisecond)
 	}
 
 	// consume stream
@@ -90,6 +94,10 @@ func StartListener(cfg config.Config, logger log.Logger, processingQueue chan *t
 				logger.Info(fmt.Sprintf("New stream msg from %d with tx hash %s", parsedMsg.SourceDomain, parsedMsg.SourceTxHash))
 
 				processingQueue <- parsedMsg
+
+				// It might help to wait a small amount of time between sending messages into the processing queue
+				// so that account sequences / nonces are set correctly
+				// time.Sleep(10 * time.Millisecond)
 			}
 		}
 	}()

cmd/ethereum/util_test.go

@@ -19,7 +19,7 @@ func init() {
 }
 
 func TestGetEthereumAccountNonce(t *testing.T) {
-	_, err := ethereum.GetEthereumAccountNonce(cfg.Networks.Destination.Ethereum.RPC, cfg.Networks.Minters[0].MinterAddress)
+	_, err := ethereum.GetEthereumAccountNonce(cfg.Networks.Destination.Ethereum.RPC, "0x4996f29b254c77972fff8f25e6f7797b3c9a0eb6")
 	require.Nil(t, err)
 }
 

cmd/noble/broadcast.go

@@ -8,6 +8,7 @@ import (
 	"fmt"
 	"io"
 	"net/http"
+	"regexp"
 	"strconv"
 	"time"
 
@@ -78,16 +79,16 @@ func Broadcast(
 	}
 
 	for attempt := 0; attempt <= cfg.Networks.Destination.Noble.BroadcastRetries; attempt++ {
-		logger.Debug(fmt.Sprintf(
+		logger.Info(fmt.Sprintf(
 			"Broadcasting %s message from %d to %d: with source tx hash %s",
 			msg.Type,
 			msg.SourceDomain,
 			msg.DestDomain,
 			msg.SourceTxHash))
 
-		// TODO Account sequence lock is implemented but gets out of sync with remote.
-		// accountSequence := sequenceMap.Next(cfg.Networks.Destination.Noble.DomainId)
-		accountNumber, accountSequence, err := GetNobleAccountNumberSequence(cfg.Networks.Destination.Noble.API, nobleAddress)
+		accountSequence := sequenceMap.Next(cfg.Networks.Destination.Noble.DomainId)
+		accountNumber, _, err := GetNobleAccountNumberSequence(cfg.Networks.Destination.Noble.API, nobleAddress)
+
 		if err != nil {
 			logger.Error("unable to retrieve account number")
 		}
@@ -134,14 +135,38 @@ func Broadcast(
 			msg.Status = types.Complete
 			return rpcResponse, nil
 		}
+
+		// check tx response code
+		logger.Error(fmt.Sprintf("received non zero : %d - %s", rpcResponse.Code, rpcResponse.Log))
+
+		if err == nil && rpcResponse.Code == 32 {
+			// on account sequence mismatch, extract correct account sequence and retry
+			pattern := `expected (\d+), got (\d+)`
+			re := regexp.MustCompile(pattern)
+			match := re.FindStringSubmatch(rpcResponse.Log)
+
+			var newAccountSequence int64
+			if len(match) == 3 {
+				// Extract the numbers from the match.
+				newAccountSequence, _ = strconv.ParseInt(match[1], 10, 0)
+			} else {
+				// otherwise, just request the account sequence
+				_, newAccountSequence, err = GetNobleAccountNumberSequence(cfg.Networks.Destination.Noble.API, nobleAddress)
+				if err != nil {
+					logger.Error("unable to retrieve account number")
+				}
+			}
+			logger.Debug(fmt.Sprintf("error during broadcast: %s", rpcResponse.Log))
+			logger.Debug(fmt.Sprintf("retrying with new account sequence: %d", newAccountSequence))
+			sequenceMap.Put(cfg.Networks.Destination.Noble.DomainId, newAccountSequence)
+		}
 		if err != nil {
 			logger.Error(fmt.Sprintf("error during broadcast: %s", err.Error()))
 			logger.Info(fmt.Sprintf("Retrying in %d seconds", cfg.Networks.Destination.Noble.BroadcastRetryInterval))
 			time.Sleep(time.Duration(cfg.Networks.Destination.Noble.BroadcastRetryInterval) * time.Second)
 			continue
 		}
-		// check tx response code
-		logger.Error(fmt.Sprintf("received non zero : %d - %s", rpcResponse.Code, rpcResponse.Log))
+
 		logger.Info(fmt.Sprintf("Retrying in %d seconds", cfg.Networks.Destination.Noble.BroadcastRetryInterval))
 		time.Sleep(time.Duration(cfg.Networks.Destination.Noble.BroadcastRetryInterval) * time.Second)
 	}

cmd/noble/listener_test.go

@@ -1,8 +1,9 @@
-package noble
+package noble_test
 
 import (
 	"cosmossdk.io/log"
 	"github.com/rs/zerolog"
+	"github.com/strangelove-ventures/noble-cctp-relayer/cmd/noble"
 	"github.com/strangelove-ventures/noble-cctp-relayer/config"
 	"github.com/strangelove-ventures/noble-cctp-relayer/types"
 	"github.com/stretchr/testify/require"
@@ -25,8 +26,7 @@ func init() {
 
 func TestStartListener(t *testing.T) {
 	cfg.Networks.Source.Noble.StartBlock = 3273557
-	cfg.Networks.Source.Noble.LookbackPeriod = 0
-	go StartListener(cfg, logger, processingQueue)
+	go noble.StartListener(cfg, logger, processingQueue)
 
 	time.Sleep(20 * time.Second)
 

cmd/process.go

@@ -3,6 +3,7 @@ package cmd
 import (
 	"bytes"
 	"cosmossdk.io/log"
+	"encoding/hex"
 	"fmt"
 	"github.com/spf13/cobra"
 	"github.com/strangelove-ventures/noble-cctp-relayer/cmd/circle"
@@ -11,6 +12,7 @@ import (
 	"github.com/strangelove-ventures/noble-cctp-relayer/config"
 	"github.com/strangelove-ventures/noble-cctp-relayer/types"
 	"os"
+	"strings"
 	"sync"
 	"time"
 )
@@ -26,7 +28,7 @@ var startCmd = &cobra.Command{
 // Store represents terminal states
 var State = types.NewStateMap()
 
-// SequenceMap maps the domain -> the equivalent minter address sequence/nonce
+// SequenceMap maps the domain -> the equivalent minter account sequence or nonce
 var sequenceMap = types.NewSequenceMap()
 
 func Start(cmd *cobra.Command, args []string) {
@@ -178,20 +180,35 @@ func filterDisabledCCTPRoutes(cfg config.Config, logger log.Logger, msg *types.M
 // filterInvalidDestinationCallers returns true if the minter is not the destination caller for the specified domain
 func filterInvalidDestinationCallers(cfg config.Config, logger log.Logger, msg *types.MessageState) bool {
 	zeroByteArr := make([]byte, 32)
-	bech32DestinationCaller, err := types.DecodeDestinationCaller(msg.DestinationCaller)
-
 	result := false
-	if err != nil {
-		result = true
-	}
-	if !bytes.Equal(msg.DestinationCaller, zeroByteArr) &&
-		bech32DestinationCaller != cfg.Networks.Minters[msg.DestDomain].MinterAddress {
-		result = true
-	}
 
-	if result {
-		logger.Info(fmt.Sprintf("Filtered tx %s because the destination caller %s is specified and it's not the minter %s",
-			msg.SourceTxHash, msg.DestinationCaller, cfg.Networks.Minters[msg.DestDomain].MinterAddress))
+	switch msg.DestDomain {
+	case 4:
+		bech32DestinationCaller, err := types.DecodeDestinationCaller(msg.DestinationCaller)
+		if err != nil {
+			result = true
+		}
+		if !bytes.Equal(msg.DestinationCaller, zeroByteArr) &&
+			bech32DestinationCaller != cfg.Networks.Minters[msg.DestDomain].MinterAddress {
+			result = true
+		}
+		if result {
+			logger.Info(fmt.Sprintf("Filtered tx %s because the destination caller %s is specified and it's not the minter %s",
+				msg.SourceTxHash, msg.DestinationCaller, cfg.Networks.Minters[msg.DestDomain].MinterAddress))
+		}
+
+	default: // minting to evm
+		decodedMinter, err := hex.DecodeString(strings.ReplaceAll(cfg.Networks.Minters[0].MinterAddress, "0x", ""))
+		if err != nil {
+			return !bytes.Equal(msg.DestinationCaller, zeroByteArr)
+		}
+
+		decodedMinterPadded := make([]byte, 32)
+		copy(decodedMinterPadded[12:], decodedMinter)
+
+		if !bytes.Equal(msg.DestinationCaller, zeroByteArr) && !bytes.Equal(msg.DestinationCaller, decodedMinterPadded) {
+			result = true
+		}
 	}
 
 	return result

config/config.go

@@ -23,7 +23,6 @@ type Config struct {
 				RPC              string `yaml:"rpc"`
 				RequestQueueSize uint32 `yaml:"request-queue-size"`
 				StartBlock       uint64 `yaml:"start-block"`
-				LookbackPeriod   uint64 `yaml:"lookback-period"`
 				Workers          uint32 `yaml:"workers"`
 				Enabled          bool   `yaml:"enabled"`
 			} `yaml:"noble"`

integration/config.go

@@ -2,6 +2,7 @@ package integration_testing
 
 import (
 	"cosmossdk.io/log"
+	"github.com/rs/zerolog"
 	"github.com/strangelove-ventures/noble-cctp-relayer/cmd/noble"
 	"github.com/strangelove-ventures/noble-cctp-relayer/config"
 	"github.com/strangelove-ventures/noble-cctp-relayer/types"
@@ -25,7 +26,7 @@ func setupTest() func() {
 	// setup
 	testCfg = Parse("../.ignore/integration.yaml")
 	cfg = config.Parse("../.ignore/testnet.yaml")
-	logger = log.NewLogger(os.Stdout)
+	logger = log.NewLogger(os.Stdout, log.LevelOption(zerolog.DebugLevel))
 
 	_, nextMinterSequence, err := noble.GetNobleAccountNumberSequence(
 		cfg.Networks.Destination.Noble.API,