ai: add threshold also to mempool

it already exists in the compliance engine as well, but including in
mempool provides stronger localized guarantee of behaviour

Author: jordanschalm

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(node.LastFinalizedHeader.View)
+			proposals := buffer.NewPendingBlocks(node.LastFinalizedHeader.View, node.ComplianceConfig.GetSkipNewProposalsThreshold())
 
 			logger := createLogger(node.Logger, node.RootChainID)
 			complianceCore, err := compliance.NewCore(

consensus/integration/nodes_test.go

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

engine/collection/compliance/core.go

@@ -196,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
@@ -210,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)

engine/collection/compliance/core_test.go

@@ -542,7 +542,7 @@ func (cs *CoreSuite) TestProposalBufferingOrder() {
 	}
 
 	// replace the engine buffer with the real one
-	cs.core.pending = realbuffer.NewPendingClusterBlocks(cs.head.Block.View)
+	cs.core.pending = realbuffer.NewPendingClusterBlocks(cs.head.Block.View, 100_000)
 
 	// process all of the descendants
 	for _, proposal := range proposals {

engine/collection/epochmgr/factories/compliance.go

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

engine/consensus/compliance/core.go

@@ -203,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
@@ -217,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)

engine/consensus/compliance/core_test.go

@@ -584,7 +584,7 @@ func (cs *CoreSuite) TestProposalBufferingOrder() {
 	}
 
 	// replace the engine buffer with the real one
-	cs.core.pending = real.NewPendingBlocks(cs.head.View)
+	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))

module/buffer.go

@@ -20,7 +20,9 @@ type BufferedProposal interface {
 type GenericPendingBlockBuffer[T BufferedProposal] interface {
 	// Add adds the input block to the block buffer.
 	// If the block already exists, or is below the finalized view, this is a no-op.
-	Add(block flow.Slashable[T])
+	// Errors returns:
+	//   - mempool.BlockViewTooFarAheadError if block.View > finalizedView + activeViewRangeSize (when activeViewRangeSize > 0)
+	Add(block flow.Slashable[T]) error
 
 	// ByID returns the block with the given ID, if it exists.
 	// Otherwise returns (nil, false)

module/buffer/pending_blocks.go

@@ -7,6 +7,7 @@ import (
 	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/module"
 	"github.com/onflow/flow-go/module/forest"
+	"github.com/onflow/flow-go/module/mempool"
 )
 
 // proposalVertex implements [forest.Vertex] for generic block proposals.
@@ -51,8 +52,9 @@ func (v proposalVertex[T]) Parent() (flow.Identifier, uint64) {
 //
 // Safe for concurrent use.
 type GenericPendingBlocks[T module.BufferedProposal] struct {
-	lock   *sync.Mutex
-	forest *forest.LevelledForest
+	lock                *sync.Mutex
+	forest              *forest.LevelledForest
+	activeViewRangeSize uint64
 }
 
 type PendingBlocks = GenericPendingBlocks[*flow.Proposal]
@@ -61,26 +63,45 @@ type PendingClusterBlocks = GenericPendingBlocks[*cluster.Proposal]
 var _ module.PendingBlockBuffer = (*PendingBlocks)(nil)
 var _ module.PendingClusterBlockBuffer = (*PendingClusterBlocks)(nil)
 
-func NewPendingBlocks(finalizedView uint64) *PendingBlocks {
+func NewPendingBlocks(finalizedView uint64, activeViewRangeSize uint64) *PendingBlocks {
 	return &PendingBlocks{
-		lock:   new(sync.Mutex),
-		forest: forest.NewLevelledForest(finalizedView),
+		lock: new(sync.Mutex),
+		// LevelledForest's lowestLevel is inclusive, so add 1 here
+		forest:              forest.NewLevelledForest(finalizedView + 1),
+		activeViewRangeSize: activeViewRangeSize,
 	}
 }
 
-func NewPendingClusterBlocks(finalizedView uint64) *PendingClusterBlocks {
+func NewPendingClusterBlocks(finalizedView uint64, activeViewRangeSize uint64) *PendingClusterBlocks {
 	return &PendingClusterBlocks{
-		lock:   new(sync.Mutex),
-		forest: forest.NewLevelledForest(finalizedView),
+		lock:                new(sync.Mutex),
+		forest:              forest.NewLevelledForest(finalizedView),
+		activeViewRangeSize: activeViewRangeSize,
 	}
 }
 
 // Add adds the input block to the block buffer.
 // If the block already exists, or is below the finalized view, this is a no-op.
-func (b *GenericPendingBlocks[T]) Add(block flow.Slashable[T]) {
+// Errors returns:
+//   - mempool.BeyondActiveRangeError if block.View > finalizedView + activeViewRangeSize (when activeViewRangeSize > 0)
+func (b *GenericPendingBlocks[T]) Add(block flow.Slashable[T]) error {
 	b.lock.Lock()
 	defer b.lock.Unlock()
+
+	blockView := block.Message.ProposalHeader().Header.View
+	finalizedView := b.highestPrunedView()
+
+	// Check if block view exceeds the active view range size
+	// If activeViewRangeSize is 0, there's no limitation
+	if b.activeViewRangeSize > 0 && blockView > finalizedView+b.activeViewRangeSize {
+		return mempool.NewBeyondActiveRangeError(
+			"block view %d exceeds active view range size: finalized view %d + range size %d = %d",
+			blockView, finalizedView, b.activeViewRangeSize, finalizedView+b.activeViewRangeSize,
+		)
+	}
+
 	b.forest.AddVertex(newProposalVertex(block))
+	return nil
 }
 
 // ByID returns the block with the given ID, if it exists.
@@ -131,3 +152,10 @@ func (b *GenericPendingBlocks[T]) Size() uint {
 	defer b.lock.Unlock()
 	return uint(b.forest.GetSize())
 }
+
+// highestPrunedView returns the highest pruned view (finalized view).
+// CAUTION: Caller must acquire the lock.
+func (b *GenericPendingBlocks[T]) highestPrunedView() uint64 {
+	// LevelledForest.LowestLevel is the lowest UNPRUNED view, so subtract 1 here
+	return b.forest.LowestLevel - 1
+}

module/buffer/pending_blocks_test.go

@@ -22,8 +22,9 @@ func TestPendingBlocksSuite(t *testing.T) {
 }
 
 func (suite *PendingBlocksSuite) SetupTest() {
-	// Initialize with finalized view 0
-	suite.buffer = NewPendingBlocks(0)
+	// Initialize with finalized view 0 and no view range limitation (0 = no limit)
+	// Individual tests that need the limitation will create their own buffers
+	suite.buffer = NewPendingBlocks(0, 0)
 }
 
 // block creates a new block proposal wrapped as Slashable.
@@ -41,7 +42,7 @@ func (suite *PendingBlocksSuite) blockWithParent(parent *flow.Header) flow.Slash
 // TestAdd tests adding blocks to the buffer.
 func (suite *PendingBlocksSuite) TestAdd() {
 	block := suite.block()
-	suite.buffer.Add(block)
+	suite.Require().NoError(suite.buffer.Add(block))
 
 	// Verify block can be retrieved by ID
 	retrieved, ok := suite.buffer.ByID(block.Message.Block.ID())
@@ -59,8 +60,8 @@ func (suite *PendingBlocksSuite) TestAdd() {
 // TestAddDuplicate verifies that adding the same block twice is a no-op.
 func (suite *PendingBlocksSuite) TestAddDuplicate() {
 	block := suite.block()
-	suite.buffer.Add(block)
-	suite.buffer.Add(block) // Add again
+	suite.Require().NoError(suite.buffer.Add(block))
+	suite.Require().NoError(suite.buffer.Add(block)) // Add again
 
 	// Should still only have one block
 	suite.Assert().Equal(uint(1), suite.buffer.Size())
@@ -74,24 +75,88 @@ func (suite *PendingBlocksSuite) TestAddDuplicate() {
 // TestAddBelowFinalizedView verifies that adding blocks below finalized view is a no-op.
 func (suite *PendingBlocksSuite) TestAddBelowFinalizedView() {
 	finalizedView := uint64(1000)
-	buffer := NewPendingBlocks(finalizedView)
+	buffer := NewPendingBlocks(finalizedView, 100_000)
 
 	// Create a block with view below finalized
 	block := suite.block()
 	block.Message.Block.ParentView = finalizedView - 10
 	block.Message.Block.View = finalizedView - 5
 
-	buffer.Add(block)
+	suite.Require().NoError(buffer.Add(block))
 
 	_, ok := buffer.ByID(block.Message.Block.ID())
 	suite.Assert().False(ok)
 	suite.Assert().Equal(uint(0), buffer.Size())
 }
 
+// TestAddExceedsActiveViewRangeSize verifies that adding blocks that exceed the active view range size returns an error.
+func (suite *PendingBlocksSuite) TestAddExceedsActiveViewRangeSize() {
+	finalizedView := uint64(1000)
+	activeViewRangeSize := uint64(100)
+	buffer := NewPendingBlocks(finalizedView, activeViewRangeSize)
+
+	// Create a parent header and then a block that exceeds the active view range size
+	parentHeader := unittest.BlockHeaderFixture()
+	parentHeader.View = finalizedView + 50
+	block := suite.blockWithParent(parentHeader)
+	block.Message.Block.View = finalizedView + activeViewRangeSize + 1
+
+	err := buffer.Add(block)
+	suite.Assert().Error(err)
+	suite.Assert().True(mempool.IsBeyondActiveRangeError(err))
+
+	// Verify block was not added
+	_, ok := buffer.ByID(block.Message.Block.ID())
+	suite.Assert().False(ok)
+	suite.Assert().Equal(uint(0), buffer.Size())
+}
+
+// TestAddWithinActiveViewRangeSize verifies that adding blocks within the active view range size succeeds.
+func (suite *PendingBlocksSuite) TestAddWithinActiveViewRangeSize() {
+	finalizedView := uint64(1000)
+	activeViewRangeSize := uint64(100)
+	buffer := NewPendingBlocks(finalizedView, activeViewRangeSize)
+
+	// Create a parent header and then a block that is exactly at the limit
+	parentHeader := unittest.BlockHeaderFixture()
+	parentHeader.View = finalizedView + 50
+	block := suite.blockWithParent(parentHeader)
+	block.Message.Block.View = finalizedView + activeViewRangeSize
+
+	err := buffer.Add(block)
+	suite.Assert().NoError(err)
+
+	// Verify block was added
+	_, ok := buffer.ByID(block.Message.Block.ID())
+	suite.Assert().True(ok)
+	suite.Assert().Equal(uint(1), buffer.Size())
+}
+
+// TestAddWithZeroActiveViewRangeSize verifies that when activeViewRangeSize is 0, there's no limitation.
+func (suite *PendingBlocksSuite) TestAddWithZeroActiveViewRangeSize() {
+	finalizedView := uint64(1000)
+	activeViewRangeSize := uint64(0) // No limitation
+	buffer := NewPendingBlocks(finalizedView, activeViewRangeSize)
+
+	// Create a parent header and then a block that is very far ahead
+	parentHeader := unittest.BlockHeaderFixture()
+	parentHeader.View = finalizedView + 500_000
+	block := suite.blockWithParent(parentHeader)
+	block.Message.Block.View = finalizedView + 1_000_000
+
+	err := buffer.Add(block)
+	suite.Assert().NoError(err)
+
+	// Verify block was added
+	_, ok := buffer.ByID(block.Message.Block.ID())
+	suite.Assert().True(ok)
+	suite.Assert().Equal(uint(1), buffer.Size())
+}
+
 // TestByID tests retrieving blocks by ID.
 func (suite *PendingBlocksSuite) TestByID() {
 	block := suite.block()
-	suite.buffer.Add(block)
+	suite.Require().NoError(suite.buffer.Add(block))
 
 	// Test retrieving existing block
 	retrieved, ok := suite.buffer.ByID(block.Message.Block.ID())
@@ -173,7 +238,7 @@ func (suite *PendingBlocksSuite) TestPruneByView() {
 		// 10% of the time, add a new unrelated block
 		if i%10 == 0 {
 			block := suite.block()
-			suite.buffer.Add(block)
+			suite.Require().NoError(suite.buffer.Add(block))
 			blocks = append(blocks, block)
 			continue
 		}
@@ -182,7 +247,7 @@ func (suite *PendingBlocksSuite) TestPruneByView() {
 		if i%2 == 1 && len(blocks) > 0 {
 			parent := blocks[rand.Intn(len(blocks))]
 			block := suite.blockWithParent(parent.Message.Block.ToHeader())
-			suite.buffer.Add(block)
+			suite.Require().NoError(suite.buffer.Add(block))
 			blocks = append(blocks, block)
 		}
 	}
@@ -210,13 +275,13 @@ func (suite *PendingBlocksSuite) TestPruneByView() {
 // TestPruneByViewBelowFinalizedView verifies that pruning below finalized view returns an error.
 func (suite *PendingBlocksSuite) TestPruneByViewBelowFinalizedView() {
 	finalizedView := uint64(100)
-	buffer := NewPendingBlocks(finalizedView)
+	buffer := NewPendingBlocks(finalizedView, 100_000)
 
 	// Add some blocks above finalized view
 	parent := unittest.BlockHeaderFixture()
 	parent.View = finalizedView + 10
 	block := suite.blockWithParent(parent)
-	buffer.Add(block)
+	suite.Require().NoError(buffer.Add(block))
 
 	// Prune at finalized view should succeed
 	err := buffer.PruneByView(finalizedView)
@@ -305,7 +370,7 @@ func (suite *PendingBlocksSuite) TestConcurrentAccess() {
 			defer wg.Done()
 			for j := 0; j < blocksPerGoroutine; j++ {
 				block := suite.block()
-				suite.buffer.Add(block)
+				suite.Require().NoError(suite.buffer.Add(block))
 			}
 		}()
 	}