Always enforce promotion filter for transactions in the prioritized layer

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

@@ -20,6 +20,7 @@
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPoolConfiguration;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPoolMetrics;
 
+import java.util.List;
 import java.util.Map;
 import java.util.NavigableMap;
 import java.util.TreeMap;
@@ -28,6 +29,10 @@
 import java.util.function.Predicate;
 import java.util.stream.Stream;
 
+/**
+ * Holds the current set of executable pending transactions, that are candidate for inclusion on
+ * next block. The pending transactions are kept sorted by paid fee descending.
+ */
 public abstract class AbstractPrioritizedTransactions extends AbstractSequentialTransactionsLayer {
   protected final TreeSet<PendingTransaction> orderByFee;
 
@@ -77,6 +82,12 @@ protected void internalReplaced(final PendingTransaction replacedTx) {
   }
 
   private boolean hasPriority(final PendingTransaction pendingTransaction) {
+    // if it does not pass the promotion filter, then has not priority
+    if (!promotionFilter(pendingTransaction)) {
+      return false;
+    }
+
+    // if there is space add it, otherwise check if it has more value than the last one
     if (orderByFee.size() < poolConfig.getMaxPrioritizedTransactions()) {
       return true;
     }
@@ -104,8 +115,9 @@ protected void internalRemove(
   }
 
   @Override
-  public PendingTransaction promote(final Predicate<PendingTransaction> promotionFilter) {
-    return null;
+  public List<PendingTransaction> promote(
+      final Predicate<PendingTransaction> promotionFilter, final long l, final int freeSlots) {
+    return List.of();
   }
 
   @Override

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

@@ -264,7 +264,7 @@ private TransactionAddedResult addToNextLayer(
         distance);
   }
 
-  private TransactionAddedResult addToNextLayer(
+  protected TransactionAddedResult addToNextLayer(
       final NavigableMap<Long, PendingTransaction> senderTxs,
       final PendingTransaction pendingTransaction,
       final int distance) {
@@ -304,7 +304,7 @@ private void evict(final long spaceToFree, final int txsToEvict) {
       while ((evictedSize < spaceToFree || txsToEvict > evictedCount)
           && !lessReadySenderTxs.isEmpty()) {
         lastTx = lessReadySenderTxs.pollLastEntry().getValue();
-        processEvict(lessReadySenderTxs, lastTx);
+        processEvict(lessReadySenderTxs, lastTx, EVICTED);
         ++evictedCount;
         evictedSize += lastTx.memorySize();
         // evicted can always be added to the next layer
@@ -371,11 +371,13 @@ protected PendingTransaction processRemove(
   }
 
   protected PendingTransaction processEvict(
-      final NavigableMap<Long, PendingTransaction> senderTxs, final PendingTransaction evictedTx) {
+      final NavigableMap<Long, PendingTransaction> senderTxs,
+      final PendingTransaction evictedTx,
+      final RemovalReason reason) {
     final PendingTransaction removedTx = pendingTransactions.remove(evictedTx.getHash());
     if (removedTx != null) {
       decreaseSpaceUsed(evictedTx);
-      metrics.incrementRemoved(evictedTx.isReceivedFromLocalSource(), EVICTED.label(), name());
+      metrics.incrementRemoved(evictedTx.isReceivedFromLocalSource(), reason.label(), name());
       internalEvict(senderTxs, removedTx);
     }
     return removedTx;
@@ -398,22 +400,20 @@ public final void blockAdded(
     nextLayer.blockAdded(feeMarket, blockHeader, maxConfirmedNonceBySender);
     maxConfirmedNonceBySender.forEach(this::confirmed);
     internalBlockAdded(blockHeader, feeMarket);
+    promoteTransactions();
   }
 
   protected abstract void internalBlockAdded(
       final BlockHeader blockHeader, final FeeMarket feeMarket);
 
   final void promoteTransactions() {
-    int freeSlots = maxTransactionsNumber() - pendingTransactions.size();
+    final int freeSlots = maxTransactionsNumber() - pendingTransactions.size();
+    final long freeSpace = cacheFreeSpace();
 
-    while (cacheFreeSpace() > 0 && freeSlots > 0) {
-      final var promotedTx = nextLayer.promote(this::promotionFilter);
-      if (promotedTx != null) {
-        processAdded(promotedTx);
-        --freeSlots;
-      } else {
-        break;
-      }
+    if (freeSlots > 0 && freeSpace > 0) {
+      nextLayer
+          .promote(this::promotionFilter, cacheFreeSpace(), freeSlots)
+          .forEach(this::processAdded);
     }
   }
 
@@ -444,8 +444,6 @@ private void confirmed(final Address sender, final long maxConfirmedNonce) {
         internalConfirmed(senderTxs, sender, maxConfirmedNonce, highestNonceRemovedTx);
       }
     }
-
-    promoteTransactions();
   }
 
   protected abstract void internalConfirmed(

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

@@ -14,6 +14,8 @@
  */
 package org.hyperledger.besu.ethereum.eth.transactions.layered;
 
+import static org.hyperledger.besu.ethereum.eth.transactions.layered.TransactionsLayer.RemovalReason.BELOW_BASE_FEE;
+
 import org.hyperledger.besu.datatypes.Wei;
 import org.hyperledger.besu.ethereum.core.BlockHeader;
 import org.hyperledger.besu.ethereum.core.Transaction;
@@ -27,17 +29,10 @@
 import java.util.Optional;
 import java.util.function.BiFunction;
 import java.util.function.Supplier;
-import java.util.stream.Collectors;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-/**
- * Holds the current set of pending transactions with the ability to iterate them based on priority
- * for mining or look-up by hash.
- *
- * <p>This class is safe for use across multiple threads.
- */
 public class BaseFeePrioritizedTransactions extends AbstractPrioritizedTransactions {
 
   private static final Logger LOG = LoggerFactory.getLogger(BaseFeePrioritizedTransactions.class);
@@ -69,6 +64,15 @@ protected int compareByFee(final PendingTransaction pt1, final PendingTransactio
         .compare(pt1, pt2);
   }
 
+  /**
+   * On base fee markets when a new block is added we can calculate the base fee for the next block
+   * and use it to keep only pending transactions willing to pay at least that fee in the
+   * prioritized layer, since only these transactions are executable, while all the other can be
+   * demoted to the next layer.
+   *
+   * @param blockHeader the header of the added block
+   * @param feeMarket the fee market
+   */
   @Override
   protected void internalBlockAdded(final BlockHeader blockHeader, final FeeMarket feeMarket) {
     final Wei newNextBlockBaseFee = calculateNextBlockBaseFee(feeMarket, blockHeader);
@@ -81,7 +85,48 @@ protected void internalBlockAdded(final BlockHeader blockHeader, final FeeMarket
 
     nextBlockBaseFee = Optional.of(newNextBlockBaseFee);
     orderByFee.clear();
-    orderByFee.addAll(pendingTransactions.values());
+
+    final var itTxsBySender = txsBySender.entrySet().iterator();
+    while (itTxsBySender.hasNext()) {
+      final var senderTxs = itTxsBySender.next().getValue();
+
+      Optional<Long> maybeFirstUnderpricedNonce = Optional.empty();
+
+      for (final var e : senderTxs.entrySet()) {
+        final PendingTransaction tx = e.getValue();
+        // it must pass the promotion filter to be prioritized
+        if (promotionFilter(tx)) {
+          orderByFee.add(tx);
+        } else {
+          // otherwise sender txs starting from this nonce need to be demoted to next layer,
+          // and we can go to next sender
+          maybeFirstUnderpricedNonce = Optional.of(e.getKey());
+          break;
+        }
+      }
+
+      maybeFirstUnderpricedNonce.ifPresent(
+          nonce -> {
+            // demote all txs after the first underpriced to the next layer, because none of them is
+            // executable now, and we can avoid sorting them until they are candidate for execution
+            // again
+            final var demoteTxs = senderTxs.tailMap(nonce, true);
+            while (!demoteTxs.isEmpty()) {
+              final PendingTransaction demoteTx = demoteTxs.pollLastEntry().getValue();
+              LOG.atTrace()
+                  .setMessage("Demoting tx {} with max gas price below next block base fee {}")
+                  .addArgument(demoteTx::toTraceLog)
+                  .addArgument(newNextBlockBaseFee::toHumanReadableString)
+                  .log();
+              processEvict(senderTxs, demoteTx, BELOW_BASE_FEE);
+              addToNextLayer(senderTxs, demoteTx, 0);
+            }
+          });
+
+      if (senderTxs.isEmpty()) {
+        itTxsBySender.remove();
+      }
+    }
   }
 
   private Wei calculateNextBlockBaseFee(final FeeMarket feeMarket, final BlockHeader blockHeader) {
@@ -101,10 +146,7 @@ protected boolean promotionFilter(final PendingTransaction pendingTransaction) {
     return nextBlockBaseFee
         .map(
             baseFee ->
-                pendingTransaction
-                    .getTransaction()
-                    .getEffectiveGasPrice(nextBlockBaseFee)
-                    .greaterOrEqualThan(baseFee))
+                pendingTransaction.getTransaction().getMaxGasPrice().greaterOrEqualThan(baseFee))
         .orElse(false);
   }
 
@@ -115,13 +157,6 @@ protected String internalLogStats() {
       return "Basefee Prioritized: Empty";
     }
 
-    final var baseFeePartition =
-        stream()
-            .map(PendingTransaction::getTransaction)
-            .collect(
-                Collectors.partitioningBy(
-                    tx -> tx.getMaxGasPrice().greaterOrEqualThan(nextBlockBaseFee.get()),
-                    Collectors.counting()));
     final Transaction highest = orderByFee.last().getTransaction();
     final Transaction lowest = orderByFee.first().getTransaction();
 
@@ -145,10 +180,6 @@ protected String internalLogStats() {
         + ", hash: "
         + lowest.getHash()
         + "], next block base fee: "
-        + nextBlockBaseFee.get().toHumanReadableString()
-        + ", above next base fee: "
-        + baseFeePartition.get(true)
-        + ", below next base fee: "
-        + baseFeePartition.get(false);
+        + nextBlockBaseFee.get().toHumanReadableString();
   }
 }

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

@@ -115,8 +115,9 @@ public OptionalLong getCurrentNonceFor(final Address sender) {
   }
 
   @Override
-  public PendingTransaction promote(final Predicate<PendingTransaction> promotionFilter) {
-    return null;
+  public List<PendingTransaction> promote(
+      final Predicate<PendingTransaction> promotionFilter, final long l, final int freeSlots) {
+    return List.of();
   }
 
   @Override

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

@@ -25,7 +25,9 @@
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPoolMetrics;
 import org.hyperledger.besu.ethereum.mainnet.feemarket.FeeMarket;
 
+import java.util.ArrayList;
 import java.util.Comparator;
+import java.util.List;
 import java.util.Map;
 import java.util.NavigableMap;
 import java.util.NavigableSet;
@@ -139,30 +141,51 @@ public Stream<PendingTransaction> stream() {
   }
 
   @Override
-  public PendingTransaction promote(final Predicate<PendingTransaction> promotionFilter) {
-
-    final var maybePromotedTx =
-        orderByMaxFee.descendingSet().stream()
-            .filter(candidateTx -> promotionFilter.test(candidateTx))
-            .findFirst();
-
-    return maybePromotedTx
-        .map(
-            promotedTx -> {
-              final var senderTxs = txsBySender.get(promotedTx.getSender());
-              // we always promote the first tx of a sender, so remove the first entry
-              senderTxs.pollFirstEntry();
-              processRemove(senderTxs, promotedTx.getTransaction(), PROMOTED);
-
-              // now that we have space, promote from the next layer
-              promoteTransactions();
-
-              if (senderTxs.isEmpty()) {
-                txsBySender.remove(promotedTx.getSender());
-              }
-              return promotedTx;
-            })
-        .orElse(null);
+  public List<PendingTransaction> promote(
+      final Predicate<PendingTransaction> promotionFilter,
+      final long freeSpace,
+      final int freeSlots) {
+    long accSpace = 0;
+    final List<PendingTransaction> promotedTxs = new ArrayList<>();
+
+    // first find all txs that can be promoted
+    search:
+    for (final var senderFirstTx : orderByMaxFee.descendingSet()) {
+      final var senderTxs = txsBySender.get(senderFirstTx.getSender());
+      for (final var candidateTx : senderTxs.values()) {
+        if (promotionFilter.test(candidateTx)) {
+          accSpace += candidateTx.memorySize();
+          if (promotedTxs.size() < freeSlots && accSpace <= freeSpace) {
+            promotedTxs.add(candidateTx);
+          } else {
+            // no room for more txs the search is over exit the loops
+            break search;
+          }
+        } else {
+          // skip remaining txs for this sender to avoid gaps
+          break;
+        }
+      }
+    }
+
+    // then remove promoted txs from this layer
+    promotedTxs.forEach(
+        promotedTx -> {
+          final var sender = promotedTx.getSender();
+          final var senderTxs = txsBySender.get(sender);
+          senderTxs.remove(promotedTx.getNonce());
+          processRemove(senderTxs, promotedTx.getTransaction(), PROMOTED);
+          if (senderTxs.isEmpty()) {
+            txsBySender.remove(sender);
+          }
+        });
+
+    if (!promotedTxs.isEmpty()) {
+      // since we removed some txs we can try to promote from next layer
+      promoteTransactions();
+    }
+
+    return promotedTxs;
   }
 
   @Override