Improve BFT resilience of BlockBuffer stores

cmd/consensus/main.go

@@ -805,7 +805,7 @@ func main() {
 		}).
 		Component("consensus compliance engine", func(node *cmd.NodeConfig) (module.ReadyDoneAware, error) {
 			// initialize the pending blocks cache
-			proposals := buffer.NewPendingBlocks()
+			proposals := buffer.NewPendingBlocks(node.LastFinalizedHeader.View, node.ComplianceConfig.GetSkipNewProposalsThreshold())
 
 			logger := createLogger(node.Logger, node.RootChainID)
 			complianceCore, err := compliance.NewCore(

consensus/integration/nodes_test.go

@@ -514,12 +514,12 @@ func createNode(
 	)
 	require.NoError(t, err)
 
-	// initialize the pending blocks cache
-	cache := buffer.NewPendingBlocks()
-
 	rootHeader, err := rootSnapshot.Head()
 	require.NoError(t, err)
 
+	// initialize the pending blocks cache
+	cache := buffer.NewPendingBlocks(rootHeader.View, modulecompliance.DefaultConfig().SkipNewProposalsThreshold)
+
 	rootQC, err := rootSnapshot.QuorumCertificate()
 	require.NoError(t, err)
 

engine/collection/compliance/core.go

@@ -93,7 +93,10 @@ func NewCore(
 	if err != nil {
 		return nil, fmt.Errorf("could not initialized finalized boundary cache: %w", err)
 	}
-	c.ProcessFinalizedBlock(final)
+	err = c.ProcessFinalizedBlock(final)
+	if err != nil {
+		return nil, fmt.Errorf("could not process finalized block: %w", err)
+	}
 
 	// log the mempool size off the bat
 	c.mempoolMetrics.MempoolEntries(metrics.ResourceClusterProposal, c.pending.Size())
@@ -193,7 +196,15 @@ func (c *Core) OnBlockProposal(proposal flow.Slashable[*cluster.Proposal]) error
 	_, found := c.pending.ByID(block.ParentID)
 	if found {
 		// add the block to the cache
-		_ = c.pending.Add(proposal)
+		if err := c.pending.Add(proposal); err != nil {
+			if mempool.IsBeyondActiveRangeError(err) {
+				// In general we expect the block buffer to use SkipNewProposalsThreshold,
+				// however since it is instantiated outside this component, we allow the thresholds to differ
+				log.Debug().Err(err).Msg("dropping block beyond block buffer active range")
+				return nil
+			}
+			return fmt.Errorf("could not add proposal to pending buffer: %w", err)
+		}
 		c.mempoolMetrics.MempoolEntries(metrics.ResourceClusterProposal, c.pending.Size())
 
 		return nil
@@ -207,7 +218,15 @@ func (c *Core) OnBlockProposal(proposal flow.Slashable[*cluster.Proposal]) error
 		return fmt.Errorf("could not check parent exists: %w", err)
 	}
 	if !exists {
-		_ = c.pending.Add(proposal)
+		if err := c.pending.Add(proposal); err != nil {
+			if mempool.IsBeyondActiveRangeError(err) {
+				// In general we expect the block buffer to use SkipNewProposalsThreshold,
+				// however since it is instantiated outside this component, we allow the thresholds to differ
+				log.Debug().Err(err).Msg("dropping block beyond block buffer active range")
+				return nil
+			}
+			return fmt.Errorf("could not add proposal to pending buffer: %w", err)
+		}
 		c.mempoolMetrics.MempoolEntries(metrics.ResourceClusterProposal, c.pending.Size())
 
 		c.sync.RequestBlock(block.ParentID, block.Height-1)
@@ -288,9 +307,6 @@ func (c *Core) processBlockAndDescendants(proposal flow.Slashable[*cluster.Propo
 		}
 	}
 
-	// drop all the children that should have been processed now
-	c.pending.DropForParent(blockID)
-
 	return nil
 }
 
@@ -363,14 +379,19 @@ func (c *Core) processBlockProposal(proposal *cluster.Proposal) error {
 
 // ProcessFinalizedBlock performs pruning of stale data based on finalization event
 // removes pending blocks below the finalized view
-func (c *Core) ProcessFinalizedBlock(finalized *flow.Header) {
+// No errors are expected during normal operation.
+func (c *Core) ProcessFinalizedBlock(finalized *flow.Header) error {
 	// remove all pending blocks at or below the finalized view
-	c.pending.PruneByView(finalized.View)
+	err := c.pending.PruneByView(finalized.View)
+	if err != nil {
+		return err
+	}
 	c.finalizedHeight.Set(finalized.Height)
 	c.finalizedView.Set(finalized.View)
 
 	// always record the metric
 	c.mempoolMetrics.MempoolEntries(metrics.ResourceClusterProposal, c.pending.Size())
+	return nil
 }
 
 // checkForAndLogOutdatedInputError checks whether error is an `engine.OutdatedInputError`.

engine/collection/compliance/core_test.go

@@ -124,7 +124,7 @@ func (cs *CommonSuite) SetupTest() {
 
 	// set up pending module mock
 	cs.pending = &module.PendingClusterBlockBuffer{}
-	cs.pending.On("Add", mock.Anything, mock.Anything).Return(true)
+	cs.pending.On("Add", mock.Anything, mock.Anything)
 	cs.pending.On("ByID", mock.Anything).Return(
 		func(blockID flow.Identifier) flow.Slashable[*cluster.Proposal] {
 			return cs.pendingDB[blockID]
@@ -143,9 +143,8 @@ func (cs *CommonSuite) SetupTest() {
 			return ok
 		},
 	)
-	cs.pending.On("DropForParent", mock.Anything).Return()
 	cs.pending.On("Size").Return(uint(0))
-	cs.pending.On("PruneByView", mock.Anything).Return()
+	cs.pending.On("PruneByView", mock.Anything).Return(nil)
 
 	closed := func() <-chan struct{} {
 		channel := make(chan struct{})
@@ -518,9 +517,6 @@ func (cs *CoreSuite) TestProcessBlockAndDescendants() {
 
 	// check that we submitted each child to hotstuff
 	cs.hotstuff.AssertExpectations(cs.T())
-
-	// make sure we drop the cache after trying to process
-	cs.pending.AssertCalled(cs.T(), "DropForParent", parent.ID())
 }
 
 func (cs *CoreSuite) TestProposalBufferingOrder() {
@@ -546,7 +542,7 @@ func (cs *CoreSuite) TestProposalBufferingOrder() {
 	}
 
 	// replace the engine buffer with the real one
-	cs.core.pending = realbuffer.NewPendingClusterBlocks()
+	cs.core.pending = realbuffer.NewPendingClusterBlocks(cs.head.Block.View, 100_000)
 
 	// process all of the descendants
 	for _, proposal := range proposals {

engine/collection/compliance/engine.go

@@ -166,6 +166,8 @@ func (e *Engine) processOnFinalizedBlock(block *model.Block) error {
 	if err != nil { // no expected errors
 		return fmt.Errorf("could not get finalized header: %w", err)
 	}
-	e.core.ProcessFinalizedBlock(finalHeader)
+	if err := e.core.ProcessFinalizedBlock(finalHeader); err != nil {
+		return fmt.Errorf("could not process finalized block: %w", err)
+	}
 	return nil
 }

engine/collection/epochmgr/factories/compliance.go

@@ -66,7 +66,11 @@ func (f *ComplianceEngineFactory) Create(
 	validator hotstuff.Validator,
 ) (*compliance.Engine, error) {
 
-	cache := buffer.NewPendingClusterBlocks()
+	final, err := clusterState.Final().Head()
+	if err != nil {
+		return nil, fmt.Errorf("could not get finalized header: %w", err)
+	}
+	cache := buffer.NewPendingClusterBlocks(final.View, f.config.GetSkipNewProposalsThreshold())
 	core, err := compliance.NewCore(
 		f.log,
 		f.engMetrics,