Merge branch 'master' into consensus-updates

main/params.go

@@ -26,6 +26,10 @@ import (
 	"github.com/ava-labs/gecko/utils/wrappers"
 )
 
+const (
+	dbVersion = "v0.1.0"
+)
+
 // Results of parsing the CLI
 var (
 	Config = node.Config{}
@@ -143,7 +147,7 @@ func init() {
 	// DB:
 	if *db && err == nil {
 		// TODO: Add better params here
-		dbPath := path.Join(*dbDir, genesis.NetworkName(Config.NetworkID))
+		dbPath := path.Join(*dbDir, genesis.NetworkName(Config.NetworkID), dbVersion)
 		db, err := leveldb.New(dbPath, 0, 0, 0)
 		Config.DB = db
 		errs.Add(err)

networking/connections.go

@@ -194,8 +194,11 @@ func (c *connections) remove(peer salticidae.PeerID, id ids.ShortID) {
 func (c *connections) removePeerID(peer salticidae.PeerID) {
 	peerID := toID(peer)
 	if id, exists := c.peerIDToID[peerID]; exists {
+		idKey := id.Key()
+
 		delete(c.peerIDToID, peerID)
-		delete(c.idToPeerID, id.Key())
+		delete(c.idToPeerID, idKey)
+		delete(c.idToIP, idKey)
 	}
 }
 
@@ -204,6 +207,7 @@ func (c *connections) removeID(id ids.ShortID) {
 	if peer, exists := c.idToPeerID[idKey]; exists {
 		delete(c.peerIDToID, toID(peer))
 		delete(c.idToPeerID, idKey)
+		delete(c.idToIP, idKey)
 	}
 }
 

networking/handshake_handlers.go

@@ -19,6 +19,8 @@ import (
 	"errors"
 	"fmt"
 	"math"
+	"strconv"
+	"strings"
 	"sync"
 	"time"
 	"unsafe"
@@ -61,9 +63,23 @@ Attempt reconnections
     node isn't connected to after awhile delete the connection.
 */
 
+// Version this avalanche instance is executing.
+var (
+	VersionPrefix    = "avalanche/"
+	VersionSeparator = "."
+	MajorVersion     = 0
+	MinorVersion     = 1
+	PatchVersion     = 0
+	ClientVersion    = fmt.Sprintf("%s%d%s%d%s%d",
+		VersionPrefix,
+		MajorVersion,
+		VersionSeparator,
+		MinorVersion,
+		VersionSeparator,
+		PatchVersion)
+)
+
 const (
-	// CurrentVersion this avalanche instance is executing.
-	CurrentVersion = "avalanche/0.0.1"
 	// MaxClockDifference allowed between connected nodes.
 	MaxClockDifference = time.Minute
 	// PeerListGossipSpacing is the amount of time to wait between pushing this
@@ -74,6 +90,10 @@ const (
 	// PeerListStakerGossipFraction calculates the fraction of stakers that are
 	// gossiped to. If set to 1, then only stakers will be gossiped to.
 	PeerListStakerGossipFraction = 2
+
+	// ConnectTimeout is the amount of time to wait before attempt to connect to
+	// an unknown peer
+	ConnectTimeout = 6 * time.Second
 	// GetVersionTimeout is the amount of time to wait before sending a
 	// getVersion message to a partially connected peer
 	GetVersionTimeout = 2 * time.Second
@@ -96,27 +116,37 @@ var (
 type Handshake struct {
 	handshakeMetrics
 
-	networkID uint32
+	networkID uint32 // ID of the network I'm running, used to prevent connecting to the wrong network
 
 	log           logging.Logger
-	vdrs          validators.Set
-	myAddr        salticidae.NetAddr
-	myID          ids.ShortID
-	net           salticidae.PeerNetwork
-	enableStaking bool // Should only be false for local tests
+	vdrs          validators.Set         // set of current validators in the AVAnet
+	myAddr        salticidae.NetAddr     // IP I communicate to peers
+	myID          ids.ShortID            // ID that identifies myself as a staker or not
+	net           salticidae.PeerNetwork // C messaging network
+	enableStaking bool                   // Should only be false for local tests
 
-	clock       timer.Clock
-	pending     Connections // Connections that I haven't gotten version messages from
-	connections Connections // Connections that I think are connected
+	clock timer.Clock
 
-	versionTimeout   timer.TimeoutManager
-	reconnectTimeout timer.TimeoutManager
+	// Connections that I have added by IP, but haven't gotten an ID from
+	requestedLock    sync.Mutex
+	requested        map[string]struct{}
+	requestedTimeout timer.TimeoutManager // keys are hashes of the ip:port string
+
+	// Connections that I have added as a peer, but haven't gotten a version
+	// message from
+	pending        Connections
+	versionTimeout timer.TimeoutManager // keys are the peer IDs
+
+	// Connections that I have gotten a valid version message from
+	connections      Connections
+	reconnectTimeout timer.TimeoutManager // keys are the peer IDs
+
+	// IPs of nodes I'm connected to will be repeatedly gossiped throughout the network
 	peerListGossiper *timer.Repeater
 
+	// If any chain is blocked on connecting to peers, track these blockers here
 	awaitingLock sync.Mutex
 	awaiting     []*networking.AwaitingConnections
-
-	requestedConnections map[string]struct{}
 }
 
 // Initialize to the c networking library. This should only be done once during
@@ -132,19 +162,34 @@ func (nm *Handshake) Initialize(
 	networkID uint32,
 ) {
 	log.AssertTrue(nm.net == nil, "Should only register network handlers once")
+
+	nm.handshakeMetrics.Initialize(log, registerer)
+
+	nm.networkID = networkID
+
 	nm.log = log
 	nm.vdrs = vdrs
 	nm.myAddr = myAddr
 	nm.myID = myID
 	nm.net = peerNet
 	nm.enableStaking = enableStaking
-	nm.networkID = networkID
+
+	nm.requested = make(map[string]struct{})
+	nm.requestedTimeout.Initialize(ConnectTimeout)
+	go nm.log.RecoverAndPanic(nm.requestedTimeout.Dispatch)
 
 	nm.pending = NewConnections()
+	nm.versionTimeout.Initialize(GetVersionTimeout)
+	go nm.log.RecoverAndPanic(nm.versionTimeout.Dispatch)
+
 	nm.connections = NewConnections()
+	nm.reconnectTimeout.Initialize(ReconnectTimeout)
+	go nm.log.RecoverAndPanic(nm.reconnectTimeout.Dispatch)
 
-	nm.requestedConnections = make(map[string]struct{})
+	nm.peerListGossiper = timer.NewRepeater(nm.gossipPeerList, PeerListGossipSpacing)
+	go nm.log.RecoverAndPanic(nm.peerListGossiper.Dispatch)
 
+	// register c message callbacks
 	net := peerNet.AsMsgNetwork()
 
 	net.RegConnHandler(salticidae.MsgNetworkConnCallback(C.connHandler), nil)
@@ -156,34 +201,95 @@ func (nm *Handshake) Initialize(
 	net.RegHandler(Version, salticidae.MsgNetworkMsgCallback(C.version), nil)
 	net.RegHandler(GetPeerList, salticidae.MsgNetworkMsgCallback(C.getPeerList), nil)
 	net.RegHandler(PeerList, salticidae.MsgNetworkMsgCallback(C.peerList), nil)
+}
 
-	nm.handshakeMetrics.Initialize(nm.log, registerer)
+// ConnectTo add the peer as a connection and connects to them.
+func (nm *Handshake) ConnectTo(peer salticidae.PeerID, stakerID ids.ShortID, addr salticidae.NetAddr) {
+	if nm.pending.ContainsPeerID(peer) || nm.connections.ContainsPeerID(peer) {
+		return
+	}
 
-	nm.versionTimeout.Initialize(GetVersionTimeout)
-	go nm.log.RecoverAndPanic(nm.versionTimeout.Dispatch)
+	nm.log.Info("Attempting to connect to %s", stakerID)
 
-	nm.reconnectTimeout.Initialize(ReconnectTimeout)
-	go nm.log.RecoverAndPanic(nm.reconnectTimeout.Dispatch)
+	nm.net.AddPeer(peer)
+	nm.net.SetPeerAddr(peer, addr)
+	nm.net.ConnPeer(peer, 600, 1)
 
-	nm.peerListGossiper = timer.NewRepeater(nm.gossipPeerList, PeerListGossipSpacing)
-	go nm.log.RecoverAndPanic(nm.peerListGossiper.Dispatch)
+	ip := toIPDesc(addr)
+	nm.pending.Add(peer, stakerID, ip)
+
+	peerBytes := toID(peer)
+	peerID := ids.NewID(peerBytes)
+
+	nm.reconnectTimeout.Put(peerID, func() {
+		nm.pending.Remove(peer, stakerID)
+		nm.connections.Remove(peer, stakerID)
+		nm.net.DelPeer(peer)
+
+		nm.numPeers.Set(float64(nm.connections.Len()))
+	})
 }
 
 // Connect ...
 func (nm *Handshake) Connect(addr salticidae.NetAddr) {
+	ip := toIPDesc(addr)
+	ipStr := ip.String()
+	if nm.pending.ContainsIP(ip) || nm.connections.ContainsIP(ip) {
+		return
+	}
+
 	if !nm.enableStaking {
-		peer := salticidae.NewPeerIDFromNetAddr(addr, false)
-		nm.net.AddPeer(peer)
-		nm.net.SetPeerAddr(peer, addr)
-		nm.net.ConnPeer(peer, 600, 1)
-		peer.Free()
-	} else {
-		ip := toIPDesc(addr)
-		nm.requestedConnections[ip.String()] = struct{}{}
+		nm.log.Info("Adding peer %s", ip)
+
+		peer := salticidae.NewPeerIDFromNetAddr(addr, true)
+		nm.ConnectTo(peer, toShortID(ip), addr)
+		return
+	}
+
+	nm.requestedLock.Lock()
+	_, exists := nm.requested[ipStr]
+	nm.requestedLock.Unlock()
+
+	if exists {
+		return
+	}
+
+	nm.log.Info("Adding peer %s", ip)
+
+	count := new(int)
+	*count = 100
+	handler := new(func())
+	*handler = func() {
+		nm.requestedLock.Lock()
+		defer nm.requestedLock.Unlock()
+
+		if *count == 100 {
+			nm.requested[ipStr] = struct{}{}
+		}
+
+		if _, exists := nm.requested[ipStr]; !exists {
+			return
+		}
+
+		if *count <= 0 {
+			delete(nm.requested, ipStr)
+			return
+		}
+		*count--
+
+		if nm.pending.ContainsIP(ip) || nm.connections.ContainsIP(ip) {
+			return
+		}
+
+		nm.log.Info("Attempting to discover peer at %s", ipStr)
 
 		msgNet := nm.net.AsMsgNetwork()
 		msgNet.Connect(addr)
+
+		ipID := ids.NewID(hashing.ComputeHash256Array([]byte(ipStr)))
+		nm.requestedTimeout.Put(ipID, *handler)
 	}
+	(*handler)()
 }
 
 // AwaitConnections ...
@@ -266,7 +372,7 @@ func (nm *Handshake) SendGetVersion(peer salticidae.PeerID) {
 // SendVersion to the requested peer
 func (nm *Handshake) SendVersion(peer salticidae.PeerID) error {
 	build := Builder{}
-	v, err := build.Version(nm.networkID, nm.clock.Unix(), toIPDesc(nm.myAddr), CurrentVersion)
+	v, err := build.Version(nm.networkID, nm.clock.Unix(), toIPDesc(nm.myAddr), ClientVersion)
 	if err != nil {
 		return fmt.Errorf("packing Version failed due to %s", err)
 	}
@@ -331,25 +437,27 @@ func connHandler(_conn *C.struct_msgnetwork_conn_t, connected C.bool, _ unsafe.P
 		return connected
 	}
 
+	HandshakeNet.requestedLock.Lock()
+	defer HandshakeNet.requestedLock.Unlock()
+
 	conn := salticidae.MsgNetworkConnFromC(salticidae.CMsgNetworkConn(_conn))
 	addr := conn.GetAddr()
 	ip := toIPDesc(addr)
-
 	ipStr := ip.String()
-	if _, exists := HandshakeNet.requestedConnections[ipStr]; !exists {
+
+	ipID := ids.NewID(hashing.ComputeHash256Array([]byte(ipStr)))
+	HandshakeNet.requestedTimeout.Remove(ipID)
+
+	if _, exists := HandshakeNet.requested[ipStr]; !exists {
 		HandshakeNet.log.Debug("connHandler called with %s", ip)
 		return true
 	}
-	delete(HandshakeNet.requestedConnections, ipStr)
+	delete(HandshakeNet.requested, ipStr)
 
 	cert := conn.GetPeerCert()
-	peer := salticidae.NewPeerIDFromX509(cert, false)
+	peer := salticidae.NewPeerIDFromX509(cert, true)
 
-	HandshakeNet.net.AddPeer(peer)
-	HandshakeNet.net.SetPeerAddr(peer, addr)
-	HandshakeNet.net.ConnPeer(peer, 600, 1)
-
-	peer.Free()
+	HandshakeNet.ConnectTo(peer, getCert(cert), addr)
 	return true
 }
 
@@ -372,8 +480,6 @@ func (nm *Handshake) connectedToPeer(conn *C.struct_peernetwork_conn_t, peer sal
 
 	nm.reconnectTimeout.Remove(peerID)
 
-	nm.pending.Add(peer, cert, utils.IPDesc{})
-
 	handler := new(func())
 	*handler = func() {
 		if nm.pending.ContainsPeerID(peer) {
@@ -388,34 +494,39 @@ func (nm *Handshake) disconnectedFromPeer(peer salticidae.PeerID) {
 	cert := ids.ShortID{}
 	if pendingCert, exists := nm.pending.GetID(peer); exists {
 		cert = pendingCert
+		nm.log.Info("Disconnected from pending peer %s", cert)
 	} else if connectedCert, exists := nm.connections.GetID(peer); exists {
 		cert = connectedCert
+		nm.log.Info("Disconnected from peer %s", cert)
 	} else {
 		return
 	}
 
-	nm.log.Info("Disconnected from %s", cert)
-
 	peerBytes := toID(peer)
 	peerID := ids.NewID(peerBytes)
 
+	nm.versionTimeout.Remove(peerID)
+	nm.connections.Remove(peer, cert)
+	nm.numPeers.Set(float64(nm.connections.Len()))
+
 	if nm.vdrs.Contains(cert) {
 		nm.reconnectTimeout.Put(peerID, func() {
+			nm.pending.Remove(peer, cert)
+			nm.connections.Remove(peer, cert)
 			nm.net.DelPeer(peer)
+
+			nm.numPeers.Set(float64(nm.connections.Len()))
 		})
+		nm.pending.Add(peer, cert, utils.IPDesc{})
 	} else {
+		nm.pending.Remove(peer, cert)
 		nm.net.DelPeer(peer)
 	}
-	nm.versionTimeout.Remove(peerID)
 
 	if !nm.enableStaking {
 		nm.vdrs.Remove(cert)
 	}
 
-	nm.pending.RemovePeerID(peer)
-	nm.connections.RemovePeerID(peer)
-	nm.numPeers.Set(float64(nm.connections.Len()))
-
 	nm.awaitingLock.Lock()
 	defer nm.awaitingLock.Unlock()
 	for _, awaiting := range HandshakeNet.awaiting {
@@ -423,6 +534,59 @@ func (nm *Handshake) disconnectedFromPeer(peer salticidae.PeerID) {
 	}
 }
 
+// checkCompatibility Check to make sure that the peer and I speak the same language.
+func (nm *Handshake) checkCompatibility(peerVersion string) bool {
+	if !strings.HasPrefix(peerVersion, VersionPrefix) {
+		nm.log.Warn("Peer attempted to connect with an invalid version prefix")
+		return false
+	}
+	peerVersion = peerVersion[len(VersionPrefix):]
+	splitPeerVersion := strings.SplitN(peerVersion, VersionSeparator, 3)
+	if len(splitPeerVersion) != 3 {
+		nm.log.Warn("Peer attempted to connect with an invalid number of subversions")
+		return false
+	}
+
+	major, err := strconv.Atoi(splitPeerVersion[0])
+	if err != nil {
+		nm.log.Warn("Peer attempted to connect with an invalid major version")
+		return false
+	}
+	minor, err := strconv.Atoi(splitPeerVersion[1])
+	if err != nil {
+		nm.log.Warn("Peer attempted to connect with an invalid minor version")
+		return false
+	}
+	patch, err := strconv.Atoi(splitPeerVersion[2])
+	if err != nil {
+		nm.log.Warn("Peer attempted to connect with an invalid patch version")
+		return false
+	}
+
+	switch {
+	case major < MajorVersion:
+		// peers major version is too low
+		return false
+	case major > MajorVersion:
+		nm.log.Warn("Peer attempted to connect with a higher major version, this client may need to be updated")
+		return false
+	}
+
+	switch {
+	case minor < MinorVersion:
+		// peers minor version is too low
+		return false
+	case minor > MinorVersion:
+		nm.log.Warn("Peer attempted to connect with a higher minor version, this client may need to be updated")
+		return false
+	}
+
+	if patch > PatchVersion {
+		nm.log.Warn("Peer is connecting with a higher patch version, this client may need to be updated")
+	}
+	return true
+}
+
 // peerHandler notifies a change to the set of connected peers
 // connected is true if a new peer is connected
 // connected is false if a formerly connected peer has disconnected
@@ -451,19 +615,27 @@ func unknownPeerHandler(_addr *C.netaddr_t, _cert *C.x509_t, _ unsafe.Pointer) {
 	HandshakeNet.log.Info("Adding peer %s", ip)
 
 	var peer salticidae.PeerID
+	var id ids.ShortID
 	if HandshakeNet.enableStaking {
 		cert := salticidae.X509FromC(salticidae.CX509(_cert))
 		peer = salticidae.NewPeerIDFromX509(cert, true)
+		id = getCert(cert)
 	} else {
 		peer = salticidae.NewPeerIDFromNetAddr(addr, true)
+		id = toShortID(ip)
 	}
 
 	peerBytes := toID(peer)
 	peerID := ids.NewID(peerBytes)
 
 	HandshakeNet.reconnectTimeout.Put(peerID, func() {
+		HandshakeNet.pending.Remove(peer, id)
+		HandshakeNet.connections.Remove(peer, id)
 		HandshakeNet.net.DelPeer(peer)
+
+		HandshakeNet.numPeers.Set(float64(HandshakeNet.connections.Len()))
 	})
+	HandshakeNet.pending.Add(peer, id, utils.IPDesc{})
 	HandshakeNet.net.AddPeer(peer)
 }
 
@@ -506,12 +678,17 @@ func version(_msg *C.struct_msg_t, _conn *C.struct_msgnetwork_conn_t, _ unsafe.P
 	conn := salticidae.PeerNetworkConnFromC(salticidae.CPeerNetworkConn(_conn))
 	peer := conn.GetPeerID(true)
 
+	peerBytes := toID(peer)
+	peerID := ids.NewID(peerBytes)
+
+	HandshakeNet.versionTimeout.Remove(peerID)
+
 	id, exists := HandshakeNet.pending.GetID(peer)
 	if !exists {
+		HandshakeNet.log.Warn("Dropping Version message because the peer isn't pending")
 		return
 	}
-
-	defer HandshakeNet.pending.Remove(peer, id)
+	HandshakeNet.pending.Remove(peer, id)
 
 	build := Builder{}
 	pMsg, err := build.Parse(Version, msg.GetPayloadByMove())
@@ -537,8 +714,8 @@ func version(_msg *C.struct_msg_t, _conn *C.struct_msgnetwork_conn_t, _ unsafe.P
 		return
 	}
 
-	if peerVersion := pMsg.Get(VersionStr).(string); !checkCompatibility(CurrentVersion, peerVersion) {
-		HandshakeNet.log.Warn("Bad version")
+	if peerVersion := pMsg.Get(VersionStr).(string); !HandshakeNet.checkCompatibility(peerVersion) {
+		HandshakeNet.log.Debug("Dropping connection due to an incompatible version from peer")
 
 		HandshakeNet.net.DelPeer(peer)
 		return
@@ -550,18 +727,12 @@ func version(_msg *C.struct_msg_t, _conn *C.struct_msgnetwork_conn_t, _ unsafe.P
 
 	HandshakeNet.SendPeerList(peer)
 	HandshakeNet.connections.Add(peer, id, ip)
-
-	peerBytes := toID(peer)
-	peerID := ids.NewID(peerBytes)
-
-	HandshakeNet.versionTimeout.Remove(peerID)
+	HandshakeNet.numPeers.Set(float64(HandshakeNet.connections.Len()))
 
 	if !HandshakeNet.enableStaking {
 		HandshakeNet.vdrs.Add(validators.NewValidator(id, 1))
 	}
 
-	HandshakeNet.numPeers.Set(float64(HandshakeNet.connections.Len()))
-
 	HandshakeNet.awaitingLock.Lock()
 	defer HandshakeNet.awaitingLock.Unlock()
 
@@ -612,20 +783,14 @@ func peerList(_msg *C.struct_msg_t, _conn *C.struct_msgnetwork_conn_t, _ unsafe.
 	cErr := salticidae.NewError()
 	for _, ip := range ips {
 		addr := salticidae.NewNetAddrFromIPPortString(ip.String(), true, &cErr)
-		if cErr.GetCode() == 0 && !HandshakeNet.myAddr.IsEq(addr) { // Make sure not to connect to myself
-			ip := toIPDesc(addr)
-			if !HandshakeNet.pending.ContainsIP(ip) && !HandshakeNet.connections.ContainsIP(ip) {
-				HandshakeNet.log.Debug("Adding peer %s", ip)
 
-				HandshakeNet.Connect(addr)
-			}
+		if cErr.GetCode() != 0 || HandshakeNet.myAddr.IsEq(addr) {
+			// Make sure not to connect to myself
+			continue
 		}
-	}
-}
 
-func getMsgCert(_conn *C.struct_msgnetwork_conn_t) ids.ShortID {
-	conn := salticidae.MsgNetworkConnFromC(salticidae.CMsgNetworkConn(_conn))
-	return getCert(conn.GetPeerCert())
+		HandshakeNet.Connect(addr)
+	}
 }
 
 func getPeerCert(_conn *C.struct_peernetwork_conn_t) ids.ShortID {
@@ -635,29 +800,16 @@ func getPeerCert(_conn *C.struct_peernetwork_conn_t) ids.ShortID {
 
 func getCert(cert salticidae.X509) ids.ShortID {
 	der := cert.GetDer(false)
-	defer der.Free()
-
 	certDS := salticidae.NewDataStreamMovedFromByteArray(der, false)
-	defer certDS.Free()
-
 	certBytes := certDS.GetDataInPlace(certDS.Size()).Get()
 	certID, err := ids.ToShortID(hashing.PubkeyBytesToAddress(certBytes))
+
+	certDS.Free()
+	der.Free()
 	HandshakeNet.log.AssertNoError(err)
 	return certID
 }
 
-// checkCompatibility Check to make sure that the peer and I speak the same language.
-func checkCompatibility(myVersion string, peerVersion string) bool {
-	// At the moment, we are all compatible.
-	return true
-}
-
-func toAddr(ip utils.IPDesc, autoFree bool) salticidae.NetAddr {
-	err := salticidae.NewError()
-	addr := salticidae.NewNetAddrFromIPPortString(ip.String(), autoFree, &err)
-	HandshakeNet.log.AssertTrue(err.GetCode() == 0, "IP Failed parsing")
-	return addr
-}
 func toShortID(ip utils.IPDesc) ids.ShortID {
 	return ids.NewShortID(hashing.ComputeHash160Array([]byte(ip.String())))
 }

node/node.go

@@ -5,7 +5,7 @@ package node
 
 // #include "salticidae/network.h"
 // void onTerm(int sig, void *);
-// void errorHandler(SalticidaeCError *, bool, void *);
+// void errorHandler(SalticidaeCError *, bool, int32_t, void *);
 import "C"
 
 import (
@@ -130,14 +130,14 @@ func onTerm(C.int, unsafe.Pointer) {
 }
 
 //export errorHandler
-func errorHandler(_err *C.struct_SalticidaeCError, fatal C.bool, _ unsafe.Pointer) {
+func errorHandler(_err *C.struct_SalticidaeCError, fatal C.bool, asyncID C.int32_t, _ unsafe.Pointer) {
 	err := (*salticidae.Error)(unsafe.Pointer(_err))
 	if fatal {
 		MainNode.Log.Fatal("Error during async call: %s", salticidae.StrError(err.GetCode()))
 		MainNode.EC.Stop()
 		return
 	}
-	MainNode.Log.Error("Error during async call: %s", salticidae.StrError(err.GetCode()))
+	MainNode.Log.Debug("Error during async with ID %d call: %s", asyncID, salticidae.StrError(err.GetCode()))
 }
 
 func (n *Node) initNetlib() error {
@@ -152,6 +152,8 @@ func (n *Node) initNetlib() error {
 
 	// Create peer network config, may have tls enabled
 	peerConfig := salticidae.NewPeerNetworkConfig()
+	peerConfig.ConnTimeout(60)
+
 	msgConfig := peerConfig.AsMsgNetworkConfig()
 	msgConfig.MaxMsgSize(maxMessageSize)
 
@@ -254,7 +256,7 @@ func (n *Node) StartConsensusServer() error {
 	// Listen for P2P messages
 	n.PeerNet.Listen(serverIP, &err)
 	if code := err.GetCode(); code != 0 {
-		return fmt.Errorf("failed to start consensus server: %s", salticidae.StrError(code))
+		return fmt.Errorf("failed to listen on consensus server at %s: %s", n.Config.StakingIP, salticidae.StrError(code))
 	}
 
 	// Start a server to handle throughput tests if configuration says to. Disabled by default.
@@ -268,7 +270,7 @@ func (n *Node) StartConsensusServer() error {
 
 		n.ClientNet.Listen(clientIP, &err)
 		if code := err.GetCode(); code != 0 {
-			return fmt.Errorf("failed to listen on xput server: %s", salticidae.StrError(code))
+			return fmt.Errorf("failed to listen on xput server at 127.0.0.1:%d: %s", n.Config.ThroughputPort, salticidae.StrError(code))
 		}
 	}
 

scripts/ansible/inventory.yml

@@ -28,6 +28,7 @@ borealis_bootstrap:
     staking_tls_enabled: true
     staking_tls_key_file: "/home/ubuntu/keys/staker.key"
     staking_tls_cert_file: "/home/ubuntu/keys/staker.crt"
+    plugin_dir: "/home/ubuntu/go/src/github.com/ava-labs/gecko/build/plugins"
     log_dir: "/home/ubuntu/.gecko"
     log_level: debug
     snow_sample_size: 3
@@ -73,6 +74,7 @@ borealis_node:
     staking_tls_enabled: true
     staking_tls_key_file: "/home/ubuntu/keys/staker.key"
     staking_tls_cert_file: "/home/ubuntu/keys/staker.crt"
+    plugin_dir: "/home/ubuntu/go/src/github.com/ava-labs/gecko/build/plugins"
     log_dir: "/home/ubuntu/.gecko"
     log_level: debug
     snow_sample_size: 3

scripts/ansible/kill_playbook.yml

@@ -0,0 +1,9 @@
+
+---
+- name: Update the network
+  connection: ssh
+  gather_facts: false
+  hosts: all
+  tasks:
+    - name: Kill Node
+      command: killall ava

scripts/ansible/ping_playbook.yml

@@ -0,0 +1,11 @@
+
+---
+- name: Update the network
+  connection: ssh
+  gather_facts: false
+  hosts: all
+  tasks:
+    - name: Ping node
+      shell: "ls"
+      environment:
+        PATH: /sbin:/usr/sbin:/bin:/usr/bin:/usr/local/bin:/snap/bin

scripts/ansible/restart_playbook.yml

@@ -7,8 +7,8 @@
   vars:
     ava_binary: ~/go/src/github.com/ava-labs/gecko/build/ava
     repo_folder: ~/go/src/github.com/ava-labs/gecko
-    repo_name: ava-labs/gecko
-    repo_branch: cascade
+    repo_name: ava-labs/gecko-internal
+    repo_branch: retry-connections
   tasks:
     - name: Kill Node
       command: killall ava
@@ -33,6 +33,6 @@
         path: "{{ log_dir }}"
         state: absent
     - name: Start node
-      shell: "nohup {{ ava_binary }} --network-id={{ network_id }} --api-admin-enabled={{ api_admin_enabled }} --api-keystore-enabled={{ api_keystore_enabled }} --api-metrics-enabled={{ api_metrics_enabled }} --ava-tx-fee={{ ava_tx_fee }} --assertions-enabled={{ assertions_enabled }} --signature-verification-enabled={{ signature_verification_enabled }} --db-enabled={{ db_enabled }} --db-dir={{ db_dir }} --http-port={{ http_port }} --http-tls-enabled={{ http_tls_enabled }} --http-tls-key-file={{ http_tls_key_file }} --http-tls-cert-file={{ http_tls_cert_file }} --bootstrap-ips={{ bootstrap_ips }} --bootstrap-ids={{ bootstrap_ids }} --public-ip={{ ansible_host }} --staking-port={{ staking_port }} --staking-tls-enabled={{ staking_tls_enabled }} --staking-tls-key-file={{ staking_tls_key_file }} --staking-tls-cert-file={{ staking_tls_cert_file }} --log-dir={{ log_dir }} --log-level={{ log_level }} --snow-sample-size={{ snow_sample_size }} --snow-quorum-size={{ snow_quorum_size }} --snow-virtuous-commit-threshold={{ snow_virtuous_commit_threshold }} --snow-rogue-commit-threshold={{ snow_rogue_commit_threshold }} --snow-avalanche-num-parents={{ snow_avalanche_num_parents }} --snow-avalanche-batch-size={{ snow_avalanche_batch_size }} --api-ipcs-enabled={{ api_ipcs_enabled }} --xput-server-enabled={{ xput_server_enabled }} --xput-server-port={{ xput_server_port }} >/dev/null 2>&1 &"
+      shell: "nohup {{ ava_binary }} --network-id={{ network_id }} --api-admin-enabled={{ api_admin_enabled }} --api-keystore-enabled={{ api_keystore_enabled }} --api-metrics-enabled={{ api_metrics_enabled }} --ava-tx-fee={{ ava_tx_fee }} --assertions-enabled={{ assertions_enabled }} --signature-verification-enabled={{ signature_verification_enabled }} --db-enabled={{ db_enabled }} --db-dir={{ db_dir }} --http-port={{ http_port }} --http-tls-enabled={{ http_tls_enabled }} --http-tls-key-file={{ http_tls_key_file }} --http-tls-cert-file={{ http_tls_cert_file }} --bootstrap-ips={{ bootstrap_ips }} --bootstrap-ids={{ bootstrap_ids }} --public-ip={{ ansible_host }} --staking-port={{ staking_port }} --staking-tls-enabled={{ staking_tls_enabled }} --staking-tls-key-file={{ staking_tls_key_file }} --staking-tls-cert-file={{ staking_tls_cert_file }} --plugin-dir={{ plugin_dir }} --log-dir={{ log_dir }} --log-level={{ log_level }} --snow-sample-size={{ snow_sample_size }} --snow-quorum-size={{ snow_quorum_size }} --snow-virtuous-commit-threshold={{ snow_virtuous_commit_threshold }} --snow-rogue-commit-threshold={{ snow_rogue_commit_threshold }} --snow-avalanche-num-parents={{ snow_avalanche_num_parents }} --snow-avalanche-batch-size={{ snow_avalanche_batch_size }} --api-ipcs-enabled={{ api_ipcs_enabled }} --xput-server-enabled={{ xput_server_enabled }} --xput-server-port={{ xput_server_port }} >/dev/null 2>&1 &"
       environment:
         PATH: /sbin:/usr/sbin:/bin:/usr/bin:/usr/local/bin:/snap/bin

scripts/ansible/update_playbook.yml

@@ -7,8 +7,8 @@
   vars:
     ava_binary: ~/go/src/github.com/ava-labs/gecko/build/ava
     repo_folder: ~/go/src/github.com/ava-labs/gecko
-    repo_name: ava-labs/gecko
-    repo_branch: cascade
+    repo_name: ava-labs/gecko-internal
+    repo_branch: retry-connections
   tasks:
     - name: Kill Node
       command: killall ava
@@ -25,6 +25,6 @@
       environment:
         PATH: /sbin:/usr/sbin:/bin:/usr/bin:/usr/local/bin:/snap/bin
     - name: Start node
-      shell: "nohup {{ ava_binary }} --network-id={{ network_id }} --api-admin-enabled={{ api_admin_enabled }} --api-keystore-enabled={{ api_keystore_enabled }} --api-metrics-enabled={{ api_metrics_enabled }} --ava-tx-fee={{ ava_tx_fee }} --assertions-enabled={{ assertions_enabled }} --signature-verification-enabled={{ signature_verification_enabled }} --db-enabled={{ db_enabled }} --db-dir={{ db_dir }} --http-port={{ http_port }} --http-tls-enabled={{ http_tls_enabled }} --http-tls-key-file={{ http_tls_key_file }} --http-tls-cert-file={{ http_tls_cert_file }} --bootstrap-ips={{ bootstrap_ips }} --bootstrap-ids={{ bootstrap_ids }} --public-ip={{ ansible_host }} --staking-port={{ staking_port }} --staking-tls-enabled={{ staking_tls_enabled }} --staking-tls-key-file={{ staking_tls_key_file }} --staking-tls-cert-file={{ staking_tls_cert_file }} --log-dir={{ log_dir }} --log-level={{ log_level }} --snow-sample-size={{ snow_sample_size }} --snow-quorum-size={{ snow_quorum_size }} --snow-virtuous-commit-threshold={{ snow_virtuous_commit_threshold }} --snow-rogue-commit-threshold={{ snow_rogue_commit_threshold }} --snow-avalanche-num-parents={{ snow_avalanche_num_parents }} --snow-avalanche-batch-size={{ snow_avalanche_batch_size }} --api-ipcs-enabled={{ api_ipcs_enabled }} --xput-server-enabled={{ xput_server_enabled }} --xput-server-port={{ xput_server_port }} >/dev/null 2>&1 &"
+      shell: "nohup {{ ava_binary }} --network-id={{ network_id }} --api-admin-enabled={{ api_admin_enabled }} --api-keystore-enabled={{ api_keystore_enabled }} --api-metrics-enabled={{ api_metrics_enabled }} --ava-tx-fee={{ ava_tx_fee }} --assertions-enabled={{ assertions_enabled }} --signature-verification-enabled={{ signature_verification_enabled }} --db-enabled={{ db_enabled }} --db-dir={{ db_dir }} --http-port={{ http_port }} --http-tls-enabled={{ http_tls_enabled }} --http-tls-key-file={{ http_tls_key_file }} --http-tls-cert-file={{ http_tls_cert_file }} --bootstrap-ips={{ bootstrap_ips }} --bootstrap-ids={{ bootstrap_ids }} --public-ip={{ ansible_host }} --staking-port={{ staking_port }} --staking-tls-enabled={{ staking_tls_enabled }} --staking-tls-key-file={{ staking_tls_key_file }} --staking-tls-cert-file={{ staking_tls_cert_file }} --plugin-dir={{ plugin_dir }} --log-dir={{ log_dir }} --log-level={{ log_level }} --snow-sample-size={{ snow_sample_size }} --snow-quorum-size={{ snow_quorum_size }} --snow-virtuous-commit-threshold={{ snow_virtuous_commit_threshold }} --snow-rogue-commit-threshold={{ snow_rogue_commit_threshold }} --snow-avalanche-num-parents={{ snow_avalanche_num_parents }} --snow-avalanche-batch-size={{ snow_avalanche_batch_size }} --api-ipcs-enabled={{ api_ipcs_enabled }} --xput-server-enabled={{ xput_server_enabled }} --xput-server-port={{ xput_server_port }} >/dev/null 2>&1 &"
       environment:
         PATH: /sbin:/usr/sbin:/bin:/usr/bin:/usr/local/bin:/snap/bin

snow/consensus/snowball/parameters.go

@@ -10,20 +10,21 @@ import (
 )
 
 const (
-	errMsg = "__________                    .___\n" +
-		"\\______   \\____________     __| _/__.__.\n" +
-		" |    |  _/\\_  __ \\__  \\   / __ <   |  |\n" +
-		" |    |   \\ |  | \\// __ \\_/ /_/ |\\___  |\n" +
-		" |______  / |__|  (____  /\\____ |/ ____|\n" +
-		"        \\/             \\/      \\/\\/\n" +
+	errMsg = "" +
+		`__________                    .___` + "\n" +
+		`\______   \____________     __| _/__.__.` + "\n" +
+		` |    |  _/\_  __ \__  \   / __ <   |  |` + "\n" +
+		` |    |   \ |  | \// __ \_/ /_/ |\___  |` + "\n" +
+		` |______  / |__|  (____  /\____ |/ ____|` + "\n" +
+		`        \/             \/      \/\/` + "\n" +
 		"\n" +
-		"🏆      🏆      🏆     🏆      🏆      🏆\n" +
-		"  ________ ________      ________________\n" +
-		" /  _____/ \\_____  \\    /  _  \\__    ___/\n" +
-		"/   \\  ___  /   |   \\  /  /_\\  \\|    |\n" +
-		"\\    \\_\\  \\/    |    \\/    |    \\    |\n" +
-		" \\______  /\\_______  /\\____|__  /____|\n" +
-		"        \\/         \\/         \\/\n"
+		`🏆      🏆      🏆     🏆      🏆      🏆` + "\n" +
+		`  ________ ________      ________________` + "\n" +
+		` /  _____/ \_____  \    /  _  \__    ___/` + "\n" +
+		`/   \  ___  /   |   \  /  /_\  \|    |` + "\n" +
+		`\    \_\  \/    |    \/    |    \    |` + "\n" +
+		` \______  /\_______  /\____|__  /____|` + "\n" +
+		`        \/         \/         \/` + "\n"
 )
 
 // Parameters required for snowball consensus

snow/engine/avalanche/bootstrapper.go

@@ -207,6 +207,7 @@ func (b *bootstrapper) finish() {
 func (b *bootstrapper) executeAll(jobs *queue.Jobs, numBlocked prometheus.Gauge) {
 	for job, err := jobs.Pop(); err == nil; job, err = jobs.Pop() {
 		numBlocked.Dec()
+		b.BootstrapConfig.Context.Log.Debug("Executing: %s", job.ID())
 		if err := jobs.Execute(job); err != nil {
 			b.BootstrapConfig.Context.Log.Warn("Error executing: %s", err)
 		}

snow/engine/avalanche/state/serializer.go

@@ -82,6 +82,10 @@ func (s *Serializer) ParseVertex(b []byte) (avacon.Vertex, error) {
 
 // BuildVertex implements the avalanche.State interface
 func (s *Serializer) BuildVertex(parentSet ids.Set, txs []snowstorm.Tx) (avacon.Vertex, error) {
+	if len(txs) == 0 {
+		return nil, errNoTxs
+	}
+
 	parentIDs := parentSet.List()
 	ids.SortIDs(parentIDs)
 	sortTxs(txs)

snow/engine/avalanche/state/vertex.go

@@ -24,6 +24,7 @@ var (
 	errExtraSpace     = errors.New("trailing buffer space")
 	errInvalidParents = errors.New("vertex contains non-sorted or duplicated parentIDs")
 	errInvalidTxs     = errors.New("vertex contains non-sorted or duplicated transactions")
+	errNoTxs          = errors.New("vertex contains no transactions")
 )
 
 type vertex struct {
@@ -45,6 +46,8 @@ func (vtx *vertex) Verify() error {
 	switch {
 	case !ids.IsSortedAndUniqueIDs(vtx.parentIDs):
 		return errInvalidParents
+	case len(vtx.txs) == 0:
+		return errNoTxs
 	case !isSortedAndUniqueTxs(vtx.txs):
 		return errInvalidTxs
 	default:
@@ -55,7 +58,7 @@ func (vtx *vertex) Verify() error {
 /*
  * Vertex:
  * Codec        | 04 Bytes
- * Chain       | 32 Bytes
+ * Chain        | 32 Bytes
  * Height       | 08 Bytes
  * NumParents   | 04 Bytes
  * Repeated (NumParents):

snow/engine/avalanche/transitive.go

@@ -316,8 +316,37 @@ func (t *Transitive) batch(txs []snowstorm.Tx, force, empty bool) {
 		}
 	}
 
-	if len(batch) > 0 || (empty && !issued) {
+	if len(batch) > 0 {
 		t.issueBatch(batch)
+	} else if empty && !issued {
+		t.issueRepoll()
+	}
+}
+
+func (t *Transitive) issueRepoll() {
+	preferredIDs := t.Consensus.Preferences().List()
+	numPreferredIDs := len(preferredIDs)
+	if numPreferredIDs == 0 {
+		t.Config.Context.Log.Error("Re-query attempt was dropped due to no pending vertices")
+		return
+	}
+
+	sampler := random.Uniform{N: len(preferredIDs)}
+	vtxID := preferredIDs[sampler.Sample()]
+
+	p := t.Consensus.Parameters()
+	vdrs := t.Config.Validators.Sample(p.K) // Validators to sample
+
+	vdrSet := ids.ShortSet{} // Validators to sample repr. as a set
+	for _, vdr := range vdrs {
+		vdrSet.Add(vdr.ID())
+	}
+
+	t.RequestID++
+	if numVdrs := len(vdrs); numVdrs == p.K && t.polls.Add(t.RequestID, vdrSet.Len()) {
+		t.Config.Sender.PullQuery(vdrSet, t.RequestID, vtxID)
+	} else if numVdrs < p.K {
+		t.Config.Context.Log.Error("Re-query for %s was dropped due to an insufficient number of validators", vtxID)
 	}
 }
 

snow/engine/avalanche/transitive_test.go

@@ -698,23 +698,12 @@ func TestEngineScheduleRepoll(t *testing.T) {
 
 	sender.PushQueryF = nil
 
-	st.buildVertex = func(_ ids.Set, txs []snowstorm.Tx) (avalanche.Vertex, error) {
-		consumers := []snowstorm.Tx{}
-		for _, tx := range txs {
-			consumers = append(consumers, tx)
-		}
-		return &Vtx{
-			parents: []avalanche.Vertex{gVtx, mVtx},
-			id:      GenerateID(),
-			txs:     consumers,
-			status:  choices.Processing,
-			bytes:   []byte{1},
-		}, nil
-	}
-
 	repolled := new(bool)
-	sender.PushQueryF = func(_ ids.ShortSet, _ uint32, _ ids.ID, _ []byte) {
+	sender.PullQueryF = func(_ ids.ShortSet, _ uint32, vtxID ids.ID) {
 		*repolled = true
+		if !vtxID.Equals(vtx.ID()) {
+			t.Fatalf("Wrong vertex queried")
+		}
 	}
 
 	te.QueryFailed(vdr.ID(), *requestID)
@@ -979,31 +968,14 @@ func TestEngineIssueRepoll(t *testing.T) {
 	te.Initialize(config)
 	te.finishBootstrapping()
 
-	newVtxID := new(ids.ID)
-
-	st.buildVertex = func(s ids.Set, txs []snowstorm.Tx) (avalanche.Vertex, error) {
-		if len(txs) != 0 {
-			t.Fatalf("Wrong vertex issued")
-		}
-		if s.Len() != 2 || !s.Contains(gVtx.ID()) || !s.Contains(mVtx.ID()) {
-			t.Fatalf("Wrong vertex issued")
-		}
-
-		vtx := &Vtx{
-			parents: []avalanche.Vertex{gVtx, mVtx},
-			id:      GenerateID(),
-			status:  choices.Processing,
-			bytes:   []byte{1},
-		}
-		*newVtxID = vtx.ID()
-		return vtx, nil
-	}
-
-	sender.PushQueryF = func(vdrs ids.ShortSet, _ uint32, vtxID ids.ID, vtx []byte) {
+	sender.PullQueryF = func(vdrs ids.ShortSet, _ uint32, vtxID ids.ID) {
 		vdrSet := ids.ShortSet{}
 		vdrSet.Add(vdr.ID())
-		if !vdrs.Equals(vdrSet) || !vtxID.Equals(*newVtxID) {
-			t.Fatalf("Wrong query message")
+		if !vdrs.Equals(vdrSet) {
+			t.Fatalf("Wrong query recipients")
+		}
+		if !vtxID.Equals(gVtx.ID()) && !vtxID.Equals(mVtx.ID()) {
+			t.Fatalf("Unknown re-query")
 		}
 	}
 

snow/engine/avalanche/voter.go

@@ -79,12 +79,23 @@ func (v *voter) bubbleVotes(votes ids.UniqueBag) ids.UniqueBag {
 			vtx := vts[0]
 			vts = vts[1:]
 
-			if status := vtx.Status(); status.Fetched() && !v.t.Consensus.VertexIssued(vtx) {
-				vts = append(vts, vtx.Parents()...)
-			} else if !status.Decided() && v.t.Consensus.VertexIssued(vtx) {
+			status := vtx.Status()
+			if !status.Fetched() {
+				v.t.Config.Context.Log.Verbo("Dropping %d vote(s) for %s because the vertex is unknown", set.Len(), vtx.ID())
+				continue
+			}
+
+			if status.Decided() {
+				v.t.Config.Context.Log.Verbo("Dropping %d vote(s) for %s because the vertex is decided", set.Len(), vtx.ID())
+				continue
+			}
+
+			if v.t.Consensus.VertexIssued(vtx) {
+				v.t.Config.Context.Log.Verbo("Applying %d vote(s) for %s", set.Len(), vtx.ID())
 				bubbledVotes.UnionSet(vtx.ID(), set)
 			} else {
-				v.t.Config.Context.Log.Debug("Dropping %d vote(s) for %s because the vertex is invalid", set.Len(), vtx.ID())
+				v.t.Config.Context.Log.Verbo("Bubbling %d vote(s) for %s because the vertex isn't issued", set.Len(), vtx.ID())
+				vts = append(vts, vtx.Parents()...)
 			}
 		}
 	}

vms/platformvm/service_test.go

@@ -9,7 +9,7 @@ import (
 )
 
 func TestAddDefaultSubnetValidator(t *testing.T) {
-	expectedJSONString := `{"startTime":"0","endtime":"0","id":null,"destination":null,"delegationFeeRate":"0","payerNonce":"0"}`
+	expectedJSONString := `{"startTime":"0","endTime":"0","id":null,"destination":null,"delegationFeeRate":"0","payerNonce":"0"}`
 	args := AddDefaultSubnetValidatorArgs{}
 	bytes, err := json.Marshal(&args)
 	if err != nil {

vms/platformvm/static_service.go

@@ -44,7 +44,7 @@ type APIAccount struct {
 // is sent when this staker is done staking.
 type APIValidator struct {
 	StartTime   json.Uint64  `json:"startTime"`
-	EndTime     json.Uint64  `json:"endtime"`
+	EndTime     json.Uint64  `json:"endTime"`
 	Weight      *json.Uint64 `json:"weight,omitempty"`
 	StakeAmount *json.Uint64 `json:"stakeAmount,omitempty"`
 	ID          ids.ShortID  `json:"id"`