Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Validate peers on startup #231

Merged
merged 23 commits into from
Mar 19, 2024
Merged

Validate peers on startup #231

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
23 commits
Select commit Hold shift + click to select a range
7a4c847
src and dst cfgs store ids
cam-schultz Mar 12, 2024
0fbcb79
canonical vdrs in own package
cam-schultz Mar 12, 2024
0c0cdfd
connect peers on startup
cam-schultz Mar 12, 2024
55b6b24
ConnectToCanonicalValidators helper
cam-schultz Mar 13, 2024
c02cf8b
add helper type
cam-schultz Mar 13, 2024
25e2472
fix linter
cam-schultz Mar 13, 2024
6bb878f
connect to signing subnet
cam-schultz Mar 13, 2024
b4e681c
use struct var
cam-schultz Mar 13, 2024
5db104b
primary network handling
cam-schultz Mar 13, 2024
67e2410
clean up peer utils
cam-schultz Mar 13, 2024
5e0b9f1
Merge branch 'main' into validate-peers
cam-schultz Mar 13, 2024
253e247
do not connect to beacon nodes
cam-schultz Mar 13, 2024
3976dbc
remove unused var
cam-schultz Mar 13, 2024
f4bc236
pass source chain cfg to db ctr
cam-schultz Mar 14, 2024
5ee36e3
update comment
cam-schultz Mar 14, 2024
f3d64ad
pr feedback
cam-schultz Mar 14, 2024
0cb6139
add vdr client to network struct
cam-schultz Mar 14, 2024
9311703
refactor connection loop
cam-schultz Mar 15, 2024
a2c0e86
remove comment
cam-schultz Mar 15, 2024
46e8fb8
cache chain->subnet mapping
cam-schultz Mar 15, 2024
9fa3f75
supported dsts already parsed
cam-schultz Mar 15, 2024
f504c2f
add helpers
cam-schultz Mar 15, 2024
d48e359
Merge branch 'main' into validate-peers
cam-schultz Mar 15, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
85 changes: 58 additions & 27 deletions config/config.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,8 +68,10 @@ type SourceBlockchain struct {
SupportedDestinations []string `mapstructure:"supported-destinations" json:"supported-destinations"`
ProcessHistoricalBlocksFromHeight uint64 `mapstructure:"process-historical-blocks-from-height" json:"process-historical-blocks-from-height"`

// convenience field to access the supported destinations after initialization
// convenience fields to access parsed data after initialization
supportedDestinations set.Set[ids.ID]
subnetID ids.ID
blockchainID ids.ID
}

type DestinationBlockchain struct {
Expand All @@ -81,6 +83,10 @@ type DestinationBlockchain struct {

// Fetched from the chain after startup
warpQuorum WarpQuorum

// convenience fields to access parsed data after initialization
subnetID ids.ID
blockchainID ids.ID
}

type WarpQuorum struct {
Expand All @@ -98,9 +104,8 @@ type Config struct {
ProcessMissedBlocks bool `mapstructure:"process-missed-blocks" json:"process-missed-blocks"`
ManualWarpMessages []*ManualWarpMessage `mapstructure:"manual-warp-messages" json:"manual-warp-messages"`

// convenience fields to access the source subnet and chain IDs after initialization
sourceSubnetIDs []ids.ID
sourceBlockchainIDs []ids.ID
Comment on lines -101 to -103
Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

These were only used to pass to the app request network constructor. With this change, it makes more sense to pass the configuration and iterate over the SourceBlockchain and DestinationBlockchain slices directly, so the parsed IDs were moved there.

// convenience field to fetch a blockchain's subnet ID
blockchainIDToSubnetID map[ids.ID]ids.ID
}

func SetDefaultConfigValues(v *viper.Viper) {
Expand Down Expand Up @@ -194,6 +199,8 @@ func (c *Config) Validate() error {
return err
}

blockchainIDToSubnetID := make(map[ids.ID]ids.ID)

// Validate the destination chains
destinationChains := set.NewSet[string](len(c.DestinationBlockchains))
for _, s := range c.DestinationBlockchains {
Expand All @@ -204,12 +211,11 @@ func (c *Config) Validate() error {
return errors.New("configured destination subnets must have unique chain IDs")
}
destinationChains.Add(s.BlockchainID)
blockchainIDToSubnetID[s.blockchainID] = s.subnetID
}

// Validate the source chains and store the source subnet and chain IDs for future use
sourceBlockchains := set.NewSet[string](len(c.SourceBlockchains))
var sourceSubnetIDs []ids.ID
var sourceBlockchainIDs []ids.ID
for _, s := range c.SourceBlockchains {
// Validate configuration
if err := s.Validate(&destinationChains); err != nil {
Expand All @@ -220,23 +226,9 @@ func (c *Config) Validate() error {
return errors.New("configured source subnets must have unique chain IDs")
}
sourceBlockchains.Add(s.BlockchainID)

// Save IDs for future use
subnetID, err := ids.FromString(s.SubnetID)
if err != nil {
return fmt.Errorf("invalid subnetID in configuration. error: %w", err)
}
sourceSubnetIDs = append(sourceSubnetIDs, subnetID)

blockchainID, err := ids.FromString(s.BlockchainID)
if err != nil {
return fmt.Errorf("invalid subnetID in configuration. error: %w", err)
}
sourceBlockchainIDs = append(sourceBlockchainIDs, blockchainID)
blockchainIDToSubnetID[s.blockchainID] = s.subnetID
}

c.sourceSubnetIDs = sourceSubnetIDs
c.sourceBlockchainIDs = sourceBlockchainIDs
c.blockchainIDToSubnetID = blockchainIDToSubnetID

// Validate the manual warp messages
for i, msg := range c.ManualWarpMessages {
Expand All @@ -248,6 +240,10 @@ func (c *Config) Validate() error {
return nil
}

func (c *Config) GetSubnetID(blockchainID ids.ID) ids.ID {
return c.blockchainIDToSubnetID[blockchainID]
}

func (m *ManualWarpMessage) GetUnsignedMessageBytes() []byte {
return m.unsignedMessageBytes
}
Expand Down Expand Up @@ -417,6 +413,18 @@ func (s *SourceBlockchain) Validate(destinationBlockchainIDs *set.Set[string]) e
}
}

// Validate and store the subnet and blockchain IDs for future use
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not for this PR, but we should consider splitting config.go up into multiple files at some point soon. I think SourceBlockchain and DestinationBlockchain could each be put in their own file still in the config package and could help improve navigation/readability.

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

agreed and some of the similarities between the source/destination can be abstracted

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

some of the similarities between the source/destination can be abstracted

Viper doesn't play nicely with this unfortunately. There may be a workaround, but I haven't looked to deeply into it.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It doesn't look too bad, but we can definitely defer to a separate ticket https://stackoverflow.com/questions/47185318/multiple-config-files-with-go-viper

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

To clarify, I think we should keep a single user-facing configuration file, but split it into multiple Go files for easier code readability.

I experimented with Matt's suggestion of composing SourceBlockchain and DestinationBlockchain using a common type, but Viper seemed to have trouble unmarshalling JSON into them. Definitely worth revisiting if we decide to go for those config improvements.

blockchainID, err := ids.FromString(s.BlockchainID)
if err != nil {
return fmt.Errorf("invalid blockchainID in configuration. error: %w", err)
}
s.blockchainID = blockchainID
subnetID, err := ids.FromString(s.SubnetID)
if err != nil {
return fmt.Errorf("invalid subnetID in configuration. error: %w", err)
}
s.subnetID = subnetID

// Validate and store the allowed destinations for future use
s.supportedDestinations = set.Set[ids.ID]{}

Expand Down Expand Up @@ -447,6 +455,14 @@ func (s *SourceBlockchain) Validate(destinationBlockchainIDs *set.Set[string]) e
return nil
}

func (s *SourceBlockchain) GetSubnetID() ids.ID {
return s.subnetID
}

func (s *SourceBlockchain) GetBlockchainID() ids.ID {
return s.blockchainID
}

// Validatees the destination subnet configuration
func (s *DestinationBlockchain) Validate() error {
if _, err := ids.FromString(s.SubnetID); err != nil {
Expand All @@ -473,9 +489,29 @@ func (s *DestinationBlockchain) Validate() error {
return fmt.Errorf("unsupported VM type for source subnet: %s", s.VM)
}

// Validate and store the subnet and blockchain IDs for future use
blockchainID, err := ids.FromString(s.BlockchainID)
if err != nil {
return fmt.Errorf("invalid blockchainID in configuration. error: %w", err)
}
s.blockchainID = blockchainID
subnetID, err := ids.FromString(s.SubnetID)
if err != nil {
return fmt.Errorf("invalid subnetID in configuration. error: %w", err)
}
s.subnetID = subnetID

return nil
}

func (s *DestinationBlockchain) GetSubnetID() ids.ID {
return s.subnetID
}

func (s *DestinationBlockchain) GetBlockchainID() ids.ID {
return s.blockchainID
}

func (s *DestinationBlockchain) initializeWarpQuorum() error {
blockchainID, err := ids.FromString(s.BlockchainID)
if err != nil {
Expand Down Expand Up @@ -514,11 +550,6 @@ func (s *DestinationBlockchain) GetRelayerAccountInfo() (*ecdsa.PrivateKey, comm
// Top-level config getters
//

// GetSourceIDs returns the Subnet and Chain IDs of all subnets configured as a source
func (c *Config) GetSourceIDs() ([]ids.ID, []ids.ID) {
return c.sourceSubnetIDs, c.sourceBlockchainIDs
}

func (c *Config) GetWarpQuorum(blockchainID ids.ID) (WarpQuorum, error) {
for _, s := range c.DestinationBlockchains {
if blockchainID.String() == s.BlockchainID {
Expand Down
17 changes: 10 additions & 7 deletions database/json_file_storage.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@ import (

"github.com/ava-labs/avalanchego/ids"
"github.com/ava-labs/avalanchego/utils/logging"
"github.com/ava-labs/awm-relayer/config"
"github.com/pkg/errors"
"go.uber.org/zap"
)
Expand All @@ -34,30 +35,32 @@ type JSONFileStorage struct {
}

// NewJSONFileStorage creates a new JSONFileStorage instance
func NewJSONFileStorage(logger logging.Logger, dir string, networks []ids.ID) (*JSONFileStorage, error) {
func NewJSONFileStorage(logger logging.Logger, dir string, sourceBlockchains []*config.SourceBlockchain) (*JSONFileStorage, error) {
storage := &JSONFileStorage{
dir: filepath.Clean(dir),
mutexes: make(map[ids.ID]*sync.RWMutex),
logger: logger,
currentState: make(map[ids.ID]chainState),
}

for _, network := range networks {
storage.currentState[network] = make(chainState)
storage.mutexes[network] = &sync.RWMutex{}
for _, sourceBlockchain := range sourceBlockchains {
sourceBlockchainID := sourceBlockchain.GetBlockchainID()
storage.currentState[sourceBlockchainID] = make(chainState)
storage.mutexes[sourceBlockchainID] = &sync.RWMutex{}
}

_, err := os.Stat(dir)
if err == nil {
// Directory already exists.
// Read the existing storage.
for _, network := range networks {
currentState, fileExists, err := storage.getCurrentState(network)
for _, sourceBlockchain := range sourceBlockchains {
sourceBlockchainID := sourceBlockchain.GetBlockchainID()
currentState, fileExists, err := storage.getCurrentState(sourceBlockchainID)
if err != nil {
return nil, err
}
if fileExists {
storage.currentState[network] = currentState
storage.currentState[sourceBlockchainID] = currentState
}
}
return storage, nil
Expand Down
71 changes: 51 additions & 20 deletions database/json_file_storage_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,15 +13,44 @@ import (

"github.com/ava-labs/avalanchego/ids"
"github.com/ava-labs/avalanchego/utils/logging"
"github.com/ava-labs/avalanchego/utils/set"
"github.com/ava-labs/awm-relayer/config"
"github.com/stretchr/testify/assert"
)

var validSourceBlockchainConfig = &config.SourceBlockchain{
RPCEndpoint: "http://test.avax.network/ext/bc/C/rpc",
WSEndpoint: "ws://test.avax.network/ext/bc/C/ws",
BlockchainID: "S4mMqUXe7vHsGiRAma6bv3CKnyaLssyAxmQ2KvFpX1KEvfFCD",
SubnetID: "2TGBXcnwx5PqiXWiqxAKUaNSqDguXNh1mxnp82jui68hxJSZAx",
VM: "evm",
MessageContracts: map[string]config.MessageProtocolConfig{
"0xd81545385803bCD83bd59f58Ba2d2c0562387F83": {
MessageFormat: config.TELEPORTER.String(),
},
},
}

func populateSourceConfig(blockchainIDs []ids.ID) []*config.SourceBlockchain {
sourceBlockchains := make([]*config.SourceBlockchain, len(blockchainIDs))
for i, id := range blockchainIDs {
sourceBlockchains[i] = validSourceBlockchainConfig
sourceBlockchains[i].BlockchainID = id.String()
}
destinationsBlockchainIDs := set.NewSet[string](1) // just needs to be non-nil
destinationsBlockchainIDs.Add(ids.GenerateTestID().String())
sourceBlockchains[0].Validate(&destinationsBlockchainIDs)
return sourceBlockchains
}

// Test that the JSON database can write and read to a single chain concurrently.
func TestConcurrentWriteReadSingleChain(t *testing.T) {
networks := []ids.ID{
ids.GenerateTestID(),
}
jsonStorage := setupJsonStorage(t, networks)
sourceBlockchains := populateSourceConfig(
[]ids.ID{
ids.GenerateTestID(),
},
)
jsonStorage := setupJsonStorage(t, sourceBlockchains)

// Test writing to the JSON database concurrently.
wg := sync.WaitGroup{}
Expand All @@ -30,16 +59,16 @@ func TestConcurrentWriteReadSingleChain(t *testing.T) {
idx := i
go func() {
defer wg.Done()
testWrite(jsonStorage, networks[0], uint64(idx))
testWrite(jsonStorage, sourceBlockchains[0].GetBlockchainID(), uint64(idx))
}()
}
wg.Wait()

// Write one final time to ensure that concurrent writes don't cause any issues.
finalTargetValue := uint64(11)
testWrite(jsonStorage, networks[0], finalTargetValue)
testWrite(jsonStorage, sourceBlockchains[0].GetBlockchainID(), finalTargetValue)

latestProcessedBlockData, err := jsonStorage.Get(networks[0], []byte(LatestProcessedBlockKey))
latestProcessedBlockData, err := jsonStorage.Get(sourceBlockchains[0].GetBlockchainID(), []byte(LatestProcessedBlockKey))
if err != nil {
t.Fatalf("failed to retrieve from JSON storage. err: %v", err)
}
Expand All @@ -52,12 +81,14 @@ func TestConcurrentWriteReadSingleChain(t *testing.T) {

// Test that the JSON database can write and read from multiple chains concurrently. Write to any given chain are not concurrent.
func TestConcurrentWriteReadMultipleChains(t *testing.T) {
networks := []ids.ID{
ids.GenerateTestID(),
ids.GenerateTestID(),
ids.GenerateTestID(),
}
jsonStorage := setupJsonStorage(t, networks)
sourceBlockchains := populateSourceConfig(
[]ids.ID{
ids.GenerateTestID(),
ids.GenerateTestID(),
ids.GenerateTestID(),
},
)
jsonStorage := setupJsonStorage(t, sourceBlockchains)

// Test writing to the JSON database concurrently.
wg := sync.WaitGroup{}
Expand All @@ -66,19 +97,19 @@ func TestConcurrentWriteReadMultipleChains(t *testing.T) {
index := i
go func() {
defer wg.Done()
testWrite(jsonStorage, networks[index], uint64(index))
testWrite(jsonStorage, sourceBlockchains[index].GetBlockchainID(), uint64(index))
}()
}
wg.Wait()

// Write one final time to ensure that concurrent writes don't cause any issues.
finalTargetValue := uint64(3)
for _, network := range networks {
testWrite(jsonStorage, network, finalTargetValue)
for _, sourceBlockchain := range sourceBlockchains {
testWrite(jsonStorage, sourceBlockchain.GetBlockchainID(), finalTargetValue)
}

for i, id := range networks {
latestProcessedBlockData, err := jsonStorage.Get(id, []byte(LatestProcessedBlockKey))
for i, sourceBlockchain := range sourceBlockchains {
latestProcessedBlockData, err := jsonStorage.Get(sourceBlockchain.GetBlockchainID(), []byte(LatestProcessedBlockKey))
if err != nil {
t.Fatalf("failed to retrieve from JSON storage. networkID: %d err: %v", i, err)
}
Expand All @@ -90,7 +121,7 @@ func TestConcurrentWriteReadMultipleChains(t *testing.T) {
}
}

func setupJsonStorage(t *testing.T, networks []ids.ID) *JSONFileStorage {
func setupJsonStorage(t *testing.T, sourceBlockchains []*config.SourceBlockchain) *JSONFileStorage {
logger := logging.NewLogger(
"awm-relayer-test",
logging.NewWrappedCore(
Expand All @@ -101,7 +132,7 @@ func setupJsonStorage(t *testing.T, networks []ids.ID) *JSONFileStorage {
)
storageDir := t.TempDir()

jsonStorage, err := NewJSONFileStorage(logger, storageDir, networks)
jsonStorage, err := NewJSONFileStorage(logger, storageDir, sourceBlockchains)
if err != nil {
t.Fatal(err)
}
Expand Down
Loading
Loading