Merge branch 'master' into amlandeep/add-random-EN-lookup-on-recipt-lookup-failure

Author: synzhu

cmd/access/node_builder/access_node_builder.go

@@ -12,6 +12,7 @@ import (
 	"github.com/spf13/pflag"
 
 	"github.com/onflow/flow-go/crypto"
+	"github.com/onflow/flow-go/module/compliance"
 
 	"github.com/onflow/flow-go/cmd"
 	"github.com/onflow/flow-go/consensus"
@@ -309,6 +310,7 @@ func (builder *FlowAccessNodeBuilder) buildFollowerEngine() *FlowAccessNodeBuild
 			builder.FollowerCore,
 			builder.SyncCore,
 			node.Tracer,
+			compliance.WithSkipNewProposalsThreshold(builder.ComplianceConfig.SkipNewProposalsThreshold),
 		)
 		if err != nil {
 			return nil, fmt.Errorf("could not create follower engine: %w", err)

cmd/bootstrap/cmd/constraints.go

@@ -50,12 +50,12 @@ func checkConstraints(partnerNodes, internalNodes []model.NodeInfo) {
 			if _, exists := internals.ByNodeID(node.NodeID); exists {
 				clusterInternalCount++
 			}
-			if clusterInternalCount <= clusterPartnerCount*2 {
-				log.Fatal().Msgf(
-					"will not bootstrap configuration without Byzantine majority within cluster: "+
-						"(partners=%d, internals=%d, min_internals=%d)",
-					clusterPartnerCount, clusterInternalCount, clusterPartnerCount*2+1)
-			}
+		}
+		if clusterInternalCount <= clusterPartnerCount*2 {
+			log.Fatal().Msgf(
+				"will not bootstrap configuration without Byzantine majority within cluster: "+
+					"(partners=%d, internals=%d, min_internals=%d)",
+				clusterPartnerCount, clusterInternalCount, clusterPartnerCount*2+1)
 		}
 		partnerCOLCount += clusterPartnerCount
 		internalCOLCount += clusterInternalCount

cmd/collection/main.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/onflow/flow-go/cmd/util/cmd/common"
 	"github.com/onflow/flow-go/model/bootstrap"
+	modulecompliance "github.com/onflow/flow-go/module/compliance"
 	"github.com/onflow/flow-go/module/mempool/herocache"
 
 	"github.com/onflow/flow-go-sdk/client"
@@ -68,9 +69,10 @@ func main() {
 		startupTimeString                      string
 		startupTime                            time.Time
 
-		followerState protocol.MutableState
-		ingestConf    = ingest.DefaultConfig()
-		rpcConf       rpc.Config
+		followerState           protocol.MutableState
+		ingestConf              = ingest.DefaultConfig()
+		rpcConf                 rpc.Config
+		clusterComplianceConfig modulecompliance.Config
 
 		pools                   *epochpool.TransactionPools // epoch-scoped transaction pools
 		followerBuffer          *buffer.PendingBlocks       // pending block cache for follower
@@ -140,6 +142,8 @@ func main() {
 			"additional fraction of replica timeout that the primary will wait for votes")
 		flags.DurationVar(&blockRateDelay, "block-rate-delay", 250*time.Millisecond,
 			"the delay to broadcast block proposal in order to control block production rate")
+		flags.Uint64Var(&clusterComplianceConfig.SkipNewProposalsThreshold,
+			"cluster-compliance-skip-proposals-threshold", modulecompliance.DefaultConfig().SkipNewProposalsThreshold, "threshold at which new proposals are discarded rather than cached, if their height is this much above local finalized height (cluster compliance engine)")
 		flags.StringVar(&startupTimeString, "hotstuff-startup-time", cmd.NotSet, "specifies date and time (in ISO 8601 format) after which the consensus participant may enter the first view (e.g (e.g 1996-04-24T15:04:05-07:00))")
 
 		// epoch qc contract flags
@@ -306,6 +310,7 @@ func main() {
 				followerCore,
 				mainChainSyncCore,
 				node.Tracer,
+				modulecompliance.WithSkipNewProposalsThreshold(node.ComplianceConfig.SkipNewProposalsThreshold),
 			)
 			if err != nil {
 				return nil, fmt.Errorf("could not create follower engine: %w", err)
@@ -425,6 +430,7 @@ func main() {
 				node.Metrics.Mempool,
 				node.State,
 				node.Storage.Transactions,
+				modulecompliance.WithSkipNewProposalsThreshold(clusterComplianceConfig.SkipNewProposalsThreshold),
 			)
 			if err != nil {
 				return nil, err

cmd/consensus/main.go

@@ -47,6 +47,7 @@ import (
 	"github.com/onflow/flow-go/module/buffer"
 	builder "github.com/onflow/flow-go/module/builder/consensus"
 	chmodule "github.com/onflow/flow-go/module/chunks"
+	modulecompliance "github.com/onflow/flow-go/module/compliance"
 	dkgmodule "github.com/onflow/flow-go/module/dkg"
 	"github.com/onflow/flow-go/module/epochs"
 	finalizer "github.com/onflow/flow-go/module/finalizer/consensus"
@@ -657,13 +658,21 @@ func main() {
 				mutableState,
 				proposals,
 				syncCore,
-				hotstuffModules.Aggregator)
+				hotstuffModules.Aggregator,
+				modulecompliance.WithSkipNewProposalsThreshold(node.ComplianceConfig.SkipNewProposalsThreshold),
+			)
 			if err != nil {
 				return nil, fmt.Errorf("could not initialize compliance core: %w", err)
 			}
 
 			// initialize the compliance engine
-			comp, err = compliance.NewEngine(node.Logger, node.Network, node.Me, prov, complianceCore)
+			comp, err = compliance.NewEngine(
+				node.Logger,
+				node.Network,
+				node.Me,
+				prov,
+				complianceCore,
+			)
 			if err != nil {
 				return nil, fmt.Errorf("could not initialize compliance engine: %w", err)
 			}

cmd/execution/main.go

@@ -61,6 +61,7 @@ import (
 	"github.com/onflow/flow-go/model/flow/filter"
 	"github.com/onflow/flow-go/module"
 	"github.com/onflow/flow-go/module/buffer"
+	"github.com/onflow/flow-go/module/compliance"
 	finalizer "github.com/onflow/flow-go/module/finalizer/consensus"
 	"github.com/onflow/flow-go/module/metrics"
 	"github.com/onflow/flow-go/module/state_synchronization"
@@ -674,6 +675,7 @@ func main() {
 				followerCore,
 				syncCore,
 				node.Tracer,
+				compliance.WithSkipNewProposalsThreshold(node.ComplianceConfig.SkipNewProposalsThreshold),
 			)
 			if err != nil {
 				return nil, fmt.Errorf("could not create follower engine: %w", err)

cmd/node_builder.go

@@ -14,6 +14,7 @@ import (
 	"github.com/spf13/pflag"
 
 	"github.com/onflow/flow-go/crypto"
+	"github.com/onflow/flow-go/module/compliance"
 
 	"github.com/onflow/flow-go/admin/commands"
 	"github.com/onflow/flow-go/fvm"
@@ -151,6 +152,9 @@ type BaseConfig struct {
 	topologyEdgeProbability         float64
 	HeroCacheMetricsEnable          bool
 	SyncCoreConfig                  synchronization.Config
+	// ComplianceConfig configures either the compliance engine (consensus nodes)
+	// or the follower engine (all other node roles)
+	ComplianceConfig compliance.Config
 }
 
 // NodeConfig contains all the derived parameters such the NodeID, private keys etc. and initialized instances of
@@ -232,5 +236,6 @@ func DefaultBaseConfig() *BaseConfig {
 		topologyEdgeProbability:         topology.MaximumEdgeProbability,
 		HeroCacheMetricsEnable:          false,
 		SyncCoreConfig:                  synchronization.DefaultConfig(),
+		ComplianceConfig:                compliance.DefaultConfig(),
 	}
 }

cmd/scaffold.go

@@ -162,6 +162,8 @@ func (fnb *FlowNodeBuilder) BaseFlags() {
 	fnb.flags.UintVar(&fnb.BaseConfig.SyncCoreConfig.MaxAttempts, "sync-max-attempts", defaultConfig.SyncCoreConfig.MaxAttempts, "the maximum number of attempts we make for each requested block/height before discarding")
 	fnb.flags.UintVar(&fnb.BaseConfig.SyncCoreConfig.MaxSize, "sync-max-size", defaultConfig.SyncCoreConfig.MaxSize, "the maximum number of blocks we request in the same block request message")
 	fnb.flags.UintVar(&fnb.BaseConfig.SyncCoreConfig.MaxRequests, "sync-max-requests", defaultConfig.SyncCoreConfig.MaxRequests, "the maximum number of requests we send during each scanning period")
+
+	fnb.flags.Uint64Var(&fnb.BaseConfig.ComplianceConfig.SkipNewProposalsThreshold, "compliance-skip-proposals-threshold", defaultConfig.ComplianceConfig.SkipNewProposalsThreshold, "threshold at which new proposals are discarded rather than cached, if their height is this much above local finalized height")
 }
 
 func (fnb *FlowNodeBuilder) EnqueuePingService() {

cmd/verification/main.go

@@ -26,6 +26,7 @@ import (
 	"github.com/onflow/flow-go/module"
 	"github.com/onflow/flow-go/module/buffer"
 	"github.com/onflow/flow-go/module/chunks"
+	"github.com/onflow/flow-go/module/compliance"
 	finalizer "github.com/onflow/flow-go/module/finalizer/consensus"
 	"github.com/onflow/flow-go/module/mempool"
 	"github.com/onflow/flow-go/module/mempool/stdmap"
@@ -328,6 +329,7 @@ func main() {
 				followerCore,
 				syncCore,
 				node.Tracer,
+				compliance.WithSkipNewProposalsThreshold(node.ComplianceConfig.SkipNewProposalsThreshold),
 			)
 			if err != nil {
 				return nil, fmt.Errorf("could not create follower engine: %w", err)

engine/collection/compliance/core.go

@@ -15,6 +15,7 @@ import (
 	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/model/messages"
 	"github.com/onflow/flow-go/module"
+	"github.com/onflow/flow-go/module/compliance"
 	"github.com/onflow/flow-go/module/metrics"
 	"github.com/onflow/flow-go/state"
 	clusterkv "github.com/onflow/flow-go/state/cluster"
@@ -28,6 +29,7 @@ import (
 // user of this object needs to ensure single thread access.
 type Core struct {
 	log               zerolog.Logger // used to log relevant actions with context
+	config            compliance.Config
 	metrics           module.EngineMetrics
 	mempoolMetrics    module.MempoolMetrics
 	collectionMetrics module.CollectionMetrics
@@ -49,10 +51,17 @@ func NewCore(
 	state clusterkv.MutableState,
 	pending module.PendingClusterBlockBuffer,
 	voteAggregator hotstuff.VoteAggregator,
+	opts ...compliance.Opt,
 ) (*Core, error) {
 
+	config := compliance.DefaultConfig()
+	for _, apply := range opts {
+		apply(&config)
+	}
+
 	c := &Core{
 		log:               log.With().Str("cluster_compliance", "core").Logger(),
+		config:            config,
 		metrics:           collector,
 		mempoolMetrics:    mempool,
 		collectionMetrics: collectionMetrics,
@@ -108,6 +117,20 @@ func (c *Core) OnBlockProposal(originID flow.Identifier, proposal *messages.Clus
 		return fmt.Errorf("could not check proposal: %w", err)
 	}
 
+	// ignore proposals which are too far ahead of our local finalized state
+	// instead, rely on sync engine to catch up finalization more effectively, and avoid
+	// large subtree of blocks to be cached.
+	final, err := c.state.Final().Head()
+	if err != nil {
+		return fmt.Errorf("could not get latest finalized header: %w", err)
+	}
+	if header.Height > final.Height && header.Height-final.Height > c.config.SkipNewProposalsThreshold {
+		log.Debug().
+			Uint64("final_height", final.Height).
+			Msg("dropping block too far ahead of locally finalized height")
+		return nil
+	}
+
 	// there are two possibilities if the proposal is neither already pending
 	// processing in the cache, nor has already been processed:
 	// 1) the proposal is unverifiable because parent or ancestor is unknown

engine/collection/compliance/core_test.go

@@ -17,6 +17,7 @@ import (
 	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/model/messages"
 	realbuffer "github.com/onflow/flow-go/module/buffer"
+	"github.com/onflow/flow-go/module/compliance"
 	"github.com/onflow/flow-go/module/metrics"
 	module "github.com/onflow/flow-go/module/mock"
 	clusterint "github.com/onflow/flow-go/state/cluster"
@@ -197,6 +198,22 @@ func (cs *ComplianceCoreSuite) TestOnBlockProposalValidParent() {
 	cs.hotstuff.AssertExpectations(cs.T())
 }
 
+func (cs *ComplianceCoreSuite) TestOnBlockProposalSkipProposalThreshold() {
+
+	// create a proposal which is far enough ahead to be dropped
+	originID := unittest.IdentifierFixture()
+	block := unittest.ClusterBlockFixture()
+	block.Header.Height = cs.head.Header.Height + compliance.DefaultConfig().SkipNewProposalsThreshold + 1
+	proposal := unittest.ClusterProposalFromBlock(&block)
+
+	err := cs.core.OnBlockProposal(originID, proposal)
+	require.NoError(cs.T(), err)
+
+	// block should be dropped - not added to state or cache
+	cs.state.AssertNotCalled(cs.T(), "Extend", mock.Anything)
+	cs.pending.AssertNotCalled(cs.T(), "Add", originID, mock.Anything)
+}
+
 func (cs *ComplianceCoreSuite) TestOnBlockProposalValidAncestor() {
 
 	// create a proposal that has two ancestors in the cache

engine/collection/epochmgr/factories/proposal.go

@@ -9,6 +9,7 @@ import (
 	"github.com/onflow/flow-go/engine/collection/compliance"
 	"github.com/onflow/flow-go/module"
 	"github.com/onflow/flow-go/module/buffer"
+	modulecompliance "github.com/onflow/flow-go/module/compliance"
 	"github.com/onflow/flow-go/network"
 	"github.com/onflow/flow-go/state/cluster"
 	"github.com/onflow/flow-go/state/protocol"
@@ -24,6 +25,7 @@ type ProposalEngineFactory struct {
 	mempoolMetrics module.MempoolMetrics
 	protoState     protocol.State
 	transactions   storage.Transactions
+	complianceOpts []modulecompliance.Opt
 }
 
 // NewFactory returns a new collection proposal engine factory.
@@ -36,6 +38,7 @@ func NewProposalEngineFactory(
 	mempoolMetrics module.MempoolMetrics,
 	protoState protocol.State,
 	transactions storage.Transactions,
+	complianceOpts ...modulecompliance.Opt,
 ) (*ProposalEngineFactory, error) {
 
 	factory := &ProposalEngineFactory{
@@ -47,6 +50,7 @@ func NewProposalEngineFactory(
 		mempoolMetrics: mempoolMetrics,
 		protoState:     protoState,
 		transactions:   transactions,
+		complianceOpts: complianceOpts,
 	}
 	return factory, nil
 }
@@ -68,6 +72,7 @@ func (f *ProposalEngineFactory) Create(
 		clusterState,
 		cache,
 		voteAggregator,
+		f.complianceOpts...,
 	)
 	if err != nil {
 		return nil, fmt.Errorf("could create cluster compliance core: %w", err)

engine/common/follower/engine.go

@@ -13,6 +13,7 @@ import (
 	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/model/messages"
 	"github.com/onflow/flow-go/module"
+	"github.com/onflow/flow-go/module/compliance"
 	"github.com/onflow/flow-go/module/metrics"
 	"github.com/onflow/flow-go/module/trace"
 	"github.com/onflow/flow-go/network"
@@ -25,6 +26,7 @@ import (
 type Engine struct {
 	unit           *engine.Unit
 	log            zerolog.Logger
+	config         compliance.Config
 	me             module.Local
 	engMetrics     module.EngineMetrics
 	mempoolMetrics module.MempoolMetrics
@@ -53,11 +55,18 @@ func New(
 	follower module.HotStuffFollower,
 	sync module.BlockRequester,
 	tracer module.Tracer,
+	opts ...compliance.Opt,
 ) (*Engine, error) {
 
+	config := compliance.DefaultConfig()
+	for _, apply := range opts {
+		apply(&config)
+	}
+
 	e := &Engine{
 		unit:           engine.NewUnit(),
 		log:            log.With().Str("engine", "follower").Logger(),
+		config:         config,
 		me:             me,
 		engMetrics:     engMetrics,
 		mempoolMetrics: mempoolMetrics,
@@ -215,6 +224,20 @@ func (e *Engine) onBlockProposal(originID flow.Identifier, proposal *messages.Bl
 		return fmt.Errorf("could not check proposal: %w", err)
 	}
 
+	// ignore proposals which are too far ahead of our local finalized state
+	// instead, rely on sync engine to catch up finalization more effectively, and avoid
+	// large subtree of blocks to be cached.
+	final, err := e.state.Final().Head()
+	if err != nil {
+		return fmt.Errorf("could not get latest finalized header: %w", err)
+	}
+	if header.Height > final.Height && header.Height-final.Height > e.config.SkipNewProposalsThreshold {
+		log.Debug().
+			Uint64("final_height", final.Height).
+			Msg("dropping block too far ahead of locally finalized height")
+		return nil
+	}
+
 	// there are two possibilities if the proposal is neither already pending
 	// processing in the cache, nor has already been processed:
 	// 1) the proposal is unverifiable because parent or ancestor is unknown