Layered Transaction Pool

.gitignore

@@ -4,7 +4,6 @@
 *~.nib
 *.iml
 *.launch
-*.swp
 *.log
 .classpath
 .DS_Store
@@ -29,4 +28,5 @@ site/
 /kubernetes/reports/
 /kubernetes/besu-*.tar.gz
 **/src/*/generated
-jitpack.yml
+jitpack.yml
+/ethereum/eth/src/test/resources/tx.csv.gz

CHANGELOG.md

@@ -9,6 +9,7 @@
 - EIP-4844: Zero blob transactions are invalid [#5425](https://github.com/hyperledger/besu/pull/5425)
 - Transaction pool flag to disable specific behaviors for locally submitted transactions [#5418](https://github.com/hyperledger/besu/pull/5418)
 - New optional feature to save the txpool content to file on shutdown and reloading it on startup [#5434](https://github.com/hyperledger/besu/pull/5434) 
+- Early access - layered transaction pool implementation [#5290](https://github.com/hyperledger/besu/pull/5290)
 
 ### Bug Fixes
 - Fix eth_feeHistory response for the case in which blockCount is higher than highestBlock requested. [#5397](https://github.com/hyperledger/besu/pull/5397)

besu/src/main/java/org/hyperledger/besu/cli/BesuCommand.java

@@ -1202,7 +1202,8 @@ static class TxPoolOptionGroup {
         description =
             "Maximum number of pending transactions that will be kept in the transaction pool (default: ${DEFAULT-VALUE})",
         arity = "1")
-    private final Integer txPoolMaxSize = TransactionPoolConfiguration.MAX_PENDING_TRANSACTIONS;
+    private final Integer txPoolMaxSize =
+        TransactionPoolConfiguration.DEFAULT_MAX_PENDING_TRANSACTIONS;
 
     @Option(
         names = {"--tx-pool-retention-hours"},

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

@@ -25,11 +25,15 @@
 import java.util.Arrays;
 import java.util.List;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import picocli.CommandLine;
 
 /** The Transaction pool Cli options. */
 public class TransactionPoolOptions
     implements CLIOptions<ImmutableTransactionPoolConfiguration.Builder> {
+  private static final Logger LOG = LoggerFactory.getLogger(TransactionPoolOptions.class);
+
   private static final String TX_MESSAGE_KEEP_ALIVE_SEC_FLAG =
       "--Xincoming-tx-messages-keep-alive-seconds";
 
@@ -47,6 +51,14 @@ public class TransactionPoolOptions
   private static final String SAVE_RESTORE_FLAG = "--tx-pool-enable-save-restore";
   private static final String SAVE_FILE = "--tx-pool-save-file";
 
+  private static final String LAYERED_TX_POOL_ENABLED_FLAG = "--Xlayered-tx-pool";
+  private static final String LAYERED_TX_POOL_LAYER_MAX_CAPACITY =
+      "--Xlayered-tx-pool-layer-max-capacity";
+  private static final String LAYERED_TX_POOL_MAX_PRIORITIZED =
+      "--Xlayered-tx-pool-max-prioritized";
+  private static final String LAYERED_TX_POOL_MAX_FUTURE_BY_SENDER =
+      "--Xlayered-tx-pool-max-future-by-sender";
+
   @CommandLine.Option(
       names = {STRICT_TX_REPLAY_PROTECTION_ENABLED_FLAG},
       paramLabel = "<Boolean>",
@@ -84,7 +96,46 @@ public class TransactionPoolOptions
           "Maximum portion of the transaction pool which a single account may occupy with future transactions (default: ${DEFAULT-VALUE})",
       arity = "1")
   private Float txPoolLimitByAccountPercentage =
-      TransactionPoolConfiguration.LIMIT_TXPOOL_BY_ACCOUNT_PERCENTAGE;
+      TransactionPoolConfiguration.DEFAULT_LIMIT_TX_POOL_BY_ACCOUNT_PERCENTAGE;
+
+  @CommandLine.Option(
+      names = {LAYERED_TX_POOL_ENABLED_FLAG},
+      paramLabel = "<Boolean>",
+      hidden = true,
+      description = "Enable the Layered Transaction Pool (default: ${DEFAULT-VALUE})",
+      arity = "0..1")
+  private Boolean layeredTxPoolEnabled =
+      TransactionPoolConfiguration.DEFAULT_LAYERED_TX_POOL_ENABLED;
+
+  @CommandLine.Option(
+      names = {LAYERED_TX_POOL_LAYER_MAX_CAPACITY},
+      paramLabel = "<Long>",
+      hidden = true,
+      description =
+          "Max amount of memory space, in bytes, that any layer within the transaction pool could occupy (default: ${DEFAULT-VALUE})",
+      arity = "1")
+  private long layeredTxPoolLayerMaxCapacity =
+      TransactionPoolConfiguration.DEFAULT_PENDING_TRANSACTIONS_LAYER_MAX_CAPACITY_BYTES;
+
+  @CommandLine.Option(
+      names = {LAYERED_TX_POOL_MAX_PRIORITIZED},
+      paramLabel = "<Int>",
+      hidden = true,
+      description =
+          "Max number of pending transactions that are prioritized and thus kept sorted (default: ${DEFAULT-VALUE})",
+      arity = "1")
+  private int layeredTxPoolMaxPrioritized =
+      TransactionPoolConfiguration.DEFAULT_MAX_PRIORITIZED_TRANSACTIONS;
+
+  @CommandLine.Option(
+      names = {LAYERED_TX_POOL_MAX_FUTURE_BY_SENDER},
+      paramLabel = "<Int>",
+      hidden = true,
+      description =
+          "Max number of future pending transactions allowed for a single sender (default: ${DEFAULT-VALUE})",
+      arity = "1")
+  private int layeredTxPoolMaxFutureBySender =
+      TransactionPoolConfiguration.DEFAULT_MAX_FUTURE_BY_SENDER;
 
   @CommandLine.Option(
       names = {DISABLE_LOCAL_TXS_FLAG},
@@ -139,19 +190,34 @@ public static TransactionPoolOptions fromConfig(final TransactionPoolConfigurati
     options.disableLocalTxs = config.getDisableLocalTransactions();
     options.saveRestoreEnabled = config.getEnableSaveRestore();
     options.saveFile = config.getSaveFile();
+    options.layeredTxPoolEnabled = config.getLayeredTxPoolEnabled();
+    options.layeredTxPoolLayerMaxCapacity = config.getPendingTransactionsLayerMaxCapacityBytes();
+    options.layeredTxPoolMaxPrioritized = config.getMaxPrioritizedTransactions();
+    options.layeredTxPoolMaxFutureBySender = config.getMaxFutureBySender();
     return options;
   }
 
   @Override
   public ImmutableTransactionPoolConfiguration.Builder toDomainObject() {
+    if (layeredTxPoolEnabled) {
+      LOG.warn(
+          "Layered transaction pool enabled, ignoring settings for "
+              + "--tx-pool-max-size and --tx-pool-limit-by-account-percentage");
+    }
+
     return ImmutableTransactionPoolConfiguration.builder()
         .strictTransactionReplayProtectionEnabled(strictTxReplayProtectionEnabled)
         .txMessageKeepAliveSeconds(txMessageKeepAliveSeconds)
         .eth65TrxAnnouncedBufferingPeriod(Duration.ofMillis(eth65TrxAnnouncedBufferingPeriod))
         .txPoolLimitByAccountPercentage(txPoolLimitByAccountPercentage)
         .disableLocalTransactions(disableLocalTxs)
         .enableSaveRestore(saveRestoreEnabled)
-        .saveFile(saveFile);
+        .saveFile(saveFile)
+        .txPoolLimitByAccountPercentage(txPoolLimitByAccountPercentage)
+        .layeredTxPoolEnabled(layeredTxPoolEnabled)
+        .pendingTransactionsLayerMaxCapacityBytes(layeredTxPoolLayerMaxCapacity)
+        .maxPrioritizedTransactions(layeredTxPoolMaxPrioritized)
+        .maxFutureBySender(layeredTxPoolMaxFutureBySender);
   }
 
   @Override
@@ -166,7 +232,14 @@ public List<String> getCLIOptions() {
         TX_MESSAGE_KEEP_ALIVE_SEC_FLAG,
         OptionParser.format(txMessageKeepAliveSeconds),
         ETH65_TX_ANNOUNCED_BUFFERING_PERIOD_FLAG,
-        OptionParser.format(eth65TrxAnnouncedBufferingPeriod));
+        OptionParser.format(eth65TrxAnnouncedBufferingPeriod),
+        LAYERED_TX_POOL_ENABLED_FLAG + "=" + layeredTxPoolEnabled,
+        LAYERED_TX_POOL_LAYER_MAX_CAPACITY,
+        OptionParser.format(layeredTxPoolLayerMaxCapacity),
+        LAYERED_TX_POOL_MAX_PRIORITIZED,
+        OptionParser.format(layeredTxPoolMaxPrioritized),
+        LAYERED_TX_POOL_MAX_FUTURE_BY_SENDER,
+        OptionParser.format(layeredTxPoolMaxFutureBySender));
   }
 
   /**

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

@@ -16,7 +16,7 @@
 
 import static java.nio.charset.StandardCharsets.UTF_8;
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.hyperledger.besu.ethereum.eth.transactions.TransactionPoolConfiguration.LIMIT_TXPOOL_BY_ACCOUNT_PERCENTAGE;
+import static org.hyperledger.besu.ethereum.eth.transactions.TransactionPoolConfiguration.DEFAULT_LIMIT_TX_POOL_BY_ACCOUNT_PERCENTAGE;
 
 import org.hyperledger.besu.cli.options.unstable.TransactionPoolOptions;
 import org.hyperledger.besu.ethereum.eth.transactions.ImmutableTransactionPoolConfiguration;
@@ -101,7 +101,7 @@ public void senderLimitedTxPoolCeiling_violated() {
     final TransactionPoolOptions options = getOptionsFromBesuCommand(cmd);
     final TransactionPoolConfiguration config = options.toDomainObject().build();
     assertThat(config.getTxPoolLimitByAccountPercentage())
-        .isEqualTo(LIMIT_TXPOOL_BY_ACCOUNT_PERCENTAGE);
+        .isEqualTo(DEFAULT_LIMIT_TX_POOL_BY_ACCOUNT_PERCENTAGE);
 
     assertThat(commandOutput.toString(UTF_8)).isEmpty();
     assertThat(commandErrorOutput.toString(UTF_8))
@@ -239,7 +239,13 @@ ImmutableTransactionPoolConfiguration.Builder createDefaultDomainObject() {
         .txPoolLimitByAccountPercentage(defaultValue.getTxPoolLimitByAccountPercentage())
         .disableLocalTransactions(defaultValue.getDisableLocalTransactions())
         .enableSaveRestore(defaultValue.getEnableSaveRestore())
-        .saveFile(defaultValue.getSaveFile());
+        .saveFile(defaultValue.getSaveFile())
+        .txPoolLimitByAccountPercentage(defaultValue.getTxPoolLimitByAccountPercentage())
+        .layeredTxPoolEnabled(defaultValue.getLayeredTxPoolEnabled())
+        .pendingTransactionsLayerMaxCapacityBytes(
+            defaultValue.getPendingTransactionsLayerMaxCapacityBytes())
+        .maxPrioritizedTransactions(defaultValue.getMaxPrioritizedTransactions())
+        .maxFutureBySender(defaultValue.getMaxFutureBySender());
   }
 
   @Override
@@ -253,7 +259,12 @@ ImmutableTransactionPoolConfiguration.Builder createCustomizedDomainObject() {
         .txPoolLimitByAccountPercentage(0.5f)
         .disableLocalTransactions(true)
         .enableSaveRestore(true)
-        .saveFile(new File("abc.xyz"));
+        .saveFile(new File("abc.xyz"))
+        .txPoolLimitByAccountPercentage(0.5f)
+        .layeredTxPoolEnabled(true)
+        .pendingTransactionsLayerMaxCapacityBytes(1_000_000L)
+        .maxPrioritizedTransactions(1000)
+        .maxFutureBySender(10);
   }
 
   @Override

ethereum/api/src/integration-test/java/org/hyperledger/besu/ethereum/api/jsonrpc/methods/fork/frontier/EthGetFilterChangesIntegrationTest.java

@@ -49,9 +49,11 @@
 import org.hyperledger.besu.ethereum.eth.manager.EthContext;
 import org.hyperledger.besu.ethereum.eth.manager.EthPeers;
 import org.hyperledger.besu.ethereum.eth.transactions.ImmutableTransactionPoolConfiguration;
+import org.hyperledger.besu.ethereum.eth.transactions.PendingTransactions;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionBroadcaster;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPool;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPoolConfiguration;
+import org.hyperledger.besu.ethereum.eth.transactions.TransactionPoolMetrics;
 import org.hyperledger.besu.ethereum.eth.transactions.sorter.GasPricePendingTransactionsSorter;
 import org.hyperledger.besu.metrics.noop.NoOpMetricsSystem;
 import org.hyperledger.besu.plugin.data.TransactionType;
@@ -84,7 +86,7 @@ public class EthGetFilterChangesIntegrationTest {
   private TransactionPool transactionPool;
   private final MetricsSystem metricsSystem = new NoOpMetricsSystem();
 
-  private GasPricePendingTransactionsSorter transactions;
+  private PendingTransactions transactions;
 
   private static final int MAX_TRANSACTIONS = 5;
   private static final KeyPair keyPair = SignatureAlgorithmFactory.getInstance().generateKeyPair();
@@ -116,7 +118,7 @@ public void setUp() {
             batchAddedListener,
             ethContext,
             new MiningParameters.Builder().minTransactionGasPrice(Wei.ZERO).build(),
-            metricsSystem,
+            new TransactionPoolMetrics(metricsSystem),
             TransactionPoolConfiguration.DEFAULT);
     final BlockchainQueries blockchainQueries =
         new BlockchainQueries(blockchain, protocolContext.getWorldStateArchive());

ethereum/api/src/integration-test/java/org/hyperledger/besu/ethereum/api/jsonrpc/methods/fork/london/EthGetFilterChangesIntegrationTest.java

@@ -49,9 +49,11 @@
 import org.hyperledger.besu.ethereum.eth.manager.EthContext;
 import org.hyperledger.besu.ethereum.eth.manager.EthPeers;
 import org.hyperledger.besu.ethereum.eth.transactions.ImmutableTransactionPoolConfiguration;
+import org.hyperledger.besu.ethereum.eth.transactions.PendingTransactions;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionBroadcaster;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPool;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPoolConfiguration;
+import org.hyperledger.besu.ethereum.eth.transactions.TransactionPoolMetrics;
 import org.hyperledger.besu.ethereum.eth.transactions.sorter.BaseFeePendingTransactionsSorter;
 import org.hyperledger.besu.metrics.noop.NoOpMetricsSystem;
 import org.hyperledger.besu.plugin.data.TransactionType;
@@ -84,7 +86,7 @@ public class EthGetFilterChangesIntegrationTest {
   private TransactionPool transactionPool;
   private final MetricsSystem metricsSystem = new NoOpMetricsSystem();
 
-  private BaseFeePendingTransactionsSorter transactions;
+  private PendingTransactions transactions;
 
   private static final int MAX_TRANSACTIONS = 5;
   private static final KeyPair keyPair = SignatureAlgorithmFactory.getInstance().generateKeyPair();
@@ -116,7 +118,7 @@ public void setUp() {
             batchAddedListener,
             ethContext,
             new MiningParameters.Builder().minTransactionGasPrice(Wei.ZERO).build(),
-            metricsSystem,
+            new TransactionPoolMetrics(metricsSystem),
             TransactionPoolConfiguration.DEFAULT);
     final BlockchainQueries blockchainQueries =
         new BlockchainQueries(blockchain, protocolContext.getWorldStateArchive());

ethereum/api/src/main/java/org/hyperledger/besu/ethereum/api/jsonrpc/internal/methods/TxPoolBesuPendingTransactions.java

@@ -25,9 +25,9 @@
 import org.hyperledger.besu.ethereum.core.Transaction;
 import org.hyperledger.besu.ethereum.eth.transactions.PendingTransactions;
 
+import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
-import java.util.Set;
 import java.util.stream.Collectors;
 
 public class TxPoolBesuPendingTransactions implements JsonRpcMethod {
@@ -56,7 +56,7 @@ public JsonRpcResponse response(final JsonRpcRequestContext requestContext) {
             .map(PendingTransactionsParams::filters)
             .orElse(Collections.emptyList());
 
-    final Set<Transaction> pendingTransactionsFiltered =
+    final Collection<Transaction> pendingTransactionsFiltered =
         pendingTransactionFilter.reduce(
             pendingTransactions.getPendingTransactions(), filters, limit);
 

ethereum/api/src/main/java/org/hyperledger/besu/ethereum/api/jsonrpc/internal/methods/TxPoolBesuStatistics.java

@@ -22,7 +22,7 @@
 import org.hyperledger.besu.ethereum.eth.transactions.PendingTransaction;
 import org.hyperledger.besu.ethereum.eth.transactions.PendingTransactions;
 
-import java.util.Set;
+import java.util.Collection;
 
 public class TxPoolBesuStatistics implements JsonRpcMethod {
 
@@ -43,7 +43,8 @@ public JsonRpcResponse response(final JsonRpcRequestContext requestContext) {
   }
 
   private PendingTransactionsStatisticsResult statistics() {
-    final Set<PendingTransaction> pendingTransaction = pendingTransactions.getPendingTransactions();
+    final Collection<PendingTransaction> pendingTransaction =
+        pendingTransactions.getPendingTransactions();
     final long localCount =
         pendingTransaction.stream().filter(PendingTransaction::isReceivedFromLocalSource).count();
     final long remoteCount = pendingTransaction.size() - localCount;

ethereum/api/src/main/java/org/hyperledger/besu/ethereum/api/jsonrpc/internal/results/PendingTransactionResult.java

@@ -31,7 +31,7 @@ public class PendingTransactionResult implements TransactionResult {
   public PendingTransactionResult(final PendingTransaction pendingTransaction) {
     hash = pendingTransaction.getHash().toString();
     isReceivedFromLocalSource = pendingTransaction.isReceivedFromLocalSource();
-    addedToPoolAt = pendingTransaction.getAddedToPoolAt();
+    addedToPoolAt = Instant.ofEpochMilli(pendingTransaction.getAddedAt());
   }
 
   @JsonGetter(value = "hash")

ethereum/api/src/main/java/org/hyperledger/besu/ethereum/api/jsonrpc/internal/results/PendingTransactionsResult.java

@@ -16,24 +16,22 @@
 
 import org.hyperledger.besu.ethereum.eth.transactions.PendingTransaction;
 
-import java.util.Set;
-import java.util.stream.Collectors;
+import java.util.Collection;
+import java.util.List;
 
 import com.fasterxml.jackson.annotation.JsonValue;
 
 public class PendingTransactionsResult implements TransactionResult {
 
-  private final Set<PendingTransactionResult> pendingTransactionResults;
+  private final List<PendingTransactionResult> pendingTransactionResults;
 
-  public PendingTransactionsResult(final Set<PendingTransaction> pendingTransactionSet) {
+  public PendingTransactionsResult(final Collection<PendingTransaction> pendingTransactionSet) {
     pendingTransactionResults =
-        pendingTransactionSet.stream()
-            .map(PendingTransactionResult::new)
-            .collect(Collectors.toSet());
+        pendingTransactionSet.stream().map(PendingTransactionResult::new).toList();
   }
 
   @JsonValue
-  public Set<PendingTransactionResult> getResults() {
+  public List<PendingTransactionResult> getResults() {
     return pendingTransactionResults;
   }
 }

ethereum/api/src/main/java/org/hyperledger/besu/ethereum/api/jsonrpc/internal/results/transaction/pool/PendingTransactionFilter.java

@@ -23,10 +23,9 @@
 import org.hyperledger.besu.ethereum.core.Transaction;
 import org.hyperledger.besu.ethereum.eth.transactions.PendingTransaction;
 
+import java.util.Collection;
 import java.util.List;
 import java.util.Optional;
-import java.util.Set;
-import java.util.stream.Collectors;
 
 /**
  * This class allows to filter a list of pending transactions
@@ -43,16 +42,16 @@ public class PendingTransactionFilter {
   public static final String VALUE_FIELD = "value";
   public static final String NONCE_FIELD = "nonce";
 
-  public Set<Transaction> reduce(
-      final Set<PendingTransaction> pendingTransactions,
+  public Collection<Transaction> reduce(
+      final Collection<PendingTransaction> pendingTransactions,
       final List<Filter> filters,
       final int limit)
       throws InvalidJsonRpcParameters {
     return pendingTransactions.stream()
         .filter(pendingTx -> applyFilters(pendingTx, filters))
         .limit(limit)
         .map(PendingTransaction::getTransaction)
-        .collect(Collectors.toSet());
+        .toList();
   }
 
   private boolean applyFilters(

ethereum/api/src/main/java/org/hyperledger/besu/ethereum/api/jsonrpc/websocket/subscription/pending/PendingTransactionSubscriptionService.java

@@ -18,11 +18,11 @@
 import org.hyperledger.besu.ethereum.api.jsonrpc.websocket.subscription.SubscriptionManager;
 import org.hyperledger.besu.ethereum.api.jsonrpc.websocket.subscription.request.SubscriptionType;
 import org.hyperledger.besu.ethereum.core.Transaction;
-import org.hyperledger.besu.ethereum.eth.transactions.PendingTransactionListener;
+import org.hyperledger.besu.ethereum.eth.transactions.PendingTransactionAddedListener;
 
 import java.util.List;
 
-public class PendingTransactionSubscriptionService implements PendingTransactionListener {
+public class PendingTransactionSubscriptionService implements PendingTransactionAddedListener {
 
   private final SubscriptionManager subscriptionManager;