Add test to check that TX selection doesn't prioritize TXs from the same sender

CHANGELOG.md

@@ -1,11 +1,13 @@
 # Changelog
 
-## Next Release
+## 24.1.0
 
 ### Breaking Changes
 
 ### Additions and Improvements
 
+ - Add option to avoid grouping transactions from a single sender when selecting candidate transactions for a block [#6022](https://github.com/hyperledger/besu/pull/6022)
+
 ### Bug Fixes
 
 ### Download Links

besu/src/main/java/org/hyperledger/besu/cli/options/stable/TransactionPoolOptions.java

@@ -42,6 +42,7 @@ public class TransactionPoolOptions implements CLIOptions<TransactionPoolConfigu
   private static final String TX_POOL_ENABLE_SAVE_RESTORE = "--tx-pool-enable-save-restore";
   private static final String TX_POOL_SAVE_FILE = "--tx-pool-save-file";
   private static final String TX_POOL_PRICE_BUMP = "--tx-pool-price-bump";
+  private static final String TX_POOL_DISABLE_SENDER_GROUPING = "--tx-pool-disable-sender-grouping";
   private static final String RPC_TX_FEECAP = "--rpc-tx-feecap";
   private static final String STRICT_TX_REPLAY_PROTECTION_ENABLED_FLAG =
       "--strict-tx-replay-protection-enabled";
@@ -88,6 +89,15 @@ public class TransactionPoolOptions implements CLIOptions<TransactionPoolConfigu
       arity = "1")
   private Percentage priceBump = TransactionPoolConfiguration.DEFAULT_PRICE_BUMP;
 
+  @CommandLine.Option(
+      names = {TX_POOL_DISABLE_SENDER_GROUPING},
+      paramLabel = "<Boolean>",
+      description =
+          "Disable sender grouping of transactions during selection. (default: ${DEFAULT-VALUE})",
+      arity = "0..1")
+  private Boolean disableSenderTXGrouping =
+      TransactionPoolConfiguration.DEFAULT_DISABLE_SENDER_TX_GROUPING;
+
   @CommandLine.Option(
       names = {RPC_TX_FEECAP},
       description =
@@ -202,6 +212,7 @@ public static TransactionPoolOptions fromConfig(final TransactionPoolConfigurati
     options.saveRestoreEnabled = config.getEnableSaveRestore();
     options.disableLocalTxs = config.getDisableLocalTransactions();
     options.priceBump = config.getPriceBump();
+    options.disableSenderTXGrouping = config.getDisableSenderTXGrouping();
     options.txFeeCap = config.getTxFeeCap();
     options.saveFile = config.getSaveFile();
     options.strictTxReplayProtectionEnabled = config.getStrictTransactionReplayProtectionEnabled();
@@ -244,6 +255,7 @@ public TransactionPoolConfiguration toDomainObject() {
         .enableSaveRestore(saveRestoreEnabled)
         .disableLocalTransactions(disableLocalTxs)
         .priceBump(priceBump)
+        .disableSenderTXGrouping(disableSenderTXGrouping)
         .txFeeCap(txFeeCap)
         .saveFile(saveFile)
         .strictTransactionReplayProtectionEnabled(strictTxReplayProtectionEnabled)

besu/src/test/java/org/hyperledger/besu/cli/options/stable/TransactionPoolOptionsTest.java

@@ -146,6 +146,20 @@ public void priceBump() {
         priceBump.toString());
   }
 
+  @Test
+  public void disableSenderTXGroupingOn() {
+    internalTestSuccess(
+        config -> assertThat(config.getDisableSenderTXGrouping()).isTrue(),
+        "--tx-pool-disable-sender-grouping=true");
+  }
+
+  @Test
+  public void disableSenderTXGroupingOff() {
+    internalTestSuccess(
+        config -> assertThat(config.getDisableSenderTXGrouping()).isFalse(),
+        "--tx-pool-disable-sender-grouping=false");
+  }
+
   @Test
   public void invalidPriceBumpShouldFail() {
     internalTestFailure(

besu/src/test/resources/everything_config.toml

@@ -177,6 +177,7 @@ strict-tx-replay-protection-enabled=true
 tx-pool-disable-locals=false
 tx-pool-enable-save-restore=true
 tx-pool-save-file="txpool.dump"
+tx-pool-disable-sender-grouping=false
 ## Layered
 tx-pool-layer-max-capacity=12345678
 tx-pool-max-prioritized=9876

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/transactions/TransactionPoolConfiguration.java

@@ -62,6 +62,7 @@ enum Implementation {
   Wei DEFAULT_RPC_TX_FEE_CAP = Wei.fromEth(1);
   boolean DEFAULT_DISABLE_LOCAL_TXS = false;
   boolean DEFAULT_ENABLE_SAVE_RESTORE = false;
+  boolean DEFAULT_DISABLE_SENDER_TX_GROUPING = false;
 
   File DEFAULT_SAVE_FILE = new File(DEFAULT_SAVE_FILE_NAME);
   long DEFAULT_PENDING_TRANSACTIONS_LAYER_MAX_CAPACITY_BYTES = 12_500_000L;
@@ -121,6 +122,11 @@ default File getSaveFile() {
     return DEFAULT_SAVE_FILE;
   }
 
+  @Value.Default
+  default Boolean getDisableSenderTXGrouping() {
+    return DEFAULT_DISABLE_SENDER_TX_GROUPING;
+  }
+
   @Value.Default
   default Implementation getTxPoolImplementation() {
     return DEFAULT_TX_POOL_IMPLEMENTATION;

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/transactions/sorter/AbstractPendingTransactionsSorter.java

@@ -252,14 +252,17 @@ public void selectTransactions(final TransactionSelector selector) {
       final Set<Transaction> transactionsToRemove = new HashSet<>();
       final Map<Address, AccountTransactionOrder> accountTransactions = new HashMap<>();
       final Iterator<PendingTransaction> prioritizedTransactions = prioritizedTransactions();
+
       while (prioritizedTransactions.hasNext()) {
         final PendingTransaction highestPriorityPendingTransaction = prioritizedTransactions.next();
         final AccountTransactionOrder accountTransactionOrder =
             accountTransactions.computeIfAbsent(
                 highestPriorityPendingTransaction.getSender(), this::createSenderTransactionOrder);
 
         for (final PendingTransaction transactionToProcess :
-            accountTransactionOrder.transactionsToProcess(highestPriorityPendingTransaction)) {
+            accountTransactionOrder.transactionsToProcess(
+                highestPriorityPendingTransaction,
+                poolConfig.getDisableSenderTXGrouping() ? 1 : Integer.MAX_VALUE)) {
           final TransactionSelectionResult result =
               selector.evaluateTransaction(transactionToProcess);
 
@@ -275,6 +278,51 @@ public void selectTransactions(final TransactionSelector selector) {
           }
         }
       }
+
+      if (poolConfig.getDisableSenderTXGrouping()) {
+        // If sender grouping of transactions is disabled, there will potentially
+        // be transactions left for the sender accounts we have selected at least 1 TX for already.
+
+        // Iterate over the remaining accounts that have at least 1 deferred transaction until there
+        // aren't any left.
+        while (accountTransactions.size() > 0) {
+
+          // List the current accounts with deferred transactions
+          List<Map.Entry<Address, AccountTransactionOrder>> accounts =
+              accountTransactions.entrySet().stream().toList();
+
+          for (Map.Entry<Address, AccountTransactionOrder> nextEntry : accounts) {
+            // Get the address and account transaction order
+            Address addr = nextEntry.getKey();
+            AccountTransactionOrder nextAccount = nextEntry.getValue();
+
+            // Get the highest priority deferred transaction and evaluate it
+            Optional<PendingTransaction> nextTransactionForSender =
+                nextAccount.getHighestPriorityDeferredTransaction();
+
+            if (nextTransactionForSender.isPresent()) {
+              PendingTransaction transactionToEvaluate = nextTransactionForSender.get();
+              final TransactionSelectionResult result =
+                  selector.evaluateTransaction(transactionToEvaluate);
+
+              if (result.discard()) {
+                transactionsToRemove.add(transactionToEvaluate.getTransaction());
+                transactionsToRemove.addAll(
+                    signalInvalidAndGetDependentTransactions(
+                        transactionToEvaluate.getTransaction()));
+              }
+
+              if (result.stop()) {
+                transactionsToRemove.forEach(this::removeTransaction);
+                return;
+              }
+            } else {
+              // This account is empty - remove from the list
+              accountTransactions.remove(addr);
+            }
+          }
+        }
+      }
       transactionsToRemove.forEach(this::removeTransaction);
     }
   }

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/transactions/sorter/AccountTransactionOrder.java

@@ -20,6 +20,7 @@
 import java.util.Comparator;
 import java.util.List;
 import java.util.NavigableSet;
+import java.util.Optional;
 import java.util.TreeSet;
 import java.util.stream.Stream;
 
@@ -50,16 +51,34 @@ public AccountTransactionOrder(final Stream<PendingTransaction> senderTransactio
    */
   public Iterable<PendingTransaction> transactionsToProcess(
       final PendingTransaction nextTransactionInPriorityOrder) {
+    return transactionsToProcess(nextTransactionInPriorityOrder, Integer.MAX_VALUE);
+  }
+
+  public Iterable<PendingTransaction> transactionsToProcess(
+      final PendingTransaction nextTransactionInPriorityOrder, final int maxTransactionsToReturn) {
     deferredTransactions.add(nextTransactionInPriorityOrder);
     final List<PendingTransaction> transactionsToApply = new ArrayList<>();
     while (!deferredTransactions.isEmpty()
         && !transactionsForSender.isEmpty()
-        && deferredTransactions.first().equals(transactionsForSender.first())) {
+        && deferredTransactions.first().equals(transactionsForSender.first())
+        && transactionsToApply.size() < maxTransactionsToReturn) {
       final PendingTransaction transaction = deferredTransactions.first();
       transactionsToApply.add(transaction);
       deferredTransactions.remove(transaction);
       transactionsForSender.remove(transaction);
     }
     return transactionsToApply;
   }
+
+  public Optional<PendingTransaction> getHighestPriorityDeferredTransaction() {
+    Optional<PendingTransaction> transactionToApply = Optional.empty();
+    if (!deferredTransactions.isEmpty()
+        && !transactionsForSender.isEmpty()
+        && deferredTransactions.first().equals(transactionsForSender.first())) {
+      transactionToApply = Optional.of(deferredTransactions.first());
+      deferredTransactions.remove(transactionToApply.get());
+      transactionsForSender.remove(transactionToApply.get());
+    }
+    return transactionToApply;
+  }
 }

ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/transactions/sorter/AbstractPendingTransactionsTestBase.java

@@ -64,6 +64,7 @@
 public abstract class AbstractPendingTransactionsTestBase {
 
   protected static final int MAX_TRANSACTIONS = 5;
+  protected static final int MAX_TRANSACTIONS_LARGE_POOL = 15;
   private static final float LIMITED_TRANSACTIONS_BY_SENDER_PERCENTAGE = 0.8f;
   protected static final Supplier<SignatureAlgorithm> SIGNATURE_ALGORITHM =
       Suppliers.memoize(SignatureAlgorithmFactory::getInstance);
@@ -92,9 +93,29 @@ public abstract class AbstractPendingTransactionsTestBase {
           .build();
   protected PendingTransactions senderLimitedTransactions =
       getPendingTransactions(senderLimitedConfig, Optional.empty());
+  protected AbstractPendingTransactionsSorter transactionsLarge =
+      getPendingTransactions(
+          ImmutableTransactionPoolConfiguration.builder()
+              .txPoolMaxSize(MAX_TRANSACTIONS_LARGE_POOL)
+              .txPoolLimitByAccountPercentage(Fraction.fromFloat(1.0f))
+              .build(),
+          Optional.empty());
+  protected AbstractPendingTransactionsSorter transactionsLargeNoSenderGrouping =
+      getPendingTransactions(
+          ImmutableTransactionPoolConfiguration.builder()
+              .txPoolMaxSize(MAX_TRANSACTIONS_LARGE_POOL)
+              .disableSenderTXGrouping(true)
+              .txPoolLimitByAccountPercentage(Fraction.fromFloat(1.0f))
+              .build(),
+          Optional.empty());
 
   protected final Transaction transaction1 = createTransaction(2);
   protected final Transaction transaction2 = createTransaction(1);
+  protected final Transaction transaction3 = createTransaction(3);
+
+  protected final Transaction transaction1Sdr2 = createTransactionSender2(1);
+  protected final Transaction transaction2Sdr2 = createTransactionSender2(2);
+  protected final Transaction transaction3Sdr2 = createTransactionSender2(3);
 
   protected final PendingTransactionAddedListener listener =
       mock(PendingTransactionAddedListener.class);
@@ -309,19 +330,86 @@ public void shouldNotNotifyDroppedListenerWhenTransactionAddedToBlock() {
   }
 
   @Test
-  public void selectTransactionsUntilSelectorRequestsNoMore() {
-    transactions.addRemoteTransaction(transaction1, Optional.empty());
-    transactions.addRemoteTransaction(transaction2, Optional.empty());
+  public void selectTransactionsInDefaultOrder() {
+    assertThat(transactionsLarge.addRemoteTransaction(transaction1, Optional.empty()))
+        .isEqualTo(ADDED);
+    assertThat(transactionsLarge.addRemoteTransaction(transaction2, Optional.empty()))
+        .isEqualTo(ADDED);
+    assertThat(transactionsLarge.addRemoteTransaction(transaction1Sdr2, Optional.empty()))
+        .isEqualTo(ADDED);
+    assertThat(transactionsLarge.addRemoteTransaction(transaction2Sdr2, Optional.empty()))
+        .isEqualTo(ADDED);
+    assertThat(transactionsLarge.addRemoteTransaction(transaction3Sdr2, Optional.empty()))
+        .isEqualTo(ADDED);
+    assertThat(transactionsLarge.addRemoteTransaction(transaction3, Optional.empty()))
+        .isEqualTo(ADDED);
 
     final List<Transaction> parsedTransactions = Lists.newArrayList();
-    transactions.selectTransactions(
+    transactionsLarge.selectTransactions(
         pendingTx -> {
           parsedTransactions.add(pendingTx.getTransaction());
-          return TransactionSelectionResult.BLOCK_OCCUPANCY_ABOVE_THRESHOLD;
+
+          if (parsedTransactions.size() == 6) {
+            return TransactionSelectionResult.BLOCK_OCCUPANCY_ABOVE_THRESHOLD;
+          }
+          return SELECTED;
         });
 
-    assertThat(parsedTransactions.size()).isEqualTo(1);
-    assertThat(parsedTransactions.get(0)).isEqualTo(transaction2);
+    assertThat(parsedTransactions.size()).isEqualTo(6);
+
+    assertThat(parsedTransactions.get(0)).isEqualTo(transaction1Sdr2);
+    assertThat(parsedTransactions.get(1)).isEqualTo(transaction2Sdr2);
+    assertThat(parsedTransactions.get(2)).isEqualTo(transaction3Sdr2);
+    assertThat(parsedTransactions.get(3))
+        .isEqualTo(transaction2); // Transaction 2 is actually the lowest nonce for this sender
+    assertThat(parsedTransactions.get(4))
+        .isEqualTo(transaction1); // Transaction 1 is the next nonce for the sender
+    assertThat(parsedTransactions.get(5))
+        .isEqualTo(transaction3); // Transaction 3 is the next nonce for the sender
+  }
+
+  @Test
+  public void selectTransactionsInOrderNoGroupBySender() {
+    assertThat(
+            transactionsLargeNoSenderGrouping.addRemoteTransaction(transaction2, Optional.empty()))
+        .isEqualTo(ADDED);
+    assertThat(
+            transactionsLargeNoSenderGrouping.addRemoteTransaction(
+                transaction1Sdr2, Optional.empty()))
+        .isEqualTo(ADDED);
+    assertThat(
+            transactionsLargeNoSenderGrouping.addRemoteTransaction(
+                transaction2Sdr2, Optional.empty()))
+        .isEqualTo(ADDED);
+    assertThat(
+            transactionsLargeNoSenderGrouping.addRemoteTransaction(transaction1, Optional.empty()))
+        .isEqualTo(ADDED);
+    assertThat(
+            transactionsLargeNoSenderGrouping.addRemoteTransaction(
+                transaction3Sdr2, Optional.empty()))
+        .isEqualTo(ADDED);
+    assertThat(
+            transactionsLargeNoSenderGrouping.addRemoteTransaction(transaction3, Optional.empty()))
+        .isEqualTo(ADDED);
+
+    final List<Transaction> parsedTransactions = Lists.newArrayList();
+    transactionsLargeNoSenderGrouping.selectTransactions(
+        pendingTx -> {
+          parsedTransactions.add(pendingTx.getTransaction());
+
+          if (parsedTransactions.size() == 6) {
+            return TransactionSelectionResult.BLOCK_OCCUPANCY_ABOVE_THRESHOLD;
+          }
+          return SELECTED;
+        });
+
+    assertThat(parsedTransactions.size()).isEqualTo(6);
+
+    // Check that by setting --tx-pool-disable-sender-grouping=true then sdr 1 hasn't monopolized
+    // the pool selection just because its transaction (transaction 3) is the first one to be parsed
+    // by the selector
+    assertThat(parsedTransactions.get(0)).isEqualTo(transaction1Sdr2);
+    assertThat(parsedTransactions.get(1)).isEqualTo(transaction2);
   }
 
   @Test
@@ -613,9 +701,18 @@ protected Transaction createTransaction(final long transactionNumber) {
     return new TransactionTestFixture()
         .value(Wei.of(transactionNumber))
         .nonce(transactionNumber)
+        .gasPrice(Wei.of(0))
         .createTransaction(KEYS1);
   }
 
+  protected Transaction createTransactionSender2(final long transactionNumber) {
+    return new TransactionTestFixture()
+        .value(Wei.of(transactionNumber))
+        .nonce(transactionNumber)
+        .gasPrice(Wei.of(0))
+        .createTransaction(KEYS2);
+  }
+
   @Test
   public void shouldEvictMultipleOldTransactions() {
     final int maxTransactionRetentionHours = 1;