Layered txpool: do not send notifications when moving tx between layers

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

CHANGELOG.md

@@ -21,6 +21,7 @@
 - Correctly drops messages that exceeds local message size limit [#5455](https://github.com/hyperledger/besu/pull/7507)
 - **DebugMetrics**: Fixed a `ClassCastException` occurring in `DebugMetrics` when handling nested metric structures. Previously, `Double` values within these structures were incorrectly cast to `Map` objects, leading to errors. This update allows for proper handling of both direct values and nested structures at the same level. Issue# [#7383](https://github.com/hyperledger/besu/pull/7383)
 - `evmtool` was not respecting the `--genesis` setting, resulting in unexpected trace results. [#7433](https://github.com/hyperledger/besu/pull/7433)
+- Layered txpool: do not send notifications when moving tx between layers [#7539](https://github.com/hyperledger/besu/pull/7539)
 
 ## 24.8.0
 

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

@@ -14,6 +14,7 @@
  */
 package org.hyperledger.besu.ethereum.eth.transactions.layered;
 
+import static org.hyperledger.besu.ethereum.eth.transactions.layered.TransactionsLayer.AddReason.MOVE;
 import static org.hyperledger.besu.ethereum.eth.transactions.layered.TransactionsLayer.RemovalReason.EVICTED;
 import static org.hyperledger.besu.ethereum.eth.transactions.layered.TransactionsLayer.RemovalReason.FOLLOW_INVALIDATED;
 
@@ -77,7 +78,7 @@ private void pushDown(
               senderTxs.remove(txToRemove.getNonce());
               processRemove(senderTxs, txToRemove.getTransaction(), FOLLOW_INVALIDATED);
             })
-        .forEach(followingTx -> nextLayer.add(followingTx, gap));
+        .forEach(followingTx -> nextLayer.add(followingTx, gap, MOVE));
   }
 
   @Override

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

@@ -19,6 +19,7 @@
 import static org.hyperledger.besu.ethereum.eth.transactions.TransactionAddedResult.ALREADY_KNOWN;
 import static org.hyperledger.besu.ethereum.eth.transactions.TransactionAddedResult.REJECTED_UNDERPRICED_REPLACEMENT;
 import static org.hyperledger.besu.ethereum.eth.transactions.TransactionAddedResult.TRY_NEXT_LAYER;
+import static org.hyperledger.besu.ethereum.eth.transactions.layered.TransactionsLayer.AddReason.MOVE;
 import static org.hyperledger.besu.ethereum.eth.transactions.layered.TransactionsLayer.RemovalReason.CONFIRMED;
 import static org.hyperledger.besu.ethereum.eth.transactions.layered.TransactionsLayer.RemovalReason.CROSS_LAYER_REPLACED;
 import static org.hyperledger.besu.ethereum.eth.transactions.layered.TransactionsLayer.RemovalReason.EVICTED;
@@ -169,7 +170,8 @@ protected abstract TransactionAddedResult canAdd(
       final PendingTransaction pendingTransaction, final int gap);
 
   @Override
-  public TransactionAddedResult add(final PendingTransaction pendingTransaction, final int gap) {
+  public TransactionAddedResult add(
+      final PendingTransaction pendingTransaction, final int gap, final AddReason addReason) {
 
     // is replacing an existing one?
     TransactionAddedResult addStatus = maybeReplaceTransaction(pendingTransaction);
@@ -178,7 +180,7 @@ public TransactionAddedResult add(final PendingTransaction pendingTransaction, f
     }
 
     if (addStatus.equals(TRY_NEXT_LAYER)) {
-      return addToNextLayer(pendingTransaction, gap);
+      return addToNextLayer(pendingTransaction, gap, addReason);
     }
 
     if (addStatus.isSuccess()) {
@@ -192,7 +194,10 @@ public TransactionAddedResult add(final PendingTransaction pendingTransaction, f
         tryFillGap(addStatus, pendingTransaction, getRemainingPromotionsPerType());
       }
 
-      ethScheduler.scheduleTxWorkerTask(() -> notifyTransactionAdded(pendingTransaction));
+      if (addReason.sendNotification()) {
+        ethScheduler.scheduleTxWorkerTask(() -> notifyTransactionAdded(pendingTransaction));
+      }
+
     } else {
       final var rejectReason = addStatus.maybeInvalidReason().orElseThrow();
       metrics.incrementRejected(pendingTransaction, rejectReason, name());
@@ -302,24 +307,26 @@ public PendingTransaction promoteFor(
   }
 
   private TransactionAddedResult addToNextLayer(
-      final PendingTransaction pendingTransaction, final int distance) {
+      final PendingTransaction pendingTransaction, final int distance, final AddReason addReason) {
     return addToNextLayer(
         txsBySender.getOrDefault(pendingTransaction.getSender(), EMPTY_SENDER_TXS),
         pendingTransaction,
-        distance);
+        distance,
+        addReason);
   }
 
   protected TransactionAddedResult addToNextLayer(
       final NavigableMap<Long, PendingTransaction> senderTxs,
       final PendingTransaction pendingTransaction,
-      final int distance) {
+      final int distance,
+      final AddReason addReason) {
     final int nextLayerDistance;
     if (senderTxs.isEmpty()) {
       nextLayerDistance = distance;
     } else {
       nextLayerDistance = (int) (pendingTransaction.getNonce() - (senderTxs.lastKey() + 1));
     }
-    return nextLayer.add(pendingTransaction, nextLayerDistance);
+    return nextLayer.add(pendingTransaction, nextLayerDistance, addReason);
   }
 
   private void processAdded(final PendingTransaction addedTx) {
@@ -353,7 +360,7 @@ private void evict(final long spaceToFree, final int txsToEvict) {
         ++evictedCount;
         evictedSize += lastTx.memorySize();
         // evicted can always be added to the next layer
-        addToNextLayer(lessReadySenderTxs, lastTx, 0);
+        addToNextLayer(lessReadySenderTxs, lastTx, 0, MOVE);
       }
 
       if (lessReadySenderTxs.isEmpty()) {

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

@@ -14,6 +14,7 @@
  */
 package org.hyperledger.besu.ethereum.eth.transactions.layered;
 
+import static org.hyperledger.besu.ethereum.eth.transactions.layered.TransactionsLayer.AddReason.MOVE;
 import static org.hyperledger.besu.ethereum.eth.transactions.layered.TransactionsLayer.RemovalReason.BELOW_BASE_FEE;
 
 import org.hyperledger.besu.datatypes.Wei;
@@ -133,7 +134,7 @@ protected void internalBlockAdded(final BlockHeader blockHeader, final FeeMarket
                   .addArgument(newNextBlockBaseFee::toHumanReadableString)
                   .log();
               processEvict(senderTxs, demoteTx, BELOW_BASE_FEE);
-              addToNextLayer(senderTxs, demoteTx, 0);
+              addToNextLayer(senderTxs, demoteTx, 0, MOVE);
             }
           });
 

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

@@ -75,7 +75,8 @@ public List<PendingTransaction> getAll() {
   }
 
   @Override
-  public TransactionAddedResult add(final PendingTransaction pendingTransaction, final int gap) {
+  public TransactionAddedResult add(
+      final PendingTransaction pendingTransaction, final int gap, final AddReason reason) {
     notifyTransactionDropped(pendingTransaction);
     metrics.incrementRemoved(pendingTransaction, DROPPED.label(), name());
     ++droppedCount;

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

@@ -20,6 +20,7 @@
 import static org.hyperledger.besu.ethereum.eth.transactions.TransactionAddedResult.ALREADY_KNOWN;
 import static org.hyperledger.besu.ethereum.eth.transactions.TransactionAddedResult.INTERNAL_ERROR;
 import static org.hyperledger.besu.ethereum.eth.transactions.TransactionAddedResult.NONCE_TOO_FAR_IN_FUTURE_FOR_SENDER;
+import static org.hyperledger.besu.ethereum.eth.transactions.layered.TransactionsLayer.AddReason.NEW;
 import static org.hyperledger.besu.ethereum.eth.transactions.layered.TransactionsLayer.RemovalReason.INVALIDATED;
 import static org.hyperledger.besu.ethereum.eth.transactions.layered.TransactionsLayer.RemovalReason.RECONCILED;
 
@@ -100,7 +101,7 @@ public synchronized TransactionAddedResult addTransaction(
     }
 
     try {
-      return prioritizedTransactions.add(pendingTransaction, (int) nonceDistance);
+      return prioritizedTransactions.add(pendingTransaction, (int) nonceDistance, NEW);
     } catch (final Throwable throwable) {
       return reconcileAndRetryAdd(
           pendingTransaction, stateSenderNonce, (int) nonceDistance, throwable);
@@ -123,7 +124,7 @@ private TransactionAddedResult reconcileAndRetryAdd(
         .log();
     reconcileSender(pendingTransaction.getSender(), stateSenderNonce);
     try {
-      return prioritizedTransactions.add(pendingTransaction, nonceDistance);
+      return prioritizedTransactions.add(pendingTransaction, nonceDistance, NEW);
     } catch (final Throwable throwable2) {
       // the error should have been solved by the reconcile, logging at higher level now
       LOG.atWarn()
@@ -210,7 +211,7 @@ private void reconcileSender(final Address sender, final long stateSenderNonce)
       final long lowestNonce = reAddTxs.getFirst().getNonce();
       final int newNonceDistance = (int) Math.max(0, lowestNonce - stateSenderNonce);
 
-      reAddTxs.forEach(ptx -> prioritizedTransactions.add(ptx, newNonceDistance));
+      reAddTxs.forEach(ptx -> prioritizedTransactions.add(ptx, newNonceDistance, NEW));
     }
 
     LOG.atDebug()

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

@@ -41,7 +41,19 @@ public interface TransactionsLayer {
 
   boolean contains(Transaction transaction);
 
-  TransactionAddedResult add(PendingTransaction pendingTransaction, int gap);
+  /**
+   * Try to add a pending transaction to this layer. The {@code addReason} is used to discriminate
+   * between a new tx that is added to the pool, or a tx that is already in the pool, but is moving
+   * internally between layers, for example, due to a promotion or demotion. The distinction is
+   * needed since we only need to send a notification for a new tx, and not when it is only an
+   * internal move.
+   *
+   * @param pendingTransaction the tx to try to add to this layer
+   * @param gap the nonce gap between the current sender nonce and the tx
+   * @param addReason define if it is a new tx or an internal move
+   * @return the result of the add operation
+   */
+  TransactionAddedResult add(PendingTransaction pendingTransaction, int gap, AddReason addReason);
 
   void remove(PendingTransaction pendingTransaction, RemovalReason reason);
 
@@ -108,6 +120,29 @@ List<PendingTransaction> promote(
 
   String logSender(Address sender);
 
+  /** Describe why we are trying to add a tx to a layer. */
+  enum AddReason {
+    /** When adding a tx, that is not present in the pool. */
+    NEW(true),
+    /** When adding a tx as result of an internal move between layers. */
+    MOVE(false);
+
+    private final boolean sendNotification;
+
+    AddReason(final boolean sendNotification) {
+      this.sendNotification = sendNotification;
+    }
+
+    /**
+     * Should we send add notification for this reason?
+     *
+     * @return true if notification should be sent
+     */
+    public boolean sendNotification() {
+      return sendNotification;
+    }
+  }
+
   enum RemovalReason {
     CONFIRMED,
     CROSS_LAYER_REPLACED,

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

@@ -17,6 +17,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 static org.hyperledger.besu.ethereum.eth.transactions.layered.TransactionsLayer.AddReason.NEW;
 
 import org.hyperledger.besu.datatypes.TransactionType;
 import org.hyperledger.besu.datatypes.Wei;
@@ -169,7 +170,7 @@ protected void shouldPrioritizeValueThenTimeAddedToPool(
             .mapToObj(
                 i -> {
                   final var lowPriceTx = lowValueTxSupplier.next();
-                  final var prioritizeResult = transactions.add(lowPriceTx, 0);
+                  final var prioritizeResult = transactions.add(lowPriceTx, 0, NEW);
 
                   assertThat(prioritizeResult).isEqualTo(ADDED);
                   assertThat(evictCollector.getEvictedTransactions()).isEmpty();
@@ -180,7 +181,7 @@ protected void shouldPrioritizeValueThenTimeAddedToPool(
     assertThat(transactions.count()).isEqualTo(MAX_TRANSACTIONS);
 
     // This should kick the oldest tx with the low gas price out, namely the first one we added
-    final var highValuePrioRes = transactions.add(highValueTx, 0);
+    final var highValuePrioRes = transactions.add(highValueTx, 0, NEW);
     assertThat(highValuePrioRes).isEqualTo(ADDED);
     assertEvicted(expectedDroppedTx);
 
@@ -195,7 +196,7 @@ protected TransactionAddedResult prioritizeTransaction(final Transaction tx) {
   }
 
   protected TransactionAddedResult prioritizeTransaction(final PendingTransaction tx) {
-    return transactions.add(tx, 0);
+    return transactions.add(tx, 0, NEW);
   }
 
   protected void assertTransactionPrioritized(final PendingTransaction tx) {

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

@@ -35,8 +35,9 @@ public String name() {
   }
 
   @Override
-  public TransactionAddedResult add(final PendingTransaction pendingTransaction, final int gap) {
-    final var res = super.add(pendingTransaction, gap);
+  public TransactionAddedResult add(
+      final PendingTransaction pendingTransaction, final int gap, final AddReason addReason) {
+    final var res = super.add(pendingTransaction, gap, addReason);
     evictedTxs.add(pendingTransaction);
     return res;
   }