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

Signed-off-by: Fabio Di Fabio <fabio.difabio@consensys.net>

CHANGELOG.md

@@ -7,6 +7,7 @@
 ### Breaking Changes
 
 ### Additions and Improvements
+- 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
 

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 =
+            "Keep a pending transaction in the txpool until its score is greater than or equal to 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

@@ -19,6 +19,7 @@
 import static org.hyperledger.besu.ethereum.eth.transactions.TransactionAddedResult.ALREADY_KNOWN;
 import static org.hyperledger.besu.ethereum.eth.transactions.TransactionAddedResult.REJECTED_UNDERPRICED_REPLACEMENT;
 import static org.hyperledger.besu.ethereum.eth.transactions.TransactionAddedResult.TRY_NEXT_LAYER;
+import static org.hyperledger.besu.ethereum.eth.transactions.layered.TransactionsLayer.RemovalReason.BELOW_MIN_SCORE;
 import static org.hyperledger.besu.ethereum.eth.transactions.layered.TransactionsLayer.RemovalReason.CONFIRMED;
 import static org.hyperledger.besu.ethereum.eth.transactions.layered.TransactionsLayer.RemovalReason.CROSS_LAYER_REPLACED;
 import static org.hyperledger.besu.ethereum.eth.transactions.layered.TransactionsLayer.RemovalReason.EVICTED;
@@ -466,8 +467,12 @@ final void promoteTransactions() {
   @Override
   public void penalize(final PendingTransaction penalizedTransaction) {
     if (pendingTransactions.containsKey(penalizedTransaction.getHash())) {
-      internalPenalize(penalizedTransaction);
-      metrics.incrementPenalized(penalizedTransaction, name());
+      if (penalizedTransaction.getScore() > poolConfig.getMinScore()) {
+        internalPenalize(penalizedTransaction);
+        metrics.incrementPenalized(penalizedTransaction, name());
+      } else {
+        remove(penalizedTransaction, BELOW_MIN_SCORE);
+      }
     } else {
       nextLayer.penalize(penalizedTransaction);
     }

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

@@ -316,7 +316,6 @@ public synchronized List<Transaction> getPriorityTransactions() {
   @Override
   public void selectTransactions(final PendingTransactions.TransactionSelector selector) {
     final List<PendingTransaction> invalidTransactions = new ArrayList<>();
-    final List<PendingTransaction> penalizedTransactions = new ArrayList<>();
     final Set<Address> skipSenders = new HashSet<>();
 
     final Map<Byte, List<SenderPendingTransactions>> candidateTxsByScore;
@@ -351,7 +350,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)
@@ -376,20 +380,13 @@ public void selectTransactions(final PendingTransactions.TransactionSelector sel
     }
 
     ethScheduler.scheduleTxWorkerTask(
-        () -> {
-          invalidTransactions.forEach(
-              invalidTx -> {
-                synchronized (this) {
-                  prioritizedTransactions.remove(invalidTx, INVALIDATED);
-                }
-              });
-          penalizedTransactions.forEach(
-              penalizedTx -> {
-                synchronized (this) {
-                  prioritizedTransactions.internalPenalize(penalizedTx);
-                }
-              });
-        });
+        () ->
+            invalidTransactions.forEach(
+                invalidTx -> {
+                  synchronized (this) {
+                    prioritizedTransactions.remove(invalidTx, INVALIDATED);
+                  }
+                }));
   }
 
   @Override

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

@@ -118,7 +118,8 @@ enum RemovalReason {
     PROMOTED,
     REPLACED,
     RECONCILED,
-    BELOW_BASE_FEE;
+    BELOW_BASE_FEE,
+    BELOW_MIN_SCORE;
 
     private final String label;
 

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

@@ -68,12 +68,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))
@@ -86,6 +88,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))
@@ -1332,7 +1335,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() {