Merge branch 'master' into haroldsphinx/1152-Update-Benchnet2-automation-and-helmchart

Author: haroldsphinx

cmd/access/node_builder/access_node_builder.go

@@ -1072,7 +1072,7 @@ func (builder *FlowAccessNodeBuilder) enqueuePublicNetworkInit() {
 //   - The passed in private key as the libp2p key
 //   - No connection gater
 //   - Default Flow libp2p pubsub options
-func (builder *FlowAccessNodeBuilder) initLibP2PFactory(networkKey crypto.PrivateKey, bindAddress string, networkMetrics module.LibP2PMetrics) p2pbuilder.LibP2PFactoryFunc {
+func (builder *FlowAccessNodeBuilder) initLibP2PFactory(networkKey crypto.PrivateKey, bindAddress string, networkMetrics module.LibP2PMetrics) p2p.LibP2PFactoryFunc {
 	return func() (p2p.LibP2PNode, error) {
 		connManager, err := connection.NewConnManager(builder.Logger, networkMetrics, builder.ConnectionManagerConfig)
 		if err != nil {
@@ -1113,6 +1113,7 @@ func (builder *FlowAccessNodeBuilder) initLibP2PFactory(networkKey crypto.Privat
 			SetPeerManagerOptions(connection.ConnectionPruningDisabled, builder.PeerUpdateInterval).
 			SetStreamCreationRetryInterval(builder.UnicastCreateStreamRetryDelay).
 			SetGossipSubTracer(meshTracer).
+			SetGossipSubScoreTracerInterval(builder.GossipSubConfig.ScoreTracerInterval).
 			Build()
 
 		if err != nil {

cmd/observer/node_builder/observer_builder.go

@@ -846,7 +846,7 @@ func (builder *ObserverServiceBuilder) validateParams() error {
 // * No connection manager
 // * No peer manager
 // * Default libp2p pubsub options
-func (builder *ObserverServiceBuilder) initLibP2PFactory(networkKey crypto.PrivateKey) p2pbuilder.LibP2PFactoryFunc {
+func (builder *ObserverServiceBuilder) initLibP2PFactory(networkKey crypto.PrivateKey) p2p.LibP2PFactoryFunc {
 	return func() (p2p.LibP2PNode, error) {
 		var pis []peer.AddrInfo
 
@@ -888,6 +888,7 @@ func (builder *ObserverServiceBuilder) initLibP2PFactory(networkKey crypto.Priva
 			}).
 			SetStreamCreationRetryInterval(builder.UnicastCreateStreamRetryDelay).
 			SetGossipSubTracer(meshTracer).
+			SetGossipSubScoreTracerInterval(builder.GossipSubConfig.ScoreTracerInterval).
 			Build()
 
 		if err != nil {

cmd/scaffold.go

@@ -185,6 +185,7 @@ func (fnb *FlowNodeBuilder) BaseFlags() {
 	fnb.flags.BoolVar(&fnb.BaseConfig.NetworkConnectionPruning, "networking-connection-pruning", defaultConfig.NetworkConnectionPruning, "enabling connection trimming")
 	fnb.flags.BoolVar(&fnb.BaseConfig.GossipSubConfig.PeerScoring, "peer-scoring-enabled", defaultConfig.GossipSubConfig.PeerScoring, "enabling peer scoring on pubsub network")
 	fnb.flags.DurationVar(&fnb.BaseConfig.GossipSubConfig.LocalMeshLogInterval, "gossipsub-local-mesh-logging-interval", defaultConfig.GossipSubConfig.LocalMeshLogInterval, "logging interval for local mesh in gossipsub")
+	fnb.flags.DurationVar(&fnb.BaseConfig.GossipSubConfig.ScoreTracerInterval, "gossipsub-score-tracer-interval", defaultConfig.GossipSubConfig.ScoreTracerInterval, "logging interval for peer score tracer in gossipsub, set to 0 to disable")
 	fnb.flags.UintVar(&fnb.BaseConfig.guaranteesCacheSize, "guarantees-cache-size", bstorage.DefaultCacheSize, "collection guarantees cache size")
 	fnb.flags.UintVar(&fnb.BaseConfig.receiptsCacheSize, "receipts-cache-size", bstorage.DefaultCacheSize, "receipts cache size")
 

engine/execution/state/state.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"sync"
 
 	"github.com/dgraph-io/badger/v2"
 
@@ -148,7 +149,8 @@ type LedgerStorageSnapshot struct {
 	ledger     ledger.Ledger
 	commitment flow.StateCommitment
 
-	readCache map[flow.RegisterID]flow.RegisterValue
+	mutex     sync.RWMutex
+	readCache map[flow.RegisterID]flow.RegisterValue // Guarded by mutex.
 }
 
 func NewLedgerStorageSnapshot(
@@ -162,16 +164,25 @@ func NewLedgerStorageSnapshot(
 	}
 }
 
-func (storage *LedgerStorageSnapshot) Get(
+func (storage *LedgerStorageSnapshot) getFromCache(
 	id flow.RegisterID,
 ) (
 	flow.RegisterValue,
-	error,
+	bool,
 ) {
-	if value, ok := storage.readCache[id]; ok {
-		return value, nil
-	}
+	storage.mutex.RLock()
+	defer storage.mutex.RUnlock()
 
+	value, ok := storage.readCache[id]
+	return value, ok
+}
+
+func (storage *LedgerStorageSnapshot) getFromLedger(
+	id flow.RegisterID,
+) (
+	flow.RegisterValue,
+	error,
+) {
 	query, err := makeSingleValueQuery(storage.commitment, id)
 	if err != nil {
 		return nil, fmt.Errorf("cannot create ledger query: %w", err)
@@ -186,14 +197,29 @@ func (storage *LedgerStorageSnapshot) Get(
 			err)
 	}
 
-	// Prevent caching of value with len zero
-	if len(value) == 0 {
-		return nil, nil
+	return value, nil
+}
+
+func (storage *LedgerStorageSnapshot) Get(
+	id flow.RegisterID,
+) (
+	flow.RegisterValue,
+	error,
+) {
+	value, ok := storage.getFromCache(id)
+	if ok {
+		return value, nil
+	}
+
+	value, err := storage.getFromLedger(id)
+	if err != nil {
+		return nil, err
 	}
 
-	// don't cache value with len zero
-	storage.readCache[id] = value
+	storage.mutex.Lock()
+	defer storage.mutex.Unlock()
 
+	storage.readCache[id] = value
 	return value, nil
 }
 

follower/follower_builder.go

@@ -576,7 +576,7 @@ func (builder *FollowerServiceBuilder) validateParams() error {
 //   - No connection manager
 //   - No peer manager
 //   - Default libp2p pubsub options
-func (builder *FollowerServiceBuilder) initLibP2PFactory(networkKey crypto.PrivateKey) p2pbuilder.LibP2PFactoryFunc {
+func (builder *FollowerServiceBuilder) initLibP2PFactory(networkKey crypto.PrivateKey) p2p.LibP2PFactoryFunc {
 	return func() (p2p.LibP2PNode, error) {
 		var pis []peer.AddrInfo
 
@@ -618,6 +618,7 @@ func (builder *FollowerServiceBuilder) initLibP2PFactory(networkKey crypto.Priva
 			}).
 			SetStreamCreationRetryInterval(builder.UnicastCreateStreamRetryDelay).
 			SetGossipSubTracer(meshTracer).
+			SetGossipSubScoreTracerInterval(builder.GossipSubConfig.ScoreTracerInterval).
 			Build()
 
 		if err != nil {

insecure/corruptlibp2p/libp2p_node_factory.go

@@ -35,7 +35,7 @@ func NewCorruptLibP2PNodeFactory(
 	topicValidatorDisabled,
 	withMessageSigning,
 	withStrictSignatureVerification bool,
-) p2pbuilder.LibP2PFactoryFunc {
+) p2p.LibP2PFactoryFunc {
 	return func() (p2p.LibP2PNode, error) {
 		if chainID != flow.BftTestnet {
 			panic("illegal chain id for using corrupt libp2p node")
@@ -70,7 +70,7 @@ func NewCorruptLibP2PNodeFactory(
 
 // CorruptGossipSubFactory returns a factory function that creates a new instance of the forked gossipsub module from
 // github.com/yhassanzadeh13/go-libp2p-pubsub for the purpose of BFT testing and attack vector implementation.
-func CorruptGossipSubFactory(routerOpts ...func(*corrupt.GossipSubRouter)) p2pbuilder.GossipSubFactoryFunc {
+func CorruptGossipSubFactory(routerOpts ...func(*corrupt.GossipSubRouter)) p2p.GossipSubFactoryFunc {
 	factory := func(ctx context.Context, logger zerolog.Logger, host host.Host, cfg p2p.PubSubAdapterConfig) (p2p.PubSubAdapter, error) {
 		adapter, router, err := NewCorruptGossipSubAdapter(ctx, logger, host, cfg)
 		for _, opt := range routerOpts {
@@ -83,21 +83,21 @@ func CorruptGossipSubFactory(routerOpts ...func(*corrupt.GossipSubRouter)) p2pbu
 
 // CorruptGossipSubConfigFactory returns a factory function that creates a new instance of the forked gossipsub config
 // from github.com/yhassanzadeh13/go-libp2p-pubsub for the purpose of BFT testing and attack vector implementation.
-func CorruptGossipSubConfigFactory(opts ...CorruptPubSubAdapterConfigOption) p2pbuilder.GossipSubAdapterConfigFunc {
+func CorruptGossipSubConfigFactory(opts ...CorruptPubSubAdapterConfigOption) p2p.GossipSubAdapterConfigFunc {
 	return func(base *p2p.BasePubSubAdapterConfig) p2p.PubSubAdapterConfig {
 		return NewCorruptPubSubAdapterConfig(base, opts...)
 	}
 }
 
 // CorruptGossipSubConfigFactoryWithInspector returns a factory function that creates a new instance of the forked gossipsub config
 // from github.com/yhassanzadeh13/go-libp2p-pubsub for the purpose of BFT testing and attack vector implementation.
-func CorruptGossipSubConfigFactoryWithInspector(inspector func(peer.ID, *corrupt.RPC) error) p2pbuilder.GossipSubAdapterConfigFunc {
+func CorruptGossipSubConfigFactoryWithInspector(inspector func(peer.ID, *corrupt.RPC) error) p2p.GossipSubAdapterConfigFunc {
 	return func(base *p2p.BasePubSubAdapterConfig) p2p.PubSubAdapterConfig {
 		return NewCorruptPubSubAdapterConfig(base, WithInspector(inspector))
 	}
 }
 
-func overrideWithCorruptGossipSub(builder p2pbuilder.NodeBuilder, opts ...CorruptPubSubAdapterConfigOption) {
+func overrideWithCorruptGossipSub(builder p2p.NodeBuilder, opts ...CorruptPubSubAdapterConfigOption) {
 	factory := CorruptGossipSubFactory()
 	builder.SetGossipSubFactory(factory, CorruptGossipSubConfigFactory(opts...))
 }

insecure/corruptlibp2p/p2p_node.go

@@ -13,7 +13,6 @@ import (
 	"github.com/onflow/flow-go/network/codec/cbor"
 	"github.com/onflow/flow-go/network/message"
 	"github.com/onflow/flow-go/network/p2p"
-	"github.com/onflow/flow-go/network/p2p/connection"
 	"github.com/onflow/flow-go/network/p2p/p2pnode"
 	validator "github.com/onflow/flow-go/network/validator/pubsub"
 )
@@ -52,7 +51,7 @@ func (n *CorruptP2PNode) Subscribe(topic channels.Topic, _ p2p.TopicValidatorFun
 }
 
 // NewCorruptLibP2PNode returns corrupted libP2PNode that will subscribe to topics using the AcceptAllTopicValidator.
-func NewCorruptLibP2PNode(logger zerolog.Logger, host host.Host, pCache *p2pnode.ProtocolPeerCache, peerManager *connection.PeerManager) p2p.LibP2PNode {
+func NewCorruptLibP2PNode(logger zerolog.Logger, host host.Host, pCache p2p.ProtocolPeerCache, peerManager p2p.PeerManager) p2p.LibP2PNode {
 	node := p2pnode.NewNode(logger, host, pCache, peerManager)
 	return &CorruptP2PNode{Node: node, logger: logger, codec: cbor.NewCodec()}
 }

insecure/corruptlibp2p/pubsub_adapter.go

@@ -11,6 +11,8 @@ import (
 	corrupt "github.com/yhassanzadeh13/go-libp2p-pubsub"
 
 	"github.com/onflow/flow-go/insecure/internal"
+	"github.com/onflow/flow-go/module/component"
+	"github.com/onflow/flow-go/module/irrecoverable"
 	"github.com/onflow/flow-go/network/p2p"
 	"github.com/onflow/flow-go/utils/logging"
 )
@@ -25,6 +27,7 @@ import (
 // implementation, it is designed to be completely isolated in the "insecure" package, and
 // totally separated from the rest of the codebase.
 type CorruptGossipSubAdapter struct {
+	component.Component
 	gossipSub *corrupt.PubSub
 	router    *corrupt.GossipSubRouter
 	logger    zerolog.Logger
@@ -116,7 +119,14 @@ func NewCorruptGossipSubAdapter(ctx context.Context, logger zerolog.Logger, h ho
 		return nil, nil, fmt.Errorf("failed to create corrupt gossipsub: %w", err)
 	}
 
+	builder := component.NewComponentManagerBuilder().
+		AddWorker(func(ctx irrecoverable.SignalerContext, ready component.ReadyFunc) {
+			ready()
+			<-ctx.Done()
+		}).Build()
+
 	adapter := &CorruptGossipSubAdapter{
+		Component: builder,
 		gossipSub: gossipSub,
 		router:    router,
 		logger:    logger,

insecure/corruptlibp2p/pubsub_adapter_config.go

@@ -97,6 +97,9 @@ func (c *CorruptPubSubAdapterConfig) WithMessageIdFunction(f func([]byte) string
 		return f(pmsg.Data)
 	}))
 }
+func (c *CorruptPubSubAdapterConfig) WithScoreTracer(_ p2p.PeerScoreTracer) {
+	// CorruptPubSub does not support score tracer. This is a no-op.
+}
 
 func (c *CorruptPubSubAdapterConfig) Build() []corrupt.Option {
 	return c.options

insecure/internal/factory.go

@@ -1,11 +1,11 @@
 package internal
 
 import (
-	"github.com/onflow/flow-go/network/p2p/p2pbuilder"
+	"github.com/onflow/flow-go/network/p2p"
 	p2ptest "github.com/onflow/flow-go/network/p2p/test"
 )
 
-func WithCorruptGossipSub(factory p2pbuilder.GossipSubFactoryFunc, config p2pbuilder.GossipSubAdapterConfigFunc) p2ptest.NodeFixtureParameterOption {
+func WithCorruptGossipSub(factory p2p.GossipSubFactoryFunc, config p2p.GossipSubAdapterConfigFunc) p2ptest.NodeFixtureParameterOption {
 	return func(p *p2ptest.NodeFixtureParameters) {
 		p.GossipSubFactory = factory
 		p.GossipSubConfig = config