Move txPool work from Processing loop event to txPool thread

src/Nethermind/Nethermind.Core.Test/Blockchain/TestBlockchain.cs

@@ -422,10 +422,13 @@ private async Task<AcceptTxResult[]> AddBlockInternal(params Transaction[] trans
         BlockTree.BlockAddedToMain -= BlockAddedToMain;
         BlockTree.BlockAddedToMain += BlockAddedToMain;
 
-        await WaitAsync(_oneAtATime, "Multiple block produced at once.");
+        await WaitAsync(_oneAtATime, "Multiple block produced at once.").ConfigureAwait(false);
         AcceptTxResult[] txResults = transactions.Select(t => TxPool.SubmitTx(t, TxHandlingOptions.None)).ToArray();
         Timestamper.Add(TimeSpan.FromSeconds(1));
-        await BlockProductionTrigger.BuildBlock();
+        var headProcessed = new SemaphoreSlim(0);
+        TxPool.TxPoolHeadChanged += (s, a) => headProcessed.Release();
+        await BlockProductionTrigger.BuildBlock().ConfigureAwait(false);
+        await headProcessed.WaitAsync().ConfigureAwait(false);
         return txResults;
     }
 

src/Nethermind/Nethermind.JsonRpc.Test/JsonRpcSocketsClientTests.cs

@@ -125,7 +125,7 @@ static async Task<int> CountNumberOfMessages(Socket socket, CancellationToken to
                 {
                     using JsonRpcResult result = JsonRpcResult.Single(RandomSuccessResponse(1_000, () => disposeCount++), default);
                     await client.SendJsonRpcResult(result);
-                    await Task.Delay(100);
+                    await Task.Delay(10);
                 }
 
                 disposeCount.Should().Be(messageCount);
@@ -220,7 +220,7 @@ async Task<int> ReadMessages(Socket socket, IList<byte[]> receivedMessages, Canc
                     await stream.WriteEndOfMessageAsync();
                     if (i % 10 == 0)
                     {
-                        await Task.Delay(100);
+                        await Task.Delay(10);
                     }
                 }
                 stream.Close();

src/Nethermind/Nethermind.TxPool.Test/TxPoolTests.cs

@@ -213,7 +213,12 @@ public void should_not_ignore_insufficient_funds_for_eip1559_transactions()
             txPool.GetPendingTransactionsCount().Should().Be(0);
             result.Should().Be(AcceptTxResult.InsufficientFunds);
             EnsureSenderBalance(tx.SenderAddress, tx.Value);
+
+            var headProcessed = new ManualResetEventSlim(false);
+            txPool.TxPoolHeadChanged += (s, a) => headProcessed.Set();
             _blockTree.BlockAddedToMain += Raise.EventWith(_blockTree, new BlockReplacementEventArgs(Build.A.Block.WithGasLimit(10000000).TestObject));
+
+            headProcessed.Wait();
             result = txPool.SubmitTx(tx, TxHandlingOptions.PersistentBroadcast);
             result.Should().Be(AcceptTxResult.InsufficientFunds);
             txPool.GetPendingTransactionsCount().Should().Be(0);
@@ -719,7 +724,12 @@ public void should_remove_txHash_from_hashCache_when_tx_removed_because_of_txPoo
             EnsureSenderBalance(higherPriorityTx);
             _txPool.SubmitTx(higherPriorityTx, TxHandlingOptions.PersistentBroadcast);
 
+            var headProcessed = new ManualResetEventSlim(false);
+            _txPool.TxPoolHeadChanged += (s, a) => headProcessed.Set();
+
             _blockTree.BlockAddedToMain += Raise.EventWith(new BlockReplacementEventArgs(Build.A.Block.TestObject));
+
+            headProcessed.Wait();
             _txPool.IsKnown(higherPriorityTx.Hash).Should().BeTrue();
             _txPool.IsKnown(transaction.Hash).Should().BeFalse();
         }

src/Nethermind/Nethermind.TxPool/ITxPool.cs

@@ -16,6 +16,8 @@ public interface ITxPool
         int GetPendingBlobTransactionsCount();
         Transaction[] GetPendingTransactions();
 
+        public event EventHandler<Block>? TxPoolHeadChanged;
+
         /// <summary>
         /// Non-blob txs grouped by sender address, sorted by nonce and later tx pool sorting
         /// </summary>

src/Nethermind/Nethermind.TxPool/NullTxPool.cs

@@ -16,6 +16,12 @@ private NullTxPool() { }
 
         public static NullTxPool Instance { get; } = new();
 
+        public event EventHandler<Block>? TxPoolHeadChanged
+        {
+            add { }
+            remove { }
+        }
+
         public int GetPendingTransactionsCount() => 0;
         public int GetPendingBlobTransactionsCount() => 0;
         public Transaction[] GetPendingTransactions() => Array.Empty<Transaction>();

src/Nethermind/Nethermind.TxPool/TxBroadcaster.cs

@@ -142,7 +142,7 @@ public void AnnounceOnce(ITxPoolPeer peer, Transaction[] txs)
             }
         }
 
-        public void OnNewHead()
+        public void OnNewHead(object? sender, Block block)
         {
             _baseFeeThreshold = CalculateBaseFeeThreshold();
             BroadcastPersistentTxs();

src/Nethermind/Nethermind.TxPool/TxPool.cs

@@ -59,6 +59,8 @@ public class TxPool : ITxPool, IDisposable
         private readonly UpdateGroupDelegate _updateBucket;
         private readonly UpdateGroupDelegate _updateBucketAdded;
 
+        public event EventHandler<Block>? TxPoolHeadChanged;
+
         /// <summary>
         /// Indexes transactions
         /// </summary>
@@ -112,6 +114,7 @@ public TxPool(IEthereumEcdsa ecdsa,
             _updateBucketAdded = UpdateBucketWithAddedTransaction;
 
             _broadcaster = new TxBroadcaster(comparer, TimerFactory.Default, txPoolConfig, chainHeadInfoProvider, logManager, transactionsGossipPolicy);
+            TxPoolHeadChanged += _broadcaster.OnNewHead;
 
             _transactions = new TxDistinctSortedPool(MemoryAllowance.MemPoolSize, comparer, logManager);
             _blobTransactions = txPoolConfig.BlobsSupport.IsPersistentStorage()
@@ -164,7 +167,7 @@ public TxPool(IEthereumEcdsa ecdsa,
             ProcessNewHeads();
         }
 
-        public Transaction[] GetPendingTransactions() => _transactions.GetSnapshot();
+        public Transaction[] GetPendingTransactions() => _transactionSnapshot ??= _transactions.GetSnapshot();
 
         public int GetPendingTransactionsCount() => _transactions.Count;
 
@@ -191,10 +194,6 @@ private void OnHeadChange(object? sender, BlockReplacementEventArgs e)
         {
             try
             {
-                // Clear snapshot
-                _transactionSnapshot = null;
-                _blobTransactionSnapshot = null;
-                _hashCache.ClearCurrentBlockCache();
                 _headBlocksChannel.Writer.TryWrite(e);
             }
             catch (Exception exception)
@@ -211,8 +210,12 @@ private void ProcessNewHeads()
             {
                 while (await _headBlocksChannel.Reader.WaitToReadAsync())
                 {
+                    _hashCache.ClearCurrentBlockCache();
                     while (_headBlocksChannel.Reader.TryRead(out BlockReplacementEventArgs? args))
                     {
+                        // Clear snapshot
+                        _transactionSnapshot = null;
+                        _blobTransactionSnapshot = null;
                         try
                         {
                             ArrayPoolList<AddressAsKey>? accountChanges = args.Block.AccountChanges;
@@ -235,7 +238,7 @@ private void ProcessNewHeads()
                             ReAddReorganisedTransactions(args.PreviousBlock);
                             RemoveProcessedTransactions(args.Block);
                             UpdateBuckets();
-                            _broadcaster.OnNewHead();
+                            TxPoolHeadChanged?.Invoke(this, args.Block);
                             Metrics.TransactionCount = _transactions.Count;
                             Metrics.BlobTransactionCount = _blobTransactions.Count;
                         }
@@ -750,6 +753,7 @@ public UInt256 GetLatestPendingNonce(Address address)
         public void Dispose()
         {
             _timer?.Dispose();
+            TxPoolHeadChanged -= _broadcaster.OnNewHead;
             _broadcaster.Dispose();
             _headInfo.HeadChanged -= OnHeadChange;
             _headBlocksChannel.Writer.Complete();