Add p2p.Network component

go.mod

@@ -11,7 +11,7 @@ require (
 	github.com/DataDog/zstd v1.5.2
 	github.com/Microsoft/go-winio v0.5.2
 	github.com/NYTimes/gziphandler v1.1.1
-	github.com/ava-labs/coreth v0.12.9-rc.0
+	github.com/ava-labs/coreth v0.12.8-rc.1.0.20231115212839-97af976b2d5a
 	github.com/ava-labs/ledger-avalanche/go v0.0.0-20231102202641-ae2ebdaeac34
 	github.com/btcsuite/btcd/btcutil v1.1.3
 	github.com/cockroachdb/pebble v0.0.0-20230209160836-829675f94811

go.sum

@@ -66,8 +66,8 @@ github.com/allegro/bigcache v1.2.1-0.20190218064605-e24eb225f156 h1:eMwmnE/GDgah
 github.com/allegro/bigcache v1.2.1-0.20190218064605-e24eb225f156/go.mod h1:Cb/ax3seSYIx7SuZdm2G2xzfwmv3TPSk2ucNfQESPXM=
 github.com/antihax/optional v1.0.0/go.mod h1:uupD/76wgC+ih3iEmQUL+0Ugr19nfwCT1kdvxnR2qWY=
 github.com/armon/consul-api v0.0.0-20180202201655-eb2c6b5be1b6/go.mod h1:grANhF5doyWs3UAsr3K4I6qtAmlQcZDesFNEHPZAzj8=
-github.com/ava-labs/coreth v0.12.9-rc.0 h1:Xvk/iJTY2MSBkkiOs9Eo92nxd67VXzRjaC/WmQXRIb0=
-github.com/ava-labs/coreth v0.12.9-rc.0/go.mod h1:rECKQfGFDeodrwGPlJSvFUJDbVr30jSMIVjQLi6pNX4=
+github.com/ava-labs/coreth v0.12.8-rc.1.0.20231115212839-97af976b2d5a h1:A4RdXCgsooZWqEDA430V7bLPTPdIoiGfd4djFUTW4Ek=
+github.com/ava-labs/coreth v0.12.8-rc.1.0.20231115212839-97af976b2d5a/go.mod h1:nYEP2+B0GDW9b8H+cRNs1nGVgB5CdXjP+lJVpQMwi6o=
 github.com/ava-labs/ledger-avalanche/go v0.0.0-20231102202641-ae2ebdaeac34 h1:mg9Uw6oZFJKytJxgxnl3uxZOs/SB8CVHg6Io4Tf99Zc=
 github.com/ava-labs/ledger-avalanche/go v0.0.0-20231102202641-ae2ebdaeac34/go.mod h1:pJxaT9bUgeRNVmNRgtCHb7sFDIRKy7CzTQVi8gGNT6g=
 github.com/aymerick/raymond v2.0.3-0.20180322193309-b565731e1464+incompatible/go.mod h1:osfaiScAUVup+UC9Nfq76eWqDhXlp+4UYaA8uhTBO6g=

network/p2p/client.go

@@ -42,10 +42,9 @@ type CrossChainAppResponseCallback func(
 type Client struct {
 	handlerID     uint64
 	handlerPrefix []byte
-	router        *Router
+	router        *router
 	sender        common.AppSender
-	// nodeSampler is used to select nodes to route AppRequestAny to
-	nodeSampler NodeSampler
+	options       *ClientOptions
 }
 
 // AppRequestAny issues an AppRequest to an arbitrary node decided by Client.
@@ -56,7 +55,7 @@ func (c *Client) AppRequestAny(
 	appRequestBytes []byte,
 	onResponse AppResponseCallback,
 ) error {
-	sampled := c.nodeSampler.Sample(ctx, 1)
+	sampled := c.options.NodeSampler.Sample(ctx, 1)
 	if len(sampled) != 1 {
 		return ErrNoPeers
 	}

network/p2p/gossip/gossip_test.go

@@ -13,8 +13,6 @@ import (
 
 	"github.com/stretchr/testify/require"
 
-	"go.uber.org/mock/gomock"
-
 	"github.com/ava-labs/avalanchego/ids"
 	"github.com/ava-labs/avalanchego/network/p2p"
 	"github.com/ava-labs/avalanchego/snow/engine/common"
@@ -117,10 +115,9 @@ func TestGossiperGossip(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			require := require.New(t)
-			ctrl := gomock.NewController(t)
 
-			responseSender := common.NewMockSender(ctrl)
-			responseRouter := p2p.NewRouter(logging.NoLog{}, responseSender, prometheus.NewRegistry(), "")
+			responseSender := &common.SenderTest{}
+			responseNetwork := p2p.NewNetwork(logging.NoLog{}, responseSender, prometheus.NewRegistry(), "")
 			responseBloom, err := NewBloomFilter(1000, 0.01)
 			require.NoError(err)
 			responseSet := testSet{
@@ -130,31 +127,30 @@ func TestGossiperGossip(t *testing.T) {
 			for _, item := range tt.responder {
 				require.NoError(responseSet.Add(item))
 			}
-			peers := &p2p.Peers{}
-			require.NoError(peers.Connected(context.Background(), ids.EmptyNodeID, nil))
 
 			handler, err := NewHandler[*testTx](responseSet, tt.config, prometheus.NewRegistry())
 			require.NoError(err)
-			_, err = responseRouter.RegisterAppProtocol(0x0, handler, peers)
+			_, err = responseNetwork.RegisterAppProtocol(0x0, handler)
 			require.NoError(err)
 
-			requestSender := common.NewMockSender(ctrl)
-			requestRouter := p2p.NewRouter(logging.NoLog{}, requestSender, prometheus.NewRegistry(), "")
-
-			gossiped := make(chan struct{})
-			requestSender.EXPECT().SendAppRequest(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
-				Do(func(ctx context.Context, nodeIDs set.Set[ids.NodeID], requestID uint32, request []byte) {
+			requestSender := &common.SenderTest{
+				SendAppRequestF: func(ctx context.Context, nodeIDs set.Set[ids.NodeID], requestID uint32, request []byte) error {
 					go func() {
-						require.NoError(responseRouter.AppRequest(ctx, ids.EmptyNodeID, requestID, time.Time{}, request))
+						require.NoError(responseNetwork.AppRequest(ctx, ids.EmptyNodeID, requestID, time.Time{}, request))
 					}()
-				}).AnyTimes()
+					return nil
+				},
+			}
 
-			responseSender.EXPECT().
-				SendAppResponse(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
-				Do(func(ctx context.Context, nodeID ids.NodeID, requestID uint32, appResponseBytes []byte) {
-					require.NoError(requestRouter.AppResponse(ctx, nodeID, requestID, appResponseBytes))
-					close(gossiped)
-				}).AnyTimes()
+			requestNetwork := p2p.NewNetwork(logging.NoLog{}, requestSender, prometheus.NewRegistry(), "")
+			require.NoError(requestNetwork.Connected(context.Background(), ids.EmptyNodeID, nil))
+
+			gossiped := make(chan struct{})
+			responseSender.SendAppResponseF = func(ctx context.Context, nodeID ids.NodeID, requestID uint32, appResponseBytes []byte) error {
+				require.NoError(requestNetwork.AppResponse(ctx, nodeID, requestID, appResponseBytes))
+				close(gossiped)
+				return nil
+			}
 
 			bloom, err := NewBloomFilter(1000, 0.01)
 			require.NoError(err)
@@ -166,7 +162,7 @@ func TestGossiperGossip(t *testing.T) {
 				require.NoError(requestSet.Add(item))
 			}
 
-			requestClient, err := requestRouter.RegisterAppProtocol(0x0, nil, peers)
+			requestClient, err := requestNetwork.RegisterAppProtocol(0x0, nil)
 			require.NoError(err)
 
 			config := Config{

network/p2p/network.go

@@ -0,0 +1,241 @@
+// Copyright (C) 2019-2023, Ava Labs, Inc. All rights reserved.
+// See the file LICENSE for licensing terms.
+
+package p2p
+
+import (
+	"context"
+	"encoding/binary"
+	"fmt"
+	"sync"
+
+	"github.com/prometheus/client_golang/prometheus"
+
+	"github.com/ava-labs/avalanchego/ids"
+	"github.com/ava-labs/avalanchego/snow/engine/common"
+	snowvalidators "github.com/ava-labs/avalanchego/snow/validators"
+	"github.com/ava-labs/avalanchego/utils/logging"
+	"github.com/ava-labs/avalanchego/utils/metric"
+	"github.com/ava-labs/avalanchego/utils/set"
+	"github.com/ava-labs/avalanchego/version"
+)
+
+var (
+	_ snowvalidators.Connector = (*Network)(nil)
+	_ common.AppHandler        = (*Network)(nil)
+	_ NodeSampler              = (*peers)(nil)
+)
+
+// ClientOption configures Client
+type ClientOption interface {
+	apply(options *ClientOptions)
+}
+
+type clientOptionFunc func(options *ClientOptions)
+
+func (o clientOptionFunc) apply(options *ClientOptions) {
+	o(options)
+}
+
+// WithPeerSampling configures Client.AppRequestAny to sample peers
+func WithPeerSampling(network *Network) ClientOption {
+	return clientOptionFunc(func(options *ClientOptions) {
+		options.NodeSampler = network.peers
+	})
+}
+
+// WithValidatorSampling configures Client.AppRequestAny to sample validators
+func WithValidatorSampling(validators *Validators) ClientOption {
+	return clientOptionFunc(func(options *ClientOptions) {
+		options.NodeSampler = validators
+	})
+}
+
+// ClientOptions holds client-configurable values
+type ClientOptions struct {
+	// NodeSampler is used to select nodes to route Client.AppRequestAny to
+	NodeSampler NodeSampler
+}
+
+// NewNetwork returns an instance of Network
+func NewNetwork(
+	log logging.Logger,
+	sender common.AppSender,
+	metrics prometheus.Registerer,
+	namespace string,
+) *Network {
+	return &Network{
+		log:       log,
+		sender:    sender,
+		metrics:   metrics,
+		namespace: namespace,
+		router:    newRouter(log),
+		peers:     &peers{},
+	}
+}
+
+// Network maintains state of the peer-to-peer network and any in-flight
+// requests
+type Network struct {
+	log       logging.Logger
+	sender    common.AppSender
+	metrics   prometheus.Registerer
+	namespace string
+
+	*router
+	peers *peers
+}
+
+func (n *Network) Connected(_ context.Context, nodeID ids.NodeID, _ *version.Application) error {
+	n.lock.Lock()
+	defer n.lock.Unlock()
+
+	n.peers.set.Add(nodeID)
+	return nil
+}
+
+func (n *Network) Disconnected(_ context.Context, nodeID ids.NodeID) error {
+	n.lock.Lock()
+	defer n.lock.Unlock()
+
+	n.peers.set.Remove(nodeID)
+	return nil
+}
+
+// RegisterAppProtocol reserves an identifier for an application protocol and
+// returns a Client that can be used to send messages for the corresponding
+// protocol.
+func (n *Network) RegisterAppProtocol(handlerID uint64, handler Handler, options ...ClientOption) (*Client, error) {
+	// TODO refactor router
+	n.router.lock.Lock()
+	defer n.router.lock.Unlock()
+
+	if _, ok := n.router.handlers[handlerID]; ok {
+		return nil, fmt.Errorf("failed to register handler id %d: %w", handlerID, ErrExistingAppProtocol)
+	}
+
+	appRequestTime, err := metric.NewAverager(
+		n.namespace,
+		fmt.Sprintf("handler_%d_app_request", handlerID),
+		"app request time (ns)",
+		n.metrics,
+	)
+	if err != nil {
+		return nil, fmt.Errorf("failed to register app request metric for handler_%d: %w", handlerID, err)
+	}
+
+	appRequestFailedTime, err := metric.NewAverager(
+		n.namespace,
+		fmt.Sprintf("handler_%d_app_request_failed", handlerID),
+		"app request failed time (ns)",
+		n.metrics,
+	)
+	if err != nil {
+		return nil, fmt.Errorf("failed to register app request failed metric for handler_%d: %w", handlerID, err)
+	}
+
+	appResponseTime, err := metric.NewAverager(
+		n.namespace,
+		fmt.Sprintf("handler_%d_app_response", handlerID),
+		"app response time (ns)",
+		n.metrics,
+	)
+	if err != nil {
+		return nil, fmt.Errorf("failed to register app response metric for handler_%d: %w", handlerID, err)
+	}
+
+	appGossipTime, err := metric.NewAverager(
+		n.namespace,
+		fmt.Sprintf("handler_%d_app_gossip", handlerID),
+		"app gossip time (ns)",
+		n.metrics,
+	)
+	if err != nil {
+		return nil, fmt.Errorf("failed to register app gossip metric for handler_%d: %w", handlerID, err)
+	}
+
+	crossChainAppRequestTime, err := metric.NewAverager(
+		n.namespace,
+		fmt.Sprintf("handler_%d_cross_chain_app_request", handlerID),
+		"cross chain app request time (ns)",
+		n.metrics,
+	)
+	if err != nil {
+		return nil, fmt.Errorf("failed to register cross-chain app request metric for handler_%d: %w", handlerID, err)
+	}
+
+	crossChainAppRequestFailedTime, err := metric.NewAverager(
+		n.namespace,
+		fmt.Sprintf("handler_%d_cross_chain_app_request_failed", handlerID),
+		"app request failed time (ns)",
+		n.metrics,
+	)
+	if err != nil {
+		return nil, fmt.Errorf("failed to register cross-chain app request failed metric for handler_%d: %w", handlerID, err)
+	}
+
+	crossChainAppResponseTime, err := metric.NewAverager(
+		n.namespace,
+		fmt.Sprintf("handler_%d_cross_chain_app_response", handlerID),
+		"cross chain app response time (ns)",
+		n.metrics,
+	)
+	if err != nil {
+		return nil, fmt.Errorf("failed to register cross-chain app response metric for handler_%d: %w", handlerID, err)
+	}
+
+	n.router.handlers[handlerID] = &meteredHandler{
+		responder: &responder{
+			Handler:   handler,
+			handlerID: handlerID,
+			log:       n.log,
+			sender:    n.sender,
+		},
+		metrics: &metrics{
+			appRequestTime:                 appRequestTime,
+			appRequestFailedTime:           appRequestFailedTime,
+			appResponseTime:                appResponseTime,
+			appGossipTime:                  appGossipTime,
+			crossChainAppRequestTime:       crossChainAppRequestTime,
+			crossChainAppRequestFailedTime: crossChainAppRequestFailedTime,
+			crossChainAppResponseTime:      crossChainAppResponseTime,
+		},
+	}
+
+	clientOptions := &ClientOptions{
+		NodeSampler: n.peers,
+	}
+
+	for _, option := range options {
+		option.apply(clientOptions)
+	}
+
+	return &Client{
+		handlerID:     handlerID,
+		handlerPrefix: binary.AppendUvarint(nil, handlerID),
+		sender:        n.sender,
+		router:        n.router,
+		options:       clientOptions,
+	}, nil
+}
+
+// peers contains the set of nodes we are connected to
+type peers struct {
+	lock sync.RWMutex
+	set  set.SampleableSet[ids.NodeID]
+}
+
+func (p *peers) has(nodeID ids.NodeID) bool {
+	p.lock.RLock()
+	defer p.lock.RUnlock()
+
+	return p.set.Contains(nodeID)
+}
+
+// Sample returns a pseudo-random sample of up to limit peers
+func (p *peers) Sample(_ context.Context, limit int) []ids.NodeID {
+	p.lock.RLock()
+	defer p.lock.RUnlock()
+
+	return p.set.Sample(limit)
+}