handle multi-msg scenario

cmd/circle/attestation.go

@@ -1,19 +1,20 @@
 package circle
 
 import (
-	"cosmossdk.io/log"
 	"encoding/json"
 	"fmt"
-	"github.com/strangelove-ventures/noble-cctp-relayer/config"
-	"github.com/strangelove-ventures/noble-cctp-relayer/types"
 	"io"
 	"net/http"
 	"time"
+
+	"cosmossdk.io/log"
+	"github.com/strangelove-ventures/noble-cctp-relayer/config"
+	"github.com/strangelove-ventures/noble-cctp-relayer/types"
 )
 
 // CheckAttestation checks the iris api for attestation status and returns true if attestation is complete
-func CheckAttestation(cfg config.Config, logger log.Logger, irisLookupId string) *types.AttestationResponse {
-	logger.Debug(fmt.Sprintf("Checking attestation for %s%s%s", cfg.Circle.AttestationBaseUrl, "0x", irisLookupId))
+func CheckAttestation(cfg config.Config, logger log.Logger, irisLookupId string, txHash string, sourceDomain, destDomain uint32) *types.AttestationResponse {
+	logger.Debug(fmt.Sprintf("Checking attestation for %s%s%s for source tx %s from %d to %d", cfg.Circle.AttestationBaseUrl, "0x", irisLookupId, txHash, sourceDomain, destDomain))
 
 	client := http.Client{Timeout: 2 * time.Second}
 

cmd/circle/attestation_test.go

@@ -1,13 +1,14 @@
 package circle_test
 
 import (
+	"os"
+	"testing"
+
 	"cosmossdk.io/log"
 	"github.com/rs/zerolog"
 	"github.com/strangelove-ventures/noble-cctp-relayer/cmd/circle"
 	"github.com/strangelove-ventures/noble-cctp-relayer/config"
 	"github.com/stretchr/testify/require"
-	"os"
-	"testing"
 )
 
 var cfg config.Config
@@ -19,12 +20,12 @@ func init() {
 }
 
 func TestAttestationIsReady(t *testing.T) {
-	resp := circle.CheckAttestation(cfg, logger, "85bbf7e65a5992e6317a61f005e06d9972a033d71b514be183b179e1b47723fe")
+	resp := circle.CheckAttestation(cfg, logger, "85bbf7e65a5992e6317a61f005e06d9972a033d71b514be183b179e1b47723fe", "", 0, 4)
 	require.NotNil(t, resp)
 	require.Equal(t, "complete", resp.Status)
 }
 
 func TestAttestationNotFound(t *testing.T) {
-	resp := circle.CheckAttestation(cfg, logger, "not an attestation")
+	resp := circle.CheckAttestation(cfg, logger, "not an attestation", "", 0, 4)
 	require.Nil(t, resp)
 }

cmd/ethereum/broadcast.go

@@ -1,24 +1,28 @@
 package ethereum
 
 import (
-	"cosmossdk.io/log"
+	"context"
 	"encoding/hex"
 	"errors"
 	"fmt"
+	"math/big"
+	"regexp"
+	"strconv"
+	"time"
+
+	"cosmossdk.io/log"
 	"github.com/ethereum/go-ethereum/accounts/abi/bind"
 	"github.com/ethereum/go-ethereum/common"
 	ethtypes "github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/ethclient"
 	"github.com/strangelove-ventures/noble-cctp-relayer/config"
 	"github.com/strangelove-ventures/noble-cctp-relayer/types"
-	"math/big"
-	"regexp"
-	"strconv"
-	"time"
 )
 
 // Broadcast broadcasts a message to Ethereum
 func Broadcast(
+	ctx context.Context,
 	cfg config.Config,
 	logger log.Logger,
 	msg *types.MessageState,
@@ -27,15 +31,28 @@ func Broadcast(
 
 	// set up eth client
 	client, err := ethclient.Dial(cfg.Networks.Destination.Ethereum.RPC)
+	if err != nil {
+		return nil, fmt.Errorf("unable to dial ethereum client: %w", err)
+	}
 	defer client.Close()
 
+	backend := NewContractBackendWrapper(client)
+
 	privEcdsaKey, ethereumAddress, err := GetEcdsaKeyAddress(cfg.Networks.Minters[0].MinterPrivateKey)
 	if err != nil {
 		return nil, err
 	}
 
 	auth, err := bind.NewKeyedTransactorWithChainID(privEcdsaKey, big.NewInt(cfg.Networks.Destination.Ethereum.ChainId))
-	messageTransmitter, err := NewMessageTransmitter(common.HexToAddress(cfg.Networks.Source.Ethereum.MessageTransmitter), client)
+	if err != nil {
+		return nil, fmt.Errorf("unable to create auth: %w", err)
+	}
+
+	messageTransmitter, err := NewMessageTransmitter(common.HexToAddress(cfg.Networks.Source.Ethereum.MessageTransmitter), backend)
+	if err != nil {
+		return nil, fmt.Errorf("unable to create message transmitter: %w", err)
+	}
+
 	attestationBytes, err := hex.DecodeString(msg.Attestation[2:])
 	if err != nil {
 		return nil, errors.New("unable to decode message attestation")
@@ -52,6 +69,42 @@ func Broadcast(
 		nonce := sequenceMap.Next(cfg.Networks.Destination.Ethereum.DomainId)
 		auth.Nonce = big.NewInt(nonce)
 
+		// TODO remove
+		nextNonce, err := GetEthereumAccountNonce(cfg.Networks.Destination.Ethereum.RPC, ethereumAddress)
+		if err != nil {
+			logger.Error("unable to retrieve account number")
+		} else {
+			auth.Nonce = big.NewInt(nextNonce)
+		}
+		// TODO end remove
+
+		// check if nonce already used
+		co := &bind.CallOpts{
+			Pending: true,
+			Context: ctx,
+		}
+
+		logger.Debug("Checking if nonce was used for broadcast to Ethereum", "source_domain", msg.SourceDomain, "nonce", msg.Nonce)
+
+		key := append(
+			common.LeftPadBytes((big.NewInt(int64(msg.SourceDomain))).Bytes(), 4),
+			common.LeftPadBytes((big.NewInt(int64(msg.Nonce))).Bytes(), 8)...,
+		)
+
+		response, nonceErr := messageTransmitter.UsedNonces(co, [32]byte(crypto.Keccak256(key)))
+		if nonceErr != nil {
+			logger.Debug("Error querying whether nonce was used.   Continuing...")
+		} else {
+			fmt.Printf("received used nonce response: %d\n", response)
+			if response.Uint64() == uint64(1) {
+				// nonce has already been used, mark as complete
+				logger.Debug(fmt.Sprintf("This source domain/nonce has already been used: %d %d",
+					msg.SourceDomain, msg.Nonce))
+				msg.Status = types.Complete
+				return nil, errors.New("receive message was already broadcasted")
+			}
+		}
+
 		// broadcast txn
 		tx, err := messageTransmitter.ReceiveMessage(
 			auth,
@@ -65,20 +118,16 @@ func Broadcast(
 			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" {
-					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)
+					msg.Status = types.Complete
+					return nil, parsedErr
 				}
 
 				match, _ := regexp.MatchString("nonce too low: next nonce [0-9]+, tx nonce [0-9]+", parsedErr.Error())
 				if match {
 					numberRegex := regexp.MustCompile("[0-9]+")
 					nextNonce, err := strconv.ParseInt(numberRegex.FindAllString(parsedErr.Error(), 1)[0], 10, 0)
 					if err != nil {
-						nonce, err = GetEthereumAccountNonce(cfg.Networks.Destination.Ethereum.RPC, ethereumAddress)
+						nextNonce, err = GetEthereumAccountNonce(cfg.Networks.Destination.Ethereum.RPC, ethereumAddress)
 						if err != nil {
 							logger.Error("unable to retrieve account number")
 						}
@@ -88,6 +137,7 @@ func Broadcast(
 			}
 
 			// if it's not the last attempt, retry
+			// TODO increase the destination.ethereum.broadcast retries (3-5) and retry interval (15s).  By checking for used nonces, there is no gas cost for failed mints.
 			if attempt != cfg.Networks.Destination.Ethereum.BroadcastRetries {
 				logger.Info(fmt.Sprintf("Retrying in %d seconds", cfg.Networks.Destination.Ethereum.BroadcastRetryInterval))
 				time.Sleep(time.Duration(cfg.Networks.Destination.Ethereum.BroadcastRetryInterval) * time.Second)

cmd/ethereum/broadcast_test.go

@@ -0,0 +1,43 @@
+package ethereum_test
+
+import (
+	"context"
+	"math/big"
+	"testing"
+
+	"github.com/ethereum/go-ethereum/accounts/abi/bind"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/crypto"
+	"github.com/ethereum/go-ethereum/ethclient"
+	"github.com/strangelove-ventures/noble-cctp-relayer/cmd/ethereum"
+	"github.com/stretchr/testify/require"
+)
+
+func TestEthUsedNonce(t *testing.T) {
+	sourceDomain := uint32(4)
+	nonce := uint64(2970)
+
+	key := append(
+		common.LeftPadBytes((big.NewInt(int64(sourceDomain))).Bytes(), 4),
+		common.LeftPadBytes((big.NewInt(int64(nonce))).Bytes(), 8)...,
+	)
+
+	require.Equal(t, []byte("\x00\x00\x00\x04\x00\x00\x00\x00\x00\x00\x0B\x9A"), key)
+
+	client, err := ethclient.Dial("https://mainnet.infura.io/v3/e44b63a541e94b30a74a97447518c0ec")
+	require.NoError(t, err)
+	defer client.Close()
+
+	messageTransmitter, err := ethereum.NewMessageTransmitter(common.HexToAddress("0x0a992d191deec32afe36203ad87d7d289a738f81"), client)
+	require.NoError(t, err)
+
+	co := &bind.CallOpts{
+		Pending: true,
+		Context: context.TODO(),
+	}
+
+	response, err := messageTransmitter.UsedNonces(co, [32]byte(crypto.Keccak256(key)))
+	require.NoError(t, err)
+
+	require.Equal(t, big.NewInt(1), response)
+}

cmd/ethereum/contract_backend_wrapper.go

@@ -0,0 +1,28 @@
+package ethereum
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/ethclient"
+)
+
+type ContractBackendWrapper struct {
+	*ethclient.Client
+}
+
+func NewContractBackendWrapper(client *ethclient.Client) *ContractBackendWrapper {
+	return &ContractBackendWrapper{
+		Client: client,
+	}
+}
+
+func (c *ContractBackendWrapper) SendTransaction(ctx context.Context, tx *types.Transaction) error {
+	json, err := tx.MarshalJSON()
+	if err != nil {
+		return err
+	}
+	fmt.Printf("SendTransaction: %+v\n\nRAW: %s\n", tx, json)
+	return c.Client.SendTransaction(ctx, tx)
+}

cmd/noble/broadcast.go

@@ -26,12 +26,14 @@ import (
 	"github.com/cosmos/cosmos-sdk/types/tx/signing"
 	xauthsigning "github.com/cosmos/cosmos-sdk/x/auth/signing"
 	xauthtx "github.com/cosmos/cosmos-sdk/x/auth/tx"
+	"github.com/strangelove-ventures/noble-cctp-relayer/cmd/noble/cosmos"
 	"github.com/strangelove-ventures/noble-cctp-relayer/config"
 	"github.com/strangelove-ventures/noble-cctp-relayer/types"
 )
 
 // Broadcast broadcasts a message to Noble
 func Broadcast(
+	ctx context.Context,
 	cfg config.Config,
 	logger log.Logger,
 	msg *types.MessageState,
@@ -49,14 +51,14 @@ func Broadcast(
 	txBuilder := sdkContext.TxConfig.NewTxBuilder()
 	attestationBytes, err := hex.DecodeString(msg.Attestation[2:])
 	if err != nil {
-		return nil, errors.New("unable to decode message attestation")
+		return nil, fmt.Errorf("unable to decode message attestation")
 	}
 
 	// get priv key
 	nobleAddress := cfg.Networks.Minters[4].MinterAddress
 	keyBz, err := hex.DecodeString(cfg.Networks.Minters[4].MinterPrivateKey)
 	if err != nil {
-		return nil, errors.New(fmt.Sprintf("Unable to parse Noble private key"))
+		return nil, fmt.Errorf("unable to parse Noble private key")
 	}
 	privKey := secp256k1.PrivKey{Key: keyBz}
 
@@ -79,7 +81,22 @@ func Broadcast(
 		return nil, errors.New("failed to set up rpc client")
 	}
 
+	cc, err := cosmos.NewProvider(cfg.Networks.Source.Noble.RPC)
+	if err != nil {
+		return nil, fmt.Errorf("unable to build cosmos provider for noble: %w", err)
+	}
+
 	for attempt := 0; attempt <= cfg.Networks.Destination.Noble.BroadcastRetries; attempt++ {
+		used, err := cc.QueryUsedNonce(ctx, msg.SourceDomain, msg.Nonce)
+		if err != nil {
+			return nil, fmt.Errorf("unable to query used nonce: %w", err)
+		}
+
+		if used {
+			msg.Status = types.Complete
+			return nil, fmt.Errorf("noble cctp minter nonce %d already used", msg.Nonce)
+		}
+
 		logger.Info(fmt.Sprintf(
 			"Broadcasting %s message from %d to %d: with source tx hash %s",
 			msg.Type,
@@ -91,7 +108,7 @@ func Broadcast(
 		accountNumber, _, err := GetNobleAccountNumberSequence(cfg.Networks.Destination.Noble.API, nobleAddress)
 
 		if err != nil {
-			logger.Error("unable to retrieve account number")
+			return nil, fmt.Errorf("failed to retrieve account number and sequence: %w", err)
 		}
 
 		sigV2 := signing.SignatureV2{
@@ -119,16 +136,18 @@ func Broadcast(
 			sdkContext.TxConfig,
 			uint64(accountSequence),
 		)
-
-		err = txBuilder.SetSignatures(sigV2)
 		if err != nil {
-			return nil, err
+			return nil, fmt.Errorf("failed to sign tx: %w", err)
+		}
+
+		if err := txBuilder.SetSignatures(sigV2); err != nil {
+			return nil, fmt.Errorf("failed to set signatures: %w", err)
 		}
 
 		// Generated Protobuf-encoded bytes.
 		txBytes, err := sdkContext.TxConfig.TxEncoder()(txBuilder.GetTx())
 		if err != nil {
-			return nil, err
+			return nil, fmt.Errorf("failed to proto encode tx: %w", err)
 		}
 
 		rpcResponse, err := rpcClient.BroadcastTxSync(context.Background(), txBytes)