Merge branch 'main' into blobdb-for-static-data

acceptance-tests/tests/src/test/java/org/hyperledger/besu/tests/acceptance/jsonrpc/EthSendRawTransactionTest.java → acceptance-tests/tests/src/test/java/org/hyperledger/besu/tests/acceptance/jsonrpc/EthSendRawTransactionAcceptanceTest.java

@@ -27,7 +27,7 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
-public class EthSendRawTransactionTest extends AcceptanceTestBase {
+public class EthSendRawTransactionAcceptanceTest extends AcceptanceTestBase {
   private static final long CHAIN_ID = 20211;
 
   private Account sender;

ethereum/api/src/main/java/org/hyperledger/besu/ethereum/api/jsonrpc/JsonRpcErrorConverter.java

@@ -55,6 +55,8 @@ public static JsonRpcError convertTransactionInvalidReason(
         return JsonRpcError.MAX_PRIORITY_FEE_PER_GAS_EXCEEDS_MAX_FEE_PER_GAS;
       case OFFCHAIN_PRIVACY_GROUP_DOES_NOT_EXIST:
         return JsonRpcError.OFFCHAIN_PRIVACY_GROUP_DOES_NOT_EXIST;
+      case INVALID_TRANSACTION_FORMAT:
+        return JsonRpcError.INVALID_TRANSACTION_TYPE;
       case TRANSACTION_ALREADY_KNOWN:
         return JsonRpcError.ETH_SEND_TX_ALREADY_KNOWN;
       case TRANSACTION_REPLACEMENT_UNDERPRICED:

ethereum/api/src/main/java/org/hyperledger/besu/ethereum/api/jsonrpc/internal/response/JsonRpcError.java

@@ -55,6 +55,7 @@ public enum JsonRpcError {
   // Transaction validation failures
   NONCE_TOO_LOW(-32001, "Nonce too low"),
   INVALID_TRANSACTION_SIGNATURE(-32002, "Invalid signature"),
+  INVALID_TRANSACTION_TYPE(-32602, "Invalid transaction type"),
   INTRINSIC_GAS_EXCEEDS_LIMIT(-32003, "Intrinsic gas exceeds gas limit"),
   TRANSACTION_UPFRONT_COST_EXCEEDS_BALANCE(-32004, "Upfront cost exceeds account balance"),
   EXCEEDS_BLOCK_GAS_LIMIT(-32005, "Transaction gas limit exceeds block gas limit"),

ethereum/api/src/test/java/org/hyperledger/besu/ethereum/api/jsonrpc/internal/methods/EthSendRawTransactionTest.java

@@ -41,6 +41,7 @@ public class EthSendRawTransactionTest {
 
   private static final String VALID_TRANSACTION =
       "0xf86d0485174876e800830222e0945aae326516b4f8fe08074b7e972e40a713048d62880de0b6b3a7640000801ba05d4e7998757264daab67df2ce6f7e7a0ae36910778a406ca73898c9899a32b9ea0674700d5c3d1d27f2e6b4469957dfd1a1c49bf92383d80717afc84eb05695d5b";
+
   @Mock private TransactionPool transactionPool;
   private EthSendRawTransaction method;
 
@@ -170,6 +171,12 @@ public void transactionWithNotWhitelistedSenderAccountIsRejected() {
         TransactionInvalidReason.TX_SENDER_NOT_AUTHORIZED, JsonRpcError.TX_SENDER_NOT_AUTHORIZED);
   }
 
+  @Test
+  public void transactionWithIncorrectTransactionTypeIsRejected() {
+    verifyErrorForInvalidTransaction(
+        TransactionInvalidReason.INVALID_TRANSACTION_FORMAT, JsonRpcError.INVALID_TRANSACTION_TYPE);
+  }
+
   @Test
   public void transactionWithFeeCapExceededIsRejected() {
     verifyErrorForInvalidTransaction(

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

@@ -26,7 +26,6 @@ private enum Status {
     DROPPED,
     TRY_NEXT_LAYER,
     ADDED,
-    REORG_SENDER,
     INTERNAL_ERROR
   }
 
@@ -43,9 +42,6 @@ private enum Status {
   public static final TransactionAddedResult TRY_NEXT_LAYER =
       new TransactionAddedResult(Status.TRY_NEXT_LAYER);
 
-  public static final TransactionAddedResult REORG_SENDER =
-      new TransactionAddedResult(Status.REORG_SENDER);
-
   public static final TransactionAddedResult DROPPED = new TransactionAddedResult(Status.DROPPED);
 
   public static final TransactionAddedResult INTERNAL_ERROR =

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

@@ -109,6 +109,15 @@ public OptionalLong getNextNonceFor(final Address sender) {
     return nextLayerRes;
   }
 
+  @Override
+  public OptionalLong getCurrentNonceFor(final Address sender) {
+    final var senderTxs = txsBySender.get(sender);
+    if (senderTxs != null) {
+      return OptionalLong.of(senderTxs.firstKey());
+    }
+    return nextLayer.getCurrentNonceFor(sender);
+  }
+
   @Override
   protected void internalNotifyAdded(
       final NavigableMap<Long, PendingTransaction> senderTxs,

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

@@ -17,7 +17,6 @@
 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.TransactionAddedResult.REORG_SENDER;
 import static org.hyperledger.besu.ethereum.eth.transactions.TransactionAddedResult.TRY_NEXT_LAYER;
 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;
@@ -274,9 +273,6 @@ private TransactionAddedResult addToNextLayer(
       nextLayerDistance = distance;
     } else {
       nextLayerDistance = (int) (pendingTransaction.getNonce() - (senderTxs.lastKey() + 1));
-      if (nextLayerDistance < 0) {
-        return REORG_SENDER;
-      }
     }
     return nextLayer.add(pendingTransaction, nextLayerDistance);
   }

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

@@ -109,6 +109,11 @@ public OptionalLong getNextNonceFor(final Address sender) {
     return OptionalLong.empty();
   }
 
+  @Override
+  public OptionalLong getCurrentNonceFor(final Address sender) {
+    return OptionalLong.empty();
+  }
+
   @Override
   public PendingTransaction promote(final Predicate<PendingTransaction> promotionFilter) {
     return null;

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

@@ -20,9 +20,8 @@
 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.TransactionAddedResult.REORG_SENDER;
 import static org.hyperledger.besu.ethereum.eth.transactions.layered.TransactionsLayer.RemovalReason.INVALIDATED;
-import static org.hyperledger.besu.ethereum.eth.transactions.layered.TransactionsLayer.RemovalReason.REORG;
+import static org.hyperledger.besu.ethereum.eth.transactions.layered.TransactionsLayer.RemovalReason.RECONCILED;
 
 import org.hyperledger.besu.datatypes.Address;
 import org.hyperledger.besu.datatypes.Hash;
@@ -39,6 +38,7 @@
 import org.hyperledger.besu.evm.account.Account;
 import org.hyperledger.besu.evm.account.AccountState;
 
+import java.util.ArrayDeque;
 import java.util.ArrayList;
 import java.util.HashSet;
 import java.util.List;
@@ -95,66 +95,140 @@ public synchronized TransactionAddedResult addLocalTransaction(
   TransactionAddedResult addTransaction(
       final PendingTransaction pendingTransaction, final Optional<Account> maybeSenderAccount) {
 
-    final long senderNonce = maybeSenderAccount.map(AccountState::getNonce).orElse(0L);
+    final long stateSenderNonce = maybeSenderAccount.map(AccountState::getNonce).orElse(0L);
 
-    logTransactionForReplayAdd(pendingTransaction, senderNonce);
+    logTransactionForReplayAdd(pendingTransaction, stateSenderNonce);
 
-    final long nonceDistance = pendingTransaction.getNonce() - senderNonce;
+    if (hasAccountNonceDisparity(pendingTransaction, stateSenderNonce)) {
+      reconcileSender(pendingTransaction.getSender(), stateSenderNonce);
+    }
+
+    final long nonceDistance = pendingTransaction.getNonce() - stateSenderNonce;
 
     final TransactionAddedResult nonceChecksResult =
-        nonceChecks(pendingTransaction, senderNonce, nonceDistance);
+        nonceChecks(pendingTransaction, stateSenderNonce, nonceDistance);
     if (nonceChecksResult != null) {
       return nonceChecksResult;
     }
 
     try {
-      TransactionAddedResult result =
-          prioritizedTransactions.add(pendingTransaction, (int) nonceDistance);
-
-      if (result.equals(REORG_SENDER)) {
-        result = reorgSenderOf(pendingTransaction, (int) nonceDistance);
-      }
-
-      return result;
+      return prioritizedTransactions.add(pendingTransaction, (int) nonceDistance);
     } catch (final Throwable throwable) {
-      // in case something unexpected happened, log this sender txs and force a reorg of his txs
+      return reconcileAndRetryAdd(
+          pendingTransaction, stateSenderNonce, (int) nonceDistance, throwable);
+    }
+  }
+
+  private TransactionAddedResult reconcileAndRetryAdd(
+      final PendingTransaction pendingTransaction,
+      final long stateSenderNonce,
+      final int nonceDistance,
+      final Throwable throwable) {
+    // in case something unexpected happened, log this sender txs, force a reconcile and retry
+    // another time
+    // ToDo: demote to debug when Layered TxPool is out of preview
+    LOG.warn(
+        "Unexpected error {} when adding transaction {}, current sender status {}",
+        throwable,
+        pendingTransaction.toTraceLog(),
+        prioritizedTransactions.logSender(pendingTransaction.getSender()));
+    LOG.warn("Stack trace", throwable);
+    reconcileSender(pendingTransaction.getSender(), stateSenderNonce);
+    try {
+      return prioritizedTransactions.add(pendingTransaction, nonceDistance);
+    } catch (final Throwable throwable2) {
       LOG.warn(
           "Unexpected error {} when adding transaction {}, current sender status {}",
           throwable,
           pendingTransaction.toTraceLog(),
           prioritizedTransactions.logSender(pendingTransaction.getSender()));
       LOG.warn("Stack trace", throwable);
-      reorgSenderOf(pendingTransaction, (int) nonceDistance);
       return INTERNAL_ERROR;
     }
   }
 
-  private TransactionAddedResult reorgSenderOf(
-      final PendingTransaction pendingTransaction, final int nonceDistance) {
-    final var existingSenderTxs = prioritizedTransactions.getAllFor(pendingTransaction.getSender());
+  /**
+   * Detect a disparity between account nonce has seen by the world state and the txpool, that could
+   * happen during the small amount of time during block import when the world state is updated
+   * while the txpool still does not process the confirmed txs, or when there is a reorg and the
+   * sender nonce goes back.
+   *
+   * @param pendingTransaction the incoming transaction to check
+   * @param stateSenderNonce account nonce from the world state
+   * @return false if the nonce for the sender has seen by the txpool matches the value of the
+   *     account nonce in the world state, true if they differ
+   */
+  private boolean hasAccountNonceDisparity(
+      final PendingTransaction pendingTransaction, final long stateSenderNonce) {
+    final OptionalLong maybeTxPoolSenderNonce =
+        prioritizedTransactions.getCurrentNonceFor(pendingTransaction.getSender());
+    if (maybeTxPoolSenderNonce.isPresent()) {
+      final long txPoolSenderNonce = maybeTxPoolSenderNonce.getAsLong();
+      if (stateSenderNonce != txPoolSenderNonce) {
+        LOG.atDebug()
+            .setMessage(
+                "Nonce disparity detected when adding pending transaction {}. "
+                    + "Account nonce from world state is {} while current txpool nonce is {}")
+            .addArgument(pendingTransaction::toTraceLog)
+            .addArgument(stateSenderNonce)
+            .addArgument(txPoolSenderNonce)
+            .log();
+        return true;
+      }
+    }
+    return false;
+  }
+
+  /**
+   * Rebuild the txpool for a sender according to the specified nonce. This is used in case the
+   * account nonce has seen by the txpool is not the correct one (see {@link
+   * LayeredPendingTransactions#hasAccountNonceDisparity(PendingTransaction, long)} for when this
+   * could happen). It works by removing all the txs for the sender and re-adding them using the
+   * passed nonce.
+   *
+   * @param sender the sender for which rebuild the txpool
+   * @param stateSenderNonce the world state account nonce to use in the txpool for the sender
+   */
+  private void reconcileSender(final Address sender, final long stateSenderNonce) {
+    final var existingSenderTxs = prioritizedTransactions.getAllFor(sender);
+    if (existingSenderTxs.isEmpty()) {
+      LOG.debug("Sender {} has no transactions to reconcile", sender);
+      return;
+    }
+
+    LOG.atDebug()
+        .setMessage("Sender {} with nonce {} has {} transaction(s) to reconcile {}")
+        .addArgument(sender)
+        .addArgument(stateSenderNonce)
+        .addArgument(existingSenderTxs::size)
+        .addArgument(() -> prioritizedTransactions.logSender(sender))
+        .log();
+
+    final var reAddTxs = new ArrayDeque<PendingTransaction>(existingSenderTxs.size());
 
     // it is more performant to invalidate backward
     for (int i = existingSenderTxs.size() - 1; i >= 0; --i) {
-      prioritizedTransactions.remove(existingSenderTxs.get(i), REORG);
+      final var ptx = existingSenderTxs.get(i);
+      prioritizedTransactions.remove(ptx, RECONCILED);
+      if (ptx.getNonce() >= stateSenderNonce) {
+        reAddTxs.addFirst(ptx);
+      }
     }
 
-    // add the new one and re-add all the previous
-    final var result = prioritizedTransactions.add(pendingTransaction, nonceDistance);
-    existingSenderTxs.forEach(ptx -> prioritizedTransactions.add(ptx, nonceDistance));
-    LOG.atTrace()
-        .setMessage(
-            "Pending transaction {} with nonce distance {} triggered a reorg for sender {} with {} existing transactions: {}")
-        .addArgument(pendingTransaction::toTraceLog)
-        .addArgument(nonceDistance)
-        .addArgument(pendingTransaction::getSender)
-        .addArgument(existingSenderTxs::size)
-        .addArgument(
-            () ->
-                existingSenderTxs.stream()
-                    .map(PendingTransaction::toTraceLog)
-                    .collect(Collectors.joining("; ")))
+    if (!reAddTxs.isEmpty()) {
+      // re-add all the previous txs
+      final long lowestNonce = reAddTxs.getFirst().getNonce();
+      final int newNonceDistance = (int) Math.max(0, lowestNonce - stateSenderNonce);
+
+      reAddTxs.forEach(ptx -> prioritizedTransactions.add(ptx, newNonceDistance));
+    }
+
+    LOG.atDebug()
+        .setMessage("Sender {} with nonce {} status after reconciliation {}")
+        .addArgument(sender)
+        .addArgument(stateSenderNonce)
+        .addArgument(() -> prioritizedTransactions.logSender(sender))
         .log();
-    return result;
   }
 
   private void logTransactionForReplayAdd(
@@ -253,7 +327,7 @@ public synchronized void selectTransactions(
         .takeWhile(unused -> !completed.get())
         .peek(
             highPrioPendingTx ->
-                LOG.atDebug()
+                LOG.atTrace()
                     .setMessage("highPrioPendingTx {}, senderTxs {}")
                     .addArgument(highPrioPendingTx::toTraceLog)
                     .addArgument(
@@ -361,7 +435,7 @@ public synchronized void manageBlockAdded(
       final List<Transaction> confirmedTransactions,
       final List<Transaction> reorgTransactions,
       final FeeMarket feeMarket) {
-    LOG.atDebug()
+    LOG.atTrace()
         .setMessage("Managing new added block {}")
         .addArgument(blockHeader::toLogString)
         .log();

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

@@ -273,6 +273,16 @@ public OptionalLong getNextNonceFor(final Address sender) {
     return OptionalLong.empty();
   }
 
+  @Override
+  public OptionalLong getCurrentNonceFor(final Address sender) {
+    final var senderTxs = txsBySender.get(sender);
+    if (senderTxs != null) {
+      final var gap = gapBySender.get(sender);
+      return OptionalLong.of(senderTxs.firstKey() - gap);
+    }
+    return nextLayer.getCurrentNonceFor(sender);
+  }
+
   @Override
   protected void internalNotifyAdded(
       final NavigableMap<Long, PendingTransaction> senderTxs,

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

@@ -57,6 +57,15 @@ void blockAdded(
 
   OptionalLong getNextNonceFor(Address sender);
 
+  /**
+   * Get the sender nonce has seen by this and the following layers
+   *
+   * @param sender the sender for which retrieve the txpool
+   * @return either the sender nonce or empty if the sender is unknown to this or the following
+   *     layers
+   */
+  OptionalLong getCurrentNonceFor(Address sender);
+
   PendingTransaction promote(Predicate<PendingTransaction> promotionFilter);
 
   long subscribeToAdded(PendingTransactionAddedListener listener);
@@ -88,7 +97,7 @@ enum RemovalReason {
     INVALIDATED,
     PROMOTED,
     REPLACED,
-    REORG;
+    RECONCILED;
 
     private final String label;