Fix: interrupt bundle selection on timeout (#1756)

* Fix: Interrupt bundle selection on timeout

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

* Apply suggestions from code review

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

* Added SELECTION_CANCELLED as interrupt reason

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

* Apply suggestions from code review

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

* Fix/interrupt bundle selection on timeout test fix (#1758)

* fix bundle test

* Rename variable

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

---------

Signed-off-by: Fabio Di Fabio <[email protected]>
Co-authored-by: Fluent Crafter <[email protected]>

Author: fab-10

besu-plugins/linea-sequencer/acceptance-tests/src/test/java/linea/plugin/acc/test/rpc/linea/BundleSelectionTimeoutTest.java

@@ -63,7 +63,7 @@ public void multipleBundleSelectionTimeout() throws Exception {
     final var mulmodExecutor = deployMulmodExecutor();
 
     final var calls =
-        IntStream.rangeClosed(1, 10)
+        IntStream.rangeClosed(1, 11)
             .mapToObj(
                 nonce ->
                     mulmodOperation(
@@ -74,16 +74,23 @@ public void multipleBundleSelectionTimeout() throws Exception {
                         BigInteger.valueOf(MAX_TX_GAS_LIMIT / 10)))
             .toArray(MulmodCall[]::new);
 
+    final var rawTxs = Arrays.stream(calls).map(MulmodCall::rawTx).toArray(String[]::new);
     final var sendBundleRequestSmall =
         new SendBundleRequest(
-            new BundleParams(new String[] {calls[0].rawTx()}, Integer.toHexString(2)));
-    final var sendBundleResponseSmall = sendBundleRequestSmall.execute(minerNode.nodeRequests());
+            new BundleParams(Arrays.copyOfRange(rawTxs, 0, 1), Integer.toHexString(2)));
 
-    final var rawTxs = Arrays.stream(calls).skip(1).map(MulmodCall::rawTx).toArray(String[]::new);
+    final var sendBundleRequestBig1 =
+        new SendBundleRequest(
+            new BundleParams(Arrays.copyOfRange(rawTxs, 1, 10), Integer.toHexString(2)));
+    // second bundle contains one tx only to be fast to execute,
+    // and ensure timeout occurs on the 2nd bundle and 3rd is not event considered
+    final var sendBundleRequestBig2 =
+        new SendBundleRequest(
+            new BundleParams(Arrays.copyOfRange(rawTxs, 1, 2), Integer.toHexString(2)));
 
-    final var sendBundleRequestBig =
-        new SendBundleRequest(new BundleParams(rawTxs, Integer.toHexString(2)));
-    final var sendBundleResponseBig = sendBundleRequestBig.execute(minerNode.nodeRequests());
+    final var sendBundleResponseSmall = sendBundleRequestSmall.execute(minerNode.nodeRequests());
+    final var sendBundleResponseBig1 = sendBundleRequestBig1.execute(minerNode.nodeRequests());
+    final var sendBundleResponseBig2 = sendBundleRequestBig2.execute(minerNode.nodeRequests());
 
     final var transferTxHash =
         accountTransactions
@@ -93,15 +100,18 @@ public void multipleBundleSelectionTimeout() throws Exception {
     assertThat(sendBundleResponseSmall.hasError()).isFalse();
     assertThat(sendBundleResponseSmall.getResult().bundleHash()).isNotBlank();
 
-    assertThat(sendBundleResponseBig.hasError()).isFalse();
-    assertThat(sendBundleResponseBig.getResult().bundleHash()).isNotBlank();
+    assertThat(sendBundleResponseBig1.hasError()).isFalse();
+    assertThat(sendBundleResponseBig1.getResult().bundleHash()).isNotBlank();
+
+    assertThat(sendBundleResponseBig2.hasError()).isFalse();
+    assertThat(sendBundleResponseBig2.getResult().bundleHash()).isNotBlank();
 
     minerNode.verify(eth.expectSuccessfulTransactionReceipt(transferTxHash.toHexString()));
 
     // first bundle is successful
     minerNode.verify(eth.expectSuccessfulTransactionReceipt(calls[0].txHash()));
 
-    // second bundle is not
+    // following bundles are not selected
     Arrays.stream(calls)
         .skip(1)
         .map(MulmodCall::txHash)

besu-plugins/linea-sequencer/sequencer/src/main/java/net/consensys/linea/sequencer/txselection/LineaTransactionSelectorFactory.java

@@ -9,11 +9,16 @@
 
 package net.consensys.linea.sequencer.txselection;
 
+import static org.hyperledger.besu.plugin.data.TransactionSelectionResult.PLUGIN_SELECTION_TIMEOUT;
+import static org.hyperledger.besu.plugin.data.TransactionSelectionResult.PLUGIN_SELECTION_TIMEOUT_INVALID_TX;
+import static org.hyperledger.besu.plugin.data.TransactionSelectionResult.SELECTION_CANCELLED;
+
 import java.time.Instant;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 import lombok.extern.slf4j.Slf4j;
 import net.consensys.linea.bundles.BundlePoolService;
@@ -29,6 +34,7 @@
 import net.consensys.linea.sequencer.txselection.selectors.TransactionEventFilter;
 import org.hyperledger.besu.datatypes.Address;
 import org.hyperledger.besu.plugin.data.ProcessableBlockHeader;
+import org.hyperledger.besu.plugin.data.TransactionSelectionResult;
 import org.hyperledger.besu.plugin.services.BlockchainService;
 import org.hyperledger.besu.plugin.services.txselection.BlockTransactionSelectionService;
 import org.hyperledger.besu.plugin.services.txselection.PluginTransactionSelector;
@@ -56,6 +62,7 @@ public class LineaTransactionSelectorFactory implements PluginTransactionSelecto
   private final AtomicReference<LineaTransactionSelector> currSelector = new AtomicReference<>();
   private final AtomicReference<Map<Address, Set<TransactionEventFilter>>> deniedEvents;
   private final AtomicReference<Map<Address, Set<TransactionEventFilter>>> deniedBundleEvents;
+  private final AtomicBoolean isSelectionInterrupted = new AtomicBoolean(false);
 
   public LineaTransactionSelectorFactory(
       final BlockchainService blockchainService,
@@ -109,8 +116,6 @@ public void selectPendingTransactions(
       // check and send liveness bundle if any
       checkAndSendLivenessBundle(bts, pendingBlockHeader.getNumber());
 
-      if (isSelectionInterrupted()) return;
-
       final var bundlesByBlockNumber =
           bundlePoolService.getBundlesByBlockNumber(pendingBlockHeader.getNumber());
 
@@ -121,20 +126,17 @@ public void selectPendingTransactions(
           .addArgument(bundlesByBlockNumber::size)
           .log();
 
-      if (isSelectionInterrupted()) return;
-
       final var selectionStartedAt = System.nanoTime();
 
       bundlesByBlockNumber.stream()
-          .takeWhile(unused -> !isSelectionInterrupted())
+          .takeWhile(unused -> !isSelectionInterrupted.get())
           .forEach(
               bundle -> {
                 final var bundleStartedAt = System.nanoTime();
                 log.trace("Starting evaluation of bundle {}", bundle.bundleIdentifier());
 
                 var maybeBadBundleRes =
                     bundle.pendingTransactions().stream()
-                        .takeWhile(unused -> !isSelectionInterrupted())
                         .map(bts::evaluatePendingTransaction)
                         .filter(evalRes -> !evalRes.selected())
                         .findFirst();
@@ -143,45 +145,46 @@ public void selectPendingTransactions(
                 final var cumulativeBundleSelectionTime = now - selectionStartedAt;
                 final var currentBundleSelectionTime = now - bundleStartedAt;
 
-                if (isSelectionInterrupted()) {
-                  log.atDebug()
-                      .setMessage(
-                          "Bundle selection interrupted while processing bundle {},"
-                              + " elapsed time: current bundle {}ms, cumulative {}ms")
-                      .addArgument(bundle::bundleIdentifier)
-                      .addArgument(() -> nanosToMillis(currentBundleSelectionTime))
-                      .addArgument(() -> nanosToMillis(cumulativeBundleSelectionTime))
-                      .log();
-                  rollback(bts);
-                } else {
-                  if (maybeBadBundleRes.isPresent()) {
+                if (maybeBadBundleRes.isPresent()) {
+                  final var notSelectedReason = maybeBadBundleRes.get();
+
+                  if (isSelectionInterrupted(notSelectedReason)) {
+                    isSelectionInterrupted.set(true);
                     log.atDebug()
                         .setMessage(
-                            "Failed bundle {}, reason {}, elapsed time: current bundle {}ms, cumulative {}ms")
+                            "Bundle selection interrupted while processing bundle {},"
+                                + " elapsed time: current bundle {}ms, cumulative {}ms")
                         .addArgument(bundle::bundleIdentifier)
-                        .addArgument(maybeBadBundleRes::get)
                         .addArgument(() -> nanosToMillis(currentBundleSelectionTime))
                         .addArgument(() -> nanosToMillis(cumulativeBundleSelectionTime))
                         .log();
-                    rollback(bts);
                   } else {
                     log.atDebug()
                         .setMessage(
-                            "Selected bundle {}, elapsed time: current bundle {}ms, cumulative {}ms")
+                            "Failed bundle {}, reason {}, elapsed time: current bundle {}ms, cumulative {}ms")
                         .addArgument(bundle::bundleIdentifier)
+                        .addArgument(notSelectedReason)
                         .addArgument(() -> nanosToMillis(currentBundleSelectionTime))
                         .addArgument(() -> nanosToMillis(cumulativeBundleSelectionTime))
                         .log();
-                    commit(bts);
                   }
+
+                  rollback(bts);
+                } else {
+                  log.atDebug()
+                      .setMessage(
+                          "Selected bundle {}, elapsed time: current bundle {}ms, cumulative {}ms")
+                      .addArgument(bundle::bundleIdentifier)
+                      .addArgument(() -> nanosToMillis(currentBundleSelectionTime))
+                      .addArgument(() -> nanosToMillis(cumulativeBundleSelectionTime))
+                      .log();
+
+                  commit(bts);
                 }
               });
     } finally {
       currSelector.set(null);
-      if (isSelectionInterrupted()) {
-        // finally consume the interrupt
-        Thread.currentThread().interrupt();
-      }
+      isSelectionInterrupted.set(false);
     }
   }
 
@@ -206,7 +209,18 @@ private void checkAndSendLivenessBundle(
               .findFirst();
 
       if (badBundleRes.isPresent()) {
-        log.debug("Failed liveness bundle {}, reason {}", livenessBundle.get(), badBundleRes);
+        final var notSelectedReason = badBundleRes.get();
+
+        if (isSelectionInterrupted(notSelectedReason)) {
+          isSelectionInterrupted.set(true);
+          log.debug(
+              "Bundle selection interrupted while processing liveness bundle {}, reason {}",
+              livenessBundle.get(),
+              notSelectedReason);
+        } else {
+          log.debug(
+              "Failed liveness bundle {}, reason {}", livenessBundle.get(), notSelectedReason);
+        }
         livenessService.get().updateUptimeMetrics(false, headBlockTimestamp);
         rollback(bts);
       } else {
@@ -231,9 +245,9 @@ private long nanosToMillis(final long nanos) {
     return TimeUnit.NANOSECONDS.toMillis(nanos);
   }
 
-  private boolean isSelectionInterrupted() {
-    // returns if the thread is interrupted without resetting the state
-    // so it can be called many times without changing the interrupt state
-    return Thread.currentThread().isInterrupted();
+  private boolean isSelectionInterrupted(final TransactionSelectionResult selectionResult) {
+    return selectionResult.equals(PLUGIN_SELECTION_TIMEOUT)
+        || selectionResult.equals(PLUGIN_SELECTION_TIMEOUT_INVALID_TX)
+        || selectionResult.equals(SELECTION_CANCELLED);
   }
 }