Merge branch 'master' into tarak/randomness-part5-ledger

Author: tarakby

.github/workflows/bench.yml

@@ -20,6 +20,11 @@ jobs:
   benchstat:
     name: Performance regression check
     runs-on: ubuntu-latest
+    # Check if the event is not triggered by a fork
+    # peter-evans/find-comment@v1 does not work on forks.
+    # see https://github.com/peter-evans/create-pull-request/blob/main/docs/concepts-guidelines.md#restrictions-on-repository-forks for details.
+    # Ideally we would like to still run the benchmark on forks, but we can't do that with the current setup.
+    if: github.event.pull_request.head.repo.full_name == github.repository
     continue-on-error: true
     steps:
       - name: Set benchmark repetitions

access/validator.go

@@ -5,12 +5,12 @@ import (
 	"fmt"
 
 	"github.com/onflow/flow-go/crypto"
+	"github.com/onflow/flow-go/state"
 
 	"github.com/onflow/cadence/runtime/parser"
 
 	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/state/protocol"
-	"github.com/onflow/flow-go/storage"
 )
 
 type Blocks interface {
@@ -29,7 +29,7 @@ func NewProtocolStateBlocks(state protocol.State) *ProtocolStateBlocks {
 func (b *ProtocolStateBlocks) HeaderByID(id flow.Identifier) (*flow.Header, error) {
 	header, err := b.state.AtBlockID(id).Head()
 	if err != nil {
-		if errors.Is(err, storage.ErrNotFound) {
+		if errors.Is(err, state.ErrUnknownSnapshotReference) {
 			return nil, nil
 		}
 

cmd/access/node_builder/access_node_builder.go

@@ -68,6 +68,7 @@ import (
 	"github.com/onflow/flow-go/network/p2p/dht"
 	"github.com/onflow/flow-go/network/p2p/middleware"
 	"github.com/onflow/flow-go/network/p2p/p2pbuilder"
+	"github.com/onflow/flow-go/network/p2p/p2pbuilder/inspector"
 	"github.com/onflow/flow-go/network/p2p/subscription"
 	"github.com/onflow/flow-go/network/p2p/tracer"
 	"github.com/onflow/flow-go/network/p2p/translator"
@@ -170,7 +171,7 @@ func DefaultAccessNodeConfig() *AccessNodeConfig {
 			BindAddress: cmd.NotSet,
 			Metrics:     metrics.NewNoopCollector(),
 		},
-		executionDataSyncEnabled: false,
+		executionDataSyncEnabled: true,
 		executionDataDir:         filepath.Join(homedir, ".flow", "execution_data"),
 		executionDataStartHeight: 0,
 		executionDataConfig: edrequester.ExecutionDataConfig{
@@ -236,7 +237,15 @@ func (builder *FlowAccessNodeBuilder) buildFollowerState() *FlowAccessNodeBuilde
 			return fmt.Errorf("only implementations of type badger.State are currently supported but read-only state has type %T", node.State)
 		}
 
-		followerState, err := badgerState.NewFollowerState(state, node.Storage.Index, node.Storage.Payloads, node.Tracer, node.ProtocolEvents, blocktimer.DefaultBlockTimer)
+		followerState, err := badgerState.NewFollowerState(
+			node.Logger,
+			node.Tracer,
+			node.ProtocolEvents,
+			state,
+			node.Storage.Index,
+			node.Storage.Payloads,
+			blocktimer.DefaultBlockTimer,
+		)
 		builder.FollowerState = followerState
 
 		return err
@@ -333,7 +342,6 @@ func (builder *FlowAccessNodeBuilder) buildFollowerEngine() *FlowAccessNodeBuild
 			builder.Validator,
 			builder.SyncCore,
 			node.Tracer,
-			modulecompliance.WithSkipNewProposalsThreshold(node.ComplianceConfig.SkipNewProposalsThreshold),
 		)
 		if err != nil {
 			return nil, fmt.Errorf("could not create follower core: %w", err)
@@ -347,6 +355,7 @@ func (builder *FlowAccessNodeBuilder) buildFollowerEngine() *FlowAccessNodeBuild
 			node.Storage.Headers,
 			builder.Finalized,
 			core,
+			followereng.WithComplianceConfigOpt(modulecompliance.WithSkipNewProposalsThreshold(node.ComplianceConfig.SkipNewProposalsThreshold)),
 		)
 		if err != nil {
 			return nil, fmt.Errorf("could not create follower engine: %w", err)
@@ -1020,7 +1029,7 @@ func (builder *FlowAccessNodeBuilder) enqueuePublicNetworkInit() {
 		}).
 		Component("public libp2p node", func(node *cmd.NodeConfig) (module.ReadyDoneAware, error) {
 
-			libP2PFactory := builder.initLibP2PFactory(builder.NodeConfig.NetworkKey, builder.PublicNetworkConfig.BindAddress, builder.PublicNetworkConfig.Metrics)
+			libP2PFactory := builder.initPublicLibP2PFactory(builder.NodeConfig.NetworkKey, builder.PublicNetworkConfig.BindAddress, builder.PublicNetworkConfig.Metrics)
 
 			var err error
 			libp2pNode, err = libP2PFactory()
@@ -1066,15 +1075,15 @@ func (builder *FlowAccessNodeBuilder) enqueuePublicNetworkInit() {
 		})
 }
 
-// initLibP2PFactory creates the LibP2P factory function for the given node ID and network key.
+// initPublicLibP2PFactory creates the LibP2P factory function for the given node ID and network key.
 // The factory function is later passed into the initMiddleware function to eventually instantiate the p2p.LibP2PNode instance
 // The LibP2P host is created with the following options:
 //   - DHT as server
 //   - The address from the node config or the specified bind address as the listen address
 //   - 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) p2p.LibP2PFactoryFunc {
+func (builder *FlowAccessNodeBuilder) initPublicLibP2PFactory(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 {
@@ -1087,6 +1096,16 @@ func (builder *FlowAccessNodeBuilder) initLibP2PFactory(networkKey crypto.Privat
 			builder.IdentityProvider,
 			builder.GossipSubConfig.LocalMeshLogInterval)
 
+		// setup RPC inspectors
+		rpcInspectorBuilder := inspector.NewGossipSubInspectorBuilder(builder.Logger, builder.SporkID, builder.GossipSubRPCInspectorsConfig, builder.GossipSubInspectorNotifDistributor)
+		rpcInspectors, err := rpcInspectorBuilder.
+			SetPublicNetwork(p2p.PublicNetworkEnabled).
+			SetMetrics(builder.Metrics.Network, builder.MetricsRegisterer).
+			SetMetricsEnabled(builder.MetricsEnabled).Build()
+		if err != nil {
+			return nil, fmt.Errorf("failed to create gossipsub rpc inspectors: %w", err)
+		}
+
 		libp2pNode, err := p2pbuilder.NewNodeBuilder(
 			builder.Logger,
 			networkMetrics,
@@ -1116,6 +1135,7 @@ func (builder *FlowAccessNodeBuilder) initLibP2PFactory(networkKey crypto.Privat
 			SetStreamCreationRetryInterval(builder.UnicastCreateStreamRetryDelay).
 			SetGossipSubTracer(meshTracer).
 			SetGossipSubScoreTracerInterval(builder.GossipSubConfig.ScoreTracerInterval).
+			SetGossipSubRPCInspectors(rpcInspectors...).
 			Build()
 
 		if err != nil {

cmd/collection/main.go

@@ -183,7 +183,15 @@ func main() {
 			if !ok {
 				return fmt.Errorf("only implementations of type badger.State are currently supported but read-only state has type %T", node.State)
 			}
-			followerState, err = badgerState.NewFollowerState(state, node.Storage.Index, node.Storage.Payloads, node.Tracer, node.ProtocolEvents, blocktimer.DefaultBlockTimer)
+			followerState, err = badgerState.NewFollowerState(
+				node.Logger,
+				node.Tracer,
+				node.ProtocolEvents,
+				state,
+				node.Storage.Index,
+				node.Storage.Payloads,
+				blocktimer.DefaultBlockTimer,
+			)
 			return err
 		}).
 		Module("transactions mempool", func(node *cmd.NodeConfig) error {
@@ -229,7 +237,7 @@ func main() {
 			return nil
 		}).
 		Component("machine account config validator", func(node *cmd.NodeConfig) (module.ReadyDoneAware, error) {
-			//@TODO use fallback logic for flowClient similar to DKG/QC contract clients
+			// @TODO use fallback logic for flowClient similar to DKG/QC contract clients
 			flowClient, err := common.FlowClient(flowClientConfigs[0])
 			if err != nil {
 				return nil, fmt.Errorf("failed to get flow client connection option for access node (0): %s %w", flowClientConfigs[0].AccessAddress, err)
@@ -317,7 +325,6 @@ func main() {
 				validator,
 				mainChainSyncCore,
 				node.Tracer,
-				modulecompliance.WithSkipNewProposalsThreshold(node.ComplianceConfig.SkipNewProposalsThreshold),
 			)
 			if err != nil {
 				return nil, fmt.Errorf("could not create follower core: %w", err)
@@ -331,6 +338,7 @@ func main() {
 				node.Storage.Headers,
 				finalizedHeader.Get(),
 				core,
+				followereng.WithComplianceConfigOpt(modulecompliance.WithSkipNewProposalsThreshold(node.ComplianceConfig.SkipNewProposalsThreshold)),
 			)
 			if err != nil {
 				return nil, fmt.Errorf("could not create follower engine: %w", err)

cmd/consensus/main.go

@@ -242,7 +242,17 @@ func main() {
 				return err
 			}
 
-			mutableState, err = badgerState.NewFullConsensusState(state, node.Storage.Index, node.Storage.Payloads, node.Tracer, node.ProtocolEvents, blockTimer, receiptValidator, sealValidator)
+			mutableState, err = badgerState.NewFullConsensusState(
+				node.Logger,
+				node.Tracer,
+				node.ProtocolEvents,
+				state,
+				node.Storage.Index,
+				node.Storage.Payloads,
+				blockTimer,
+				receiptValidator,
+				sealValidator,
+			)
 			return err
 		}).
 		Module("random beacon key", func(node *cmd.NodeConfig) error {
@@ -377,7 +387,7 @@ func main() {
 			return nil
 		}).
 		Component("machine account config validator", func(node *cmd.NodeConfig) (module.ReadyDoneAware, error) {
-			//@TODO use fallback logic for flowClient similar to DKG/QC contract clients
+			// @TODO use fallback logic for flowClient similar to DKG/QC contract clients
 			flowClient, err := common.FlowClient(flowClientConfigs[0])
 			if err != nil {
 				return nil, fmt.Errorf("failed to get flow client connection option for access node (0): %s %w", flowClientConfigs[0].AccessAddress, err)

cmd/execution_builder.go

@@ -216,7 +216,15 @@ func (exeNode *ExecutionNode) LoadMutableFollowerState(node *NodeConfig) error {
 		return fmt.Errorf("only implementations of type badger.State are currently supported but read-only state has type %T", node.State)
 	}
 	var err error
-	exeNode.followerState, err = badgerState.NewFollowerState(bState, node.Storage.Index, node.Storage.Payloads, node.Tracer, node.ProtocolEvents, blocktimer.DefaultBlockTimer)
+	exeNode.followerState, err = badgerState.NewFollowerState(
+		node.Logger,
+		node.Tracer,
+		node.ProtocolEvents,
+		bState,
+		node.Storage.Index,
+		node.Storage.Payloads,
+		blocktimer.DefaultBlockTimer,
+	)
 	return err
 }
 
@@ -899,7 +907,6 @@ func (exeNode *ExecutionNode) LoadFollowerEngine(
 		validator,
 		exeNode.syncCore,
 		node.Tracer,
-		compliance.WithSkipNewProposalsThreshold(node.ComplianceConfig.SkipNewProposalsThreshold),
 	)
 	if err != nil {
 		return nil, fmt.Errorf("could not create follower core: %w", err)
@@ -913,6 +920,7 @@ func (exeNode *ExecutionNode) LoadFollowerEngine(
 		node.Storage.Headers,
 		exeNode.finalizedHeader.Get(),
 		core,
+		followereng.WithComplianceConfigOpt(compliance.WithSkipNewProposalsThreshold(node.ComplianceConfig.SkipNewProposalsThreshold)),
 	)
 	if err != nil {
 		return nil, fmt.Errorf("could not create follower engine: %w", err)

cmd/node_builder.go

@@ -6,9 +6,6 @@ import (
 	"path/filepath"
 	"time"
 
-	"github.com/onflow/flow-go/network/p2p/distributor"
-	"github.com/onflow/flow-go/network/p2p/p2pbuilder"
-
 	"github.com/dgraph-io/badger/v2"
 	madns "github.com/multiformats/go-multiaddr-dns"
 	"github.com/prometheus/client_golang/prometheus"
@@ -29,9 +26,11 @@ import (
 	"github.com/onflow/flow-go/network/codec/cbor"
 	"github.com/onflow/flow-go/network/p2p"
 	"github.com/onflow/flow-go/network/p2p/connection"
+	"github.com/onflow/flow-go/network/p2p/distributor"
 	"github.com/onflow/flow-go/network/p2p/dns"
-	"github.com/onflow/flow-go/network/p2p/inspector/validation"
 	"github.com/onflow/flow-go/network/p2p/middleware"
+	"github.com/onflow/flow-go/network/p2p/p2pbuilder"
+	inspectorbuilder "github.com/onflow/flow-go/network/p2p/p2pbuilder/inspector"
 	"github.com/onflow/flow-go/network/p2p/unicast"
 	"github.com/onflow/flow-go/state/protocol"
 	"github.com/onflow/flow-go/state/protocol/events"
@@ -185,7 +184,10 @@ type NetworkConfig struct {
 	// that are not part of protocol state should be trimmed
 	// TODO: solely a fallback mechanism, can be removed upon reliable behavior in production.
 	NetworkConnectionPruning bool
-	GossipSubConfig          *p2pbuilder.GossipSubConfig
+	// GossipSubConfig core gossipsub configuration.
+	GossipSubConfig *p2pbuilder.GossipSubConfig
+	// GossipSubRPCInspectorsConfig configuration for all gossipsub RPC control message inspectors.
+	GossipSubRPCInspectorsConfig *inspectorbuilder.GossipSubRPCInspectorsConfig
 	// PreferredUnicastProtocols list of unicast protocols in preferred order
 	PreferredUnicastProtocols       []string
 	NetworkReceivedMessageCacheSize uint32
@@ -199,11 +201,8 @@ type NetworkConfig struct {
 	UnicastCreateStreamRetryDelay time.Duration
 	// size of the queue for notifications about new peers in the disallow list.
 	DisallowListNotificationCacheSize uint32
-	// size of the queue for notifications about gossipsub RPC inspections.
-	GossipSubRPCInspectorNotificationCacheSize uint32
-	GossipSubRPCInspectorCacheSize             uint32
-	UnicastRateLimitersConfig                  *UnicastRateLimitersConfig
-	GossipSubRPCValidationConfigs              *p2pbuilder.GossipSubRPCValidationConfigs
+	// UnicastRateLimitersConfig configuration for all unicast rate limiters.
+	UnicastRateLimitersConfig *UnicastRateLimitersConfig
 }
 
 // UnicastRateLimitersConfig unicast rate limiter configuration for the message and bandwidth rate limiters.
@@ -301,27 +300,13 @@ func DefaultBaseConfig() *BaseConfig {
 				BandwidthRateLimit:  0,
 				BandwidthBurstLimit: middleware.LargeMsgMaxUnicastMsgSize,
 			},
-			GossipSubRPCValidationConfigs: &p2pbuilder.GossipSubRPCValidationConfigs{
-				NumberOfWorkers: validation.DefaultNumberOfWorkers,
-				GraftLimits: map[string]int{
-					validation.DiscardThresholdMapKey: validation.DefaultGraftDiscardThreshold,
-					validation.SafetyThresholdMapKey:  validation.DefaultGraftSafetyThreshold,
-					validation.RateLimitMapKey:        validation.DefaultGraftRateLimit,
-				},
-				PruneLimits: map[string]int{
-					validation.DiscardThresholdMapKey: validation.DefaultPruneDiscardThreshold,
-					validation.SafetyThresholdMapKey:  validation.DefaultPruneSafetyThreshold,
-					validation.RateLimitMapKey:        validation.DefaultPruneRateLimit,
-				},
-			},
-			DNSCacheTTL:                                dns.DefaultTimeToLive,
-			LibP2PResourceManagerConfig:                p2pbuilder.DefaultResourceManagerConfig(),
-			ConnectionManagerConfig:                    connection.DefaultConnManagerConfig(),
-			NetworkConnectionPruning:                   connection.ConnectionPruningEnabled,
-			GossipSubConfig:                            p2pbuilder.DefaultGossipSubConfig(),
-			GossipSubRPCInspectorNotificationCacheSize: distributor.DefaultGossipSubInspectorNotificationQueueCacheSize,
-			GossipSubRPCInspectorCacheSize:             validation.DefaultControlMsgValidationInspectorQueueCacheSize,
-			DisallowListNotificationCacheSize:          distributor.DefaultDisallowListNotificationQueueCacheSize,
+			GossipSubConfig:                   p2pbuilder.DefaultGossipSubConfig(),
+			GossipSubRPCInspectorsConfig:      inspectorbuilder.DefaultGossipSubRPCInspectorsConfig(),
+			DNSCacheTTL:                       dns.DefaultTimeToLive,
+			LibP2PResourceManagerConfig:       p2pbuilder.DefaultResourceManagerConfig(),
+			ConnectionManagerConfig:           connection.DefaultConnManagerConfig(),
+			NetworkConnectionPruning:          connection.ConnectionPruningEnabled,
+			DisallowListNotificationCacheSize: distributor.DefaultDisallowListNotificationQueueCacheSize,
 		},
 		nodeIDHex:        NotSet,
 		AdminAddr:        NotSet,