Interrupt candidate tx execution on block creation timeout

Signed-off-by: Fabio Di Fabio <fabio.difabio@consensys.net>

Author: fab-10

CHANGELOG.md

@@ -1,8 +1,6 @@
 # Changelog
 
 ## [Unreleased]
-- Add configuration of Consolidation Request Contract Address via genesis configuration [#7647](https://github.com/hyperledger/besu/pull/7647)
-
 
 ### Upcoming Breaking Changes
 - k8s (KUBERNETES) Nat method is now deprecated and will be removed in a future release
@@ -14,7 +12,8 @@
 - Remove privacy test classes support [#7569](https://github.com/hyperledger/besu/pull/7569)
 - Add Blob Transaction Metrics [#7622](https://github.com/hyperledger/besu/pull/7622)
 - Implemented support for emptyBlockPeriodSeconds in QBFT [#6965](https://github.com/hyperledger/besu/pull/6965)
-
+- Add configuration of Consolidation Request Contract Address via genesis configuration [#7647](https://github.com/hyperledger/besu/pull/7647)
+- Interrupt pending transaction processing on block creation timeout [#7673](https://github.com/hyperledger/besu/pull/7673)
 
 ### Bug fixes
 - Fix mounted data path directory permissions for besu user [#7575](https://github.com/hyperledger/besu/pull/7575)

ethereum/blockcreation/src/main/java/org/hyperledger/besu/ethereum/blockcreation/txselection/BlockTransactionSelector.java

@@ -15,6 +15,7 @@
 package org.hyperledger.besu.ethereum.blockcreation.txselection;
 
 import static org.hyperledger.besu.plugin.data.TransactionSelectionResult.BLOCK_SELECTION_TIMEOUT;
+import static org.hyperledger.besu.plugin.data.TransactionSelectionResult.BLOCK_SELECTION_TIMEOUT_INVALID_TX;
 import static org.hyperledger.besu.plugin.data.TransactionSelectionResult.INVALID_TX_EVALUATION_TOO_LONG;
 import static org.hyperledger.besu.plugin.data.TransactionSelectionResult.SELECTED;
 import static org.hyperledger.besu.plugin.data.TransactionSelectionResult.TX_EVALUATION_TOO_LONG;
@@ -52,9 +53,11 @@
 import org.hyperledger.besu.plugin.services.tracer.BlockAwareOperationTracer;
 import org.hyperledger.besu.plugin.services.txselection.PluginTransactionSelector;
 
+import java.time.Duration;
 import java.util.List;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.FutureTask;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -97,11 +100,12 @@ public class BlockTransactionSelector {
       new TransactionSelectionResults();
   private final List<AbstractTransactionSelector> transactionSelectors;
   private final PluginTransactionSelector pluginTransactionSelector;
-  private final BlockAwareOperationTracer pluginOperationTracer;
+  private final BlockAwareOperationTracer operationTracer;
   private final EthScheduler ethScheduler;
   private final AtomicBoolean isTimeout = new AtomicBoolean(false);
   private final long blockTxsSelectionMaxTime;
   private WorldUpdater blockWorldStateUpdater;
+  private volatile TransactionEvaluationContext currTxEvaluationContext;
 
   public BlockTransactionSelector(
       final MiningParameters miningParameters,
@@ -139,7 +143,8 @@ public BlockTransactionSelector(
             transactionPool);
     transactionSelectors = createTransactionSelectors(blockSelectionContext);
     this.pluginTransactionSelector = pluginTransactionSelector;
-    this.pluginOperationTracer = pluginTransactionSelector.getOperationTracer();
+    this.operationTracer =
+        new InterruptibleOperationTracer(pluginTransactionSelector.getOperationTracer());
     blockWorldStateUpdater = worldState.updater();
     blockTxsSelectionMaxTime = miningParameters.getBlockTxsSelectionMaxTime();
   }
@@ -178,15 +183,17 @@ public TransactionSelectionResults buildTransactionListForBlock() {
   }
 
   private void timeLimitedSelection() {
-    final var txSelection =
-        ethScheduler.scheduleBlockCreationTask(
+    final var txSelectionTask =
+        new FutureTask<Void>(
             () ->
                 blockSelectionContext
                     .transactionPool()
-                    .selectTransactions(this::evaluateTransaction));
+                    .selectTransactions(this::evaluateTransaction),
+            null);
+    ethScheduler.scheduleBlockCreationTask(txSelectionTask);
 
     try {
-      txSelection.get(blockTxsSelectionMaxTime, TimeUnit.MILLISECONDS);
+      txSelectionTask.get(blockTxsSelectionMaxTime, TimeUnit.MILLISECONDS);
     } catch (InterruptedException | ExecutionException e) {
       if (isCancelled.get()) {
         throw new CancellationException("Cancelled during transaction selection");
@@ -197,6 +204,9 @@ private void timeLimitedSelection() {
       synchronized (isTimeout) {
         isTimeout.set(true);
       }
+
+      cancelEvaluatingTxWithGraceTime(txSelectionTask);
+
       LOG.warn(
           "Interrupting the selection of transactions for block inclusion as it exceeds the maximum configured duration of "
               + blockTxsSelectionMaxTime
@@ -205,6 +215,40 @@ private void timeLimitedSelection() {
     }
   }
 
+  private void cancelEvaluatingTxWithGraceTime(final FutureTask<Void> txSelectionTask) {
+    final long elapsedTime =
+        currTxEvaluationContext.getEvaluationTimer().elapsed(TimeUnit.MILLISECONDS);
+    // adding 100ms so we are sure it take strictly more than the block selection max time
+    final long txRemainingTime = (blockTxsSelectionMaxTime - elapsedTime) + 100;
+
+    LOG.atDebug()
+        .setMessage(
+            "Transaction {} is processing for {}ms, giving it {}ms grace time, before considering it taking too much time to execute")
+        .addArgument(currTxEvaluationContext.getPendingTransaction()::toTraceLog)
+        .addArgument(elapsedTime)
+        .addArgument(txRemainingTime)
+        .log();
+
+    ethScheduler.scheduleFutureTask(
+        () -> {
+          if (!txSelectionTask.isDone()) {
+            LOG.atDebug()
+                .setMessage(
+                    "Transaction {} is still processing after the grace time, total processing time {}ms,"
+                        + " greater than max block selection time of {}ms, forcing an interrupt")
+                .addArgument(currTxEvaluationContext.getPendingTransaction()::toTraceLog)
+                .addArgument(
+                    () ->
+                        currTxEvaluationContext.getEvaluationTimer().elapsed(TimeUnit.MILLISECONDS))
+                .addArgument(blockTxsSelectionMaxTime)
+                .log();
+
+            txSelectionTask.cancel(true);
+          }
+        },
+        Duration.ofMillis(txRemainingTime));
+  }
+
   /**
    * Evaluates a list of transactions and updates the selection results accordingly. If a
    * transaction is not selected during the evaluation, it is updated as not selected in the
@@ -236,6 +280,7 @@ private TransactionSelectionResult evaluateTransaction(
 
     final TransactionEvaluationContext evaluationContext =
         createTransactionEvaluationContext(pendingTransaction);
+    currTxEvaluationContext = evaluationContext;
 
     TransactionSelectionResult selectionResult = evaluatePreProcessing(evaluationContext);
     if (!selectionResult.selected()) {
@@ -337,7 +382,7 @@ private TransactionProcessingResult processTransaction(
         blockSelectionContext.pendingBlockHeader(),
         pendingTransaction.getTransaction(),
         blockSelectionContext.miningBeneficiary(),
-        pluginOperationTracer,
+        operationTracer,
         blockHashLookup,
         false,
         TransactionValidationParams.mining(),
@@ -422,14 +467,10 @@ private TransactionSelectionResult handleTransactionNotSelected(
     final var pendingTransaction = evaluationContext.getPendingTransaction();
 
     // check if this tx took too much to evaluate, and in case it was invalid remove it from the
-    // pool, otherwise penalize it.
+    // pool, otherwise penalize it. Not synchronized since there is no state change here.
     final TransactionSelectionResult actualResult =
         isTimeout.get()
-            ? transactionTookTooLong(evaluationContext, selectionResult)
-                ? selectionResult.discard()
-                    ? INVALID_TX_EVALUATION_TOO_LONG
-                    : TX_EVALUATION_TOO_LONG
-                : BLOCK_SELECTION_TIMEOUT
+            ? rewriteSelectionResultForTimeout(evaluationContext, selectionResult)
             : selectionResult;
 
     transactionSelectionResults.updateNotSelected(evaluationContext.getTransaction(), actualResult);
@@ -446,6 +487,34 @@ private TransactionSelectionResult handleTransactionNotSelected(
     return actualResult;
   }
 
+  /**
+   * In case of a block creation timeout, we rewrite the selection result, so we can easily spot
+   * what happened looking at the transaction selection results.
+   *
+   * @param evaluationContext The current selection session data.
+   * @param selectionResult The result of the transaction selection process.
+   * @return the rewritten selection result
+   */
+  private TransactionSelectionResult rewriteSelectionResultForTimeout(
+      final TransactionEvaluationContext evaluationContext,
+      final TransactionSelectionResult selectionResult) {
+
+    if (transactionTookTooLong(evaluationContext, selectionResult)) {
+      return selectionResult.discard() ? INVALID_TX_EVALUATION_TOO_LONG : TX_EVALUATION_TOO_LONG;
+    }
+
+    return selectionResult.discard() ? BLOCK_SELECTION_TIMEOUT_INVALID_TX : BLOCK_SELECTION_TIMEOUT;
+  }
+
+  /**
+   * Check if the evaluation of this tx took more than the block creation max time, because if true
+   * we want to penalize it. We penalize it, instead of directly removing, because it could happen
+   * that the tx will evaluate in time next time. Invalid txs are always removed.
+   *
+   * @param evaluationContext The current selection session data.
+   * @param selectionResult The result of the transaction selection process.
+   * @return true if the evaluation of this tx took more than the block creation max time
+   */
   private boolean transactionTookTooLong(
       final TransactionEvaluationContext evaluationContext,
       final TransactionSelectionResult selectionResult) {

ethereum/blockcreation/src/main/java/org/hyperledger/besu/ethereum/blockcreation/txselection/InterruptibleOperationTracer.java

@@ -0,0 +1,143 @@
+/*
+ * 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.blockcreation.txselection;
+
+import org.hyperledger.besu.datatypes.Address;
+import org.hyperledger.besu.datatypes.Transaction;
+import org.hyperledger.besu.datatypes.Wei;
+import org.hyperledger.besu.evm.frame.ExceptionalHaltReason;
+import org.hyperledger.besu.evm.frame.MessageFrame;
+import org.hyperledger.besu.evm.log.Log;
+import org.hyperledger.besu.evm.operation.Operation;
+import org.hyperledger.besu.evm.worldstate.WorldView;
+import org.hyperledger.besu.plugin.data.BlockBody;
+import org.hyperledger.besu.plugin.data.BlockHeader;
+import org.hyperledger.besu.plugin.data.ProcessableBlockHeader;
+import org.hyperledger.besu.plugin.services.tracer.BlockAwareOperationTracer;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+
+import org.apache.tuweni.bytes.Bytes;
+
+public class InterruptibleOperationTracer implements BlockAwareOperationTracer {
+  private final BlockAwareOperationTracer delegate;
+
+  public InterruptibleOperationTracer(final BlockAwareOperationTracer delegate) {
+    this.delegate = delegate;
+  }
+
+  @Override
+  public void traceStartBlock(final BlockHeader blockHeader, final BlockBody blockBody) {
+    delegate.traceStartBlock(blockHeader, blockBody);
+  }
+
+  @Override
+  public void traceEndBlock(final BlockHeader blockHeader, final BlockBody blockBody) {
+    delegate.traceEndBlock(blockHeader, blockBody);
+  }
+
+  @Override
+  public void traceStartBlock(final ProcessableBlockHeader processableBlockHeader) {
+    delegate.traceStartBlock(processableBlockHeader);
+  }
+
+  @Override
+  public boolean isExtendedTracing() {
+    return delegate.isExtendedTracing();
+  }
+
+  @Override
+  public void tracePreExecution(final MessageFrame frame) {
+    checkInterrupt();
+    delegate.tracePreExecution(frame);
+  }
+
+  @Override
+  public void tracePostExecution(
+      final MessageFrame frame, final Operation.OperationResult operationResult) {
+    checkInterrupt();
+    delegate.tracePostExecution(frame, operationResult);
+  }
+
+  @Override
+  public void tracePrecompileCall(
+      final MessageFrame frame, final long gasRequirement, final Bytes output) {
+    checkInterrupt();
+    delegate.tracePrecompileCall(frame, gasRequirement, output);
+  }
+
+  @Override
+  public void traceAccountCreationResult(
+      final MessageFrame frame, final Optional<ExceptionalHaltReason> haltReason) {
+    checkInterrupt();
+    delegate.traceAccountCreationResult(frame, haltReason);
+  }
+
+  @Override
+  public void tracePrepareTransaction(final WorldView worldView, final Transaction transaction) {
+    delegate.tracePrepareTransaction(worldView, transaction);
+  }
+
+  @Override
+  public void traceStartTransaction(final WorldView worldView, final Transaction transaction) {
+    delegate.traceStartTransaction(worldView, transaction);
+  }
+
+  @Override
+  public void traceBeforeRewardTransaction(
+      final WorldView worldView, final Transaction tx, final Wei miningReward) {
+    delegate.traceBeforeRewardTransaction(worldView, tx, miningReward);
+  }
+
+  @Override
+  public void traceEndTransaction(
+      final WorldView worldView,
+      final Transaction tx,
+      final boolean status,
+      final Bytes output,
+      final List<Log> logs,
+      final long gasUsed,
+      final Set<Address> selfDestructs,
+      final long timeNs) {
+    delegate.traceEndTransaction(
+        worldView, tx, status, output, logs, gasUsed, selfDestructs, timeNs);
+  }
+
+  @Override
+  public void traceContextEnter(final MessageFrame frame) {
+    checkInterrupt();
+    delegate.traceContextEnter(frame);
+  }
+
+  @Override
+  public void traceContextReEnter(final MessageFrame frame) {
+    checkInterrupt();
+    delegate.traceContextReEnter(frame);
+  }
+
+  @Override
+  public void traceContextExit(final MessageFrame frame) {
+    checkInterrupt();
+    delegate.traceContextExit(frame);
+  }
+
+  private void checkInterrupt() {
+    if (Thread.interrupted()) {
+      throw new RuntimeException(new InterruptedException("Transaction execution interrupted"));
+    }
+  }
+}

ethereum/blockcreation/src/main/java/org/hyperledger/besu/ethereum/blockcreation/txselection/selectors/ProcessingResultTransactionSelector.java

@@ -112,6 +112,7 @@ private TransactionSelectionResult transactionSelectionResultForInvalidResult(
   private boolean isTransientValidationError(final TransactionInvalidReason invalidReason) {
     return invalidReason.equals(TransactionInvalidReason.UPFRONT_COST_EXCEEDS_BALANCE)
         || invalidReason.equals(TransactionInvalidReason.GAS_PRICE_BELOW_CURRENT_BASE_FEE)
-        || invalidReason.equals(TransactionInvalidReason.NONCE_TOO_HIGH);
+        || invalidReason.equals(TransactionInvalidReason.NONCE_TOO_HIGH)
+        || invalidReason.equals(TransactionInvalidReason.EXECUTION_INTERRUPTED);
   }
 }