Merge branch 'refs/heads/7311-add-peertask-foundation-code' into 7311-add-GetReceiptsFromPeerTask

pull/7638/head
Matilda Clerke 1 month ago
commit 17edf6759b
  1. 1
      CHANGELOG.md
  2. 2
      besu/src/main/java/org/hyperledger/besu/services/BesuEventsImpl.java
  3. 3
      ethereum/api/src/main/java/org/hyperledger/besu/ethereum/api/jsonrpc/websocket/subscription/pending/PendingTransactionDroppedSubscriptionService.java
  4. 19
      ethereum/api/src/test/java/org/hyperledger/besu/ethereum/api/jsonrpc/internal/methods/EthCallTest.java
  5. 15
      ethereum/api/src/test/java/org/hyperledger/besu/ethereum/api/jsonrpc/websocket/subscription/pending/PendingTransactionDroppedSubscriptionServiceTest.java
  6. 25
      ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/transactions/PeerTransactionTracker.java
  7. 2
      ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/transactions/PendingTransactionDroppedListener.java
  8. 32
      ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/transactions/RemovalReason.java
  9. 11
      ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/transactions/TransactionPool.java
  10. 2
      ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/transactions/layered/AbstractPrioritizedTransactions.java
  11. 6
      ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/transactions/layered/AbstractSequentialTransactionsLayer.java
  12. 31
      ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/transactions/layered/AbstractTransactionsLayer.java
  13. 2
      ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/transactions/layered/BaseFeePrioritizedTransactions.java
  14. 11
      ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/transactions/layered/EndLayer.java
  15. 4
      ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/transactions/layered/LayeredPendingTransactions.java
  16. 76
      ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/transactions/layered/LayeredRemovalReason.java
  17. 4
      ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/transactions/layered/ReadyTransactions.java
  18. 8
      ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/transactions/layered/SparseTransactions.java
  19. 2
      ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/transactions/layered/TransactionsLayer.java
  20. 24
      ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/transactions/sorter/AbstractPendingTransactionsSorter.java
  21. 45
      ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/transactions/sorter/SequencedRemovalReason.java
  22. 2
      ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/manager/peertask/PeerTaskExecutorTest.java
  23. 37
      ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/transactions/PeerTransactionTrackerTest.java
  24. 6
      ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/transactions/layered/LayeredPendingTransactionsTest.java
  25. 5
      ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/transactions/layered/LayersTest.java
  26. 2
      ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/transactions/layered/ReplayTest.java
  27. 26
      ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/transactions/sorter/AbstractPendingTransactionsTestBase.java

@ -8,6 +8,7 @@
### Upcoming Breaking Changes
### Additions and Improvements
- Fine tune already seen txs tracker when a tx is removed from the pool [#7755](https://github.com/hyperledger/besu/pull/7755)
### Bug fixes

@ -133,7 +133,7 @@ public class BesuEventsImpl implements BesuEvents {
public long addTransactionDroppedListener(
final TransactionDroppedListener transactionDroppedListener) {
return transactionPool.subscribeDroppedTransactions(
transactionDroppedListener::onTransactionDropped);
(transaction, reason) -> transactionDroppedListener.onTransactionDropped(transaction));
}
@Override

@ -20,6 +20,7 @@ import org.hyperledger.besu.ethereum.api.jsonrpc.websocket.subscription.Subscrip
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.PendingTransactionDroppedListener;
import org.hyperledger.besu.ethereum.eth.transactions.RemovalReason;
import java.util.List;
@ -34,7 +35,7 @@ public class PendingTransactionDroppedSubscriptionService
}
@Override
public void onTransactionDropped(final Transaction transaction) {
public void onTransactionDropped(final Transaction transaction, final RemovalReason reason) {
notifySubscribers(transaction.getHash());
}

@ -157,12 +157,12 @@ public class EthCallTest {
when(result.isSuccessful()).thenReturn(true);
when(result.getValidationResult()).thenReturn(ValidationResult.valid());
when(result.getOutput()).thenReturn(Bytes.of(1));
verify(transactionSimulator).process(any(), any(), any(), mapperCaptor.capture(), any());
verify(transactionSimulator)
.process(eq(callParameter()), any(), any(), mapperCaptor.capture(), any());
assertThat(mapperCaptor.getValue().apply(mock(MutableWorldState.class), Optional.of(result)))
.isEqualTo(Optional.of(expectedResponse));
assertThat(response).usingRecursiveComparison().isEqualTo(expectedResponse);
verify(transactionSimulator).process(eq(callParameter()), any(), any(), any(), any());
verify(blockchainQueries, atLeastOnce()).getBlockchain();
verifyNoMoreInteractions(blockchainQueries);
}
@ -174,10 +174,9 @@ public class EthCallTest {
// Expect a revert error with no decoded reason (error doesn't begin "Error(string)" so ignored)
final String abiHexString = "0x1234";
final JsonRpcError expectedError = new JsonRpcError(REVERT_ERROR, abiHexString);
final JsonRpcResponse expectedResponse = new JsonRpcErrorResponse(null, expectedError);
final JsonRpcErrorResponse expectedResponse = new JsonRpcErrorResponse(null, expectedError);
assertThat(((JsonRpcErrorResponse) expectedResponse).getError().getMessage())
.isEqualTo("Execution reverted");
assertThat(expectedResponse.getError().getMessage()).isEqualTo("Execution reverted");
mockTransactionProcessorSuccessResult(expectedResponse);
when(blockchainQueries.getBlockchain()).thenReturn(blockchain);
@ -214,9 +213,9 @@ public class EthCallTest {
// bytes are invalid ABI)
final String abiHexString = "0x08c379a002d36d";
final JsonRpcError expectedError = new JsonRpcError(REVERT_ERROR, abiHexString);
final JsonRpcResponse expectedResponse = new JsonRpcErrorResponse(null, expectedError);
final JsonRpcErrorResponse expectedResponse = new JsonRpcErrorResponse(null, expectedError);
assertThat(((JsonRpcErrorResponse) expectedResponse).getError().getMessage())
assertThat(expectedResponse.getError().getMessage())
.isEqualTo("Execution reverted: ABI decode error");
mockTransactionProcessorSuccessResult(expectedResponse);
@ -254,9 +253,9 @@ public class EthCallTest {
final String abiHexString =
"0x08c379a00000000000000000000000000000000000000000000000000000000000000020000000000000000000000000000000000000000000000000000000000000002545524332303a207472616e736665722066726f6d20746865207a65726f2061646472657373000000000000000000000000000000000000000000000000000000";
final JsonRpcError expectedError = new JsonRpcError(REVERT_ERROR, abiHexString);
final JsonRpcResponse expectedResponse = new JsonRpcErrorResponse(null, expectedError);
final JsonRpcErrorResponse expectedResponse = new JsonRpcErrorResponse(null, expectedError);
assertThat(((JsonRpcErrorResponse) expectedResponse).getError().getMessage())
assertThat(expectedResponse.getError().getMessage())
.isEqualTo("Execution reverted: ERC20: transfer from the zero address");
mockTransactionProcessorSuccessResult(expectedResponse);
@ -279,8 +278,6 @@ public class EthCallTest {
when(result.result()).thenReturn(processingResult);
verify(transactionSimulator).process(any(), any(), any(), mapperCaptor.capture(), any());
System.out.println(result);
System.out.println(expectedResponse);
assertThat(mapperCaptor.getValue().apply(mock(MutableWorldState.class), Optional.of(result)))
.isEqualTo(Optional.of(expectedResponse));

@ -30,6 +30,7 @@ import org.hyperledger.besu.ethereum.api.jsonrpc.websocket.subscription.request.
import org.hyperledger.besu.ethereum.chain.Blockchain;
import org.hyperledger.besu.ethereum.core.Block;
import org.hyperledger.besu.ethereum.core.Transaction;
import org.hyperledger.besu.ethereum.eth.transactions.RemovalReason;
import java.util.Arrays;
import java.util.HashMap;
@ -47,6 +48,18 @@ public class PendingTransactionDroppedSubscriptionServiceTest {
private static final Hash TX_ONE =
Hash.fromHexString("0x15876958423545c3c7b0fcf9be8ffb543305ee1b43db87ed380dcf0cd16589f7");
private static final RemovalReason DUMMY_REMOVAL_REASON =
new RemovalReason() {
@Override
public String label() {
return "";
}
@Override
public boolean stopTracking() {
return false;
}
};
@Mock private SubscriptionManager subscriptionManager;
@Mock private Blockchain blockchain;
@ -65,7 +78,7 @@ public class PendingTransactionDroppedSubscriptionServiceTest {
setUpSubscriptions(subscriptionIds);
final Transaction pending = transaction(TX_ONE);
service.onTransactionDropped(pending);
service.onTransactionDropped(pending, DUMMY_REMOVAL_REASON);
verifyNoInteractions(block);
verifyNoInteractions(blockchain);

@ -34,7 +34,8 @@ import java.util.stream.Collectors;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class PeerTransactionTracker implements EthPeer.DisconnectCallback {
public class PeerTransactionTracker
implements EthPeer.DisconnectCallback, PendingTransactionDroppedListener {
private static final Logger LOG = LoggerFactory.getLogger(PeerTransactionTracker.class);
private static final int MAX_TRACKED_SEEN_TRANSACTIONS = 100_000;
@ -122,13 +123,14 @@ public class PeerTransactionTracker implements EthPeer.DisconnectCallback {
}
private <T> Set<T> createTransactionsSet() {
return Collections.newSetFromMap(
new LinkedHashMap<>(1 << 4, 0.75f, true) {
@Override
protected boolean removeEldestEntry(final Map.Entry<T, Boolean> eldest) {
return size() > MAX_TRACKED_SEEN_TRANSACTIONS;
}
});
return Collections.synchronizedSet(
Collections.newSetFromMap(
new LinkedHashMap<>(16, 0.75f, true) {
@Override
protected boolean removeEldestEntry(final Map.Entry<T, Boolean> eldest) {
return size() > MAX_TRACKED_SEEN_TRANSACTIONS;
}
}));
}
@Override
@ -175,4 +177,11 @@ public class PeerTransactionTracker implements EthPeer.DisconnectCallback {
private String logPeerSet(final Set<EthPeer> peers) {
return peers.stream().map(EthPeer::getLoggableId).collect(Collectors.joining(","));
}
@Override
public void onTransactionDropped(final Transaction transaction, final RemovalReason reason) {
if (reason.stopTracking()) {
seenTransactions.values().stream().forEach(st -> st.remove(transaction.getHash()));
}
}
}

@ -19,5 +19,5 @@ import org.hyperledger.besu.ethereum.core.Transaction;
@FunctionalInterface
public interface PendingTransactionDroppedListener {
void onTransactionDropped(Transaction transaction);
void onTransactionDropped(Transaction transaction, final RemovalReason reason);
}

@ -0,0 +1,32 @@
/*
* Copyright contributors to Hyperledger Besu.
*
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
* an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
* specific language governing permissions and limitations under the License.
*
* SPDX-License-Identifier: Apache-2.0
*/
package org.hyperledger.besu.ethereum.eth.transactions;
/** The reason why a pending tx has been removed */
public interface RemovalReason {
/**
* Return a label that identify this reason to be used in the metric system.
*
* @return a label
*/
String label();
/**
* Return true if we should stop tracking the tx as already seen
*
* @return true if no more tracking is needed
*/
boolean stopTracking();
}

@ -137,7 +137,8 @@ public class TransactionPool implements BlockAddedObserver {
initializeBlobMetrics();
initLogForReplay();
subscribePendingTransactions(this::mapBlobsOnTransactionAdded);
subscribeDroppedTransactions(this::unmapBlobsOnTransactionDropped);
subscribeDroppedTransactions(
(transaction, reason) -> unmapBlobsOnTransactionDropped(transaction));
}
private void initLogForReplay() {
@ -720,7 +721,9 @@ public class TransactionPool implements BlockAddedObserver {
void subscribe() {
onAddedListenerId = pendingTransactions.subscribePendingTransactions(this::onAdded);
onDroppedListenerId = pendingTransactions.subscribeDroppedTransactions(this::onDropped);
onDroppedListenerId =
pendingTransactions.subscribeDroppedTransactions(
(transaction, reason) -> onDropped(transaction, reason));
}
void unsubscribe() {
@ -728,8 +731,8 @@ public class TransactionPool implements BlockAddedObserver {
pendingTransactions.unsubscribeDroppedTransactions(onDroppedListenerId);
}
private void onDropped(final Transaction transaction) {
onDroppedListeners.forEach(listener -> listener.onTransactionDropped(transaction));
private void onDropped(final Transaction transaction, final RemovalReason reason) {
onDroppedListeners.forEach(listener -> listener.onTransactionDropped(transaction, reason));
}
private void onAdded(final Transaction transaction) {

@ -137,7 +137,7 @@ public abstract class AbstractPrioritizedTransactions extends AbstractSequential
protected void internalRemove(
final NavigableMap<Long, PendingTransaction> senderTxs,
final PendingTransaction removedTx,
final RemovalReason removalReason) {
final LayeredRemovalReason removalReason) {
orderByFee.remove(removedTx);
}

@ -15,8 +15,8 @@
package org.hyperledger.besu.ethereum.eth.transactions.layered;
import static org.hyperledger.besu.ethereum.eth.transactions.layered.AddReason.MOVE;
import static org.hyperledger.besu.ethereum.eth.transactions.layered.RemovalReason.LayerMoveReason.EVICTED;
import static org.hyperledger.besu.ethereum.eth.transactions.layered.RemovalReason.LayerMoveReason.FOLLOW_INVALIDATED;
import static org.hyperledger.besu.ethereum.eth.transactions.layered.LayeredRemovalReason.LayerMoveReason.EVICTED;
import static org.hyperledger.besu.ethereum.eth.transactions.layered.LayeredRemovalReason.LayerMoveReason.FOLLOW_INVALIDATED;
import org.hyperledger.besu.datatypes.Address;
import org.hyperledger.besu.ethereum.eth.manager.EthScheduler;
@ -24,7 +24,7 @@ import org.hyperledger.besu.ethereum.eth.transactions.BlobCache;
import org.hyperledger.besu.ethereum.eth.transactions.PendingTransaction;
import org.hyperledger.besu.ethereum.eth.transactions.TransactionPoolConfiguration;
import org.hyperledger.besu.ethereum.eth.transactions.TransactionPoolMetrics;
import org.hyperledger.besu.ethereum.eth.transactions.layered.RemovalReason.PoolRemovalReason;
import org.hyperledger.besu.ethereum.eth.transactions.layered.LayeredRemovalReason.PoolRemovalReason;
import java.util.Map;
import java.util.NavigableMap;

@ -20,12 +20,12 @@ import static org.hyperledger.besu.ethereum.eth.transactions.TransactionAddedRes
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.AddReason.MOVE;
import static org.hyperledger.besu.ethereum.eth.transactions.layered.RemovalReason.LayerMoveReason.EVICTED;
import static org.hyperledger.besu.ethereum.eth.transactions.layered.RemovalReason.PoolRemovalReason.BELOW_MIN_SCORE;
import static org.hyperledger.besu.ethereum.eth.transactions.layered.RemovalReason.PoolRemovalReason.CONFIRMED;
import static org.hyperledger.besu.ethereum.eth.transactions.layered.RemovalReason.PoolRemovalReason.CROSS_LAYER_REPLACED;
import static org.hyperledger.besu.ethereum.eth.transactions.layered.RemovalReason.PoolRemovalReason.REPLACED;
import static org.hyperledger.besu.ethereum.eth.transactions.layered.RemovalReason.RemovedFrom.POOL;
import static org.hyperledger.besu.ethereum.eth.transactions.layered.LayeredRemovalReason.LayerMoveReason.EVICTED;
import static org.hyperledger.besu.ethereum.eth.transactions.layered.LayeredRemovalReason.PoolRemovalReason.BELOW_MIN_SCORE;
import static org.hyperledger.besu.ethereum.eth.transactions.layered.LayeredRemovalReason.PoolRemovalReason.CONFIRMED;
import static org.hyperledger.besu.ethereum.eth.transactions.layered.LayeredRemovalReason.PoolRemovalReason.CROSS_LAYER_REPLACED;
import static org.hyperledger.besu.ethereum.eth.transactions.layered.LayeredRemovalReason.PoolRemovalReason.REPLACED;
import static org.hyperledger.besu.ethereum.eth.transactions.layered.LayeredRemovalReason.RemovedFrom.POOL;
import org.hyperledger.besu.datatypes.Address;
import org.hyperledger.besu.datatypes.Hash;
@ -295,7 +295,9 @@ public abstract class AbstractTransactionsLayer implements TransactionsLayer {
if (remainingPromotionsPerType[txType.ordinal()] > 0) {
senderTxs.pollFirstEntry();
processRemove(
senderTxs, candidateTx.getTransaction(), RemovalReason.LayerMoveReason.PROMOTED);
senderTxs,
candidateTx.getTransaction(),
LayeredRemovalReason.LayerMoveReason.PROMOTED);
metrics.incrementRemoved(candidateTx, "promoted", name());
if (senderTxs.isEmpty()) {
@ -386,7 +388,7 @@ public abstract class AbstractTransactionsLayer implements TransactionsLayer {
decreaseCounters(replacedTx);
metrics.incrementRemoved(replacedTx, REPLACED.label(), name());
internalReplaced(replacedTx);
notifyTransactionDropped(replacedTx);
notifyTransactionDropped(replacedTx, REPLACED);
}
protected abstract void internalReplaced(final PendingTransaction replacedTx);
@ -415,7 +417,7 @@ public abstract class AbstractTransactionsLayer implements TransactionsLayer {
protected PendingTransaction processRemove(
final NavigableMap<Long, PendingTransaction> senderTxs,
final Transaction transaction,
final RemovalReason removalReason) {
final LayeredRemovalReason removalReason) {
final PendingTransaction removedTx = pendingTransactions.remove(transaction.getHash());
if (removedTx != null) {
@ -423,7 +425,7 @@ public abstract class AbstractTransactionsLayer implements TransactionsLayer {
metrics.incrementRemoved(removedTx, removalReason.label(), name());
internalRemove(senderTxs, removedTx, removalReason);
if (removalReason.removedFrom().equals(POOL)) {
notifyTransactionDropped(removedTx);
notifyTransactionDropped(removedTx, removalReason);
}
}
return removedTx;
@ -432,7 +434,7 @@ public abstract class AbstractTransactionsLayer implements TransactionsLayer {
protected PendingTransaction processEvict(
final NavigableMap<Long, PendingTransaction> senderTxs,
final PendingTransaction evictedTx,
final RemovalReason reason) {
final LayeredRemovalReason reason) {
final PendingTransaction removedTx = pendingTransactions.remove(evictedTx.getHash());
if (removedTx != null) {
decreaseCounters(evictedTx);
@ -545,7 +547,7 @@ public abstract class AbstractTransactionsLayer implements TransactionsLayer {
protected abstract void internalRemove(
final NavigableMap<Long, PendingTransaction> senderTxs,
final PendingTransaction pendingTransaction,
final RemovalReason removalReason);
final LayeredRemovalReason removalReason);
protected abstract PendingTransaction getEvictable();
@ -606,9 +608,10 @@ public abstract class AbstractTransactionsLayer implements TransactionsLayer {
listener -> listener.onTransactionAdded(pendingTransaction.getTransaction()));
}
protected void notifyTransactionDropped(final PendingTransaction pendingTransaction) {
protected void notifyTransactionDropped(
final PendingTransaction pendingTransaction, final LayeredRemovalReason reason) {
onDroppedListeners.forEach(
listener -> listener.onTransactionDropped(pendingTransaction.getTransaction()));
listener -> listener.onTransactionDropped(pendingTransaction.getTransaction(), reason));
}
@Override

@ -15,7 +15,7 @@
package org.hyperledger.besu.ethereum.eth.transactions.layered;
import static org.hyperledger.besu.ethereum.eth.transactions.layered.AddReason.MOVE;
import static org.hyperledger.besu.ethereum.eth.transactions.layered.RemovalReason.LayerMoveReason.DEMOTED;
import static org.hyperledger.besu.ethereum.eth.transactions.layered.LayeredRemovalReason.LayerMoveReason.DEMOTED;
import org.hyperledger.besu.datatypes.Wei;
import org.hyperledger.besu.ethereum.core.BlockHeader;

@ -14,7 +14,7 @@
*/
package org.hyperledger.besu.ethereum.eth.transactions.layered;
import static org.hyperledger.besu.ethereum.eth.transactions.layered.RemovalReason.PoolRemovalReason.DROPPED;
import static org.hyperledger.besu.ethereum.eth.transactions.layered.LayeredRemovalReason.PoolRemovalReason.DROPPED;
import org.hyperledger.besu.datatypes.Address;
import org.hyperledger.besu.datatypes.Hash;
@ -25,7 +25,7 @@ import org.hyperledger.besu.ethereum.eth.transactions.PendingTransactionAddedLis
import org.hyperledger.besu.ethereum.eth.transactions.PendingTransactionDroppedListener;
import org.hyperledger.besu.ethereum.eth.transactions.TransactionAddedResult;
import org.hyperledger.besu.ethereum.eth.transactions.TransactionPoolMetrics;
import org.hyperledger.besu.ethereum.eth.transactions.layered.RemovalReason.PoolRemovalReason;
import org.hyperledger.besu.ethereum.eth.transactions.layered.LayeredRemovalReason.PoolRemovalReason;
import org.hyperledger.besu.ethereum.mainnet.feemarket.FeeMarket;
import org.hyperledger.besu.util.Subscribers;
@ -78,7 +78,7 @@ public class EndLayer implements TransactionsLayer {
@Override
public TransactionAddedResult add(
final PendingTransaction pendingTransaction, final int gap, final AddReason reason) {
notifyTransactionDropped(pendingTransaction);
notifyTransactionDropped(pendingTransaction, DROPPED);
metrics.incrementRemoved(pendingTransaction, DROPPED.label(), name());
++droppedCount;
return TransactionAddedResult.DROPPED;
@ -152,9 +152,10 @@ public class EndLayer implements TransactionsLayer {
onDroppedListeners.unsubscribe(id);
}
protected void notifyTransactionDropped(final PendingTransaction pendingTransaction) {
protected void notifyTransactionDropped(
final PendingTransaction pendingTransaction, final LayeredRemovalReason reason) {
onDroppedListeners.forEach(
listener -> listener.onTransactionDropped(pendingTransaction.getTransaction()));
listener -> listener.onTransactionDropped(pendingTransaction.getTransaction(), reason));
}
@Override

@ -21,8 +21,8 @@ import static org.hyperledger.besu.ethereum.eth.transactions.TransactionAddedRes
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.AddReason.NEW;
import static org.hyperledger.besu.ethereum.eth.transactions.layered.RemovalReason.PoolRemovalReason.INVALIDATED;
import static org.hyperledger.besu.ethereum.eth.transactions.layered.RemovalReason.PoolRemovalReason.RECONCILED;
import static org.hyperledger.besu.ethereum.eth.transactions.layered.LayeredRemovalReason.PoolRemovalReason.INVALIDATED;
import static org.hyperledger.besu.ethereum.eth.transactions.layered.LayeredRemovalReason.PoolRemovalReason.RECONCILED;
import org.hyperledger.besu.datatypes.Address;
import org.hyperledger.besu.datatypes.Hash;

@ -14,10 +14,12 @@
*/
package org.hyperledger.besu.ethereum.eth.transactions.layered;
import org.hyperledger.besu.ethereum.eth.transactions.RemovalReason;
import java.util.Locale;
/** The reason why a pending tx has been removed */
public interface RemovalReason {
interface LayeredRemovalReason extends RemovalReason {
/**
* From where the tx has been removed
*
@ -25,13 +27,6 @@ public interface RemovalReason {
*/
RemovedFrom removedFrom();
/**
* Return a label that identify this reason to be used in the metric system.
*
* @return a label
*/
String label();
/** There are 2 kinds of removals, from a layer and from the pool. */
enum RemovedFrom {
/**
@ -50,37 +45,53 @@ public interface RemovalReason {
}
/** The reason why the tx has been removed from the pool */
enum PoolRemovalReason implements RemovalReason {
/** Tx removed since it is confirmed on chain, as part of an imported block. */
CONFIRMED,
/** Tx removed since it has been replaced by another one added in the same layer. */
REPLACED,
/** Tx removed since it has been replaced by another one added in another layer. */
CROSS_LAYER_REPLACED,
/** Tx removed when the pool is full, to make space for new incoming txs. */
DROPPED,
enum PoolRemovalReason implements LayeredRemovalReason {
/**
* Tx removed since it is confirmed on chain, as part of an imported block. Keep tracking since
* makes no sense to reprocess a confirmed.
*/
CONFIRMED(false),
/**
* Tx removed since it has been replaced by another one added in the same layer. Keep tracking
* since makes no sense to reprocess a replaced tx.
*/
REPLACED(false),
/**
* Tx removed since it has been replaced by another one added in another layer. Keep tracking
* since makes no sense to reprocess a replaced tx.
*/
CROSS_LAYER_REPLACED(false),
/**
* Tx removed when the pool is full, to make space for new incoming txs. Stop tracking it so we
* could re-accept it in the future.
*/
DROPPED(true),
/**
* Tx removed since found invalid after it was added to the pool, for example during txs
* selection for a new block proposal.
* selection for a new block proposal. Keep tracking since we do not want to reprocess an
* invalid tx.
*/
INVALIDATED,
INVALIDATED(false),
/**
* Special case, when for a sender, discrepancies are found between the world state view and the
* pool view, then all the txs for this sender are removed and added again. Discrepancies, are
* rare, and can happen during a short windows when a new block is being imported and the world
* state being updated.
* state being updated. Keep tracking since it is removed and re-added.
*/
RECONCILED,
RECONCILED(false),
/**
* When a pending tx is penalized its score is decreased, if at some point its score is lower
* than the configured minimum then the pending tx is removed from the pool.
* than the configured minimum then the pending tx is removed from the pool. Stop tracking it so
* we could re-accept it in the future.
*/
BELOW_MIN_SCORE;
BELOW_MIN_SCORE(true);
private final String label;
private final boolean stopTracking;
PoolRemovalReason() {
PoolRemovalReason(final boolean stopTracking) {
this.label = name().toLowerCase(Locale.ROOT);
this.stopTracking = stopTracking;
}
@Override
@ -92,10 +103,15 @@ public interface RemovalReason {
public String label() {
return label;
}
@Override
public boolean stopTracking() {
return stopTracking;
}
}
/** The reason why the tx has been moved across layers */
enum LayerMoveReason implements RemovalReason {
enum LayerMoveReason implements LayeredRemovalReason {
/**
* When the current layer is full, and this tx needs to be moved to the lower layer, in order to
* free space.
@ -132,5 +148,15 @@ public interface RemovalReason {
public String label() {
return label;
}
/**
* We need to continue to track a tx when is moved between layers
*
* @return always false
*/
@Override
public boolean stopTracking() {
return false;
}
}
}

@ -14,7 +14,7 @@
*/
package org.hyperledger.besu.ethereum.eth.transactions.layered;
import static org.hyperledger.besu.ethereum.eth.transactions.layered.RemovalReason.LayerMoveReason.PROMOTED;
import static org.hyperledger.besu.ethereum.eth.transactions.layered.LayeredRemovalReason.LayerMoveReason.PROMOTED;
import org.hyperledger.besu.datatypes.Address;
import org.hyperledger.besu.ethereum.core.BlockHeader;
@ -109,7 +109,7 @@ public class ReadyTransactions extends AbstractSequentialTransactionsLayer {
protected void internalRemove(
final NavigableMap<Long, PendingTransaction> senderTxs,
final PendingTransaction removedTx,
final RemovalReason removalReason) {
final LayeredRemovalReason removalReason) {
orderByMaxFee.remove(removedTx);
if (!senderTxs.isEmpty()) {
orderByMaxFee.add(senderTxs.firstEntry().getValue());

@ -14,8 +14,8 @@
*/
package org.hyperledger.besu.ethereum.eth.transactions.layered;
import static org.hyperledger.besu.ethereum.eth.transactions.layered.RemovalReason.LayerMoveReason.PROMOTED;
import static org.hyperledger.besu.ethereum.eth.transactions.layered.RemovalReason.PoolRemovalReason.INVALIDATED;
import static org.hyperledger.besu.ethereum.eth.transactions.layered.LayeredRemovalReason.LayerMoveReason.PROMOTED;
import static org.hyperledger.besu.ethereum.eth.transactions.layered.LayeredRemovalReason.PoolRemovalReason.INVALIDATED;
import org.hyperledger.besu.datatypes.Address;
import org.hyperledger.besu.ethereum.core.BlockHeader;
@ -26,7 +26,7 @@ import org.hyperledger.besu.ethereum.eth.transactions.PendingTransaction;
import org.hyperledger.besu.ethereum.eth.transactions.TransactionAddedResult;
import org.hyperledger.besu.ethereum.eth.transactions.TransactionPoolConfiguration;
import org.hyperledger.besu.ethereum.eth.transactions.TransactionPoolMetrics;
import org.hyperledger.besu.ethereum.eth.transactions.layered.RemovalReason.PoolRemovalReason;
import org.hyperledger.besu.ethereum.eth.transactions.layered.LayeredRemovalReason.PoolRemovalReason;
import org.hyperledger.besu.ethereum.mainnet.feemarket.FeeMarket;
import java.util.ArrayList;
@ -278,7 +278,7 @@ public class SparseTransactions extends AbstractTransactionsLayer {
protected void internalRemove(
final NavigableMap<Long, PendingTransaction> senderTxs,
final PendingTransaction removedTx,
final RemovalReason removalReason) {
final LayeredRemovalReason removalReason) {
sparseEvictionOrder.remove(removedTx);

@ -22,7 +22,7 @@ import org.hyperledger.besu.ethereum.eth.transactions.PendingTransaction;
import org.hyperledger.besu.ethereum.eth.transactions.PendingTransactionAddedListener;
import org.hyperledger.besu.ethereum.eth.transactions.PendingTransactionDroppedListener;
import org.hyperledger.besu.ethereum.eth.transactions.TransactionAddedResult;
import org.hyperledger.besu.ethereum.eth.transactions.layered.RemovalReason.PoolRemovalReason;
import org.hyperledger.besu.ethereum.eth.transactions.layered.LayeredRemovalReason.PoolRemovalReason;
import org.hyperledger.besu.ethereum.mainnet.feemarket.FeeMarket;
import java.util.List;

@ -18,6 +18,10 @@ import static org.hyperledger.besu.ethereum.eth.transactions.TransactionAddedRes
import static org.hyperledger.besu.ethereum.eth.transactions.TransactionAddedResult.ALREADY_KNOWN;
import static org.hyperledger.besu.ethereum.eth.transactions.TransactionAddedResult.NONCE_TOO_FAR_IN_FUTURE_FOR_SENDER;
import static org.hyperledger.besu.ethereum.eth.transactions.TransactionAddedResult.REJECTED_UNDERPRICED_REPLACEMENT;
import static org.hyperledger.besu.ethereum.eth.transactions.sorter.SequencedRemovalReason.EVICTED;
import static org.hyperledger.besu.ethereum.eth.transactions.sorter.SequencedRemovalReason.INVALID;
import static org.hyperledger.besu.ethereum.eth.transactions.sorter.SequencedRemovalReason.REPLACED;
import static org.hyperledger.besu.ethereum.eth.transactions.sorter.SequencedRemovalReason.TIMED_EVICTION;
import org.hyperledger.besu.datatypes.Address;
import org.hyperledger.besu.datatypes.Hash;
@ -160,7 +164,7 @@ public abstract class AbstractPendingTransactionsSorter implements PendingTransa
.setMessage("Evicted {} due to age")
.addArgument(transactionInfo::toTraceLog)
.log();
removeTransaction(transactionInfo.getTransaction());
removeTransaction(transactionInfo.getTransaction(), TIMED_EVICTION);
});
}
@ -196,9 +200,9 @@ public abstract class AbstractPendingTransactionsSorter implements PendingTransa
return transactionAddedStatus;
}
void removeTransaction(final Transaction transaction) {
void removeTransaction(final Transaction transaction, final SequencedRemovalReason reason) {
removeTransaction(transaction, false);
notifyTransactionDropped(transaction);
notifyTransactionDropped(transaction, reason);
}
@Override
@ -256,12 +260,12 @@ public abstract class AbstractPendingTransactionsSorter implements PendingTransa
}
if (result.stop()) {
transactionsToRemove.forEach(this::removeTransaction);
transactionsToRemove.forEach(tx -> removeTransaction(tx, INVALID));
return;
}
}
}
transactionsToRemove.forEach(this::removeTransaction);
transactionsToRemove.forEach(tx -> removeTransaction(tx, INVALID));
}
}
@ -324,7 +328,7 @@ public abstract class AbstractPendingTransactionsSorter implements PendingTransa
.setMessage("Tracked transaction by sender {}")
.addArgument(pendingTxsForSender::toTraceLog)
.log();
maybeReplacedTransaction.ifPresent(this::removeTransaction);
maybeReplacedTransaction.ifPresent(tx -> removeTransaction(tx, REPLACED));
return ADDED;
}
@ -354,8 +358,10 @@ public abstract class AbstractPendingTransactionsSorter implements PendingTransa
pendingTransactionSubscribers.forEach(listener -> listener.onTransactionAdded(transaction));
}
private void notifyTransactionDropped(final Transaction transaction) {
transactionDroppedListeners.forEach(listener -> listener.onTransactionDropped(transaction));
private void notifyTransactionDropped(
final Transaction transaction, final SequencedRemovalReason reason) {
transactionDroppedListeners.forEach(
listener -> listener.onTransactionDropped(transaction, reason));
}
@Override
@ -491,7 +497,7 @@ public abstract class AbstractPendingTransactionsSorter implements PendingTransa
// remove backward to avoid gaps
for (int i = txsToEvict.size() - 1; i >= 0; i--) {
removeTransaction(txsToEvict.get(i).getTransaction());
removeTransaction(txsToEvict.get(i).getTransaction(), EVICTED);
}
}

@ -0,0 +1,45 @@
/*
* Copyright contributors to Hyperledger Besu.
*
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
* an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
* specific language governing permissions and limitations under the License.
*
* SPDX-License-Identifier: Apache-2.0
*/
package org.hyperledger.besu.ethereum.eth.transactions.sorter;
import org.hyperledger.besu.ethereum.eth.transactions.RemovalReason;
import java.util.Locale;
/** The reason why a pending tx has been removed */
enum SequencedRemovalReason implements RemovalReason {
EVICTED(true),
TIMED_EVICTION(true),
REPLACED(false),
INVALID(false);
private final String label;
private final boolean stopTracking;
SequencedRemovalReason(final boolean stopTracking) {
this.label = name().toLowerCase(Locale.ROOT);
this.stopTracking = stopTracking;
}
@Override
public String label() {
return label;
}
@Override
public boolean stopTracking() {
return stopTracking;
}
}

@ -107,7 +107,7 @@ public class PeerTaskExecutorTest {
PeerTaskExecutorResult<Object> result = peerTaskExecutor.executeAgainstPeer(peerTask, ethPeer);
Assertions.assertNotNull(result);
Assertions.assertTrue(result.result().isEmpty());
Assertions.assertTrue(result.result().isPresent());
Assertions.assertEquals(PeerTaskExecutorResponseCode.INVALID_RESPONSE, result.responseCode());
}

@ -65,6 +65,28 @@ public class PeerTransactionTrackerTest {
assertThat(tracker.claimTransactionsToSendToPeer(ethPeer2)).containsOnly(transaction3);
}
@Test
public void shouldStopTrackingSeenTransactionsWhenRemovalReasonSaysSo() {
tracker.markTransactionsAsSeen(ethPeer1, ImmutableSet.of(transaction2));
assertThat(tracker.hasSeenTransaction(transaction2.getHash())).isTrue();
tracker.onTransactionDropped(transaction2, createRemovalReason(true));
assertThat(tracker.hasSeenTransaction(transaction2.getHash())).isFalse();
}
@Test
public void shouldKeepTrackingSeenTransactionsWhenRemovalReasonSaysSo() {
tracker.markTransactionsAsSeen(ethPeer1, ImmutableSet.of(transaction2));
assertThat(tracker.hasSeenTransaction(transaction2.getHash())).isTrue();
tracker.onTransactionDropped(transaction2, createRemovalReason(false));
assertThat(tracker.hasSeenTransaction(transaction2.getHash())).isTrue();
}
@Test
public void shouldExcludeAlreadySeenTransactionsAsACollectionFromTransactionsToSend() {
tracker.markTransactionsAsSeen(ethPeer1, ImmutableSet.of(transaction1, transaction2));
@ -125,4 +147,19 @@ public class PeerTransactionTrackerTest {
assertThat(tracker.hasPeerSeenTransaction(ethPeer1, transaction1)).isFalse();
assertThat(tracker.hasPeerSeenTransaction(ethPeer2, transaction2)).isFalse();
}
private RemovalReason createRemovalReason(final boolean stopTracking) {
return new RemovalReason() {
@Override
public String label() {
return "";
}
@Override
public boolean stopTracking() {
return stopTracking;
}
};
}
}

@ -22,8 +22,8 @@ import static org.hyperledger.besu.ethereum.eth.transactions.TransactionAddedRes
import static org.hyperledger.besu.ethereum.eth.transactions.TransactionAddedResult.REJECTED_UNDERPRICED_REPLACEMENT;
import static org.hyperledger.besu.ethereum.eth.transactions.layered.AddReason.MOVE;
import static org.hyperledger.besu.ethereum.eth.transactions.layered.AddReason.NEW;
import static org.hyperledger.besu.ethereum.eth.transactions.layered.RemovalReason.PoolRemovalReason.DROPPED;
import static org.hyperledger.besu.ethereum.eth.transactions.layered.RemovalReason.PoolRemovalReason.REPLACED;
import static org.hyperledger.besu.ethereum.eth.transactions.layered.LayeredRemovalReason.PoolRemovalReason.DROPPED;
import static org.hyperledger.besu.ethereum.eth.transactions.layered.LayeredRemovalReason.PoolRemovalReason.REPLACED;
import static org.hyperledger.besu.ethereum.transaction.TransactionInvalidReason.GAS_PRICE_BELOW_CURRENT_BASE_FEE;
import static org.hyperledger.besu.ethereum.transaction.TransactionInvalidReason.UPFRONT_COST_EXCEEDS_BALANCE;
import static org.hyperledger.besu.plugin.data.TransactionSelectionResult.BLOB_PRICE_BELOW_CURRENT_MIN;
@ -281,7 +281,7 @@ public class LayeredPendingTransactionsTest extends BaseTransactionPoolTest {
assertThat(smallLayers.evictedCollector.getEvictedTransactions())
.map(PendingTransaction::getTransaction)
.contains(firstTxs.get(0));
verify(droppedListener).onTransactionDropped(firstTxs.get(0));
verify(droppedListener).onTransactionDropped(firstTxs.get(0), DROPPED);
}
@Test

@ -21,13 +21,13 @@ import static org.hyperledger.besu.datatypes.TransactionType.ACCESS_LIST;
import static org.hyperledger.besu.datatypes.TransactionType.BLOB;
import static org.hyperledger.besu.datatypes.TransactionType.EIP1559;
import static org.hyperledger.besu.datatypes.TransactionType.FRONTIER;
import static org.hyperledger.besu.ethereum.eth.transactions.layered.LayeredRemovalReason.PoolRemovalReason.INVALIDATED;
import static org.hyperledger.besu.ethereum.eth.transactions.layered.LayersTest.Sender.S1;
import static org.hyperledger.besu.ethereum.eth.transactions.layered.LayersTest.Sender.S2;
import static org.hyperledger.besu.ethereum.eth.transactions.layered.LayersTest.Sender.S3;
import static org.hyperledger.besu.ethereum.eth.transactions.layered.LayersTest.Sender.S4;
import static org.hyperledger.besu.ethereum.eth.transactions.layered.LayersTest.Sender.SP1;
import static org.hyperledger.besu.ethereum.eth.transactions.layered.LayersTest.Sender.SP2;
import static org.hyperledger.besu.ethereum.eth.transactions.layered.RemovalReason.PoolRemovalReason.INVALIDATED;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@ -1409,7 +1409,8 @@ public class LayersTest extends BaseTransactionPoolTest {
this.pending = new LayeredPendingTransactions(poolConfig, this.prio, ethScheduler);
this.pending.subscribePendingTransactions(notificationsChecker::collectAddNotification);
this.pending.subscribeDroppedTransactions(notificationsChecker::collectDropNotification);
this.pending.subscribeDroppedTransactions(
(tx, reason) -> notificationsChecker.collectDropNotification(tx));
}
@Override

@ -16,7 +16,7 @@ package org.hyperledger.besu.ethereum.eth.transactions.layered;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.fail;
import static org.hyperledger.besu.ethereum.eth.transactions.layered.RemovalReason.PoolRemovalReason.INVALIDATED;
import static org.hyperledger.besu.ethereum.eth.transactions.layered.LayeredRemovalReason.PoolRemovalReason.INVALIDATED;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;

@ -18,6 +18,10 @@ 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.ALREADY_KNOWN;
import static org.hyperledger.besu.ethereum.eth.transactions.TransactionAddedResult.REJECTED_UNDERPRICED_REPLACEMENT;
import static org.hyperledger.besu.ethereum.eth.transactions.sorter.SequencedRemovalReason.EVICTED;
import static org.hyperledger.besu.ethereum.eth.transactions.sorter.SequencedRemovalReason.INVALID;
import static org.hyperledger.besu.ethereum.eth.transactions.sorter.SequencedRemovalReason.REPLACED;
import static org.hyperledger.besu.ethereum.eth.transactions.sorter.SequencedRemovalReason.TIMED_EVICTION;
import static org.hyperledger.besu.plugin.data.TransactionSelectionResult.SELECTED;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
@ -292,9 +296,9 @@ public abstract class AbstractPendingTransactionsTestBase {
transactions.subscribeDroppedTransactions(droppedListener);
transactions.removeTransaction(transaction1);
transactions.removeTransaction(transaction1, TIMED_EVICTION);
verify(droppedListener).onTransactionDropped(transaction1);
verify(droppedListener).onTransactionDropped(transaction1, TIMED_EVICTION);
}
@Test
@ -304,13 +308,13 @@ public abstract class AbstractPendingTransactionsTestBase {
final long id = transactions.subscribeDroppedTransactions(droppedListener);
transactions.removeTransaction(transaction1);
transactions.removeTransaction(transaction1, EVICTED);
verify(droppedListener).onTransactionDropped(transaction1);
verify(droppedListener).onTransactionDropped(transaction1, EVICTED);
transactions.unsubscribeDroppedTransactions(id);
transactions.removeTransaction(transaction2);
transactions.removeTransaction(transaction2, EVICTED);
verifyNoMoreInteractions(droppedListener);
}
@ -321,9 +325,9 @@ public abstract class AbstractPendingTransactionsTestBase {
transactions.subscribeDroppedTransactions(droppedListener);
transactions.removeTransaction(transaction1);
transactions.removeTransaction(transaction1, REPLACED);
verify(droppedListener).onTransactionDropped(transaction1);
verify(droppedListener).onTransactionDropped(transaction1, REPLACED);
}
@Test
@ -473,7 +477,7 @@ public abstract class AbstractPendingTransactionsTestBase {
public void shouldReturnEmptyOptionalAsMaximumNonceWhenLastTransactionForSenderRemoved() {
final Transaction transaction = transactionWithNonceAndSender(1, KEYS1);
transactions.addTransaction(createRemotePendingTransaction(transaction), Optional.empty());
transactions.removeTransaction(transaction);
transactions.removeTransaction(transaction, INVALID);
assertThat(transactions.getNextNonceForSender(SENDER1)).isEmpty();
}
@ -822,6 +826,8 @@ public abstract class AbstractPendingTransactionsTestBase {
.build(),
Optional.of(clock));
twoHourEvictionTransactionPool.subscribeDroppedTransactions(droppedListener);
twoHourEvictionTransactionPool.addTransaction(
createRemotePendingTransaction(transaction1, clock.millis()), Optional.empty());
assertThat(twoHourEvictionTransactionPool.size()).isEqualTo(1);
@ -832,6 +838,7 @@ public abstract class AbstractPendingTransactionsTestBase {
twoHourEvictionTransactionPool.evictOldTransactions();
assertThat(twoHourEvictionTransactionPool.size()).isEqualTo(1);
assertThat(metricsSystem.getCounterValue(REMOVED_COUNTER, REMOTE, DROPPED)).isEqualTo(1);
verify(droppedListener).onTransactionDropped(transaction1, TIMED_EVICTION);
}
@Test
@ -949,7 +956,8 @@ public abstract class AbstractPendingTransactionsTestBase {
public void shouldPrioritizeGasPriceThenTimeAddedToPool() {
// Make sure the 100 gas price TX isn't dropped
transactions.subscribeDroppedTransactions(
transaction -> assertThat(transaction.getGasPrice().get().toLong()).isLessThan(100));
(transaction, reason) ->
assertThat(transaction.getGasPrice().get().toLong()).isLessThan(100));
// Fill the pool with transactions from random senders
final List<Transaction> lowGasPriceTransactions =

Loading…
Cancel
Save