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,

engine/consensus/compliance/core.go

@@ -100,7 +100,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)
+	}
 
 	c.mempoolMetrics.MempoolEntries(metrics.ResourceProposal, c.pending.Size())
 
@@ -200,7 +203,15 @@ func (c *Core) OnBlockProposal(proposal flow.Slashable[*flow.Proposal]) error {
 	_, found := c.pending.ByID(header.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.ResourceProposal, c.pending.Size())
 
 		return nil
@@ -214,7 +225,15 @@ func (c *Core) OnBlockProposal(proposal flow.Slashable[*flow.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.ResourceProposal, c.pending.Size())
 
 		c.sync.RequestBlock(header.ParentID, header.Height-1)
@@ -300,9 +319,6 @@ func (c *Core) processBlockAndDescendants(proposal flow.Slashable[*flow.Proposal
 		}
 	}
 
-	// drop all the children that should have been processed now
-	c.pending.DropForParent(blockID)
-
 	return nil
 }
 
@@ -392,14 +408,19 @@ func (c *Core) processBlockProposal(proposal *flow.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.ResourceProposal, c.pending.Size())
+	return nil
 }
 
 // checkForAndLogOutdatedInputError checks whether error is an `engine.OutdatedInputError`.

engine/consensus/compliance/core_test.go

@@ -200,7 +200,7 @@ func (cs *CommonSuite) SetupTest() {
 
 	// set up pending module mock
 	cs.pending = &module.PendingBlockBuffer{}
-	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[*flow.Proposal] {
 			return cs.pendingDB[blockID]
@@ -219,9 +219,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)
 
 	// set up hotstuff module mock
 	cs.hotstuff = module.NewHotStuff(cs.T())
@@ -565,9 +564,6 @@ func (cs *CoreSuite) TestProcessBlockAndDescendants() {
 		Message:  proposal0,
 	})
 	require.NoError(cs.T(), err, "should pass handling children")
-
-	// make sure we drop the cache after trying to process
-	cs.pending.AssertCalled(cs.T(), "DropForParent", parent.ID())
 }
 
 func (cs *CoreSuite) TestProposalBufferingOrder() {
@@ -588,7 +584,7 @@ func (cs *CoreSuite) TestProposalBufferingOrder() {
 	}
 
 	// replace the engine buffer with the real one
-	cs.core.pending = real.NewPendingBlocks()
+	cs.core.pending = real.NewPendingBlocks(cs.head.View, 100_000)
 
 	// check that we request the ancestor block each time
 	cs.sync.On("RequestBlock", missingBlock.ID(), missingBlock.Height).Times(len(proposals))

engine/consensus/compliance/engine.go

@@ -180,6 +180,9 @@ 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)
+	err = e.core.ProcessFinalizedBlock(finalHeader)
+	if err != nil {
+		return fmt.Errorf("could not process finalized block: %w", err)
+	}
 	return nil
 }

model/cluster/block.go

@@ -82,10 +82,7 @@ func NewRootBlock(untrusted UntrustedBlock) (*Block, error) {
 // Proposal represents a signed proposed block in collection node cluster consensus.
 //
 //structwrite:immutable - mutations allowed only within the constructor.
-type Proposal struct {
-	Block           Block
-	ProposerSigData []byte
-}
+type Proposal = flow.GenericProposal[Payload]
 
 // UntrustedProposal is an untrusted input-only representation of a cluster.Proposal,
 // used for construction.
@@ -137,12 +134,6 @@ func NewRootProposal(untrusted UntrustedProposal) (*Proposal, error) {
 
 }
 
-// ProposalHeader converts the proposal into a compact [ProposalHeader] representation,
-// where the payload is compressed to a hash reference.
-func (p *Proposal) ProposalHeader() *flow.ProposalHeader {
-	return &flow.ProposalHeader{Header: p.Block.ToHeader(), ProposerSigData: p.ProposerSigData}
-}
-
 // BlockResponse is the same as flow.BlockResponse, but for cluster
 // consensus. It contains a list of structurally validated cluster block proposals
 // that should correspond to the request.

model/flow/block.go

@@ -163,13 +163,19 @@ func (s BlockStatus) String() string {
 	return [...]string{"BLOCK_UNKNOWN", "BLOCK_FINALIZED", "BLOCK_SEALED"}[s]
 }
 
+// GenericProposal represents a generic Flow proposal structure parameterized by a block payload type.
+// It includes the block header metadata, block payload, and proposer signature.
+type GenericProposal[T HashablePayload] struct {
+	// Block is the block being proposed.
+	Block GenericBlock[T]
+	// ProposerSigData is the signature (vote) from the proposer of this block.
+	ProposerSigData []byte
+}
+
 // Proposal is a signed proposal that includes the block payload, in addition to the required header and signature.
 //
 //structwrite:immutable - mutations allowed only within the constructor
-type Proposal struct {
-	Block           Block
-	ProposerSigData []byte
-}
+type Proposal = GenericProposal[Payload]
 
 // UntrustedProposal is an untrusted input-only representation of a Proposal,
 // used for construction.
@@ -223,7 +229,7 @@ func NewRootProposal(untrusted UntrustedProposal) (*Proposal, error) {
 
 // ProposalHeader converts the proposal into a compact [ProposalHeader] representation,
 // where the payload is compressed to a hash reference.
-func (b *Proposal) ProposalHeader() *ProposalHeader {
+func (b *GenericProposal[T]) ProposalHeader() *ProposalHeader {
 	return &ProposalHeader{Header: b.Block.ToHeader(), ProposerSigData: b.ProposerSigData}
 }