fix(merge): bound dangling block cache growth

src/Nethermind/Nethermind.Merge.Plugin.Test/BlockCacheServiceTest.cs

@@ -0,0 +1,78 @@
+// SPDX-FileCopyrightText: 2022 Demerzel Solutions Limited
+// SPDX-License-Identifier: LGPL-3.0-only
+
+using FluentAssertions;
+using Nethermind.Core;
+using Nethermind.Core.Crypto;
+using Nethermind.Core.Test.Builders;
+using Nethermind.Merge.Plugin.Handlers;
+using NUnit.Framework;
+
+namespace Nethermind.Merge.Plugin.Test;
+
+[TestFixture]
+public class BlockCacheServiceTest
+{
+    [Test]
+    public void prunes_highest_unprotected_block_and_returns_false_when_added_block_is_pruned()
+    {
+        BlockCacheService blockCacheService = new(2);
+        Block block1 = Build.A.Block.WithNumber(1).TestObject;
+        Block block2 = Build.A.Block.WithNumber(2).TestObject;
+        Block block3 = Build.A.Block.WithNumber(3).TestObject;
+        Hash256 block1Hash = block1.GetOrCalculateHash();
+        Hash256 block2Hash = block2.GetOrCalculateHash();
+        Hash256 block3Hash = block3.GetOrCalculateHash();
+
+        blockCacheService.TryAddBlock(block1).Should().BeTrue();
+        blockCacheService.TryAddBlock(block2).Should().BeTrue();
+        blockCacheService.TryAddBlock(block3).Should().BeFalse();
+
+        blockCacheService.BlockCache.Should().HaveCount(2);
+        blockCacheService.BlockCache.ContainsKey(block1Hash).Should().BeTrue();
+        blockCacheService.BlockCache.ContainsKey(block2Hash).Should().BeTrue();
+        blockCacheService.BlockCache.ContainsKey(block3Hash).Should().BeFalse();
+    }
+
+    [Test]
+    public void preserves_head_and_finalized_blocks_when_pruning()
+    {
+        BlockCacheService blockCacheService = new(2);
+        Block finalizedBlock = Build.A.Block.WithNumber(1).TestObject;
+        Block block2 = Build.A.Block.WithNumber(2).TestObject;
+        Block headBlock = Build.A.Block.WithNumber(3).TestObject;
+        Hash256 finalizedHash = finalizedBlock.GetOrCalculateHash();
+        Hash256 block2Hash = block2.GetOrCalculateHash();
+        Hash256 headHash = headBlock.GetOrCalculateHash();
+        blockCacheService.FinalizedHash = finalizedHash;
+        blockCacheService.HeadBlockHash = headHash;
+
+        blockCacheService.TryAddBlock(finalizedBlock).Should().BeTrue();
+        blockCacheService.TryAddBlock(block2).Should().BeTrue();
+        blockCacheService.TryAddBlock(headBlock).Should().BeTrue();
+
+        blockCacheService.BlockCache.Should().HaveCount(2);
+        blockCacheService.BlockCache.ContainsKey(finalizedHash).Should().BeTrue();
+        blockCacheService.BlockCache.ContainsKey(block2Hash).Should().BeFalse();
+        blockCacheService.BlockCache.ContainsKey(headHash).Should().BeTrue();
+    }
+
+    [Test]
+    public void preserves_protected_blocks_when_all_candidates_are_protected()
+    {
+        BlockCacheService blockCacheService = new(1);
+        Block finalizedBlock = Build.A.Block.WithNumber(1).TestObject;
+        Block headBlock = Build.A.Block.WithNumber(2).TestObject;
+        Hash256 finalizedHash = finalizedBlock.GetOrCalculateHash();
+        Hash256 headHash = headBlock.GetOrCalculateHash();
+        blockCacheService.FinalizedHash = finalizedHash;
+        blockCacheService.HeadBlockHash = headHash;
+
+        blockCacheService.TryAddBlock(finalizedBlock).Should().BeTrue();
+        blockCacheService.TryAddBlock(headBlock).Should().BeTrue();
+
+        blockCacheService.BlockCache.Should().HaveCount(2);
+        blockCacheService.BlockCache.ContainsKey(finalizedHash).Should().BeTrue();
+        blockCacheService.BlockCache.ContainsKey(headHash).Should().BeTrue();
+    }
+}

src/Nethermind/Nethermind.Merge.Plugin.Test/InvalidChainTracker/InvalidChainTrackerTest.cs

@@ -181,7 +181,7 @@ public void givenAnInvalidBlock_ifParentIsNotPostMerge_thenLastValidHashShouldBe
         Hash256 invalidBlock = Keccak.Compute("A");
         BlockHeader parentBlockHeader = new BlockHeaderBuilder().TestObject;
 
-        blockCacheService.BlockCache[parentBlockHeader.GetOrCalculateHash()] = new Block(parentBlockHeader);
+        blockCacheService.TryAddBlock(new Block(parentBlockHeader));
 
         IPoSSwitcher poSSwitcher = Substitute.For<IPoSSwitcher>();
         poSSwitcher.IsPostMerge(parentBlockHeader).Returns(false);
@@ -203,8 +203,8 @@ public void givenAnInvalidBlock_WithUnknownParent_thenGetParentFromCache()
         BlockHeader blockHeader = new BlockHeaderBuilder()
             .WithParentHash(parentBlockHeader.GetOrCalculateHash()).TestObject;
 
-        blockCacheService.BlockCache[blockHeader.GetOrCalculateHash()] = new Block(blockHeader);
-        blockCacheService.BlockCache[parentBlockHeader.GetOrCalculateHash()] = new Block(parentBlockHeader);
+        blockCacheService.TryAddBlock(new Block(blockHeader));
+        blockCacheService.TryAddBlock(new Block(parentBlockHeader));
 
         IPoSSwitcher alwaysPos = Substitute.For<IPoSSwitcher>();
         alwaysPos.IsPostMerge(Arg.Any<BlockHeader>()).Returns(true);

src/Nethermind/Nethermind.Merge.Plugin/Handlers/BlockCacheService.cs

@@ -1,15 +1,105 @@
 // SPDX-FileCopyrightText: 2022 Demerzel Solutions Limited
 // SPDX-License-Identifier: LGPL-3.0-only
 
+using System;
 using System.Collections.Concurrent;
+using System.Collections.Generic;
 using Nethermind.Core;
 using Nethermind.Core.Crypto;
+using Nethermind.Crypto;
 
 namespace Nethermind.Merge.Plugin.Handlers;
 
+/// <summary>
+/// Stores payload blocks while their ancestry is resolved during merge sync.
+/// </summary>
 public class BlockCacheService : IBlockCacheService
 {
-    public ConcurrentDictionary<Hash256AsKey, Block> BlockCache { get; } = new();
+    private readonly int _maxCachedBlocks;
+    private readonly ConcurrentDictionary<Hash256AsKey, Block> _blockCache = new();
+
+    /// <summary>
+    /// Initializes a block cache with the default merge sync bound.
+    /// </summary>
+    public BlockCacheService()
+        : this((int)(Reorganization.MaxDepth * 2 + 16))
+    {
+    }
+
+    /// <summary>
+    /// Initializes a block cache with the provided maximum number of cached blocks.
+    /// </summary>
+    /// <param name="maxCachedBlocks">Maximum number of cached blocks before pruning unprotected entries.</param>
+    public BlockCacheService(int maxCachedBlocks)
+    {
+        ArgumentOutOfRangeException.ThrowIfNegativeOrZero(maxCachedBlocks);
+        _maxCachedBlocks = maxCachedBlocks;
+    }
+
+    /// <inheritdoc />
+    public IReadOnlyDictionary<Hash256AsKey, Block> BlockCache => _blockCache;
+
+    /// <inheritdoc />
     public Hash256? FinalizedHash { get; set; }
+
+    /// <inheritdoc />
     public Hash256? HeadBlockHash { get; set; }
+
+    /// <inheritdoc />
+    public bool TryAddBlock(Block block)
+    {
+        Hash256 blockHash = block.GetOrCalculateHash();
+        if (!_blockCache.TryAdd(blockHash, block))
+        {
+            return false;
+        }
+
+        bool blockRemainsCached = true;
+        // The cache bound is small, so pruning scans it instead of maintaining a second ordered index.
+        while (_blockCache.Count > _maxCachedBlocks &&
+               TryGetHighestNumberedUnprotectedBlock(out Hash256AsKey blockHashToRemove))
+        {
+            bool removed = _blockCache.TryRemove(blockHashToRemove, out _);
+            if (removed && Equals(blockHashToRemove.Value, blockHash))
+            {
+                blockRemainsCached = false;
+            }
+        }
+
+        return blockRemainsCached;
+    }
+
+    /// <inheritdoc />
+    public bool TryRemoveBlock(Hash256 blockHash) => _blockCache.TryRemove(blockHash, out _);
+
+    /// <inheritdoc />
+    public void Clear() => _blockCache.Clear();
+
+    private bool TryGetHighestNumberedUnprotectedBlock(out Hash256AsKey blockHash)
+    {
+        Hash256AsKey highestNumberedHash = default;
+        long highestBlockNumber = long.MinValue;
+        bool foundBlock = false;
+
+        foreach (KeyValuePair<Hash256AsKey, Block> cachedBlock in _blockCache)
+        {
+            if (IsProtected(cachedBlock.Key))
+            {
+                continue;
+            }
+
+            if (!foundBlock || cachedBlock.Value.Number > highestBlockNumber)
+            {
+                highestNumberedHash = cachedBlock.Key;
+                highestBlockNumber = cachedBlock.Value.Number;
+                foundBlock = true;
+            }
+        }
+
+        blockHash = highestNumberedHash;
+        return foundBlock;
+    }
+
+    private bool IsProtected(Hash256AsKey blockHash) =>
+        Equals(blockHash.Value, FinalizedHash) || Equals(blockHash.Value, HeadBlockHash);
 }

src/Nethermind/Nethermind.Merge.Plugin/Handlers/IBlockCacheService.cs

@@ -1,15 +1,45 @@
 // SPDX-FileCopyrightText: 2022 Demerzel Solutions Limited
 // SPDX-License-Identifier: LGPL-3.0-only
 
-using System.Collections.Concurrent;
+using System.Collections.Generic;
 using Nethermind.Core;
 using Nethermind.Core.Crypto;
 
 namespace Nethermind.Merge.Plugin.Handlers;
 
 public interface IBlockCacheService
 {
-    public ConcurrentDictionary<Hash256AsKey, Block> BlockCache { get; }
+    /// <summary>
+    /// Blocks cached while payload ancestry is resolved during merge sync.
+    /// </summary>
+    public IReadOnlyDictionary<Hash256AsKey, Block> BlockCache { get; }
+
+    /// <summary>
+    /// Finalized block hash protected from cache pruning.
+    /// </summary>
     Hash256? FinalizedHash { get; set; }
+
+    /// <summary>
+    /// Head block hash protected from cache pruning.
+    /// </summary>
     Hash256? HeadBlockHash { get; set; }
+
+    /// <summary>
+    /// Adds a block to the bounded cache.
+    /// </summary>
+    /// <param name="block">Block to add.</param>
+    /// <returns><see langword="true"/> if the block remains cached; otherwise <see langword="false"/>.</returns>
+    bool TryAddBlock(Block block);
+
+    /// <summary>
+    /// Removes a block from the cache.
+    /// </summary>
+    /// <param name="blockHash">Hash of the block to remove.</param>
+    /// <returns><see langword="true"/> if the block was removed; otherwise <see langword="false"/>.</returns>
+    bool TryRemoveBlock(Hash256 blockHash);
+
+    /// <summary>
+    /// Removes all cached blocks.
+    /// </summary>
+    void Clear();
 }

src/Nethermind/Nethermind.Merge.Plugin/Handlers/NewPayloadHandler.cs

@@ -171,7 +171,7 @@ public async Task<ResultWrapper<PayloadStatusV1>> HandleAsync(ExecutionPayload r
             }
 
             if (_logger.IsInfo) _logger.Info($"Insert block into cache without parent {block}");
-            _blockCacheService.BlockCache.TryAdd(block.Hash!, block);
+            _blockCacheService.TryAddBlock(block);
             return NewPayloadV1Result.Syncing;
         }
 
@@ -478,13 +478,14 @@ private bool TryInsertDanglingBlock(Block block)
             if (current is null)
             {
                 // block not part of beacon pivot chain, save in cache
-                _blockCacheService.BlockCache.TryAdd(block.Hash!, block);
+                _blockCacheService.TryAddBlock(block);
                 return false;
             }
 
             while (stack.TryPop(out Block? child))
             {
                 _blockTree.Insert(child, BlockTreeInsertBlockOptions.SaveHeader, insertHeaderOptions);
+                _blockCacheService.TryRemoveBlock(child.Hash!);
             }
 
             _beaconPivot.ProcessDestination = block.Header;

src/Nethermind/Nethermind.Merge.Plugin/Synchronization/BeaconSync.cs

@@ -38,7 +38,7 @@ public void StopSyncing()
             if (!_isInBeaconModeControl)
             {
                 _beaconPivot.RemoveBeaconPivot();
-                _blockCacheService.BlockCache.Clear();
+                _blockCacheService.Clear();
             }
 
             _isInBeaconModeControl = true;

src/Nethermind/Nethermind.Synchronization.Test/E2ESyncTests.cs

@@ -562,8 +562,8 @@ public async Task SyncUntilFinished(IContainer server, CancellationToken cancell
             IBlockTree otherBlockTree = server.Resolve<IBlockTree>();
             Block finalizedBlock = otherBlockTree.FindBlock(otherBlockTree.Head!.Number - 250)!;
             Block headBlock = otherBlockTree.Head!;
-            blockCacheService.BlockCache.TryAdd(new Hash256AsKey(finalizedBlock.Hash!), finalizedBlock);
-            blockCacheService.BlockCache.TryAdd(new Hash256AsKey(headBlock.Hash!), headBlock);
+            blockCacheService.TryAddBlock(finalizedBlock);
+            blockCacheService.TryAddBlock(headBlock);
             blockCacheService.FinalizedHash = finalizedBlock.Hash!;
 
             await preMergeTestEnv.WaitForSyncMode(mode => mode != SyncMode.UpdatingPivot, cancellationToken);