Merge pull request #145 from ethpandaops/pk910/clear-stale-blocks

cleanup old blocks from cache in pruning routine

indexer/beacon/blockcache.go

@@ -212,6 +212,23 @@ func (cache *blockCache) getPruningBlocks(minInMemorySlot phase0.Slot) []*Block
 	return blocks
 }
 
+// getCleanupBlocks returns the blocks that can be cleaned up based on the given finalized slot.
+func (cache *blockCache) getCleanupBlocks(finalizedSlot phase0.Slot) []*Block {
+	cache.cacheMutex.RLock()
+	defer cache.cacheMutex.RUnlock()
+
+	blocks := []*Block{}
+	for slot, slotBlocks := range cache.slotMap {
+		if slot >= finalizedSlot {
+			continue
+		}
+
+		blocks = append(blocks, slotBlocks...)
+	}
+
+	return blocks
+}
+
 // getForkBlocks returns a slice of blocks that belong to the specified forkId.
 func (cache *blockCache) getForkBlocks(forkId ForkKey) []*Block {
 	cache.cacheMutex.RLock()

indexer/beacon/pruning.go

@@ -369,6 +369,13 @@ func (indexer *Indexer) processCachePruning(prunedEpochStats, prunedEpochStates
 		pruneBlock.block.block = nil
 	}
 
+	// remove all blocks in the finalized block range from the cache
+	finalizedSlot := chainState.EpochToSlot(indexer.lastFinalizedEpoch)
+	cleanupBlocks := indexer.blockCache.getCleanupBlocks(finalizedSlot)
+	for _, block := range cleanupBlocks {
+		indexer.blockCache.removeBlock(block)
+	}
+
 	// clean up epoch stats cache
 	for _, epochStats := range indexer.epochCache.getEpochStatsBeforeEpoch(minInMemoryEpoch) {
 		if epochStats.dependentState != nil {
@@ -379,6 +386,11 @@ func (indexer *Indexer) processCachePruning(prunedEpochStats, prunedEpochStates
 			epochStats.pruneValues()
 			prunedEpochStats++
 		}
+
+		if epochStats.epoch < indexer.lastFinalizedEpoch {
+			// remove from epoch stats cache
+			indexer.epochCache.removeEpochStats(epochStats)
+		}
 	}
 
 	prunedEpochStates += indexer.epochCache.removeUnreferencedEpochStates()