Priority senders (hyperledger#5959)

Signed-off-by: Fabio Di Fabio <[email protected]>
Co-authored-by: Sally MacFarlane <[email protected]>

CHANGELOG.md

@@ -6,7 +6,11 @@
 
 ### Breaking Changes
 
+### Deprecations
+- `--tx-pool-disable-locals` has been deprecated for removal in favor of `--tx-pool-no-local-priority`, no semantic change, only a renaming [#5959](https://github.com/hyperledger/besu/pull/5959)
+
 ### Additions and Improvements
+- New option `--tx-pool-priority-senders` to specify a list of senders, that has the effect to prioritize any transactions sent by these senders from any source [#5959](https://github.com/hyperledger/besu/pull/5959)
 
 ### Bug Fixes
 
@@ -46,7 +50,7 @@ By default, the txpool is tuned for mainnet usage, but if you are using private
 - Layered transaction pool implementation is now stable and enabled by default. If you want still to use the legacy implementation, use `--tx-pool=legacy`. 
   By default, the new transaction pool is capped at using 25MB of memory, this limit can be raised using `--layered-tx-pool-layer-max-capacity` options  [#5772](https://github.com/hyperledger/besu/pull/5772)
 - Tune G1GC to reduce Besu memory footprint, and new `besu-untuned` start scripts to run without any specific G1GC flags [#5879](https://github.com/hyperledger/besu/pull/5879)
-- Reduce `engine_forkchoiceUpdatedV?` response time by asynchronously process block added events in the transaction pool [#5909](https://github.com/hyperledger/besu/pull/5909) 
+- Reduce `engine_forkchoiceUpdatedV?` response time by asynchronously process block added events in the transaction pool [#5909](https://github.com/hyperledger/besu/pull/5909)
 
 ### Bug Fixes
 - do not create ignorable storage on revert storage-variables subcommand [#5830](https://github.com/hyperledger/besu/pull/5830) 
@@ -55,7 +59,6 @@ By default, the txpool is tuned for mainnet usage, but if you are using private
 
 ### Download Links
 https://hyperledger.jfrog.io/artifactory/besu-binaries/besu/23.10.0/besu-23.10.0.tar.gz / sha256: 3c75f3792bfdb0892705b378f0b8bfc14ef6cecf1d8afe711d8d8687ed6687cf
-
 https://hyperledger.jfrog.io/artifactory/besu-binaries/besu/23.10.0/besu-23.10.0.zip / sha256: d5dafff4c3cbf104bf75b34a9f108dcdd7b08d2759de75ec65cd997f38f52866
 
 ## 23.7.3

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

@@ -24,6 +24,7 @@
 import org.hyperledger.besu.cli.converter.PercentageConverter;
 import org.hyperledger.besu.cli.options.CLIOptions;
 import org.hyperledger.besu.cli.util.CommandLineUtils;
+import org.hyperledger.besu.datatypes.Address;
 import org.hyperledger.besu.datatypes.Wei;
 import org.hyperledger.besu.ethereum.eth.transactions.ImmutableTransactionPoolConfiguration;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPoolConfiguration;
@@ -32,19 +33,25 @@
 
 import java.io.File;
 import java.util.List;
+import java.util.Set;
 
 import picocli.CommandLine;
 
 /** The Transaction pool Cli stable options. */
 public class TransactionPoolOptions implements CLIOptions<TransactionPoolConfiguration> {
   private static final String TX_POOL_IMPLEMENTATION = "--tx-pool";
+  /** Use TX_POOL_NO_LOCAL_PRIORITY instead */
+  @Deprecated(forRemoval = true)
   private static final String TX_POOL_DISABLE_LOCALS = "--tx-pool-disable-locals";
+
+  private static final String TX_POOL_NO_LOCAL_PRIORITY = "--tx-pool-no-local-priority";
   private static final String TX_POOL_ENABLE_SAVE_RESTORE = "--tx-pool-enable-save-restore";
   private static final String TX_POOL_SAVE_FILE = "--tx-pool-save-file";
   private static final String TX_POOL_PRICE_BUMP = "--tx-pool-price-bump";
   private static final String RPC_TX_FEECAP = "--rpc-tx-feecap";
   private static final String STRICT_TX_REPLAY_PROTECTION_ENABLED_FLAG =
       "--strict-tx-replay-protection-enabled";
+  private static final String TX_POOL_PRIORITY_SENDERS = "--tx-pool-priority-senders";
 
   @CommandLine.Option(
       names = {TX_POOL_IMPLEMENTATION},
@@ -54,13 +61,13 @@ public class TransactionPoolOptions implements CLIOptions<TransactionPoolConfigu
   private TransactionPoolConfiguration.Implementation txPoolImplementation = LAYERED;
 
   @CommandLine.Option(
-      names = {TX_POOL_DISABLE_LOCALS},
+      names = {TX_POOL_NO_LOCAL_PRIORITY, TX_POOL_DISABLE_LOCALS},
       paramLabel = "<Boolean>",
       description =
-          "Set to true if transactions sent via RPC should have the same checks and not be prioritized over remote ones (default: ${DEFAULT-VALUE})",
+          "Set to true if senders of transactions sent via RPC should not have priority (default: ${DEFAULT-VALUE})",
       fallbackValue = "true",
       arity = "0..1")
-  private Boolean disableLocalTxs = TransactionPoolConfiguration.DEFAULT_DISABLE_LOCAL_TXS;
+  private Boolean noLocalPriority = TransactionPoolConfiguration.DEFAULT_NO_LOCAL_PRIORITY;
 
   @CommandLine.Option(
       names = {TX_POOL_ENABLE_SAVE_RESTORE},
@@ -104,6 +111,15 @@ public class TransactionPoolOptions implements CLIOptions<TransactionPoolConfigu
       arity = "0..1")
   private Boolean strictTxReplayProtectionEnabled = false;
 
+  @CommandLine.Option(
+      names = {TX_POOL_PRIORITY_SENDERS},
+      split = ",",
+      paramLabel = "Comma separated list of addresses",
+      description =
+          "Pending transactions sent exclusively by these addresses, from any source, are prioritized and only evicted after all others. If not specified, then only the senders submitting transactions via RPC have priority (default: ${DEFAULT-VALUE})",
+      arity = "1..*")
+  private Set<Address> prioritySenders = TransactionPoolConfiguration.DEFAULT_PRIORITY_SENDERS;
+
   @CommandLine.ArgGroup(
       validate = false,
       heading = "@|bold Tx Pool Layered Implementation Options|@%n")
@@ -200,11 +216,12 @@ public static TransactionPoolOptions fromConfig(final TransactionPoolConfigurati
     final TransactionPoolOptions options = TransactionPoolOptions.create();
     options.txPoolImplementation = config.getTxPoolImplementation();
     options.saveRestoreEnabled = config.getEnableSaveRestore();
-    options.disableLocalTxs = config.getDisableLocalTransactions();
+    options.noLocalPriority = config.getNoLocalPriority();
     options.priceBump = config.getPriceBump();
     options.txFeeCap = config.getTxFeeCap();
     options.saveFile = config.getSaveFile();
     options.strictTxReplayProtectionEnabled = config.getStrictTransactionReplayProtectionEnabled();
+    options.prioritySenders = config.getPrioritySenders();
     options.layeredOptions.txPoolLayerMaxCapacity =
         config.getPendingTransactionsLayerMaxCapacityBytes();
     options.layeredOptions.txPoolMaxPrioritized = config.getMaxPrioritizedTransactions();
@@ -242,11 +259,12 @@ public TransactionPoolConfiguration toDomainObject() {
     return ImmutableTransactionPoolConfiguration.builder()
         .txPoolImplementation(txPoolImplementation)
         .enableSaveRestore(saveRestoreEnabled)
-        .disableLocalTransactions(disableLocalTxs)
+        .noLocalPriority(noLocalPriority)
         .priceBump(priceBump)
         .txFeeCap(txFeeCap)
         .saveFile(saveFile)
         .strictTransactionReplayProtectionEnabled(strictTxReplayProtectionEnabled)
+        .prioritySenders(prioritySenders)
         .pendingTransactionsLayerMaxCapacityBytes(layeredOptions.txPoolLayerMaxCapacity)
         .maxPrioritizedTransactions(layeredOptions.txPoolMaxPrioritized)
         .maxFutureBySender(layeredOptions.txPoolMaxFutureBySender)

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

@@ -20,6 +20,7 @@
 
 import org.hyperledger.besu.cli.options.AbstractCLIOptionsTest;
 import org.hyperledger.besu.cli.options.OptionParser;
+import org.hyperledger.besu.datatypes.Address;
 import org.hyperledger.besu.datatypes.Wei;
 import org.hyperledger.besu.ethereum.eth.transactions.ImmutableTransactionPoolConfiguration;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPoolConfiguration;
@@ -73,20 +74,20 @@ public void pendingTransactionRetentionPeriod() {
 
   @Test
   public void disableLocalsDefault() {
-    internalTestSuccess(config -> assertThat(config.getDisableLocalTransactions()).isFalse());
+    internalTestSuccess(config -> assertThat(config.getNoLocalPriority()).isFalse());
   }
 
   @Test
   public void disableLocalsOn() {
     internalTestSuccess(
-        config -> assertThat(config.getDisableLocalTransactions()).isTrue(),
+        config -> assertThat(config.getNoLocalPriority()).isTrue(),
         "--tx-pool-disable-locals=true");
   }
 
   @Test
   public void disableLocalsOff() {
     internalTestSuccess(
-        config -> assertThat(config.getDisableLocalTransactions()).isFalse(),
+        config -> assertThat(config.getNoLocalPriority()).isFalse(),
         "--tx-pool-disable-locals=false");
   }
 
@@ -214,6 +215,61 @@ public void failIfLayeredOptionsWhenLegacySelectedByArg() {
         "--tx-pool-max-prioritized=1000");
   }
 
+  @Test
+  public void byDefaultNoPrioritySenders() {
+    internalTestSuccess(config -> assertThat(config.getPrioritySenders()).isEmpty());
+  }
+
+  @Test
+  public void onePrioritySenderWorks() {
+    final Address prioritySender = Address.fromHexString("0xABC123");
+    internalTestSuccess(
+        config -> assertThat(config.getPrioritySenders()).containsExactly(prioritySender),
+        "--tx-pool-priority-senders",
+        prioritySender.toHexString());
+  }
+
+  @Test
+  public void morePrioritySendersWorks() {
+    final Address prioritySender1 = Address.fromHexString("0xABC123");
+    final Address prioritySender2 = Address.fromHexString("0xDEF456");
+    final Address prioritySender3 = Address.fromHexString("0x789000");
+    internalTestSuccess(
+        config ->
+            assertThat(config.getPrioritySenders())
+                .containsExactly(prioritySender1, prioritySender2, prioritySender3),
+        "--tx-pool-priority-senders",
+        prioritySender1.toHexString()
+            + ","
+            + prioritySender2.toHexString()
+            + ","
+            + prioritySender3.toHexString());
+  }
+
+  @Test
+  public void atLeastOnePrioritySenders() {
+    internalTestFailure(
+        "Missing required parameter for option '--tx-pool-priority-senders' at index 0 (Comma separated list of addresses)",
+        "--tx-pool-priority-senders");
+  }
+
+  @Test
+  public void malformedListOfPrioritySenders() {
+    final Address prioritySender1 = Address.fromHexString("0xABC123");
+    final Address prioritySender2 = Address.fromHexString("0xDEF456");
+    final Address prioritySender3 = Address.fromHexString("0x789000");
+    internalTestFailure(
+        "Invalid value for option '--tx-pool-priority-senders' at index 0 (Comma separated list of addresses): "
+            + "cannot convert '0x0000000000000000000000000000000000abc123;0x0000000000000000000000000000000000def456' "
+            + "to Address (java.lang.IllegalArgumentException: Invalid odd-length hex binary representation)",
+        "--tx-pool-priority-senders",
+        prioritySender1.toHexString()
+            + ";"
+            + prioritySender2.toHexString()
+            + ","
+            + prioritySender3.toHexString());
+  }
+
   @Override
   protected TransactionPoolConfiguration createDefaultDomainObject() {
     return TransactionPoolConfiguration.DEFAULT;

besu/src/test/resources/everything_config.toml

@@ -175,7 +175,8 @@ tx-pool="layered"
 tx-pool-price-bump=13
 rpc-tx-feecap=2000000000000000000
 strict-tx-replay-protection-enabled=true
-tx-pool-disable-locals=false
+tx-pool-no-local-priority=false
+tx-pool-priority-senders=["0xABC0000000000000000000000000000000001234","0xDEF0000000000000000000000000000000001234"]
 tx-pool-enable-save-restore=true
 tx-pool-save-file="txpool.dump"
 ## Layered

consensus/merge/src/test/java/org/hyperledger/besu/consensus/merge/blockcreation/MergeCoordinatorTest.java

@@ -58,11 +58,11 @@
 import org.hyperledger.besu.ethereum.core.BlockHeaderTestFixture;
 import org.hyperledger.besu.ethereum.core.Difficulty;
 import org.hyperledger.besu.ethereum.core.MiningParameters;
-import org.hyperledger.besu.ethereum.core.Transaction;
 import org.hyperledger.besu.ethereum.core.TransactionTestFixture;
 import org.hyperledger.besu.ethereum.eth.manager.EthContext;
 import org.hyperledger.besu.ethereum.eth.sync.backwardsync.BackwardSyncContext;
 import org.hyperledger.besu.ethereum.eth.transactions.ImmutableTransactionPoolConfiguration;
+import org.hyperledger.besu.ethereum.eth.transactions.PendingTransaction;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionBroadcaster;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPool;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPoolConfiguration;
@@ -314,8 +314,8 @@ public void exceptionDuringBuildingBlockShouldNotBeInvalid()
             invocation -> {
               if (retries.getAndIncrement() < txPerBlock) {
                 // a new transaction every time a block is built
-                transactions.addLocalTransaction(
-                    createTransaction(retries.get() - 1), Optional.empty());
+                transactions.addTransaction(
+                    createLocalTransaction(retries.get() - 1), Optional.empty());
               } else {
                 // when we have 5 transactions finalize block creation
                 willThrow.finalizeProposalById(
@@ -387,8 +387,8 @@ public void shouldContinueBuildingBlocksUntilFinalizeIsCalled()
             invocation -> {
               if (retries.getAndIncrement() < 5) {
                 // a new transaction every time a block is built
-                transactions.addLocalTransaction(
-                    createTransaction(retries.get() - 1), Optional.empty());
+                transactions.addTransaction(
+                    createLocalTransaction(retries.get() - 1), Optional.empty());
               } else {
                 // when we have 5 transactions finalize block creation
                 coordinator.finalizeProposalById(
@@ -506,7 +506,7 @@ public void shouldRetryBlockCreationOnRecoverableError()
         .when(mergeContext)
         .putPayloadById(any());
 
-    transactions.addLocalTransaction(createTransaction(0), Optional.empty());
+    transactions.addTransaction(createLocalTransaction(0), Optional.empty());
 
     var payloadId =
         coordinator.preparePayload(
@@ -643,8 +643,8 @@ public void shouldNotStartAnotherBlockCreationJobIfCalledAgainWithTheSamePayload
             invocation -> {
               if (retries.getAndIncrement() < 5) {
                 // a new transaction every time a block is built
-                transactions.addLocalTransaction(
-                    createTransaction(retries.get() - 1), Optional.empty());
+                transactions.addTransaction(
+                    createLocalTransaction(retries.get() - 1), Optional.empty());
               } else {
                 // when we have 5 transactions finalize block creation
                 coordinator.finalizeProposalById(
@@ -1022,20 +1022,24 @@ private BlockHeader nextBlockHeader(
         .buildHeader();
   }
 
-  private Transaction createTransaction(final long transactionNumber) {
-    return new TransactionTestFixture()
-        .value(Wei.of(transactionNumber + 1))
-        .to(Optional.of(Address.ZERO))
-        .gasLimit(53000L)
-        .gasPrice(
-            Wei.fromHexString("0x00000000000000000000000000000000000000000000000000000013b9aca00"))
-        .maxFeePerGas(
-            Optional.of(
+  private PendingTransaction createLocalTransaction(final long transactionNumber) {
+    return PendingTransaction.newPendingTransaction(
+        new TransactionTestFixture()
+            .value(Wei.of(transactionNumber + 1))
+            .to(Optional.of(Address.ZERO))
+            .gasLimit(53000L)
+            .gasPrice(
                 Wei.fromHexString(
-                    "0x00000000000000000000000000000000000000000000000000000013b9aca00")))
-        .maxPriorityFeePerGas(Optional.of(Wei.of(100_000)))
-        .nonce(transactionNumber)
-        .createTransaction(KEYS1);
+                    "0x00000000000000000000000000000000000000000000000000000013b9aca00"))
+            .maxFeePerGas(
+                Optional.of(
+                    Wei.fromHexString(
+                        "0x00000000000000000000000000000000000000000000000000000013b9aca00")))
+            .maxPriorityFeePerGas(Optional.of(Wei.of(100_000)))
+            .nonce(transactionNumber)
+            .createTransaction(KEYS1),
+        true,
+        true);
   }
 
   private static BlockHeader mockBlockHeader() {

datatypes/src/main/java/org/hyperledger/besu/datatypes/PendingTransaction.java

@@ -30,6 +30,13 @@ public interface PendingTransaction {
    */
   boolean isReceivedFromLocalSource();
 
+  /**
+   * Should this transaction be prioritized?
+   *
+   * @return true if it is a transaction with priority
+   */
+  boolean hasPriority();
+
   /**
    * Timestamp in millisecond when this transaction has been added to the pool
    *

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

@@ -50,6 +50,7 @@
 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.PendingTransaction;
 import org.hyperledger.besu.ethereum.eth.transactions.PendingTransactions;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionBroadcaster;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPool;
@@ -90,7 +91,8 @@ public class EthGetFilterChangesIntegrationTest {
 
   private static final int MAX_TRANSACTIONS = 5;
   private static final KeyPair keyPair = SignatureAlgorithmFactory.getInstance().generateKeyPair();
-  private final Transaction transaction = createTransaction(1);
+  private final PendingTransaction pendingTransaction =
+      new PendingTransaction.Local((createTransaction(1)));
   private FilterManager filterManager;
   private EthGetFilterChanges method;
 
@@ -193,7 +195,7 @@ public void shouldReturnHashesIfNewBlocks() {
     JsonRpcResponse actual = method.response(request);
     assertThat(actual).usingRecursiveComparison().isEqualTo(expected);
 
-    final Block block = appendBlock(transaction);
+    final Block block = appendBlock(pendingTransaction.getTransaction());
 
     // We've added one block, so there should be one new hash.
     expected = new JsonRpcSuccessResponse(null, Lists.newArrayList(block.getHash().toString()));
@@ -223,11 +225,12 @@ public void shouldReturnHashesIfNewPendingTransactions() {
     JsonRpcResponse actual = method.response(request);
     assertThat(actual).usingRecursiveComparison().isEqualTo(expected);
 
-    transactions.addRemoteTransaction(transaction, Optional.empty());
+    transactions.addTransaction(pendingTransaction, Optional.empty());
 
     // We've added one transaction, so there should be one new hash.
     expected =
-        new JsonRpcSuccessResponse(null, Lists.newArrayList(String.valueOf(transaction.getHash())));
+        new JsonRpcSuccessResponse(
+            null, Lists.newArrayList(String.valueOf(pendingTransaction.getHash())));
     actual = method.response(request);
     assertThat(actual).usingRecursiveComparison().isEqualTo(expected);