Layered txpool tuning for blob transactions (#6940)

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

CHANGELOG.md

@@ -42,7 +42,7 @@
 - Add RPC errors metric [#6919](https://github.com/hyperledger/besu/pull/6919/)
 - Add `rlp decode` subcommand to decode IBFT/QBFT extraData to validator list [#6895](https://github.com/hyperledger/besu/pull/6895)
 - Allow users to specify which plugins are registered [#6700](https://github.com/hyperledger/besu/pull/6700)
-
+- Layered txpool tuning for blob transactions [#6940](https://github.com/hyperledger/besu/pull/6940)
 
 ### Bug fixes
 - Fix txpool dump/restore race condition [#6665](https://github.com/hyperledger/besu/pull/6665)

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

@@ -27,6 +27,7 @@ import org.hyperledger.besu.cli.converter.PercentageConverter;
 import org.hyperledger.besu.cli.util.CommandLineUtils;
 import org.hyperledger.besu.config.GenesisConfigOptions;
 import org.hyperledger.besu.datatypes.Address;
+import org.hyperledger.besu.datatypes.TransactionType;
 import org.hyperledger.besu.datatypes.Wei;
 import org.hyperledger.besu.ethereum.eth.transactions.ImmutableTransactionPoolConfiguration;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPoolConfiguration;
@@ -37,6 +38,7 @@ import org.hyperledger.besu.util.number.Percentage;
 import java.io.File;
 import java.time.Duration;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 
 import picocli.CommandLine;
@@ -155,6 +157,8 @@ public class TransactionPoolOptions implements CLIOptions<TransactionPoolConfigu
   static class Layered {
     private static final String TX_POOL_LAYER_MAX_CAPACITY = "--tx-pool-layer-max-capacity";
     private static final String TX_POOL_MAX_PRIORITIZED = "--tx-pool-max-prioritized";
+    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";
 
     @CommandLine.Option(
@@ -175,6 +179,16 @@ public class TransactionPoolOptions implements CLIOptions<TransactionPoolConfigu
     Integer txPoolMaxPrioritized =
         TransactionPoolConfiguration.DEFAULT_MAX_PRIORITIZED_TRANSACTIONS;
 
+    @CommandLine.Option(
+        names = {TX_POOL_MAX_PRIORITIZED_BY_TYPE},
+        paramLabel = "MAP<TYPE,INTEGER>",
+        split = ",",
+        description =
+            "Max number of pending transactions, of a specific type, that are prioritized and thus kept sorted (default: ${DEFAULT-VALUE})",
+        arity = "1")
+    Map<TransactionType, Integer> txPoolMaxPrioritizedByType =
+        TransactionPoolConfiguration.DEFAULT_MAX_PRIORITIZED_TRANSACTIONS_BY_TYPE;
+
     @CommandLine.Option(
         names = {TX_POOL_MAX_FUTURE_BY_SENDER},
         paramLabel = MANDATORY_INTEGER_FORMAT_HELP,
@@ -297,6 +311,8 @@ public class TransactionPoolOptions implements CLIOptions<TransactionPoolConfigu
     options.layeredOptions.txPoolLayerMaxCapacity =
         config.getPendingTransactionsLayerMaxCapacityBytes();
     options.layeredOptions.txPoolMaxPrioritized = config.getMaxPrioritizedTransactions();
+    options.layeredOptions.txPoolMaxPrioritizedByType =
+        config.getMaxPrioritizedTransactionsByType();
     options.layeredOptions.txPoolMaxFutureBySender = config.getMaxFutureBySender();
     options.sequencedOptions.txPoolLimitByAccountPercentage =
         config.getTxPoolLimitByAccountPercentage();
@@ -354,6 +370,7 @@ public class TransactionPoolOptions implements CLIOptions<TransactionPoolConfigu
         .minGasPrice(minGasPrice)
         .pendingTransactionsLayerMaxCapacityBytes(layeredOptions.txPoolLayerMaxCapacity)
         .maxPrioritizedTransactions(layeredOptions.txPoolMaxPrioritized)
+        .maxPrioritizedTransactionsByType(layeredOptions.txPoolMaxPrioritizedByType)
         .maxFutureBySender(layeredOptions.txPoolMaxFutureBySender)
         .txPoolLimitByAccountPercentage(sequencedOptions.txPoolLimitByAccountPercentage)
         .txPoolMaxSize(sequencedOptions.txPoolMaxSize)

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

@@ -21,11 +21,14 @@ import static org.hyperledger.besu.ethereum.eth.transactions.TransactionPoolConf
 
 import org.hyperledger.besu.cli.converter.DurationMillisConverter;
 import org.hyperledger.besu.datatypes.Address;
+import org.hyperledger.besu.datatypes.TransactionType;
 import org.hyperledger.besu.datatypes.Wei;
 import org.hyperledger.besu.ethereum.eth.transactions.ImmutableTransactionPoolConfiguration;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPoolConfiguration;
 import org.hyperledger.besu.util.number.Percentage;
 
+import java.io.IOException;
+import java.nio.file.Path;
 import java.time.Duration;
 
 import org.junit.jupiter.api.Test;
@@ -369,6 +372,52 @@ public class TransactionPoolOptionsTest
         "-1");
   }
 
+  @Test
+  public void maxPrioritizedTxsPerType() {
+    final int maxBlobs = 2;
+    final int maxFrontier = 200;
+    internalTestSuccess(
+        config -> {
+          assertThat(config.getMaxPrioritizedTransactionsByType().get(TransactionType.BLOB))
+              .isEqualTo(maxBlobs);
+          assertThat(config.getMaxPrioritizedTransactionsByType().get(TransactionType.FRONTIER))
+              .isEqualTo(maxFrontier);
+        },
+        "--tx-pool-max-prioritized-by-type",
+        "BLOB=" + maxBlobs + ",FRONTIER=" + maxFrontier);
+  }
+
+  @Test
+  public void maxPrioritizedTxsPerTypeConfigFile() throws IOException {
+    final int maxBlobs = 2;
+    final int maxFrontier = 200;
+    final Path tempConfigFilePath =
+        createTempFile(
+            "config",
+            String.format(
+                """
+    tx-pool-max-prioritized-by-type=["BLOB=%s","FRONTIER=%s"]
+    """,
+                maxBlobs, maxFrontier));
+    internalTestSuccess(
+        config -> {
+          assertThat(config.getMaxPrioritizedTransactionsByType().get(TransactionType.BLOB))
+              .isEqualTo(maxBlobs);
+          assertThat(config.getMaxPrioritizedTransactionsByType().get(TransactionType.FRONTIER))
+              .isEqualTo(maxFrontier);
+        },
+        "--config-file",
+        tempConfigFilePath.toString());
+  }
+
+  @Test
+  public void maxPrioritizedTxsPerTypeWrongTxType() {
+    internalTestFailure(
+        "Invalid value for option '--tx-pool-max-prioritized-by-type' (MAP<TYPE,INTEGER>): expected one of [FRONTIER, ACCESS_LIST, EIP1559, BLOB] (case-insensitive) but was 'WRONG_TYPE'",
+        "--tx-pool-max-prioritized-by-type",
+        "WRONG_TYPE=1");
+  }
+
   @Override
   protected TransactionPoolConfiguration createDefaultDomainObject() {
     return TransactionPoolConfiguration.DEFAULT;

besu/src/test/resources/everything_config.toml

@@ -185,6 +185,7 @@ tx-pool-save-file="txpool.dump"
 ## Layered
 tx-pool-layer-max-capacity=12345678
 tx-pool-max-prioritized=9876
+tx-pool-max-prioritized-by-type=["BLOB=10","FRONTIER=100"]
 tx-pool-max-future-by-sender=321
 ## Legacy/Sequenced
 tx-pool-retention-hours=999

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

@@ -149,6 +149,16 @@ public class TransactionPool implements BlockAddedObserver {
                     .map(Address::toHexString)
                     .collect(Collectors.joining(",")))
         .log();
+    // log the max prioritized txs by type
+    LOG_FOR_REPLAY
+        .atTrace()
+        .setMessage("{}")
+        .addArgument(
+            () ->
+                configuration.getMaxPrioritizedTransactionsByType().entrySet().stream()
+                    .map(e -> e.getKey().name() + "=" + e.getValue())
+                    .collect(Collectors.joining(",")))
+        .log();
   }
 
   @VisibleForTesting

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

@@ -15,6 +15,7 @@
 package org.hyperledger.besu.ethereum.eth.transactions;
 
 import org.hyperledger.besu.datatypes.Address;
+import org.hyperledger.besu.datatypes.TransactionType;
 import org.hyperledger.besu.datatypes.Wei;
 import org.hyperledger.besu.plugin.services.TransactionPoolValidatorService;
 import org.hyperledger.besu.plugin.services.txvalidator.PluginTransactionPoolValidator;
@@ -24,6 +25,8 @@ import org.hyperledger.besu.util.number.Percentage;
 
 import java.io.File;
 import java.time.Duration;
+import java.util.EnumMap;
+import java.util.Map;
 import java.util.Set;
 
 import org.immutables.value.Value;
@@ -71,6 +74,8 @@ public interface TransactionPoolConfiguration {
   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;
+  EnumMap<TransactionType, Integer> DEFAULT_MAX_PRIORITIZED_TRANSACTIONS_BY_TYPE =
+      new EnumMap<>(Map.of(TransactionType.BLOB, 6));
   int DEFAULT_MAX_FUTURE_BY_SENDER = 200;
   Implementation DEFAULT_TX_POOL_IMPLEMENTATION = Implementation.LAYERED;
   Set<Address> DEFAULT_PRIORITY_SENDERS = Set.of();
@@ -148,6 +153,11 @@ public interface TransactionPoolConfiguration {
     return DEFAULT_MAX_PRIORITIZED_TRANSACTIONS;
   }
 
+  @Value.Default
+  default Map<TransactionType, Integer> getMaxPrioritizedTransactionsByType() {
+    return DEFAULT_MAX_PRIORITIZED_TRANSACTIONS_BY_TYPE;
+  }
+
   @Value.Default
   default int getMaxFutureBySender() {
     return DEFAULT_MAX_FUTURE_BY_SENDER;

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

@@ -15,6 +15,7 @@
 package org.hyperledger.besu.ethereum.eth.transactions.layered;
 
 import org.hyperledger.besu.datatypes.Address;
+import org.hyperledger.besu.datatypes.TransactionType;
 import org.hyperledger.besu.ethereum.core.MiningParameters;
 import org.hyperledger.besu.ethereum.eth.transactions.BlobCache;
 import org.hyperledger.besu.ethereum.eth.transactions.PendingTransaction;
@@ -87,6 +88,15 @@ public abstract class AbstractPrioritizedTransactions extends AbstractSequential
   }
 
   private boolean hasPriority(final PendingTransaction pendingTransaction) {
+    // check if there is space for that tx type
+    final var txType = pendingTransaction.getTransaction().getType();
+    if (txCountByType[txType.ordinal()]
+        >= poolConfig
+            .getMaxPrioritizedTransactionsByType()
+            .getOrDefault(txType, Integer.MAX_VALUE)) {
+      return false;
+    }
+
     // if it does not pass the promotion filter, then has not priority
     if (!promotionFilter(pendingTransaction)) {
       return false;
@@ -123,10 +133,32 @@ public abstract class AbstractPrioritizedTransactions extends AbstractSequential
   public List<PendingTransaction> promote(
       final Predicate<PendingTransaction> promotionFilter,
       final long freeSpace,
-      final int freeSlots) {
+      final int freeSlots,
+      final int[] remainingPromotionsPerType) {
     return List.of();
   }
 
+  /**
+   * Here the max number of txs of a specific type that can be promoted, is defined by the
+   * configuration, so we return the difference between the configured max and the current count of
+   * txs for each type
+   *
+   * @return an array containing the max amount of txs that can be promoted for each type
+   */
+  @Override
+  protected int[] getRemainingPromotionsPerType() {
+    final var allTypes = TransactionType.values();
+    final var remainingPromotionsPerType = new int[allTypes.length];
+    for (int i = 0; i < allTypes.length; i++) {
+      remainingPromotionsPerType[i] =
+          poolConfig
+                  .getMaxPrioritizedTransactionsByType()
+                  .getOrDefault(allTypes[i], Integer.MAX_VALUE)
+              - txCountByType[i];
+    }
+    return remainingPromotionsPerType;
+  }
+
   @Override
   public Stream<PendingTransaction> stream() {
     return orderByFee.descendingSet().stream();

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

@@ -61,6 +61,13 @@ import org.slf4j.LoggerFactory;
 public abstract class AbstractTransactionsLayer implements TransactionsLayer {
   private static final Logger LOG = LoggerFactory.getLogger(AbstractTransactionsLayer.class);
   private static final NavigableMap<Long, PendingTransaction> EMPTY_SENDER_TXS = new TreeMap<>();
+  private static final int[] UNLIMITED_PROMOTIONS_PER_TYPE =
+      new int[TransactionType.values().length];
+
+  static {
+    Arrays.fill(UNLIMITED_PROMOTIONS_PER_TYPE, Integer.MAX_VALUE);
+  }
+
   protected final TransactionPoolConfiguration poolConfig;
   protected final TransactionsLayer nextLayer;
   protected final BiFunction<PendingTransaction, PendingTransaction, Boolean>
@@ -170,7 +177,7 @@ public abstract class AbstractTransactionsLayer implements TransactionsLayer {
 
       if (!maybeFull()) {
         // if there is space try to see if the added tx filled some gaps
-        tryFillGap(addStatus, pendingTransaction);
+        tryFillGap(addStatus, pendingTransaction, getRemainingPromotionsPerType());
       }
 
       notifyTransactionAdded(pendingTransaction);
@@ -207,16 +214,21 @@ public abstract class AbstractTransactionsLayer implements TransactionsLayer {
   }
 
   private void tryFillGap(
-      final TransactionAddedResult addStatus, final PendingTransaction pendingTransaction) {
+      final TransactionAddedResult addStatus,
+      final PendingTransaction pendingTransaction,
+      final int[] remainingPromotionsPerType) {
     // it makes sense to fill gaps only if the add is not a replacement and this layer does not
     // allow gaps
     if (!addStatus.isReplacement() && !gapsAllowed()) {
       final PendingTransaction promotedTx =
-          nextLayer.promoteFor(pendingTransaction.getSender(), pendingTransaction.getNonce());
+          nextLayer.promoteFor(
+              pendingTransaction.getSender(),
+              pendingTransaction.getNonce(),
+              remainingPromotionsPerType);
       if (promotedTx != null) {
         processAdded(promotedTx);
         if (!maybeFull()) {
-          tryFillGap(ADDED, promotedTx);
+          tryFillGap(ADDED, promotedTx, remainingPromotionsPerType);
         }
       }
     }
@@ -251,22 +263,30 @@ public abstract class AbstractTransactionsLayer implements TransactionsLayer {
       final PendingTransaction pendingTransaction);
 
   @Override
-  public PendingTransaction promoteFor(final Address sender, final long nonce) {
+  public PendingTransaction promoteFor(
+      final Address sender, final long nonce, final int[] remainingPromotionsPerType) {
     final var senderTxs = txsBySender.get(sender);
     if (senderTxs != null) {
       long expectedNonce = nonce + 1;
       if (senderTxs.firstKey() == expectedNonce) {
-        final PendingTransaction promotedTx = senderTxs.pollFirstEntry().getValue();
-        processRemove(senderTxs, promotedTx.getTransaction(), PROMOTED);
-        metrics.incrementRemoved(promotedTx, "promoted", name());
-
-        if (senderTxs.isEmpty()) {
-          txsBySender.remove(sender);
+        final var candidateTx = senderTxs.firstEntry().getValue();
+        final var txType = candidateTx.getTransaction().getType();
+
+        if (remainingPromotionsPerType[txType.ordinal()] > 0) {
+          senderTxs.pollFirstEntry();
+          processRemove(senderTxs, candidateTx.getTransaction(), PROMOTED);
+          metrics.incrementRemoved(candidateTx, "promoted", name());
+
+          if (senderTxs.isEmpty()) {
+            txsBySender.remove(sender);
+          }
+          --remainingPromotionsPerType[txType.ordinal()];
+          return candidateTx;
         }
-        return promotedTx;
+        return null;
       }
     }
-    return nextLayer.promoteFor(sender, nonce);
+    return nextLayer.promoteFor(sender, nonce, remainingPromotionsPerType);
   }
 
   private TransactionAddedResult addToNextLayer(
@@ -425,11 +445,24 @@ public abstract class AbstractTransactionsLayer implements TransactionsLayer {
 
     if (freeSlots > 0 && freeSpace > 0) {
       nextLayer
-          .promote(this::promotionFilter, cacheFreeSpace(), freeSlots)
+          .promote(
+              this::promotionFilter, cacheFreeSpace(), freeSlots, getRemainingPromotionsPerType())
           .forEach(this::processAdded);
     }
   }
 
+  /**
+   * How many txs of a specified type can be promoted? This make sense when a max number of txs of a
+   * type can be included in a single block (ex. blob txs), to avoid filling the layer with more txs
+   * than the useful ones. By default, there are no limits, but each layer can define its own
+   * policy.
+   *
+   * @return an array containing the max amount of txs that can be promoted for each type
+   */
+  protected int[] getRemainingPromotionsPerType() {
+    return Arrays.copyOf(UNLIMITED_PROMOTIONS_PER_TYPE, UNLIMITED_PROMOTIONS_PER_TYPE.length);
+  }
+
   private void confirmed(final Address sender, final long maxConfirmedNonce) {
     final var senderTxs = txsBySender.get(sender);
 

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

@@ -149,6 +149,7 @@ public class BaseFeePrioritizedTransactions extends AbstractPrioritizedTransacti
 
   @Override
   protected boolean promotionFilter(final PendingTransaction pendingTransaction) {
+
     // check if the tx is willing to pay at least the base fee
     if (nextBlockBaseFee
         .map(pendingTransaction.getTransaction().getMaxGasPrice()::lessThan)

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

@@ -122,7 +122,8 @@ public class EndLayer implements TransactionsLayer {
   public List<PendingTransaction> promote(
       final Predicate<PendingTransaction> promotionFilter,
       final long freeSpace,
-      final int freeSlots) {
+      final int freeSlots,
+      final int[] remainingPromotionsPerType) {
     return List.of();
   }
 
@@ -152,7 +153,8 @@ public class EndLayer implements TransactionsLayer {
   }
 
   @Override
-  public PendingTransaction promoteFor(final Address sender, final long nonce) {
+  public PendingTransaction promoteFor(
+      final Address sender, final long nonce, final int[] remainingPromotionsPerType) {
     return null;
   }
 

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

@@ -146,7 +146,8 @@ public class ReadyTransactions extends AbstractSequentialTransactionsLayer {
   public List<PendingTransaction> promote(
       final Predicate<PendingTransaction> promotionFilter,
       final long freeSpace,
-      final int freeSlots) {
+      final int freeSlots,
+      final int[] remainingPromotionsPerType) {
     long accumulatedSpace = 0;
     final List<PendingTransaction> promotedTxs = new ArrayList<>();
 
@@ -155,10 +156,12 @@ public class ReadyTransactions extends AbstractSequentialTransactionsLayer {
     for (final var senderFirstTx : orderByMaxFee.descendingSet()) {
       final var senderTxs = txsBySender.get(senderFirstTx.getSender());
       for (final var candidateTx : senderTxs.values()) {
-        if (promotionFilter.test(candidateTx)) {
+        final var txType = candidateTx.getTransaction().getType();
+        if (promotionFilter.test(candidateTx) && remainingPromotionsPerType[txType.ordinal()] > 0) {
           accumulatedSpace += candidateTx.memorySize();
           if (promotedTxs.size() < freeSlots && accumulatedSpace <= freeSpace) {
             promotedTxs.add(candidateTx);
+            --remainingPromotionsPerType[txType.ordinal()];
           } else {
             // no room for more txs the search is over exit the loops
             break search;

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

@@ -147,7 +147,8 @@ public class SparseTransactions extends AbstractTransactionsLayer {
   public List<PendingTransaction> promote(
       final Predicate<PendingTransaction> promotionFilter,
       final long freeSpace,
-      final int freeSlots) {
+      final int freeSlots,
+      final int[] remainingPromotionsPerType) {
     long accumulatedSpace = 0;
     final List<PendingTransaction> promotedTxs = new ArrayList<>();
 
@@ -158,11 +159,12 @@ public class SparseTransactions extends AbstractTransactionsLayer {
       final var senderSeqTxs = getSequentialSubset(txsBySender.get(sender));
 
       for (final var candidateTx : senderSeqTxs.values()) {
-
-        if (promotionFilter.test(candidateTx)) {
+        final var txType = candidateTx.getTransaction().getType();
+        if (promotionFilter.test(candidateTx) && remainingPromotionsPerType[txType.ordinal()] > 0) {
           accumulatedSpace += candidateTx.memorySize();
           if (promotedTxs.size() < freeSlots && accumulatedSpace <= freeSpace) {
             promotedTxs.add(candidateTx);
+            --remainingPromotionsPerType[txType.ordinal()];
           } else {
             // no room for more txs the search is over exit the loops
             break search;

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

@@ -70,7 +70,10 @@ public interface TransactionsLayer {
   OptionalLong getCurrentNonceFor(Address sender);
 
   List<PendingTransaction> promote(
-      Predicate<PendingTransaction> promotionFilter, final long freeSpace, final int freeSlots);
+      Predicate<PendingTransaction> promotionFilter,
+      final long freeSpace,
+      final int freeSlots,
+      final int[] remainingPromotionsPerType);
 
   long subscribeToAdded(PendingTransactionAddedListener listener);
 
@@ -80,7 +83,7 @@ public interface TransactionsLayer {
 
   void unsubscribeFromDropped(long id);
 
-  PendingTransaction promoteFor(Address sender, long nonce);
+  PendingTransaction promoteFor(Address sender, long nonce, final int[] remainingPromotionsPerType);
 
   void notifyAdded(PendingTransaction pendingTransaction);
 

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

@@ -18,6 +18,7 @@ import static org.assertj.core.api.Assertions.assertThat;
 import static org.hyperledger.besu.ethereum.eth.transactions.TransactionAddedResult.ADDED;
 import static org.hyperledger.besu.ethereum.eth.transactions.TransactionAddedResult.DROPPED;
 
+import org.hyperledger.besu.datatypes.TransactionType;
 import org.hyperledger.besu.datatypes.Wei;
 import org.hyperledger.besu.ethereum.core.BlockHeader;
 import org.hyperledger.besu.ethereum.core.MiningParameters;
@@ -30,8 +31,10 @@ import org.hyperledger.besu.ethereum.eth.transactions.TransactionPoolMetrics;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPoolReplacementHandler;
 
 import java.util.ArrayList;
+import java.util.EnumMap;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Map;
 import java.util.function.BiFunction;
 import java.util.stream.IntStream;
 
@@ -39,6 +42,8 @@ import org.junit.jupiter.api.Test;
 
 public abstract class AbstractPrioritizedTransactionsTestBase extends BaseTransactionPoolTest {
   protected static final int MAX_TRANSACTIONS = 5;
+  protected static final EnumMap<TransactionType, Integer> MAX_TRANSACTIONS_BY_TYPE =
+      new EnumMap<>(Map.of(TransactionType.BLOB, 2));
   protected final TransactionPoolMetrics txPoolMetrics = new TransactionPoolMetrics(metricsSystem);
   protected final EvictCollectorLayer evictCollector = new EvictCollectorLayer(txPoolMetrics);
   protected final MiningParameters miningParameters =
@@ -49,6 +54,7 @@ public abstract class AbstractPrioritizedTransactionsTestBase extends BaseTransa
       getSorter(
           ImmutableTransactionPoolConfiguration.builder()
               .maxPrioritizedTransactions(MAX_TRANSACTIONS)
+              .maxPrioritizedTransactionsByType(MAX_TRANSACTIONS_BY_TYPE)
               .maxFutureBySender(MAX_TRANSACTIONS)
               .build(),
           miningParameters);

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

@@ -35,6 +35,7 @@ import org.hyperledger.besu.ethereum.eth.transactions.TransactionPoolConfigurati
 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.Optional;
@@ -111,6 +112,9 @@ public class BaseFeePrioritizedTransactionsTest extends AbstractPrioritizedTrans
         originalTransaction.getType(),
         originalTransaction.getNonce(),
         originalTransaction.getMaxGasPrice().multiply(2),
+        originalTransaction.getMaxGasPrice().multiply(2).divide(10),
+        originalTransaction.getPayload().size(),
+        originalTransaction.getBlobCount(),
         keys);
   }
 
@@ -218,4 +222,72 @@ public class BaseFeePrioritizedTransactionsTest extends AbstractPrioritizedTrans
     shouldPrioritizeValueThenTimeAddedToPool(
         lowValueTxs.iterator(), highGasPriceTransaction, lowValueTxs.get(0));
   }
+
+  @Test
+  public void maxNumberOfTxsForTypeIsEnforced() {
+    final var limitedType = MAX_TRANSACTIONS_BY_TYPE.entrySet().iterator().next();
+    final var maxNumber = limitedType.getValue();
+    final var addedTxs = new ArrayList<Transaction>(maxNumber);
+    for (int i = 0; i < maxNumber; i++) {
+      final var tx =
+          createTransaction(
+              limitedType.getKey(),
+              0,
+              DEFAULT_MIN_GAS_PRICE,
+              DEFAULT_MIN_GAS_PRICE.divide(10),
+              0,
+              1,
+              SIGNATURE_ALGORITHM.get().generateKeyPair());
+      addedTxs.add(tx);
+      assertThat(prioritizeTransaction(tx)).isEqualTo(ADDED);
+    }
+
+    final var overflowTx =
+        createTransaction(
+            limitedType.getKey(),
+            0,
+            DEFAULT_MIN_GAS_PRICE,
+            DEFAULT_MIN_GAS_PRICE.divide(10),
+            0,
+            1,
+            SIGNATURE_ALGORITHM.get().generateKeyPair());
+    assertThat(prioritizeTransaction(overflowTx)).isEqualTo(DROPPED);
+
+    addedTxs.forEach(this::assertTransactionPrioritized);
+    assertTransactionNotPrioritized(overflowTx);
+  }
+
+  @Test
+  public void maxNumberOfTxsForTypeWithReplacement() {
+    final var limitedType = MAX_TRANSACTIONS_BY_TYPE.entrySet().iterator().next();
+    final var maxNumber = limitedType.getValue();
+    final var addedTxs = new ArrayList<Transaction>(maxNumber);
+    for (int i = 0; i < maxNumber; i++) {
+      final var tx =
+          createTransaction(
+              limitedType.getKey(),
+              i,
+              DEFAULT_MIN_GAS_PRICE,
+              DEFAULT_MIN_GAS_PRICE.divide(10),
+              0,
+              1,
+              KEYS1);
+      addedTxs.add(tx);
+      assertThat(prioritizeTransaction(tx)).isEqualTo(ADDED);
+    }
+
+    final var replacedTx = addedTxs.get(0);
+    final var replacementTx = createTransactionReplacement(replacedTx, KEYS1);
+    final var txAddResult = prioritizeTransaction(replacementTx);
+
+    assertThat(txAddResult.isReplacement()).isTrue();
+    assertThat(txAddResult.maybeReplacedTransaction())
+        .map(PendingTransaction::getTransaction)
+        .contains(replacedTx);
+
+    addedTxs.remove(replacedTx);
+    addedTxs.forEach(this::assertTransactionPrioritized);
+    assertTransactionNotPrioritized(replacedTx);
+    assertTransactionPrioritized(replacementTx);
+  }
 }

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

@@ -61,6 +61,7 @@ public class BaseTransactionPoolTest {
   protected final Transaction transaction0 = createTransaction(0);
   protected final Transaction transaction1 = createTransaction(1);
   protected final Transaction transaction2 = createTransaction(2);
+  protected final Transaction blobTransaction0 = createEIP4844Transaction(0, KEYS1, 1, 1);
 
   protected final StubMetricsSystem metricsSystem = new StubMetricsSystem();
 
@@ -103,13 +104,33 @@ public class BaseTransactionPoolTest {
         keys);
   }
 
+  protected Transaction createTransactionOfSize(
+      final long nonce, final Wei maxGasPrice, final int txSize, final KeyPair keys) {
+
+    final TransactionType txType =
+        TransactionType.values()[
+            randomizeTxType.nextInt(txSize < blobTransaction0.getSize() ? 3 : 4)];
+
+    final Transaction baseTx =
+        createTransaction(txType, nonce, maxGasPrice, maxGasPrice.divide(10), 0, 1, keys);
+    final int payloadSize = txSize - baseTx.getSize();
+
+    return createTransaction(
+        txType, nonce, maxGasPrice, maxGasPrice.divide(10), payloadSize, 1, keys);
+  }
+
   protected Transaction createTransaction(
       final long nonce, final Wei maxGasPrice, final int payloadSize, final KeyPair keys) {
 
-    // ToDo 4844: include BLOB tx here
-    final TransactionType txType = TransactionType.values()[randomizeTxType.nextInt(3)];
+    final TransactionType txType = TransactionType.values()[randomizeTxType.nextInt(4)];
 
-    return createTransaction(txType, nonce, maxGasPrice, payloadSize, keys);
+    return switch (txType) {
+      case FRONTIER, ACCESS_LIST, EIP1559 ->
+          createTransaction(txType, nonce, maxGasPrice, payloadSize, keys);
+      case BLOB ->
+          createTransaction(
+              txType, nonce, maxGasPrice, maxGasPrice.divide(10), payloadSize, 1, keys);
+    };
   }
 
   protected Transaction createTransaction(
@@ -157,6 +178,7 @@ public class BaseTransactionPoolTest {
       tx.maxFeePerGas(Optional.of(maxGasPrice))
           .maxPriorityFeePerGas(Optional.of(maxPriorityFeePerGas));
       if (type.supportsBlob() && blobCount > 0) {
+        tx.maxFeePerBlobGas(Optional.of(maxGasPrice));
         final var versionHashes =
             IntStream.range(0, blobCount)
                 .mapToObj(i -> new VersionedHash((byte) 1, Hash.ZERO))
@@ -186,7 +208,9 @@ public class BaseTransactionPoolTest {
         originalTransaction.getType(),
         originalTransaction.getNonce(),
         originalTransaction.getMaxGasPrice().multiply(2),
+        originalTransaction.getMaxGasPrice().multiply(2).divide(10),
         0,
+        1,
         keys);
   }
 

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

@@ -15,6 +15,7 @@
 package org.hyperledger.besu.ethereum.eth.transactions.layered;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.hyperledger.besu.datatypes.TransactionType.BLOB;
 import static org.hyperledger.besu.ethereum.eth.transactions.TransactionAddedResult.ADDED;
 import static org.hyperledger.besu.ethereum.eth.transactions.TransactionAddedResult.ALREADY_KNOWN;
 import static org.hyperledger.besu.ethereum.eth.transactions.TransactionAddedResult.NONCE_TOO_FAR_IN_FUTURE_FOR_SENDER;
@@ -55,6 +56,7 @@ import org.hyperledger.besu.plugin.data.TransactionSelectionResult;
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 import java.util.OptionalLong;
 import java.util.function.BiFunction;
@@ -68,7 +70,8 @@ import org.junit.jupiter.params.provider.MethodSource;
 public class LayeredPendingTransactionsTest extends BaseTransactionPoolTest {
 
   protected static final int MAX_TRANSACTIONS = 5;
-  protected static final int MAX_CAPACITY_BYTES = 10_000;
+  protected static final int MAX_PRIORITIZED_BLOB_TRANSACTIONS = MAX_TRANSACTIONS + 1;
+  protected static final int MAX_CAPACITY_BYTES = 150_000;
   protected static final Wei DEFAULT_BASE_FEE = Wei.of(100);
   protected static final int LIMITED_TRANSACTIONS_BY_SENDER = 4;
   protected static final String REMOTE = "remote";
@@ -82,20 +85,31 @@ public class LayeredPendingTransactionsTest extends BaseTransactionPoolTest {
   private final TransactionPoolConfiguration poolConf =
       ImmutableTransactionPoolConfiguration.builder()
           .maxPrioritizedTransactions(MAX_TRANSACTIONS)
+          .maxPrioritizedTransactionsByType(Map.of(BLOB, MAX_PRIORITIZED_BLOB_TRANSACTIONS))
           .maxFutureBySender(MAX_TRANSACTIONS)
-          .pendingTransactionsLayerMaxCapacityBytes(MAX_CAPACITY_BYTES)
           .build();
 
   private final TransactionPoolConfiguration senderLimitedConfig =
       ImmutableTransactionPoolConfiguration.builder()
           .maxPrioritizedTransactions(MAX_TRANSACTIONS)
+          .maxPrioritizedTransactionsByType(Map.of(BLOB, MAX_PRIORITIZED_BLOB_TRANSACTIONS))
+          .maxFutureBySender(LIMITED_TRANSACTIONS_BY_SENDER)
+          .build();
+
+  private final TransactionPoolConfiguration smallPoolConfig =
+      ImmutableTransactionPoolConfiguration.builder()
+          .maxPrioritizedTransactions(MAX_TRANSACTIONS)
+          .maxPrioritizedTransactionsByType(Map.of(BLOB, MAX_PRIORITIZED_BLOB_TRANSACTIONS))
           .maxFutureBySender(LIMITED_TRANSACTIONS_BY_SENDER)
           .pendingTransactionsLayerMaxCapacityBytes(MAX_CAPACITY_BYTES)
           .build();
+
   private LayeredPendingTransactions senderLimitedTransactions;
   private LayeredPendingTransactions pendingTransactions;
+  private LayeredPendingTransactions smallPendingTransactions;
   private CreatedLayers senderLimitedLayers;
   private CreatedLayers layers;
+  private CreatedLayers smallLayers;
   private TransactionPoolMetrics txPoolMetrics;
 
   private static BlockHeader mockBlockHeader() {
@@ -151,12 +165,16 @@ public class LayeredPendingTransactionsTest extends BaseTransactionPoolTest {
 
     layers = createLayers(poolConf);
     senderLimitedLayers = createLayers(senderLimitedConfig);
+    smallLayers = createLayers(smallPoolConfig);
 
     pendingTransactions = new LayeredPendingTransactions(poolConf, layers.prioritizedTransactions);
 
     senderLimitedTransactions =
         new LayeredPendingTransactions(
             senderLimitedConfig, senderLimitedLayers.prioritizedTransactions);
+
+    smallPendingTransactions =
+        new LayeredPendingTransactions(smallPoolConfig, smallLayers.prioritizedTransactions);
   }
 
   @Test
@@ -211,41 +229,43 @@ public class LayeredPendingTransactionsTest extends BaseTransactionPoolTest {
   public void evictTransactionsWhenSizeLimitExceeded() {
     final List<Transaction> firstTxs = new ArrayList<>(MAX_TRANSACTIONS);
 
-    pendingTransactions.subscribeDroppedTransactions(droppedListener);
+    smallPendingTransactions.subscribeDroppedTransactions(droppedListener);
 
     for (int i = 0; i < MAX_TRANSACTIONS; i++) {
       final Account sender = mock(Account.class);
       when(sender.getNonce()).thenReturn((long) i);
       final var tx =
-          createTransaction(
+          createTransactionOfSize(
               i,
-              DEFAULT_MIN_GAS_PRICE.multiply(2 * (i + 1)),
-              (int) poolConf.getPendingTransactionsLayerMaxCapacityBytes() + 1,
+              DEFAULT_BASE_FEE.add(i),
+              (int) smallPoolConfig.getPendingTransactionsLayerMaxCapacityBytes() + 1,
               SIGNATURE_ALGORITHM.get().generateKeyPair());
-      pendingTransactions.addTransaction(createRemotePendingTransaction(tx), Optional.of(sender));
+      smallPendingTransactions.addTransaction(
+          createRemotePendingTransaction(tx), Optional.of(sender));
       firstTxs.add(tx);
-      assertTransactionPending(pendingTransactions, tx);
+      assertTransactionPending(smallPendingTransactions, tx);
     }
 
-    assertThat(pendingTransactions.size()).isEqualTo(MAX_TRANSACTIONS);
+    assertThat(smallPendingTransactions.size()).isEqualTo(MAX_TRANSACTIONS);
 
     final Transaction lastBigTx =
-        createTransaction(
+        createTransactionOfSize(
             0,
             DEFAULT_MIN_GAS_PRICE.multiply(1000),
-            (int) poolConf.getPendingTransactionsLayerMaxCapacityBytes(),
+            (int) smallPoolConfig.getPendingTransactionsLayerMaxCapacityBytes(),
             SIGNATURE_ALGORITHM.get().generateKeyPair());
     final Account lastSender = mock(Account.class);
     when(lastSender.getNonce()).thenReturn(0L);
-    pendingTransactions.addTransaction(
+    smallPendingTransactions.addTransaction(
         createRemotePendingTransaction(lastBigTx), Optional.of(lastSender));
-    assertTransactionPending(pendingTransactions, lastBigTx);
+    assertTransactionPending(smallPendingTransactions, lastBigTx);
 
-    assertTransactionNotPending(pendingTransactions, firstTxs.get(0));
+    assertTransactionNotPending(smallPendingTransactions, firstTxs.get(0));
     assertThat(
-            getRemovedCount(REMOTE, NO_PRIORITY, DROPPED.label(), layers.evictedCollector.name()))
+            getRemovedCount(
+                REMOTE, NO_PRIORITY, DROPPED.label(), smallLayers.evictedCollector.name()))
         .isEqualTo(1);
-    assertThat(layers.evictedCollector.getEvictedTransactions())
+    assertThat(smallLayers.evictedCollector.getEvictedTransactions())
         .map(PendingTransaction::getTransaction)
         .contains(firstTxs.get(0));
     verify(droppedListener).onTransactionDropped(firstTxs.get(0));

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

@@ -15,6 +15,10 @@
 package org.hyperledger.besu.ethereum.eth.transactions.layered;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.hyperledger.besu.datatypes.TransactionType.ACCESS_LIST;
+import static org.hyperledger.besu.datatypes.TransactionType.BLOB;
+import static org.hyperledger.besu.datatypes.TransactionType.EIP1559;
+import static org.hyperledger.besu.datatypes.TransactionType.FRONTIER;
 import static org.hyperledger.besu.ethereum.eth.transactions.layered.LayersTest.Sender.S1;
 import static org.hyperledger.besu.ethereum.eth.transactions.layered.LayersTest.Sender.S2;
 import static org.hyperledger.besu.ethereum.eth.transactions.layered.LayersTest.Sender.S3;
@@ -27,6 +31,7 @@ import static org.mockito.Mockito.when;
 
 import org.hyperledger.besu.crypto.KeyPair;
 import org.hyperledger.besu.datatypes.Address;
+import org.hyperledger.besu.datatypes.TransactionType;
 import org.hyperledger.besu.datatypes.Wei;
 import org.hyperledger.besu.ethereum.core.BlockHeader;
 import org.hyperledger.besu.ethereum.core.MiningParameters;
@@ -51,7 +56,6 @@ import java.util.Optional;
 import java.util.OptionalLong;
 import java.util.stream.Stream;
 
-import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.MethodSource;
@@ -62,51 +66,29 @@ public class LayersTest extends BaseTransactionPoolTest {
   private static final Wei BASE_FEE = Wei.ONE;
   private static final Wei MIN_GAS_PRICE = BASE_FEE;
 
-  private final TransactionPoolConfiguration poolConfig =
+  private static final TransactionPoolConfiguration DEFAULT_TX_POOL_CONFIG =
       ImmutableTransactionPoolConfiguration.builder()
           .maxPrioritizedTransactions(MAX_PRIO_TRANSACTIONS)
+          .maxPrioritizedTransactionsByType(Map.of(BLOB, 1))
           .maxFutureBySender(MAX_FUTURE_FOR_SENDER)
           .pendingTransactionsLayerMaxCapacityBytes(
-              new PendingTransaction.Remote(createEIP1559Transaction(0, KEYS1, 1)).memorySize() * 3)
+              new PendingTransaction.Remote(
+                          new BaseTransactionPoolTest().createEIP1559Transaction(0, KEYS1, 1))
+                      .memorySize()
+                  * 3L)
           .build();
 
-  private final TransactionPoolMetrics txPoolMetrics = new TransactionPoolMetrics(metricsSystem);
-
-  private final EvictCollectorLayer evictCollector = new EvictCollectorLayer(txPoolMetrics);
-  private final SparseTransactions sparseTransactions =
-      new SparseTransactions(
-          poolConfig,
-          evictCollector,
-          txPoolMetrics,
-          this::transactionReplacementTester,
-          new BlobCache());
-
-  private final ReadyTransactions readyTransactions =
-      new ReadyTransactions(
-          poolConfig,
-          sparseTransactions,
-          txPoolMetrics,
-          this::transactionReplacementTester,
-          new BlobCache());
-
-  private final BaseFeePrioritizedTransactions prioritizedTransactions =
-      new BaseFeePrioritizedTransactions(
-          poolConfig,
-          LayersTest::mockBlockHeader,
-          readyTransactions,
-          txPoolMetrics,
-          this::transactionReplacementTester,
-          FeeMarket.london(0L),
-          new BlobCache(),
-          MiningParameters.newDefault().setMinTransactionGasPrice(MIN_GAS_PRICE));
-
-  private final LayeredPendingTransactions pendingTransactions =
-      new LayeredPendingTransactions(poolConfig, prioritizedTransactions);
-
-  @AfterEach
-  void reset() {
-    pendingTransactions.reset();
-  }
+  private static final TransactionPoolConfiguration BLOB_TX_POOL_CONFIG =
+      ImmutableTransactionPoolConfiguration.builder()
+          .maxPrioritizedTransactions(MAX_PRIO_TRANSACTIONS)
+          .maxPrioritizedTransactionsByType(Map.of(BLOB, 1))
+          .maxFutureBySender(MAX_FUTURE_FOR_SENDER)
+          .pendingTransactionsLayerMaxCapacityBytes(
+              new PendingTransaction.Remote(
+                          new BaseTransactionPoolTest().createEIP4844Transaction(0, KEYS1, 1, 1))
+                      .memorySize()
+                  * 3L)
+          .build();
 
   @ParameterizedTest
   @MethodSource("providerAddTransactions")
@@ -168,7 +150,51 @@ public class LayersTest extends BaseTransactionPoolTest {
     assertScenario(scenario);
   }
 
+  @ParameterizedTest
+  @MethodSource("providerMaxPrioritizedByType")
+  void maxPrioritizedByType(final Scenario scenario) {
+    assertScenario(scenario, BLOB_TX_POOL_CONFIG);
+  }
+
   private void assertScenario(final Scenario scenario) {
+    assertScenario(scenario, DEFAULT_TX_POOL_CONFIG);
+  }
+
+  private void assertScenario(
+      final Scenario scenario, final TransactionPoolConfiguration poolConfig) {
+    final TransactionPoolMetrics txPoolMetrics = new TransactionPoolMetrics(metricsSystem);
+
+    final EvictCollectorLayer evictCollector = new EvictCollectorLayer(txPoolMetrics);
+    final SparseTransactions sparseTransactions =
+        new SparseTransactions(
+            poolConfig,
+            evictCollector,
+            txPoolMetrics,
+            (pt1, pt2) -> transactionReplacementTester(poolConfig, pt1, pt2),
+            new BlobCache());
+
+    final ReadyTransactions readyTransactions =
+        new ReadyTransactions(
+            poolConfig,
+            sparseTransactions,
+            txPoolMetrics,
+            (pt1, pt2) -> transactionReplacementTester(poolConfig, pt1, pt2),
+            new BlobCache());
+
+    final BaseFeePrioritizedTransactions prioritizedTransactions =
+        new BaseFeePrioritizedTransactions(
+            poolConfig,
+            LayersTest::mockBlockHeader,
+            readyTransactions,
+            txPoolMetrics,
+            (pt1, pt2) -> transactionReplacementTester(poolConfig, pt1, pt2),
+            FeeMarket.london(0L),
+            new BlobCache(),
+            MiningParameters.newDefault().setMinTransactionGasPrice(MIN_GAS_PRICE));
+
+    final LayeredPendingTransactions pendingTransactions =
+        new LayeredPendingTransactions(poolConfig, prioritizedTransactions);
+
     scenario.execute(
         pendingTransactions,
         prioritizedTransactions,
@@ -1180,17 +1206,56 @@ public class LayersTest extends BaseTransactionPoolTest {
                 .expectedDroppedForSender(S3, 0)));
   }
 
+  static Stream<Arguments> providerMaxPrioritizedByType() {
+    return Stream.of(
+        Arguments.of(
+            new Scenario("first blob tx is prioritized")
+                .addForSender(S1, BLOB, 0)
+                .expectedPrioritizedForSender(S1, 0)),
+        Arguments.of(
+            new Scenario("multiple senders only first blob tx is prioritized")
+                .addForSender(S1, BLOB, 0)
+                .addForSender(S2, BLOB, 0)
+                .expectedPrioritizedForSender(S1, 0)
+                .expectedReadyForSender(S2, 0)),
+        Arguments.of(
+            new Scenario("same sender following blob txs are moved to ready")
+                .addForSender(S1, BLOB, 0, 1, 2)
+                .expectedPrioritizedForSender(S1, 0)
+                .expectedReadyForSender(S1, 1, 2)),
+        Arguments.of(
+            new Scenario("promoting txs respect prioritized count limit")
+                .addForSender(S1, BLOB, 0, 1, 2)
+                .expectedPrioritizedForSender(S1, 0)
+                .expectedReadyForSender(S1, 1, 2)
+                .confirmedForSenders(S1, 0)
+                .expectedPrioritizedForSender(S1, 1)
+                .expectedReadyForSender(S1, 2)),
+        Arguments.of(
+            new Scenario("filling gaps respect prioritized count limit")
+                .addForSender(S1, BLOB, 1)
+                .expectedSparseForSender(S1, 1)
+                .addForSender(S1, BLOB, 0)
+                .expectedPrioritizedForSender(S1, 0)
+                .expectedSparseForSender(S1, 1)),
+        Arguments.of(
+            new Scenario("promoting to ready is unbounded")
+                .addForSender(S1, BLOB, 0, 1, 2, 3, 4, 5, 6)
+                .expectedPrioritizedForSender(S1, 0)
+                .expectedReadyForSender(S1, 1, 2, 3)
+                .expectedSparseForSender(S1, 4, 5, 6)
+                .confirmedForSenders(S1, 3)
+                .expectedPrioritizedForSender(S1, 4)
+                .expectedReadyForSender(S1, 5, 6)
+                .expectedSparseForSenders()));
+  }
+
   private static BlockHeader mockBlockHeader() {
     final BlockHeader blockHeader = mock(BlockHeader.class);
     when(blockHeader.getBaseFee()).thenReturn(Optional.of(BASE_FEE));
     return blockHeader;
   }
 
-  private boolean transactionReplacementTester(
-      final PendingTransaction pt1, final PendingTransaction pt2) {
-    return transactionReplacementTester(poolConfig, pt1, pt2);
-  }
-
   private static boolean transactionReplacementTester(
       final TransactionPoolConfiguration poolConfig,
       final PendingTransaction pt1,
@@ -1235,10 +1300,14 @@ public class LayersTest extends BaseTransactionPoolTest {
     }
 
     Scenario addForSender(final Sender sender, final long... nonce) {
+      return addForSender(sender, EIP1559, nonce);
+    }
+
+    Scenario addForSender(final Sender sender, final TransactionType type, final long... nonce) {
       Arrays.stream(nonce)
           .forEach(
               n -> {
-                final var pendingTx = getOrCreate(sender, n);
+                final var pendingTx = getOrCreate(sender, type, n);
                 actions.add(
                     (pending, prio, ready, sparse, dropped) -> {
                       final Account mockSender = mock(Account.class);
@@ -1290,22 +1359,49 @@ public class LayersTest extends BaseTransactionPoolTest {
       assertExpectedDropped(dropped, lastExpectedDropped);
     }
 
-    private PendingTransaction getOrCreate(final Sender sender, final long nonce) {
+    private PendingTransaction getOrCreate(
+        final Sender sender, final TransactionType type, final long nonce) {
       return txsBySender
           .get(sender)
-          .computeIfAbsent(nonce, n -> createEIP1559PendingTransactions(sender, n));
+          .computeIfAbsent(
+              nonce,
+              n ->
+                  switch (type) {
+                    case FRONTIER -> createFrontierPendingTransaction(sender, n);
+                    case ACCESS_LIST -> createAccessListPendingTransaction(sender, n);
+                    case EIP1559 -> createEIP1559PendingTransaction(sender, n);
+                    case BLOB -> createBlobPendingTransaction(sender, n);
+                  });
     }
 
     private PendingTransaction get(final Sender sender, final long nonce) {
       return txsBySender.get(sender).get(nonce);
     }
 
-    private PendingTransaction createEIP1559PendingTransactions(
+    private PendingTransaction createFrontierPendingTransaction(
+        final Sender sender, final long nonce) {
+      return createRemotePendingTransaction(
+          createTransaction(FRONTIER, nonce, Wei.ONE, 0, sender.key), sender.hasPriority);
+    }
+
+    private PendingTransaction createAccessListPendingTransaction(
+        final Sender sender, final long nonce) {
+      return createRemotePendingTransaction(
+          createTransaction(ACCESS_LIST, nonce, Wei.ONE, 0, sender.key), sender.hasPriority);
+    }
+
+    private PendingTransaction createEIP1559PendingTransaction(
         final Sender sender, final long nonce) {
       return createRemotePendingTransaction(
           createEIP1559Transaction(nonce, sender.key, sender.gasFeeMultiplier), sender.hasPriority);
     }
 
+    private PendingTransaction createBlobPendingTransaction(final Sender sender, final long nonce) {
+      return createRemotePendingTransaction(
+          createEIP4844Transaction(nonce, sender.key, sender.gasFeeMultiplier, 1),
+          sender.hasPriority);
+    }
+
     public Scenario expectedPrioritizedForSender(final Sender sender, final long... nonce) {
       lastExpectedPrioritized = expectedForSender(sender, nonce);
       final var expectedCopy = List.copyOf(lastExpectedPrioritized);
@@ -1469,7 +1565,7 @@ public class LayersTest extends BaseTransactionPoolTest {
       Arrays.stream(nonce)
           .forEach(
               n -> {
-                final var pendingTx = getOrCreate(sender, n);
+                final var pendingTx = getOrCreate(sender, EIP1559, n);
                 actions.add(
                     (pending, prio, ready, sparse, dropped) -> prio.remove(pendingTx, INVALIDATED));
               });

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

@@ -22,6 +22,7 @@ import static org.mockito.Mockito.when;
 
 import org.hyperledger.besu.crypto.SignatureAlgorithmFactory;
 import org.hyperledger.besu.datatypes.Address;
+import org.hyperledger.besu.datatypes.TransactionType;
 import org.hyperledger.besu.datatypes.Wei;
 import org.hyperledger.besu.ethereum.core.BlockHeader;
 import org.hyperledger.besu.ethereum.core.MiningParameters;
@@ -48,11 +49,13 @@ import java.math.BigInteger;
 import java.nio.charset.StandardCharsets;
 import java.time.Instant;
 import java.util.Arrays;
+import java.util.EnumMap;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.function.BiFunction;
+import java.util.stream.Collectors;
 import java.util.zip.GZIPInputStream;
 
 import com.google.common.base.Splitter;
@@ -117,6 +120,7 @@ public class ReplayTest {
       final TransactionPoolConfiguration poolConfig =
           ImmutableTransactionPoolConfiguration.builder()
               .prioritySenders(readPrioritySenders(br.readLine()))
+              .maxPrioritizedTransactionsByType(readMaxPrioritizedByType(br.readLine()))
               .build();
 
       final AbstractPrioritizedTransactions prioritizedTransactions =
@@ -161,6 +165,17 @@ public class ReplayTest {
     }
   }
 
+  private Map<TransactionType, Integer> readMaxPrioritizedByType(final String line) {
+    return Arrays.stream(line.split(","))
+        .map(e -> e.split("="))
+        .collect(
+            Collectors.toMap(
+                a -> TransactionType.valueOf(a[0]),
+                a -> Integer.parseInt(a[1]),
+                (a, b) -> a,
+                () -> new EnumMap<>(TransactionType.class)));
+  }
+
   private List<Address> readPrioritySenders(final String line) {
     return Arrays.stream(line.split(",")).map(Address::fromHexString).toList();
   }