Merge #4118

4118: [Follower] Skipping proposals too far ahead of locally finalized view r=durkmurder a=durkmurder

### Context

This PR removes `compliance.Config` from `follower.Core` and moves it to `follower.Engine` which uses `SkipNewProposalsThreshold` to drop proposals that are too far ahead in future.

Co-authored-by: Yurii Oleksyshyn <[email protected]>

Author: bors[bot]

cmd/access/node_builder/access_node_builder.go

@@ -334,7 +334,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)
@@ -348,6 +347,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)

cmd/collection/main.go

@@ -317,7 +317,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 +330,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/execution_builder.go

@@ -899,7 +899,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 +912,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/observer/node_builder/observer_builder.go

@@ -35,7 +35,6 @@ import (
 	"github.com/onflow/flow-go/engine/access/rpc"
 	"github.com/onflow/flow-go/engine/access/rpc/backend"
 	"github.com/onflow/flow-go/engine/common/follower"
-	followereng "github.com/onflow/flow-go/engine/common/follower"
 	synceng "github.com/onflow/flow-go/engine/common/synchronization"
 	"github.com/onflow/flow-go/engine/protocol"
 	"github.com/onflow/flow-go/model/encodable"
@@ -185,7 +184,7 @@ type ObserverServiceBuilder struct {
 	SyncEngineParticipantsProviderFactory func() module.IdentifierProvider
 
 	// engines
-	FollowerEng *followereng.ComplianceEngine
+	FollowerEng *follower.ComplianceEngine
 	SyncEng     *synceng.Engine
 
 	// Public network
@@ -356,7 +355,7 @@ func (builder *ObserverServiceBuilder) buildFollowerEngine() *ObserverServiceBui
 		if node.HeroCacheMetricsEnable {
 			heroCacheCollector = metrics.FollowerCacheMetrics(node.MetricsRegisterer)
 		}
-		core, err := followereng.NewComplianceCore(
+		core, err := follower.NewComplianceCore(
 			node.Logger,
 			node.Metrics.Mempool,
 			heroCacheCollector,
@@ -366,20 +365,20 @@ func (builder *ObserverServiceBuilder) buildFollowerEngine() *ObserverServiceBui
 			builder.Validator,
 			builder.SyncCore,
 			node.Tracer,
-			compliance.WithSkipNewProposalsThreshold(builder.ComplianceConfig.SkipNewProposalsThreshold),
 		)
 		if err != nil {
 			return nil, fmt.Errorf("could not create follower core: %w", err)
 		}
 
-		builder.FollowerEng, err = followereng.NewComplianceLayer(
+		builder.FollowerEng, err = follower.NewComplianceLayer(
 			node.Logger,
 			node.Network,
 			node.Me,
 			node.Metrics.Engine,
 			node.Storage.Headers,
 			builder.Finalized,
 			core,
+			follower.WithComplianceConfigOpt(compliance.WithSkipNewProposalsThreshold(builder.ComplianceConfig.SkipNewProposalsThreshold)),
 			follower.WithChannel(channels.PublicReceiveBlocks),
 		)
 		if err != nil {

cmd/verification_builder.go

@@ -381,7 +381,6 @@ func (v *VerificationNodeBuilder) LoadComponentsAndModules() {
 				validator,
 				syncCore,
 				node.Tracer,
-				modulecompliance.WithSkipNewProposalsThreshold(node.ComplianceConfig.SkipNewProposalsThreshold),
 			)
 			if err != nil {
 				return nil, fmt.Errorf("could not create follower core: %w", err)
@@ -395,6 +394,7 @@ func (v *VerificationNodeBuilder) LoadComponentsAndModules() {
 				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)

engine/collection/compliance/core.go

@@ -150,10 +150,10 @@ func (c *Core) OnBlockProposal(originID flow.Identifier, proposal *messages.Clus
 	// 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.
-	if header.Height > finalHeight+c.config.SkipNewProposalsThreshold {
+	if header.View > finalView+c.config.SkipNewProposalsThreshold {
 		log.Debug().
 			Uint64("skip_new_proposals_threshold", c.config.SkipNewProposalsThreshold).
-			Msg("dropping block too far ahead of locally finalized height")
+			Msg("dropping block too far ahead of locally finalized view")
 		return nil
 	}
 

engine/common/follower/compliance_core.go

@@ -13,7 +13,6 @@ import (
 	"github.com/onflow/flow-go/engine/common/follower/pending_tree"
 	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/module"
-	"github.com/onflow/flow-go/module/compliance"
 	"github.com/onflow/flow-go/module/component"
 	"github.com/onflow/flow-go/module/irrecoverable"
 	"github.com/onflow/flow-go/module/trace"
@@ -41,7 +40,6 @@ type ComplianceCore struct {
 	*component.ComponentManager
 	log                 zerolog.Logger
 	mempoolMetrics      module.MempoolMetrics
-	config              compliance.Config
 	tracer              module.Tracer
 	pendingCache        *cache.Cache
 	pendingTree         *pending_tree.PendingTree
@@ -66,17 +64,11 @@ func NewComplianceCore(log zerolog.Logger,
 	validator hotstuff.Validator,
 	sync module.BlockRequester,
 	tracer module.Tracer,
-	opts ...compliance.Opt,
 ) (*ComplianceCore, error) {
 	onEquivocation := func(block, otherBlock *flow.Block) {
 		finalizationConsumer.OnDoubleProposeDetected(model.BlockFromFlow(block.Header), model.BlockFromFlow(otherBlock.Header))
 	}
 
-	config := compliance.DefaultConfig()
-	for _, apply := range opts {
-		apply(&config)
-	}
-
 	finalizedBlock, err := state.Final().Head()
 	if err != nil {
 		return nil, fmt.Errorf("could not query finalized block: %w", err)
@@ -92,7 +84,6 @@ func NewComplianceCore(log zerolog.Logger,
 		validator:           validator,
 		sync:                sync,
 		tracer:              tracer,
-		config:              config,
 		certifiedRangesChan: make(chan CertifiedBlocks, defaultCertifiedRangeChannelCapacity),
 		finalizedBlocksChan: make(chan *flow.Header, defaultFinalizedBlocksChannelCapacity),
 	}

engine/common/follower/compliance_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/component"
 	"github.com/onflow/flow-go/module/irrecoverable"
 	"github.com/onflow/flow-go/module/metrics"
@@ -30,6 +31,13 @@ func WithChannel(channel channels.Channel) EngineOption {
 	}
 }
 
+// WithComplianceConfigOpt applies compliance config opt to internal config
+func WithComplianceConfigOpt(opt compliance.Opt) EngineOption {
+	return func(e *ComplianceEngine) {
+		opt(&e.config)
+	}
+}
+
 // defaultBatchProcessingWorkers number of concurrent workers that process incoming blocks.
 const defaultBatchProcessingWorkers = 4
 
@@ -65,6 +73,7 @@ type ComplianceEngine struct {
 	me                         module.Local
 	engMetrics                 module.EngineMetrics
 	con                        network.Conduit
+	config                     compliance.Config
 	channel                    channels.Channel
 	headers                    storage.Headers
 	pendingProposals           *fifoqueue.FifoQueue        // queue for fresh proposals
@@ -106,6 +115,7 @@ func NewComplianceLayer(
 		log:                        log.With().Str("engine", "follower").Logger(),
 		me:                         me,
 		engMetrics:                 engMetrics,
+		config:                     compliance.DefaultConfig(),
 		channel:                    channels.ReceiveBlocks,
 		pendingProposals:           pendingBlocks,
 		syncedBlocks:               syncedBlocks,
@@ -327,6 +337,13 @@ func (e *ComplianceEngine) submitConnectedBatch(log zerolog.Logger, latestFinali
 		log.Debug().Msgf("dropping range [%d, %d] below finalized view %d", blocks[0].Header.View, lastBlock.View, latestFinalizedView)
 		return
 	}
+	if lastBlock.View > latestFinalizedView+e.config.SkipNewProposalsThreshold {
+		log.Debug().
+			Uint64("skip_new_proposals_threshold", e.config.SkipNewProposalsThreshold).
+			Msgf("dropping range [%d, %d] too far ahead of locally finalized view %d",
+				blocks[0].Header.View, lastBlock.View, latestFinalizedView)
+		return
+	}
 	log.Debug().Msgf("submitting sub-range with views [%d, %d] for further processing", blocks[0].Header.View, lastBlock.View)
 
 	select {

engine/consensus/compliance/core.go

@@ -158,10 +158,10 @@ func (c *Core) OnBlockProposal(originID flow.Identifier, proposal *messages.Bloc
 	// 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.
-	if header.Height > finalHeight+c.config.SkipNewProposalsThreshold {
+	if header.View > finalView+c.config.SkipNewProposalsThreshold {
 		log.Debug().
 			Uint64("skip_new_proposals_threshold", c.config.SkipNewProposalsThreshold).
-			Msg("dropping block too far ahead of locally finalized height")
+			Msg("dropping block too far ahead of locally finalized view")
 		return nil
 	}
 

engine/consensus/compliance/core_test.go

@@ -321,7 +321,7 @@ func (cs *CoreSuite) TestOnBlockProposalSkipProposalThreshold() {
 	// create a proposal which is far enough ahead to be dropped
 	originID := cs.participants[1].NodeID
 	block := unittest.BlockFixture()
-	block.Header.Height = cs.head.Height + compliance.DefaultConfig().SkipNewProposalsThreshold + 1
+	block.Header.View = cs.head.View + compliance.DefaultConfig().SkipNewProposalsThreshold + 1
 	proposal := unittest.ProposalFromBlock(&block)
 
 	err := cs.core.OnBlockProposal(originID, proposal)