Process messages concurrently

.github/workflows/codeql.yml

@@ -37,6 +37,8 @@ jobs:
     steps:
     - name: Checkout repository
       uses: actions/checkout@v4
+      with:
+          submodules: recursive
 
     # Initializes the CodeQL tools for scanning.
     - name: Initialize CodeQL

.github/workflows/e2e.yml

@@ -19,6 +19,8 @@ jobs:
     steps:
       - name: Checkout awm-relayer repository
         uses: actions/checkout@v4
+        with:
+          submodules: recursive
 
       - name: Set Go version
         run: |
@@ -45,6 +47,8 @@ jobs:
 
       - name: Checkout awm-relayer repository
         uses: actions/checkout@v4
+        with:
+          submodules: recursive
 
       - name: Run E2E Tests
         run: AVALANCHEGO_BUILD_PATH=/tmp/e2e-test/avalanchego DATA_DIR=/tmp/e2e-test/data ./scripts/e2e_test.sh

.github/workflows/linter.yml

@@ -19,6 +19,8 @@ jobs:
     steps:
     - name: Checkout repository
       uses: actions/checkout@v4
+      with:
+          submodules: recursive
 
     - name: Set Go version
       run: |

.github/workflows/release.yml

@@ -17,6 +17,7 @@ jobs:
         with:
           fetch-depth: 0
           path: awm-relayer
+          submodules: recursive
 
       - name: Set Go version
         run: |

.github/workflows/snyk.yml

@@ -18,6 +18,7 @@ jobs:
         uses: actions/checkout@v4
         with:
           path: awm-relayer
+          submodules: recursive
 
       - name: Run Snyk
         uses: snyk/actions/golang@master

.github/workflows/test.yml

@@ -19,6 +19,8 @@ jobs:
     steps:
       - name: Checkout awm-relayer repository
         uses: actions/checkout@v4
+        with:
+          submodules: recursive
 
       - name: Set Go version
         run: |

.gitignore

@@ -9,3 +9,7 @@ relayer-config.json
 main.log
 server.log
 *.test
+
+# Foundry outputs
+cache/
+out/

.gitmodules

@@ -0,0 +1,6 @@
+[submodule "tests/contracts/lib/teleporter"]
+	path = tests/contracts/lib/teleporter
+	url = https://github.com/ava-labs/teleporter
+[submodule "tests/contracts/lib/forge-std"]
+	path = tests/contracts/lib/forge-std
+	url = https://github.com/foundry-rs/forge-std

.golangci.yml

@@ -3,13 +3,13 @@
 run:
   timeout: 3m
   tests: true
-  # skip auto-generated files.
-  skip-files:
-    - ".*mock.*"
 
 issues:
   # Maximum count of issues with the same text. Set to 0 to disable. Default is 3.
   max-same-issues: 0
+  # skip auto-generated files.
+  exclude-files:
+    - ".*mock.*"
 
 linters:
   disable-all: true

README.md

@@ -94,6 +94,10 @@ awm-relayer --version                                   Display awm-relayer vers
 awm-relayer --help                                      Display awm-relayer usage and exit.
 ```
 
+### Initialize the repository
+
+- Get all submodules: `git submodule update --init --recursive`
+
 ### Building
 
 Before building, be sure to install Go, which is required even if you're just building the Docker image.

database/utils.go

@@ -45,7 +45,7 @@ func CalculateStartingBlockHeight(
 	} else if err != nil {
 		// Otherwise, we've encountered an unknown database error
 		logger.Error(
-			"failed to get latest block from database",
+			"Failed to get latest block from database",
 			zap.String("relayerID", relayerID.ID.String()),
 			zap.Error(err),
 		)

main/main.go

@@ -19,11 +19,14 @@ import (
 	"github.com/ava-labs/avalanchego/utils/logging"
 	"github.com/ava-labs/awm-relayer/config"
 	"github.com/ava-labs/awm-relayer/database"
+	"github.com/ava-labs/awm-relayer/ethclient"
 	"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/ethereum/go-ethereum/common"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"go.uber.org/atomic"
@@ -122,7 +125,7 @@ func main() {
 	if logLevel <= logging.Debug {
 		networkLogLevel = logLevel
 	}
-	network, responseChans, err := peers.NewNetwork(
+	network, err := peers.NewNetwork(
 		networkLogLevel,
 		registerer,
 		&cfg,
@@ -200,18 +203,27 @@ func main() {
 	ticker := utils.NewTicker(cfg.DBWriteIntervalSeconds)
 	go ticker.Run()
 
-	manualWarpMessages := make(map[ids.ID][]*relayerTypes.WarpLogInfo)
+	// Gather manual Warp messages specified in the configuration
+	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.String("warpMessageBytes", hex.EncodeToString(msg.GetUnsignedMessageBytes())),
+				zap.Error(err),
+			)
+			panic(err)
+		}
+		warpLogInfo := relayerTypes.WarpMessageInfo{
+			SourceAddress:   msg.GetSourceAddress(),
+			UnsignedMessage: unsignedMsg,
 		}
 		manualWarpMessages[sourceBlockchainID] = append(manualWarpMessages[sourceBlockchainID], &warpLogInfo)
 	}
 
-	// Create relayers for each of the subnets configured as a source
+	// Create listeners for each of the subnets configured as a source
 	errGroup, ctx := errgroup.WithContext(context.Background())
 	for _, s := range cfg.SourceBlockchains {
 		blockchainID, err := ids.FromString(s.BlockchainID)
@@ -229,16 +241,15 @@ func main() {
 
 		// errgroup will cancel the context when the first goroutine returns an error
 		errGroup.Go(func() error {
-			// runRelayer runs until it errors or the context is cancelled by another goroutine
-			return runRelayer(
+			// runListener runs until it errors or the context is cancelled by another goroutine
+			return runListener(
 				ctx,
 				logger,
 				metrics,
 				db,
 				ticker,
 				*subnetInfo,
 				network,
-				responseChans[blockchainID],
 				destinationClients,
 				messageCreator,
 				health,
@@ -254,76 +265,178 @@ func main() {
 	)
 }
 
-// runRelayer creates a relayer instance for a subnet. It listens for warp messages on that subnet, and handles delivery to the destination
-func runRelayer(
+// runListener creates a Listener instance and the ApplicationRelayers for a subnet.
+// The Listener listens for warp messages on that subnet, and the ApplicationRelayers handle delivery to the destination
+func runListener(
 	ctx context.Context,
 	logger logging.Logger,
 	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 {
-	logger.Info(
-		"Creating relayer",
-		zap.String("originBlockchainID", sourceSubnetInfo.BlockchainID),
+	// Dial the eth client
+	ethClient, err := ethclient.DialWithConfig(
+		context.Background(),
+		sourceBlockchain.RPCEndpoint.BaseURL,
+		sourceBlockchain.RPCEndpoint.HTTPHeaders,
+		sourceBlockchain.RPCEndpoint.QueryParams,
 	)
+	if err != nil {
+		logger.Error(
+			"Failed to connect to node via RPC",
+			zap.String("blockchainID", sourceBlockchain.BlockchainID),
+			zap.Error(err),
+		)
+		return err
+	}
 
-	listener, err := relayer.NewListener(
+	// Create the ApplicationRelayers
+	applicationRelayers, minHeight, err := createApplicationRelayers(
+		ctx,
 		logger,
 		metrics,
 		db,
 		ticker,
-		sourceSubnetInfo,
+		sourceBlockchain,
 		network,
-		responseChan,
-		destinationClients,
 		messageCreator,
+		cfg,
+		ethClient,
+		destinationClients,
+	)
+	if err != nil {
+		logger.Error(
+			"Failed to create application relayers",
+			zap.String("blockchainID", sourceBlockchain.BlockchainID),
+			zap.Error(err),
+		)
+		return err
+	}
+	logger.Info(
+		"Created application relayers",
+		zap.String("blockchainID", sourceBlockchain.BlockchainID),
+	)
+
+	// Create the Listener
+	listener, err := relayer.NewListener(
+		logger,
+		sourceBlockchain,
 		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)),
+	err = listener.ProcessManualWarpMessages(logger, manualWarpMessages, sourceBlockchain)
+	if err != nil {
+		logger.Error(
+			"Failed to process manual Warp messages",
+			zap.String("blockchainID", sourceBlockchain.BlockchainID),
+			zap.Error(err),
 		)
-		err := listener.RouteManualWarpMessage(warpMessage)
-		if err != nil {
-			logger.Error(
-				"Failed to relay manual Warp message. Continuing.",
-				zap.Error(err),
-				zap.String("warpMessageBytes", hex.EncodeToString(warpMessage.UnsignedMsgBytes)),
-			)
-			continue
-		}
 	}
 
 	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
 	// Will only return on error or context cancellation
 	return listener.ProcessLogs(ctx)
 }
 
+func createApplicationRelayers(
+	ctx context.Context,
+	logger logging.Logger,
+	metrics *relayer.ApplicationRelayerMetrics,
+	db database.RelayerDatabase,
+	ticker *utils.Ticker,
+	sourceBlockchain config.SourceBlockchain,
+	network *peers.AppRequestNetwork,
+	messageCreator message.Creator,
+	cfg *config.Config,
+	srcEthClient ethclient.Client,
+	destinationClients map[ids.ID]vms.DestinationClient,
+) (map[common.Hash]*relayer.ApplicationRelayer, uint64, error) {
+	// Create the ApplicationRelayers
+	logger.Info(
+		"Creating application relayers",
+		zap.String("originBlockchainID", sourceBlockchain.BlockchainID),
+	)
+	applicationRelayers := make(map[common.Hash]*relayer.ApplicationRelayer)
+
+	currentHeight, err := srcEthClient.BlockNumber(context.Background())
+	if err != nil {
+		logger.Error(
+			"Failed to get current block height",
+			zap.Error(err),
+		)
+		return nil, 0, err
+	}
+
+	// Each ApplicationRelayer determines its starting height based on the database state.
+	// The Listener begins processing messages starting from the minimum height across all of the ApplicationRelayers
+	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 nil, 0, err
+		}
+		if minHeight == 0 || height < minHeight {
+			minHeight = height
+		}
+		applicationRelayer, err := relayer.NewApplicationRelayer(
+			logger,
+			metrics,
+			network,
+			messageCreator,
+			relayerID,
+			db,
+			ticker,
+			destinationClients[relayerID.DestinationBlockchainID],
+			sourceBlockchain,
+			height,
+			cfg,
+		)
+		if err != nil {
+			logger.Error(
+				"Failed to create application relayer",
+				zap.String("relayerID", relayerID.ID.String()),
+				zap.Error(err),
+			)
+			return nil, 0, err
+		}
+		applicationRelayers[relayerID.ID] = applicationRelayer
+	}
+	return applicationRelayers, minHeight, nil
+}
+
 func startMetricsServer(logger logging.Logger, gatherer prometheus.Gatherer, port uint16) {
 	http.Handle("/metrics", promhttp.HandlerFor(gatherer, promhttp.HandlerOpts{}))