Merge pull request #297 from koinos/294-protocol-version

Protocol version

go.mod

@@ -22,6 +22,7 @@ require (
 
 require (
 	filippo.io/bigmod v0.0.1 // indirect
+	github.com/Masterminds/semver/v3 v3.2.1 // indirect
 	github.com/benbjohnson/clock v1.3.5 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/btcsuite/btcd v0.20.1-beta // indirect

go.sum

@@ -545,6 +545,8 @@ github.com/BurntSushi/toml v1.2.1/go.mod h1:CxXYINrC8qIiEnFrOxCa7Jy5BFHlXnUU2pbi
 github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802/go.mod h1:IVnqGOEym/WlBOVXweHU+Q+/VP0lqqI8lqeDx9IjBqo=
 github.com/DATA-DOG/go-sqlmock v1.3.3/go.mod h1:f/Ixk793poVmq4qj/V1dPUg2JEAKC73Q5eFN3EC/SaM=
 github.com/JohnCGriffin/overflow v0.0.0-20211019200055-46fa312c352c/go.mod h1:X0CRv0ky0k6m906ixxpzmDRLvX58TFUKS2eePweuyxk=
+github.com/Masterminds/semver/v3 v3.2.1 h1:RN9w6+7QoMeJVGyfmbcgs28Br8cvmnucEXnY0rYXWg0=
+github.com/Masterminds/semver/v3 v3.2.1/go.mod h1:qvl/7zhW3nngYb5+80sSMF+FG2BjYrf8m9wsX0PNOMQ=
 github.com/OneOfOne/xxhash v1.2.2/go.mod h1:HSdplMjZKSmBqAxg5vPj2TmRDmfkzw+cTzAElWljhcU=
 github.com/StackExchange/wmi v0.0.0-20180116203802-5d049714c4a6/go.mod h1:3eOhrUMpNV+6aFIbp5/iudMxNCF27Vw2OZgy4xEx0Fg=
 github.com/VictoriaMetrics/fastcache v1.6.0/go.mod h1:0qHz5QP0GMX4pfmMA/zt5RgfNuXJrTP0zS7DqpHGGTw=

internal/node/node.go

@@ -113,6 +113,7 @@ func NewKoinosP2PNode(ctx context.Context, listenAddr string, localRPC rpc.Local
 		// performance issues.
 		libp2p.EnableNATService(),
 		libp2p.ConnectionGater(node.PeerErrorHandler),
+		libp2p.ProtocolVersion(p2p.KoinosProtocolVersionString()),
 	}
 
 	host, err := libp2p.New(options...)
@@ -191,6 +192,7 @@ func NewKoinosP2PNode(ctx context.Context, listenAddr string, localRPC rpc.Local
 	node.ConnectionManager = p2p.NewConnectionManager(
 		node.Host,
 		node.localRPC,
+		&config.ConnectionManagerOptions,
 		&config.PeerConnectionOptions,
 		node,
 		node.Options.InitialPeers,

internal/options/config.go

@@ -2,21 +2,23 @@ package options
 
 // Config is the entire configuration file
 type Config struct {
-	NodeOptions             NodeOptions
-	PeerConnectionOptions   PeerConnectionOptions
-	PeerErrorHandlerOptions PeerErrorHandlerOptions
-	GossipToggleOptions     GossipToggleOptions
-	ApplicatorOptions       ApplicatorOptions
+	NodeOptions              NodeOptions
+	PeerConnectionOptions    PeerConnectionOptions
+	PeerErrorHandlerOptions  PeerErrorHandlerOptions
+	GossipToggleOptions      GossipToggleOptions
+	ApplicatorOptions        ApplicatorOptions
+	ConnectionManagerOptions ConnectionManagerOptions
 }
 
 // NewConfig creates a new Config
 func NewConfig() *Config {
 	config := Config{
-		NodeOptions:             *NewNodeOptions(),
-		PeerConnectionOptions:   *NewPeerConnectionOptions(),
-		PeerErrorHandlerOptions: *NewPeerErrorHandlerOptions(),
-		GossipToggleOptions:     *NewGossipToggleOptions(),
-		ApplicatorOptions:       *NewApplicatorOptions(),
+		NodeOptions:              *NewNodeOptions(),
+		PeerConnectionOptions:    *NewPeerConnectionOptions(),
+		PeerErrorHandlerOptions:  *NewPeerErrorHandlerOptions(),
+		GossipToggleOptions:      *NewGossipToggleOptions(),
+		ApplicatorOptions:        *NewApplicatorOptions(),
+		ConnectionManagerOptions: *NewConnectionManagerOptions(),
 	}
 	return &config
 }

internal/options/connection_manager_options.go

@@ -0,0 +1,22 @@
+package options
+
+import "time"
+
+const (
+	protocolVersionRetryTimeDefault = time.Millisecond * 50
+	protocolVersionTimeoutDefault   = time.Second * 5
+)
+
+// ConnectionManagerOptions are options for ConnectionManager
+type ConnectionManagerOptions struct {
+	ProtocolVersionRetryTime time.Duration
+	ProtocolVersionTimeout   time.Duration
+}
+
+// NewConnectionManagerOptions returns default initialized ConnectionManagerOptions
+func NewConnectionManagerOptions() *ConnectionManagerOptions {
+	return &ConnectionManagerOptions{
+		ProtocolVersionRetryTime: protocolVersionRetryTimeDefault,
+		ProtocolVersionTimeout:   protocolVersionTimeoutDefault,
+	}
+}

internal/options/error_handler_options.go

@@ -28,6 +28,7 @@ const (
 	processRequestTimeoutErrorScoreDefault   = 0
 	forkBombErrorScoreDefault                = errorScoreThresholdDefault * 2
 	maxHeightErrorScoreDefault               = blockApplicationErrorScoreDefault
+	protocolMistmatchErrorScoreDefault       = errorScoreThresholdDefault * 2
 	unknownErrorScoreDefault                 = blockApplicationErrorScoreDefault
 )
 
@@ -56,6 +57,7 @@ type PeerErrorHandlerOptions struct {
 	ProcessRequestTimeoutErrorScore   uint64
 	ForkBombErrorScore                uint64
 	MaxHeightErrorScore               uint64
+	ProtocolMismatchErrorScore        uint64
 	UnknownErrorScore                 uint64
 }
 
@@ -84,6 +86,7 @@ func NewPeerErrorHandlerOptions() *PeerErrorHandlerOptions {
 		ProcessRequestTimeoutErrorScore:   processRequestTimeoutErrorScoreDefault,
 		ForkBombErrorScore:                forkBombErrorScoreDefault,
 		MaxHeightErrorScore:               maxHeightErrorScoreDefault,
+		ProtocolMismatchErrorScore:        protocolMistmatchErrorScoreDefault,
 		UnknownErrorScore:                 unknownErrorScoreDefault,
 	}
 }

internal/p2p/connection_manager.go

@@ -2,11 +2,15 @@ package p2p
 
 import (
 	"context"
+	"errors"
 	"fmt"
+	"strings"
 	"time"
 
+	"github.com/Masterminds/semver/v3"
 	log "github.com/koinos/koinos-log-golang/v2"
 	"github.com/koinos/koinos-p2p/internal/options"
+	"github.com/koinos/koinos-p2p/internal/p2perrors"
 	"github.com/koinos/koinos-p2p/internal/rpc"
 
 	gorpc "github.com/libp2p/go-libp2p-gorpc"
@@ -49,6 +53,7 @@ type ConnectionManager struct {
 	client *gorpc.Client
 
 	localRPC    rpc.LocalRPC
+	opts        *options.ConnectionManagerOptions
 	peerOpts    *options.PeerConnectionOptions
 	libProvider LastIrreversibleBlockProvider
 	applicator  *Applicator
@@ -68,6 +73,7 @@ type ConnectionManager struct {
 func NewConnectionManager(
 	host host.Host,
 	localRPC rpc.LocalRPC,
+	managerOpts *options.ConnectionManagerOptions,
 	peerOpts *options.PeerConnectionOptions,
 	libProvider LastIrreversibleBlockProvider,
 	initialPeers []peer.AddrInfo,
@@ -79,6 +85,7 @@ func NewConnectionManager(
 		client:               gorpc.NewClient(host, rpc.PeerRPCID),
 		server:               gorpc.NewServer(host, rpc.PeerRPCID),
 		localRPC:             localRPC,
+		opts:                 managerOpts,
 		peerOpts:             peerOpts,
 		libProvider:          libProvider,
 		applicator:           applicator,
@@ -172,6 +179,52 @@ func (c *ConnectionManager) IsConnected(ctx context.Context, pid peer.ID) bool {
 	}
 }
 
+func (c *ConnectionManager) readProtocolVersion(pid peer.ID) (string, error) {
+	peerVersion, err := c.host.Peerstore().Get(pid, "ProtocolVersion")
+	if err != nil {
+		return "", err
+	}
+
+	switch peerVersion := peerVersion.(type) {
+	case string:
+		return peerVersion, nil
+	default:
+		return "", p2perrors.ErrProtocolMismatch
+	}
+}
+
+func (c *ConnectionManager) GetProtocolVersion(ctx context.Context, pid peer.ID) (*semver.Version, error) {
+	versionCtx, cancel := context.WithTimeout(ctx, c.opts.ProtocolVersionTimeout)
+	defer cancel()
+
+	for {
+		versionString, err := c.readProtocolVersion(pid)
+		if err != nil {
+			if errors.Is(err, p2perrors.ErrProtocolMismatch) {
+				return nil, err
+			}
+		} else if len(versionString) > 0 {
+			if !strings.HasPrefix(versionString, koinosProtocolPrefix) {
+				return nil, p2perrors.ErrProtocolMismatch
+			}
+
+			parts := strings.Split(versionString, "/")
+			version, err := semver.NewVersion(parts[len(parts)-1])
+			if err != nil {
+				return nil, p2perrors.ErrProtocolMismatch
+			}
+
+			return version, nil
+		}
+
+		select {
+		case <-time.After(c.opts.ProtocolVersionRetryTime):
+		case <-versionCtx.Done():
+			return nil, p2perrors.ErrProtocolMissing
+		}
+	}
+}
+
 func (c *ConnectionManager) handleConnected(ctx context.Context, msg connectionMessage) {
 	pid := msg.conn.RemotePeer()
 	s := fmt.Sprintf("%s/p2p/%s", msg.conn.RemoteMultiaddr(), pid)
@@ -189,6 +242,7 @@ func (c *ConnectionManager) handleConnected(ctx context.Context, msg connectionM
 				c.peerErrorChan,
 				c.peerOpts,
 				c.applicator,
+				c,
 			),
 			conn:   msg.conn,
 			cancel: cancel,

internal/p2p/error_handler.go

@@ -130,7 +130,7 @@ func (p *PeerErrorHandler) handleError(ctx context.Context, peerErr PeerError) {
 			}
 		}
 
-		if !errors.Is(peerErr.err, p2perrors.ErrChainIDMismatch) {
+		if !errors.Is(peerErr.err, p2perrors.ErrChainIDMismatch) && !errors.Is(peerErr.err, p2perrors.ErrProtocolMismatch) {
 			log.Infof("Encountered peer error: %s, %s. Current error score: %v", peerErr.id, peerErr.err.Error(), p.errorScores[ipAddr].score)
 		}
 
@@ -182,6 +182,8 @@ func (p *PeerErrorHandler) getScoreForError(err error) uint64 {
 		return p.opts.ChainNotConnectedErrorScore
 	case errors.Is(err, p2perrors.ErrCheckpointMismatch):
 		return p.opts.CheckpointMismatchErrorScore
+	case errors.Is(err, p2perrors.ErrProtocolMismatch):
+		return p.opts.ProtocolMismatchErrorScore
 
 	// Errors that should only originate from the local process or local node
 	case errors.Is(err, p2perrors.ErrLocalRPC):

internal/p2p/peer_connection.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/Masterminds/semver/v3"
 	log "github.com/koinos/koinos-log-golang/v2"
 	"github.com/koinos/koinos-p2p/internal/options"
 	"github.com/koinos/koinos-p2p/internal/p2perrors"
@@ -20,23 +21,38 @@ type signalRequestBlocks struct{}
 // PeerConnection handles the sync portion of a connection to a peer
 type PeerConnection struct {
 	id       peer.ID
+	version  *semver.Version
 	isSynced bool
 	opts     *options.PeerConnectionOptions
 
 	requestBlockChan chan signalRequestBlocks
 
-	libProvider   LastIrreversibleBlockProvider
-	localRPC      rpc.LocalRPC
-	peerRPC       rpc.RemoteRPC
-	applicator    *Applicator
-	peerErrorChan chan<- PeerError
+	libProvider     LastIrreversibleBlockProvider
+	localRPC        rpc.LocalRPC
+	peerRPC         rpc.RemoteRPC
+	applicator      *Applicator
+	peerErrorChan   chan<- PeerError
+	versionProvider ProtocolVersionProvider
 }
 
 func (p *PeerConnection) requestBlocks() {
 	p.requestBlockChan <- signalRequestBlocks{}
 }
 
 func (p *PeerConnection) handshake(ctx context.Context) error {
+	// Check Peer's protocol version
+	version, err := p.versionProvider.GetProtocolVersion(ctx, p.id)
+	if err == nil {
+		p.version = version
+	} else {
+		// TODO: Remove to reject when protocol is missing
+		if errors.Is(err, p2perrors.ErrProtocolMissing) {
+			p.version = semver.New(0, 0, 0, "", "")
+		} else {
+			return err
+		}
+	}
+
 	// Get my chain id
 	rpcContext, cancelLocalGetChainID := context.WithTimeout(ctx, p.opts.LocalRPCTimeout)
 	defer cancelLocalGetChainID()
@@ -244,7 +260,8 @@ func NewPeerConnection(
 	peerRPC rpc.RemoteRPC,
 	peerErrorChan chan<- PeerError,
 	opts *options.PeerConnectionOptions,
-	applicator *Applicator) *PeerConnection {
+	applicator *Applicator,
+	versionProvider ProtocolVersionProvider) *PeerConnection {
 	return &PeerConnection{
 		id:               id,
 		isSynced:         false,
@@ -255,5 +272,6 @@ func NewPeerConnection(
 		peerRPC:          peerRPC,
 		applicator:       applicator,
 		peerErrorChan:    peerErrorChan,
+		versionProvider:  versionProvider,
 	}
 }

internal/p2p/protocol_version.go

@@ -0,0 +1,28 @@
+package p2p
+
+import "github.com/Masterminds/semver/v3"
+
+const (
+	koinosProtocolPrefix = "koinos/p2p/"
+	versionMajor         = 1
+	versionMinor         = 0
+	versionPatch         = 0
+	versionPrerelease    = ""
+	versionMetadata      = ""
+)
+
+var (
+	koinosProtocolVersion *semver.Version
+)
+
+func KoinosProtocolVersion() *semver.Version {
+	if koinosProtocolVersion == nil {
+		koinosProtocolVersion = semver.New(versionMajor, versionMinor, versionPatch, versionPrerelease, versionMetadata)
+	}
+
+	return koinosProtocolVersion
+}
+
+func KoinosProtocolVersionString() string {
+	return koinosProtocolPrefix + KoinosProtocolVersion().String()
+}

internal/p2p/protocol_version_provider.go

@@ -0,0 +1,13 @@
+package p2p
+
+import (
+	"context"
+
+	"github.com/Masterminds/semver/v3"
+	"github.com/libp2p/go-libp2p/core/peer"
+)
+
+// ProtocolVersionProvider is an interface for a peer's protocol version to the PeerConnection
+type ProtocolVersionProvider interface {
+	GetProtocolVersion(ctx context.Context, pid peer.ID) (*semver.Version, error)
+}

internal/p2perrors/errors.go

@@ -64,4 +64,10 @@ var (
 
 	// ErrBlockState represents when chain cannot create a new block state node
 	ErrBlockState = errors.New("could not create new block state node")
+
+	// ErrProtocolMismatch represents when a peer's protocol version does match ours
+	ErrProtocolMismatch = errors.New("protocol version mismatch")
+
+	// ErrProtocolMissing represents when a peer's protocol version is missing
+	ErrProtocolMissing = errors.New("protocol version is missing")
 )