parallelize txpool test + fixes for parallel blockchain tests

src/Nethermind/Nethermind.Blockchain.Test/BlockhashProviderTests.cs

@@ -438,6 +438,7 @@ public void BlockhashStore_uses_custom_ring_buffer_size()
     }
 
     [Test, MaxTime(Timeout.MaxTestTime)]
+    [NonParallelizable]
     public async Task Prefetches_come_in_wrong_order()
     {
         const int chainLength = 261;

src/Nethermind/Nethermind.Blockchain.Test/Builders/FilterBuilder.cs

@@ -11,14 +11,15 @@ namespace Nethermind.Blockchain.Test.Builders
 {
     public class FilterBuilder
     {
-        private static int _id;
+        private int _id;
         private BlockParameter _fromBlock = new(BlockParameterType.Latest);
         private BlockParameter _toBlock = new(BlockParameterType.Latest);
         private AddressFilter _address = AddressFilter.AnyAddress;
         private SequenceTopicsFilter _topicsFilter = new();
 
-        private FilterBuilder()
+        private FilterBuilder(int id)
         {
+            _id = id;
         }
 
         public static FilterBuilder New()
@@ -29,9 +30,9 @@ public static FilterBuilder New()
 
         public static FilterBuilder New(ref int currentFilterIndex)
         {
-            _id = currentFilterIndex;
+            int id = currentFilterIndex;
             currentFilterIndex++;
-            return new FilterBuilder();
+            return new FilterBuilder(id);
         }
 
         public FilterBuilder WithId(int id)

src/Nethermind/Nethermind.Blockchain.Test/Filters/FilterManagerTests.cs

@@ -21,11 +21,11 @@
 
 namespace Nethermind.Blockchain.Test.Filters;
 
+[Parallelizable(ParallelScope.None)]
 public class FilterManagerTests
 {
     private FilterStore _filterStore = null!;
-    private IBranchProcessor _branchProcessor = null!;
-    private IMainProcessingContext _mainProcessingContext = null!;
+    private TestMainProcessingContext _mainProcessingContext = null!;
     private ITxPool _txPool = null!;
     private ILogManager _logManager = null!;
     private FilterManager _filterManager = null!;
@@ -36,10 +36,8 @@ public class FilterManagerTests
     public void Setup()
     {
         _currentFilterId = 0;
-        _filterStore = new FilterStore(new TimerFactory(), 20, 10);
-        _branchProcessor = Substitute.For<IBranchProcessor>();
-        _mainProcessingContext = Substitute.For<IMainProcessingContext>();
-        _mainProcessingContext.BranchProcessor.Returns(_branchProcessor);
+        _filterStore = new FilterStore(new TimerFactory(), 400, 100);
+        _mainProcessingContext = new TestMainProcessingContext();
         _txPool = Substitute.For<ITxPool>();
         _logManager = LimboLogs.Instance;
     }
@@ -55,7 +53,7 @@ public async Task removing_filter_removes_data()
     {
         LogsShouldNotBeEmpty(static _ => { }, static _ => { });
         _filterManager.GetLogs(0).Should().NotBeEmpty();
-        await Task.Delay(60);
+        await Task.Delay(600);
         _filterManager.GetLogs(0).Should().BeEmpty();
     }
 
@@ -257,6 +255,7 @@ public void logs_should_be_empty_for_existing_block_and_addresses_and_non_existi
     [Test, MaxTime(Timeout.MaxTestTime)]
     [TestCase(1, 1)]
     [TestCase(5, 3)]
+    [NonParallelizable]
     public void logs_should_have_correct_log_indexes(int filtersCount, int logsPerTx)
     {
         const int txCount = 10;
@@ -330,12 +329,12 @@ private void Assert(IEnumerable<Action<FilterBuilder>> filterBuilders,
         _filterStore.SaveFilters(filters.OfType<BlockFilter>());
         _filterManager = new FilterManager(_filterStore, _mainProcessingContext, _txPool, _logManager);
 
-        _branchProcessor.BlockProcessed += Raise.EventWith(_branchProcessor, new BlockProcessedEventArgs(block, []));
+        _mainProcessingContext.TestBranchProcessor.RaiseBlockProcessed(new BlockProcessedEventArgs(block, []));
 
         int index = 1;
         foreach (TxReceipt receipt in receipts)
         {
-            _mainProcessingContext.TransactionProcessed += Raise.EventWith(_branchProcessor,
+            _mainProcessingContext.RaiseTransactionProcessed(
                 new TxProcessedEventArgs(index, Build.A.Transaction.TestObject, block.Header, receipt));
             index++;
         }

src/Nethermind/Nethermind.Blockchain.Test/Filters/TestProcessingContext.cs

@@ -0,0 +1,55 @@
+// SPDX-FileCopyrightText: 2025 Demerzel Solutions Limited
+// SPDX-License-Identifier: LGPL-3.0-only
+
+#nullable enable
+
+using System;
+using System.Collections.Generic;
+using System.Threading;
+using Nethermind.Consensus.Processing;
+using Nethermind.Core;
+using Nethermind.Evm.State;
+using Nethermind.Evm.Tracing;
+using Nethermind.Evm.TransactionProcessing;
+
+namespace Nethermind.Blockchain.Test.Filters;
+
+/// <summary>
+/// Test implementation of IBranchProcessor that allows manual event raising.
+/// </summary>
+internal class TestBranchProcessor : IBranchProcessor
+{
+    public event EventHandler<BlockProcessedEventArgs>? BlockProcessed;
+    public event EventHandler<BlocksProcessingEventArgs>? BlocksProcessing { add { } remove { } }
+    public event EventHandler<BlockEventArgs>? BlockProcessing { add { } remove { } }
+
+    public Block[] Process(BlockHeader? baseBlock, IReadOnlyList<Block> suggestedBlocks,
+        ProcessingOptions processingOptions, IBlockTracer blockTracer, CancellationToken token = default)
+        => [];
+
+    public void RaiseBlockProcessed(BlockProcessedEventArgs args)
+        => BlockProcessed?.Invoke(this, args);
+}
+
+/// <summary>
+/// Test implementation of IMainProcessingContext that allows manual event raising.
+/// </summary>
+internal class TestMainProcessingContext : IMainProcessingContext
+{
+    private readonly TestBranchProcessor _branchProcessor = new();
+
+    public ITransactionProcessor TransactionProcessor => null!;
+    public IBranchProcessor BranchProcessor => _branchProcessor;
+    public IBlockProcessor BlockProcessor => null!;
+    public IBlockchainProcessor BlockchainProcessor => null!;
+    public IBlockProcessingQueue BlockProcessingQueue => null!;
+    public IWorldState WorldState => null!;
+    public IGenesisLoader GenesisLoader => null!;
+
+    public event EventHandler<TxProcessedEventArgs>? TransactionProcessed;
+
+    public TestBranchProcessor TestBranchProcessor => _branchProcessor;
+
+    public void RaiseTransactionProcessed(TxProcessedEventArgs args)
+        => TransactionProcessed?.Invoke(this, args);
+}

src/Nethermind/Nethermind.Blockchain.Test/Find/LogFinderTests.cs

@@ -318,6 +318,7 @@ public void complex_filter(LogFilter filter, int expectedCount, bool withBloomDb
     }
 
     [Test, MaxTime(Timeout.MaxTestTime)]
+    [NonParallelizable]
     public async Task Throw_log_finder_operation_canceled_after_given_timeout([Values(2, 0.01)] double waitTime)
     {
         var timeout = TimeSpan.FromMilliseconds(Timeout.MaxWaitTime);
@@ -335,7 +336,7 @@ public async Task Throw_log_finder_operation_canceled_after_given_timeout([Value
 
         if (waitTime > 1)
         {
-            action.Should().Throw<AggregateException>().WithInnerException<OperationCanceledException>();
+            action.Should().Throw<OperationCanceledException>();
         }
         else
         {

src/Nethermind/Nethermind.Network/P2P/ProtocolHandlers/SyncPeerProtocolHandlerBase.cs

@@ -20,7 +20,6 @@
 using Nethermind.Network.P2P.Subprotocols.Eth.V62;
 using Nethermind.Network.P2P.Subprotocols.Eth.V62.Messages;
 using Nethermind.Network.P2P.Subprotocols.Eth.V63.Messages;
-using Nethermind.Serialization.Rlp;
 using Nethermind.Stats;
 using Nethermind.Stats.Model;
 using Nethermind.Synchronization;
@@ -50,7 +49,6 @@ public abstract class SyncPeerProtocolHandlerBase : ZeroProtocolHandlerBase, ISy
 
         protected Hash256 _remoteHeadBlockHash;
         protected readonly ITimestamper _timestamper;
-        protected readonly TxDecoder _txDecoder;
 
         protected readonly MessageQueue<GetBlockHeadersMessage, IOwnedReadOnlyList<BlockHeader?>> _headersRequests;
         protected readonly MessageQueue<GetBlockBodiesMessage, (OwnedBlockBodies, long)> _bodiesRequests;
@@ -67,7 +65,6 @@ protected SyncPeerProtocolHandlerBase(ISession session,
         {
             SyncServer = syncServer ?? throw new ArgumentNullException(nameof(syncServer));
             _timestamper = Timestamper.Default;
-            _txDecoder = TxDecoder.Instance;
             _headersRequests = new MessageQueue<GetBlockHeadersMessage, IOwnedReadOnlyList<BlockHeader>>(Send);
             _bodiesRequests = new MessageQueue<GetBlockBodiesMessage, (OwnedBlockBodies, long)>(Send);
         }

src/Nethermind/Nethermind.Shutter.Test/ShutterIntegrationTests.cs

@@ -115,7 +115,9 @@ public async Task Can_increment_metric_on_missed_keys()
             time += (long)ShutterTestsCommon.SlotLength.TotalSeconds;
         }
 
-        Assert.That(Metrics.ShutterKeysMissed, Is.EqualTo(5));
+        // ImproveBlock tasks run in the background and may not have completed yet
+        // when GetPayload returns (it only waits 50ms), so poll until all increments land.
+        Assert.That(() => Metrics.ShutterKeysMissed, Is.EqualTo((ulong)5).After(5000, 50));
     }
 
 }

src/Nethermind/Nethermind.Synchronization.Test/FastSync/StateSyncFeedHealingTests.cs

@@ -19,6 +19,7 @@
 namespace Nethermind.Synchronization.Test.FastSync;
 
 [Parallelizable(ParallelScope.All)]
+[FixtureLifeCycle(LifeCycle.InstancePerTestCase)]
 public class StateSyncFeedHealingTests : StateSyncFeedTestsBase
 {
     [Test]

src/Nethermind/Nethermind.Trie.Test/Pruning/TreeStoreTests.cs

@@ -897,6 +897,7 @@ public void After_commit_should_have_has_root()
 
         [Test]
         [Retry(3)]
+        [NonParallelizable]
         public async Task Will_RemovePastKeys_OnSnapshot()
         {
             MemDb memDb = new();
@@ -936,6 +937,8 @@ public async Task Will_RemovePastKeys_OnSnapshot()
         }
 
         [Test]
+        [Retry(3)]
+        [NonParallelizable]
         public async Task Will_Trigger_ReorgBoundaryEvent_On_Prune()
         {
             // TODO: Check why slow

src/Nethermind/Nethermind.Trie.Test/TrieTests.cs

@@ -28,6 +28,7 @@ namespace Nethermind.Trie.Test
 {
     [TestFixture]
     [Parallelizable(ParallelScope.All)]
+    [FixtureLifeCycle(LifeCycle.InstancePerTestCase)]
     public class TrieTests
     {
         private ILogger _logger;
@@ -1082,6 +1083,8 @@ public void Fuzz_accounts_with_reorganizations(
         }
 
         [TestCaseSource(nameof(FuzzAccountsWithStorageScenarios))]
+        [Retry(3)]
+        [NonParallelizable]
         public void Fuzz_accounts_with_storage(
             (TrieStoreConfigurations trieStoreConfigurations,
                 int accountsCount,

src/Nethermind/Nethermind.TxPool.Test/BlobTxStorageTests.cs

@@ -10,6 +10,7 @@
 namespace Nethermind.TxPool.Test;
 
 [TestFixture]
+[Parallelizable(ParallelScope.All)]
 public class BlobTxStorageTests
 {
     [Test]

src/Nethermind/Nethermind.TxPool.Test/Collections/SortedPoolTests.cs

@@ -20,6 +20,8 @@
 namespace Nethermind.TxPool.Test.Collections
 {
     [TestFixture]
+    [Parallelizable(ParallelScope.All)]
+    [FixtureLifeCycle(LifeCycle.InstancePerTestCase)]
     public class SortedPoolTests
     {
         private const int Capacity = 16;

src/Nethermind/Nethermind.TxPool.Test/CompetingTransactionEqualityComparerTests.cs

@@ -9,6 +9,7 @@
 
 namespace Nethermind.TxPool.Test
 {
+    [Parallelizable(ParallelScope.All)]
     public class CompetingTransactionEqualityComparerTests
     {
         public static IEnumerable TestCases

src/Nethermind/Nethermind.TxPool.Test/DelegatedAccountFilterTest.cs

@@ -19,14 +19,15 @@
 
 namespace Nethermind.TxPool.Test;
 
+[Parallelizable(ParallelScope.All)]
 internal class DelegatedAccountFilterTest
 {
     [Test]
     public void Accept_SenderIsNotDelegated_ReturnsAccepted()
     {
         IChainHeadSpecProvider headInfoProvider = Substitute.For<IChainHeadSpecProvider>();
         headInfoProvider.GetCurrentHeadSpec().Returns(Prague.Instance);
-        TxDistinctSortedPool standardPool = new TxDistinctSortedPool(MemoryAllowance.MemPoolSize, Substitute.For<IComparer<Transaction>>(), NullLogManager.Instance);
+        TxDistinctSortedPool standardPool = new TxDistinctSortedPool(new TxPoolConfig().Size, Substitute.For<IComparer<Transaction>>(), NullLogManager.Instance);
         TxDistinctSortedPool blobPool = new BlobTxDistinctSortedPool(10, Substitute.For<IComparer<Transaction>>(), NullLogManager.Instance);
         DelegatedAccountFilter filter = new(headInfoProvider, standardPool, blobPool, Substitute.For<IReadOnlyStateProvider>(), new DelegationCache());
         Transaction transaction = Build.A.Transaction.SignedAndResolved(new EthereumEcdsa(0), TestItem.PrivateKeyA).TestObject;
@@ -46,7 +47,7 @@ public void Accept_SenderIsDelegatedWithNoTransactionsInPool_ReturnsAccepted()
         stateProvider.InsertCode(code, TestItem.AddressA);
         IChainHeadSpecProvider headInfoProvider = Substitute.For<IChainHeadSpecProvider>();
         headInfoProvider.GetCurrentHeadSpec().Returns(Prague.Instance);
-        TxDistinctSortedPool standardPool = new TxDistinctSortedPool(MemoryAllowance.MemPoolSize, Substitute.For<IComparer<Transaction>>(), NullLogManager.Instance);
+        TxDistinctSortedPool standardPool = new TxDistinctSortedPool(new TxPoolConfig().Size, Substitute.For<IComparer<Transaction>>(), NullLogManager.Instance);
         TxDistinctSortedPool blobPool = new BlobTxDistinctSortedPool(10, Substitute.For<IComparer<Transaction>>(), NullLogManager.Instance);
         DelegatedAccountFilter filter = new(headInfoProvider, standardPool, blobPool, stateProvider, new DelegationCache());
         Transaction transaction = Build.A.Transaction.SignedAndResolved(new EthereumEcdsa(0), TestItem.PrivateKeyA).TestObject;
@@ -62,7 +63,7 @@ public void Accept_SenderIsDelegatedWithOneTransactionInPoolWithSameNonce_Return
     {
         IChainHeadSpecProvider headInfoProvider = Substitute.For<IChainHeadSpecProvider>();
         headInfoProvider.GetCurrentHeadSpec().Returns(Prague.Instance);
-        TxDistinctSortedPool standardPool = new TxDistinctSortedPool(MemoryAllowance.MemPoolSize, Substitute.For<IComparer<Transaction>>(), NullLogManager.Instance);
+        TxDistinctSortedPool standardPool = new TxDistinctSortedPool(new TxPoolConfig().Size, Substitute.For<IComparer<Transaction>>(), NullLogManager.Instance);
         TxDistinctSortedPool blobPool = new BlobTxDistinctSortedPool(10, Substitute.For<IComparer<Transaction>>(), NullLogManager.Instance);
         Transaction inPool = Build.A.Transaction.SignedAndResolved(new EthereumEcdsa(0), TestItem.PrivateKeyA).TestObject;
         standardPool.TryInsert(inPool.Hash, inPool);
@@ -84,7 +85,7 @@ public void Accept_SenderIsDelegatedWithOneTransactionInPoolWithDifferentNonce_R
     {
         IChainHeadSpecProvider headInfoProvider = Substitute.For<IChainHeadSpecProvider>();
         headInfoProvider.GetCurrentHeadSpec().Returns(Prague.Instance);
-        TxDistinctSortedPool standardPool = new TxDistinctSortedPool(MemoryAllowance.MemPoolSize, Substitute.For<IComparer<Transaction>>(), NullLogManager.Instance);
+        TxDistinctSortedPool standardPool = new TxDistinctSortedPool(new TxPoolConfig().Size, Substitute.For<IComparer<Transaction>>(), NullLogManager.Instance);
         TxDistinctSortedPool blobPool = new BlobTxDistinctSortedPool(10, Substitute.For<IComparer<Transaction>>(), NullLogManager.Instance);
         Transaction inPool = Build.A.Transaction.SignedAndResolved(new EthereumEcdsa(0), TestItem.PrivateKeyA).TestObject;
         standardPool.TryInsert(inPool.Hash, inPool);
@@ -111,7 +112,7 @@ public void Accept_Eip7702IsNotActivated_ReturnsExpected(bool isActive, AcceptTx
     {
         IChainHeadSpecProvider headInfoProvider = Substitute.For<IChainHeadSpecProvider>();
         headInfoProvider.GetCurrentHeadSpec().Returns(isActive ? Prague.Instance : Cancun.Instance);
-        TxDistinctSortedPool standardPool = new TxDistinctSortedPool(MemoryAllowance.MemPoolSize, Substitute.For<IComparer<Transaction>>(), NullLogManager.Instance);
+        TxDistinctSortedPool standardPool = new TxDistinctSortedPool(new TxPoolConfig().Size, Substitute.For<IComparer<Transaction>>(), NullLogManager.Instance);
         TxDistinctSortedPool blobPool = new BlobTxDistinctSortedPool(10, Substitute.For<IComparer<Transaction>>(), NullLogManager.Instance);
         Transaction inPool = Build.A.Transaction.WithNonce(0).SignedAndResolved(new EthereumEcdsa(0), TestItem.PrivateKeyA).TestObject;
         standardPool.TryInsert(inPool.Hash, inPool);
@@ -140,7 +141,7 @@ public void Accept_SenderHasPendingDelegation_OnlyAcceptsIfNonceIsExactMatch(int
     {
         IChainHeadSpecProvider headInfoProvider = Substitute.For<IChainHeadSpecProvider>();
         headInfoProvider.GetCurrentHeadSpec().Returns(Prague.Instance);
-        TxDistinctSortedPool standardPool = new TxDistinctSortedPool(MemoryAllowance.MemPoolSize, Substitute.For<IComparer<Transaction>>(), NullLogManager.Instance);
+        TxDistinctSortedPool standardPool = new TxDistinctSortedPool(new TxPoolConfig().Size, Substitute.For<IComparer<Transaction>>(), NullLogManager.Instance);
         TxDistinctSortedPool blobPool = new BlobTxDistinctSortedPool(10, Substitute.For<IComparer<Transaction>>(), NullLogManager.Instance);
         DelegationCache pendingDelegations = new();
         pendingDelegations.IncrementDelegationCount(TestItem.AddressA);
@@ -161,7 +162,7 @@ public void Accept_AuthorityHasPendingTransaction_ReturnsDelegatorHasPendingTx(b
     {
         IChainHeadSpecProvider headInfoProvider = Substitute.For<IChainHeadSpecProvider>();
         headInfoProvider.GetCurrentHeadSpec().Returns(Prague.Instance);
-        TxDistinctSortedPool standardPool = new TxDistinctSortedPool(MemoryAllowance.MemPoolSize, Substitute.For<IComparer<Transaction>>(), NullLogManager.Instance);
+        TxDistinctSortedPool standardPool = new TxDistinctSortedPool(new TxPoolConfig().Size, Substitute.For<IComparer<Transaction>>(), NullLogManager.Instance);
         TxDistinctSortedPool blobPool = new BlobTxDistinctSortedPool(10, Substitute.For<IComparer<Transaction>>(), NullLogManager.Instance);
         DelegatedAccountFilter filter = new(headInfoProvider, standardPool, blobPool, Substitute.For<IReadOnlyStateProvider>(), new());
         Transaction transaction;
@@ -200,7 +201,7 @@ public void Accept_SetCodeTxHasAuthorityWithPendingTx_ReturnsDelegatorHasPending
     {
         IChainHeadSpecProvider headInfoProvider = Substitute.For<IChainHeadSpecProvider>();
         headInfoProvider.GetCurrentHeadSpec().Returns(Prague.Instance);
-        TxDistinctSortedPool standardPool = new TxDistinctSortedPool(MemoryAllowance.MemPoolSize, Substitute.For<IComparer<Transaction>>(), NullLogManager.Instance);
+        TxDistinctSortedPool standardPool = new TxDistinctSortedPool(new TxPoolConfig().Size, Substitute.For<IComparer<Transaction>>(), NullLogManager.Instance);
         TxDistinctSortedPool blobPool = new BlobTxDistinctSortedPool(10, Substitute.For<IComparer<Transaction>>(), NullLogManager.Instance);
         DelegationCache pendingDelegations = new();
         pendingDelegations.IncrementDelegationCount(TestItem.AddressA);

src/Nethermind/Nethermind.TxPool.Test/NonceManagerTests.cs

@@ -19,6 +19,8 @@
 
 namespace Nethermind.TxPool.Test;
 
+[Parallelizable(ParallelScope.All)]
+[FixtureLifeCycle(LifeCycle.InstancePerTestCase)]
 public class NonceManagerTests
 {
     private ISpecProvider _specProvider;

src/Nethermind/Nethermind.TxPool.Test/ReceiptStorageTests.cs

@@ -19,6 +19,8 @@ namespace Nethermind.TxPool.Test
 {
     [TestFixture(true)]
     [TestFixture(false)]
+    [Parallelizable(ParallelScope.All)]
+    [FixtureLifeCycle(LifeCycle.InstancePerTestCase)]
     public class ReceiptStorageTests
     {
         private readonly bool _useEip2718;