Broadcast best persistent txs instead of random

src/Nethermind/Nethermind.TxPool.Test/TxBroadcasterTests.cs

@@ -0,0 +1,106 @@
+//  Copyright (c) 2022 Demerzel Solutions Limited
+//  This file is part of the Nethermind library.
+// 
+//  The Nethermind library is free software: you can redistribute it and/or modify
+//  it under the terms of the GNU Lesser General Public License as published by
+//  the Free Software Foundation, either version 3 of the License, or
+//  (at your option) any later version.
+// 
+//  The Nethermind library is distributed in the hope that it will be useful,
+//  but WITHOUT ANY WARRANTY; without even the implied warranty of
+//  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+//  GNU Lesser General Public License for more details.
+// 
+//  You should have received a copy of the GNU Lesser General Public License
+//  along with the Nethermind. If not, see <http://www.gnu.org/licenses/>.
+// 
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Runtime.CompilerServices;
+using FluentAssertions;
+using Nethermind.Blockchain;
+using Nethermind.Blockchain.Comparers;
+using Nethermind.Core;
+using Nethermind.Core.Extensions;
+using Nethermind.Core.Specs;
+using Nethermind.Core.Test.Builders;
+using Nethermind.Core.Timers;
+using Nethermind.Crypto;
+using Nethermind.Logging;
+using Nethermind.Specs;
+using NSubstitute;
+using NUnit.Framework;
+
+[assembly: InternalsVisibleTo("Nethermind.Blockchain.Test")]
+
+namespace Nethermind.TxPool.Test;
+
+[TestFixture]
+public class TxBroadcasterTests
+{
+    private ILogManager _logManager;
+    private ISpecProvider _specProvider;
+    private IBlockTree _blockTree;
+    private IComparer<Transaction> _comparer;
+    private TxBroadcaster _broadcaster;
+    private EthereumEcdsa _ethereumEcdsa;
+    private TxPoolConfig _txPoolConfig;
+
+    [SetUp]
+    public void Setup()
+    {
+        _logManager = LimboLogs.Instance;
+        _specProvider = RopstenSpecProvider.Instance;
+        _ethereumEcdsa = new EthereumEcdsa(_specProvider.ChainId, _logManager);
+        _blockTree = Substitute.For<IBlockTree>();
+        _comparer = new TransactionComparerProvider(_specProvider, _blockTree).GetDefaultComparer();
+        _txPoolConfig = new TxPoolConfig();
+    }
+
+    [TestCase(0)]
+    [TestCase(1)]
+    [TestCase(2)]
+    [TestCase(99)]
+    [TestCase(100)]
+    [TestCase(101)]
+    [TestCase(1000)]
+    [TestCase(-10)]
+    public void should_pick_best_persistent_txs_to_broadcast(int threshold)
+    {
+        _txPoolConfig = new TxPoolConfig() { PeerNotificationThreshold = threshold };
+        _broadcaster = new TxBroadcaster(_comparer, TimerFactory.Default, _txPoolConfig, _logManager);
+
+        int addedTxsCount = TestItem.PrivateKeys.Length;
+        Transaction[] transactions = new Transaction[addedTxsCount];
+
+        for (int i = 0; i < addedTxsCount; i++)
+        {
+            transactions[i] = Build.A.Transaction
+                .WithSenderAddress(TestItem.PrivateKeys[i].Address)
+                .WithGasPrice(i.GWei())
+                .SignedAndResolved(_ethereumEcdsa, TestItem.PrivateKeys[i])
+                .TestObject;
+
+            _broadcaster.StartBroadcast(transactions[i]);
+        }
+
+        _broadcaster.GetSnapshot().Length.Should().Be(addedTxsCount);
+
+        ITxPoolPeer txPoolPeer = Substitute.For<ITxPoolPeer>();
+        List<Transaction> pickedTxs = _broadcaster.GetTxsToSend(txPoolPeer, ArraySegment<Transaction>.Empty).ToList();
+
+        int expectedCount = threshold <= 0 ? 0 : Math.Min(addedTxsCount * threshold / 100 + 1, addedTxsCount);
+        pickedTxs.Count.Should().Be(expectedCount);
+
+        List<Transaction> expectedTxs = new();
+
+        for (int i = 1; i <= expectedCount; i++)
+        {
+            expectedTxs.Add(transactions[addedTxsCount - i]);
+        }
+
+        expectedTxs.Should().BeEquivalentTo(pickedTxs);
+    }
+}

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

@@ -701,6 +701,83 @@ public void should_not_broadcast_own_transactions_that_faded_out_and_came_back()
             Assert.AreEqual(0, _txPool.GetOwnPendingTransactions().Length);
         }
 
+        [TestCase(1, 0)]
+        [TestCase(2, 0)]
+        [TestCase(2, 1)]
+        [TestCase(10, 0)]
+        [TestCase(10, 1)]
+        [TestCase(10, 5)]
+        [TestCase(10, 8)]
+        [TestCase(10, 9)]
+        public void should_remove_stale_txs_from_persistent_transactions(int numberOfTxs, int nonceIncludedInBlock)
+        {
+            _txPool = CreatePool();
+
+            Transaction[] transactions = new Transaction[numberOfTxs];
+            EnsureSenderBalance(TestItem.AddressA, UInt256.MaxValue);
+
+            for (int i = 0; i < numberOfTxs; i++)
+            {
+                transactions[i] = Build.A.Transaction
+                    .WithNonce((UInt256)i)
+                    .WithGasLimit(GasCostOf.Transaction)
+                    .WithGasPrice(10.GWei())
+                    .SignedAndResolved(_ethereumEcdsa, TestItem.PrivateKeyA)
+                    .TestObject;
+                _txPool.SubmitTx(transactions[i], TxHandlingOptions.PersistentBroadcast);
+            }
+            _txPool.GetOwnPendingTransactions().Length.Should().Be(numberOfTxs);
+
+            Block block = Build.A.Block.WithTransactions(transactions[nonceIncludedInBlock]).TestObject;
+            BlockReplacementEventArgs blockReplacementEventArgs = new(block, null);
+
+            ManualResetEvent manualResetEvent = new(false);
+            _txPool.RemoveTransaction(Arg.Do<Keccak>(t => manualResetEvent.Set()));
+            _blockTree.BlockAddedToMain += Raise.EventWith(new object(), blockReplacementEventArgs);
+            manualResetEvent.WaitOne(TimeSpan.FromMilliseconds(200));
+
+            // transactions[nonceIncludedInBlock] was included in the block and should be removed, as well as all lower nonces.
+            _txPool.GetOwnPendingTransactions().Length.Should().Be(numberOfTxs - nonceIncludedInBlock - 1);
+        }
+
+        [Test]
+        public void broadcaster_should_work_well_when_there_are_no_txs_in_persistent_txs_from_sender_of_tx_included_in_block()
+        {
+            _txPool = CreatePool();
+
+            Transaction transactionA = Build.A.Transaction
+                .WithNonce(0)
+                .WithGasLimit(GasCostOf.Transaction)
+                .WithGasPrice(10.GWei())
+                .SignedAndResolved(_ethereumEcdsa, TestItem.PrivateKeyA)
+                .TestObject;
+            EnsureSenderBalance(transactionA);
+            _txPool.SubmitTx(transactionA, TxHandlingOptions.None);
+
+            Transaction transactionB = Build.A.Transaction
+                .WithNonce(0)
+                .WithGasLimit(GasCostOf.Transaction)
+                .WithGasPrice(10.GWei())
+                .SignedAndResolved(_ethereumEcdsa, TestItem.PrivateKeyB)
+                .TestObject;
+            EnsureSenderBalance(transactionB);
+            _txPool.SubmitTx(transactionB, TxHandlingOptions.PersistentBroadcast);
+
+            _txPool.GetPendingTransactions().Length.Should().Be(2);
+            _txPool.GetOwnPendingTransactions().Length.Should().Be(1);
+
+            Block block = Build.A.Block.WithTransactions(transactionA).TestObject;
+            BlockReplacementEventArgs blockReplacementEventArgs = new(block, null);
+
+            ManualResetEvent manualResetEvent = new(false);
+            _txPool.RemoveTransaction(Arg.Do<Keccak>(t => manualResetEvent.Set()));
+            _blockTree.BlockAddedToMain += Raise.EventWith(new object(), blockReplacementEventArgs);
+            manualResetEvent.WaitOne(TimeSpan.FromMilliseconds(200));
+
+            _txPool.GetPendingTransactions().Length.Should().Be(1);
+            _txPool.GetOwnPendingTransactions().Length.Should().Be(1);
+        }
+
         [Test]
         public async Task should_remove_transactions_concurrently()
         {

src/Nethermind/Nethermind.TxPool/Collections/SortedPool.cs

@@ -145,6 +145,21 @@ public bool TryTakeFirst(out TValue first)
             return TryRemove(GetKey(sortedValues.Min), out first);
         }
 
+        /// <summary>
+        /// Returns specified number of elements from the start of supplied comparer order.
+        /// </summary>
+        [MethodImpl(MethodImplOptions.Synchronized)]
+        public IEnumerable<TValue> GetFirsts(int numberOfValues)
+        {
+            SortedSet<TValue> sortedValues = new(_sortedComparer);
+            foreach (KeyValuePair<TGroupKey, SortedSet<TValue>> bucket in _buckets)
+            {
+                sortedValues.Add(bucket.Value.Max!);
+            }
+
+            return sortedValues.Take(numberOfValues);
+        }
+
         /// <summary>
         /// Gets last element in supplied comparer order.
         /// </summary>
@@ -210,6 +225,36 @@ private bool TryRemove(TKey key, bool evicted, out TValue value, out ICollection
         [MethodImpl(MethodImplOptions.Synchronized)]
         public bool TryRemove(TKey key) => TryRemove(key, out _, out _);
 
+        /// <summary>
+        /// Tries to get elements matching predicated criteria, iterating through SortedSet with break on first mismatch.
+        /// </summary>
+        /// <param name="groupKey">Given GroupKey, which elements are checked.</param>
+        /// <param name="where">Predicated criteria.</param>
+        /// <returns>Elements matching predicated criteria.</returns>
+        [MethodImpl(MethodImplOptions.Synchronized)]
+        public IEnumerable<TValue> TakeWhile(TGroupKey groupKey, Predicate<TValue> where)
+        {
+            if (_buckets.TryGetValue(groupKey, out SortedSet<TValue>? bucket))
+            {
+                using SortedSet<TValue>.Enumerator enumerator = bucket!.GetEnumerator();
+                List<TValue>? list = null;
+
+                while (enumerator.MoveNext())
+                {
+                    if (!where(enumerator.Current))
+                    {
+                        break;
+                    }
+
+                    list ??= new List<TValue>();
+                    list.Add(enumerator.Current);
+                }
+
+                return list ?? Enumerable.Empty<TValue>();
+            }
+            return Enumerable.Empty<TValue>();
+        }
+
         /// <summary>
         /// Tries to get element.
         /// </summary>

src/Nethermind/Nethermind.TxPool/TxBroadcaster.cs

@@ -22,6 +22,7 @@
 using Nethermind.Core;
 using Nethermind.Core.Crypto;
 using Nethermind.Core.Timers;
+using Nethermind.Int256;
 using Nethermind.Logging;
 using Nethermind.TxPool.Collections;
 
@@ -111,11 +112,22 @@ public void StopBroadcast(Keccak txHash)
         {
             if (_persistentTxs.Count != 0)
             {
-                bool wasIncluded = _persistentTxs.TryRemove(txHash, out Transaction _);
-                if (wasIncluded)
+                bool hasBeenRemoved = _persistentTxs.TryRemove(txHash, out Transaction _);
+                if (hasBeenRemoved)
                 {
                     if (_logger.IsTrace) _logger.Trace(
-                        $"Transaction {txHash} removed from broadcaster after block inclusion");
+                        $"Transaction {txHash} removed from broadcaster");
+                }
+            }
+        }
+
+        public void EnsureStopBroadcastUpToNonce(Address address, UInt256 nonce)
+        {
+            if (_persistentTxs.Count != 0)
+            {
+                foreach (Transaction tx in _persistentTxs.TakeWhile(address, t => t.Nonce <= nonce))
+                {
+                    StopBroadcast(tx.Hash!);
                 }
             }
         }
@@ -124,15 +136,13 @@ private void TimerOnElapsed(object sender, EventArgs args)
         {
             void NotifyPeers()
             {
-                Transaction[] persistentTxs = _persistentTxs.GetSnapshot();
-
                 _txsToSend = Interlocked.Exchange(ref _accumulatedTemporaryTxs, _txsToSend);
 
                 if (_logger.IsDebug) _logger.Debug($"Broadcasting transactions to all peers");
 
                 foreach ((_, ITxPoolPeer peer) in _peers)
                 {
-                    Notify(peer, GetTxsToSend(peer, persistentTxs, _txsToSend));
+                    Notify(peer, GetTxsToSend(peer, _txsToSend));
                 }
 
                 _txsToSend.Clear();
@@ -142,13 +152,20 @@ void NotifyPeers()
             _timer.Enabled = true;
         }
 
-        private IEnumerable<Transaction> GetTxsToSend(ITxPoolPeer peer, IReadOnlyList<Transaction> persistentTxs, IEnumerable<Transaction> txsToSend)
+        internal IEnumerable<Transaction> GetTxsToSend(ITxPoolPeer peer, IEnumerable<Transaction> txsToSend)
         {
-            for (int i = 0; i < persistentTxs.Count; i++)
+            if (_txPoolConfig.PeerNotificationThreshold > 0)
             {
-                if (_txPoolConfig.PeerNotificationThreshold >= Random.Value.Next(1, 100))
+                // PeerNotificationThreshold is a declared in config percent of transactions in persistent broadcast,
+                // which will be sent when timer elapse. numberOfPersistentTxsToBroadcast is equal to
+                // PeerNotificationThreshold multiplicated by number of transactions in persistent broadcast, rounded up.
+                int numberOfPersistentTxsToBroadcast =
+                    Math.Min(_txPoolConfig.PeerNotificationThreshold * _persistentTxs.Count / 100 + 1,
+                        _persistentTxs.Count);
+
+                foreach (Transaction tx in _persistentTxs.GetFirsts(numberOfPersistentTxsToBroadcast))
                 {
-                    yield return persistentTxs[i];
+                    yield return tx;
                 }
             }
 

src/Nethermind/Nethermind.TxPool/TxPool.cs

@@ -221,6 +221,8 @@ private void RemoveProcessedTransactions(IReadOnlyList<Transaction> blockTransac
                 {
                     eip1559Txs++;
                 }
+
+                _broadcaster.EnsureStopBroadcastUpToNonce(blockTransactions[i].SenderAddress!, blockTransactions[i].Nonce);
             }
 
             if (transactionsInBlock != 0)