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) {
@@ -295,6 +302,23 @@ func (s *DestinationSubnet) GetNodeRPCEndpoint() string {
 	return fmt.Sprintf("%s/ext/bc/%s/rpc", baseUrl, chainID)
 }
 
+// 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
+	}
+	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
+	}
+	return fmt.Sprintf("%s/ext/bc/%s/rpc", baseUrl, chainID)
+}
+
 // Constructs a WS endpoint for the subnet.
 // If the WSEndpoint field is set in the configuration, returns that directly.
 // Otherwise, constructs the endpoint from the APINodeHost, APINodePort, and EncryptConnection fields,

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,13 @@
+package database
+
+import "github.com/ava-labs/avalanchego/ids"
+
+const (
+	LatestSeenBlockKey = "latestSeenBlock"
+)
+
+// 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
+}

database/json_file_storage.go

@@ -0,0 +1,172 @@
+package database
+
+import (
+	"encoding/json"
+	"os"
+	"path/filepath"
+	"sync"
+
+	"github.com/ava-labs/avalanchego/ids"
+	"github.com/ava-labs/avalanchego/utils/logging"
+	"github.com/pkg/errors"
+	"go.uber.org/zap"
+)
+
+var _ RelayerDatabase = &JSONFileStorage{}
+
+var (
+	fileDoesNotExistErr = errors.New("JSON database file does not exist")
+)
+
+type chainState map[string]string
+
+// JSONFileStorage implements RelayerDatabase
+type JSONFileStorage struct {
+	// the directory where the json files are stored
+	dir string
+
+	// Each network has its own mutex
+	// The chainIDs used to index the JSONFileStorage are created at initialization
+	// and are not modified afterwards, so we don't need to lock the map itself.
+	mutexes map[ids.ID]*sync.RWMutex
+	logger  logging.Logger
+}
+
+// NewJSONFileStorage creates a new JSONFileStorage instance
+func NewJSONFileStorage(logger logging.Logger, dir string, networks []ids.ID) (*JSONFileStorage, error) {
+	storage := &JSONFileStorage{
+		dir:     filepath.Clean(dir),
+		mutexes: make(map[ids.ID]*sync.RWMutex),
+		logger:  logger,
+	}
+
+	for _, network := range networks {
+		storage.mutexes[network] = &sync.RWMutex{}
+	}
+
+	_, err := os.Stat(dir)
+	if err == nil {
+		// dir already exist
+		// return the existing storage
+		return storage, nil
+	}
+
+	// 0755: The owner can read, write, execute.
+	// Everyone else can read and execute but not modify the file.
+	err = os.MkdirAll(dir, 0755)
+	if err != nil {
+		storage.logger.Error("failed to create dir",
+			zap.Error(err))
+		return nil, err
+	}
+
+	return storage, nil
+}
+
+// Get the latest chain state from the json database, and retrieve the value from the key
+func (s *JSONFileStorage) Get(chainID ids.ID, key []byte) ([]byte, error) {
+	currentState := make(chainState)
+	fileExists, err := s.read(chainID, &currentState)
+	if err != nil {
+		s.logger.Error(
+			"failed to read file",
+			zap.Error(err),
+		)
+		return nil, err
+	}
+	if !fileExists {
+		return nil, fileDoesNotExistErr
+	}
+
+	return []byte(currentState[string(key)]), nil
+}
+
+// Put the value into the json database. Read the current chain state and overwrite the key, if it exists
+func (s *JSONFileStorage) Put(chainID ids.ID, key []byte, value []byte) error {
+	currentState := make(chainState)
+	_, err := s.read(chainID, &currentState)
+	if err != nil {
+		s.logger.Error(
+			"failed to read file",
+			zap.Error(err),
+		)
+		return err
+	}
+
+	currentState[string(key)] = string(value)
+
+	return s.write(chainID, currentState)
+}
+
+// write value into the file
+func (s *JSONFileStorage) write(network ids.ID, v interface{}) error {
+	mutex, ok := s.mutexes[network]
+	if !ok {
+		return errors.New("network does not exist")
+	}
+
+	mutex.Lock()
+	defer mutex.Unlock()
+
+	fnlPath := filepath.Join(s.dir, network.String()+".json")
+	tmpPath := fnlPath + ".tmp"
+
+	b, err := json.MarshalIndent(v, "", "\t")
+	if err != nil {
+		return err
+	}
+
+	// write marshaled data to the temp file
+	// if write failed, the original file is not affected
+	// 0644 Only the owner can read and write.
+	// Everyone else can only read. No one can execute the file.
+	if err := os.WriteFile(tmpPath, b, 0644); err != nil {
+		return errors.Wrap(err, "failed to write file")
+	}
+
+	// move final file into place
+	if err := os.Rename(tmpPath, fnlPath); err != nil {
+		return errors.Wrap(err, "failed to rename file")
+	}
+
+	return nil
+}
+
+// Read from disk and unmarshal into v
+// Returns a bool indicating whether the file exists, and an error.
+// If an error is returned, the bool should be ignored.
+func (s *JSONFileStorage) read(network ids.ID, v interface{}) (bool, error) {
+	mutex, ok := s.mutexes[network]
+	if !ok {
+		return false, errors.New("network does not exist")
+	}
+
+	mutex.RLock()
+	defer mutex.RUnlock()
+
+	path := filepath.Join(s.dir, network.String()+".json")
+
+	// If the file does not exist, return false, but do not return an error as this
+	// is an expected case
+	if _, err := os.Stat(path); errors.Is(err, os.ErrNotExist) {
+		s.logger.Debug(
+			"file does not exist",
+			zap.String("path", path),
+			zap.Error(err),
+		)
+		return false, nil
+	}
+
+	b, err := os.ReadFile(path)
+	if err != nil {
+		return false, errors.Wrap(err, "failed to read file")
+	}
+
+	// unmarshal data
+	if err = json.Unmarshal(b, &v); err != nil {
+		return false, errors.Wrap(err, "failed to unmarshal json file")
+	}
+
+	// Return true to indicate that the file exists and we read from it successfully
+	return true, nil
+}