Check if message already delivered

messages/message_manager.go

@@ -20,6 +20,7 @@ import (
 // for each message protocol, and performs the sending to the destination chain.
 type MessageManager interface {
 	// ShouldSendMessage returns true if the message should be sent to the destination chain
+	// If an error is returned, the boolean should be ignored by the caller.
 	ShouldSendMessage(warpMessageInfo *vmtypes.WarpMessageInfo, destinationChainID ids.ID) (bool, error)
 	// SendMessage sends the signed message to the destination chain. The payload parsed according to
 	// the VM rules is also passed in, since MessageManager does not assume any particular VM

messages/teleporter/message.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/ava-labs/subnet-evm/accounts/abi"
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/ava-labs/avalanchego/ids"
 )
 
 // TeleporterMessage contains the Teleporter message, including
@@ -34,6 +35,13 @@ type ReceiveCrossChainMessageInput struct {
 	RelayerRewardAddress common.Address `json:"relayerRewardAddress"`
 }
 
+// MessageReceivedInput is the input to the MessageReceived
+// in the contract deployed on the receiving chain
+type MessageReceivedInput struct {
+	OriginChainID ids.ID  `json:"relayerRewardAddress"`
+	MessageID	 *big.Int `json:"messageID"`
+}
+
 // unpack Teleporter message bytes according to EVM ABI encoding rules
 func unpackTeleporterMessage(messageBytes []byte) (*TeleporterMessage, error) {
 	args := abi.Arguments{
@@ -60,3 +68,13 @@ func unpackTeleporterMessage(messageBytes []byte) (*TeleporterMessage, error) {
 func packReceiverMessage(inputStruct ReceiveCrossChainMessageInput) ([]byte, error) {
 	return EVMTeleporterContractABI.Pack("receiveCrossChainMessage", inputStruct.RelayerRewardAddress)
 }
+
+func packMessageReceivedMessage(inputStruct MessageReceivedInput) ([]byte, error) {
+	return EVMTeleporterContractABI.Pack("messageReceived", inputStruct.OriginChainID, inputStruct.MessageID)
+}
+
+func unpackMessageReceivedResult(result []byte) (bool, error) {
+	var success bool
+	err := EVMTeleporterContractABI.UnpackIntoInterface(&success, "messageReceived", result)
+	return success, err
+}

messages/teleporter/message_manager.go

@@ -4,7 +4,9 @@
 package teleporter
 
 import (
+	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 
 	"github.com/ava-labs/avalanchego/cache"
@@ -14,6 +16,8 @@ import (
 	"github.com/ava-labs/awm-relayer/config"
 	"github.com/ava-labs/awm-relayer/vms"
 	"github.com/ava-labs/awm-relayer/vms/vmtypes"
+	"github.com/ava-labs/subnet-evm/ethclient"
+	"github.com/ava-labs/subnet-evm/interfaces"
 	"github.com/ethereum/go-ethereum/common"
 	"go.uber.org/zap"
 )
@@ -70,6 +74,24 @@ func NewMessageManager(
 	}, nil
 }
 
+func isDestination(messageDestinationID ids.ID, allowedDestinationID ids.ID) bool {
+	return messageDestinationID == allowedDestinationID
+}
+
+func isAllowedRelayer(allowedRelayers []common.Address, eoa common.Address) bool {
+	// If no allowed relayer addresses were set, then anyone can relay it.
+	if len(allowedRelayers) == 0 {
+		return true
+	}
+
+	for _, addr := range allowedRelayers {
+		if addr == eoa {
+			return true
+		}
+	}
+	return false
+}
+
 // ShouldSendMessage returns true if the message should be sent to the destination chain
 func (m *messageManager) ShouldSendMessage(warpMessageInfo *vmtypes.WarpMessageInfo, destinationChainID ids.ID) (bool, error) {
 	// Unpack the teleporter message and add it to the cache
@@ -87,13 +109,59 @@ func (m *messageManager) ShouldSendMessage(warpMessageInfo *vmtypes.WarpMessageI
 	if !ok {
 		return false, fmt.Errorf("relayer not configured to deliver to destination. destinationChainID=%s", destinationChainID.String())
 	}
-	if !destinationClient.Allowed(destinationChainID, teleporterMessage.AllowedRelayerAddresses) {
+	senderAddress := destinationClient.SenderAddress()
+	if !isAllowedRelayer(teleporterMessage.AllowedRelayerAddresses, senderAddress) {
+		m.logger.Info("Relayer EOA not allowed to deliver this message.")
+		return false, nil
+	}
+	if !isDestination(destinationChainID, destinationClient.DestinationChainID()) {
 		m.logger.Info(
-			"Relayer not allowed to deliver to chain.",
+			"Destination chain ID for message not supported by relayer.",
 			zap.String("destinationChainID", destinationChainID.String()),
 		)
 		return false, nil
 	}
+
+	// Check if the message has already been delivered to the destination chain
+	client, ok := destinationClient.Client().(ethclient.Client)
+	if !ok {
+		m.logger.Info("Destination client is not an Ethereum client.")
+		return false, errors.New("destination client is not an Ethereum client")
+	}
+
+	data, err := packMessageReceivedMessage(MessageReceivedInput{
+		OriginChainID: warpMessageInfo.WarpUnsignedMessage.SourceChainID,
+		MessageID:     teleporterMessage.MessageID,
+	})
+	if err != nil {
+		m.logger.Info("Failed packing messageReceived call data.")
+		return false, err
+	}
+	protocolAddress := common.BytesToAddress(m.protocolAddress[:])
+	callMessage := interfaces.CallMsg{
+		To:   &protocolAddress,
+		From: senderAddress,
+		Data: data,
+	}
+	result, err := client.CallContract(context.Background(), callMessage, nil)
+	if err != nil {
+		m.logger.Info("Failed calling messageReceived method on destination chain.")
+		return false, err
+	}
+	// check the contract call result
+	received, err := unpackMessageReceivedResult(result)
+	if err != nil {
+		return false, err
+	}
+	if received {
+		m.logger.Info(
+			"Message already deliverd to destination.",
+			zap.String("destinationChainID", destinationChainID.String()),
+			zap.String("teleporterMessageID", teleporterMessage.MessageID.String()),
+		)
+		return false, nil
+	}
+
 	// Cache the message so it can be reused in SendMessage
 	m.teleporterMessageCache.Put(warpMessageInfo.WarpUnsignedMessage.ID(), teleporterMessage)
 	return true, nil

vms/destination_client.go

@@ -22,8 +22,14 @@ type DestinationClient interface {
 	// TODO: Make generic for any VM.
 	SendTx(signedMessage *warp.Message, toAddress string, gasLimit uint64, callData []byte) error
 
-	// Allowed checks if the relayer is allowed to relay the message according to the VM rules and the message metadata
-	Allowed(chainID ids.ID, allowedRelayers []common.Address) bool
+	// Client returns the underlying client for the destination chain
+	Client() interface{}
+
+	// SenderAddress returns the address of the relayer on the destination chain
+	SenderAddress() common.Address
+
+	// DestinationChainID returns the ID of the destination chain
+	DestinationChainID() ids.ID
 }
 
 func NewDestinationClient(logger logging.Logger, subnetInfo config.DestinationSubnet) (DestinationClient, error) {

vms/evm/destination_client.go

@@ -95,28 +95,28 @@ func NewDestinationClient(logger logging.Logger, subnetInfo config.DestinationSu
 	}, nil
 }
 
-func (tdc *destinationClient) SendTx(signedMessage *avalancheWarp.Message,
+func (c *destinationClient) SendTx(signedMessage *avalancheWarp.Message,
 	toAddress string,
 	gasLimit uint64,
 	callData []byte) error {
 	// Synchronize teleporter message requests to the same destination chain so that message ordering is preserved
-	tdc.lock.Lock()
-	defer tdc.lock.Unlock()
+	c.lock.Lock()
+	defer c.lock.Unlock()
 	// We need the global 32-byte representation of the destination chain ID, as well as the destination's configured chainID
 	// Without the destination's configured chainID, transaction signature verification will fail
-	destinationChainIDBigInt, err := tdc.client.ChainID(context.Background())
+	destinationChainIDBigInt, err := c.client.ChainID(context.Background())
 	if err != nil {
-		tdc.logger.Error(
+		c.logger.Error(
 			"Failed to get chain ID from destination chain endpoint",
 			zap.Error(err),
 		)
 		return err
 	}
 
 	// Get the current base fee estimation, which is based on the previous blocks gas usage.
-	baseFee, err := tdc.client.EstimateBaseFee(context.Background())
+	baseFee, err := c.client.EstimateBaseFee(context.Background())
 	if err != nil {
-		tdc.logger.Error(
+		c.logger.Error(
 			"Failed to get base fee",
 			zap.Error(err),
 		)
@@ -125,9 +125,9 @@ func (tdc *destinationClient) SendTx(signedMessage *avalancheWarp.Message,
 
 	// Get the suggested gas tip cap of the network
 	// TODO: Add a configurable ceiling to this value
-	gasTipCap, err := tdc.client.SuggestGasTipCap(context.Background())
+	gasTipCap, err := c.client.SuggestGasTipCap(context.Background())
 	if err != nil {
-		tdc.logger.Error(
+		c.logger.Error(
 			"Failed to get gas tip cap",
 			zap.Error(err),
 		)
@@ -146,7 +146,7 @@ func (tdc *destinationClient) SendTx(signedMessage *avalancheWarp.Message,
 	// Construct the actual transaction to broadcast on the destination chain
 	tx := types.NewTx(&types.DynamicFeeTx{
 		ChainID:   destinationChainIDBigInt,
-		Nonce:     tdc.currentNonce,
+		Nonce:     c.currentNonce,
 		To:        &to,
 		Gas:       gasLimit,
 		GasFeeCap: gasFeeCap,
@@ -163,17 +163,17 @@ func (tdc *destinationClient) SendTx(signedMessage *avalancheWarp.Message,
 
 	// Sign and send the transaction on the destination chain
 	signer := types.LatestSignerForChainID(destinationChainIDBigInt)
-	signedTx, err := types.SignTx(tx, signer, tdc.pk)
+	signedTx, err := types.SignTx(tx, signer, c.pk)
 	if err != nil {
-		tdc.logger.Error(
+		c.logger.Error(
 			"Failed to sign transaction",
 			zap.Error(err),
 		)
 		return err
 	}
 
-	if err := tdc.client.SendTransaction(context.Background(), signedTx); err != nil {
-		tdc.logger.Error(
+	if err := c.client.SendTransaction(context.Background(), signedTx); err != nil {
+		c.logger.Error(
 			"Failed to send transaction",
 			zap.Error(err),
 		)
@@ -182,43 +182,23 @@ func (tdc *destinationClient) SendTx(signedMessage *avalancheWarp.Message,
 
 	// Increment the nonce to use on the destination chain now that we've sent
 	// a transaction using the current value.
-	tdc.currentNonce++
-	tdc.logger.Info(
+	c.currentNonce++
+	c.logger.Info(
 		"Sent transaction",
 		zap.String("txID", signedTx.Hash().String()),
 	)
 
 	return nil
 }
 
-func (tdc *destinationClient) isDestination(chainID ids.ID) bool {
-	if chainID != tdc.destinationChainID {
-		tdc.logger.Info(
-			"Destination chain ID for message not supported by relayer.",
-			zap.String("chainID", chainID.String()),
-		)
-		return false
-	}
-	return true
+func (c *destinationClient) Client() interface{} {
+	return c.client
 }
 
-func (tdc *destinationClient) isAllowedRelayer(allowedRelayers []common.Address) bool {
-	// If no allowed relayer addresses were set, then anyone can relay it.
-	if len(allowedRelayers) == 0 {
-		return true
-	}
-
-	for _, addr := range allowedRelayers {
-		if addr == tdc.eoa {
-			return true
-		}
-	}
-
-	tdc.logger.Info("Relayer EOA not allowed to deliver this message.")
-	return false
+func (c *destinationClient) SenderAddress() common.Address {
+	return c.eoa
 }
 
-func (tdc *destinationClient) Allowed(chainID ids.ID, allowedRelayers []common.Address) bool {
-	return tdc.isDestination(chainID) &&
-		tdc.isAllowedRelayer(allowedRelayers)
+func (c *destinationClient) DestinationChainID() ids.ID {
+	return c.destinationChainID
 }