Remove tx from pool when its score is lower than a configured value

CHANGELOG.md

@@ -8,6 +8,7 @@
 
 ### Additions and Improvements
 - Update Java and Gradle dependecies [#7571](https://github.com/hyperledger/besu/pull/7571)
+- Layered txpool: new options `--tx-pool-min-score` to remove a tx from pool when its score is lower than the specified value [#7576](https://github.com/hyperledger/besu/pull/7576)
 
 ### Bug fixes
 - Layered txpool: do not send notifications when moving tx between layers [#7539](https://github.com/hyperledger/besu/pull/7539)

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

@@ -160,6 +160,7 @@ static class Layered {
     private static final String TX_POOL_MAX_PRIORITIZED_BY_TYPE =
         "--tx-pool-max-prioritized-by-type";
     private static final String TX_POOL_MAX_FUTURE_BY_SENDER = "--tx-pool-max-future-by-sender";
+    private static final String TX_POOL_MIN_SCORE = "--tx-pool-min-score";
 
     @CommandLine.Option(
         names = {TX_POOL_LAYER_MAX_CAPACITY},
@@ -196,6 +197,15 @@ static class Layered {
             "Max number of future pending transactions allowed for a single sender (default: ${DEFAULT-VALUE})",
         arity = "1")
     Integer txPoolMaxFutureBySender = TransactionPoolConfiguration.DEFAULT_MAX_FUTURE_BY_SENDER;
+
+    @CommandLine.Option(
+        names = {TX_POOL_MIN_SCORE},
+        paramLabel = "<Byte>",
+        description =
+            "Remove a pending transaction from the txpool if its score is lower than this value."
+                + "Accepts values between -128 and 127 (default: ${DEFAULT-VALUE})",
+        arity = "1")
+    Byte minScore = TransactionPoolConfiguration.DEFAULT_TX_POOL_MIN_SCORE;
   }
 
   @CommandLine.ArgGroup(
@@ -314,6 +324,7 @@ public static TransactionPoolOptions fromConfig(final TransactionPoolConfigurati
     options.layeredOptions.txPoolMaxPrioritizedByType =
         config.getMaxPrioritizedTransactionsByType();
     options.layeredOptions.txPoolMaxFutureBySender = config.getMaxFutureBySender();
+    options.layeredOptions.minScore = config.getMinScore();
     options.sequencedOptions.txPoolLimitByAccountPercentage =
         config.getTxPoolLimitByAccountPercentage();
     options.sequencedOptions.txPoolMaxSize = config.getTxPoolMaxSize();
@@ -372,6 +383,7 @@ public TransactionPoolConfiguration toDomainObject() {
         .maxPrioritizedTransactions(layeredOptions.txPoolMaxPrioritized)
         .maxPrioritizedTransactionsByType(layeredOptions.txPoolMaxPrioritizedByType)
         .maxFutureBySender(layeredOptions.txPoolMaxFutureBySender)
+        .minScore(layeredOptions.minScore)
         .txPoolLimitByAccountPercentage(sequencedOptions.txPoolLimitByAccountPercentage)
         .txPoolMaxSize(sequencedOptions.txPoolMaxSize)
         .pendingTxRetentionPeriod(sequencedOptions.pendingTxRetentionPeriod)

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

@@ -418,6 +418,24 @@ public void maxPrioritizedTxsPerTypeWrongTxType() {
         "WRONG_TYPE=1");
   }
 
+  @Test
+  public void minScoreWorks() {
+    final byte minScore = -10;
+    internalTestSuccess(
+        config -> assertThat(config.getMinScore()).isEqualTo(minScore),
+        "--tx-pool-min-score",
+        Byte.toString(minScore));
+  }
+
+  @Test
+  public void minScoreNonByteValueReturnError() {
+    final var overflowMinScore = Integer.toString(-300);
+    internalTestFailure(
+        "Invalid value for option '--tx-pool-min-score': '" + overflowMinScore + "' is not a byte",
+        "--tx-pool-min-score",
+        overflowMinScore);
+  }
+
   @Override
   protected TransactionPoolConfiguration createDefaultDomainObject() {
     return TransactionPoolConfiguration.DEFAULT;

besu/src/test/resources/everything_config.toml

@@ -195,6 +195,7 @@ tx-pool-retention-hours=999
 tx-pool-max-size=1234
 tx-pool-limit-by-account-percentage=0.017
 tx-pool-min-gas-price=1000
+tx-pool-min-score=100
 
 # Revert Reason
 revert-reason-enabled=false

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

@@ -80,6 +80,7 @@ enum Implementation {
   Implementation DEFAULT_TX_POOL_IMPLEMENTATION = Implementation.LAYERED;
   Set<Address> DEFAULT_PRIORITY_SENDERS = Set.of();
   Wei DEFAULT_TX_POOL_MIN_GAS_PRICE = Wei.of(1000);
+  byte DEFAULT_TX_POOL_MIN_SCORE = -128;
 
   TransactionPoolConfiguration DEFAULT = ImmutableTransactionPoolConfiguration.builder().build();
 
@@ -173,6 +174,11 @@ default Wei getMinGasPrice() {
     return DEFAULT_TX_POOL_MIN_GAS_PRICE;
   }
 
+  @Value.Default
+  default byte getMinScore() {
+    return DEFAULT_TX_POOL_MIN_SCORE;
+  }
+
   @Value.Default
   default TransactionPoolValidatorService getTransactionPoolValidatorService() {
     return new TransactionPoolValidatorService() {

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/transactions/layered/AbstractTransactionsLayer.java

@@ -21,6 +21,7 @@
 import static org.hyperledger.besu.ethereum.eth.transactions.TransactionAddedResult.TRY_NEXT_LAYER;
 import static org.hyperledger.besu.ethereum.eth.transactions.layered.AddReason.MOVE;
 import static org.hyperledger.besu.ethereum.eth.transactions.layered.RemovalReason.LayerMoveReason.EVICTED;
+import static org.hyperledger.besu.ethereum.eth.transactions.layered.RemovalReason.PoolRemovalReason.BELOW_MIN_SCORE;
 import static org.hyperledger.besu.ethereum.eth.transactions.layered.RemovalReason.PoolRemovalReason.CONFIRMED;
 import static org.hyperledger.besu.ethereum.eth.transactions.layered.RemovalReason.PoolRemovalReason.CROSS_LAYER_REPLACED;
 import static org.hyperledger.besu.ethereum.eth.transactions.layered.RemovalReason.PoolRemovalReason.REPLACED;
@@ -481,6 +482,9 @@ public void penalize(final PendingTransaction penalizedTransaction) {
     if (pendingTransactions.containsKey(penalizedTransaction.getHash())) {
       internalPenalize(penalizedTransaction);
       metrics.incrementPenalized(penalizedTransaction, name());
+      if (penalizedTransaction.getScore() < poolConfig.getMinScore()) {
+        remove(penalizedTransaction, BELOW_MIN_SCORE);
+      }
     } else {
       nextLayer.penalize(penalizedTransaction);
     }

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/transactions/layered/LayeredPendingTransactions.java

@@ -42,7 +42,6 @@
 import org.hyperledger.besu.plugin.data.TransactionSelectionResult;
 
 import java.util.ArrayDeque;
-import java.util.ArrayList;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
@@ -316,7 +315,6 @@ public synchronized List<Transaction> getPriorityTransactions() {
 
   @Override
   public void selectTransactions(final PendingTransactions.TransactionSelector selector) {
-    final List<PendingTransaction> penalizedTransactions = new ArrayList<>();
     final Set<Address> skipSenders = new HashSet<>();
 
     final Map<Byte, List<SenderPendingTransactions>> candidateTxsByScore;
@@ -356,7 +354,12 @@ public void selectTransactions(final PendingTransactions.TransactionSelector sel
             }
 
             if (selectionResult.penalize()) {
-              penalizedTransactions.add(candidatePendingTx);
+              ethScheduler.scheduleTxWorkerTask(
+                  () -> {
+                    synchronized (this) {
+                      prioritizedTransactions.penalize(candidatePendingTx);
+                    }
+                  });
               LOG.atTrace()
                   .setMessage("Transaction {} penalized")
                   .addArgument(candidatePendingTx::toTraceLog)
@@ -379,15 +382,6 @@ public void selectTransactions(final PendingTransactions.TransactionSelector sel
         }
       }
     }
-
-    ethScheduler.scheduleTxWorkerTask(
-        () ->
-            penalizedTransactions.forEach(
-                penalizedTx -> {
-                  synchronized (this) {
-                    prioritizedTransactions.penalize(penalizedTx);
-                  }
-                }));
   }
 
   @Override

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/transactions/layered/RemovalReason.java

@@ -52,25 +52,30 @@ enum RemovedFrom {
   /** The reason why the tx has been removed from the pool */
   enum PoolRemovalReason implements RemovalReason {
     /** Tx removed since it is confirmed on chain, as part of an imported block. */
-    CONFIRMED(),
+    CONFIRMED,
     /** Tx removed since it has been replaced by another one added in the same layer. */
-    REPLACED(),
+    REPLACED,
     /** Tx removed since it has been replaced by another one added in another layer. */
-    CROSS_LAYER_REPLACED(),
+    CROSS_LAYER_REPLACED,
     /** Tx removed when the pool is full, to make space for new incoming txs. */
-    DROPPED(),
+    DROPPED,
     /**
      * Tx removed since found invalid after it was added to the pool, for example during txs
      * selection for a new block proposal.
      */
-    INVALIDATED(),
+    INVALIDATED,
     /**
      * Special case, when for a sender, discrepancies are found between the world state view and the
      * pool view, then all the txs for this sender are removed and added again. Discrepancies, are
      * rare, and can happen during a short windows when a new block is being imported and the world
      * state being updated.
      */
-    RECONCILED();
+    RECONCILED,
+    /**
+     * When a pending tx is penalized its score is decreased, if at some point its score is lower
+     * than the configured minimum then the pending tx is removed from the pool.
+     */
+    BELOW_MIN_SCORE;
 
     private final String label;
 
@@ -95,22 +100,22 @@ enum LayerMoveReason implements RemovalReason {
      * When the current layer is full, and this tx needs to be moved to the lower layer, in order to
      * free space.
      */
-    EVICTED(),
+    EVICTED,
     /**
      * Specific to sequential layers, when a tx is removed because found invalid, then if the sender
      * has other txs with higher nonce, then a gap is created, and since sequential layers do not
      * permit gaps, txs following the invalid one need to be moved to lower layers.
      */
-    FOLLOW_INVALIDATED(),
+    FOLLOW_INVALIDATED,
     /**
      * When a tx is moved to the upper layer, since it satisfies all the requirement to be promoted.
      */
-    PROMOTED(),
+    PROMOTED,
     /**
      * When a tx is moved to the lower layer, since it, or a preceding one from the same sender,
      * does not respect anymore the requisites to stay in this layer.
      */
-    DEMOTED();
+    DEMOTED;
 
     private final String label;
 

ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/transactions/layered/LayersTest.java

@@ -74,12 +74,14 @@ public class LayersTest extends BaseTransactionPoolTest {
   private static final int MAX_FUTURE_FOR_SENDER = 10;
   private static final Wei BASE_FEE = Wei.ONE;
   private static final Wei MIN_GAS_PRICE = BASE_FEE;
+  private static final byte MIN_SCORE = 125;
 
   private static final TransactionPoolConfiguration DEFAULT_TX_POOL_CONFIG =
       ImmutableTransactionPoolConfiguration.builder()
           .maxPrioritizedTransactions(MAX_PRIO_TRANSACTIONS)
           .maxPrioritizedTransactionsByType(Map.of(BLOB, 1))
           .maxFutureBySender(MAX_FUTURE_FOR_SENDER)
+          .minScore(MIN_SCORE)
           .pendingTransactionsLayerMaxCapacityBytes(
               new PendingTransaction.Remote(
                           new BaseTransactionPoolTest().createEIP1559Transaction(0, KEYS1, 1))
@@ -92,6 +94,7 @@ public class LayersTest extends BaseTransactionPoolTest {
           .maxPrioritizedTransactions(MAX_PRIO_TRANSACTIONS)
           .maxPrioritizedTransactionsByType(Map.of(BLOB, 1))
           .maxFutureBySender(MAX_FUTURE_FOR_SENDER)
+          .minScore(MIN_SCORE)
           .pendingTransactionsLayerMaxCapacityBytes(
               new PendingTransaction.Remote(
                           new BaseTransactionPoolTest().createEIP4844Transaction(0, KEYS1, 1, 1))
@@ -1293,7 +1296,17 @@ static Stream<Arguments> providerPenalized() {
                 .penalizeForSender(S2, 1)
                 .addForSender(S2, 2)
                 .expectedReadyForSenders(S1, 0, S1, 1, S2, 1)
-                .expectedSparseForSender(S2, 2)));
+                .expectedSparseForSender(S2, 2)),
+        Arguments.of(
+            new Scenario("remove below min score")
+                .addForSender(S1, 0) // score 127
+                .expectedPrioritizedForSender(S1, 0)
+                .penalizeForSender(S1, 0) // score 126
+                .expectedPrioritizedForSender(S1, 0)
+                .penalizeForSender(S1, 0) // score 125
+                .expectedPrioritizedForSender(S1, 0)
+                .penalizeForSender(S1, 0) // score 124, removed since decreased score < MIN_SCORE
+                .expectedPrioritizedForSenders()));
   }
 
   private static BlockHeader mockBlockHeader() {