[PIE-1710] Expose a CLI option to configure the life time of transact…

…ion messages. (#1610)

* [PIE-1710] Expose a CLI option to configure the life time of transaction messages.

- add a cli option `--tx-pool-message-keep-alive-seconds`

* rename cli option

* change cli option description

* spotless apply

* Update everything_config.toml

ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/transactions/TransactionPool.java

@@ -55,6 +55,8 @@
 public class TransactionPool implements BlockAddedObserver {
 
   private static final Logger LOG = getLogger();
+  public static final int DEFAULT_TX_MSG_KEEP_ALIVE = 60;
+
   private static final long SYNC_TOLERANCE = 100L;
   private static final String REMOTE = "remote";
   private static final String LOCAL = "local";

ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/transactions/TransactionPoolFactory.java

@@ -34,7 +34,8 @@ public static TransactionPool createTransactionPool(
       final MetricsSystem metricsSystem,
       final SyncState syncState,
       final int maxTransactionRetentionHours,
-      final Wei minTransactionGasPrice) {
+      final Wei minTransactionGasPrice,
+      final int txMessageKeepAliveSeconds) {
 
     final PendingTransactions pendingTransactions =
         new PendingTransactions(
@@ -65,7 +66,8 @@ public static TransactionPool createTransactionPool(
                 metricsSystem.createCounter(
                     PantheonMetricCategory.TRANSACTION_POOL,
                     "transactions_messages_skipped_total",
-                    "Total number of transactions messages skipped by the processor.")));
+                    "Total number of transactions messages skipped by the processor.")),
+            txMessageKeepAliveSeconds);
 
     ethContext.getEthMessages().subscribe(EthPV62.TRANSACTIONS, transactionsMessageHandler);
     protocolContext.getBlockchain().observeBlockAdded(transactionPool);

ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/transactions/TransactionsMessageHandler.java

@@ -24,15 +24,17 @@
 
 class TransactionsMessageHandler implements MessageCallback {
 
-  private static final Duration TX_KEEP_ALIVE = Duration.ofMinutes(1);
   private final TransactionsMessageProcessor transactionsMessageProcessor;
   private final EthScheduler scheduler;
+  private final Duration txMsgKeepAlive;
 
   public TransactionsMessageHandler(
       final EthScheduler scheduler,
-      final TransactionsMessageProcessor transactionsMessageProcessor) {
+      final TransactionsMessageProcessor transactionsMessageProcessor,
+      final int txMsgKeepAliveSeconds) {
     this.scheduler = scheduler;
     this.transactionsMessageProcessor = transactionsMessageProcessor;
+    this.txMsgKeepAlive = Duration.ofSeconds(txMsgKeepAliveSeconds);
   }
 
   @Override
@@ -42,6 +44,6 @@ public void exec(final EthMessage message) {
     scheduler.scheduleTxWorkerTask(
         () ->
             transactionsMessageProcessor.processTransactionsMessage(
-                message.getPeer(), transactionsMessage, startedAt, TX_KEEP_ALIVE));
+                message.getPeer(), transactionsMessage, startedAt, txMsgKeepAlive));
   }
 }

ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/manager/EthProtocolManagerTest.java

@@ -52,6 +52,7 @@
 import tech.pegasys.pantheon.ethereum.eth.messages.TransactionsMessage;
 import tech.pegasys.pantheon.ethereum.eth.sync.state.SyncState;
 import tech.pegasys.pantheon.ethereum.eth.transactions.PendingTransactions;
+import tech.pegasys.pantheon.ethereum.eth.transactions.TransactionPool;
 import tech.pegasys.pantheon.ethereum.eth.transactions.TransactionPoolFactory;
 import tech.pegasys.pantheon.ethereum.mainnet.MainnetProtocolSchedule;
 import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
@@ -1082,7 +1083,8 @@ public void transactionMessagesGoToTheCorrectExecutor() {
           metricsSystem,
           mock(SyncState.class),
           PendingTransactions.DEFAULT_TX_RETENTION_HOURS,
-          Wei.ZERO);
+          Wei.ZERO,
+          TransactionPool.DEFAULT_TX_MSG_KEEP_ALIVE);
 
       // Send just a transaction message.
       final PeerConnection peer = setupPeer(ethManager, (cap, msg, connection) -> {});

ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/transactions/TestNode.java

@@ -152,7 +152,8 @@ public TestNode(
             metricsSystem,
             syncState,
             PendingTransactions.DEFAULT_TX_RETENTION_HOURS,
-            Wei.ZERO);
+            Wei.ZERO,
+            TransactionPool.DEFAULT_TX_MSG_KEEP_ALIVE);
 
     networkRunner.start();
     selfPeer = DefaultPeer.fromEnodeURL(network.getLocalEnode().get());

pantheon/src/main/java/tech/pegasys/pantheon/cli/PantheonCommand.java

@@ -51,6 +51,7 @@
 import tech.pegasys.pantheon.ethereum.eth.sync.SynchronizerConfiguration;
 import tech.pegasys.pantheon.ethereum.eth.sync.TrailingPeerRequirements;
 import tech.pegasys.pantheon.ethereum.eth.transactions.PendingTransactions;
+import tech.pegasys.pantheon.ethereum.eth.transactions.TransactionPool;
 import tech.pegasys.pantheon.ethereum.graphql.GraphQLConfiguration;
 import tech.pegasys.pantheon.ethereum.jsonrpc.JsonRpcConfiguration;
 import tech.pegasys.pantheon.ethereum.jsonrpc.RpcApi;
@@ -566,6 +567,13 @@ void setBannedNodeIds(final List<String> values) {
       arity = "1")
   private final Integer pendingTxRetentionPeriod = PendingTransactions.DEFAULT_TX_RETENTION_HOURS;
 
+  @Option(
+      names = {"--tx-pool-keep-alive-seconds"},
+      paramLabel = MANDATORY_INTEGER_FORMAT_HELP,
+      description = "Keep alive of transactions in seconds (default: ${DEFAULT-VALUE})",
+      arity = "1")
+  private final Integer txMessageKeepAliveSeconds = TransactionPool.DEFAULT_TX_MSG_KEEP_ALIVE;
+
   // Inner class so we can get to loggingLevel.
   public class PantheonExceptionHandler
       extends CommandLine.AbstractHandler<List<Object>, PantheonExceptionHandler>
@@ -817,6 +825,7 @@ PantheonController<?> buildController() {
               new MiningParameters(coinbase, minTransactionGasPrice, extraData, isMiningEnabled))
           .maxPendingTransactions(txPoolMaxSize)
           .pendingTransactionRetentionPeriod(pendingTxRetentionPeriod)
+          .txMessageKeepAliveSeconds(txMessageKeepAliveSeconds)
           .nodePrivateKeyFile(nodePrivateKeyFile())
           .metricsSystem(metricsSystem.get())
           .privacyParameters(privacyParameters())

pantheon/src/main/java/tech/pegasys/pantheon/controller/PantheonControllerBuilder.java

@@ -74,6 +74,7 @@ public abstract class PantheonControllerBuilder<C> {
   protected Clock clock;
   protected Integer maxPendingTransactions;
   protected Integer pendingTransactionRetentionPeriod;
+  protected Integer txMessageKeepAliveSeconds = TransactionPool.DEFAULT_TX_MSG_KEEP_ALIVE;
   protected KeyPair nodeKeys;
   private StorageProvider storageProvider;
   private final List<Runnable> shutdownActions = new ArrayList<>();
@@ -159,6 +160,12 @@ public PantheonControllerBuilder<C> pendingTransactionRetentionPeriod(
     return this;
   }
 
+  public PantheonControllerBuilder<C> txMessageKeepAliveSeconds(
+      final int txMessageKeepAliveSeconds) {
+    this.txMessageKeepAliveSeconds = txMessageKeepAliveSeconds;
+    return this;
+  }
+
   public PantheonController<C> build() throws IOException {
     checkNotNull(genesisConfig, "Missing genesis config");
     checkNotNull(syncConfig, "Missing sync config");
@@ -235,7 +242,8 @@ public PantheonController<C> build() throws IOException {
             metricsSystem,
             syncState,
             pendingTransactionRetentionPeriod,
-            miningParameters.getMinTransactionGasPrice());
+            miningParameters.getMinTransactionGasPrice(),
+            txMessageKeepAliveSeconds);
 
     final MiningCoordinator miningCoordinator =
         createMiningCoordinator(

pantheon/src/test/java/tech/pegasys/pantheon/cli/CommandTestAbstract.java

@@ -134,6 +134,8 @@ public void initMocks() throws Exception {
     when(mockControllerBuilder.maxPendingTransactions(anyInt())).thenReturn(mockControllerBuilder);
     when(mockControllerBuilder.pendingTransactionRetentionPeriod(anyInt()))
         .thenReturn(mockControllerBuilder);
+    when(mockControllerBuilder.txMessageKeepAliveSeconds(anyInt()))
+        .thenReturn(mockControllerBuilder);
     when(mockControllerBuilder.nodePrivateKeyFile(any())).thenReturn(mockControllerBuilder);
     when(mockControllerBuilder.metricsSystem(any())).thenReturn(mockControllerBuilder);
     when(mockControllerBuilder.privacyParameters(any())).thenReturn(mockControllerBuilder);

pantheon/src/test/java/tech/pegasys/pantheon/cli/PantheonCommandTest.java

@@ -2621,4 +2621,27 @@ public void pendingTransactionRetentionPeriod() {
     assertThat(commandOutput.toString()).isEmpty();
     assertThat(commandErrorOutput.toString()).isEmpty();
   }
+
+  @Test
+  public void txMessageKeepAliveSeconds() {
+    final int txMessageKeepAliveSeconds = 999;
+    parseCommand("--tx-pool-keep-alive-seconds", String.valueOf(txMessageKeepAliveSeconds));
+
+    verify(mockControllerBuilder).txMessageKeepAliveSeconds(intArgumentCaptor.capture());
+    verify(mockControllerBuilder).txMessageKeepAliveSeconds(eq(txMessageKeepAliveSeconds));
+
+    assertThat(commandOutput.toString()).isEmpty();
+    assertThat(commandErrorOutput.toString()).isEmpty();
+  }
+
+  @Test
+  public void txMessageKeepAliveSecondsWithInvalidInputShouldFail() {
+    parseCommand("--tx-pool-keep-alive-seconds", "acbd");
+
+    verifyZeroInteractions(mockRunnerBuilder);
+
+    assertThat(commandOutput.toString()).isEmpty();
+    assertThat(commandErrorOutput.toString())
+        .contains("Invalid value for option '--tx-pool-keep-alive-seconds': 'acbd' is not an int");
+  }
 }

pantheon/src/test/resources/everything_config.toml

@@ -97,3 +97,4 @@ privacy-precompiled-address=9
 tx-pool-retention-hours=999
 
 tx-pool-max-size=1234
+tx-pool-keep-alive-seconds=60