Inject synchronous Vertx as dependency for engine api (#4289)

Inject vertx dependency into EngineJsonRpcMethods instead of creating a new one. 

Improves testability, before some tests failed with too many files open because vertx inside EngineJsonRpcMethods was not being closed on test's afterEach hook

Signed-off-by: Pedro Novais <jpvnovais@gmail.com>

besu/src/main/java/org/hyperledger/besu/RunnerBuilder.java

@@ -142,6 +142,7 @@ import com.google.common.base.Preconditions;
 import com.google.common.base.Strings;
 import graphql.GraphQL;
 import io.vertx.core.Vertx;
+import io.vertx.core.VertxOptions;
 import org.apache.tuweni.bytes.Bytes;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -986,6 +987,9 @@ public class RunnerBuilder {
       final Map<String, BesuPlugin> namedPlugins,
       final Path dataDir,
       final RpcEndpointServiceImpl rpcEndpointServiceImpl) {
+    // sync vertx for engine consensus API, to process requests in FIFO order;
+    final Vertx consensusEngineServer = Vertx.vertx(new VertxOptions().setWorkerPoolSize(1));
+
     final Map<String, JsonRpcMethod> methods =
         new JsonRpcMethodsFactory()
             .methods(
@@ -1012,7 +1016,8 @@ public class RunnerBuilder {
                 natService,
                 namedPlugins,
                 dataDir,
-                besuController.getProtocolManager().ethContext().getEthPeers());
+                besuController.getProtocolManager().ethContext().getEthPeers(),
+                consensusEngineServer);
     methods.putAll(besuController.getAdditionalJsonRpcMethods(jsonRpcApis));
 
     var pluginMethods = rpcEndpointServiceImpl.getPluginMethods(jsonRpcConfiguration.getRpcApis());

ethereum/api/src/integration-test/java/org/hyperledger/besu/ethereum/api/jsonrpc/JsonRpcTestMethodsFactory.java

@@ -55,6 +55,9 @@ import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 
+import io.vertx.core.Vertx;
+import io.vertx.core.VertxOptions;
+
 /** Provides a facade to construct the JSON-RPC component. */
 public class JsonRpcTestMethodsFactory {
 
@@ -180,6 +183,7 @@ public class JsonRpcTestMethodsFactory {
             natService,
             new HashMap<>(),
             dataDir,
-            ethPeers);
+            ethPeers,
+            Vertx.vertx(new VertxOptions().setWorkerPoolSize(1)));
   }
 }

ethereum/api/src/main/java/org/hyperledger/besu/ethereum/api/jsonrpc/methods/ExecutionEngineJsonRpcMethods.java

@@ -30,7 +30,6 @@ import java.util.Map;
 import java.util.Optional;
 
 import io.vertx.core.Vertx;
-import io.vertx.core.VertxOptions;
 
 public class ExecutionEngineJsonRpcMethods extends ApiGroupJsonRpcMethods {
 
@@ -38,13 +37,14 @@ public class ExecutionEngineJsonRpcMethods extends ApiGroupJsonRpcMethods {
 
   private final Optional<MergeMiningCoordinator> mergeCoordinator;
   private final ProtocolContext protocolContext;
-
   private final EthPeers ethPeers;
+  private final Vertx consensusEngineServer;
 
   ExecutionEngineJsonRpcMethods(
       final MiningCoordinator miningCoordinator,
       final ProtocolContext protocolContext,
-      final EthPeers ethPeers) {
+      final EthPeers ethPeers,
+      final Vertx consensusEngineServer) {
     this.mergeCoordinator =
         Optional.ofNullable(miningCoordinator)
             .filter(mc -> mc.isCompatibleWithEngineApi())
@@ -52,6 +52,7 @@ public class ExecutionEngineJsonRpcMethods extends ApiGroupJsonRpcMethods {
 
     this.protocolContext = protocolContext;
     this.ethPeers = ethPeers;
+    this.consensusEngineServer = consensusEngineServer;
   }
 
   @Override
@@ -61,15 +62,17 @@ public class ExecutionEngineJsonRpcMethods extends ApiGroupJsonRpcMethods {
 
   @Override
   protected Map<String, JsonRpcMethod> create() {
-    Vertx syncVertx = Vertx.vertx(new VertxOptions().setWorkerPoolSize(1));
     if (mergeCoordinator.isPresent()) {
       return mapOf(
-          new EngineGetPayload(syncVertx, protocolContext, blockResultFactory),
-          new EngineNewPayload(syncVertx, protocolContext, mergeCoordinator.get(), ethPeers),
-          new EngineForkchoiceUpdated(syncVertx, protocolContext, mergeCoordinator.get()),
-          new EngineExchangeTransitionConfiguration(syncVertx, protocolContext));
+          new EngineGetPayload(consensusEngineServer, protocolContext, blockResultFactory),
+          new EngineNewPayload(
+              consensusEngineServer, protocolContext, mergeCoordinator.get(), ethPeers),
+          new EngineForkchoiceUpdated(
+              consensusEngineServer, protocolContext, mergeCoordinator.get()),
+          new EngineExchangeTransitionConfiguration(consensusEngineServer, protocolContext));
     } else {
-      return mapOf(new EngineExchangeTransitionConfiguration(syncVertx, protocolContext));
+      return mapOf(
+          new EngineExchangeTransitionConfiguration(consensusEngineServer, protocolContext));
     }
   }
 }

ethereum/api/src/main/java/org/hyperledger/besu/ethereum/api/jsonrpc/methods/JsonRpcMethodsFactory.java

@@ -46,6 +46,8 @@ import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 
+import io.vertx.core.Vertx;
+
 public class JsonRpcMethodsFactory {
 
   public Map<String, JsonRpcMethod> methods(
@@ -72,7 +74,8 @@ public class JsonRpcMethodsFactory {
       final NatService natService,
       final Map<String, BesuPlugin> namedPlugins,
       final Path dataDir,
-      final EthPeers ethPeers) {
+      final EthPeers ethPeers,
+      final Vertx consensusEngineServer) {
     final Map<String, JsonRpcMethod> enabled = new HashMap<>();
 
     if (!rpcApis.isEmpty()) {
@@ -94,7 +97,8 @@ public class JsonRpcMethodsFactory {
                   blockchainQueries, protocolSchedule, metricsSystem, transactionPool, dataDir),
               new EeaJsonRpcMethods(
                   blockchainQueries, protocolSchedule, transactionPool, privacyParameters),
-              new ExecutionEngineJsonRpcMethods(miningCoordinator, protocolContext, ethPeers),
+              new ExecutionEngineJsonRpcMethods(
+                  miningCoordinator, protocolContext, ethPeers, consensusEngineServer),
               new GoQuorumJsonRpcPrivacyMethods(
                   blockchainQueries, protocolSchedule, transactionPool, privacyParameters),
               new EthJsonRpcMethods(

ethereum/api/src/test/java/org/hyperledger/besu/ethereum/api/jsonrpc/AbstractJsonRpcHttpServiceTest.java

@@ -61,6 +61,7 @@ import java.util.Optional;
 import java.util.Set;
 
 import io.vertx.core.Vertx;
+import io.vertx.core.VertxOptions;
 import okhttp3.MediaType;
 import okhttp3.OkHttpClient;
 import org.junit.After;
@@ -84,6 +85,7 @@ public abstract class AbstractJsonRpcHttpServiceTest {
           RpcApis.TRACE.name());
 
   protected final Vertx vertx = Vertx.vertx();
+  protected final Vertx syncVertx = Vertx.vertx(new VertxOptions().setWorkerPoolSize(1));
   protected JsonRpcHttpService service;
   protected OkHttpClient client;
   protected String baseUrl;
@@ -188,7 +190,8 @@ public abstract class AbstractJsonRpcHttpServiceTest {
             natService,
             new HashMap<>(),
             folder.getRoot().toPath(),
-            mock(EthPeers.class));
+            mock(EthPeers.class),
+            syncVertx);
   }
 
   protected void startService() throws Exception {
@@ -224,5 +227,6 @@ public abstract class AbstractJsonRpcHttpServiceTest {
     client.connectionPool().evictAll();
     service.stop().join();
     vertx.close();
+    syncVertx.close();
   }
 }

ethereum/api/src/test/java/org/hyperledger/besu/ethereum/api/jsonrpc/JsonRpcHttpServiceHostAllowlistTest.java

@@ -122,7 +122,8 @@ public class JsonRpcHttpServiceHostAllowlistTest {
                     natService,
                     new HashMap<>(),
                     folder.getRoot().toPath(),
-                    mock(EthPeers.class)));
+                    mock(EthPeers.class),
+                    vertx));
     service = createJsonRpcHttpService();
     service.start().join();
 

ethereum/api/src/test/java/org/hyperledger/besu/ethereum/api/jsonrpc/JsonRpcHttpServiceLoginTest.java

@@ -87,6 +87,7 @@ import org.mockito.junit.MockitoJUnitRunner;
 
 @RunWith(MockitoJUnitRunner.StrictStubs.class)
 public class JsonRpcHttpServiceLoginTest {
+
   @ClassRule public static final TemporaryFolder folder = new TemporaryFolder();
 
   private static final Vertx vertx = Vertx.vertx();
@@ -151,7 +152,8 @@ public class JsonRpcHttpServiceLoginTest {
                     natService,
                     new HashMap<>(),
                     folder.getRoot().toPath(),
-                    mock(EthPeers.class)));
+                    mock(EthPeers.class),
+                    vertx));
     service = createJsonRpcHttpService();
     jwtAuth = service.authenticationService.get().getJwtAuthProvider();
     service.start().join();

ethereum/api/src/test/java/org/hyperledger/besu/ethereum/api/jsonrpc/JsonRpcHttpServiceRpcApisTest.java

@@ -81,6 +81,7 @@ import org.mockito.junit.MockitoJUnitRunner;
 
 @RunWith(MockitoJUnitRunner.class)
 public class JsonRpcHttpServiceRpcApisTest {
+
   @Rule public final TemporaryFolder folder = new TemporaryFolder();
 
   private final Vertx vertx = Vertx.vertx();
@@ -220,7 +221,8 @@ public class JsonRpcHttpServiceRpcApisTest {
                     natService,
                     new HashMap<>(),
                     folder.getRoot().toPath(),
-                    mock(EthPeers.class)));
+                    mock(EthPeers.class),
+                    vertx));
     final JsonRpcHttpService jsonRpcHttpService =
         new JsonRpcHttpService(
             vertx,
@@ -318,7 +320,8 @@ public class JsonRpcHttpServiceRpcApisTest {
                     natService,
                     new HashMap<>(),
                     folder.getRoot().toPath(),
-                    mock(EthPeers.class)));
+                    mock(EthPeers.class),
+                    vertx));
     final JsonRpcHttpService jsonRpcHttpService =
         new JsonRpcHttpService(
             vertx,

ethereum/api/src/test/java/org/hyperledger/besu/ethereum/api/jsonrpc/JsonRpcHttpServiceTestBase.java

@@ -130,7 +130,8 @@ public class JsonRpcHttpServiceTestBase {
                     natService,
                     new HashMap<>(),
                     folder.getRoot().toPath(),
-                    ethPeersMock));
+                    ethPeersMock,
+                    vertx));
     service = createJsonRpcHttpService(createLimitedJsonRpcConfig());
     service.start().join();
 

ethereum/api/src/test/java/org/hyperledger/besu/ethereum/api/jsonrpc/JsonRpcHttpServiceTlsClientAuthTest.java

@@ -137,7 +137,8 @@ public class JsonRpcHttpServiceTlsClientAuthTest {
                     natService,
                     Collections.emptyMap(),
                     folder.getRoot().toPath(),
-                    mock(EthPeers.class)));
+                    mock(EthPeers.class),
+                    vertx));
 
     System.setProperty("javax.net.ssl.trustStore", CLIENT_AS_CA_CERT.getKeyStoreFile().toString());
     System.setProperty(

ethereum/api/src/test/java/org/hyperledger/besu/ethereum/api/jsonrpc/JsonRpcHttpServiceTlsMisconfigurationTest.java

@@ -125,7 +125,8 @@ public class JsonRpcHttpServiceTlsMisconfigurationTest {
                     natService,
                     Collections.emptyMap(),
                     folder.getRoot().toPath(),
-                    mock(EthPeers.class)));
+                    mock(EthPeers.class),
+                    vertx));
   }
 
   @After

ethereum/api/src/test/java/org/hyperledger/besu/ethereum/api/jsonrpc/JsonRpcHttpServiceTlsTest.java

@@ -127,7 +127,8 @@ public class JsonRpcHttpServiceTlsTest {
                     natService,
                     Collections.emptyMap(),
                     folder.getRoot().toPath(),
-                    mock(EthPeers.class)));
+                    mock(EthPeers.class),
+                    vertx));
     service = createJsonRpcHttpService(createJsonRpcConfig());
     service.start().join();
     baseUrl = service.url();

ethereum/api/src/test/java/org/hyperledger/besu/ethereum/api/jsonrpc/websocket/WebSocketServiceLoginTest.java

@@ -184,7 +184,8 @@ public class WebSocketServiceLoginTest {
                     natService,
                     new HashMap<>(),
                     folder.getRoot().toPath(),
-                    mock(EthPeers.class)));
+                    mock(EthPeers.class),
+                    vertx));
 
     websocketMethods.putAll(rpcMethods);
     webSocketMessageHandlerSpy =