Merge branch 'master' of github.com:p2p-org/dshackle

.editorconfig

@@ -1,2 +1,3 @@
 [*.kt]
-continuation_indent_size = 4
+continuation_indent_size = 4
+disabled_rules=no-wildcard-imports

.gitignore

@@ -11,4 +11,5 @@ testsetup/
 *.iws
 
 # nix
-env
+env
+/dshackle-cli/node_modules/

.gitmodules

@@ -1,3 +1,6 @@
 [submodule "emerald-java-client"]
 	path = emerald-java-client
 	url = git@github.com:p2p-org/emerald-java-client.git
+[submodule "dshackle-cli/grpc"]
+	path = dshackle-cli/grpc
+	url = https://github.com/emeraldpay/emerald-grpc.git

build.gradle

@@ -164,7 +164,7 @@ jib {
         }
     }
     container {
-        jvmFlags = ['-Xms1024m']
+        jvmFlags = ['-Xms1024m', '-XX:MaxRAMPercentage=60']
         mainClass = 'io.emeraldpay.dshackle.StarterKt'
         args = []
         ports = ['2448', '2449', '8545']

dshackle-cli/README.md

@@ -0,0 +1,21 @@
+# DSHACKLE CLI
+
+A CLI tool to verify the [dshacle](https://github.com/emeraldpay/dshackle) installation.
+
+## Usage
+```
+npx dchackle-cli-tool [-p|--print] <dshackle instance url>
+```
+
+### Options
+-p | --print - will print the describe response as is
+
+## Example
+
+```
+> dchackle-cli-tool localhost:2449
+Connecting to: localhost:2449...
+Connected to  localhost:2449
+CHAIN_ETHEREUM -> AVAIL_OK
+CHAIN_KOVAN -> AVAIL_OK
+```

dshackle-cli/bin/dshackle-cli-tool

@@ -0,0 +1,4 @@
+#!/usr/bin/env node
+
+require = require('esm')(module /*, options*/);
+require('../src/cli').cli(process.argv);

dshackle-cli/package.json

@@ -0,0 +1,29 @@
+{
+  "name": "dshackle-cli-tool",
+  "version": "1.0.4",
+  "description": "",
+  "main": "src/index.js",
+  "scripts": {
+    "test": "echo \"Error: no test specified\" && exit 1"
+  },
+  "bin": {
+    "@mfomenkov/dshackle-cli-tool": "bin/dshackle-cli-tool",
+    "dshackle-cli-tool": "bin/dshackle-cli-tool"
+  },
+  "keywords": [],
+  "author": "",
+  "license": "MIT",
+  "dependencies": {
+    "@grpc/grpc-js": "^1.6.12",
+    "@grpc/proto-loader": "^0.7.2",
+    "arg": "^5.0.2",
+    "cli-color": "^2.0.3",
+    "esm": "^3.2.25",
+    "inquirer": "^9.1.1"
+  },
+  "files": [
+    "bin/",
+    "src/",
+    "grpc/"
+  ]
+}

dshackle-cli/src/cli.js

@@ -0,0 +1,57 @@
+import {describe} from "./grpc-clent";
+import arg from 'arg';
+import clc from "cli-color";
+import util from "util";
+
+export function cli(args) {
+    let opts = parseArgumentsIntoOptions(args);
+    if (!opts.url) {
+        console.log("Err: URL not specified!")
+        return
+    }
+    describe(opts.url, (error, response) => {
+        if (error) {
+            console.error(clc.red('Connection to ' + opts.url + ' failed! [' + error.message + ']'));
+        } else {
+            console.error(clc.green('Connected to ', opts.url));
+            if (opts.print) {
+                console.log(util.inspect(response, false, null, true /* enable colors */));
+            } else {
+                response.chains.forEach(function (item) {
+                    var state = item.status.availability;
+
+                    switch (state) {
+                        case 'AVAIL_OK':
+                            state = clc.green(state);
+                            break
+                        case 'AVAIL_UNKNOWN':
+                        case 'AVAIL_UNAVAILABLE':
+                            state = clc.red(state);
+                            break
+                        default:
+                            state = clc.yellow(state);
+                    }
+                    console.log(item.status.chain + ' -> ' + clc.bold(state))
+                })
+            }
+        }
+    })
+}
+
+function parseArgumentsIntoOptions(rawArgs) {
+    const args = arg(
+        {
+            '--print': Boolean,
+            '-p': '--print'
+        },
+        {
+            argv: rawArgs.slice(2),
+        }
+    );
+    return {
+        print: args['--print'] || false,
+        url: args._[0]
+    };
+}
+
+

dshackle-cli/src/grpc-clent.js

@@ -0,0 +1,25 @@
+const grpc = require("@grpc/grpc-js");
+const path = require('path')
+const protoLoader = require("@grpc/proto-loader");
+
+const PROTO_PATH = path.join(__dirname, "../grpc/proto/blockchain.proto");
+
+const options = {
+    keepCase: true,
+    longs: String,
+    enums: String,
+    defaults: true,
+    oneofs: true,
+};
+
+const packageDefinition = protoLoader.loadSync(PROTO_PATH, options);
+const emerald = grpc.loadPackageDefinition(packageDefinition).emerald
+
+export function describe(url, handler) {
+    const client = new emerald.Blockchain(
+        url,
+        grpc.credentials.createInsecure()
+    );
+    console.log('Connecting to: ' + url + '...')
+    client.Describe({}, handler);
+}

src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt

@@ -183,6 +183,9 @@ class UpstreamsConfigReader(
                 http.tls = authConfigReader.readClientTls(node)
             }
         }
+        getValueAsBool(connConfigNode, "prefer-http")?.let {
+            connection.preferHttp = it
+        }
         getMapping(connConfigNode, "ws")?.let { node ->
             getValueAsString(node, "url")?.let { url ->
                 val ws = UpstreamsConfig.WsEndpoint(URI(url))
@@ -256,7 +259,7 @@ class UpstreamsConfigReader(
     }
 
     internal fun readUpstreamGrpc(
-        upNode: MappingNode,
+        upNode: MappingNode
     ) {
         if (hasAny(upNode, "chain")) {
             log.warn("Chain should be not applied to gRPC upstream")
@@ -309,7 +312,8 @@ class UpstreamsConfigReader(
             }?.filterNotNull()?.toSet() ?: emptySet()
 
             UpstreamsConfig.Methods(
-                enabled, disabled
+                enabled,
+                disabled
             )
         }
     }

src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt

@@ -283,7 +283,7 @@ class BlockchainRpc(
             .tag("method", method)
             .publishPercentileHistogram()
             .register(Metrics.globalRegistry)
-        val nativeItemResponseErr = Counter.builder("request.grpc.native.request")
+        val nativeItemResponseErr = Counter.builder("request.grpc.native.request.err")
             .tag("chain", chain.chainCode)
             .tag("method", method)
             .register(Metrics.globalRegistry)

src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt

@@ -20,6 +20,7 @@ import io.emeraldpay.dshackle.FileResolver
 import io.emeraldpay.dshackle.Global
 import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.upstream.BlockValidator
 import io.emeraldpay.dshackle.upstream.CurrentMultistreamHolder
 import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.HttpRpcFactory
@@ -33,6 +34,7 @@ import io.emeraldpay.dshackle.upstream.bitcoin.ExtractBlock
 import io.emeraldpay.dshackle.upstream.bitcoin.ZMQServer
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
 import io.emeraldpay.dshackle.upstream.calls.ManagedCallMethods
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumBlockValidator
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosRpcUpstream
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumRpcUpstream
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsFactory
@@ -56,7 +58,7 @@ import javax.annotation.PostConstruct
 open class ConfiguredUpstreams(
     @Autowired private val currentUpstreams: CurrentMultistreamHolder,
     @Autowired private val fileResolver: FileResolver,
-    @Autowired private val config: UpstreamsConfig,
+    @Autowired private val config: UpstreamsConfig
 ) {
 
     private val log = LoggerFactory.getLogger(ConfiguredUpstreams::class.java)
@@ -159,7 +161,7 @@ open class ConfiguredUpstreams(
             return null
         }
         val urls = ArrayList<URI>()
-        val connectorFactory = buildEthereumConnectorFactory(config.id!!, execution, chain, urls, NoChoiceWithPriorityForkChoice(conn.upstreamRating))
+        val connectorFactory = buildEthereumConnectorFactory(config.id!!, execution, chain, urls, NoChoiceWithPriorityForkChoice(conn.upstreamRating), BlockValidator.ALWAYS_VALID)
         val methods = buildMethods(config, chain)
         if (connectorFactory == null) {
             return null
@@ -228,7 +230,7 @@ open class ConfiguredUpstreams(
         val urls = ArrayList<URI>()
         val methods = buildMethods(config, chain)
 
-        val connectorFactory = buildEthereumConnectorFactory(config.id!!, conn, chain, urls, MostWorkForkChoice())
+        val connectorFactory = buildEthereumConnectorFactory(config.id!!, conn, chain, urls, MostWorkForkChoice(), EthereumBlockValidator())
         if (connectorFactory == null) {
             return null
         }
@@ -297,11 +299,11 @@ open class ConfiguredUpstreams(
         }
     }
 
-    private fun buildEthereumConnectorFactory(id: String, conn: UpstreamsConfig.EthereumConnection, chain: Chain, urls: ArrayList<URI>, forkChoice: ForkChoice): EthereumConnectorFactory? {
+    private fun buildEthereumConnectorFactory(id: String, conn: UpstreamsConfig.EthereumConnection, chain: Chain, urls: ArrayList<URI>, forkChoice: ForkChoice, blockValidator: BlockValidator): EthereumConnectorFactory? {
         val wsFactoryApi = buildWsFactory(id, chain, conn, urls)
         val httpFactory = buildHttpFactory(conn, urls)
         log.info("Using ${chain.chainName} upstream, at ${urls.joinToString()}")
-        val connectorFactory = EthereumConnectorFactory(conn.preferHttp, wsFactoryApi, httpFactory, forkChoice)
+        val connectorFactory = EthereumConnectorFactory(conn.preferHttp, wsFactoryApi, httpFactory, forkChoice, blockValidator)
         if (!connectorFactory.isValid()) {
             log.warn("Upstream configuration is invalid (probably no http endpoint)")
             return null

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHead.kt

@@ -19,6 +19,7 @@ import io.emeraldpay.dshackle.Defaults
 import io.emeraldpay.dshackle.data.BlockContainer
 import io.emeraldpay.dshackle.reader.Reader
 import io.emeraldpay.dshackle.upstream.AbstractHead
+import io.emeraldpay.dshackle.upstream.BlockValidator
 import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
@@ -28,8 +29,9 @@ import org.slf4j.LoggerFactory
 import reactor.core.publisher.Mono
 
 open class DefaultEthereumHead(
-    forkChoice: ForkChoice
-) : Head, AbstractHead(forkChoice, EthereumBlockValidator()) {
+    forkChoice: ForkChoice,
+    blockValidator: BlockValidator
+) : Head, AbstractHead(forkChoice, blockValidator) {
 
     companion object {
         private val log = LoggerFactory.getLogger(DefaultEthereumHead::class.java)

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt

@@ -17,6 +17,7 @@
 package io.emeraldpay.dshackle.upstream.ethereum
 
 import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.upstream.BlockValidator
 import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
@@ -32,8 +33,9 @@ import java.util.concurrent.Executors
 class EthereumRpcHead(
     private val api: Reader<JsonRpcRequest, JsonRpcResponse>,
     forkChoice: ForkChoice,
+    blockValidator: BlockValidator,
     private val interval: Duration = Duration.ofSeconds(10),
-) : DefaultEthereumHead(forkChoice), Lifecycle {
+) : DefaultEthereumHead(forkChoice, blockValidator), Lifecycle {
 
     companion object {
         val scheduler =

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt

@@ -16,6 +16,7 @@
  */
 package io.emeraldpay.dshackle.upstream.ethereum
 
+import io.emeraldpay.dshackle.upstream.BlockValidator
 import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcWsClient
 import org.slf4j.LoggerFactory
@@ -25,8 +26,9 @@ import reactor.core.publisher.Flux
 
 class EthereumWsHead(
     private val ws: WsConnection,
-    forkChoice: ForkChoice
-) : DefaultEthereumHead(forkChoice), Lifecycle {
+    forkChoice: ForkChoice,
+    blockValidator: BlockValidator
+) : DefaultEthereumHead(forkChoice, blockValidator), Lifecycle {
 
     private val log = LoggerFactory.getLogger(EthereumWsHead::class.java)
 

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt

@@ -58,6 +58,7 @@ import java.time.Duration
 import java.time.Instant
 import java.util.Base64
 import java.util.concurrent.Executors
+import java.util.concurrent.ScheduledFuture
 import java.util.concurrent.TimeUnit
 import java.util.concurrent.TimeoutException
 import java.util.concurrent.atomic.AtomicBoolean
@@ -85,6 +86,7 @@ open class WsConnection(
         // > io.netty.handler.codec.http.websocketx.CorruptedWebSocketFrameException: Max frame length of 65536 has been exceeded
         // It's unclear what is the right limit here, but 5mb seems to be working (1mb isn't always working)
         private const val DEFAULT_FRAME_SIZE = 5 * 1024 * 1024
+        private const val RESET_BACKOFF_TIMEOUT = 10 * 1000
 
         // The max size from multiple frames that may represent a single message
         // Accept up to 15Mb messages, because Geth is using 15mb, though it's not clear what should be a right value
@@ -96,12 +98,15 @@ open class WsConnection(
 
     private var reconnectBackoff: BackOff = ExponentialBackOff().also {
         it.initialInterval = Duration.ofMillis(100).toMillis()
-        it.maxInterval = Duration.ofMinutes(1).toMillis()
+        it.maxInterval = Duration.ofMinutes(5).toMillis()
     }
     private var currentBackOff = reconnectBackoff.start()
 
     private val parser = ResponseWSParser()
 
+    private val resetBackoffExecutor = Executors.newScheduledThreadPool(2)
+    private var resetBackoffTask: ScheduledFuture<Unit>? = null
+
     private val blocks = Sinks
         .many()
         .multicast()
@@ -147,6 +152,7 @@ open class WsConnection(
         if (alreadyReconnecting) {
             return
         }
+
         // rpcSend is already CANCELLED, since the subscription owned by the previous connection is gone
         // so we need to create a new Sink. Emit Complete is probably useless, and just in case
         rpcSend.tryEmitComplete()
@@ -154,7 +160,11 @@ open class WsConnection(
             .many()
             .unicast()
             .onBackpressureBuffer<JsonRpcRequest>()
+        resetBackoffTask?.cancel(false)
         val retryInterval = currentBackOff.nextBackOff()
+        resetBackoffTask = resetBackoffExecutor.schedule<Unit>({
+            currentBackOff = reconnectBackoff.start()
+        }, RESET_BACKOFF_TIMEOUT + retryInterval, TimeUnit.MILLISECONDS)
         if (retryInterval == BackOffExecution.STOP) {
             log.warn("Reconnect backoff exhausted. Permanently closing the connection")
             return
@@ -165,7 +175,8 @@ open class WsConnection(
                 reconnecting.set(false)
                 connectInternal()
             },
-            retryInterval, TimeUnit.MILLISECONDS
+            retryInterval,
+            TimeUnit.MILLISECONDS
         )
     }
 
@@ -224,9 +235,6 @@ open class WsConnection(
     }
 
     fun handle(inbound: WebsocketInbound, outbound: WebsocketOutbound): Publisher<Void> {
-        // restart backoff after connection
-        currentBackOff = reconnectBackoff.start()
-
         val consumer = inbound
             .aggregateFrames(msgSizeLimit)
             .receiveFrames()
@@ -298,7 +306,10 @@ open class WsConnection(
     fun onRpc(msg: ResponseWSParser.WsResponse): Mono<Void> {
         return if (msg.id.isNumber()) {
             val resp = JsonRpcResponse(
-                msg.value, msg.error, msg.id, null
+                msg.value,
+                msg.error,
+                msg.id,
+                null
             )
             Mono.fromCallable {
                 val status = rpcReceive.tryEmitNext(resp)
@@ -377,10 +388,7 @@ open class WsConnection(
     fun sendRpc(request: JsonRpcRequest) {
         // submit to upstream in a separate thread, to free current thread (needs for subscription, etc)
         sendExecutor.execute {
-            val result = rpcSend.tryEmitNext(request)
-            if (result.isFailure) {
-                log.warn("Failed to send RPC request: $result")
-            }
+            rpcSend.emitNext(request) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED }
         }
     }
 

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnectorFactory.kt

@@ -1,5 +1,6 @@
 package io.emeraldpay.dshackle.upstream.ethereum.connectors
 
+import io.emeraldpay.dshackle.upstream.BlockValidator
 import io.emeraldpay.dshackle.upstream.DefaultUpstream
 import io.emeraldpay.dshackle.upstream.HttpFactory
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstreamValidator
@@ -12,7 +13,8 @@ open class EthereumConnectorFactory(
     private val preferHttp: Boolean,
     private val wsFactory: EthereumWsFactory?,
     private val httpFactory: HttpFactory?,
-    private val forkChoice: ForkChoice
+    private val forkChoice: ForkChoice,
+    private val blockValidator: BlockValidator
 ) : ConnectorFactory {
     private val log = LoggerFactory.getLogger(EthereumConnectorFactory::class.java)
 
@@ -25,11 +27,11 @@ open class EthereumConnectorFactory(
 
     override fun create(upstream: DefaultUpstream, validator: EthereumUpstreamValidator, chain: Chain): EthereumConnector {
         if (wsFactory != null && !preferHttp) {
-            return EthereumWsConnector(wsFactory, upstream, validator, chain, forkChoice)
+            return EthereumWsConnector(wsFactory, upstream, validator, chain, forkChoice, blockValidator)
         }
         if (httpFactory == null) {
             throw java.lang.IllegalArgumentException("Can't create rpc connector if no http factory set")
         }
-        return EthereumRpcConnector(httpFactory.create(upstream.getId(), chain), wsFactory, upstream.getId(), forkChoice)
+        return EthereumRpcConnector(httpFactory.create(upstream.getId(), chain), wsFactory, upstream.getId(), forkChoice, blockValidator)
     }
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumRpcConnector.kt

@@ -3,6 +3,7 @@ package io.emeraldpay.dshackle.upstream.ethereum.connectors
 import io.emeraldpay.dshackle.cache.Caches
 import io.emeraldpay.dshackle.cache.CachesEnabled
 import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.upstream.BlockValidator
 import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.MergedHead
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumRpcHead
@@ -20,7 +21,8 @@ class EthereumRpcConnector(
     private val directReader: Reader<JsonRpcRequest, JsonRpcResponse>,
     wsFactory: EthereumWsFactory?,
     id: String,
-    forkChoice: ForkChoice
+    forkChoice: ForkChoice,
+    blockValidator: BlockValidator
 ) : EthereumConnector, CachesEnabled {
     private val conn: WsConnection?
     private val head: Head
@@ -33,14 +35,14 @@ class EthereumRpcConnector(
         if (wsFactory != null) {
             // do not set upstream to the WS, since it doesn't control the RPC upstream
             conn = wsFactory.create(null, null)
-            val wsHead = EthereumWsHead(conn, forkChoice)
+            val wsHead = EthereumWsHead(conn, forkChoice, blockValidator)
             // receive bew blocks through WebSockets, but also periodically verify with RPC in case if WS failed
-            val rpcHead = EthereumRpcHead(directReader, forkChoice, Duration.ofSeconds(60))
+            val rpcHead = EthereumRpcHead(directReader, forkChoice, blockValidator, Duration.ofSeconds(60))
             head = MergedHead(listOf(rpcHead, wsHead), forkChoice)
         } else {
             conn = null
             log.warn("Setting up connector for $id upstream with RPC-only access, less effective than WS+RPC")
-            head = EthereumRpcHead(directReader, forkChoice)
+            head = EthereumRpcHead(directReader, forkChoice, blockValidator)
         }
     }
 

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt

@@ -1,12 +1,10 @@
 package io.emeraldpay.dshackle.upstream.ethereum.connectors
 
 import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.upstream.BlockValidator
 import io.emeraldpay.dshackle.upstream.DefaultUpstream
 import io.emeraldpay.dshackle.upstream.Head
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstreamValidator
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsFactory
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsHead
-import io.emeraldpay.dshackle.upstream.ethereum.WsConnection
+import io.emeraldpay.dshackle.upstream.ethereum.*
 import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
@@ -18,7 +16,8 @@ class EthereumWsConnector(
     upstream: DefaultUpstream,
     validator: EthereumUpstreamValidator,
     chain: Chain,
-    forkChoice: ForkChoice
+    forkChoice: ForkChoice,
+    blockValidator: BlockValidator
 ) : EthereumConnector {
     private val conn: WsConnection
     private val api: Reader<JsonRpcRequest, JsonRpcResponse>
@@ -26,7 +25,7 @@ class EthereumWsConnector(
 
     init {
         conn = wsFactory.create(upstream, validator)
-        head = EthereumWsHead(conn, forkChoice)
+        head = EthereumWsHead(conn, forkChoice, blockValidator)
         api = JsonRpcWsClient(conn)
     }
 

src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt

@@ -27,6 +27,7 @@ import io.netty.resolver.DefaultAddressResolverGroup
 import org.slf4j.LoggerFactory
 import reactor.core.publisher.Mono
 import reactor.netty.http.client.HttpClient
+import reactor.netty.resources.ConnectionProvider
 import java.io.ByteArrayInputStream
 import java.security.KeyStore
 import java.security.cert.CertificateFactory
@@ -53,7 +54,11 @@ class JsonRpcHttpClient(
     private val httpClient: HttpClient
 
     init {
-        var build = HttpClient.create()
+        val connectionProvider = ConnectionProvider.builder("dshackleConnectionPool")
+            .maxConnections(1000)
+            .pendingAcquireMaxCount(5000)
+            .build()
+        var build = HttpClient.create(connectionProvider)
             .resolver(DefaultAddressResolverGroup.INSTANCE)
 
         build = build.headers { h ->

src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy

@@ -48,7 +48,7 @@ class FilteredApisSpec extends Specification {
             def httpFactory = Mock(HttpFactory) {
                 create(_, _) >> TestingCommons.api().tap { it.id = "${i++}" }
             }
-            def connectorFactory = new EthereumConnectorFactory(false, null, httpFactory, new MostWorkForkChoice())
+            def connectorFactory = new EthereumConnectorFactory(false, null, httpFactory, new MostWorkForkChoice(), BlockValidator.@Companion.ALWAYS_VALID)
             new EthereumRpcUpstream(
                     "test",
                     Chain.ETHEREUM,

src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHeadSpec.groovy

@@ -20,6 +20,7 @@ import com.fasterxml.jackson.databind.ObjectMapper
 import io.emeraldpay.dshackle.Global
 import io.emeraldpay.dshackle.data.BlockContainer
 import io.emeraldpay.dshackle.test.TestingCommons
+import io.emeraldpay.dshackle.upstream.BlockValidator
 import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice
 import io.emeraldpay.etherjar.domain.BlockHash
 import io.emeraldpay.etherjar.rpc.json.BlockJson
@@ -31,7 +32,7 @@ import java.time.Instant
 
 class DefaultEthereumHeadSpec extends Specification {
 
-    DefaultEthereumHead head = new DefaultEthereumHead(new MostWorkForkChoice())
+    DefaultEthereumHead head = new DefaultEthereumHead(new MostWorkForkChoice(), BlockValidator.@Companion.ALWAYS_VALID)
     ObjectMapper objectMapper = Global.objectMapper
 
     def blocks = (10L..20L).collect { i ->