Process missed blocks on startup

config/config.go

@@ -42,6 +42,7 @@ type SourceSubnet struct {
 	APINodeHost       string                           `mapstructure:"api-node-host" json:"api-node-host"`
 	APINodePort       uint32                           `mapstructure:"api-node-port" json:"api-node-port"`
 	EncryptConnection bool                             `mapstructure:"encrypt-connection" json:"encrypt-connection"`
+	RPCEndpoint       string                           `mapstructure:"rpc-endpoint" json:"rpc-endpoint"`
 	WSEndpoint        string                           `mapstructure:"ws-endpoint" json:"ws-endpoint"`
 	MessageContracts  map[string]MessageProtocolConfig `mapstructure:"message-contracts" json:"message-contracts"`
 }
@@ -62,6 +63,7 @@ type Config struct {
 	NetworkID          uint32              `mapstructure:"network-id" json:"network-id"`
 	PChainAPIURL       string              `mapstructure:"p-chain-api-url" json:"p-chain-api-url"`
 	EncryptConnection  bool                `mapstructure:"encrypt-connection" json:"encrypt-connection"`
+	StorageLocation    string              `mapstructure:"storage-location" json:"storage-location"`
 	SourceSubnets      []SourceSubnet      `mapstructure:"source-subnets" json:"source-subnets"`
 	DestinationSubnets []DestinationSubnet `mapstructure:"destination-subnets" json:"destination-subnets"`
 }
@@ -71,6 +73,7 @@ func SetDefaultConfigValues(v *viper.Viper) {
 	v.SetDefault(NetworkIDKey, constants.MainnetID)
 	v.SetDefault(PChainAPIURLKey, "https://api.avax.network")
 	v.SetDefault(EncryptConnectionKey, true)
+	v.SetDefault(StorageLocationKey, "./.awm-relayer-storage")
 }
 
 // BuildConfig constructs the relayer config using Viper.
@@ -99,6 +102,7 @@ func BuildConfig(v *viper.Viper) (Config, bool, error) {
 	cfg.NetworkID = v.GetUint32(NetworkIDKey)
 	cfg.PChainAPIURL = v.GetString(PChainAPIURLKey)
 	cfg.EncryptConnection = v.GetBool(EncryptConnectionKey)
+	cfg.StorageLocation = v.GetString(StorageLocationKey)
 	if err := v.UnmarshalKey(DestinationSubnetsKey, &cfg.DestinationSubnets); err != nil {
 		return Config{}, false, fmt.Errorf("failed to unmarshal destination subnets: %v", err)
 	}
@@ -215,6 +219,9 @@ func (s *SourceSubnet) Validate() error {
 	if _, err := url.ParseRequestURI(s.GetNodeWSEndpoint()); err != nil {
 		return fmt.Errorf("invalid relayer subscribe URL in source subnet configuration: %v", err)
 	}
+	if _, err := url.ParseRequestURI(s.GetNodeRPCEndpoint()); err != nil {
+		return fmt.Errorf("invalid relayer RPC URL in source subnet configuration: %v", err)
+	}
 
 	// Validate the VM specific settings
 	switch ParseVM(s.VM) {
@@ -267,15 +274,25 @@ func (s *DestinationSubnet) Validate() error {
 	return nil
 }
 
-func constructURL(protocol string, host string, port uint32, encrypt bool) string {
+func constructURL(protocol string, host string, port uint32, encrypt bool, chainIDStr string, subnetIDStr string) string {
+	var protocolPathMap = map[string]string{
+		"http": "rpc",
+		"ws":   "ws",
+	}
+	path := protocolPathMap[protocol]
+
 	if encrypt {
 		protocol = protocol + "s"
 	}
 	portStr := ""
 	if port != 0 {
 		portStr = fmt.Sprintf(":%d", port)
 	}
-	return fmt.Sprintf("%s://%s%s", protocol, host, portStr)
+	subnetID, _ := ids.FromString(subnetIDStr) // already validated in Validate()
+	if subnetID == constants.PrimaryNetworkID {
+		chainIDStr = cChainIdentifierString
+	}
+	return fmt.Sprintf("%s://%s%s/ext/bc/%s/%s", protocol, host, portStr, chainIDStr, path)
 }
 
 // Constructs an RPC endpoint for the subnet.
@@ -286,13 +303,38 @@ func (s *DestinationSubnet) GetNodeRPCEndpoint() string {
 	if s.RPCEndpoint != "" {
 		return s.RPCEndpoint
 	}
-	baseUrl := constructURL("http", s.APINodeHost, s.APINodePort, s.EncryptConnection)
-	chainID := s.ChainID
-	subnetID, _ := ids.FromString(s.SubnetID) // already validated in Validate()
-	if subnetID == constants.PrimaryNetworkID {
-		chainID = cChainIdentifierString
+
+	// Save this result for future use
+	s.RPCEndpoint = constructURL(
+		"http",
+		s.APINodeHost,
+		s.APINodePort,
+		s.EncryptConnection,
+		s.ChainID,
+		s.SubnetID,
+	)
+	return s.RPCEndpoint
+}
+
+// Constructs an RPC endpoint for the subnet.
+// If the RPCEndpoint field is set in the configuration, returns that directly.
+// Otherwise, constructs the endpoint from the APINodeHost, APINodePort, and EncryptConnection fields,
+// following the /ext/bc/{chainID}/rpc format.
+func (s *SourceSubnet) GetNodeRPCEndpoint() string {
+	if s.RPCEndpoint != "" {
+		return s.RPCEndpoint
 	}
-	return fmt.Sprintf("%s/ext/bc/%s/rpc", baseUrl, chainID)
+
+	// Save this result for future use
+	s.RPCEndpoint = constructURL(
+		"http",
+		s.APINodeHost,
+		s.APINodePort,
+		s.EncryptConnection,
+		s.ChainID,
+		s.SubnetID,
+	)
+	return s.RPCEndpoint
 }
 
 // Constructs a WS endpoint for the subnet.
@@ -303,13 +345,17 @@ func (s *SourceSubnet) GetNodeWSEndpoint() string {
 	if s.WSEndpoint != "" {
 		return s.WSEndpoint
 	}
-	baseUrl := constructURL("ws", s.APINodeHost, s.APINodePort, s.EncryptConnection)
-	chainID := s.ChainID
-	subnetID, _ := ids.FromString(s.SubnetID) // already validated in Validate()
-	if subnetID == constants.PrimaryNetworkID {
-		chainID = cChainIdentifierString
-	}
-	return fmt.Sprintf("%s/ext/bc/%s/ws", baseUrl, chainID)
+
+	// Save this result for future use
+	s.WSEndpoint = constructURL(
+		"ws",
+		s.APINodeHost,
+		s.APINodePort,
+		s.EncryptConnection,
+		s.ChainID,
+		s.SubnetID,
+	)
+	return s.WSEndpoint
 }
 
 // Get the private key and derive the wallet address from a relayer's configured private key for a given destination subnet.

config/config_test.go

@@ -124,10 +124,11 @@ func TestGetDestinationRPCEndpoint(t *testing.T) {
 	}
 }
 
-func TestGetSourceSubnetWSEndpoint(t *testing.T) {
+func TestGetSourceSubnetEndpoints(t *testing.T) {
 	testCases := []struct {
-		s              SourceSubnet
-		expectedResult string
+		s                 SourceSubnet
+		expectedWsResult  string
+		expectedRpcResult string
 	}{
 		{
 			s: SourceSubnet{
@@ -137,7 +138,8 @@ func TestGetSourceSubnetWSEndpoint(t *testing.T) {
 				ChainID:           testChainID,
 				SubnetID:          testSubnetID,
 			},
-			expectedResult: fmt.Sprintf("ws://127.0.0.1:9650/ext/bc/%s/ws", testChainID),
+			expectedWsResult:  fmt.Sprintf("ws://127.0.0.1:9650/ext/bc/%s/ws", testChainID),
+			expectedRpcResult: fmt.Sprintf("http://127.0.0.1:9650/ext/bc/%s/rpc", testChainID),
 		},
 		{
 			s: SourceSubnet{
@@ -147,7 +149,8 @@ func TestGetSourceSubnetWSEndpoint(t *testing.T) {
 				ChainID:           testChainID,
 				SubnetID:          testSubnetID,
 			},
-			expectedResult: fmt.Sprintf("wss://127.0.0.1:9650/ext/bc/%s/ws", testChainID),
+			expectedWsResult:  fmt.Sprintf("wss://127.0.0.1:9650/ext/bc/%s/ws", testChainID),
+			expectedRpcResult: fmt.Sprintf("https://127.0.0.1:9650/ext/bc/%s/rpc", testChainID),
 		},
 		{
 			s: SourceSubnet{
@@ -157,7 +160,8 @@ func TestGetSourceSubnetWSEndpoint(t *testing.T) {
 				ChainID:           testChainID,
 				SubnetID:          testSubnetID,
 			},
-			expectedResult: fmt.Sprintf("ws://api.avax.network/ext/bc/%s/ws", testChainID),
+			expectedWsResult:  fmt.Sprintf("ws://api.avax.network/ext/bc/%s/ws", testChainID),
+			expectedRpcResult: fmt.Sprintf("http://api.avax.network/ext/bc/%s/rpc", testChainID),
 		},
 		{
 			s: SourceSubnet{
@@ -167,7 +171,8 @@ func TestGetSourceSubnetWSEndpoint(t *testing.T) {
 				ChainID:           testChainID,
 				SubnetID:          primarySubnetID,
 			},
-			expectedResult: "ws://127.0.0.1:9650/ext/bc/C/ws",
+			expectedWsResult:  "ws://127.0.0.1:9650/ext/bc/C/ws",
+			expectedRpcResult: "http://127.0.0.1:9650/ext/bc/C/rpc",
 		},
 		{
 			s: SourceSubnet{
@@ -176,15 +181,17 @@ func TestGetSourceSubnetWSEndpoint(t *testing.T) {
 				APINodePort:       9650,
 				ChainID:           testChainID,
 				SubnetID:          testSubnetID,
-				WSEndpoint:        "wss://subnets.avax.network/mysubnet/ws", // overrides all other settings used to construct the endpoint
+				WSEndpoint:        "wss://subnets.avax.network/mysubnet/ws",    // overrides all other settings used to construct the endpoint
+				RPCEndpoint:       "https://subnets.avax.network/mysubnet/rpc", // overrides all other settings used to construct the endpoint
 			},
-			expectedResult: "wss://subnets.avax.network/mysubnet/ws",
+			expectedWsResult:  "wss://subnets.avax.network/mysubnet/ws",
+			expectedRpcResult: "https://subnets.avax.network/mysubnet/rpc",
 		},
 	}
 
 	for i, testCase := range testCases {
-		res := testCase.s.GetNodeWSEndpoint()
-		assert.Equal(t, testCase.expectedResult, res, fmt.Sprintf("test case %d failed", i))
+		assert.Equal(t, testCase.expectedWsResult, testCase.s.GetNodeWSEndpoint(), fmt.Sprintf("test case %d failed", i))
+		assert.Equal(t, testCase.expectedRpcResult, testCase.s.GetNodeRPCEndpoint(), fmt.Sprintf("test case %d failed", i))
 	}
 }
 

config/keys.go

@@ -13,4 +13,5 @@ const (
 	DestinationSubnetsKey = "destination-subnets"
 	EncryptConnectionKey  = "encrypt-connection"
 	AccountPrivateKeyKey  = "account-private-key"
+	StorageLocationKey    = "storage-location"
 )

database/database.go

@@ -0,0 +1,25 @@
+// Copyright (C) 2023, Ava Labs, Inc. All rights reserved.
+// See the file LICENSE for licensing terms.
+
+package database
+
+import (
+	"github.com/ava-labs/avalanchego/ids"
+	"github.com/pkg/errors"
+)
+
+const (
+	LatestProcessedBlockKey = "latestProcessedBlock"
+)
+
+var (
+	ErrKeyNotFound              = errors.New("key not found")
+	ErrChainNotFound            = errors.New("no database for chain")
+	ErrDatabaseMisconfiguration = errors.New("database misconfiguration")
+)
+
+// RelayerDatabase is a key-value store for relayer state, with each chainID maintaining its own state
+type RelayerDatabase interface {
+	Get(chainID ids.ID, key []byte) ([]byte, error)
+	Put(chainID ids.ID, key []byte, value []byte) error
+}