Merge pull request #5759 from ethereum-optimism/aj/integrate-extended-peerstore

op-node: Add gossip scores to peer store and opp2p_peers output

op-node/p2p/host.go

@@ -7,6 +7,9 @@ import (
 	"sync"
 	"time"
 
+	"github.com/ethereum-optimism/optimism/op-service/clock"
+
+	"github.com/ethereum-optimism/optimism/op-node/p2p/store"
 	libp2p "github.com/libp2p/go-libp2p"
 	lconf "github.com/libp2p/go-libp2p/config"
 	"github.com/libp2p/go-libp2p/core/connmgr"
@@ -132,11 +135,16 @@ func (conf *Config) Host(log log.Logger, reporter metrics.Reporter) (host.Host,
 		return nil, fmt.Errorf("failed to derive pubkey from network priv key: %w", err)
 	}
 
-	ps, err := pstoreds.NewPeerstore(context.Background(), conf.Store, pstoreds.DefaultOpts())
+	basePs, err := pstoreds.NewPeerstore(context.Background(), conf.Store, pstoreds.DefaultOpts())
 	if err != nil {
 		return nil, fmt.Errorf("failed to open peerstore: %w", err)
 	}
 
+	ps, err := store.NewExtendedPeerstore(context.Background(), log, clock.SystemClock, basePs, conf.Store)
+	if err != nil {
+		return nil, fmt.Errorf("failed to open extended peerstore: %w", err)
+	}
+
 	if err := ps.AddPrivKey(pid, conf.Priv); err != nil {
 		return nil, fmt.Errorf("failed to set up peerstore with priv key: %w", err)
 	}

op-node/p2p/mocks/ConnectionGater.go

@@ -142,20 +142,6 @@ func (_m *ConnectionGater) InterceptUpgraded(_a0 network.Conn) (bool, control.Di
 	return r0, r1
 }
 
-// IsBlocked provides a mock function with given fields: p
-func (_m *ConnectionGater) IsBlocked(p peer.ID) bool {
-	ret := _m.Called(p)
-
-	var r0 bool
-	if rf, ok := ret.Get(0).(func(peer.ID) bool); ok {
-		r0 = rf(p)
-	} else {
-		r0 = ret.Get(0).(bool)
-	}
-
-	return r0
-}
-
 // ListBlockedAddrs provides a mock function with given fields:
 func (_m *ConnectionGater) ListBlockedAddrs() []net.IP {
 	ret := _m.Called()

op-node/p2p/mocks/Peerstore.go

@@ -6,6 +6,8 @@ import (
 	mock "github.com/stretchr/testify/mock"
 
 	peer "github.com/libp2p/go-libp2p/core/peer"
+
+	store "github.com/ethereum-optimism/optimism/op-node/p2p/store"
 )
 
 // Peerstore is an autogenerated mock type for the Peerstore type
@@ -43,6 +45,20 @@ func (_m *Peerstore) Peers() peer.IDSlice {
 	return r0
 }
 
+// SetScore provides a mock function with given fields: _a0, _a1, _a2
+func (_m *Peerstore) SetScore(_a0 peer.ID, _a1 store.ScoreType, _a2 float64) error {
+	ret := _m.Called(_a0, _a1, _a2)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(peer.ID, store.ScoreType, float64) error); ok {
+		r0 = rf(_a0, _a1, _a2)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
 type mockConstructorTestingTNewPeerstore interface {
 	mock.TestingT
 	Cleanup(func())

op-node/p2p/peer_scorer.go

@@ -6,6 +6,7 @@ import (
 	"strconv"
 	"strings"
 
+	"github.com/ethereum-optimism/optimism/op-node/p2p/store"
 	log "github.com/ethereum/go-ethereum/log"
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
 	peer "github.com/libp2p/go-libp2p/core/peer"
@@ -91,6 +92,8 @@ type Peerstore interface {
 
 	// Peers returns all of the peer IDs stored across all inner stores.
 	Peers() peer.IDSlice
+
+	SetScore(peer.ID, store.ScoreType, float64) error
 }
 
 // Scorer is a peer scorer that scores peers based on application-specific metrics.
@@ -123,6 +126,12 @@ func (s *scorer) SnapshotHook() pubsub.ExtendedPeerScoreInspectFn {
 		}
 		// Now set the new scores.
 		for id, snap := range m {
+			scores := make(map[store.ScoreType]float64)
+			scores[store.TypeGossip] = snap.Score
+
+			if err := s.peerStore.SetScore(id, store.TypeGossip, snap.Score); err != nil {
+				s.log.Warn("Unable to update peer gossip score", "err", err)
+			}
 			band := s.bandScoreThresholds.Bucket(snap.Score)
 			scoreMap[band] += 1
 			s.gater.Update(id, snap.Score)

op-node/p2p/peer_scorer_test.go

@@ -5,6 +5,7 @@ import (
 
 	p2p "github.com/ethereum-optimism/optimism/op-node/p2p"
 	p2pMocks "github.com/ethereum-optimism/optimism/op-node/p2p/mocks"
+	"github.com/ethereum-optimism/optimism/op-node/p2p/store"
 	"github.com/ethereum-optimism/optimism/op-node/testlog"
 	log "github.com/ethereum/go-ethereum/log"
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
@@ -78,6 +79,9 @@ func (testSuite *PeerScorerTestSuite) TestScorer_SnapshotHook() {
 	// Mock the peer gater call
 	testSuite.mockGater.On("Update", peer.ID("peer1"), float64(-100)).Return(nil).Once()
 
+	// Expect updating the peer store
+	testSuite.mockStore.On("SetScore", peer.ID("peer1"), store.TypeGossip, float64(-100)).Return(nil).Once()
+
 	// The metricer should then be called with the peer score band map
 	testSuite.mockMetricer.On("SetPeerScores", map[string]float64{
 		"friend":   0,
@@ -94,6 +98,8 @@ func (testSuite *PeerScorerTestSuite) TestScorer_SnapshotHook() {
 
 	// Change the peer score now to a different band
 	testSuite.mockGater.On("Update", peer.ID("peer1"), float64(0)).Return(nil).Once()
+	// Expect updating the peer store
+	testSuite.mockStore.On("SetScore", peer.ID("peer1"), store.TypeGossip, float64(0)).Return(nil).Once()
 
 	// The metricer should then be called with the peer score band map
 	testSuite.mockMetricer.On("SetPeerScores", map[string]float64{
@@ -124,6 +130,8 @@ func (testSuite *PeerScorerTestSuite) TestScorer_SnapshotHookBlocksPeer() {
 
 	// Mock the peer gater call
 	testSuite.mockGater.On("Update", peer.ID("peer1"), float64(-101)).Return(nil)
+	// Expect updating the peer store
+	testSuite.mockStore.On("SetScore", peer.ID("peer1"), store.TypeGossip, float64(-101)).Return(nil).Once()
 
 	// The metricer should then be called with the peer score band map
 	testSuite.mockMetricer.On("SetPeerScores", map[string]float64{

op-node/p2p/peer_scores.go

 package p2p
 
 import (
+	"github.com/ethereum-optimism/optimism/op-node/p2p/store"
 	log "github.com/ethereum/go-ethereum/log"
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
 	host "github.com/libp2p/go-libp2p/core/host"
@@ -14,8 +15,13 @@ func ConfigurePeerScoring(h host.Host, g ConnectionGater, gossipConf GossipSetup
 	peerScoreThresholds := NewPeerScoreThresholds()
 	banEnabled := gossipConf.BanPeers()
 	peerGater := NewPeerGater(g, log, banEnabled)
-	scorer := NewScorer(peerGater, h.Peerstore(), m, gossipConf.PeerBandScorer(), log)
 	opts := []pubsub.Option{}
+	eps, ok := h.Peerstore().(store.ExtendedPeerstore)
+	if !ok {
+		log.Warn("Disabling peer scoring. Peerstore does not support peer scores")
+		return opts
+	}
+	scorer := NewScorer(peerGater, eps, m, gossipConf.PeerBandScorer(), log)
 	// Check the app specific score since libp2p doesn't export it's [validate] function :/
 	if peerScoreParams != nil && peerScoreParams.AppSpecificScore != nil {
 		opts = []pubsub.Option{

op-node/p2p/peer_scores_test.go

@@ -7,11 +7,18 @@ import (
 	"testing"
 	"time"
 
+	"github.com/ethereum-optimism/optimism/op-service/clock"
+
 	p2p "github.com/ethereum-optimism/optimism/op-node/p2p"
 	p2pMocks "github.com/ethereum-optimism/optimism/op-node/p2p/mocks"
+	"github.com/ethereum-optimism/optimism/op-node/p2p/store"
 	testlog "github.com/ethereum-optimism/optimism/op-node/testlog"
-
+	ds "github.com/ipfs/go-datastore"
+	"github.com/ipfs/go-datastore/sync"
+	"github.com/libp2p/go-libp2p/core/network"
+	"github.com/libp2p/go-libp2p/core/peerstore"
 	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
 	suite "github.com/stretchr/testify/suite"
 
 	log "github.com/ethereum/go-ethereum/log"
@@ -50,11 +57,29 @@ func TestPeerScores(t *testing.T) {
 	suite.Run(t, new(PeerScoresTestSuite))
 }
 
+type customPeerstoreNetwork struct {
+	network.Network
+	ps peerstore.Peerstore
+}
+
+func (c *customPeerstoreNetwork) Peerstore() peerstore.Peerstore {
+	return c.ps
+}
+
+func (c *customPeerstoreNetwork) Close() error {
+	_ = c.ps.Close()
+	return c.Network.Close()
+}
+
 // getNetHosts generates a slice of hosts using the [libp2p/go-libp2p] library.
 func getNetHosts(testSuite *PeerScoresTestSuite, ctx context.Context, n int) []host.Host {
 	var out []host.Host
+	log := testlog.Logger(testSuite.T(), log.LvlError)
 	for i := 0; i < n; i++ {
-		netw := tswarm.GenSwarm(testSuite.T())
+		swarm := tswarm.GenSwarm(testSuite.T())
+		eps, err := store.NewExtendedPeerstore(ctx, log, clock.SystemClock, swarm.Peerstore(), sync.MutexWrap(ds.NewMapDatastore()))
+		netw := &customPeerstoreNetwork{swarm, eps}
+		require.NoError(testSuite.T(), err)
 		h := bhost.NewBlankHost(netw)
 		testSuite.T().Cleanup(func() { h.Close() })
 		out = append(out, h)

op-node/p2p/rpc_api.go

@@ -5,6 +5,7 @@ import (
 	"net"
 	"time"
 
+	"github.com/ethereum-optimism/optimism/op-node/p2p/store"
 	"github.com/libp2p/go-libp2p/core/network"
 	"github.com/libp2p/go-libp2p/core/peer"
 
@@ -28,6 +29,8 @@ type PeerInfo struct {
 	Latency       time.Duration         `json:"latency"`
 
 	GossipBlocks bool `json:"gossipBlocks"` // if the peer is in our gossip topic
+
+	PeerScores store.PeerScores `json:"scores"`
 }
 
 type PeerDump struct {

op-node/p2p/rpc_server.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	decredSecp "github.com/decred/dcrd/dcrec/secp256k1/v4"
+	"github.com/ethereum-optimism/optimism/op-node/p2p/store"
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
 	"github.com/libp2p/go-libp2p-testing/netutil"
 	"github.com/libp2p/go-libp2p/core/connmgr"
@@ -108,6 +109,11 @@ func dumpPeer(id peer.ID, nw network.Network, pstore peerstore.Peerstore, connMg
 			info.NodeID = enode.PubkeyToIDV4((*decredSecp.PublicKey)(typedPub).ToECDSA())
 		}
 	}
+	if eps, ok := pstore.(store.ExtendedPeerstore); ok {
+		if dat, err := eps.GetPeerScores(id); err == nil {
+			info.PeerScores = dat
+		}
+	}
 	if dat, err := pstore.Get(id, "ProtocolVersion"); err == nil {
 		protocolVersion, ok := dat.(string)
 		if ok {

op-node/p2p/store/iface.go

@@ -6,7 +6,7 @@ import (
 )
 
 type PeerScores struct {
-	Gossip float64
+	Gossip float64 `json:"gossip"`
 }
 
 type ScoreType int