Introduce a more versatile txpool lifecycle logging (#9333)

Signed-off-by: Fabio Di Fabio <[email protected]>

Author: fab-10

ethereum/blockcreation/src/test/java/org/hyperledger/besu/ethereum/blockcreation/PoWMinerExecutorTest.java

@@ -109,12 +109,15 @@ private TransactionPool createTransactionPool() {
 
     final EthContext ethContext = mock(EthContext.class, RETURNS_DEEP_STUBS);
     when(ethContext.getEthPeers().subscribeConnect(any())).thenReturn(1L);
+    final var protocolContext = mock(ProtocolContext.class, RETURNS_DEEP_STUBS);
+    final var blockHeader = Optional.of(mockBlockHeader());
+    when(protocolContext.getBlockchain().getBlockHeader(any())).thenReturn(blockHeader);
 
     final TransactionPool transactionPool =
         new TransactionPool(
             () -> pendingTransactions,
             mock(ProtocolSchedule.class),
-            mock(ProtocolContext.class),
+            protocolContext,
             mock(TransactionBroadcaster.class),
             ethContext,
             new TransactionPoolMetrics(new NoOpMetricsSystem()),

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

@@ -15,6 +15,9 @@
 package org.hyperledger.besu.ethereum.eth.transactions;
 
 import static org.hyperledger.besu.ethereum.eth.transactions.PendingTransaction.MAX_SCORE;
+import static org.hyperledger.besu.ethereum.eth.transactions.TransactionPoolStructuredLogUtils.logInvalid;
+import static org.hyperledger.besu.ethereum.eth.transactions.TransactionPoolStructuredLogUtils.logStart;
+import static org.hyperledger.besu.ethereum.eth.transactions.TransactionPoolStructuredLogUtils.logStop;
 import static org.hyperledger.besu.ethereum.transaction.TransactionInvalidReason.CHAIN_HEAD_NOT_AVAILABLE;
 import static org.hyperledger.besu.ethereum.transaction.TransactionInvalidReason.CHAIN_HEAD_WORLD_STATE_NOT_AVAILABLE;
 import static org.hyperledger.besu.ethereum.transaction.TransactionInvalidReason.INTERNAL_ERROR;
@@ -58,9 +61,9 @@
 import java.io.FileReader;
 import java.io.FileWriter;
 import java.io.IOException;
-import java.math.BigInteger;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Comparator;
@@ -100,7 +103,6 @@
  */
 public class TransactionPool implements BlockAddedObserver {
   private static final Logger LOG = LoggerFactory.getLogger(TransactionPool.class);
-  private static final Logger LOG_FOR_REPLAY = LoggerFactory.getLogger("LOG_FOR_REPLAY");
   private final Supplier<PendingTransactions> pendingTransactionsSupplier;
   private final BlobCache cacheForBlobsOfTransactionsAddedToABlock;
   private volatile PendingTransactions pendingTransactions = new DisabledPendingTransactions();
@@ -147,46 +149,6 @@ public TransactionPool(
     subscribeDroppedTransactions(transactionBroadcaster);
   }
 
-  private void initLogForReplay() {
-    LOG_FOR_REPLAY.trace("START");
-    // log the initial block header data
-    LOG_FOR_REPLAY
-        .atTrace()
-        .setMessage("{},{},{},{}")
-        .addArgument(() -> getChainHeadBlockHeader().map(BlockHeader::getNumber).orElse(0L))
-        .addArgument(
-            () ->
-                getChainHeadBlockHeader()
-                    .flatMap(BlockHeader::getBaseFee)
-                    .map(Wei::getAsBigInteger)
-                    .orElse(BigInteger.ZERO))
-        .addArgument(() -> getChainHeadBlockHeader().map(BlockHeader::getGasUsed).orElse(0L))
-        .addArgument(() -> getChainHeadBlockHeader().map(BlockHeader::getGasLimit).orElse(0L))
-        .log();
-    // log the priority senders
-    LOG_FOR_REPLAY
-        .atTrace()
-        .setMessage("{}")
-        .addArgument(
-            () ->
-                configuration.getPrioritySenders().stream()
-                    .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();
-    // log configuration: minScore
-    LOG_FOR_REPLAY.atTrace().setMessage("{}").addArgument(configuration::getMinScore).log();
-  }
-
   @VisibleForTesting
   void handleConnect(final EthPeer peer) {
     transactionBroadcaster.relayTransactionPoolTo(
@@ -196,17 +158,20 @@ void handleConnect(final EthPeer peer) {
   public ValidationResult<TransactionInvalidReason> addTransactionViaApi(
       final Transaction transaction) {
 
-    final var result = addTransaction(transaction, true, MAX_SCORE);
+    final boolean hasPriority = isPriorityTransaction(transaction, true);
+    final var result = addTransaction(transaction, true, hasPriority, MAX_SCORE);
     if (result.isValid()) {
       localSenders.add(transaction.getSender());
       transactionBroadcaster.onTransactionsAdded(List.of(transaction));
+    } else {
+      logInvalid(transaction, result, true, hasPriority);
     }
     return result;
   }
 
   public Map<Hash, ValidationResult<TransactionInvalidReason>> addRemoteTransactions(
       final Collection<Transaction> transactions) {
-    final long started = System.currentTimeMillis();
+    final long started = System.nanoTime();
     final int initialCount = transactions.size();
     final List<Transaction> addedTransactions = new ArrayList<>(initialCount);
     LOG.trace("Adding {} remote transactions", initialCount);
@@ -217,27 +182,25 @@ public Map<Hash, ValidationResult<TransactionInvalidReason>> addRemoteTransactio
                 Collectors.toMap(
                     Transaction::getHash,
                     transaction -> {
-                      final var result = addTransaction(transaction, false, MAX_SCORE);
+                      final boolean hasPriority = isPriorityTransaction(transaction, false);
+                      final var result = addTransaction(transaction, false, hasPriority, MAX_SCORE);
                       if (result.isValid()) {
                         addedTransactions.add(transaction);
+                      } else {
+                        logInvalid(transaction, result, false, hasPriority);
                       }
                       return result;
                     },
                     (transaction1, transaction2) -> transaction1));
 
     if (isEnabled()) {
-      LOG_FOR_REPLAY
-          .atTrace()
-          .setMessage("S,{}")
-          .addArgument(() -> pendingTransactions.logStats())
-          .log();
+      TransactionPoolStructuredLogUtils.logStats(pendingTransactions);
     }
 
     LOG.atTrace()
-        .setMessage(
-            "Added {} transactions to the pool in {}ms, {} not added, current pool stats {}")
+        .setMessage("Added {} transactions to the pool in {}, {} not added, current pool stats {}")
         .addArgument(addedTransactions::size)
-        .addArgument(() -> System.currentTimeMillis() - started)
+        .addArgument(() -> Duration.ofNanos(System.nanoTime() - started))
         .addArgument(() -> initialCount - addedTransactions.size())
         .addArgument(pendingTransactions::logStats)
         .log();
@@ -249,9 +212,10 @@ public Map<Hash, ValidationResult<TransactionInvalidReason>> addRemoteTransactio
   }
 
   private ValidationResult<TransactionInvalidReason> addTransaction(
-      final Transaction baseTransaction, final boolean isLocal, final byte score) {
-
-    final boolean hasPriority = isPriorityTransaction(baseTransaction, isLocal);
+      final Transaction baseTransaction,
+      final boolean isLocal,
+      final boolean hasPriority,
+      final byte score) {
 
     if (pendingTransactions.containsTransaction(baseTransaction)) {
       LOG.atTrace()
@@ -649,7 +613,7 @@ static ValidationResultAndAccount invalid(final TransactionInvalidReason reason)
 
   public CompletableFuture<Void> setEnabled() {
     if (!isEnabled()) {
-      initLogForReplay();
+      logStart(configuration, getChainHeadBlockHeader());
       pendingTransactions = pendingTransactionsSupplier.get();
       pendingTransactionsListenersProxy.subscribe();
       isPoolEnabled.set(true);
@@ -696,7 +660,7 @@ public CompletableFuture<Void> setDisabled() {
                     return null;
                   });
       pendingTransactions = new DisabledPendingTransactions();
-      LOG_FOR_REPLAY.trace("STOP");
+      logStop();
       return saveOperation;
     }
     return CompletableFuture.completedFuture(null);
@@ -918,9 +882,9 @@ private void executeLoadFromDisk() {
                                       Bytes.fromBase64String(
                                           line.substring(scoreStr.length() + 1))),
                                   EncodingContext.POOLED_TRANSACTION);
-
+                          final boolean hasPriority = isPriorityTransaction(tx, isLocal);
                           final ValidationResult<TransactionInvalidReason> result =
-                              addTransaction(tx, isLocal, score);
+                              addTransaction(tx, isLocal, hasPriority, score);
                           return result.isValid() ? "OK" : result.getInvalidReason().name();
                         })
                     .collect(Collectors.groupingBy(Function.identity(), Collectors.counting()));