Track subnet uptimes (ava-labs#1427)

chains/manager.go

@@ -681,6 +681,7 @@ func (m *manager) createAvalancheChain(
 		sb.afterBootstrapped(),
 		m.ConsensusGossipFrequency,
 		m.ResourceTracker,
+		validators.UnhandledSubnetConnector, // avalanche chains don't use subnet connector
 	)
 	if err != nil {
 		return nil, fmt.Errorf("error initializing network handler: %w", err)
@@ -843,8 +844,13 @@ func (m *manager) createSnowmanChain(
 		return nil, fmt.Errorf("problem initializing event dispatcher: %w", err)
 	}
 
-	// first vm to be init is P-Chain once, which provides validator interface to all ProposerVMs
-	var bootstrapFunc func()
+	var (
+		bootstrapFunc   func()
+		subnetConnector = validators.UnhandledSubnetConnector
+	)
+	// If [m.validatorState] is nil then we are creating the P-Chain. Since the
+	// P-Chain is the first chain to be created, we can use it to initialize
+	// required interfaces for the other chains
 	if m.validatorState == nil {
 		valState, ok := vm.(validators.State)
 		if !ok {
@@ -878,6 +884,12 @@ func (m *manager) createSnowmanChain(
 		bootstrapFunc = func() {
 			close(m.unblockChainCreatorCh)
 		}
+
+		// Set up the subnet connector for the P-Chain
+		subnetConnector, ok = vm.(validators.SubnetConnector)
+		if !ok {
+			return nil, fmt.Errorf("expected validators.SubnetConnector but got %T", vm)
+		}
 	}
 
 	// Initialize the ProposerVM and the vm wrapped inside it
@@ -942,6 +954,7 @@ func (m *manager) createSnowmanChain(
 		sb.afterBootstrapped(),
 		m.ConsensusGossipFrequency,
 		m.ResourceTracker,
+		subnetConnector,
 	)
 	if err != nil {
 		return nil, fmt.Errorf("couldn't initialize message handler: %w", err)

message/internal_msg_builder.go

@@ -406,6 +406,25 @@ func InternalConnected(nodeID ids.NodeID, nodeVersion *version.Application) Inbo
 	}
 }
 
+// ConnectedSubnet contains the subnet ID of the subnet that the node is
+// connected to.
+type ConnectedSubnet struct {
+	SubnetID ids.ID
+}
+
+// InternalConnectedSubnet returns a message that indicates the node with [nodeID] is
+// connected to the subnet with the given [subnetID].
+func InternalConnectedSubnet(nodeID ids.NodeID, subnetID ids.ID) InboundMessage {
+	return &inboundMessage{
+		nodeID: nodeID,
+		op:     ConnectedSubnetOp,
+		message: &ConnectedSubnet{
+			SubnetID: subnetID,
+		},
+		expiration: mockable.MaxTime,
+	}
+}
+
 type Disconnected struct{}
 
 func InternalDisconnected(nodeID ids.NodeID) InboundMessage {

message/ops.go

@@ -58,6 +58,7 @@ const (
 	CrossChainAppResponseOp
 	// Internal:
 	ConnectedOp
+	ConnectedSubnetOp
 	DisconnectedOp
 	NotifyOp
 	GossipRequestOp
@@ -116,6 +117,7 @@ var (
 		CrossChainAppRequestFailedOp,
 		CrossChainAppResponseOp,
 		ConnectedOp,
+		ConnectedSubnetOp,
 		DisconnectedOp,
 		NotifyOp,
 		GossipRequestOp,
@@ -153,6 +155,7 @@ var (
 		ChitsOp,
 		// Internal
 		ConnectedOp,
+		ConnectedSubnetOp,
 		DisconnectedOp,
 	}
 
@@ -273,6 +276,8 @@ func (op Op) String() string {
 		// Internal
 	case ConnectedOp:
 		return "connected"
+	case ConnectedSubnetOp:
+		return "connected_subnet"
 	case DisconnectedOp:
 		return "disconnected"
 	case NotifyOp:

message/outbound_msg_builder.go

@@ -35,7 +35,7 @@ type OutboundMsgBuilder interface {
 
 	Ping() (OutboundMessage, error)
 
-	Pong(uptimePercentage uint8) (OutboundMessage, error)
+	Pong(uptimePercentage uint32) (OutboundMessage, error)
 
 	GetStateSummaryFrontier(
 		chainID ids.ID,
@@ -179,12 +179,12 @@ func (b *outMsgBuilder) Ping() (OutboundMessage, error) {
 	)
 }
 
-func (b *outMsgBuilder) Pong(uptimePercentage uint8) (OutboundMessage, error) {
+func (b *outMsgBuilder) Pong(uptimePercentage uint32) (OutboundMessage, error) {
 	return b.builder.createOutbound(
 		&p2ppb.Message{
 			Message: &p2ppb.Message_Pong{
 				Pong: &p2ppb.Pong{
-					UptimePct: uint32(uptimePercentage),
+					UptimePct: uptimePercentage,
 				},
 			},
 		},

network/network.go

@@ -25,7 +25,6 @@ import (
 	"github.com/ava-labs/avalanchego/network/dialer"
 	"github.com/ava-labs/avalanchego/network/peer"
 	"github.com/ava-labs/avalanchego/network/throttling"
-	"github.com/ava-labs/avalanchego/snow/engine/common"
 	"github.com/ava-labs/avalanchego/snow/networking/router"
 	"github.com/ava-labs/avalanchego/snow/networking/sender"
 	"github.com/ava-labs/avalanchego/snow/validators"
@@ -45,9 +44,10 @@ const (
 )
 
 var (
-	_                      sender.ExternalSender = (*network)(nil)
-	_                      Network               = (*network)(nil)
-	errNoPrimaryValidators                       = errors.New("no default subnet validators")
+	_ sender.ExternalSender = (*network)(nil)
+	_ Network               = (*network)(nil)
+
+	errMissingPrimaryValidators = errors.New("missing primary validator set")
 )
 
 // Network defines the functionality of the networking library.
@@ -60,7 +60,6 @@ type Network interface {
 	health.Checker
 
 	peer.Network
-	common.SubnetTracker
 
 	// StartClose this network and all existing connections it has. Calling
 	// StartClose multiple times is handled gracefully.
@@ -164,7 +163,7 @@ func NewNetwork(
 ) (Network, error) {
 	primaryNetworkValidators, ok := config.Validators.Get(constants.PrimaryNetworkID)
 	if !ok {
-		return nil, errNoPrimaryValidators
+		return nil, errMissingPrimaryValidators
 	}
 
 	inboundMsgThrottler, err := throttling.NewInboundMsgThrottler(
@@ -527,12 +526,16 @@ func (n *network) Peers(peerID ids.NodeID) ([]ids.NodeID, []ips.ClaimedIPPort, e
 }
 
 func (n *network) Pong(nodeID ids.NodeID) (message.OutboundMessage, error) {
-	uptimePercentFloat, err := n.config.UptimeCalculator.CalculateUptimePercent(nodeID)
+	// TODO: expand this message for tracking subnet uptimes.
+	uptimePercentFloat, err := n.config.UptimeCalculator.CalculateUptimePercent(
+		nodeID,
+		constants.PrimaryNetworkID,
+	)
 	if err != nil {
 		uptimePercentFloat = 0
 	}
 
-	uptimePercentInt := uint8(uptimePercentFloat * 100)
+	uptimePercentInt := uint32(uptimePercentFloat * 100)
 	return n.peerConfig.MessageCreator.Pong(uptimePercentInt)
 }
 
@@ -641,22 +644,6 @@ func (n *network) ManuallyTrack(nodeID ids.NodeID, ip ips.IPPort) {
 	}
 }
 
-func (n *network) TracksSubnet(nodeID ids.NodeID, subnetID ids.ID) bool {
-	if n.config.MyNodeID == nodeID {
-		return subnetID == constants.PrimaryNetworkID || n.config.WhitelistedSubnets.Contains(subnetID)
-	}
-
-	n.peersLock.RLock()
-	defer n.peersLock.RUnlock()
-
-	peer, connected := n.connectedPeers.GetByID(nodeID)
-	if !connected {
-		return false
-	}
-	trackedSubnets := peer.TrackedSubnets()
-	return subnetID == constants.PrimaryNetworkID || trackedSubnets.Contains(subnetID)
-}
-
 // getPeers returns a slice of connected peers from a set of [nodeIDs].
 //
 // - [nodeIDs] the IDs of the peers that should be returned if they are

network/peer/message_queue_test.go

@@ -31,7 +31,7 @@ func TestMessageQueue(t *testing.T) {
 
 	// Assert that the messages are popped in the same order they were pushed
 	for i := 0; i < numToSend; i++ {
-		m, err := mc.Pong(uint8(i))
+		m, err := mc.Pong(uint32(i))
 		require.NoError(err)
 		msgs = append(msgs, m)
 	}

network/peer/test_network.go

@@ -25,7 +25,7 @@ type testNetwork struct {
 	signer    crypto.Signer
 	subnets   ids.Set
 
-	uptime uint8
+	uptime uint32
 }
 
 // NewTestNetwork creates and returns a new TestNetwork
@@ -36,7 +36,7 @@ func NewTestNetwork(
 	version *version.Application,
 	signer crypto.Signer,
 	subnets ids.Set,
-	uptime uint8,
+	uptime uint32,
 ) Network {
 	return &testNetwork{
 		mc:        mc,

node/node.go

@@ -741,7 +741,6 @@ func (n *Node) initVMs() error {
 			Config: config.Config{
 				Chains:                          n.chainManager,
 				Validators:                      vdrs,
-				SubnetTracker:                   n.Net,
 				UptimeLockedCalculator:          n.uptimeCalculator,
 				StakingEnabled:                  n.Config.EnableStaking,
 				WhitelistedSubnets:              n.Config.WhitelistedSubnets,

scripts/mocks.mockgen.txt

@@ -12,13 +12,15 @@ github.com/ava-labs/avalanchego/snow/engine/snowman=Engine=snow/engine/snowman/m
 github.com/ava-labs/avalanchego/snow/engine/snowman/block=ChainVM=snow/engine/snowman/block/mocks/chain_vm.go
 github.com/ava-labs/avalanchego/snow/engine/snowman/block=BuildBlockWithContextChainVM=snow/engine/snowman/block/mocks/build_block_with_context_vm.go
 github.com/ava-labs/avalanchego/snow/engine/snowman/block=StateSyncableVM=snow/engine/snowman/block/mocks/state_syncable_vm.go
+github.com/ava-labs/avalanchego/snow/networking/handler=Handler=snow/networking/handler/mock_handler.go
 github.com/ava-labs/avalanchego/snow/networking/timeout=Manager=snow/networking/timeout/mock_manager.go
 github.com/ava-labs/avalanchego/snow/networking/router=Router=snow/networking/router/mock_router.go
 github.com/ava-labs/avalanchego/snow/networking/sender=ExternalSender=snow/networking/sender/mock_external_sender.go
 github.com/ava-labs/avalanchego/snow/networking/tracker=Targeter=snow/networking/tracker/mock_targeter.go
 github.com/ava-labs/avalanchego/snow/networking/tracker=Tracker=snow/networking/tracker/mock_resource_tracker.go
 github.com/ava-labs/avalanchego/snow/uptime=Calculator=snow/uptime/mock_calculator.go
 github.com/ava-labs/avalanchego/snow/validators=State=snow/validators/mock_state.go
+github.com/ava-labs/avalanchego/snow/validators=SubnetConnector=snow/validators/mock_subnet_connector.go
 github.com/ava-labs/avalanchego/utils/filesystem=Reader=utils/filesystem/mock_io.go
 github.com/ava-labs/avalanchego/utils/hashing=Hasher=utils/hashing/mock_hasher.go
 github.com/ava-labs/avalanchego/utils/logging=Logger=utils/logging/mock_logger.go

snow/networking/handler/handler.go

@@ -104,6 +104,8 @@ type handler struct {
 	numDispatchersClosed int
 	// Closed when this handler and [engine] are done shutting down
 	closed chan struct{}
+
+	subnetConnector validators.SubnetConnector
 }
 
 // Initialize this consensus handler
@@ -115,6 +117,7 @@ func New(
 	preemptTimeouts chan struct{},
 	gossipFrequency time.Duration,
 	resourceTracker tracker.ResourceTracker,
+	subnetConnector validators.SubnetConnector,
 ) (Handler, error) {
 	h := &handler{
 		ctx:              ctx,
@@ -127,6 +130,7 @@ func New(
 		closingChan:      make(chan struct{}),
 		closed:           make(chan struct{}),
 		resourceTracker:  resourceTracker,
+		subnetConnector:  subnetConnector,
 	}
 
 	var err error
@@ -641,6 +645,9 @@ func (h *handler) handleSyncMsg(ctx context.Context, msg message.InboundMessage)
 	case *message.Connected:
 		return engine.Connected(ctx, nodeID, msg.NodeVersion)
 
+	case *message.ConnectedSubnet:
+		return h.subnetConnector.ConnectedSubnet(ctx, nodeID, msg.SubnetID)
+
 	case *message.Disconnected:
 		return engine.Disconnected(ctx, nodeID)