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

CHANGELOG.md

@@ -1,5 +1,16 @@
 # Changelog
 
+## 24.1.0
+
+### Breaking Changes
+
+### Additions and Improvements
+
+
+### Bug Fixes
+
+### Download Links
+
 ## 23.10.2
 
 ### Breaking Changes
@@ -18,6 +29,7 @@
 - Upgrade EVM Reference Tests to v13 (Cancun) [#6114](https://github.com/hyperledger/besu/pull/6114)
 - Add `yParity` to GraphQL and JSON-RPC for relevant querise. [6119](https://github.com/hyperledger/besu/pull/6119)
 - Force tx replacement price bump to zero when zero base fee market is configured or `--min-gas-price` is set to 0. This allows for easier tx replacement in networks where there is not gas price. [#6079](https://github.com/hyperledger/besu/pull/6079)
+- Add `--Xtx-pool-disable-sender-grouping` 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
 - Upgrade netty to address CVE-2023-44487, CVE-2023-34462 [#6100](https://github.com/hyperledger/besu/pull/6100)

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

@@ -208,6 +208,9 @@ static class Unstable {
     private static final String ETH65_TX_ANNOUNCED_BUFFERING_PERIOD_FLAG =
         "--Xeth65-tx-announced-buffering-period-milliseconds";
 
+    private static final String TX_POOL_DISABLE_SENDER_GROUPING =
+        "--Xtx-pool-disable-sender-grouping";
+
     @CommandLine.Option(
         names = {TX_MESSAGE_KEEP_ALIVE_SEC_FLAG},
         paramLabel = "<INTEGER>",
@@ -228,6 +231,16 @@ static class Unstable {
         arity = "1")
     private Duration eth65TrxAnnouncedBufferingPeriod =
         TransactionPoolConfiguration.Unstable.ETH65_TRX_ANNOUNCED_BUFFERING_PERIOD;
+
+    @CommandLine.Option(
+        names = {TX_POOL_DISABLE_SENDER_GROUPING},
+        paramLabel = "<Boolean>",
+        hidden = true,
+        description =
+            "Disable sender grouping of transactions during selection. (default: ${DEFAULT-VALUE})",
+        arity = "0..1")
+    private Boolean disableSenderTxGrouping =
+        TransactionPoolConfiguration.DEFAULT_DISABLE_SENDER_TX_GROUPING;
   }
 
   private TransactionPoolOptions() {}
@@ -269,6 +282,8 @@ public static TransactionPoolOptions fromConfig(final TransactionPoolConfigurati
         config.getUnstable().getTxMessageKeepAliveSeconds();
     options.unstableOptions.eth65TrxAnnouncedBufferingPeriod =
         config.getUnstable().getEth65TrxAnnouncedBufferingPeriod();
+    options.unstableOptions.disableSenderTxGrouping =
+        config.getUnstable().getDisableSenderTxGrouping();
 
     return options;
   }
@@ -322,6 +337,7 @@ public TransactionPoolConfiguration toDomainObject() {
             ImmutableTransactionPoolConfiguration.Unstable.builder()
                 .txMessageKeepAliveSeconds(unstableOptions.txMessageKeepAliveSeconds)
                 .eth65TrxAnnouncedBufferingPeriod(unstableOptions.eth65TrxAnnouncedBufferingPeriod)
+                .disableSenderTxGrouping(unstableOptions.disableSenderTxGrouping)
                 .build())
         .build();
   }

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

@@ -317,6 +317,20 @@ public void eth65TrxAnnouncedBufferingPeriodWithInvalidInputShouldFail2() {
         "-1");
   }
 
+  @Test
+  public void disableSenderTxGroupingOn() {
+    internalTestSuccess(
+        config -> assertThat(config.getUnstable().getDisableSenderTxGrouping()).isTrue(),
+        "--Xtx-pool-disable-sender-grouping=true");
+  }
+
+  @Test
+  public void disableSenderTxGroupingOff() {
+    internalTestSuccess(
+        config -> assertThat(config.getUnstable().getDisableSenderTxGrouping()).isFalse(),
+        "--Xtx-pool-disable-sender-grouping=false");
+  }
+
   @Override
   protected TransactionPoolConfiguration createDefaultDomainObject() {
     return TransactionPoolConfiguration.DEFAULT;

besu/src/test/resources/everything_config.toml

@@ -188,6 +188,8 @@ tx-pool-max-future-by-sender=321
 tx-pool-retention-hours=999
 tx-pool-max-size=1234
 tx-pool-limit-by-account-percentage=0.017
+## Experimental
+Xtx-pool-disable-sender-grouping=false
 
 # Revert Reason
 revert-reason-enabled=false

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

@@ -47,6 +47,11 @@ default Duration getEth65TrxAnnouncedBufferingPeriod() {
     default int getTxMessageKeepAliveSeconds() {
       return DEFAULT_TX_MSG_KEEP_ALIVE;
     }
+
+    @Value.Default
+    default Boolean getDisableSenderTxGrouping() {
+      return DEFAULT_DISABLE_SENDER_TX_GROUPING;
+    }
   }
 
   enum Implementation {
@@ -63,6 +68,7 @@ enum Implementation {
   Wei DEFAULT_RPC_TX_FEE_CAP = Wei.fromEth(1);
   boolean DEFAULT_NO_LOCAL_PRIORITY = 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;
   int DEFAULT_MAX_PRIORITIZED_TRANSACTIONS = 2000;

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

@@ -246,14 +246,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.getUnstable().getDisableSenderTxGrouping() ? 1 : Integer.MAX_VALUE)) {
           final TransactionSelectionResult result =
               selector.evaluateTransaction(transactionToProcess);
 

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

@@ -50,11 +50,17 @@ 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);

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

@@ -65,6 +65,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);
@@ -93,9 +94,33 @@ 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)
+              .txPoolLimitByAccountPercentage(Fraction.fromFloat(1.0f))
+              .unstable(
+                  ImmutableTransactionPoolConfiguration.Unstable.builder()
+                      .disableSenderTxGrouping(true)
+                      .build())
+              .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);
@@ -109,6 +134,7 @@ abstract AbstractPendingTransactionsSorter getPendingTransactions(
 
   @Test
   public void shouldReturnExclusivelyLocalTransactionsWhenAppropriate() {
+
     final Transaction localTransaction0 = createTransaction(0);
     transactions.addTransaction(createLocalPendingTransaction(localTransaction0), Optional.empty());
     assertThat(transactions.size()).isEqualTo(1);
@@ -320,6 +346,104 @@ public void shouldNotNotifyDroppedListenerWhenTransactionAddedToBlock() {
     verifyNoInteractions(droppedListener);
   }
 
+  @Test
+  public void selectTransactionsInDefaultOrder() {
+    assertThat(
+            transactionsLarge.addTransaction(
+                createRemotePendingTransaction(transaction1), Optional.empty()))
+        .isEqualTo(ADDED);
+    assertThat(
+            transactionsLarge.addTransaction(
+                createRemotePendingTransaction(transaction2), Optional.empty()))
+        .isEqualTo(ADDED);
+    assertThat(
+            transactionsLarge.addTransaction(
+                createRemotePendingTransaction(transaction1Sdr2), Optional.empty()))
+        .isEqualTo(ADDED);
+    assertThat(
+            transactionsLarge.addTransaction(
+                createRemotePendingTransaction(transaction2Sdr2), Optional.empty()))
+        .isEqualTo(ADDED);
+    assertThat(
+            transactionsLarge.addTransaction(
+                createRemotePendingTransaction(transaction3Sdr2), Optional.empty()))
+        .isEqualTo(ADDED);
+    assertThat(
+            transactionsLarge.addTransaction(
+                createRemotePendingTransaction(transaction3), Optional.empty()))
+        .isEqualTo(ADDED);
+
+    final List<Transaction> parsedTransactions = Lists.newArrayList();
+    transactionsLarge.selectTransactions(
+        pendingTx -> {
+          parsedTransactions.add(pendingTx.getTransaction());
+
+          if (parsedTransactions.size() == 6) {
+            return TransactionSelectionResult.BLOCK_OCCUPANCY_ABOVE_THRESHOLD;
+          }
+          return SELECTED;
+        });
+
+    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.addTransaction(
+                createRemotePendingTransaction(transaction2), Optional.empty()))
+        .isEqualTo(ADDED);
+    assertThat(
+            transactionsLargeNoSenderGrouping.addTransaction(
+                createRemotePendingTransaction(transaction1Sdr2), Optional.empty()))
+        .isEqualTo(ADDED);
+    assertThat(
+            transactionsLargeNoSenderGrouping.addTransaction(
+                createRemotePendingTransaction(transaction2Sdr2), Optional.empty()))
+        .isEqualTo(ADDED);
+    assertThat(
+            transactionsLargeNoSenderGrouping.addTransaction(
+                createRemotePendingTransaction(transaction1), Optional.empty()))
+        .isEqualTo(ADDED);
+    assertThat(
+            transactionsLargeNoSenderGrouping.addTransaction(
+                createRemotePendingTransaction(transaction3Sdr2), Optional.empty()))
+        .isEqualTo(ADDED);
+    assertThat(
+            transactionsLargeNoSenderGrouping.addTransaction(
+                createRemotePendingTransaction(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
   public void selectTransactionsUntilSelectorRequestsNoMore() {
     transactions.addTransaction(createRemotePendingTransaction(transaction1), Optional.empty());
@@ -489,6 +613,10 @@ public void shouldReplaceTransactionWithSameSenderAndNonce_multipleReplacements(
             transactions.addTransaction(
                 createRemotePendingTransaction(independentTx), Optional.empty()))
         .isEqualTo(ADDED);
+    assertThat(
+            transactions.addTransaction(
+                createRemotePendingTransaction(independentTx), Optional.empty()))
+        .isEqualTo(ADDED);
 
     // All tx's except the last duplicate should be removed
     replacedTransactions.forEach(this::assertTransactionNotPending);
@@ -659,9 +787,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);
+  }
+
   private PendingTransaction createRemotePendingTransaction(
       final Transaction transaction, final long addedAt) {
     return PendingTransaction.newPendingTransaction(transaction, false, false, addedAt);