Process messages concurrently

go.mod

@@ -1,6 +1,6 @@
 module github.com/ava-labs/awm-relayer
 
-go 1.21.10
+go 1.22.3
 
 require (
 	github.com/ava-labs/avalanche-network-runner v1.7.6

main/main.go

@@ -21,9 +21,12 @@ import (
 	"github.com/ava-labs/awm-relayer/database"
 	"github.com/ava-labs/awm-relayer/peers"
 	"github.com/ava-labs/awm-relayer/relayer"
+	"github.com/ava-labs/awm-relayer/types"
 	relayerTypes "github.com/ava-labs/awm-relayer/types"
 	"github.com/ava-labs/awm-relayer/utils"
 	"github.com/ava-labs/awm-relayer/vms"
+	"github.com/ava-labs/subnet-evm/ethclient"
+	"github.com/ethereum/go-ethereum/common"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"go.uber.org/atomic"
@@ -97,7 +100,7 @@ func main() {
 	if logLevel <= logging.Debug {
 		networkLogLevel = logLevel
 	}
-	network, responseChans, err := peers.NewNetwork(
+	network, err := peers.NewNetwork(
 		networkLogLevel,
 		registerer,
 		&cfg,
@@ -175,13 +178,20 @@ func main() {
 	ticker := utils.NewTicker(cfg.DBWriteIntervalSeconds)
 	go ticker.Run()
 
-	manualWarpMessages := make(map[ids.ID][]*relayerTypes.WarpLogInfo)
+	manualWarpMessages := make(map[ids.ID][]*relayerTypes.WarpMessageInfo)
 	for _, msg := range cfg.ManualWarpMessages {
 		sourceBlockchainID := msg.GetSourceBlockchainID()
-
-		warpLogInfo := relayerTypes.WarpLogInfo{
-			SourceAddress:    msg.GetSourceAddress(),
-			UnsignedMsgBytes: msg.GetUnsignedMessageBytes(),
+		unsignedMsg, err := types.UnpackWarpMessage(msg.GetUnsignedMessageBytes())
+		if err != nil {
+			logger.Error(
+				"Failed to unpack manual Warp message",
+				zap.Error(err),
+			)
+			panic(err)
+		}
+		warpLogInfo := relayerTypes.WarpMessageInfo{
+			SourceAddress:   msg.GetSourceAddress(),
+			UnsignedMessage: unsignedMsg,
 		}
 		manualWarpMessages[sourceBlockchainID] = append(manualWarpMessages[sourceBlockchainID], &warpLogInfo)
 	}
@@ -213,7 +223,6 @@ func main() {
 				ticker,
 				*subnetInfo,
 				network,
-				responseChans[blockchainID],
 				destinationClients,
 				messageCreator,
 				health,
@@ -236,62 +245,130 @@ func runRelayer(
 	metrics *relayer.ApplicationRelayerMetrics,
 	db database.RelayerDatabase,
 	ticker *utils.Ticker,
-	sourceSubnetInfo config.SourceBlockchain,
+	sourceBlockchain config.SourceBlockchain,
 	network *peers.AppRequestNetwork,
-	responseChan chan message.InboundMessage,
 	destinationClients map[ids.ID]vms.DestinationClient,
 	messageCreator message.Creator,
 	relayerHealth *atomic.Bool,
-	manualWarpMessages []*relayerTypes.WarpLogInfo,
+	manualWarpMessages []*relayerTypes.WarpMessageInfo,
 	cfg *config.Config,
 ) error {
+	// Create the application relayers
 	logger.Info(
-		"Creating relayer",
-		zap.String("originBlockchainID", sourceSubnetInfo.BlockchainID),
+		"Creating application relayers",
+		zap.String("originBlockchainID", sourceBlockchain.BlockchainID),
 	)
+	ethClient, err := ethclient.Dial(sourceBlockchain.RPCEndpoint)
+	if err != nil {
+		logger.Error(
+			"Failed to connect to node via RPC",
+			zap.String("blockchainID", sourceBlockchain.BlockchainID),
+			zap.Error(err),
+		)
+		return err
+	}
+	currentHeight, err := ethClient.BlockNumber(context.Background())
+	if err != nil {
+		logger.Error(
+			"Failed to get current block height",
+			zap.Error(err),
+		)
+		return err
+	}
+	applicationRelayers := make(map[common.Hash]*relayer.ApplicationRelayer)
+	minHeight := uint64(0)
+
+	for _, relayerID := range database.GetSourceBlockchainRelayerIDs(&sourceBlockchain) {
+		height, err := database.CalculateStartingBlockHeight(
+			logger,
+			db,
+			relayerID,
+			sourceBlockchain.ProcessHistoricalBlocksFromHeight,
+			currentHeight,
+		)
+		if err != nil {
+			logger.Error(
+				"Failed to calculate starting block height",
+				zap.String("relayerID", relayerID.ID.String()),
+				zap.Error(err),
+			)
+			return err
+		}
+		if minHeight == 0 || height < minHeight {
+			minHeight = height
+		}
+		applicationRelayer, err := relayer.NewApplicationRelayer(
+			logger,
+			metrics,
+			network,
+			messageCreator,
+			relayerID,
+			db,
+			ticker,
+			sourceBlockchain,
+			height,
+			cfg,
+		)
+		if err != nil {
+			logger.Error(
+				"Failed to create application relayer",
+				zap.String("relayerID", relayerID.ID.String()),
+				zap.Error(err),
+			)
+			return err
+		}
+		applicationRelayers[relayerID.ID] = applicationRelayer
+	}
 
+	logger.Info(
+		"Creating listener",
+		zap.String("originBlockchainID", sourceBlockchain.BlockchainID),
+	)
 	listener, err := relayer.NewListener(
 		logger,
 		metrics,
 		db,
 		ticker,
-		sourceSubnetInfo,
+		sourceBlockchain,
 		network,
-		responseChan,
 		destinationClients,
 		messageCreator,
 		relayerHealth,
 		cfg,
+		applicationRelayers,
+		minHeight,
+		ethClient,
 	)
 	if err != nil {
 		return fmt.Errorf("failed to create listener instance: %w", err)
 	}
 	logger.Info(
 		"Created listener",
-		zap.String("blockchainID", sourceSubnetInfo.BlockchainID),
+		zap.String("blockchainID", sourceBlockchain.BlockchainID),
 	)
 
 	// Send any messages that were specified in the configuration
 	for _, warpMessage := range manualWarpMessages {
 		logger.Info(
 			"Relaying manual Warp message",
-			zap.String("blockchainID", sourceSubnetInfo.BlockchainID),
-			zap.String("warpMessageBytes", hex.EncodeToString(warpMessage.UnsignedMsgBytes)),
+			zap.String("blockchainID", sourceBlockchain.BlockchainID),
+			zap.String("warpMessageBytes", hex.EncodeToString(warpMessage.UnsignedMessage.Bytes())),
 		)
-		err := listener.RouteManualWarpMessage(warpMessage)
+		appRelayer, handler, err := listener.GetAppRelayerMessageHandler(0, warpMessage)
 		if err != nil {
 			logger.Error(
-				"Failed to relay manual Warp message. Continuing.",
+				"Failed to parse manual Warp message. Continuing.",
 				zap.Error(err),
-				zap.String("warpMessageBytes", hex.EncodeToString(warpMessage.UnsignedMsgBytes)),
+				zap.String("warpMessageBytes", hex.EncodeToString(warpMessage.UnsignedMessage.Bytes())),
 			)
 			continue
 		}
+		appRelayer.ProcessMessage(handler)
 	}
 
 	logger.Info(
 		"Listener initialized. Listening for messages to relay.",
-		zap.String("originBlockchainID", sourceSubnetInfo.BlockchainID),
+		zap.String("originBlockchainID", sourceBlockchain.BlockchainID),
 	)
 
 	// Wait for logs from the subscribed node

messages/message_manager.go

@@ -6,63 +6,35 @@
 package messages
 
 import (
-	"fmt"
-
 	"github.com/ava-labs/avalanchego/ids"
-	"github.com/ava-labs/avalanchego/utils/logging"
 	"github.com/ava-labs/avalanchego/vms/platformvm/warp"
-	"github.com/ava-labs/awm-relayer/config"
-	offchainregistry "github.com/ava-labs/awm-relayer/messages/off-chain-registry"
-	"github.com/ava-labs/awm-relayer/messages/teleporter"
-	"github.com/ava-labs/awm-relayer/vms"
 	"github.com/ethereum/go-ethereum/common"
 )
 
 // MessageManager is specific to each message protocol. The interface handles choosing which messages to send
 // for each message protocol, and performs the sending to the destination chain.
 type MessageManager interface {
+	NewMessageHandler(unsignedMessage *warp.UnsignedMessage) (MessageHandler, error)
+}
+
+type MessageHandler 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(unsignedMessage *warp.UnsignedMessage, destinationBlockchainID ids.ID) (bool, error)
+	ShouldSendMessage(destinationBlockchainID 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
 	SendMessage(signedMessage *warp.Message, destinationBlockchainID ids.ID) error
 
 	// GetMessageRoutingInfo returns the source chain ID, origin sender address, destination chain ID, and destination address
-	GetMessageRoutingInfo(unsignedMessage *warp.UnsignedMessage) (
+	GetMessageRoutingInfo() (
 		ids.ID,
 		common.Address,
 		ids.ID,
 		common.Address,
 		error,
 	)
-}
 
-// NewMessageManager constructs a MessageManager for a particular message protocol, defined by the message protocol address and config
-// Note that DestinationClients may be invoked concurrently by many MessageManagers, so it is assumed that they are implemented in a thread-safe way
-func NewMessageManager(
-	logger logging.Logger,
-	messageProtocolAddress common.Address,
-	messageProtocolConfig config.MessageProtocolConfig,
-	destinationClients map[ids.ID]vms.DestinationClient,
-) (MessageManager, error) {
-	format := messageProtocolConfig.MessageFormat
-	switch config.ParseMessageProtocol(format) {
-	case config.TELEPORTER:
-		return teleporter.NewMessageManager(
-			logger,
-			messageProtocolAddress,
-			messageProtocolConfig,
-			destinationClients,
-		)
-	case config.OFF_CHAIN_REGISTRY:
-		return offchainregistry.NewMessageManager(
-			logger,
-			messageProtocolConfig,
-			destinationClients,
-		)
-	default:
-		return nil, fmt.Errorf("invalid message format %s", format)
-	}
+	// GetUnsignedMessage returns the unsigned message
+	GetUnsignedMessage() *warp.UnsignedMessage
 }